webmail-client/plugins/webmail_plugin_calendar/webmail_plugin_calendar/caldav.py

"""Minimal CalDAV client for two-way sync (SPEC §A.3).

Speaks CalDAV over the SDK's SSRF-guarded HTTP: RFC 6764 discovery
(current-user-principal → calendar-home-set → calendar collections), an RFC 4791
``calendar-query`` REPORT to pull events, and PUT/DELETE (with ETags) to push local
changes. iCloud redirects to a per-user partition host, so requests follow a few
redirects (each still SSRF-validated by the underlying guard).

The pure helpers — ``parse_multistatus_calendars``, ``parse_report``,
``vevent_to_fields``, ``build_ics`` — take/return plain data and are unit-tested
without a network.
"""

from __future__ import annotations

import logging
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any
from urllib.parse import urljoin

from defusedxml import ElementTree as SafeET

logger = logging.getLogger("plg.calendar.caldav")

# Well-known base URLs for providers that don't need a user-supplied server URL.
PROVIDER_BASE = {"icloud": "https://caldav.icloud.com/"}

_DAV = "{DAV:}"
_CAL = "{urn:ietf:params:xml:ns:caldav}"
_CS = "{http://calendarserver.org/ns/}"
_ICAL = "{http://apple.com/ns/ical/}"

_PRINCIPAL_BODY = (
    b'<?xml version="1.0" encoding="utf-8"?>'
    b'<d:propfind xmlns:d="DAV:"><d:prop><d:current-user-principal/></d:prop></d:propfind>'
)
_HOME_BODY = (
    b'<?xml version="1.0" encoding="utf-8"?>'
    b'<d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav">'
    b"<d:prop><c:calendar-home-set/></d:prop></d:propfind>"
)
_CAL_LIST_BODY = (
    b'<?xml version="1.0" encoding="utf-8"?>'
    b'<d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav"'
    b' xmlns:cs="http://calendarserver.org/ns/" xmlns:ical="http://apple.com/ns/ical/">'
    b"<d:prop><d:resourcetype/><d:displayname/><ical:calendar-color/><cs:getctag/>"
    b"<d:current-user-privilege-set/><c:supported-calendar-component-set/></d:prop></d:propfind>"
)
_CTAG_BODY = (
    b'<?xml version="1.0" encoding="utf-8"?>'
    b'<d:propfind xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/">'
    b"<d:prop><cs:getctag/></d:prop></d:propfind>"
)
_EVENTS_QUERY = (
    b'<?xml version="1.0" encoding="utf-8"?>'
    b'<c:calendar-query xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav">'
    b"<d:prop><d:getetag/><c:calendar-data/></d:prop>"
    b'<c:filter><c:comp-filter name="VCALENDAR"><c:comp-filter name="VEVENT"/>'
    b"</c:comp-filter></c:filter></c:calendar-query>"
)


class CalDavError(Exception):
    """A CalDAV request failed (auth, network, or unexpected status)."""


@dataclass
class DavCalendar:
    url: str
    name: str
    color: str = "#3b82f6"
    ctag: str = ""
    read_only: bool = False


@dataclass
class DavEvent:
    href: str
    etag: str
    ics: str


# --- pure parsing/building (unit-tested) -------------------------------------


def _text(el: Any, path: str) -> str:
    found = el.find(path)
    return (found.text or "").strip() if found is not None and found.text else ""


def _propstat_ok(response: Any) -> Any | None:
    """Return the ``<prop>`` element of the HTTP-200 propstat in a response, or None."""
    for propstat in response.findall(f"{_DAV}propstat"):
        status = _text(propstat, f"{_DAV}status")
        if " 200 " in status:
            return propstat.find(f"{_DAV}prop")
    # Some servers omit the status; fall back to the first prop.
    ps = response.find(f"{_DAV}propstat")
    return ps.find(f"{_DAV}prop") if ps is not None else None


