from __future__ import annotations
import base64
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from typing import Mapping, Protocol
from xml.etree import ElementTree
class CalDAVError(RuntimeError):
pass
class CalDAVSyncUnsupported(CalDAVError):
pass
class CalDAVPreconditionFailed(CalDAVError):
pass
class CalDAVNotFound(CalDAVError):
pass
class CalDAVTransport(Protocol):
def __call__(self, method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: int) -> tuple[int, Mapping[str, str], bytes]:
...
@dataclass(frozen=True, slots=True)
class CalDAVObject:
href: str
etag: str | None = None
calendar_data: str | None = None
deleted: bool = False
@dataclass(frozen=True, slots=True)
class CalDAVReportResult:
objects: list[CalDAVObject] = field(default_factory=list)
sync_token: str | None = None
ctag: str | None = None
@dataclass(frozen=True, slots=True)
class CalDAVWriteResult:
href: str
etag: str | None = None
status: int = 0
@dataclass(frozen=True, slots=True)
class CalDAVDiscoveryCalendar:
collection_url: str
href: str
display_name: str | None = None
color: str | None = None
ctag: str | None = None
sync_token: str | None = None
@dataclass(frozen=True, slots=True)
class _DAVDiscoveryResponse:
href: str
display_name: str | None = None
color: str | None = None
ctag: str | None = None
sync_token: str | None = None
is_calendar: bool = False
principal_hrefs: tuple[str, ...] = ()
calendar_home_set_hrefs: tuple[str, ...] = ()
supported_components: tuple[str, ...] = ()
class CalDAVClient:
def __init__(
self,
*,
collection_url: str,
username: str | None = None,
password: str | None = None,
bearer_token: str | None = None,
timeout: int = 30,
transport: CalDAVTransport | None = None,
) -> None:
self.collection_url = ensure_collection_url(collection_url)
self.username = username
self.password = password
self.bearer_token = bearer_token
self.timeout = timeout
self.transport = transport or urllib_transport
def propfind_collection(self) -> CalDAVReportResult:
body = b"""
"""
payload = self.request("PROPFIND", self.collection_url, body=body, depth="0", expected={207})
return parse_multistatus(payload)
def discover_calendars(self) -> list[CalDAVDiscoveryCalendar]:
start_url = self.collection_url
calendars: dict[str, CalDAVDiscoveryCalendar] = {}
home_urls: list[str] = []
principal_urls: list[str] = []
visited_urls: set[tuple[str, str]] = set()
errors: list[str] = []
def add_home_href(base_url: str, href: str) -> None:
url = ensure_collection_url(absolute_dav_url(base_url, href))
if url not in home_urls:
home_urls.append(url)
def add_principal_href(base_url: str, href: str) -> None:
url = ensure_collection_url(absolute_dav_url(base_url, href))
if url not in principal_urls:
principal_urls.append(url)
def add_calendar(base_url: str, response: _DAVDiscoveryResponse) -> None:
if not response.is_calendar:
return
if response.supported_components and "VEVENT" not in response.supported_components:
return
url = ensure_collection_url(absolute_dav_url(base_url, response.href or base_url))
calendars[url] = CalDAVDiscoveryCalendar(
collection_url=url,
href=response.href,
display_name=response.display_name,
color=response.color,
ctag=response.ctag,
sync_token=response.sync_token,
)
def propfind(url: str, depth: str) -> list[_DAVDiscoveryResponse]:
key = (url, depth)
if key in visited_urls:
return []
visited_urls.add(key)
return self.propfind_discovery(url, depth=depth)
try:
for response in propfind(start_url, "0"):
add_calendar(start_url, response)
for href in response.calendar_home_set_hrefs:
add_home_href(start_url, href)
for href in response.principal_hrefs:
add_principal_href(start_url, href)
except CalDAVError as exc:
errors.append(str(exc))
for principal_url in principal_urls[:6]:
try:
for response in propfind(principal_url, "0"):
for href in response.calendar_home_set_hrefs:
add_home_href(principal_url, href)
except CalDAVError as exc:
errors.append(str(exc))
if not home_urls:
home_urls.append(start_url)
for home_url in home_urls:
try:
for response in propfind(home_url, "1"):
add_calendar(home_url, response)
except CalDAVError as exc:
errors.append(str(exc))
if not calendars and errors:
raise CalDAVError(f"CalDAV discovery did not find any calendar collections: {errors[0]}")
return sorted(calendars.values(), key=lambda item: ((item.display_name or item.collection_url).lower(), item.collection_url))
def propfind_discovery(self, url: str, *, depth: str) -> list[_DAVDiscoveryResponse]:
body = b"""
"""
payload = self.request("PROPFIND", ensure_collection_url(url), body=body, depth=depth, expected={207})
return parse_discovery_multistatus(payload)
def list_objects(self) -> CalDAVReportResult:
body = b"""
"""
payload = self.request("REPORT", self.collection_url, body=body, depth="1", expected={207})
return parse_multistatus(payload)
def sync_collection(self, sync_token: str) -> CalDAVReportResult:
body = f"""
{xml_escape(sync_token)}
1
""".encode("utf-8")
try:
payload = self.request("REPORT", self.collection_url, body=body, depth="1", expected={207})
except CalDAVError as exc:
raise CalDAVSyncUnsupported(str(exc)) from exc
return parse_multistatus(payload)
def fetch_object(self, href: str) -> str:
payload = self.request("GET", self.object_url(href), body=None, depth=None, expected={200})
return payload.decode("utf-8")
def put_object(self, href: str, ics: str, *, etag: str | None = None, create: bool = False, overwrite: bool = False) -> CalDAVWriteResult:
headers = {"Content-Type": "text/calendar; charset=utf-8"}
if create:
headers["If-None-Match"] = "*"
elif etag and not overwrite:
headers["If-Match"] = etag
elif not overwrite:
raise CalDAVPreconditionFailed("Refusing CalDAV PUT without an ETag")
status, response_headers, _payload = self.request_raw(
"PUT",
self.object_url(href),
body=ics.encode("utf-8"),
depth=None,
expected={200, 201, 204},
extra_headers=headers,
)
return CalDAVWriteResult(href=href, etag=response_etag(response_headers), status=status)
def delete_object(self, href: str, *, etag: str | None = None, overwrite: bool = False) -> CalDAVWriteResult:
headers: dict[str, str] = {}
if etag and not overwrite:
headers["If-Match"] = etag
elif not overwrite:
raise CalDAVPreconditionFailed("Refusing CalDAV DELETE without an ETag")
status, response_headers, _payload = self.request_raw(
"DELETE",
self.object_url(href),
body=None,
depth=None,
expected={200, 202, 204, 404},
extra_headers=headers,
)
return CalDAVWriteResult(href=href, etag=response_etag(response_headers), status=status)
def object_url(self, href: str) -> str:
return urllib.parse.urljoin(self.collection_url, href)
def request(self, method: str, url: str, *, body: bytes | None, depth: str | None, expected: set[int]) -> bytes:
_status, _headers, payload = self.request_raw(method, url, body=body, depth=depth, expected=expected)
return payload
def request_raw(
self,
method: str,
url: str,
*,
body: bytes | None,
depth: str | None,
expected: set[int],
extra_headers: Mapping[str, str] | None = None,
) -> tuple[int, Mapping[str, str], bytes]:
headers: dict[str, str] = {
"Accept": "application/xml,text/calendar,*/*",
"User-Agent": "govoplan-calendar-caldav/0.1",
}
if body is not None:
headers["Content-Type"] = "application/xml; charset=utf-8"
if depth is not None:
headers["Depth"] = depth
if self.bearer_token:
headers["Authorization"] = f"Bearer {self.bearer_token}"
elif self.username and self.password:
token = base64.b64encode(f"{self.username}:{self.password}".encode("utf-8")).decode("ascii")
headers["Authorization"] = f"Basic {token}"
if extra_headers:
headers.update(dict(extra_headers))
status, response_headers, payload = self.transport(method, url, headers, body, self.timeout)
if status not in expected:
if status == 404:
raise CalDAVNotFound(f"{method} {url} returned HTTP {status}")
if status == 412:
raise CalDAVPreconditionFailed(f"{method} {url} failed because the remote resource changed")
raise CalDAVError(f"{method} {url} returned HTTP {status}")
return status, response_headers, payload
def urllib_transport(method: str, url: str, headers: Mapping[str, str], body: bytes | None, timeout: int) -> tuple[int, Mapping[str, str], bytes]:
try:
request = urllib.request.Request(url, data=body, headers=dict(headers), method=method)
with urllib.request.urlopen(request, timeout=timeout) as response:
return response.status, dict(response.headers.items()), response.read()
except urllib.error.HTTPError as exc:
return exc.code, dict(exc.headers.items()), exc.read()
except urllib.error.URLError as exc:
raise CalDAVError(f"{method} {url} failed: {exc.reason}") from exc
except ValueError as exc:
raise CalDAVError(f"{method} {url} failed: {exc}") from exc
def parse_multistatus(payload: bytes) -> CalDAVReportResult:
try:
root = ElementTree.fromstring(payload)
except ElementTree.ParseError as exc:
raise CalDAVError(f"Invalid CalDAV XML response: {exc}") from exc
objects: list[CalDAVObject] = []
sync_token = first_child_text(root, "sync-token")
ctag = first_child_text(root, "getctag")
for response in child_elements(root, "response"):
href = first_child_text(response, "href")
if not href:
continue
deleted = False
etag = None
calendar_data = None
for propstat in child_elements(response, "propstat"):
status = first_child_text(propstat, "status") or ""
prop = first_child(propstat, "prop")
if prop is None:
continue
if " 404 " in status or status.endswith(" 404"):
deleted = True
continue
if " 200 " not in status and not status.endswith(" 200"):
continue
etag = first_child_text(prop, "getetag") or etag
calendar_data = first_child_text(prop, "calendar-data") or calendar_data
sync_token = first_child_text(prop, "sync-token") or sync_token
ctag = first_child_text(prop, "getctag") or ctag
objects.append(CalDAVObject(href=href, etag=strip_weak_etag(etag), calendar_data=calendar_data, deleted=deleted))
return CalDAVReportResult(objects=objects, sync_token=sync_token, ctag=ctag)
def parse_discovery_multistatus(payload: bytes) -> list[_DAVDiscoveryResponse]:
try:
root = ElementTree.fromstring(payload)
except ElementTree.ParseError as exc:
raise CalDAVError(f"Invalid CalDAV XML response: {exc}") from exc
responses: list[_DAVDiscoveryResponse] = []
for response in child_elements(root, "response"):
href = first_child_text(response, "href")
if not href:
continue
display_name = None
color = None
ctag = None
sync_token = None
is_calendar = False
principal_hrefs: list[str] = []
calendar_home_set_hrefs: list[str] = []
supported_components: list[str] = []
for propstat in child_elements(response, "propstat"):
status = first_child_text(propstat, "status") or ""
if status and " 200 " not in status and not status.endswith(" 200") and " 207 " not in status:
continue
prop = first_child(propstat, "prop")
if prop is None:
continue
for item in prop:
name = local_name(item.tag)
if name == "displayname" and item.text:
display_name = item.text.strip() or display_name
elif name == "calendar-color" and item.text:
color = item.text.strip() or color
elif name == "getctag" and item.text:
ctag = item.text.strip() or ctag
elif name == "sync-token" and item.text:
sync_token = item.text.strip() or sync_token
elif name == "resourcetype":
is_calendar = is_calendar or any(local_name(child.tag) == "calendar" for child in item)
elif name in {"current-user-principal", "principal-URL"}:
principal_hrefs.extend(nested_href_texts(item))
elif name == "calendar-home-set":
calendar_home_set_hrefs.extend(nested_href_texts(item))
elif name == "supported-calendar-component-set":
for component in item:
if local_name(component.tag) == "comp":
component_name = (component.attrib.get("name") or "").strip().upper()
if component_name:
supported_components.append(component_name)
responses.append(
_DAVDiscoveryResponse(
href=href,
display_name=display_name,
color=color,
ctag=ctag,
sync_token=sync_token,
is_calendar=is_calendar,
principal_hrefs=tuple(dict.fromkeys(principal_hrefs)),
calendar_home_set_hrefs=tuple(dict.fromkeys(calendar_home_set_hrefs)),
supported_components=tuple(dict.fromkeys(supported_components)),
)
)
return responses
def ensure_collection_url(value: str) -> str:
value = value.strip()
if value and "://" not in value and not value.startswith("/"):
value = f"https://{value}"
return value if value.endswith("/") else f"{value}/"
def absolute_dav_url(base_url: str, href: str) -> str:
return urllib.parse.urljoin(ensure_collection_url(base_url), href)
def strip_weak_etag(value: str | None) -> str | None:
if value is None:
return None
return value.strip()
def response_etag(headers: Mapping[str, str]) -> str | None:
for key, value in headers.items():
if key.lower() == "etag" and value:
return strip_weak_etag(value)
return None
def xml_escape(value: str) -> str:
return value.replace("&", "&").replace("<", "<").replace(">", ">")
def first_child(element: ElementTree.Element, name: str) -> ElementTree.Element | None:
for child in element:
if local_name(child.tag) == name:
return child
return None
def first_child_text(element: ElementTree.Element, name: str) -> str | None:
found = first_child(element, name)
if found is None or found.text is None:
return None
return found.text.strip()
def child_elements(element: ElementTree.Element, name: str) -> list[ElementTree.Element]:
return [child for child in element if local_name(child.tag) == name]
def nested_href_texts(element: ElementTree.Element) -> list[str]:
hrefs: list[str] = []
for child in element.iter():
if local_name(child.tag) == "href" and child.text and child.text.strip():
hrefs.append(child.text.strip())
return hrefs
def local_name(tag: str) -> str:
return tag.rsplit("}", 1)[-1] if "}" in tag else tag