#!/usr/bin/env python3 """Secret-holding, read-only CalDAV and IMAP service endpoints.""" import argparse import base64 import datetime as dt import email import html import http.client import imaplib import json import os import re import socket import ssl import sys import urllib.parse import xml.etree.ElementTree as ET from email.header import decode_header from pathlib import Path MAX_REQUEST = 16_384 MAX_BODY = 50_000 CONTROL = re.compile(r"[\x00-\x1f\x7f]") UID = re.compile(r"[1-9][0-9]{0,18}\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=256): if not isinstance(value, str) or not value or len(value) > maximum or CONTROL.search(value): raise RequestError(f"invalid {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 decoded_header(value): result = [] for part, charset in decode_header(value or ""): if isinstance(part, bytes): result.append(part.decode(charset or "utf-8", errors="replace")) else: result.append(part) return " ".join("".join(result).split())[:2000] def unfold_ical(raw): return re.sub(r"\r?\n[ \t]", "", raw).splitlines() def unescape_ical(value): return ( value.replace("\\n", "\n") .replace("\\N", "\n") .replace("\\,", ",") .replace("\\;", ";") .replace("\\\\", "\\") ) def parse_events(raw, href): events = [] current = None for line in unfold_ical(raw): if line == "BEGIN:VEVENT": current = {"href": href} elif line == "END:VEVENT" and current is not None: events.append(current) current = None elif current is not None and ":" in line: key, value = line.split(":", 1) name = key.split(";", 1)[0] mapped = { "UID": "uid", "SUMMARY": "summary", "DTSTART": "start", "DTEND": "end", "DESCRIPTION": "description", "LOCATION": "location", }.get(name) if mapped and mapped not in current: current[mapped] = unescape_ical(value)[:MAX_BODY] return events def calendar_request(request): operation = request.get("operation") if operation == "capabilities" and set(request) == {"operation"}: return {"ok": True, "capability": "calendar-read", "methods": ["REPORT"], "writes": False} if operation != "events" or set(request) - {"operation", "calendar", "start", "end"}: raise RequestError("unsupported calendar request") if request.get("calendar", "public") != "public": raise RequestError("calendar is not allowlisted") start = timestamp(request.get("start"), "start") end = timestamp(request.get("end"), "end") if end <= start or end - start > dt.timedelta(days=366): raise RequestError("calendar range must be positive and at most 366 days") base_url = os.environ["ORGA_CALENDAR_URL"].rstrip("/") + "/" parsed = urllib.parse.urlsplit(base_url) if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: raise RuntimeError("calendar URL must be credential-free HTTPS") body = """ """ % (start.strftime("%Y%m%dT%H%M%SZ"), end.strftime("%Y%m%dT%H%M%SZ")) user = os.environ["ORGA_CALENDAR_USER"] auth = base64.b64encode(f"{user}:{secret('ORGA_CALENDAR_PASSWORD_FILE')}".encode()).decode() connection = http.client.HTTPSConnection(parsed.hostname, parsed.port or 443, context=ssl.create_default_context(), timeout=20) try: connection.request("REPORT", parsed.path, body=body.encode(), headers={ "Authorization": f"Basic {auth}", "Content-Type": "application/xml; charset=utf-8", "Depth": "1", }) response = connection.getresponse() payload = response.read(2_000_001) finally: connection.close() if len(payload) > 2_000_000: raise RuntimeError("calendar response is too large") if response.status != 207: raise RuntimeError(f"calendar server returned HTTP {response.status}") root = ET.fromstring(payload) events = [] for item in root.findall("{DAV:}response"): href = item.findtext("{DAV:}href", "")[:1000] for data in item.findall(".//{urn:ietf:params:xml:ns:caldav}calendar-data"): events.extend(parse_events(data.text or "", href)) return {"ok": True, "trust": "UNTRUSTED_EXTERNAL_DATA", "calendar": "public", "events": events[:500], "truncated": len(events) > 500} def mail_login(): connection = imaplib.IMAP4_SSL(os.environ["ORGA_MAIL_HOST"], timeout=20) connection.login(os.environ["ORGA_MAIL_USER"], secret("ORGA_MAIL_PASSWORD_FILE")) return connection def select_readonly(connection, folder): folder = text(folder, "folder", 256) status, _ = connection.select(folder, readonly=True) if status != "OK": raise RequestError("folder is unavailable") def message_body(message): candidates = [] for part in message.walk() if message.is_multipart() else [message]: if part.get_content_maintype() == "multipart" or part.get_filename(): continue if part.get_content_type() not in {"text/plain", "text/html"}: continue payload = part.get_payload(decode=True) or b"" charset = part.get_content_charset() or "utf-8" rendered = payload.decode(charset, errors="replace") if part.get_content_type() == "text/html": rendered = html.unescape(re.sub(r"<[^>]+>", " ", rendered)) rendered = " ".join(rendered.split()) candidates.append((part.get_content_type() != "text/plain", rendered)) candidates.sort(key=lambda candidate: candidate[0]) return (candidates[0][1] if candidates else "")[:MAX_BODY] def message_summary(uid, message, include_body=False): result = { "uid": uid, "from": decoded_header(message.get("From")), "to": decoded_header(message.get("To")), "cc": decoded_header(message.get("Cc")), "date": (message.get("Date") or "")[:1000], "subject": decoded_header(message.get("Subject")), "attachments": [decoded_header(part.get_filename()) for part in message.walk() if part.get_filename()][:100], } if include_body: result["body"] = message_body(message) return result def mail_request(request): operation = request.get("operation") if operation == "capabilities" and set(request) == {"operation"}: return {"ok": True, "capability": "mail-read", "mailbox_mode": "EXAMINE", "fetch_mode": "BODY.PEEK", "writes": False} connection = mail_login() try: if operation == "folders" and set(request) == {"operation"}: status, values = connection.list() if status != "OK": raise RuntimeError("IMAP LIST failed") folders = [value.decode(errors="replace")[:1000] for value in (values or [])] return {"ok": True, "trust": "UNTRUSTED_EXTERNAL_DATA", "folders": folders[:500]} if operation == "search" and not set(request) - {"operation", "folder", "since", "limit"}: select_readonly(connection, request.get("folder", "INBOX")) try: since = dt.date.fromisoformat(text(request.get("since"), "since", 10)) except ValueError as error: raise RequestError("invalid since date") from error limit = request.get("limit", 20) if not isinstance(limit, int) or not 1 <= limit <= 50: raise RequestError("limit must be between 1 and 50") status, values = connection.uid("search", None, "SINCE", since.strftime("%d-%b-%Y")) if status != "OK": raise RuntimeError("IMAP UID SEARCH failed") uids = list(reversed((values[0] or b"").split()))[:limit] messages = [] for uid_bytes in uids: status, data = connection.uid("fetch", uid_bytes, "(BODY.PEEK[HEADER.FIELDS (FROM TO CC DATE SUBJECT)] BODYSTRUCTURE)") if status != "OK" or not data or not isinstance(data[0], tuple): continue messages.append(message_summary(uid_bytes.decode(), email.message_from_bytes(data[0][1]))) return {"ok": True, "trust": "UNTRUSTED_EXTERNAL_DATA", "folder": request.get("folder", "INBOX"), "messages": messages} if operation == "show" and not set(request) - {"operation", "folder", "uid"}: select_readonly(connection, request.get("folder", "INBOX")) uid = text(request.get("uid"), "uid", 19) if not UID.fullmatch(uid): raise RequestError("invalid uid") status, data = connection.uid("fetch", uid, "(BODY.PEEK[])") if status != "OK" or not data or not isinstance(data[0], tuple): raise RequestError("message was not found") return {"ok": True, "trust": "UNTRUSTED_EXTERNAL_DATA", "folder": request.get("folder", "INBOX"), "message": message_summary(uid, email.message_from_bytes(data[0][1]), True)} raise RequestError("unsupported mail request") finally: try: connection.logout() except imaplib.IMAP4.error: pass def serve(capability): listener = socket.socket(fileno=0) handler = calendar_request if capability == "calendar" else mail_request 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 = handler(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(): parser = argparse.ArgumentParser() parser.add_argument("--capability", required=True, choices=["calendar", "mail"]) args = parser.parse_args() serve(args.capability) if __name__ == "__main__": main()