def parse_href(xml_text: str, prop_tag: str) -> str | None:
    """First ``<prop_tag><d:href>`` value in a multistatus (discovery steps)."""
    root = SafeET.fromstring(xml_text)
    for response in root.iter(f"{_DAV}response"):
        prop = _propstat_ok(response)
        if prop is None:
            continue
        target = prop.find(prop_tag)
        if target is not None:
            href = target.find(f"{_DAV}href")
            if href is not None and href.text:
                return href.text.strip()
    return None


def parse_multistatus_calendars(xml_text: str, base_url: str) -> list[DavCalendar]:
    """Calendar collections from a Depth-1 PROPFIND on the calendar-home-set."""
    root = SafeET.fromstring(xml_text)
    out: list[DavCalendar] = []
    for response in root.iter(f"{_DAV}response"):
        prop = _propstat_ok(response)
        href_el = response.find(f"{_DAV}href")
        if prop is None or href_el is None or not href_el.text:
            continue
        resourcetype = prop.find(f"{_DAV}resourcetype")
        if resourcetype is None or resourcetype.find(f"{_CAL}calendar") is None:
            continue  # not a calendar collection (e.g. the home node itself, inbox/outbox)
        # Only calendars that hold events.
        comps = prop.find(f"{_CAL}supported-calendar-component-set")
        if comps is not None:
            names = {c.get("name") for c in comps.findall(f"{_CAL}comp")}
            if names and "VEVENT" not in names:
                continue
        name = _text(prop, f"{_DAV}displayname") or "Calendar"
        color = (_text(prop, f"{_ICAL}calendar-color") or "#3b82f6")[:7]
        ctag = _text(prop, f"{_CS}getctag")
        # read-only unless the privilege set grants write.
        read_only = True
        privset = prop.find(f"{_DAV}current-user-privilege-set")
        if privset is None:
            read_only = False  # server didn't report; assume writable
        else:
            for priv in privset.iter(f"{_DAV}privilege"):
                if priv.find(f"{_DAV}write") is not None or priv.find(f"{_DAV}all") is not None:
                    read_only = False
                    break
        out.append(
            DavCalendar(
                url=urljoin(base_url, href_el.text.strip()),
                name=name,
                color=color or "#3b82f6",
                ctag=ctag,
                read_only=read_only,
            )
        )
    return out


def parse_report(xml_text: str, base_url: str) -> list[DavEvent]:
    """(href, etag, calendar-data) tuples from a calendar-query/multiget REPORT."""
    root = SafeET.fromstring(xml_text)
    out: list[DavEvent] = []
    for response in root.iter(f"{_DAV}response"):
        href_el = response.find(f"{_DAV}href")
        prop = _propstat_ok(response)
        if href_el is None or not href_el.text or prop is None:
            continue
        cdata = _text(prop, f"{_CAL}calendar-data")
        if not cdata:
            continue
        out.append(
            DavEvent(
                href=urljoin(base_url, href_el.text.strip()),
                etag=_text(prop, f"{_DAV}getetag").strip('"'),
                ics=cdata,
            )
        )
    return out


def _aware(dt: Any) -> datetime:
    if not isinstance(dt, datetime):
        dt = datetime(dt.year, dt.month, dt.day)
    return dt if dt.tzinfo else dt.replace(tzinfo=UTC)


