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

fix(installer): audit credential retirement

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 15:46:20 +02:00
parent 613fb15a80
commit d005040a8e
3 changed files with 100 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
src/govoplan_addresses/backend/manifest.py
View File

@@ -1,7 +1,10 @@
from __future__ import annotations
from dataclasses import replace
from pathlib import Path
from sqlalchemy import inspect
from govoplan_addresses.backend.capabilities import (
CAPABILITY_ADDRESSES_CONTACT_WRITER,
CAPABILITY_ADDRESSES_LOOKUP,
@@ -25,6 +28,47 @@ from govoplan_core.core.modules import (
from govoplan_core.db.base import Base
_addresses_table_retirement_provider = drop_table_retirement_provider(
addresses_models.AddressSyncDiagnostic,
addresses_models.AddressSyncConflict,
addresses_models.AddressSyncTombstone,
addresses_models.AddressSyncSource,
addresses_models.AddressListEntry,
addresses_models.AddressList,
addresses_models.ContactPostalAddress,
addresses_models.ContactPhone,
addresses_models.ContactEmail,
addresses_models.Contact,
addresses_models.AddressBook,
label="Addresses",
)
def _addresses_retirement_provider(session: object | None, module_id: str):
plan = _addresses_table_retirement_provider(session, module_id)
base_executor = plan.destroy_data_executor
if base_executor is None:
return plan
def executor(execute_session: object, execute_module_id: str) -> None:
if not hasattr(execute_session, "get_bind") or not hasattr(execute_session, "query"):
raise RuntimeError("No database session is available for Addresses credential retirement.")
if inspect(execute_session.get_bind()).has_table(addresses_models.AddressSyncSource.__tablename__):
from govoplan_addresses.backend.service import audit_address_credentials_for_retirement
audit_address_credentials_for_retirement(execute_session)
base_executor(execute_session, execute_module_id)
return replace(
plan,
destroy_data_warnings=(
*plan.destroy_data_warnings,
"Addresses-owned encrypted connector credentials are audited and deleted with the sync-source table.",
),
destroy_data_executor=executor,
)
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
@@ -128,20 +172,7 @@ manifest = ModuleManifest(
metadata=Base.metadata,
script_location=str(Path(__file__).with_name("migrations") / "versions"),
retirement_supported=True,
retirement_provider=drop_table_retirement_provider(
addresses_models.AddressSyncDiagnostic,
addresses_models.AddressSyncConflict,
addresses_models.AddressSyncTombstone,
addresses_models.AddressSyncSource,
addresses_models.AddressListEntry,
addresses_models.AddressList,
addresses_models.ContactPostalAddress,
addresses_models.ContactPhone,
addresses_models.ContactEmail,
addresses_models.Contact,
addresses_models.AddressBook,
label="Addresses",
),
retirement_provider=_addresses_retirement_provider,
retirement_notes="Destructive retirement drops address-owned database tables after the installer captures a database snapshot.",
),
capability_factories={

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

@@ -417,6 +417,25 @@ def _record_sync_credential_deletion_audit(
return True
def audit_address_credentials_for_retirement(session: Session) -> int:
"""Audit credentials that the ensuing destructive table drop deletes."""
sources = session.query(AddressSyncSource).order_by(AddressSyncSource.id.asc()).all()
audited = 0
for source in sources:
if _record_sync_credential_deletion_audit(
session,
sync_source=source,
tenant_id=source.tenant_id,
user_id=None,
api_key_id=None,
deletion_reason="module_data_retired",
):
audited += 1
session.flush()
return audited
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:

37
tests/test_addresses_service.py
View File

@@ -3,7 +3,7 @@ from __future__ import annotations
import unittest
from unittest.mock import patch
from sqlalchemy import create_engine
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import sessionmaker
from govoplan_core.core.change_sequence import ChangeSequenceEntry
@@ -1118,5 +1118,40 @@ END:VCARD
self.assertIsNotNone(persisted)
self.assertIn("secret_encrypted", (persisted.metadata_ or {})["carddav"])
def test_destructive_module_retirement_audits_credentials_before_tables_are_dropped(self) -> None:
book = create_address_book(
self.session,
self.principal,
AddressBookCreateRequest(scope_type="user", name="Retired module"),
)
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/module-retirement/",
auth_type="bearer",
bearer_token="do-not-audit-this",
),
)
self.session.commit()
source_id = source.id
retirement_provider = manifest.migration_spec.retirement_provider
assert retirement_provider is not None
plan = retirement_provider(self.session, "addresses")
assert plan.destroy_data_executor is not None
with patch("govoplan_addresses.backend.service.audit_event") as audit:
plan.destroy_data_executor(self.session, "addresses")
audit.assert_called_once()
audit_call = audit.call_args.kwargs
self.assertEqual(audit_call["object_id"], source_id)
self.assertEqual(audit_call["details"]["deletion_reason"], "module_data_retired")
self.assertNotIn("do-not-audit-this", repr(audit_call))
self.assertFalse(inspect(self.engine).has_table("addresses_sync_sources"))
if __name__ == "__main__":
unittest.main()