From 1df2a1a730772d89b8634e5db6d29dcd6f9da5e9 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 12:11:58 +0200 Subject: [PATCH] Harden Calendar connector trust boundaries --- README.md | 10 +- pyproject.toml | 2 +- src/govoplan_calendar/backend/caldav.py | 28 ++- src/govoplan_calendar/backend/service.py | 206 +++++++++++++++++----- tests/test_caldav.py | 209 ++++++++++++++++++++++- tests/test_caldav_security.py | 17 ++ 6 files changed, 415 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index e811936..e11ff9d 100644 --- a/README.md +++ b/README.md @@ -40,9 +40,13 @@ The first backend implementation stores VEVENT data in two layers: CalDAV sync is implemented as a calendar-owned backend primitive. A CalDAV source records the remote collection URL, sync token, ETag/ctag state, username, credential reference, sync interval, sync direction, and conflict policy. -Credentials can be supplied transiently for manual sync, referenced from -environment variables with `env:NAME`, stored through a platform secret provider -when one is available, or stored as encrypted calendar-owned credentials. +Credentials can be supplied transiently for manual sync. Persisted API-managed +sources accept only a password or token: Calendar stores an opaque, tenant- and +source-bound local reference, backed by the platform secret provider when one is +available or by an encrypted calendar-owned credential otherwise. Caller-selected +environment and external-provider references are rejected. Trusted deployment +code may resolve an `env:NAME` reference only through the separate deployment +configuration helper. Inbound sync uses CalDAV `calendar-query` for full sync and `sync-collection` when a sync token exists. It imports all VEVENT components in a resource and diff --git a/pyproject.toml b/pyproject.toml index 2932bf9..a870e4d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ requires-python = ">=3.12" license = { file = "LICENSE" } authors = [{ name = "GovOPlaN" }] dependencies = [ - "govoplan-core>=0.1.8", + "govoplan-core>=0.1.9", "govoplan-access>=0.1.8", "defusedxml>=0.7,<1", "icalendar>=7.2", diff --git a/src/govoplan_calendar/backend/caldav.py b/src/govoplan_calendar/backend/caldav.py index 1c31551..6af20dd 100644 --- a/src/govoplan_calendar/backend/caldav.py +++ b/src/govoplan_calendar/backend/caldav.py @@ -9,6 +9,12 @@ from dataclasses import dataclass, field from typing import Any, Mapping, Protocol from defusedxml import ElementTree as SafeElementTree +from govoplan_core.security.outbound_http import ( + OutboundHttpError, + bounded_response_bytes, + build_outbound_http_opener, + validate_outbound_http_url, +) class CalDAVError(RuntimeError): @@ -353,20 +359,31 @@ class CalDAVClient: 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: + url = validate_outbound_http_url(url, label="CalDAV URL") 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)) + opener = build_outbound_http_opener(_SameOriginRedirectHandler(url)) with opener.open(request, timeout=timeout) as response: # noqa: S310 - validated CalDAV URL; redirects remain on origin. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected - return response.status, dict(response.headers.items()), response.read() + response_headers = dict(response.headers.items()) + return response.status, response_headers, bounded_response_bytes( + response, + headers=response_headers, + label="CalDAV response", + ) except urllib.error.HTTPError as exc: - return exc.code, dict(exc.headers.items()), exc.read() + response_headers = dict(exc.headers.items()) + try: + payload = bounded_response_bytes(exc, headers=response_headers, label="CalDAV error response") + except OutboundHttpError as policy_exc: + raise CalDAVError(f"{method} {url} failed: {policy_exc}") from policy_exc + return exc.code, response_headers, payload except urllib.error.URLError as exc: raise CalDAVError(f"{method} {url} failed: {exc.reason}") from exc - except ValueError as exc: + except (OutboundHttpError, ValueError) as exc: raise CalDAVError(f"{method} {url} failed: {exc}") from exc @@ -542,7 +559,8 @@ class _SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler): del fp, msg, headers try: candidate = validate_http_url(newurl) - except CalDAVError: + candidate = validate_outbound_http_url(candidate, label="CalDAV redirect URL") + except (CalDAVError, OutboundHttpError): return None if _url_origin(urllib.parse.urlparse(candidate)) != self._source_origin: return None diff --git a/src/govoplan_calendar/backend/service.py b/src/govoplan_calendar/backend/service.py index 8b66903..43f5615 100644 --- a/src/govoplan_calendar/backend/service.py +++ b/src/govoplan_calendar/backend/service.py @@ -17,6 +17,13 @@ from defusedxml import ElementTree as SafeElementTree from sqlalchemy import or_ from sqlalchemy.orm import Session +from govoplan_core.security.outbound_http import ( + OutboundHttpError, + bounded_response_bytes, + build_outbound_http_opener, + validate_outbound_http_url, +) + from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVReportResult, CalDAVSyncUnsupported, ensure_collection_url from govoplan_calendar.backend.db.models import CalendarCollection, CalendarEvent, CalendarSyncCredential, CalendarSyncSource from govoplan_calendar.backend.ical import expand_event_occurrences, parse_vevent, parse_vevents @@ -198,11 +205,12 @@ class _SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[no-untyped-def] try: candidate = validate_http_url(newurl, label="Calendar source redirect URL") + candidate = validate_outbound_http_url(candidate, label="Calendar source redirect URL") candidate_origin = _parsed_http_origin( urllib.parse.urlparse(candidate), label="Calendar source redirect URL", ) - except CalendarError: + except (CalendarError, OutboundHttpError): return None if candidate_origin != self._source_origin: return None @@ -696,10 +704,13 @@ def discover_caldav_calendars(session: Session, *, tenant_id: str, payload: Cale auth_type = payload.auth_type or (source.auth_type if source else "none") username = payload.username if payload.username is not None else (source.username if source else None) credential_ref = payload.credential_ref if payload.credential_ref is not None else (source.credential_ref if source else None) + if payload.credential_ref is not None and (source is None or payload.credential_ref != source.credential_ref): + raise CalendarError( + "Caller-supplied credential references are not accepted; provide a password/token " + "or select an existing sync source" + ) secret = caldav_secret_from_payload(auth_type=auth_type, password=payload.password, bearer_token=payload.bearer_token) - if secret is None and credential_ref: - secret = resolve_caldav_credential_ref(session, tenant_id=tenant_id, credential_ref=credential_ref) - if secret is None and source is not None: + if secret is None and source is not None and credential_ref == source.credential_ref: secret = resolve_caldav_secret(session, source=source) if auth_type == "basic": @@ -729,6 +740,10 @@ def discover_caldav_calendars(session: Session, *, tenant_id: str, payload: Cale def create_sync_source(session: Session, *, tenant_id: str, user_id: str | None, payload: CalendarSyncSourceCreateRequest) -> CalendarSyncSource: + if payload.credential_ref is not None: + raise CalendarError( + "Caller-supplied credential references are not accepted; provide a password or bearer token" + ) calendar = get_calendar(session, tenant_id=tenant_id, calendar_id=payload.calendar_id) calendar = ( session.query(CalendarCollection) @@ -812,6 +827,10 @@ def update_sync_source(session: Session, *, tenant_id: str, source_id: str, payl .with_for_update() .one() ) + if payload.credential_ref is not None and payload.credential_ref != source.credential_ref: + raise CalendarError( + "Caller-supplied credential references are not accepted; provide a replacement password or bearer token" + ) normalized_collection_url = ( normalize_sync_source_url(source.source_kind, payload.collection_url) if payload.collection_url is not None @@ -1064,7 +1083,12 @@ def retire_sync_source(session: Session, *, tenant_id: str, source: CalendarSync reason="CalDAV source was retired before queued external changes were delivered", ) source.deleted_at = deleted_at - delete_caldav_credential(session, tenant_id=tenant_id, credential_ref=source.credential_ref) + delete_caldav_credential( + session, + tenant_id=tenant_id, + source_id=source.id, + credential_ref=source.credential_ref, + ) def retire_caldav_source(session: Session, *, tenant_id: str, source: CalendarSyncSource, deleted_at: datetime) -> None: @@ -1095,12 +1119,12 @@ def store_caldav_credential( source: CalendarSyncSource, secret: str, ) -> str: - provider = secret_provider() - name = f"caldav:{source.id}:{source.auth_type}" - if provider is not None: - return str(provider.store_secret(scope=f"calendar:{tenant_id}", name=name, value=secret)) - - existing = internal_caldav_credential(session, tenant_id=tenant_id, credential_ref=source.credential_ref) + existing = internal_caldav_credential( + session, + tenant_id=tenant_id, + source_id=source.id, + credential_ref=source.credential_ref, + ) if existing is None: existing = CalendarSyncCredential( tenant_id=tenant_id, @@ -1111,25 +1135,47 @@ def store_caldav_credential( ) session.add(existing) session.flush() + + provider = secret_provider() + name = f"caldav:{source.id}:{source.auth_type}" + if provider is not None: + provider_ref = str(provider.store_secret(scope=f"calendar:{tenant_id}", name=name, value=secret)) + existing.secret_encrypted = None + existing.metadata_ = {"source_id": source.id, "provider_ref": provider_ref} + else: + existing.secret_encrypted = encrypt_secret(secret) + existing.metadata_ = {"source_id": source.id} existing.credential_kind = credential_kind_for_auth_type(source.auth_type) existing.label = source.display_name or source.collection_url - existing.secret_encrypted = encrypt_secret(secret) existing.deleted_at = None - existing.metadata_ = {"source_id": source.id} session.flush() return f"{CALDAV_INTERNAL_CREDENTIAL_PREFIX}{existing.id}" -def delete_caldav_credential(session: Session, *, tenant_id: str, credential_ref: str | None) -> None: +def delete_caldav_credential( + session: Session, + *, + tenant_id: str, + source_id: str, + credential_ref: str | None, +) -> None: if not credential_ref: return - provider = secret_provider() - if provider is not None and not credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX) and not credential_ref.startswith(CALDAV_ENV_CREDENTIAL_PREFIX): - provider.delete_secret(credential_ref) + credential = internal_caldav_credential( + session, + tenant_id=tenant_id, + source_id=source_id, + credential_ref=credential_ref, + ) + if credential is None: + # Legacy external/env references have no locally provable ownership and + # must never be passed to a provider delete operation. return - credential = internal_caldav_credential(session, tenant_id=tenant_id, credential_ref=credential_ref) - if credential is not None: - credential.deleted_at = utcnow() + provider_ref = _credential_provider_ref(credential) + provider = secret_provider() + if provider is not None and provider_ref: + provider.delete_secret(provider_ref) + credential.deleted_at = utcnow() def resolve_caldav_secret( @@ -1147,27 +1193,59 @@ def resolve_caldav_secret( return bearer_token if not source.credential_ref: return None - if source.credential_ref.startswith(CALDAV_ENV_CREDENTIAL_PREFIX): - return os.environ.get(source.credential_ref.removeprefix(CALDAV_ENV_CREDENTIAL_PREFIX)) - provider = secret_provider() - if provider is not None and not source.credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX): - secret = provider.read_secret(source.credential_ref) - if secret: - return secret - credential = internal_caldav_credential(session, tenant_id=source.tenant_id, credential_ref=source.credential_ref) - return decrypt_secret(credential.secret_encrypted) if credential and credential.secret_encrypted else None + credential = internal_caldav_credential( + session, + tenant_id=source.tenant_id, + source_id=source.id, + credential_ref=source.credential_ref, + ) + if credential is None: + raise CalendarError( + "The sync source credential reference is not a server-owned credential for this tenant/source" + ) + provider_ref = _credential_provider_ref(credential) + if provider_ref: + provider = secret_provider() + if provider is None: + return None + return provider.read_secret(provider_ref) + return decrypt_secret(credential.secret_encrypted) if credential.secret_encrypted else None -def resolve_caldav_credential_ref(session: Session, *, tenant_id: str, credential_ref: str) -> str | None: - if credential_ref.startswith(CALDAV_ENV_CREDENTIAL_PREFIX): - return os.environ.get(credential_ref.removeprefix(CALDAV_ENV_CREDENTIAL_PREFIX)) - provider = secret_provider() - if provider is not None and not credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX): - secret = provider.read_secret(credential_ref) - if secret: - return secret - credential = internal_caldav_credential(session, tenant_id=tenant_id, credential_ref=credential_ref) - return decrypt_secret(credential.secret_encrypted) if credential and credential.secret_encrypted else None +def resolve_caldav_credential_ref( + session: Session, + *, + tenant_id: str, + source_id: str, + credential_ref: str, +) -> str | None: + credential = internal_caldav_credential( + session, + tenant_id=tenant_id, + source_id=source_id, + credential_ref=credential_ref, + ) + if credential is None: + return None + provider_ref = _credential_provider_ref(credential) + if provider_ref: + provider = secret_provider() + return provider.read_secret(provider_ref) if provider is not None else None + return decrypt_secret(credential.secret_encrypted) if credential.secret_encrypted else None + + +def resolve_trusted_deployment_caldav_credential_ref(credential_ref: str) -> str | None: + """Resolve an env reference only for trusted, out-of-band deployment code. + + HTTP/API source configuration deliberately never calls this function. + """ + + if not credential_ref.startswith(CALDAV_ENV_CREDENTIAL_PREFIX): + raise CalendarError("Trusted deployment credential references must use the env: prefix") + env_name = credential_ref.removeprefix(CALDAV_ENV_CREDENTIAL_PREFIX) + if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", env_name): + raise CalendarError("Trusted deployment credential reference contains an invalid environment variable name") + return os.environ.get(env_name) def caldav_client_for_source( @@ -1242,22 +1320,37 @@ def http_request( url, label="Calendar source request URL", ) + try: + url = validate_outbound_http_url(url, label="Calendar source request URL") + except OutboundHttpError as exc: + raise CalendarError(str(exc)) from exc request = urllib.request.Request( # noqa: S310 - URL is validated and origin-confined. url, data=data, method=method, headers=headers or {}, ) - opener = urllib.request.build_opener(_SameOriginRedirectHandler(credential_origin)) + opener = build_outbound_http_opener(_SameOriginRedirectHandler(credential_origin)) try: with opener.open(request, timeout=timeout) as response: # noqa: S310 - validated, origin-confined Calendar URL. # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected response_headers = {key.lower(): value for key, value in response.headers.items()} - payload = response.read().decode(response_headers.get("content-charset") or "utf-8", errors="replace") + payload = bounded_response_bytes( + response, + headers=response_headers, + label="Calendar source response", + ).decode(response_headers.get("content-charset") or "utf-8", errors="replace") return int(response.status), response_headers, payload except urllib.error.HTTPError as exc: if exc.code == 304: return 304, {key.lower(): value for key, value in exc.headers.items()}, "" - detail = exc.read().decode("utf-8", errors="replace") + try: + detail = bounded_response_bytes( + exc, + headers={key.lower(): value for key, value in exc.headers.items()}, + label="Calendar source error response", + ).decode("utf-8", errors="replace") + except OutboundHttpError as policy_exc: + raise CalendarError(str(policy_exc)) from policy_exc raise CalendarError(f"HTTP {exc.code} from calendar source: {detail or exc.reason}") from exc except urllib.error.URLError as exc: raise CalendarError(f"Calendar source request failed: {exc.reason}") from exc @@ -1287,11 +1380,17 @@ def source_auth_headers( return {} -def internal_caldav_credential(session: Session, *, tenant_id: str, credential_ref: str | None) -> CalendarSyncCredential | None: - if not credential_ref: +def internal_caldav_credential( + session: Session, + *, + tenant_id: str, + credential_ref: str | None, + source_id: str | None = None, +) -> CalendarSyncCredential | None: + if not credential_ref or not credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX): return None credential_id = credential_ref.removeprefix(CALDAV_INTERNAL_CREDENTIAL_PREFIX) - return ( + credential = ( session.query(CalendarSyncCredential) .filter( CalendarSyncCredential.tenant_id == tenant_id, @@ -1300,6 +1399,16 @@ def internal_caldav_credential(session: Session, *, tenant_id: str, credential_r ) .first() ) + if credential is None or source_id is None: + return credential + metadata = credential.metadata_ if isinstance(credential.metadata_, dict) else {} + return credential if metadata.get("source_id") == source_id else None + + +def _credential_provider_ref(credential: CalendarSyncCredential) -> str | None: + metadata = credential.metadata_ if isinstance(credential.metadata_, dict) else {} + value = metadata.get("provider_ref") + return str(value) if isinstance(value, str) and value.strip() else None def credential_kind_for_auth_type(auth_type: str) -> str: @@ -2390,6 +2499,9 @@ def mark_calendar_sync_source(calendar: CalendarCollection, source: CalendarSync def caldav_source_response(source: CalendarSyncSource) -> dict[str, Any]: + has_server_owned_credential = bool( + source.credential_ref and source.credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX) + ) return { "id": source.id, "tenant_id": source.tenant_id, @@ -2399,8 +2511,8 @@ def caldav_source_response(source: CalendarSyncSource) -> dict[str, Any]: "display_name": source.display_name, "auth_type": source.auth_type, "username": source.username, - "credential_ref": source.credential_ref, - "has_credential": bool(source.credential_ref), + "credential_ref": None, + "has_credential": has_server_owned_credential, "sync_enabled": bool(source.sync_enabled), "sync_interval_seconds": int(source.sync_interval_seconds or CALDAV_DEFAULT_SYNC_INTERVAL_SECONDS), "sync_direction": source.sync_direction, diff --git a/tests/test_caldav.py b/tests/test_caldav.py index ffd1e01..d8bbcdf 100644 --- a/tests/test_caldav.py +++ b/tests/test_caldav.py @@ -11,20 +11,32 @@ from govoplan_access.backend.db import models as access_models # noqa: F401 - p from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVObject, CalDAVPreconditionFailed, CalDAVReportResult, CalDAVWriteResult, parse_multistatus from govoplan_calendar.backend.db.models import CalendarEvent, CalendarOutboxOperation, CalendarSyncCredential from govoplan_calendar.backend.outbox import dispatch_calendar_outbox -from govoplan_calendar.backend.schemas import CalendarCalDavSourceCreateRequest, CalendarCollectionCreateRequest, CalendarEventCreateRequest, CalendarEventUpdateRequest +from govoplan_calendar.backend.schemas import ( + CalendarCalDavDiscoveryRequest, + CalendarCalDavSourceCreateRequest, + CalendarCalDavSourceUpdateRequest, + CalendarCollectionCreateRequest, + CalendarEventCreateRequest, + CalendarEventUpdateRequest, +) from govoplan_calendar.backend.service import ( CALDAV_INTERNAL_CREDENTIAL_PREFIX, CalendarError, caldav_client_for_source, + caldav_source_response, create_calendar, create_caldav_source, create_event, delete_calendar, delete_event, + discover_caldav_calendars, list_caldav_sources, list_freebusy, + resolve_caldav_credential_ref, + resolve_trusted_deployment_caldav_credential_ref, sync_caldav_source, sync_due_caldav_sources, + update_caldav_source, update_event, ) from govoplan_core.db.base import Base @@ -286,6 +298,201 @@ class CalDAVSyncTests(unittest.TestCase): self.assertEqual(client.username, "ada") self.assertEqual(client.password, "secret") + def test_api_source_and_discovery_reject_caller_selected_credential_references(self) -> None: + session = self.session() + session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) + calendar = create_calendar( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCollectionCreateRequest(name="Remote"), + ) + + with self.assertRaisesRegex(CalendarError, "Caller-supplied credential references"): + create_caldav_source( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCalDavSourceCreateRequest( + calendar_id=calendar.id, + collection_url="https://attacker.example.test/cal", + auth_type="bearer", + credential_ref="env:MASTER_KEY_B64", + ), + ) + with self.assertRaisesRegex(CalendarError, "Caller-supplied credential references"): + discover_caldav_calendars( + session, + tenant_id="tenant-1", + payload=CalendarCalDavDiscoveryRequest( + url="https://attacker.example.test/cal", + auth_type="bearer", + credential_ref="vault:another-tenant", + ), + ) + source = create_caldav_source( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCalDavSourceCreateRequest( + calendar_id=calendar.id, + collection_url="https://dav.example.test/cal", + ), + ) + with self.assertRaisesRegex(CalendarError, "Caller-supplied credential references"): + update_caldav_source( + session, + tenant_id="tenant-1", + source_id=source.id, + payload=CalendarCalDavSourceUpdateRequest( + credential_ref="env:MASTER_KEY_B64", + ), + ) + + def test_provider_credentials_are_wrapped_in_tenant_and_source_owned_rows(self) -> None: + class Provider: + def __init__(self) -> None: + self.values: dict[str, str] = {} + self.deleted: list[str] = [] + + def store_secret(self, *, scope: str, name: str, value: str) -> str: + reference = f"vault:{scope}:{name}" + self.values[reference] = value + return reference + + def read_secret(self, secret_ref: str) -> str | None: + return self.values.get(secret_ref) + + def delete_secret(self, secret_ref: str) -> None: + self.deleted.append(secret_ref) + + provider = Provider() + session = self.session() + session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) + calendar = create_calendar( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCollectionCreateRequest(name="Remote"), + ) + with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): + source = create_caldav_source( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCalDavSourceCreateRequest( + calendar_id=calendar.id, + collection_url="https://dav.example.test/cal", + auth_type="basic", + username="ada", + password="secret", + ), + ) + session.commit() + credential = session.query(CalendarSyncCredential).one() + provider_ref = str((credential.metadata_ or {})["provider_ref"]) + + self.assertTrue(source.credential_ref.startswith(CALDAV_INTERNAL_CREDENTIAL_PREFIX)) + self.assertNotEqual(source.credential_ref, provider_ref) + self.assertEqual(caldav_client_for_source(session, source).password, "secret") + self.assertIsNone(caldav_source_response(source)["credential_ref"]) + self.assertTrue(caldav_source_response(source)["has_credential"]) + self.assertIsNone( + resolve_caldav_credential_ref( + session, + tenant_id="another-tenant", + source_id=source.id, + credential_ref=source.credential_ref, + ) + ) + self.assertIsNone( + resolve_caldav_credential_ref( + session, + tenant_id="tenant-1", + source_id="another-source", + credential_ref=source.credential_ref, + ) + ) + + other_calendar = create_calendar( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCollectionCreateRequest(name="Other remote"), + ) + other_source = create_caldav_source( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCalDavSourceCreateRequest( + calendar_id=other_calendar.id, + collection_url="https://dav.example.test/other-cal", + ), + ) + other_source.auth_type = "basic" + other_source.username = "mallory" + other_source.credential_ref = source.credential_ref + with self.assertRaisesRegex(CalendarError, "not a server-owned credential"): + caldav_client_for_source(session, other_source) + delete_calendar(session, tenant_id="tenant-1", calendar_id=other_calendar.id) + self.assertEqual(provider.deleted, []) + + delete_calendar(session, tenant_id="tenant-1", calendar_id=calendar.id) + + self.assertEqual(provider.deleted, [provider_ref]) + + def test_legacy_unowned_refs_are_neither_read_nor_deleted(self) -> None: + class Provider: + def __init__(self) -> None: + self.read: list[str] = [] + self.deleted: list[str] = [] + + def read_secret(self, secret_ref: str) -> str | None: + self.read.append(secret_ref) + return "other-tenant-secret" + + def delete_secret(self, secret_ref: str) -> None: + self.deleted.append(secret_ref) + + provider = Provider() + session = self.session() + session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) + calendar = create_calendar( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCollectionCreateRequest(name="Remote"), + ) + source = create_caldav_source( + session, + tenant_id="tenant-1", + user_id=None, + payload=CalendarCalDavSourceCreateRequest( + calendar_id=calendar.id, + collection_url="https://dav.example.test/cal", + ), + ) + source.auth_type = "bearer" + source.credential_ref = "vault:another-tenant" + session.commit() + + with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider): + with self.assertRaisesRegex(CalendarError, "not a server-owned credential"): + caldav_client_for_source(session, source) + delete_calendar(session, tenant_id="tenant-1", calendar_id=calendar.id) + + self.assertEqual(provider.read, []) + self.assertEqual(provider.deleted, []) + + def test_trusted_deployment_env_resolution_is_explicit_and_separate(self) -> None: + with patch.dict("os.environ", {"CALDAV_DEPLOYMENT_TOKEN": "trusted-token"}): + self.assertEqual( + resolve_trusted_deployment_caldav_credential_ref("env:CALDAV_DEPLOYMENT_TOKEN"), + "trusted-token", + ) + with self.assertRaisesRegex(CalendarError, "must use the env: prefix"): + resolve_trusted_deployment_caldav_credential_ref("vault:token") + def test_delete_calendar_retires_caldav_source_and_credential(self) -> None: session = self.session() session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant")) diff --git a/tests/test_caldav_security.py b/tests/test_caldav_security.py index 993c545..75d2e7c 100644 --- a/tests/test_caldav_security.py +++ b/tests/test_caldav_security.py @@ -5,6 +5,7 @@ import threading import unittest from collections.abc import Iterator from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from unittest.mock import patch from govoplan_calendar.backend.caldav import ( CalDAVClient, @@ -29,6 +30,22 @@ def running_http_server(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]: class CalDAVUrlSecurityTests(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( + CalDAVError, + "non-public network", + ): + urllib_transport("GET", "https://dav.example.test/event.ics", {}, None, 2) + socket_factory.assert_not_called() + def test_discovery_href_must_remain_on_configured_origin(self) -> None: base_url = "https://dav.example.test/calendars/ada/"