def vevent_to_fields(ics_text: str) -> dict[str, Any] | None:
    """Map the first VEVENT in an ICS resource to plg_calendar_events columns."""
    from icalendar import Calendar as ICalendar

    try:
        cal = ICalendar.from_ical(ics_text)
    except Exception:  # noqa: BLE001 - malformed resource; skip it
        return None
    comp = next((c for c in cal.walk("VEVENT")), None)
    if comp is None:
        return None
    start = comp.get("dtstart")
    if start is None:
        return None
    all_day = not isinstance(start.dt, datetime)
    dtstart = _aware(start.dt)
    end = comp.get("dtend")
    dtend = _aware(end.dt) if end is not None else None
    rrule = comp.get("rrule")
    return {
        "ical_uid": str(comp.get("uid") or f"{uuid.uuid4()}@caldav"),
        "sequence": int(comp.get("sequence") or 0),
        "summary": str(comp.get("summary") or ""),
        "description": str(comp.get("description") or ""),
        "location": str(comp.get("location") or ""),
        "dtstart": dtstart,
        "dtend": dtend,
        "all_day": all_day,
        "rrule": rrule.to_ical().decode() if rrule is not None else None,
        "status": str(comp.get("status") or "confirmed").lower(),
        "raw_ical": ics_text,
    }


def build_ics(event: dict[str, Any]) -> str:
    """A single-VEVENT VCALENDAR for PUTting a locally created/edited event."""
    from icalendar import Calendar as ICalendar
    from icalendar import Event as IEvent
    from icalendar.prop import vRecur

    cal = ICalendar()
    cal.add("prodid", "-//Ultimo//Calendar//EN")
    cal.add("version", "2.0")
    ev = IEvent()
    ev.add("uid", event["ical_uid"])
    ev.add("sequence", int(event.get("sequence") or 0))
    ev.add("summary", event.get("summary") or "")
    if event.get("description"):
        ev.add("description", event["description"])
    if event.get("location"):
        ev.add("location", event["location"])
    start, end = event["dtstart"], event.get("dtend")
    if event.get("all_day"):
        ev.add("dtstart", start.date() if isinstance(start, datetime) else start)
        if end is not None:
            ev.add("dtend", end.date() if isinstance(end, datetime) else end)
    else:
        ev.add("dtstart", start)
        if end is not None:
            ev.add("dtend", end)
    if event.get("rrule"):
        ev.add("rrule", vRecur.from_ical(str(event["rrule"])))
    for ex in event.get("exdates") or []:
        ev.add("exdate", ex.date() if event.get("all_day") and isinstance(ex, datetime) else ex)
    ev.add("dtstamp", datetime.now(UTC))
    cal.add_component(ev)
    return cal.to_ical().decode()


# --- network operations ------------------------------------------------------


async def _request(
    http: Any,
    method: str,
    url: str,
    *,
    creds: tuple[str, str],
    depth: str | None = None,
    body: bytes | None = None,
    extra_headers: dict[str, str] | None = None,
) -> Any:
    """One CalDAV request, following up to 5 redirects (iCloud partition hosts).

    Each hop goes through ``sdk.http`` (SSRF-guarded). Returns the final response."""
    headers: dict[str, str] = {"User-Agent": "Ultimo-Webmail/1.0"}
    if body is not None and method in ("PROPFIND", "REPORT"):
        headers["Content-Type"] = "application/xml; charset=utf-8"
    if depth is not None:
        headers["Depth"] = depth
    if extra_headers:
        headers.update(extra_headers)
    current = url
    resp = None
    for _ in range(6):
        try:
            resp = await http.request(
                method, current, headers=headers, content=body, auth=creds
            )
        except Exception as exc:  # transport error or SSRF guard rejection
            name = type(exc).__name__.lower()
            msg = str(exc).lower()
            if "ssrf" in name or "public" in msg or "private" in msg or "not allowed" in msg:
                raise CalDavError("private_host") from exc
            raise CalDavError("unreachable") from exc
        if resp.status in (301, 302, 307, 308):
            location = resp.headers.get("location")
            if not location:
                break
            current = urljoin(current, location)
            continue
        return resp
    return resp


