#!/usr/bin/env python3 """Secret-holding Orga CalDAV write service. Full Radicale admin: create/update/delete VEVENTs in any collection, not just calendar/public (see the "05-calendar-admin-full-access" rights rule in hosts/pinix/localServices/calendar/radicale.nix). Update and delete always require a concrete --collection/--uid the caller already knows -- there is no listing, search, or wildcard/bulk-delete operation here, by design. """ import base64 import datetime as dt import http.client import json import os import re import socket import ssl import sys import urllib.parse import uuid from pathlib import Path MAX_REQUEST = 16_384 CONTROL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]") UID = re.compile(r"[A-Za-z0-9._@+-]{1,255}\Z") # Alnum-first segments block ".."-style traversal outright; up to three # segments covers every collection shape seen in radicale.nix ({user}, # {user}/{cal}, calendar/public) without hardcoding a specific one. COLLECTION = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}(/[A-Za-z0-9][A-Za-z0-9._-]{0,63}){0,2}\Z") class RequestError(Exception): pass def secret(path_variable): value = Path(os.environ[path_variable]).read_text(encoding="utf-8").strip() if not value: raise RuntimeError("configured credential is empty") return value def text(value, name, maximum, allow_empty=False): if not isinstance(value, str) or len(value) > maximum or CONTROL.search(value): raise RequestError(f"invalid {name}") if not allow_empty and not value.strip(): raise RequestError(f"missing {name}") return value def timestamp(value, name): value = text(value, name, 40) try: parsed = dt.datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as error: raise RequestError(f"invalid {name}") from error if parsed.tzinfo is None: raise RequestError(f"{name} must include a timezone") return parsed.astimezone(dt.timezone.utc) def collection_path(value): value = text(value, "collection", 200) if not COLLECTION.fullmatch(value): raise RequestError("invalid collection") return value def event_uid(value): value = text(value, "uid", 255) if not UID.fullmatch(value): raise RequestError("invalid uid") return value def ics_escape(value): return ( value.replace("\\", "\\\\") .replace(";", "\\;") .replace(",", "\\,") .replace("\r\n", "\\n") .replace("\n", "\\n") .replace("\r", "\\n") ) def build_vevent(uid, dtstamp, dtstart, dtend, title, description, location): lines = [ "BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//0ff.space//orga-agent calendar-write//DE", "CALSCALE:GREGORIAN", "BEGIN:VEVENT", f"UID:{uid}", f"DTSTAMP:{dtstamp}", f"DTSTART:{dtstart}", f"DTEND:{dtend}", f"SUMMARY:{ics_escape(title)}", ] if description: lines.append(f"DESCRIPTION:{ics_escape(description)}") if location: lines.append(f"LOCATION:{ics_escape(location)}") lines += ["END:VEVENT", "END:VCALENDAR", ""] return "\r\n".join(lines) def base_url(): base = os.environ["ORGA_CALENDAR_URL"].rstrip("/") parsed = urllib.parse.urlsplit(base) if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: raise RuntimeError("calendar URL must be credential-free HTTPS") return parsed def auth_header(): user = os.environ["ORGA_CALENDAR_USER"] token = base64.b64encode(f"{user}:{secret('ORGA_CALENDAR_PASSWORD_FILE')}".encode()).decode() return f"Basic {token}" def item_href(parsed, collection, uid): return f"{parsed.path.rstrip('/')}/{collection}/{uid}.ics" def fetch_etag(connection, href): connection.request("GET", href, headers={"Authorization": auth_header()}) response = connection.getresponse() response.read(2_000_001) if response.status == 404: return None if response.status != 200: raise RuntimeError(f"calendar server returned HTTP {response.status} while reading event") etag = response.getheader("ETag") if not etag: raise RuntimeError("calendar server did not return an ETag for the event") return etag def put_event(connection, href, ics_body, if_match=None, if_none_match=None): headers = { "Authorization": auth_header(), "Content-Type": "text/calendar; charset=utf-8", } if if_match: headers["If-Match"] = if_match if if_none_match: headers["If-None-Match"] = if_none_match connection.request("PUT", href, body=ics_body.encode("utf-8"), headers=headers) response = connection.getresponse() response.read(65_536) if response.status not in (200, 201, 204): raise RuntimeError(f"calendar server returned HTTP {response.status} while writing event") def delete_event(connection, href, if_match): connection.request("DELETE", href, headers={"Authorization": auth_header(), "If-Match": if_match}) response = connection.getresponse() response.read(65_536) if response.status not in (200, 204): raise RuntimeError(f"calendar server returned HTTP {response.status} while deleting event") def open_connection(parsed): return http.client.HTTPSConnection(parsed.hostname, parsed.port or 443, context=ssl.create_default_context(), timeout=20) def validate_event_fields(request): title = text(request.get("title"), "title", 500) description = text(request.get("description", ""), "description", 20_000, allow_empty=True) location = text(request.get("location", ""), "location", 1000, allow_empty=True) start = timestamp(request.get("start"), "start") end = timestamp(request.get("end"), "end") if end <= start or end - start > dt.timedelta(days=31): raise RequestError("event duration must be positive and at most 31 days") return title, description, location, start, end def calendar_write_request(request): # Every field is validated before base_url()/open_connection() ever run, # so a bad request never touches the network or the secret. operation = request.get("operation") if operation == "capabilities" and set(request) == {"operation"}: return { "ok": True, "capability": "calendar-write", "writes": True, "scope": "admin: any collection", "operations": ["create", "update", "delete"], } if operation == "create": if set(request) - {"operation", "collection", "title", "start", "end", "description", "location"}: raise RequestError("unsupported create request") collection = collection_path(request.get("collection")) title, description, location, start, end = validate_event_fields(request) elif operation == "update": if set(request) - {"operation", "collection", "uid", "title", "start", "end", "description", "location"}: raise RequestError("unsupported update request") collection = collection_path(request.get("collection")) uid = event_uid(request.get("uid")) title, description, location, start, end = validate_event_fields(request) elif operation == "delete": if set(request) - {"operation", "collection", "uid"}: raise RequestError("unsupported delete request") collection = collection_path(request.get("collection")) uid = event_uid(request.get("uid")) else: raise RequestError("unsupported calendar-write request") parsed = base_url() connection = open_connection(parsed) try: if operation == "create": uid = f"{uuid.uuid4()}@{parsed.hostname}" dtstamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ") href = item_href(parsed, collection, uid) body = build_vevent( uid, dtstamp, start.strftime("%Y%m%dT%H%M%SZ"), end.strftime("%Y%m%dT%H%M%SZ"), title, description, location, ) # Fresh uuid4 UID makes a collision virtually impossible, but # If-None-Match: * still guarantees this can never silently # overwrite an existing item. put_event(connection, href, body, if_none_match="*") return {"ok": True, "operation": "create", "collection": collection, "uid": uid} if operation == "update": href = item_href(parsed, collection, uid) etag = fetch_etag(connection, href) if etag is None: raise RequestError("event was not found; refusing to create one under 'update'") dtstamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%dT%H%M%SZ") body = build_vevent( uid, dtstamp, start.strftime("%Y%m%dT%H%M%SZ"), end.strftime("%Y%m%dT%H%M%SZ"), title, description, location, ) put_event(connection, href, body, if_match=etag) return {"ok": True, "operation": "update", "collection": collection, "uid": uid} href = item_href(parsed, collection, uid) etag = fetch_etag(connection, href) if etag is None: raise RequestError("event was not found") delete_event(connection, href, if_match=etag) return {"ok": True, "operation": "delete", "collection": collection, "uid": uid} finally: connection.close() def serve(): listener = socket.socket(fileno=0) while True: connection, _ = listener.accept() with connection: try: raw = b"" while not raw.endswith(b"\n") and len(raw) <= MAX_REQUEST: chunk = connection.recv(4096) if not chunk: break raw += chunk if len(raw) > MAX_REQUEST: raise RequestError("request is too large") request = json.loads(raw) if not isinstance(request, dict): raise RequestError("request must be an object") response = calendar_write_request(request) except (RequestError, json.JSONDecodeError) as error: response = {"ok": False, "error": str(error)} except Exception: response = {"ok": False, "error": "upstream service request failed"} connection.sendall(json.dumps(response, ensure_ascii=False).encode("utf-8")) def main(): serve() if __name__ == "__main__": sys.exit(main() or 0)