"""WebDAV XML: reading request bodies, writing multistatus answers (RFC 4918, 4791, 6352, 6578). Properties are ``{namespace}name`` strings, as ElementTree names them.""" from __future__ import annotations import xml.etree.ElementTree as ET from dataclasses import dataclass, field from datetime import UTC, datetime from defusedxml import ElementTree as SafeET D = "DAV:" C = "urn:ietf:params:xml:ns:caldav" CR = "urn:ietf:params:xml:ns:carddav" CS = "http://calendarserver.org/ns/" A = "http://apple.com/ns/ical/" for _prefix, _ns in (("d", D), ("cal", C), ("card", CR), ("cs", CS), ("ical", A)): ET.register_namespace(_prefix, _ns) def q(ns: str, name: str) -> str: return f"{{{ns}}}{name}" class BadRequest(Exception): pass def parse(body: bytes) -> ET.Element | None: if not body.strip(): return None try: root = SafeET.fromstring(body) except Exception as err: # noqa: BLE001 - any malformed XML is the client's error raise BadRequest(str(err)) from err return root # type: ignore[no-any-return] @dataclass class PropRequest: """PROPFIND: which properties (None = all).""" names: list[str] | None = None names_only: bool = False def propfind_request(root: ET.Element | None) -> PropRequest: if root is None or root.find(q(D, "allprop")) is not None: return PropRequest() if root.find(q(D, "propname")) is not None: return PropRequest(names_only=True) prop = root.find(q(D, "prop")) if prop is None: return PropRequest() return PropRequest(names=[child.tag for child in prop]) def requested_props(root: ET.Element) -> list[str] | None: prop = root.find(q(D, "prop")) if prop is None: return None return [child.tag for child in prop] @dataclass class Report: # calendar-query | calendar-multiget | addressbook-query | addressbook-multiget # | sync-collection kind: str props: list[str] | None hrefs: list[str] = field(default_factory=list) start: datetime | None = None end: datetime | None = None sync_token: str = "" limit: int | None = None def _ical_time(value: str | None) -> datetime | None: if not value: return None try: return datetime.strptime(value.strip(), "%Y%m%dT%H%M%SZ").replace(tzinfo=UTC) except ValueError: return None def report_request(root: ET.Element) -> Report: tag = root.tag names = { q(C, "calendar-query"): "calendar-query", q(C, "calendar-multiget"): "calendar-multiget", q(CR, "addressbook-query"): "addressbook-query", q(CR, "addressbook-multiget"): "addressbook-multiget", q(D, "sync-collection"): "sync-collection", } kind = names.get(tag) if kind is None: raise BadRequest(f"unsupported report {tag}") rep = Report(kind=kind, props=requested_props(root)) rep.hrefs = [(h.text or "").strip() for h in root.findall(q(D, "href"))] if kind == "calendar-query": tr = root.find(f".//{q(C, 'time-range')}") if tr is not None: rep.start = _ical_time(tr.get("start")) rep.end = _ical_time(tr.get("end")) if kind == "sync-collection": rep.sync_token = (root.findtext(q(D, "sync-token")) or "").strip() nresults = root.find(f"{q(D, 'limit')}/{q(D, 'nresults')}") if nresults is not None and (nresults.text or "").strip().isdigit(): rep.limit = int((nresults.text or "").strip()) return rep @dataclass class PropPatch: set: dict[str, ET.Element] = field(default_factory=dict) remove: list[str] = field(default_factory=list) def proppatch_request(root: ET.Element | None) -> PropPatch: out = PropPatch() if root is None: return out for action in root: prop = action.find(q(D, "prop")) if prop is None: continue for child in prop: if action.tag == q(D, "set"): out.set[child.tag] = child elif action.tag == q(D, "remove"): out.remove.append(child.tag) return out def mkcol_props(root: ET.Element | None) -> dict[str, ET.Element]: """MKCALENDAR / extended MKCOL: the properties to set on the new collection.""" if root is None: return {} out: dict[str, ET.Element] = {} for sett in root.findall(q(D, "set")): prop = sett.find(q(D, "prop")) if prop is not None: for child in prop: out[child.tag] = child return out # ---- answers -------------------------------------------------------------------------- # A property's value: text, a list of child elements, or an element builder. PropValue = str | list[ET.Element] | None class Multistatus: def __init__(self) -> None: self.root = ET.Element(q(D, "multistatus")) def response( self, href: str, found: dict[str, PropValue], missing: list[str] | None = None, status: int | None = None, ) -> None: resp = ET.SubElement(self.root, q(D, "response")) ET.SubElement(resp, q(D, "href")).text = href if status is not None: ET.SubElement(resp, q(D, "status")).text = _status_line(status) return if found: ps = ET.SubElement(resp, q(D, "propstat")) prop = ET.SubElement(ps, q(D, "prop")) for name, value in found.items(): el = ET.SubElement(prop, name) if isinstance(value, str): el.text = value elif isinstance(value, list): el.extend(value) ET.SubElement(ps, q(D, "status")).text = _status_line(200) if missing: ps = ET.SubElement(resp, q(D, "propstat")) prop = ET.SubElement(ps, q(D, "prop")) for name in missing: ET.SubElement(prop, name) ET.SubElement(ps, q(D, "status")).text = _status_line(404) def sync_token(self, token: str) -> None: ET.SubElement(self.root, q(D, "sync-token")).text = token def to_bytes(self) -> bytes: body: bytes = ET.tostring(self.root, encoding="utf-8") return b'\n' + body def el( tag: str, text: str | None = None, children: list[ET.Element] | None = None, attrs: dict[str, str] | None = None, ) -> ET.Element: e = ET.Element(tag, attrs or {}) if text is not None: e.text = text if children: e.extend(children) return e def href(path: str) -> ET.Element: return el(q(D, "href"), path) def error_body(condition: str) -> bytes: root = ET.Element(q(D, "error")) ET.SubElement(root, condition) return b'\n' + bytes(ET.tostring(root, encoding="utf-8")) _REASONS = { 200: "OK", 201: "Created", 204: "No Content", 207: "Multi-Status", 403: "Forbidden", 404: "Not Found", 409: "Conflict", 412: "Precondition Failed", 507: "Insufficient Storage", } def _status_line(code: int) -> str: return f"HTTP/1.1 {code} {_REASONS.get(code, '')}".rstrip()