fix(secrets): audit connector credential deletion

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:

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()