webmail-client/backend/app/dav/xml.py

"""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'<?xml version="1.0" encoding="utf-8"?>\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'<?xml version="1.0" encoding="utf-8"?>\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()