Confine CardDAV requests to configured origins

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)
+        request = urllib.request.Request(  # noqa: S310 - URL is validated and origin-confined.
-        with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310 - validated CardDAV HTTP(S) URL.  # nosec B310
+            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)
+    parsed = urllib.parse.urlparse(value.strip())
-    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+    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:

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(

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()