From 613fb15a80b311f4771a80d726d83a778c65bf5f Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 15:45:39 +0200 Subject: [PATCH] fix(secrets): audit connector credential deletion --- src/govoplan_addresses/backend/service.py | 62 +++++++++++++++++++++ tests/test_addresses_service.py | 66 +++++++++++++++++++++++ 2 files changed, 128 insertions(+) diff --git a/src/govoplan_addresses/backend/service.py b/src/govoplan_addresses/backend/service.py index 42e2f8b..0b771de 100644 --- a/src/govoplan_addresses/backend/service.py +++ b/src/govoplan_addresses/backend/service.py @@ -12,6 +12,7 @@ from sqlalchemy import and_, false, func, or_ from sqlalchemy.orm import Session 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.secrets import decrypt_secret, encrypt_secret @@ -352,9 +353,70 @@ def delete_sync_source(session: Session, principal: ApiPrincipal, sync_source_id book.sync_status = None book.sync_error = None book.updated_by_account_id = _account_id(principal) + _audit_sync_credential_deletion(session, principal, sync_source) session.delete(sync_source) +def _audit_sync_credential_deletion( + session: Session, + principal: ApiPrincipal, + sync_source: AddressSyncSource, +) -> None: + """Audit removal of source-owned credential material in the DB transaction. + + CardDAV credentials are encrypted inside the sync-source row. Deleting that + row therefore deletes the credential atomically with the connector. Legacy + credential references are detached from the source, but are never resolved + or sent to an external provider because ownership cannot be proven. + """ + + tenant_id = _tenant_id_or_none(principal) + user = getattr(principal, "user", None) + _record_sync_credential_deletion_audit( + session, + sync_source=sync_source, + tenant_id=tenant_id, + user_id=getattr(user, "id", None), + api_key_id=getattr(principal, "api_key_id", None), + deletion_reason="sync_source_deleted", + ) + + +def _record_sync_credential_deletion_audit( + session: Session, + *, + sync_source: AddressSyncSource, + tenant_id: str | None, + user_id: str | None, + api_key_id: str | None, + deletion_reason: str, +) -> bool: + auth = _carddav_auth_metadata(sync_source.metadata_) + if auth.get("secret_encrypted"): + storage_backend = "encrypted_database" + elif auth.get("credential_ref"): + storage_backend = "legacy_reference" + else: + return False + audit_event( + session, + tenant_id=tenant_id, + user_id=user_id, + api_key_id=api_key_id, + action="addresses.sync_credential_deleted", + scope="tenant" if tenant_id is not None else "system", + object_type="address_sync_credential", + object_id=sync_source.id, + details={ + "sync_source_id": sync_source.id, + "connector_type": sync_source.connector_type, + "storage_backend": storage_backend, + "deletion_reason": deletion_reason, + }, + ) + return True + + def start_sync_attempt(session: Session, principal: ApiPrincipal, sync_source_id: str) -> AddressSyncSource: sync_source = get_visible_sync_source(session, principal, sync_source_id) if not sync_source.enabled: diff --git a/tests/test_addresses_service.py b/tests/test_addresses_service.py index 9e2d1b3..c72d41e 100644 --- a/tests/test_addresses_service.py +++ b/tests/test_addresses_service.py @@ -1051,6 +1051,72 @@ END:VCARD self.assertFalse(book.read_only) self.assertEqual([item.id for item in list_contacts(self.session, self.principal, address_book_id=book.id)], [contact.id]) + def test_delete_sync_source_deletes_and_audits_encrypted_credential_atomically(self) -> None: + book = create_address_book( + self.session, + self.principal, + AddressBookCreateRequest(scope_type="user", name="Audited remote"), + ) + self.session.commit() + self.session.refresh(book) + source = create_carddav_sync_source( + self.session, + self.principal, + book.id, + AddressCardDavSourceCreateRequest( + collection_url="https://carddav.example.local/addressbooks/audited/", + auth_type="basic", + username="ada", + password="do-not-audit-this", + ), + ) + self.session.commit() + source_id = source.id + + with patch("govoplan_addresses.backend.service.audit_event") as audit: + delete_sync_source(self.session, self.principal, source_id) + self.session.commit() + + self.assertIsNone(self.session.get(AddressSyncSource, source_id)) + audit.assert_called_once() + audit_call = audit.call_args.kwargs + self.assertEqual(audit_call["action"], "addresses.sync_credential_deleted") + self.assertEqual(audit_call["object_id"], source_id) + self.assertEqual(audit_call["details"]["storage_backend"], "encrypted_database") + self.assertNotIn("credential_ref", audit_call["details"]) + self.assertNotIn("secret_encrypted", audit_call["details"]) + self.assertNotIn("do-not-audit-this", repr(audit_call)) + + def test_delete_sync_source_is_not_staged_when_credential_audit_fails(self) -> None: + book = create_address_book( + self.session, + self.principal, + AddressBookCreateRequest(scope_type="user", name="Audit failure"), + ) + self.session.commit() + self.session.refresh(book) + source = create_carddav_sync_source( + self.session, + self.principal, + book.id, + AddressCardDavSourceCreateRequest( + collection_url="https://carddav.example.local/addressbooks/audit-failure/", + auth_type="bearer", + bearer_token="do-not-audit-this", + ), + ) + self.session.commit() + + with patch( + "govoplan_addresses.backend.service.audit_event", + side_effect=RuntimeError("audit unavailable"), + ), self.assertRaisesRegex(RuntimeError, "audit unavailable"): + delete_sync_source(self.session, self.principal, source.id) + + self.session.rollback() + persisted = self.session.get(AddressSyncSource, source.id) + self.assertIsNotNone(persisted) + self.assertIn("secret_encrypted", (persisted.metadata_ or {})["carddav"]) if __name__ == "__main__": unittest.main()