GovOPlaN/govoplan-calendar
1
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity

Harden Calendar connector trust boundaries

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 12:11:58 +02:00
parent a52e35b52b
commit 1df2a1a730
6 changed files with 415 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

209
tests/test_caldav.py
View File

@@ -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"))

17
tests/test_caldav_security.py
View File

@@ -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/"