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

Harden CardDAV credential and network boundaries

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 12:12:18 +02:00
parent b13e5760c8
commit 8b4cf362ca
6 changed files with 247 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@@ -11,7 +11,7 @@ requires-python = ">=3.12"
authors = [{ name = "GovOPlaN" }]
dependencies = [
"defusedxml>=0.7.1",
"govoplan-core>=0.1.8",
"govoplan-core>=0.1.9",
]
[tool.setuptools.packages.find]

28
src/govoplan_addresses/backend/carddav.py
View File

@@ -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 AddressCardDAVError(RuntimeError):
@@ -326,20 +332,31 @@ 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:
url = validate_outbound_http_url(url, label="CardDAV 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 CardDAV URL; redirects remain on origin. # nosec B310
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="CardDAV 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="CardDAV error response")
except OutboundHttpError as policy_exc:
raise AddressCardDAVError(f"{method} {url} failed: {policy_exc}") from policy_exc
return exc.code, response_headers, payload
except urllib.error.URLError as exc:
raise AddressCardDAVError(f"{method} {url} failed: {exc.reason}") from exc
except ValueError as exc:
except (OutboundHttpError, ValueError) as exc:
raise AddressCardDAVError(f"{method} {url} failed: {exc}") from exc
@@ -493,7 +510,8 @@ class _SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler):
del fp, msg, headers
try:
candidate = validate_http_url(newurl)
except AddressCardDAVError:
candidate = validate_outbound_http_url(candidate, label="CardDAV redirect URL")
except (AddressCardDAVError, OutboundHttpError):
return None
if _url_origin(urllib.parse.urlparse(candidate)) != self._source_origin:
return None

3
src/govoplan_addresses/backend/router.py
View File

@@ -103,6 +103,7 @@ from govoplan_addresses.backend.service import (
restore_contact,
resolve_sync_conflict,
preview_sync_source,
public_address_sync_metadata,
run_sync_source,
start_sync_attempt,
update_address_book,
@@ -217,7 +218,7 @@ def _sync_source_response(sync_source: AddressSyncSource) -> AddressSyncSourceRe
"last_success_at": sync_source.last_success_at,
"last_error": sync_source.last_error,
"last_diagnostic": sync_source.last_diagnostic,
"metadata": sync_source.metadata_ or {},
"metadata": public_address_sync_metadata(sync_source.metadata_),
"created_at": sync_source.created_at,
"updated_at": sync_source.updated_at,
}

92
src/govoplan_addresses/backend/service.py
View File

