feat(audit): add governed DSAR coverage

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"]

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.module_guards import (
-from govoplan_core.core.modules import DocumentationCondition, DocumentationTopic, FrontendModule, MigrationSpec, ModuleContext, ModuleManifest
+    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(
-            ViewSurface(id="audit.admin.tenant", module_id="audit", kind="section", label="Tenant audit", order=100),
+                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(
+        script_location=str(Path(__file__).with_name("migrations") / "versions"),
-            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.",),
+        known_limits=(
-        owned_concepts=("audit record", "audit retention", "audit evidence bundle", "transactional event outbox"),
+            "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"),

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