nixos-config/hosts/blunix/localServices/orga-agent/preview.py

#!/usr/bin/env python3
"""Validate and canonicalize Orga write previews; never performs a write."""

import argparse
import datetime as dt
import json
import re
import sys
from email.utils import getaddresses
from pathlib import Path

CONTROL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
UID = re.compile(r"[A-Za-z0-9._@+-]{1,255}\Z")
ADDRESS = re.compile(
    r"[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+@"
    r"[A-Za-z0-9](?:[A-Za-z0-9.-]{0,251}[A-Za-z0-9])?\Z"
)


def fail(message):
    raise ValueError(message)


def string(value, name, maximum, allow_empty=False):
    if not isinstance(value, str) or len(value) > maximum or CONTROL.search(value):
        fail(f"invalid {name}")
    if not allow_empty and not value.strip():
        fail(f"missing {name}")
    return value


def exact_keys(data, required, optional=()):
    if not isinstance(data, dict) or set(data) - set(required) - set(optional) or set(required) - set(data):
        fail("input fields do not match the operation schema")


def addresses(value, name):
    if not isinstance(value, list) or len(value) > 50:
        fail(f"invalid {name}")
    result = []
    for item in value:
        item = string(item, name, 320)
        parsed = getaddresses([item])
        if (
            len(parsed) != 1
            or not ADDRESS.fullmatch(parsed[0][1])
            or "\n" in item
            or "\r" in item
        ):
            fail(f"invalid {name} address")
        result.append(item)
    return result


def iso(value, name):
    value = string(value, name, 40)
    try:
        parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as error:
        raise ValueError(f"invalid {name}") from error
    if parsed.tzinfo is None:
        fail(f"{name} must include a timezone")
    return value, parsed


def validate(data):
    operation = data.get("operation") if isinstance(data, dict) else None
    if operation == "mail-send":
        exact_keys(data, ["operation", "to", "cc", "bcc", "subject", "body"], ["attachments"])
        if data.get("attachments", []) != []:
            fail("attachments are not supported by the safe preview")
        canonical = {
            "operation": operation,
            "to": addresses(data["to"], "to"),
            "cc": addresses(data["cc"], "cc"),
            "bcc": addresses(data["bcc"], "bcc"),
            "subject": string(data["subject"], "subject", 998),
            "body": string(data["body"], "body", 100_000),
            "attachments": [],
        }
        if not canonical["to"]:
            fail("at least one recipient is required")
    elif operation in {"calendar-create", "calendar-update"}:
        required = ["operation", "calendar", "title", "start", "end", "description", "location"]
        if operation == "calendar-update":
            required.append("uid")
        exact_keys(data, required)
        start_value, start = iso(data["start"], "start")
        end_value, end = iso(data["end"], "end")
        if end <= start or end - start > dt.timedelta(days=31):
            fail("event duration must be positive and at most 31 days")
        canonical = {
            "operation": operation, "calendar": data["calendar"],
            "title": string(data["title"], "title", 500), "start": start_value,
            "end": end_value, "description": string(data["description"], "description", 20_000, True),
            "location": string(data["location"], "location", 1000, True),
        }
        if data["calendar"] != "public":
            fail("calendar is not allowlisted")
        if operation == "calendar-update":
            canonical["uid"] = string(data["uid"], "uid", 255)
            if not UID.fullmatch(canonical["uid"]):
                fail("invalid uid")
    elif operation == "calendar-delete":
        exact_keys(data, ["operation", "calendar", "uid"])
        canonical = {"operation": operation, "calendar": data["calendar"], "uid": string(data["uid"], "uid", 255)}
        if canonical["calendar"] != "public" or not UID.fullmatch(canonical["uid"]):
            fail("calendar or uid is not allowlisted")
    else:
        fail("unsupported operation")
    return {"status": "PREVIEW_ONLY_REQUIRES_ORCHESTRATOR_CONFIRMATION", "executable": False, "request": canonical}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", required=True, help="JSON request file; use - for stdin")
    args = parser.parse_args()
    try:
        raw = sys.stdin.read(200_001) if args.input == "-" else Path(args.input).read_text(encoding="utf-8")
        if len(raw) > 200_000:
            fail("input is too large")
        print(json.dumps(validate(json.loads(raw)), ensure_ascii=False, indent=2))
    except (OSError, ValueError, json.JSONDecodeError) as error:
        print(f"orga-write-preview: {error}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main())