from __future__ import annotations
import base64
import posixpath
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Mapping, Protocol
from defusedxml import ElementTree as SafeElementTree
class AddressCardDAVError(RuntimeError):
pass
class AddressCardDAVSyncUnsupported(AddressCardDAVError):
pass
class AddressCardDAVNotFound(AddressCardDAVError):
pass
class AddressCardDAVPreconditionFailed(AddressCardDAVError):
pass
class AddressCardDAVTransport(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 AddressCardDAVObject:
href: str
etag: str | None = None
address_data: str | None = None
deleted: bool = False
@dataclass(frozen=True, slots=True)
class AddressCardDAVReportResult:
objects: list[AddressCardDAVObject] = field(default_factory=list)
sync_token: str | None = None
ctag: str | None = None
@dataclass(frozen=True, slots=True)
class AddressCardDAVAddressBook:
collection_url: str
href: str
display_name: str | None = None
description: str | None = None
ctag: str | None = None
sync_token: str | None = None
@dataclass(frozen=True, slots=True)
class AddressCardDAVWriteResult:
href: str
etag: str | None = None
status: int = 0
@dataclass(frozen=True, slots=True)
class _DiscoveryResponse:
href: str
display_name: str | None = None
description: str | None = None
ctag: str | None = None
sync_token: str | None = None
is_addressbook: bool = False
principal_hrefs: tuple[str, ...] = ()
addressbook_home_set_hrefs: tuple[str, ...] = ()
@dataclass(slots=True)
class _DiscoveryDraft:
display_name: str | None = None
description: str | None = None
ctag: str | None = None
sync_token: str | None = None
is_addressbook: bool = False
principal_hrefs: list[str] = field(default_factory=list)
addressbook_home_set_hrefs: list[str] = field(default_factory=list)
class AddressCardDAVClient:
def __init__(
self,
*,
collection_url: str,
username: str | None = None,
password: str | None = None,
bearer_token: str | None = None,
timeout: int = 30,
transport: AddressCardDAVTransport | 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) -> AddressCardDAVReportResult:
body = b"""
"""
payload = self.request("PROPFIND", self.collection_url, body=body, depth="0", expected={207})
return parse_multistatus(payload)
def discover_addressbooks(self) -> list[AddressCardDAVAddressBook]:
start_url = self.collection_url
addressbooks: dict[str, AddressCardDAVAddressBook] = {}
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_addressbook(base_url: str, response: _DiscoveryResponse) -> None:
if not response.is_addressbook:
return
url = ensure_collection_url(absolute_dav_url(base_url, response.href or base_url))
addressbooks[url] = AddressCardDAVAddressBook(
collection_url=url,
href=response.href,
display_name=response.display_name,
description=response.description,
ctag=response.ctag,
sync_token=response.sync_token,
)
def propfind(url: str, depth: str) -> list[_DiscoveryResponse]:
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_addressbook(start_url, response)
for href in response.addressbook_home_set_hrefs:
add_home_href(start_url, href)
for href in response.principal_hrefs:
add_principal_href(start_url, href)
except AddressCardDAVError 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.addressbook_home_set_hrefs:
add_home_href(principal_url, href)
except AddressCardDAVError 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_addressbook(home_url, response)
except AddressCardDAVError as exc:
errors.append(str(exc))
if not addressbooks and errors:
raise AddressCardDAVError(f"CardDAV discovery did not find any address books: {errors[0]}")
return sorted(addressbooks.values(), key=lambda item: ((item.display_name or item.collection_url).lower(), item.collection_url))
def propfind_discovery(self, url: str, *, depth: str) -> list[_DiscoveryResponse]:
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) -> AddressCardDAVReportResult:
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) -> AddressCardDAVReportResult:
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 AddressCardDAVError as exc:
raise AddressCardDAVSyncUnsupported(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, vcard: str, *, etag: str | None = None, create: bool = False, overwrite: bool = False) -> AddressCardDAVWriteResult:
headers = {"Content-Type": "text/vcard; charset=utf-8"}
if create:
headers["If-None-Match"] = "*"
elif etag and not overwrite:
headers["If-Match"] = etag
elif not overwrite:
raise AddressCardDAVPreconditionFailed(f"PUT {href} requires an ETag or explicit overwrite.")
status, response_headers, _payload = self.request_raw(
"PUT",
self.object_url(href),
body=vcard.encode("utf-8"),
depth=None,
expected={200, 201, 204},
extra_headers=headers,
)
return AddressCardDAVWriteResult(href=href, etag=response_etag(response_headers), status=status)
def delete_object(self, href: str, *, etag: str | None = None, overwrite: bool = False) -> AddressCardDAVWriteResult:
headers: dict[str, str] = {}
if etag and not overwrite:
headers["If-Match"] = etag
elif not overwrite:
raise AddressCardDAVPreconditionFailed(f"DELETE {href} requires an ETag or explicit overwrite.")
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 AddressCardDAVWriteResult(href=href, etag=response_etag(response_headers), status=status)
def object_url(self, href: str) -> str:
candidate = same_origin_dav_url(
self.collection_url,
href,
label="CardDAV object href",
)
collection_parts = urllib.parse.urlparse(self.collection_url)
candidate_parts = urllib.parse.urlparse(candidate)
collection_path = posixpath.normpath(urllib.parse.unquote(collection_parts.path))
candidate_path = posixpath.normpath(urllib.parse.unquote(candidate_parts.path))
collection_prefix = collection_path.rstrip("/") + "/"
if not candidate_path.startswith(collection_prefix) or candidate_path == collection_path:
raise AddressCardDAVError("CardDAV object href must remain inside the configured collection path")
return candidate
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/vcard,*/*",
"User-Agent": "govoplan-addresses-carddav/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 == 412:
raise AddressCardDAVPreconditionFailed(f"{method} {url} returned HTTP {status}")
if status == 404:
raise AddressCardDAVNotFound(f"{method} {url} returned HTTP {status}")
raise AddressCardDAVError(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]:
url = validate_http_url(url)
try:
request = urllib.request.Request( # noqa: S310 - URL is validated and origin-confined.
url,
data=body,
headers=dict(headers),
method=method,
)
opener = urllib.request.build_opener(_SameOriginRedirectHandler(url))
with opener.open(request, timeout=timeout) as response: # noqa: S310 - validated CardDAV URL; redirects remain on origin. # nosec B310
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 AddressCardDAVError(f"{method} {url} failed: {exc.reason}") from exc
except ValueError as exc:
raise AddressCardDAVError(f"{method} {url} failed: {exc}") from exc
def parse_multistatus(payload: bytes) -> AddressCardDAVReportResult:
try:
root = SafeElementTree.fromstring(payload)
except SafeElementTree.ParseError as exc:
raise AddressCardDAVError(f"Invalid CardDAV XML response: {exc}") from exc
objects: list[AddressCardDAVObject] = []
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
address_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
address_data = first_child_text(prop, "address-data") or address_data
sync_token = first_child_text(prop, "sync-token") or sync_token
ctag = first_child_text(prop, "getctag") or ctag
objects.append(AddressCardDAVObject(href=href, etag=strip_weak_etag(etag), address_data=address_data, deleted=deleted))
return AddressCardDAVReportResult(objects=objects, sync_token=sync_token, ctag=ctag)
def parse_discovery_multistatus(payload: bytes) -> list[_DiscoveryResponse]:
try:
root = SafeElementTree.fromstring(payload)
except SafeElementTree.ParseError as exc:
raise AddressCardDAVError(f"Invalid CardDAV XML response: {exc}") from exc
return [parsed for response in child_elements(root, "response") if (parsed := _parse_discovery_response(response)) is not None]
def _parse_discovery_response(response: Any) -> _DiscoveryResponse | None:
href = first_child_text(response, "href")
if not href:
return None
draft = _DiscoveryDraft()
for propstat in child_elements(response, "propstat"):
_apply_discovery_propstat(draft, propstat)
return _DiscoveryResponse(
href=href,
display_name=draft.display_name,
description=draft.description,
ctag=draft.ctag,
sync_token=draft.sync_token,
is_addressbook=draft.is_addressbook,
principal_hrefs=dedupe_tuple(draft.principal_hrefs),
addressbook_home_set_hrefs=dedupe_tuple(draft.addressbook_home_set_hrefs),
)
def _apply_discovery_propstat(draft: _DiscoveryDraft, 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: _DiscoveryDraft, 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
elif name == "addressbook-description" and text:
draft.description = text
elif name == "getctag" and text:
draft.ctag = text
elif name == "sync-token" and text:
draft.sync_token = text
elif name == "resourcetype":
draft.is_addressbook = draft.is_addressbook or any(local_name(child.tag) == "addressbook" for child in item)
elif name in {"current-user-principal", "principal-URL"}:
draft.principal_hrefs.extend(nested_href_texts(item))
elif name == "addressbook-home-set":
draft.addressbook_home_set_hrefs.extend(nested_href_texts(item))
def dedupe_tuple(values: list[str]) -> tuple[str, ...]:
return tuple(dict.fromkeys(values))
def ensure_collection_url(value: str) -> str:
value = value.strip()
if value and "://" not in value and not value.startswith("/"):
value = f"https://{value}"
url = validate_http_url(value)
return url if url.endswith("/") else f"{url}/"
def validate_http_url(value: str) -> str:
parsed = urllib.parse.urlparse(value.strip())
if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc or not parsed.hostname:
raise AddressCardDAVError("CardDAV URL must be an absolute HTTP(S) URL")
if parsed.username or parsed.password:
raise AddressCardDAVError("CardDAV URL must not include embedded credentials")
if parsed.query or parsed.fragment:
raise AddressCardDAVError("CardDAV URL must not include a query or fragment")
_url_origin(parsed)
return urllib.parse.urlunparse(parsed)
def absolute_dav_url(base_url: str, href: str) -> str:
return same_origin_dav_url(base_url, href, label="CardDAV discovery href")
def same_origin_dav_url(base_url: str, href: str, *, label: str) -> str:
base = ensure_collection_url(base_url)
candidate = validate_http_url(urllib.parse.urljoin(base, href))
if _url_origin(urllib.parse.urlparse(candidate)) != _url_origin(urllib.parse.urlparse(base)):
raise AddressCardDAVError(f"{label} must use the configured collection origin")
return candidate
def _url_origin(parsed: urllib.parse.ParseResult) -> tuple[str, str, int]:
try:
port = parsed.port
except ValueError as exc:
raise AddressCardDAVError("CardDAV URL has an invalid port") from exc
scheme = parsed.scheme.lower()
if port is None:
port = 443 if scheme == "https" else 80
return scheme, (parsed.hostname or "").lower(), port
class _SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler):
def __init__(self, source_url: str) -> None:
super().__init__()
self._source_origin = _url_origin(urllib.parse.urlparse(validate_http_url(source_url)))
def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def]
del fp, msg, headers
try:
candidate = validate_http_url(newurl)
except AddressCardDAVError:
return None
if _url_origin(urllib.parse.urlparse(candidate)) != self._source_origin:
return None
method = req.get_method()
data = req.data
if code == 303 and method != "HEAD":
method, data = "GET", None
elif code in {301, 302} and method == "POST":
method, data = "GET", None
forwarded_headers = {
key: value
for key, value in req.header_items()
if key.casefold() not in {"host", "content-length"}
}
return urllib.request.Request( # noqa: S310 - candidate is validated and same-origin.
candidate,
data=data,
headers=forwarded_headers,
origin_req_host=req.origin_req_host,
unverifiable=True,
method=method,
)
def strip_weak_etag(value: str | None) -> str | None:
return value.strip() if value else None
def response_etag(headers: Mapping[str, str]) -> str | None:
for key, value in headers.items():
if key.lower() == "etag":
return strip_weak_etag(value)
return None
def xml_escape(value: str) -> str:
return value.replace("&", "&").replace("<", "<").replace(">", ">")
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: 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: Any, name: str) -> list[Any]:
return [child for child in element if local_name(child.tag) == name]
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():
hrefs.append(child.text.strip())
return hrefs
def local_name(tag: str) -> str:
return tag.rsplit("}", 1)[-1] if "}" in tag else tag