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

#!/usr/bin/env python3
"""Unprivileged clients for the local, secret-holding Orga read services."""

import argparse
import json
import socket
import sys

MAX_RESPONSE = 1_000_000


def request(socket_path, payload):
    encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8") + b"\n"
    if len(encoded) > 16_384:
        raise ValueError("request is too large")
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as connection:
        connection.settimeout(30)
        connection.connect(socket_path)
        connection.sendall(encoded)
        connection.shutdown(socket.SHUT_WR)
        chunks = []
        size = 0
        while True:
            chunk = connection.recv(65_536)
            if not chunk:
                break
            size += len(chunk)
            if size > MAX_RESPONSE:
                raise ValueError("service response is too large")
            chunks.append(chunk)
    response = json.loads(b"".join(chunks))
    if not response.get("ok"):
        raise RuntimeError(response.get("error", "service request failed"))
    return response


def calendar_parser():
    parser = argparse.ArgumentParser(description="Read the allowlisted Radicale calendar")
    parser.add_argument("--socket", default="/run/orga-calendar-read.sock", help=argparse.SUPPRESS)
    sub = parser.add_subparsers(dest="operation", required=True)
    sub.add_parser("capabilities")
    events = sub.add_parser("events")
    events.add_argument("--calendar", default="public", choices=["public"])
    events.add_argument("--start", required=True, help="ISO-8601 timestamp with timezone")
    events.add_argument("--end", required=True, help="ISO-8601 timestamp with timezone")
    return parser


def calendar_write_parser():
    parser = argparse.ArgumentParser(
        description="Create, update or delete Radicale calendar events (full admin, any collection)"
    )
    parser.add_argument("--socket", default="/run/orga-calendar-write.sock", help=argparse.SUPPRESS)
    sub = parser.add_subparsers(dest="operation", required=True)
    sub.add_parser("capabilities")

    create = sub.add_parser("create")
    create.add_argument("--collection", required=True, help="e.g. calendar/public or <user>/<calendar>")
    create.add_argument("--title", required=True)
    create.add_argument("--start", required=True, help="ISO-8601 timestamp with timezone")
    create.add_argument("--end", required=True, help="ISO-8601 timestamp with timezone")
    create.add_argument("--description", default="")
    create.add_argument("--location", default="")

    update = sub.add_parser("update")
    update.add_argument("--collection", required=True)
    update.add_argument("--uid", required=True, help="concrete event UID -- required, no wildcard updates")
    update.add_argument("--title", required=True)
    update.add_argument("--start", required=True, help="ISO-8601 timestamp with timezone")
    update.add_argument("--end", required=True, help="ISO-8601 timestamp with timezone")
    update.add_argument("--description", default="")
    update.add_argument("--location", default="")

    delete = sub.add_parser("delete")
    delete.add_argument("--collection", required=True)
    delete.add_argument("--uid", required=True, help="concrete event UID -- required, no wildcard/bulk delete")
    return parser


def mail_parser():
    parser = argparse.ArgumentParser(description="Read Zoho mail without changing mailbox state")
    parser.add_argument("--socket", default="/run/orga-mail-read.sock", help=argparse.SUPPRESS)
    sub = parser.add_subparsers(dest="operation", required=True)
    sub.add_parser("capabilities")
    sub.add_parser("folders")
    search = sub.add_parser("search")
    search.add_argument("--folder", default="INBOX")
    search.add_argument("--since", required=True, help="YYYY-MM-DD")
    search.add_argument("--limit", type=int, default=20)
    show = sub.add_parser("show")
    show.add_argument("--folder", default="INBOX")
    show.add_argument("--uid", required=True)
    return parser


def mail_draft_parser():
    parser = argparse.ArgumentParser(
        description="Manage mail drafts in the account's real IMAP Drafts folder "
        "(write access limited to that folder; no SMTP, no send)"
    )
    parser.add_argument("--socket", default="/run/orga-mail-draft.sock", help=argparse.SUPPRESS)
    sub = parser.add_subparsers(dest="operation", required=True)
    sub.add_parser("capabilities")
    sub.add_parser("list")

    def add_content_args(target):
        target.add_argument("--to", action="append", required=True, help="repeatable, at least one recipient")
        target.add_argument("--cc", action="append", default=[])
        target.add_argument("--bcc", action="append", default=[])
        target.add_argument("--subject", required=True)
        target.add_argument("--body", required=True)
        target.add_argument("--attachment", action="append", default=[], dest="attachments",
                             help="repeatable; filename or description, no file content")

    create = sub.add_parser("create")
    add_content_args(create)

    update = sub.add_parser("update")
    update.add_argument("--id", required=True)
    add_content_args(update)

    show = sub.add_parser("show")
    show.add_argument("--id", required=True)

    delete = sub.add_parser("delete")
    delete.add_argument("--id", required=True)
    return parser


def main():
    if len(sys.argv) < 2 or sys.argv[1] not in {"calendar", "calendar-write", "mail", "mail-draft"}:
        print("internal usage: client.py calendar|calendar-write|mail|mail-draft ...", file=sys.stderr)
        return 2
    capability = sys.argv.pop(1)
    if capability == "calendar":
        parser = calendar_parser()
    elif capability == "calendar-write":
        parser = calendar_write_parser()
    elif capability == "mail-draft":
        parser = mail_draft_parser()
    else:
        parser = mail_parser()
    args = parser.parse_args()
    payload = vars(args)
    socket_path = payload.pop("socket")
    try:
        response = request(socket_path, payload)
    except (OSError, ValueError, RuntimeError, json.JSONDecodeError) as error:
        print(f"orga-{capability}: {error}", file=sys.stderr)
        return 1
    print(json.dumps(response, ensure_ascii=False, indent=2))
    return 0


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