@@ -1,8 +1,10 @@
from __future__ import annotations
import copy
from collections.abc import Iterable
from dataclasses import dataclass, field
import os
import re
import urllib.parse
from typing import Any
@@ -256,6 +258,8 @@ def create_sync_source(
principal: ApiPrincipal,
address_book_id: str,
payload: AddressSyncSourceCreateRequest,
*,
trusted_connector_metadata: bool = False,
) -> AddressSyncSource:
book = get_visible_address_book(session, principal, address_book_id)
if book.deleted_at is not None:
@@ -266,6 +270,8 @@ def create_sync_source(
raise AddressBookError("Sync connector type is required.")
if not display_name:
raise AddressBookError("Sync source display name is required.")
if connector_type.casefold() == "carddav" and not trusted_connector_metadata:
_assert_api_carddav_metadata_safe(payload.metadata)
read_only = _read_only_from_sync_direction(payload.sync_direction, payload.read_only)
sync_source = AddressSyncSource(
tenant_id=book.tenant_id,
@@ -325,7 +331,11 @@ def update_sync_source(
if "remote_revision" in payload.model_fields_set:
sync_source.remote_revision = _trim(payload.remote_revision)
if "metadata" in payload.model_fields_set:
sync_source.metadata_ = payload.metadata or {}
metadata = payload.metadata or {}
if sync_source.connector_type.casefold() == "carddav":
_assert_api_carddav_metadata_safe(metadata)
metadata = _merge_server_owned_carddav_metadata(sync_source.metadata_, metadata)
sync_source.metadata_ = metadata
sync_source.updated_by_account_id = _account_id(principal)
_apply_sync_source_to_book(sync_source.address_book, sync_source)
return sync_source
@@ -588,6 +598,7 @@ def discover_carddav_address_books(
principal: ApiPrincipal,
payload: AddressCardDavDiscoveryRequest,
) -> list[AddressCardDAVAddressBook]:
_assert_no_caller_carddav_credential_ref(payload.credential_ref)
source = get_visible_sync_source(session, principal, payload.source_id) if payload.source_id else None
client = _carddav_client_from_payload(session, principal, payload, source=source)
return client.discover_addressbooks()
@@ -599,6 +610,7 @@ def create_carddav_sync_source(
address_book_id: str,
payload: AddressCardDavSourceCreateRequest,
) -> AddressSyncSource:
_assert_no_caller_carddav_credential_ref(payload.credential_ref)
collection_url = ensure_collection_url(payload.collection_url)
display_name = _trim(payload.display_name) or "CardDAV address book"
metadata = _carddav_metadata(
@@ -606,7 +618,7 @@ def create_carddav_sync_source(
username=payload.username,
password=_secret_value(payload.password),
bearer_token=_secret_value(payload.bearer_token),
credential_ref=payload.credential_ref,
credential_ref=None,
collection_url=collection_url,
)
return create_sync_source(
@@ -624,6 +636,7 @@ def create_carddav_sync_source(
remote_revision=payload.remote_revision,
metadata=metadata,
),
trusted_connector_metadata=True,
)
@@ -1308,6 +1321,7 @@ def _carddav_client_from_payload(
*,
source: AddressSyncSource | None = None,
) -> AddressCardDAVClient:
_assert_no_caller_carddav_credential_ref(payload.credential_ref)
url = payload.url or (source.external_address_book_ref if source else "")
metadata = dict(source.metadata_ or {}) if source else {}
auth = dict(metadata.get("carddav") or {})
@@ -1391,13 +1405,83 @@ def _resolve_carddav_secret(
return password
if auth_type == "bearer" and bearer_token:
return bearer_token
if credential_ref and credential_ref.startswith(CARDDAV_SECRET_ENV_PREFIX):
return os.environ.get(credential_ref.removeprefix(CARDDAV_SECRET_ENV_PREFIX))
if credential_ref:
raise AddressBookError(
"The CardDAV credential reference is not a server-owned credential; provide a replacement password or token"
)
if encrypted:
return decrypt_secret(encrypted)
return None
def resolve_trusted_deployment_carddav_credential_ref(credential_ref: str) -> str | None:
"""Resolve env-backed credentials only for trusted deployment code.
API-managed discovery and sync-source paths deliberately never call this
function, so a tenant user cannot select arbitrary process environment
variables as connector credentials.
"""
if not credential_ref.startswith(CARDDAV_SECRET_ENV_PREFIX):
raise AddressBookError("Trusted deployment credential references must use the env: prefix")
env_name = credential_ref.removeprefix(CARDDAV_SECRET_ENV_PREFIX)
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", env_name):
raise AddressBookError("Trusted deployment credential reference contains an invalid environment variable name")
return os.environ.get(env_name)
def public_address_sync_metadata(metadata: object) -> dict[str, Any]:
payload = copy.deepcopy(metadata) if isinstance(metadata, dict) else {}
auth = payload.get("carddav")
if not isinstance(auth, dict):
return payload
had_credential = bool(auth.get("secret_encrypted") or auth.get("credential_ref"))
auth.pop("secret_encrypted", None)
auth.pop("credential_ref", None)
auth["has_credential"] = had_credential
return payload
def _assert_no_caller_carddav_credential_ref(credential_ref: str | None) -> None:
if _trim(credential_ref):
raise AddressBookError(
"Caller-supplied credential references are not accepted; provide a password or bearer token"
)
def _carddav_auth_metadata(metadata: object) -> dict[str, Any]:
if not isinstance(metadata, dict):
return {}
carddav = metadata.get("carddav")
return carddav if isinstance(carddav, dict) else {}
def _assert_api_carddav_metadata_safe(metadata: object) -> None:
auth = _carddav_auth_metadata(metadata)
if auth.get("credential_ref") or auth.get("secret_encrypted"):
raise AddressBookError(
"CardDAV credential references and encrypted secrets are server-managed; provide credentials through the CardDAV source endpoint"
)
def _merge_server_owned_carddav_metadata(existing: object, incoming: dict[str, Any]) -> dict[str, Any]:
merged = copy.deepcopy(incoming)
existing_auth = _carddav_auth_metadata(existing)
server_owned = {
key: existing_auth[key]
for key in ("credential_ref", "secret_encrypted")
if existing_auth.get(key)
}
if not server_owned:
return merged
auth = merged.get("carddav")
if not isinstance(auth, dict):
auth = {}
merged["carddav"] = auth
auth.update(server_owned)
return merged
def _secret_value(value: Any | None) -> str | None:
if value is None:
return None

117
tests/test_addresses_service.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
import unittest
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
@@ -33,6 +34,7 @@ from govoplan_addresses.backend.db.models import (
)
from govoplan_addresses.backend.schemas import (
AddressBookCreateRequest,
AddressCardDavDiscoveryRequest,
AddressListCreateRequest,
AddressListEntryCreateRequest,
AddressCardDavSourceCreateRequest,
@@ -48,6 +50,7 @@ from govoplan_addresses.backend.schemas import (
ContactPostalAddressPayload,
)
from govoplan_addresses.backend.manifest import manifest
from govoplan_addresses.backend.router import _sync_source_response
from govoplan_addresses.backend.service import (
AddressBookError,
address_book_contact_counts,
@@ -61,6 +64,7 @@ from govoplan_addresses.backend.service import (
delete_address_list_entry,
delete_contact,
delete_sync_source,
discover_carddav_address_books,
export_address_book_vcard,
import_vcards,
list_address_list_entries,
@@ -81,6 +85,7 @@ from govoplan_addresses.backend.service import (
start_sync_attempt,
finish_sync_attempt,
update_sync_source,
resolve_trusted_deployment_carddav_credential_ref,
)
@@ -787,6 +792,118 @@ END:VCARD
self.assertEqual(contacts[0].display_name, "Ada Remote")
self.assertEqual(contacts[0].emails[0].email, "ada.remote@example.local")
def test_carddav_credentials_are_server_managed_and_public_metadata_is_redacted(self) -> None:
book = create_address_book(
self.session,
self.principal,
AddressBookCreateRequest(scope_type="user", name="Secured remote"),
)
self.session.commit()
self.session.refresh(book)
with self.assertRaisesRegex(AddressBookError, "Caller-supplied credential references"):
create_carddav_sync_source(
self.session,
self.principal,
book.id,
AddressCardDavSourceCreateRequest(
collection_url="https://attacker.example.test/addressbooks/personal/",
auth_type="bearer",
credential_ref="env:MASTER_KEY_B64",
),
)
with self.assertRaisesRegex(AddressBookError, "Caller-supplied credential references"):
discover_carddav_address_books(
self.session,
self.principal,
AddressCardDavDiscoveryRequest(
url="https://attacker.example.test/addressbooks/",
auth_type="bearer",
credential_ref="env:MASTER_KEY_B64",
),
)
with self.assertRaisesRegex(AddressBookError, "server-managed"):
create_sync_source(
self.session,
self.principal,
book.id,
AddressSyncSourceCreateRequest(
connector_type="carddav",
display_name="Injected",
metadata={"carddav": {"auth_type": "bearer", "credential_ref": "env:DATABASE_URL"}},
),
)
source = create_carddav_sync_source(
self.session,
self.principal,
book.id,
AddressCardDavSourceCreateRequest(
collection_url="https://dav.example.test/addressbooks/personal/",
auth_type="basic",
username="ada",
password="secret",
),
)
stored_auth = dict((source.metadata_ or {})["carddav"])
self.assertIn("secret_encrypted", stored_auth)
self.assertNotEqual(stored_auth["secret_encrypted"], "secret")
response_auth = _sync_source_response(source).metadata["carddav"]
self.assertNotIn("secret_encrypted", response_auth)
self.assertNotIn("credential_ref", response_auth)
self.assertTrue(response_auth["has_credential"])
with self.assertRaisesRegex(AddressBookError, "server-managed"):
update_sync_source(
self.session,
self.principal,
source.id,
AddressSyncSourceUpdateRequest(
metadata={"carddav": {"auth_type": "bearer", "secret_encrypted": "copied-ciphertext"}}
),
)
def test_legacy_carddav_env_reference_is_not_resolved_by_runtime_sync(self) -> None:
book = create_address_book(
self.session,
self.principal,
AddressBookCreateRequest(scope_type="user", name="Legacy remote"),
)
self.session.commit()
self.session.refresh(book)
source = create_sync_source(
self.session,
self.principal,
book.id,
AddressSyncSourceCreateRequest(
connector_type="carddav",
display_name="Legacy",
external_address_book_ref="https://attacker.example.test/addressbooks/personal/",
),
)
source.metadata_ = {
"carddav": {
"auth_type": "bearer",
"credential_ref": "env:MASTER_KEY_B64",
}
}
with patch.dict("os.environ", {"MASTER_KEY_B64": "must-not-leave-process"}), self.assertRaisesRegex(
AddressBookError,
"not a server-owned credential",
):
run_sync_source(self.session, self.principal, source.id)
def test_trusted_deployment_carddav_env_resolution_is_explicit(self) -> None:
with patch.dict("os.environ", {"CARDDAV_DEPLOYMENT_TOKEN": "trusted-token"}):
self.assertEqual(
resolve_trusted_deployment_carddav_credential_ref("env:CARDDAV_DEPLOYMENT_TOKEN"),
"trusted-token",
)
with self.assertRaisesRegex(AddressBookError, "must use the env: prefix"):
resolve_trusted_deployment_carddav_credential_ref("vault:token")
def test_carddav_two_way_pushes_local_creates_updates_and_deletes(self) -> None:
book = create_address_book(self.session, self.principal, AddressBookCreateRequest(scope_type="user", name="Writable Remote"))
self.session.commit()

16
tests/test_carddav_security.py
View File

@@ -37,6 +37,22 @@ def running_http_server(handler: type[BaseHTTPRequestHandler]) -> Iterator[str]:
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/"