govoplan-addresses/tests/test_carddav_security.py

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_transport_revalidates_dns_at_connection_time(self) -> None:
        public = [(2, 1, 6, "", ("93.184.216.34", 443))]
        private = [(2, 1, 6, "", ("127.0.0.1", 443))]
        with patch.dict(
            "os.environ",
            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
        ), patch(
            "govoplan_core.security.outbound_http.socket.getaddrinfo",
            side_effect=(public, private),
        ), patch("govoplan_core.security.outbound_http.socket.socket") as socket_factory, self.assertRaisesRegex(
            AddressCardDAVError,
            "non-public network",
        ):
            urllib_transport("GET", "https://dav.example.test/contact.vcf", {}, None, 2)
        socket_factory.assert_not_called()

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