From 22d12d674b4581c13c23460e547eb95ec527d821 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 03:16:38 +0200 Subject: [PATCH] Confine CardDAV requests to configured origins --- src/govoplan_addresses/backend/carddav.py | 87 +++++++++++- src/govoplan_addresses/backend/router.py | 2 +- tests/test_carddav_security.py | 166 ++++++++++++++++++++++ 3 files changed, 248 insertions(+), 7 deletions(-) create mode 100644 tests/test_carddav_security.py diff --git a/src/govoplan_addresses/backend/carddav.py b/src/govoplan_addresses/backend/carddav.py index 30277d9..adf697d 100644 --- a/src/govoplan_addresses/backend/carddav.py +++ b/src/govoplan_addresses/backend/carddav.py @@ -1,6 +1,7 @@ from __future__ import annotations import base64 +import posixpath import urllib.error import urllib.parse import urllib.request @@ -269,7 +270,19 @@ class AddressCardDAVClient: return AddressCardDAVWriteResult(href=href, etag=response_etag(response_headers), status=status) def object_url(self, href: str) -> str: - return urllib.parse.urljoin(self.collection_url, href) + 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) @@ -313,8 +326,14 @@ class AddressCardDAVClient: 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(url, data=body, headers=dict(headers), method=method) - with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - validated CardDAV HTTP(S) URL. # nosec B310 + 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() @@ -431,16 +450,72 @@ def ensure_collection_url(value: str) -> str: def validate_http_url(value: str) -> str: - parsed = urllib.parse.urlparse(value) - if parsed.scheme not in {"http", "https"} or not parsed.netloc: + 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 urllib.parse.urljoin(ensure_collection_url(base_url), href) + 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: diff --git a/src/govoplan_addresses/backend/router.py b/src/govoplan_addresses/backend/router.py index c5fa203..de689d4 100644 --- a/src/govoplan_addresses/backend/router.py +++ b/src/govoplan_addresses/backend/router.py @@ -618,7 +618,7 @@ def api_discover_carddav_address_books( principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): - _require_scope(principal, "addresses:sync:read") + _require_scope(principal, "addresses:sync:write") try: address_books = discover_carddav_address_books(session, principal, payload) return AddressCardDavDiscoveryResponse( diff --git a/tests/test_carddav_security.py b/tests/test_carddav_security.py new file mode 100644 index 0000000..b2e9e88 --- /dev/null +++ b/tests/test_carddav_security.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import contextlib +import threading +import unittest +from collections.abc import Iterator +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import patch + +from fastapi import HTTPException + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef + +from govoplan_addresses.backend.carddav import ( + AddressCardDAVClient, + AddressCardDAVError, + absolute_dav_url, + urllib_transport, +) +from govoplan_addresses.backend.router import api_discover_carddav_address_books +from govoplan_addresses.backend.schemas import AddressCardDavDiscoveryRequest + + +@contextlib.contextmanager +def running_http_server(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]: + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + yield f"http://{host}:{port}" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=2) + + +class CardDAVUrlSecurityTests(unittest.TestCase): + def test_discovery_href_must_remain_on_configured_origin(self) -> None: + base_url = "https://dav.example.test/addressbooks/ada/" + + self.assertEqual( + absolute_dav_url(base_url, "/principals/users/ada/"), + "https://dav.example.test/principals/users/ada/", + ) + with self.assertRaisesRegex(AddressCardDAVError, "configured collection origin"): + absolute_dav_url(base_url, "https://evil.example.test/steal/") + with self.assertRaisesRegex(AddressCardDAVError, "query or fragment"): + absolute_dav_url(base_url, "/principals/users/ada/?token=secret") + + def test_object_href_must_remain_inside_configured_collection(self) -> None: + client = AddressCardDAVClient(collection_url="https://dav.example.test/addressbooks/ada") + + with self.assertRaisesRegex(AddressCardDAVError, "collection origin"): + client.object_url("https://evil.example.test/steal.vcf") + with self.assertRaisesRegex(AddressCardDAVError, "collection path"): + client.object_url("https://dav.example.test/addressbooks/other/steal.vcf") + with self.assertRaisesRegex(AddressCardDAVError, "collection path"): + client.object_url("/addressbooks/ada/%2e%2e/other/steal.vcf") + with self.assertRaisesRegex(AddressCardDAVError, "query or fragment"): + client.object_url("/addressbooks/ada/contact.vcf?download=1") + self.assertEqual( + client.object_url("/addressbooks/ada/contact.vcf"), + "https://dav.example.test/addressbooks/ada/contact.vcf", + ) + + def test_transport_refuses_redirect_before_forwarding_authorization(self) -> None: + forwarded_authorization: list[str | None] = [] + + class TargetHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + forwarded_authorization.append(self.headers.get("Authorization")) + self.send_response(200) + self.end_headers() + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(TargetHandler) as target_url: + class RedirectHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + self.send_response(302) + self.send_header("Location", f"{target_url}/stolen.vcf") + self.end_headers() + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(RedirectHandler) as redirect_url: + status, _headers, _body = urllib_transport( + "GET", + f"{redirect_url}/contact.vcf", + {"Authorization": "Bearer top-secret"}, + None, + 2, + ) + + self.assertEqual(status, 302) + self.assertEqual(forwarded_authorization, []) + + def test_transport_preserves_same_origin_redirects(self) -> None: + forwarded_authorization: list[str | None] = [] + + class RedirectHandler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API + if self.path == "/contact.vcf": + self.send_response(302) + self.send_header("Location", "/redirected.vcf") + self.end_headers() + return + forwarded_authorization.append(self.headers.get("Authorization")) + self.send_response(200) + self.end_headers() + self.wfile.write(b"contact") + + def log_message(self, _format: str, *_args: object) -> None: + return + + with running_http_server(RedirectHandler) as source_url: + status, _headers, body = urllib_transport( + "GET", + f"{source_url}/contact.vcf", + {"Authorization": "Bearer expected"}, + None, + 2, + ) + + self.assertEqual(status, 200) + self.assertEqual(body, b"contact") + self.assertEqual(forwarded_authorization, ["Bearer expected"]) + + +class CardDAVDiscoveryAuthorizationTests(unittest.TestCase): + def test_sync_read_alone_cannot_start_authenticated_discovery(self) -> None: + principal = ApiPrincipal( + principal=PrincipalRef( + account_id="account-read-only", + membership_id="membership-read-only", + tenant_id="tenant-1", + scopes=frozenset({"addresses:sync:read"}), + ), + account=object(), + user=object(), + ) + + with patch("govoplan_addresses.backend.router.discover_carddav_address_books") as discover: + with self.assertRaises(HTTPException) as raised: + api_discover_carddav_address_books( + AddressCardDavDiscoveryRequest( + url="https://dav.example.test/", + auth_type="basic", + username="reader", + password="secret", + ), + principal, + object(), # type: ignore[arg-type] - scope rejection precedes session use + ) + + self.assertEqual(raised.exception.status_code, 403) + self.assertEqual(raised.exception.detail, "Missing scope: addresses:sync:write") + discover.assert_not_called() + + +if __name__ == "__main__": + unittest.main()