GovOPlaN/govoplan-calendar
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:10 +02:00
parent c071235ad7
commit 371a00aff5
8 changed files with 561 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
src/govoplan_calendar/backend/caldav.py
View File

@@ -5,8 +5,7 @@ import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from typing import Mapping, Protocol
from xml.etree import ElementTree
from typing import Any, Mapping, Protocol
from defusedxml import ElementTree as SafeElementTree
@@ -77,6 +76,18 @@ class _DAVDiscoveryResponse:
supported_components: tuple[str, ...] = ()
@dataclass(slots=True)
class _DAVDiscoveryDraft:
display_name: str | None = None
color: str | None = None
ctag: str | None = None
sync_token: str | None = None
is_calendar: bool = False
principal_hrefs: list[str] = field(default_factory=list)
calendar_home_set_hrefs: list[str] = field(default_factory=list)
supported_components: list[str] = field(default_factory=list)
class CalDAVClient:
def __init__(
self,
@@ -313,7 +324,7 @@ def urllib_transport(method: str, url: str, headers: Mapping[str, str], body: by
url = validate_http_url(url)
try:
request = urllib.request.Request(url, data=body, headers=dict(headers), method=method)
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - validated CalDAV HTTP(S) URL. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - validated CalDAV HTTP(S) URL. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
return response.status, dict(response.headers.items()), response.read()
except urllib.error.HTTPError as exc:
return exc.code, dict(exc.headers.items()), exc.read()
@@ -361,62 +372,86 @@ def parse_discovery_multistatus(payload: bytes) -> list[_DAVDiscoveryResponse]:
root = SafeElementTree.fromstring(payload)
except SafeElementTree.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:
return [
parsed
for response in child_elements(root, "response")
if (parsed := _parse_discovery_response(response)) is not None
]
def _parse_discovery_response(response: Any) -> _DAVDiscoveryResponse | None:
href = first_child_text(response, "href")
if not href:
return None
draft = _DAVDiscoveryDraft()
for propstat in child_elements(response, "propstat"):
_apply_discovery_propstat(draft, propstat)
return _DAVDiscoveryResponse(
href=href,
display_name=draft.display_name,
color=draft.color,
ctag=draft.ctag,
sync_token=draft.sync_token,
is_calendar=draft.is_calendar,
principal_hrefs=dedupe_tuple(draft.principal_hrefs),
calendar_home_set_hrefs=dedupe_tuple(draft.calendar_home_set_hrefs),
supported_components=dedupe_tuple(draft.supported_components),
)
def _apply_discovery_propstat(draft: _DAVDiscoveryDraft, propstat: Any) -> None:
if not discovery_propstat_is_success(propstat):
return
prop = first_child(propstat, "prop")
if prop is None:
return
for item in prop:
_apply_discovery_property(draft, item)
def discovery_propstat_is_success(propstat: Any) -> bool:
status = first_child_text(propstat, "status") or ""
return not status or " 200 " in status or status.endswith(" 200") or " 207 " in status
def _apply_discovery_property(draft: _DAVDiscoveryDraft, item: Any) -> None:
name = local_name(item.tag)
text = item.text.strip() if item.text else ""
if name == "displayname" and text:
draft.display_name = text or draft.display_name
elif name == "calendar-color" and text:
draft.color = text or draft.color
elif name == "getctag" and text:
draft.ctag = text or draft.ctag
elif name == "sync-token" and text:
draft.sync_token = text or draft.sync_token
elif name == "resourcetype":
draft.is_calendar = draft.is_calendar or discovery_resource_is_calendar(item)
elif name in {"current-user-principal", "principal-URL"}:
draft.principal_hrefs.extend(nested_href_texts(item))
elif name == "calendar-home-set":
draft.calendar_home_set_hrefs.extend(nested_href_texts(item))
elif name == "supported-calendar-component-set":
draft.supported_components.extend(supported_calendar_component_names(item))
def discovery_resource_is_calendar(item: Any) -> bool:
return any(local_name(child.tag) == "calendar" for child in item)
def supported_calendar_component_names(item: Any) -> list[str]:
names: list[str] = []
for component in item:
if local_name(component.tag) != "comp":
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
component_name = (component.attrib.get("name") or "").strip().upper()
if component_name:
names.append(component_name)
return names
def dedupe_tuple(values: list[str]) -> tuple[str, ...]:
return tuple(dict.fromkeys(values))
def ensure_collection_url(value: str) -> str:
@@ -457,25 +492,25 @@ def xml_escape(value: str) -> str:
return value.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
def first_child(element: ElementTree.Element, name: str) -> ElementTree.Element | None:
def first_child(element: Any, name: str) -> Any | 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:
def first_child_text(element: Any, 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]:
def child_elements(element: Any, name: str) -> list[Any]:
return [child for child in element if local_name(child.tag) == name]
def nested_href_texts(element: ElementTree.Element) -> list[str]:
def nested_href_texts(element: Any) -> list[str]:
hrefs: list[str] = []
for child in element.iter():
if local_name(child.tag) == "href" and child.text and child.text.strip():