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

Integrate address sources with credential envelopes

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-28 19:33:08 +02:00
parent 93dddbb8c5
commit eab24750f9
6 changed files with 362 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -41,6 +41,8 @@ from govoplan_addresses.backend.schemas import (
AddressCardDavDiscoveryRequest,
AddressCardDavDiscoveryResponse,
AddressCardDavSourceCreateRequest,
AddressCredentialEnvelopeListResponse,
AddressCredentialEnvelopeResponse,
AddressSyncAttemptFinishRequest,
AddressSyncConflictCreateRequest,
AddressSyncConflictListResponse,
@@ -69,6 +71,7 @@ from govoplan_addresses.backend.schemas import (
)
from govoplan_addresses.backend.service import (
AddressBookError,
available_address_credentials,
address_book_contact_counts,
address_list_entry_counts,
create_address_book,
@@ -639,6 +642,25 @@ def api_discover_carddav_address_books(
raise _error(AddressBookError(str(exc))) from exc
@router.get("/credentials", response_model=AddressCredentialEnvelopeListResponse)
def api_list_address_credentials(
source_id: str | None = Query(default=None),
principal: ApiPrincipal = Depends(get_api_principal),
session: Session = Depends(get_session),
):
_require_scope(principal, "addresses:sync:write")
return AddressCredentialEnvelopeListResponse(
credentials=[
AddressCredentialEnvelopeResponse.model_validate(item)
for item in available_address_credentials(
session,
principal,
source_id=source_id,
)
]
)
@router.post("/address-books/{book_id}/carddav/sources", response_model=AddressSyncSourceResponse, status_code=status.HTTP_201_CREATED)
def api_create_carddav_sync_source(
book_id: str,

20
src/govoplan_addresses/backend/schemas.py
View File

@@ -439,6 +439,26 @@ class AddressCardDavDiscoveryResponse(BaseModel):
address_books: list[AddressCardDavAddressBookResponse]
class AddressCredentialEnvelopeResponse(BaseModel):
id: str
scope_type: str
scope_id: str | None = None
name: str
description: str | None = None
credential_kind: str
public_data: dict[str, Any] = Field(default_factory=dict)
secret_keys: list[str] = Field(default_factory=list)
secret_configured: bool = False
allowed_modules: list[str] = Field(default_factory=list)
inherit_to_lower_scopes: bool = False
is_active: bool = True
revision: str
class AddressCredentialEnvelopeListResponse(BaseModel):
credentials: list[AddressCredentialEnvelopeResponse] = Field(default_factory=list)
class AddressCardDavSourceCreateRequest(BaseModel):
collection_url: str = Field(min_length=1, max_length=2000)
display_name: str | None = Field(default=None, max_length=255)

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

@@ -15,6 +15,14 @@ from govoplan_core.auth import ApiPrincipal
from govoplan_core.audit.logging import audit_event
from govoplan_core.core.change_sequence import record_change
from govoplan_core.db.base import utcnow
from govoplan_core.security.credential_envelopes import (
CredentialAccessContext,
CredentialEnvelopeError,
ResolvedCredentialEnvelope,
credential_envelope_summary,
list_credential_envelopes,
resolve_credential_envelope,
)
from govoplan_core.security.secrets import decrypt_secret, encrypt_secret
from govoplan_addresses.backend.carddav import (
AddressCardDAVAddressBook,
@@ -68,6 +76,39 @@ class AddressBookError(ValueError):
pass
CORE_CREDENTIAL_ENVELOPE_PREFIX = "credential-envelope:"
def address_credential_context(
*,
tenant_id: str,
source_id: str | None = None,
) -> CredentialAccessContext:
return CredentialAccessContext(
tenant_id=tenant_id,
target_scope_type="tenant",
target_scope_id=tenant_id,
module_id="addresses",
server_ref=f"addresses:{source_id}" if source_id else None,
)
def available_address_credentials(
session: Session,
principal: ApiPrincipal,
*,
source_id: str | None = None,
) -> list[dict[str, Any]]:
context = address_credential_context(
tenant_id=principal.tenant_id,
source_id=source_id,
)
return [
credential_envelope_summary(row)
for row in list_credential_envelopes(session, context=context)
]
@dataclass(frozen=True, slots=True)
class VCardImportResult:
contacts: list[Contact]
@@ -693,6 +734,8 @@ def create_carddav_sync_source(
payload: AddressCardDavSourceCreateRequest,
) -> AddressSyncSource:
_assert_no_caller_carddav_credential_ref(payload.credential_ref)
if payload.credential_ref and (payload.password is not None or payload.bearer_token is not None):
raise AddressBookError("Select a reusable credential or enter a new secret, not both.")
collection_url = ensure_collection_url(payload.collection_url)
display_name = _trim(payload.display_name) or "CardDAV address book"
metadata = _carddav_metadata(
@@ -700,10 +743,10 @@ def create_carddav_sync_source(
username=payload.username,
password=_secret_value(payload.password),
bearer_token=_secret_value(payload.bearer_token),
credential_ref=None,
credential_ref=payload.credential_ref,
collection_url=collection_url,
)
return create_sync_source(
source = create_sync_source(
session,
principal,
address_book_id,
@@ -720,6 +763,19 @@ def create_carddav_sync_source(
),
trusted_connector_metadata=True,
)
reusable = _resolve_core_address_credential(
session,
tenant_id=principal.tenant_id,
source_id=source.id,
credential_ref=payload.credential_ref,
)
if reusable is not None:
source_metadata = dict(source.metadata_ or {})
auth = dict(source_metadata.get("carddav") or {})
auth["username"] = _credential_username(reusable) or auth.get("username")
source_metadata["carddav"] = auth
source.metadata_ = source_metadata
return source
def preview_sync_source(
@@ -1131,8 +1187,6 @@ def _plan_matching_carddav_revision(
etag=item.etag,
),
)
def _plan_carddav_remote_update(
*,
item: AddressCardDAVObject,
@@ -1618,7 +1672,23 @@ def _carddav_client_from_payload(
password = _secret_value(payload.password)
bearer_token = _secret_value(payload.bearer_token)
credential_ref = payload.credential_ref if payload.credential_ref is not None else auth.get("credential_ref")
secret = _resolve_carddav_secret(auth_type=auth_type, password=password, bearer_token=bearer_token, credential_ref=credential_ref, encrypted=auth.get("secret_encrypted"))
reusable = _resolve_core_address_credential(
session,
tenant_id=principal.tenant_id,
source_id=source.id if source else None,
credential_ref=credential_ref,
)
username = username or _credential_username(reusable)
secret = _resolve_carddav_secret(
session=session,
tenant_id=principal.tenant_id,
source_id=source.id if source else None,
auth_type=auth_type,
password=password,
bearer_token=bearer_token,
credential_ref=credential_ref,
encrypted=auth.get("secret_encrypted"),
)
return _new_carddav_client(url, auth_type=auth_type, username=username, secret=secret)
@@ -1629,12 +1699,21 @@ def _carddav_client_for_source(
password: str | None = None,
bearer_token: str | None = None,
) -> AddressCardDAVClient:
del session
metadata = dict(sync_source.metadata_ or {})
auth = dict(metadata.get("carddav") or {})
auth_type = str(auth.get("auth_type") or "none")
username = auth.get("username")
reusable = _resolve_core_address_credential(
session,
tenant_id=sync_source.tenant_id or "",
source_id=sync_source.id,
credential_ref=auth.get("credential_ref"),
)
username = username or _credential_username(reusable)
secret = _resolve_carddav_secret(
session=session,
tenant_id=sync_source.tenant_id or "",
source_id=sync_source.id,
auth_type=auth_type,
password=password,
bearer_token=bearer_token,
@@ -1683,6 +1762,9 @@ def _carddav_metadata(
def _resolve_carddav_secret(
*,
session: Session,
tenant_id: str,
source_id: str | None,
auth_type: str,
password: str | None,
bearer_token: str | None,
@@ -1694,14 +1776,74 @@ def _resolve_carddav_secret(
if auth_type == "bearer" and bearer_token:
return bearer_token
if credential_ref:
raise AddressBookError(
"The CardDAV credential reference is not a server-owned credential; provide a replacement password or token"
reusable = _resolve_core_address_credential(
session,
tenant_id=tenant_id,
source_id=source_id,
credential_ref=credential_ref,
)
if reusable is None:
raise AddressBookError(
"The CardDAV credential reference is not a server-owned credential or visible reusable credential envelope."
)
return _credential_secret(reusable, auth_type=auth_type)
if encrypted:
return decrypt_secret(encrypted)
return None
def _core_credential_id(credential_ref: str | None) -> str | None:
if not credential_ref or not credential_ref.startswith(CORE_CREDENTIAL_ENVELOPE_PREFIX):
return None
value = credential_ref.removeprefix(CORE_CREDENTIAL_ENVELOPE_PREFIX).strip()
return value or None
def _resolve_core_address_credential(
session: Session,
*,
tenant_id: str,
source_id: str | None,
credential_ref: str | None,
) -> ResolvedCredentialEnvelope | None:
credential_id = _core_credential_id(credential_ref)
if credential_id is None:
return None
try:
return resolve_credential_envelope(
session,
credential_id=credential_id,
context=address_credential_context(
tenant_id=tenant_id,
source_id=source_id,
),
)
except CredentialEnvelopeError as exc:
raise AddressBookError(
"The selected reusable credential is unavailable to this CardDAV source."
) from exc
def _credential_username(credential: ResolvedCredentialEnvelope | None) -> str | None:
if credential is None:
return None
value = credential.public_data.get("username")
return _trim(value)
def _credential_secret(credential: ResolvedCredentialEnvelope, *, auth_type: str) -> str | None:
keys = (
("password", "secret", "token")
if auth_type == "basic"
else ("access_token", "bearer_token", "token", "password", "secret")
)
for key in keys:
value = credential.secret_data.get(key)
if value is not None and str(value):
return str(value)
return None
def resolve_trusted_deployment_carddav_credential_ref(credential_ref: str) -> str | None:
"""Resolve env-backed credentials only for trusted deployment code.
@@ -1724,6 +1866,7 @@ def public_address_sync_metadata(metadata: object) -> dict[str, Any]:
if not isinstance(auth, dict):
return payload
had_credential = bool(auth.get("secret_encrypted") or auth.get("credential_ref"))
auth["credential_envelope_id"] = _core_credential_id(auth.get("credential_ref"))
auth.pop("secret_encrypted", None)
auth.pop("credential_ref", None)
auth["has_credential"] = had_credential
@@ -1731,9 +1874,10 @@ def public_address_sync_metadata(metadata: object) -> dict[str, Any]:
def _assert_no_caller_carddav_credential_ref(credential_ref: str | None) -> None:
if _trim(credential_ref):
value = _trim(credential_ref)
if value and not value.startswith(CORE_CREDENTIAL_ENVELOPE_PREFIX):
raise AddressBookError(
"Caller-supplied credential references are not accepted; provide a password or bearer token"
"Caller-supplied credential references are accepted only for reusable credential envelopes."
)