From 869891f7eca6c36ad852f3ea08a83dff6c362e35 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 21 Aug 2026 04:14:22 +0200 Subject: [PATCH] feat(audit): add governed DSAR coverage --- src/govoplan_audit/backend/dsar_provider.py | 539 ++++++++++++++++++++ src/govoplan_audit/backend/manifest.py | 124 ++++- tests/test_dsar_provider.py | 440 ++++++++++++++++ 3 files changed, 1092 insertions(+), 11 deletions(-) create mode 100644 src/govoplan_audit/backend/dsar_provider.py create mode 100644 tests/test_dsar_provider.py diff --git a/src/govoplan_audit/backend/dsar_provider.py b/src/govoplan_audit/backend/dsar_provider.py new file mode 100644 index 0000000..7b5bf39 --- /dev/null +++ b/src/govoplan_audit/backend/dsar_provider.py @@ -0,0 +1,539 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy.orm import Session + +from govoplan_audit.backend.db.models import ( + AuditEvidenceBundle, + AuditLog, + AuditOutboxDelivery, + AuditOutboxEvent, +) +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarExecutionResultRef, + DsarRecordRef, + DsarSubjectRef, + dsar_capability_name, +) + + +AUDIT_DSAR_CAPABILITY = dsar_capability_name("audit") +_MAX_RECORDS = 5_000 +_CONFLICT = object() +_TRACE_KEYS = ( + "correlation_id", + "causation_id", + "request_id", + "run_id", + "trace_id", +) +_REFERENCE_KEYS = ( + "evidence_ref", + "legal_basis_ref", + "policy_decision_ref", + "policy_ref", + "source_ref", +) +_RESOURCE_TYPES = frozenset( + { + "audit_actor_record", + "audit_event_actor_record", + "audit_replay_attribution", + "audit_evidence_bundle_attribution", + } +) + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + actor_ids: tuple[str, ...] + log_id: str | None + event_id: str | None + delivery_id: str | None + bundle_id: str | None + + +class AuditDsarProvider: + provider_id = "audit" + module_id = "audit" + + def search_subject( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + ) -> Sequence[DsarRecordRef]: + db = _session(session) + selectors = _subject_selectors(subject) + if selectors is None: + return () + + records: list[DsarRecordRef] = [] + if not (selectors.event_id or selectors.delivery_id or selectors.bundle_id): + logs = db.query(AuditLog).filter( + AuditLog.scope == "tenant", + AuditLog.tenant_id == tenant_id, + AuditLog.user_id.in_(selectors.actor_ids), + ) + if selectors.log_id: + logs = logs.filter(AuditLog.id == selectors.log_id) + records.extend( + _log_record(row) + for row in _limited(logs, AuditLog, "actor audit records") + ) + + if not (selectors.log_id or selectors.delivery_id or selectors.bundle_id): + events = db.query(AuditOutboxEvent).filter( + AuditOutboxEvent.payload["tenant"]["id"].as_string() == tenant_id, + AuditOutboxEvent.payload["actor"]["id"] + .as_string() + .in_(selectors.actor_ids), + ) + if selectors.event_id: + events = events.filter( + (AuditOutboxEvent.id == selectors.event_id) + | (AuditOutboxEvent.event_id == selectors.event_id) + ) + records.extend( + _event_record(row) + for row in _limited(events, AuditOutboxEvent, "actor event records") + ) + + if not (selectors.log_id or selectors.event_id or selectors.bundle_id): + deliveries = ( + db.query(AuditOutboxDelivery, AuditOutboxEvent) + .join( + AuditOutboxEvent, + AuditOutboxDelivery.outbox_event_id == AuditOutboxEvent.id, + ) + .filter( + AuditOutboxEvent.payload["tenant"]["id"].as_string() == tenant_id, + AuditOutboxDelivery.last_replayed_by.in_(selectors.actor_ids), + ) + ) + if selectors.delivery_id: + deliveries = deliveries.filter( + AuditOutboxDelivery.id == selectors.delivery_id + ) + rows = ( + deliveries.order_by( + AuditOutboxDelivery.created_at, + AuditOutboxDelivery.id, + ) + .limit(_MAX_RECORDS + 1) + .all() + ) + if len(rows) > _MAX_RECORDS: + raise ValueError( + "Audit DSAR replay-attribution limit exceeded; narrow the selectors." + ) + records.extend( + _delivery_record(delivery, event) for delivery, event in rows + ) + + if not (selectors.log_id or selectors.event_id or selectors.delivery_id): + bundles = db.query(AuditEvidenceBundle).filter( + AuditEvidenceBundle.scope == "tenant", + AuditEvidenceBundle.tenant_id == tenant_id, + AuditEvidenceBundle.requested_by.in_(selectors.actor_ids), + ) + if selectors.bundle_id: + bundles = bundles.filter(AuditEvidenceBundle.id == selectors.bundle_id) + records.extend( + _bundle_record(row) + for row in _limited( + bundles, + AuditEvidenceBundle, + "evidence-bundle attribution", + ) + ) + + if len(records) > _MAX_RECORDS: + raise ValueError( + "Audit DSAR combined result limit exceeded; narrow the selectors." + ) + order = { + "audit_actor_record": 10, + "audit_event_actor_record": 20, + "audit_replay_attribution": 30, + "audit_evidence_bundle_attribution": 40, + } + return tuple( + sorted( + records, + key=lambda item: (order[item.resource_type], item.resource_id), + ) + ) + + def plan_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + records: Sequence[DsarRecordRef], + ) -> Sequence[DsarErasureActionRef]: + del tenant_id + _session(session) + if _subject_selectors(subject) is None: + raise ValueError("Audit DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + actions.append( + DsarErasureActionRef( + action_id=f"audit:retain:{record.resource_type}:{record.resource_id}", + provider_id=self.provider_id, + module_id=self.module_id, + kind="retain", + resource_type=record.resource_type, + resource_id=record.resource_id, + title=f"Retain {record.title}", + rationale=record.retention_reason + or "Audit evidence must remain immutable.", + executable=False, + ) + ) + return tuple(actions) + + def execute_erasure( + self, + session: object, + *, + tenant_id: str, + subject: DsarSubjectRef, + actions: Sequence[DsarErasureActionRef], + request_id: str, + ) -> Sequence[DsarExecutionResultRef]: + del tenant_id + _session(session) + if _subject_selectors(subject) is None: + raise ValueError("Audit DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind != "retain": + raise ValueError("Audit DSAR publishes retain-only actions.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Immutable Audit evidence remains under the configured " + "retention and legal-hold policy." + ), + evidence={"request_id": request_id}, + ) + ) + return tuple(results) + + +def _subject_selectors(subject: DsarSubjectRef) -> _SubjectSelectors | None: + references = subject.external_references + values = { + "account_id": _coalesce( + subject.account_id, + references.get("audit.account"), + references.get("access.account"), + ), + "membership_id": _coalesce( + subject.membership_id, + references.get("audit.membership"), + references.get("tenancy.membership"), + ), + "identity_id": _coalesce( + subject.identity_id, + references.get("audit.identity"), + references.get("identity.id"), + ), + "user_id": _coalesce( + references.get("audit.user"), + references.get("access.user"), + references.get("idm.user"), + ), + "log_id": _coalesce( + references.get("audit.log"), + references.get("audit.record"), + ), + "event_id": _coalesce( + references.get("audit.event"), + references.get("audit.outbox_event"), + ), + "delivery_id": _coalesce( + references.get("audit.delivery"), + references.get("audit.outbox_delivery"), + ), + "bundle_id": _coalesce( + references.get("audit.evidence_bundle"), + references.get("audit.bundle"), + ), + } + if any(value is _CONFLICT for value in values.values()): + return None + actor_ids = tuple( + dict.fromkeys( + value + for key in ("account_id", "membership_id", "identity_id", "user_id") + if (value := _optional_string(values[key])) + ) + ) + if not actor_ids: + return None + return _SubjectSelectors( + actor_ids=actor_ids, + log_id=_optional_string(values["log_id"]), + event_id=_optional_string(values["event_id"]), + delivery_id=_optional_string(values["delivery_id"]), + bundle_id=_optional_string(values["bundle_id"]), + ) + + +def _coalesce(*values: str | None) -> str | None | object: + normalized = {str(value).strip() for value in values if str(value or "").strip()} + if len(normalized) > 1: + return _CONFLICT + return next(iter(normalized), None) + + +def _optional_string(value: object) -> str | None: + return value if isinstance(value, str) and value else None + + +def _limited(query, model, label: str): + rows = query.order_by(model.created_at, model.id).limit(_MAX_RECORDS + 1).all() + if len(rows) > _MAX_RECORDS: + raise ValueError(f"Audit DSAR {label} limit exceeded; narrow the selectors.") + return rows + + +def _log_record(row: AuditLog) -> DsarRecordRef: + details = row.details if isinstance(row.details, Mapping) else {} + return _record( + "audit_actor_record", + row.id, + f"Audit action: {row.action[:100]}", + { + "scope": row.scope, + "actor_user_id": row.user_id, + "action": row.action, + "object_type": row.object_type, + "object_id": row.object_id, + "trace_context": _selected_context(details, _TRACE_KEYS), + "policy_and_source_references": _selected_context( + details, + _REFERENCE_KEYS, + ), + "recorded_at": _iso(row.created_at), + }, + observed_at=row.created_at, + reason=( + "Audit actions are immutable accountability evidence. Arbitrary " + "details and credentials are excluded from the access projection." + ), + ) + + +def _event_record(row: AuditOutboxEvent) -> DsarRecordRef: + payload = row.payload if isinstance(row.payload, Mapping) else {} + actor = _object_projection(payload.get("actor")) + subject = _object_projection(payload.get("subject")) + resource = _object_projection(payload.get("resource")) + return _record( + "audit_event_actor_record", + row.id, + f"Platform event attribution: {row.event_type[:200]}", + { + "event_id": row.event_id, + "event_type": row.event_type, + "module_id": row.module_id, + "correlation_id": row.correlation_id, + "causation_id": row.causation_id, + "classification": row.classification, + "actor": actor, + "subject": subject, + "resource": resource, + "occurred_at": _bounded_string(payload.get("occurred_at"), 100), + "status": row.status, + "attempts": row.attempts, + "next_attempt_at": _iso(row.next_attempt_at), + "dispatched_at": _iso(row.dispatched_at), + "recorded_at": _iso(row.created_at), + }, + observed_at=row.created_at, + reason=( + "Platform event envelopes and actor attribution remain immutable. " + "The event payload, institutional context, and delivery errors are excluded." + ), + ) + + +def _delivery_record( + row: AuditOutboxDelivery, + event: AuditOutboxEvent, +) -> DsarRecordRef: + return _record( + "audit_replay_attribution", + row.id, + "Audit outbox replay attribution", + { + "activity": "replayed_platform_event_delivery", + "event_id": event.event_id, + "event_type": event.event_type, + "module_id": event.module_id, + "consumer_id": row.consumer_id, + "status": row.status, + "policy_decision_ref": row.policy_decision_ref, + "replay_count": row.replay_count, + "last_replayed_at": _iso(row.last_replayed_at), + "recorded_at": _iso(row.created_at), + }, + observed_at=row.last_replayed_at or row.updated_at, + reason=( + "Manual replay attribution is immutable operational evidence. Replay " + "reasons, delivery keys, and provider errors are excluded." + ), + ) + + +def _bundle_record(row: AuditEvidenceBundle) -> DsarRecordRef: + return _record( + "audit_evidence_bundle_attribution", + row.id, + "Audit evidence-bundle request attribution", + { + "activity": "requested_audit_evidence_bundle", + "scope": row.scope, + "status": row.status, + "record_count": row.record_count, + "reference_count": row.reference_count, + "bundle_sha256": row.bundle_sha256, + "generated_at": _iso(row.generated_at), + "downloaded_at": _iso(row.downloaded_at), + "error_code": row.error_code, + "requested_at": _iso(row.created_at), + }, + observed_at=row.updated_at, + reason=( + "Evidence-bundle request attribution and verification hashes are " + "immutable. Request and bundle payloads are excluded." + ), + ) + + +def _record( + resource_type: str, + resource_id: str, + title: str, + data: Mapping[str, object], + *, + observed_at: datetime | None, + reason: str, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="audit", + module_id="audit", + resource_type=resource_type, + resource_id=resource_id, + category="immutable_accountability_evidence", + title=title, + data=data, + observed_at=_aware(observed_at), + immutable_evidence=True, + retention_reason=reason, + ) + + +def _selected_context( + details: Mapping[object, object], + allowed_keys: Sequence[str], +) -> dict[str, object]: + return { + key: _bounded_reference(details[key], depth=0) + for key in allowed_keys + if key in details + } + + +def _bounded_reference(value: object, *, depth: int) -> object: + if depth > 4: + return {"redacted": True, "reason": "depth_limit"} + if value is None or isinstance(value, (bool, int, float)): + return value + if isinstance(value, str): + return value[:2_048] + if isinstance(value, Mapping): + return { + str(key)[:200]: _bounded_reference(nested, depth=depth + 1) + for key, nested in list(value.items())[:50] + if not _sensitive_key(str(key)) + } + if isinstance(value, (list, tuple)): + return [_bounded_reference(item, depth=depth + 1) for item in value[:50]] + return {"redacted": True, "type": type(value).__name__} + + +def _object_projection(value: object) -> dict[str, str | None] | None: + if not isinstance(value, Mapping): + return None + return { + "type": _bounded_string(value.get("type"), 100), + "id": _bounded_string(value.get("id"), 255), + } + + +def _bounded_string(value: object, limit: int) -> str | None: + return str(value)[:limit] if value is not None else None + + +def _sensitive_key(value: str) -> bool: + normalized = value.strip().casefold().replace("-", "_") + return any( + part in normalized + for part in ( + "authorization", + "cookie", + "credential", + "password", + "secret", + "token", + ) + ) + + +def _iso(value: datetime | None) -> str | None: + aware = _aware(value) + return aware.isoformat() if aware else None + + +def _aware(value: datetime | None) -> datetime | None: + if value is None or value.tzinfo is not None: + return value + return value.replace(tzinfo=timezone.utc) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Audit DSAR requires a SQLAlchemy Session.") + return value + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "audit" or record.module_id != "audit": + raise ValueError("Audit DSAR cannot plan a foreign provider record.") + if record.resource_type not in _RESOURCE_TYPES or not record.resource_id: + raise ValueError("Audit DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "audit" or action.module_id != "audit": + raise ValueError("Audit DSAR cannot execute a foreign provider action.") + if not action.action_id.startswith("audit:"): + raise ValueError("Audit DSAR action identity is invalid.") + + +__all__ = ["AUDIT_DSAR_CAPABILITY", "AuditDsarProvider"] diff --git a/src/govoplan_audit/backend/manifest.py b/src/govoplan_audit/backend/manifest.py index 16ee469..3427580 100644 --- a/src/govoplan_audit/backend/manifest.py +++ b/src/govoplan_audit/backend/manifest.py @@ -3,14 +3,30 @@ from __future__ import annotations from pathlib import Path from govoplan_audit.backend.db import models as audit_models # noqa: F401 - populate Audit ORM metadata +from govoplan_audit.backend.dsar_provider import ( + AUDIT_DSAR_CAPABILITY, + AuditDsarProvider, +) from govoplan_core.core.access import ( CAPABILITY_AUDIT_RECORDER, CAPABILITY_AUDIT_RETENTION, CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) -from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard -from govoplan_core.core.modules import DocumentationCondition, DocumentationTopic, FrontendModule, MigrationSpec, ModuleContext, ModuleManifest +from govoplan_core.core.module_guards import ( + drop_table_retirement_provider, + persistent_table_uninstall_guard, +) +from govoplan_core.core.modules import ( + CapabilityDocumentation, + DocumentationCondition, + DocumentationTopic, + FrontendModule, + MigrationSpec, + ModuleContext, + ModuleInterfaceProvider, + ModuleManifest, +) from govoplan_core.core.provider_governance import declared_module_architecture from govoplan_core.core.events import CAPABILITY_PLATFORM_EVENT_OUTBOX from govoplan_core.core.views import ViewSurface @@ -51,15 +67,71 @@ def _event_outbox(context: ModuleContext): ) +def _dsar_provider(_context: ModuleContext) -> AuditDsarProvider: + return AuditDsarProvider() + + manifest = ModuleManifest( id="audit", name="Audit", version="0.1.18", permissions=AUDIT_PERMISSIONS, role_templates=AUDIT_ROLE_TEMPLATES, - required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), + required_capabilities=( + CAPABILITY_AUTH_PRINCIPAL_RESOLVER, + CAPABILITY_AUTH_PERMISSION_EVALUATOR, + ), + provides_interfaces=( + ModuleInterfaceProvider(name=AUDIT_DSAR_CAPABILITY, version="0.1.0"), + ), route_factory=_route_factory, documentation=( + DocumentationTopic( + id="audit.data-subject-requests", + title="Audit data-subject requests", + summary=( + "Include minimized subject-linked accountability evidence in " + "access packages while preserving immutable retention." + ), + body=( + "Audit correlates exact account, membership, identity, or user " + "identifiers only within the active tenant. It contributes actor " + "audit records, structured platform-event actor envelopes, manual " + "replay attribution, and evidence-bundle request attribution. The " + "projection preserves action, object, event, trace, source, policy, " + "status, timing, and verification references needed to explain the " + "subject's involvement. Arbitrary audit details, event payloads, " + "institutional context, delivery keys and errors, replay reasons, " + "request payloads, generated bundle payloads, and credentials are " + "excluded. Optional exact record references narrow results but never " + "replace actor corroboration. System and cross-tenant evidence is not " + "included in tenant requests. All Audit erasure actions are retain-only " + "and non-executable because the records are immutable accountability " + "evidence governed by retention and legal hold." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("data_subject", "auditor", "security_officer", "operator"), + related_modules=("core", "access", "policy", "ops"), + metadata={ + "help_contexts": [ + "audit.admin.tenant", + "audit.event-details", + "audit.evidence.export", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "export_actor_evidence": ( + "Returns minimized actor attribution and stable references, " + "never arbitrary evidence payloads." + ), + "retain_audit_evidence": ( + "Keeps immutable evidence under configured retention and " + "legal-hold policy." + ), + }, + }, + ), DocumentationTopic( id="audit.read-authorized-evidence", title="Read authorized audit evidence", @@ -103,7 +175,9 @@ manifest = ModuleManifest( audience=("auditor", "security_officer", "operator"), conditions=( DocumentationCondition(required_scopes=("audit:evidence:export",)), - DocumentationCondition(required_scopes=("audit:system_evidence:export",)), + DocumentationCondition( + required_scopes=("audit:system_evidence:export",) + ), ), related_modules=("policy", "files"), metadata={ @@ -125,16 +199,26 @@ manifest = ModuleManifest( module_id="audit", package_name="@govoplan/audit-webui", view_surfaces=( - ViewSurface(id="audit.admin.system", module_id="audit", kind="section", label="System audit", order=90), - ViewSurface(id="audit.admin.tenant", module_id="audit", kind="section", label="Tenant audit", order=100), + ViewSurface( + id="audit.admin.system", + module_id="audit", + kind="section", + label="System audit", + order=90, + ), + ViewSurface( + id="audit.admin.tenant", + module_id="audit", + kind="section", + label="Tenant audit", + order=100, + ), ), ), migration_spec=MigrationSpec( module_id="audit", metadata=Base.metadata, - script_location=str( - Path(__file__).with_name("migrations") / "versions" - ), + script_location=str(Path(__file__).with_name("migrations") / "versions"), retirement_supported=True, retirement_provider=drop_table_retirement_provider( audit_models.AuditEvidenceBundle, @@ -158,6 +242,17 @@ manifest = ModuleManifest( CAPABILITY_AUDIT_RECORDER: _audit_recorder, CAPABILITY_AUDIT_RETENTION: _audit_retention, CAPABILITY_PLATFORM_EVENT_OUTBOX: _event_outbox, + AUDIT_DSAR_CAPABILITY: _dsar_provider, + }, + capability_documentation={ + AUDIT_DSAR_CAPABILITY: CapabilityDocumentation( + label="Audit data-subject request provider", + summary=( + "Exports minimized tenant actor evidence with retain-only " + "erasure outcomes." + ), + contract_version="0.1.0", + ), }, architecture=declared_module_architecture( layer="governance_accountability", @@ -165,8 +260,15 @@ manifest = ModuleManifest( maturity="vertical_slice", documentation_ref="docs/AUDIT_TRACE_CONTEXT.md", test_ref="tests/test_audit_module_contract.py", - known_limits=("Cross-deployment long-term archive transfer remains deployment-specific.",), - owned_concepts=("audit record", "audit retention", "audit evidence bundle", "transactional event outbox"), + known_limits=( + "Cross-deployment long-term archive transfer remains deployment-specific.", + ), + owned_concepts=( + "audit record", + "audit retention", + "audit evidence bundle", + "transactional event outbox", + ), non_owned_concepts=("domain record", "policy decision", "external effect"), recovery_docs=("README.md",), security_docs=("docs/AUDIT_TRACE_CONTEXT.md", "docs/EVIDENCE_BUNDLES.md"), diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..412af7b --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,440 @@ +from __future__ import annotations + +import json +import unittest + +from sqlalchemy import Column, String, Table, create_engine +from sqlalchemy.orm import Session + +from govoplan_audit.backend.db.models import ( + AuditEvidenceBundle, + AuditLog, + AuditOutboxDelivery, + AuditOutboxEvent, +) +from govoplan_audit.backend.dsar_provider import ( + AUDIT_DSAR_CAPABILITY, + AuditDsarProvider, +) +from govoplan_audit.backend.manifest import manifest +from govoplan_core.core.dsar import ( + DsarErasureActionRef, + DsarProvider, + DsarRecordRef, + DsarSubjectRef, +) +from govoplan_core.db.base import Base +from govoplan_core.privacy.dsar_workflow import ( + DataSubjectRequest, + create_data_subject_request, + search_data_subject_request, +) + + +class _Registry: + def __init__(self, provider: AuditDsarProvider, *, active: bool = True) -> None: + self.provider = provider + self.active = active + + def capability_names(self): + return (AUDIT_DSAR_CAPABILITY,) + + def capability_owner(self, name): + self._assert_capability(name) + return "audit" + + def tenant_entitlement_resolver(self): + active = self.active + + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type( + "State", + (), + {"effective_modules": ("audit",) if active else ()}, + )() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del session, kwargs + self._assert_capability(name) + return self.provider + + def manifests(self): + return (type("Manifest", (), {"id": "audit"})(),) + + @staticmethod + def _assert_capability(name: str) -> None: + if name != AUDIT_DSAR_CAPABILITY: + raise KeyError(name) + + +class AuditDsarProviderTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:") + if "access_users" not in Base.metadata.tables: + Table( + "access_users", + Base.metadata, + Column("id", String(36), primary_key=True), + ) + if "access_api_keys" not in Base.metadata.tables: + Table( + "access_api_keys", + Base.metadata, + Column("id", String(36), primary_key=True), + ) + Base.metadata.create_all( + self.engine, + tables=[ + Base.metadata.tables["access_users"], + Base.metadata.tables["access_api_keys"], + AuditLog.__table__, + AuditOutboxEvent.__table__, + AuditOutboxDelivery.__table__, + AuditEvidenceBundle.__table__, + DataSubjectRequest.__table__, + ], + ) + self.session = Session(self.engine) + self.provider = AuditDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + self._seed() + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + @staticmethod + def _event_payload( + *, + event_id: str, + tenant_id: str, + actor_id: str, + ) -> dict[str, object]: + return { + "type": "case.decision.recorded", + "module_id": "cases", + "payload": { + "authorization_token": "event-secret-do-not-export", + "private_case_data": "third-party-data-do-not-export", + }, + "occurred_at": "2026-08-21T10:00:00+00:00", + "event_id": event_id, + "correlation_id": f"trace-{event_id}", + "causation_id": None, + "actor": {"type": "user", "id": actor_id, "label": "Private name"}, + "tenant": {"id": tenant_id, "label": "Private tenant label"}, + "subject": {"type": "case", "id": "case-1", "label": "Private subject"}, + "resource": { + "type": "decision", + "id": "decision-1", + "label": "Private resource", + }, + "classification": "confidential", + "institutional_context": { + "authorization": "institutional-secret-do-not-export" + }, + } + + def _seed(self) -> None: + event = AuditOutboxEvent( + id="outbox-1", + event_id="event-1", + event_type="case.decision.recorded", + module_id="cases", + correlation_id="trace-event-1", + causation_id=None, + classification="confidential", + payload=self._event_payload( + event_id="event-1", + tenant_id="tenant-1", + actor_id="user-1", + ), + status="dispatched", + attempts=1, + ) + self.session.add_all( + ( + AuditLog( + id="audit-log-1", + scope="tenant", + tenant_id="tenant-1", + user_id="user-1", + action="case.decision.recorded", + object_type="case", + object_id="case-1", + details={ + "correlation_id": "trace-audit-1", + "policy_decision_ref": "policy:decision-1", + "source_ref": "cases:case-1:v4", + "message_body": "audit-secret-do-not-export", + "authorization_token": "audit-token-do-not-export", + }, + ), + AuditLog( + id="audit-log-other", + scope="tenant", + tenant_id="tenant-1", + user_id="user-other", + action="other.action", + object_type="case", + object_id="case-other", + details={"private": "other actor"}, + ), + AuditLog( + id="audit-log-other-tenant", + scope="tenant", + tenant_id="tenant-2", + user_id="user-1", + action="other.tenant.action", + object_type="case", + object_id="case-other-tenant", + details={"private": "other tenant"}, + ), + event, + AuditOutboxDelivery( + id="delivery-1", + outbox_event_id="outbox-1", + consumer_id="reporting.audit-consumer", + delivery_key="event-1:reporting.audit-consumer", + policy_decision_ref="policy:delivery-1", + status="delivered", + attempts=1, + replay_count=1, + last_replayed_by="user-1", + last_replay_reason="private-replay-reason-do-not-export", + last_error="private-delivery-error-do-not-export", + ), + AuditEvidenceBundle( + id="bundle-1", + scope="tenant", + tenant_id="tenant-1", + requested_by="user-1", + status="ready", + request_payload={"secret": "request-secret-do-not-export"}, + bundle_payload={"secret": "bundle-secret-do-not-export"}, + bundle_sha256="a" * 64, + record_count=1, + reference_count=2, + ), + ) + ) + + @staticmethod + def _subject() -> DsarSubjectRef: + return DsarSubjectRef( + account_id="account-1", + membership_id="membership-1", + external_references={"audit.user": "user-1"}, + ) + + def test_search_is_tenant_actor_scoped_and_minimized(self) -> None: + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=self._subject(), + ) + + self.assertEqual( + [ + "audit_actor_record", + "audit_event_actor_record", + "audit_replay_attribution", + "audit_evidence_bundle_attribution", + ], + [record.resource_type for record in records], + ) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("trace-audit-1", exported) + self.assertIn("policy:decision-1", exported) + self.assertIn("decision-1", exported) + self.assertIn("a" * 64, exported) + for excluded in ( + "audit-secret-do-not-export", + "audit-token-do-not-export", + "event-secret-do-not-export", + "third-party-data-do-not-export", + "institutional-secret-do-not-export", + "private-replay-reason-do-not-export", + "private-delivery-error-do-not-export", + "request-secret-do-not-export", + "bundle-secret-do-not-export", + "audit-log-other", + "audit-log-other-tenant", + ): + self.assertNotIn(excluded, exported) + + def test_exact_references_narrow_and_alias_conflicts_fail_closed(self) -> None: + log = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={ + "audit.user": "user-1", + "audit.log": "audit-log-1", + } + ), + ) + event = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={ + "audit.user": "user-1", + "audit.event": "event-1", + } + ), + ) + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={"audit.account": "account-other"}, + ), + ) + reference_only = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(external_references={"audit.log": "audit-log-1"}), + ) + + self.assertEqual(["audit-log-1"], [item.resource_id for item in log]) + self.assertEqual(["outbox-1"], [item.resource_id for item in event]) + self.assertEqual((), conflict) + self.assertEqual((), reference_only) + + def test_erasure_is_retain_only_and_execution_is_blocked(self) -> None: + subject = self._subject() + records = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=subject, + ) + actions = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=records, + ) + self.assertTrue(actions) + self.assertTrue(all(action.kind == "retain" for action in actions)) + self.assertTrue(all(not action.executable for action in actions)) + + results = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=actions, + request_id="dsar-1", + ) + self.assertTrue(all(result.status == "blocked" for result in results)) + self.assertIsNotNone(self.session.get(AuditLog, "audit-log-1")) + self.assertIsNotNone(self.session.get(AuditOutboxEvent, "outbox-1")) + + def test_foreign_records_and_actions_are_rejected(self) -> None: + subject = self._subject() + with self.assertRaisesRegex(ValueError, "foreign provider record"): + self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + records=( + DsarRecordRef( + provider_id="cases", + module_id="cases", + resource_type="audit_actor_record", + resource_id="audit-log-1", + category="evidence", + title="Foreign evidence", + ), + ), + ) + with self.assertRaisesRegex(ValueError, "foreign provider action"): + self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=subject, + actions=( + DsarErasureActionRef( + action_id="cases:retain:audit:audit-log-1", + provider_id="cases", + module_id="cases", + kind="retain", + resource_type="audit_actor_record", + resource_id="audit-log-1", + title="Retain evidence", + rationale="Evidence", + executable=False, + ), + ), + request_id="dsar-1", + ) + + def test_core_workflow_and_manifest_register_provider(self) -> None: + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-AUDIT-1", + request_kind="access", + subject=self._subject(), + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider), + row=row, + expected_revision=1, + ) + self.assertEqual([AUDIT_DSAR_CAPABILITY], row.coverage["provider_capabilities"]) + self.assertEqual(4, row.search_result["record_count"]) + + inactive = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-AUDIT-2", + request_kind="access", + subject=self._subject(), + purpose="Respond to a verified request.", + legal_basis="Article 15 GDPR", + due_at=None, + requested_by_account_id="privacy-officer", + ) + self.session.commit() + search_data_subject_request( + self.session, + registry=_Registry(self.provider, active=False), + row=inactive, + expected_revision=1, + ) + self.assertEqual( + [AUDIT_DSAR_CAPABILITY], + inactive.coverage["inactive_provider_capabilities"], + ) + + self.assertIn(AUDIT_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn(AUDIT_DSAR_CAPABILITY, manifest.capability_documentation) + self.assertIn( + AUDIT_DSAR_CAPABILITY, + {item.name for item in manifest.provides_interfaces}, + ) + self.assertTrue( + any( + topic.id == "audit.data-subject-requests" + and {"admin", "user"}.issubset(topic.documentation_types) + for topic in manifest.documentation + ) + ) + + +if __name__ == "__main__": + unittest.main()