async def discover_calendars(
    http: Any, base_url: str, creds: tuple[str, str]
) -> list[DavCalendar]:
    """RFC 6764 discovery: principal → calendar-home-set → calendar collections."""
    resp = await _request(http, "PROPFIND", base_url, creds=creds, depth="0", body=_PRINCIPAL_BODY)
    if resp is None or resp.status in (401, 403):
        raise CalDavError("auth_failed")
    if resp.status not in (207, 200):
        raise CalDavError(f"principal lookup failed ({resp.status})")
    principal = parse_href(resp.text, f"{_DAV}current-user-principal") or base_url
    principal_url = urljoin(base_url, principal)

    resp = await _request(
        http, "PROPFIND", principal_url, creds=creds, depth="0", body=_HOME_BODY
    )
    if resp is None or resp.status not in (207, 200):
        raise CalDavError("home-set lookup failed")
    home = parse_href(resp.text, f"{_CAL}calendar-home-set")
    home_url = urljoin(principal_url, home) if home else principal_url

    resp = await _request(http, "PROPFIND", home_url, creds=creds, depth="1", body=_CAL_LIST_BODY)
    if resp is None or resp.status not in (207, 200):
        raise CalDavError("calendar list failed")
    return parse_multistatus_calendars(resp.text, home_url)


async def get_ctag(http: Any, calendar_url: str, creds: tuple[str, str]) -> str:
    resp = await _request(http, "PROPFIND", calendar_url, creds=creds, depth="0", body=_CTAG_BODY)
    if resp is None or resp.status not in (207, 200):
        return ""
    root = SafeET.fromstring(resp.text)
    for response in root.iter(f"{_DAV}response"):
        prop = _propstat_ok(response)
        if prop is not None:
            return _text(prop, f"{_CS}getctag")
    return ""


async def pull_events(http: Any, calendar_url: str, creds: tuple[str, str]) -> list[DavEvent]:
    resp = await _request(
        http, "REPORT", calendar_url, creds=creds, depth="1", body=_EVENTS_QUERY
    )
    if resp is None or resp.status not in (207, 200):
        raise CalDavError(f"event report failed ({resp.status if resp else 'no response'})")
    return parse_report(resp.text, calendar_url)


def _if_match(etag: str) -> str:
    """Format a stored (quote-stripped) ETag as an HTTP entity-tag for If-Match.

    RFC 7232 requires the value be a quoted-string; Sabre/NextCloud reject a bare
    token with 412. Weak validators (``W/"…"``) and already-quoted values pass through."""
    etag = etag.strip()
    if etag.startswith("W/") or (etag.startswith('"') and etag.endswith('"')):
        return etag
    return f'"{etag}"'


async def put_event(
    http: Any, url: str, ics: str, etag: str | None, creds: tuple[str, str]
) -> str:
    """Create/update an event resource; return the server's new ETag (may be empty)."""
    extra = {"Content-Type": "text/calendar; charset=utf-8"}
    if etag:
        extra["If-Match"] = _if_match(etag)  # update: only if unchanged since we last saw it
    else:
        extra["If-None-Match"] = "*"  # create: only if it doesn't already exist
    resp = await _request(http, "PUT", url, creds=creds, body=ics.encode(), extra_headers=extra)
    if resp is None or resp.status not in (200, 201, 204):
        raise CalDavError(f"PUT failed ({resp.status if resp else 'no response'})")
    return (resp.headers.get("etag") or "").strip('"')


async def delete_event(http: Any, url: str, etag: str | None, creds: tuple[str, str]) -> None:
    extra = {"If-Match": _if_match(etag)} if etag else {}
    resp = await _request(http, "DELETE", url, creds=creds, extra_headers=extra)
    # 404/412 (already gone / changed) are treated as done — the local intent (remove) holds.
    if resp is None or resp.status not in (200, 204, 404, 412):
        raise CalDavError(f"DELETE failed ({resp.status})")


def event_href(calendar_url: str, ical_uid: str) -> str:
    """Deterministic resource path for a locally created event (UID.ics under the collection)."""
    base = calendar_url if calendar_url.endswith("/") else calendar_url + "/"
    safe = "".join(ch for ch in ical_uid if ch.isalnum() or ch in "-_.@") or uuid.uuid4().hex
    return urljoin(base, f"{safe}.ics")