feat(policy): add governed DSAR coverage

src/govoplan_policy/backend/dsar_provider.py

@@ -0,0 +1,208 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import datetime, timezone
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import (
+    DsarErasureActionRef,
+    DsarExecutionResultRef,
+    DsarRecordRef,
+    DsarSubjectRef,
+    dsar_capability_name,
+)
+from govoplan_policy.backend.db.models import PolicyOverride
+
+
+POLICY_DSAR_CAPABILITY = dsar_capability_name("policy")
+_MAX_RECORDS = 5_000
+_CONFLICT = object()
+
+
+class PolicyDsarProvider:
+    provider_id = "policy"
+    module_id = "policy"
+
+    def search_subject(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+    ) -> Sequence[DsarRecordRef]:
+        db = _session(session)
+        selectors = _selectors(subject)
+        if selectors is None:
+            return ()
+        account_id, override_id = selectors
+        query = db.query(PolicyOverride).filter(
+            PolicyOverride.tenant_id == tenant_id,
+            or_(
+                PolicyOverride.created_by == account_id,
+                PolicyOverride.updated_by == account_id,
+            ),
+        )
+        if override_id:
+            query = query.filter(PolicyOverride.id == override_id)
+        rows = (
+            query.order_by(PolicyOverride.created_at, PolicyOverride.id)
+            .limit(_MAX_RECORDS + 1)
+            .all()
+        )
+        if len(rows) > _MAX_RECORDS:
+            raise ValueError("Policy DSAR result limit exceeded; narrow selectors.")
+        return tuple(_record(row, account_id) for row in rows)
+
+    def plan_erasure(
+        self,
+        session: object,
+        *,
+        tenant_id: str,
+        subject: DsarSubjectRef,
+        records: Sequence[DsarRecordRef],
+    ) -> Sequence[DsarErasureActionRef]:
+        del tenant_id
+        _session(session)
+        if _selectors(subject) is None:
+            raise ValueError("Policy DSAR subject selectors conflict.")
+        actions = []
+        for record in records:
+            _validate_record(record)
+            actions.append(
+                DsarErasureActionRef(
+                    action_id=f"policy:retain:{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 "Policy-change attribution remains governance evidence."
+                    ),
+                    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 _selectors(subject) is None:
+            raise ValueError("Policy DSAR subject selectors conflict.")
+        results = []
+        for action in actions:
+            _validate_action(action)
+            if action.executable or action.kind != "retain":
+                raise ValueError("Policy DSAR publishes retain actions only.")
+            results.append(
+                DsarExecutionResultRef(
+                    action_id=action.action_id,
+                    status="blocked",
+                    summary="Policy-change attribution remains governance evidence.",
+                    evidence={"request_id": request_id},
+                )
+            )
+        return tuple(results)
+
+
+def _selectors(subject: DsarSubjectRef) -> tuple[str, str | None] | None:
+    references = subject.external_references
+    account = _coalesce(
+        subject.account_id,
+        references.get("policy.account"),
+        references.get("access.account"),
+    )
+    override_id = _coalesce(
+        references.get("policy.override"), references.get("policy.override_id")
+    )
+    if account is _CONFLICT or override_id is _CONFLICT:
+        return None
+    if not isinstance(account, str) or not account:
+        return None
+    return account, override_id if isinstance(override_id, str) else None
+
+
+def _record(row: PolicyOverride, account_id: str) -> DsarRecordRef:
+    activities = []
+    if row.created_by == account_id:
+        activities.append("created_policy_override")
+    if row.updated_by == account_id:
+        activities.append("updated_policy_override")
+    return DsarRecordRef(
+        provider_id="policy",
+        module_id="policy",
+        resource_type="policy_override_actor_attribution",
+        resource_id=row.id,
+        category="policy_governance_attribution",
+        title="Policy override actor attribution",
+        data={
+            "override_id": row.id,
+            "policy_family": row.policy_family,
+            "scope_type": row.scope_type,
+            "revision": row.revision,
+            "activities": activities,
+            "created_at": _iso(row.created_at),
+            "updated_at": _iso(row.updated_at),
+        },
+        observed_at=_aware(row.updated_at),
+        immutable_evidence=True,
+        retention_reason=(
+            "Policy-change attribution is retained for governance and accountability."
+        ),
+    )
+
+
+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 _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("Policy DSAR requires a SQLAlchemy Session.")
+    return value
+
+
+def _validate_record(record: DsarRecordRef) -> None:
+    if record.provider_id != "policy" or record.module_id != "policy":
+        raise ValueError("Policy DSAR cannot plan a foreign provider record.")
+    if (
+        record.resource_type != "policy_override_actor_attribution"
+        or not record.resource_id
+    ):
+        raise ValueError("Policy DSAR record identity is invalid.")
+
+
+def _validate_action(action: DsarErasureActionRef) -> None:
+    if action.provider_id != "policy" or action.module_id != "policy":
+        raise ValueError("Policy DSAR cannot execute a foreign provider action.")
+    if not action.action_id.startswith("policy:retain:"):
+        raise ValueError("Policy DSAR action identity is invalid.")
+
+
+__all__ = ["POLICY_DSAR_CAPABILITY", "PolicyDsarProvider"]

src/govoplan_policy/backend/manifest.py

@@ -37,6 +37,10 @@ from govoplan_core.core.provider_governance import declared_module_architecture
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_policy.backend.db import models as policy_models
+from govoplan_policy.backend.dsar_provider import (
+    POLICY_DSAR_CAPABILITY,
+    PolicyDsarProvider,
+)
 
 
 def _route_factory(context: ModuleContext):
@@ -125,6 +129,10 @@ def _campaign_archive_encryption_policy(context: ModuleContext) -> object:
     return CampaignArchiveEncryptionPolicyProvider()
 
 
+def _dsar_provider(_context: ModuleContext) -> PolicyDsarProvider:
+    return PolicyDsarProvider()
+
+
 ACCESS_EXPLANATION_SUBJECT_SCOPE = "policy:access_explanation:select_user"
 POLICY_IMPACT_DETAILS_SCOPE = "policy:impact:details"
 
@@ -193,9 +201,39 @@ manifest = ModuleManifest(
             name=CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS,
             version="1.0.0",
         ),
+        ModuleInterfaceProvider(name=POLICY_DSAR_CAPABILITY, version="0.1.0"),
     ),
     route_factory=_route_factory,
     documentation=(
+        DocumentationTopic(
+            id="policy.data-subject-requests",
+            title="Policy data-subject requests",
+            summary=(
+                "Export policy-change attribution without disclosing policy documents "
+                "or scoped subject identifiers."
+            ),
+            body=(
+                "Policy correlates only an exact account identifier within the active "
+                "tenant and can narrow an already verified search to one override. It "
+                "returns minimized creation and update activity with the policy family, "
+                "scope type, revision, and timestamps. Policy values, target and scope "
+                "keys, scope identifiers, and decision provenance are not included. "
+                "System-scoped overrides are not projected into a tenant request. Policy "
+                "change attribution remains governance evidence and is retained rather "
+                "than automatically erased."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("user", "policy_admin", "privacy_officer", "auditor"),
+            related_modules=("core", "access", "audit"),
+            metadata={
+                "help_contexts": ["privacy.data-subject-requests"],
+                "consequence_classes": {
+                    "export_policy_attribution": "Returns minimized policy activity for the exact account.",
+                    "retain_policy_evidence": "Preserves policy governance accountability.",
+                },
+            },
+        ),
         DocumentationTopic(
             id="policy.access-explanation-subjects",
             title="Choose subjects for access diagnostics",
@@ -310,7 +348,12 @@ manifest = ModuleManifest(
                 "The secure baseline permits AES only. An authorized policy administrator may explicitly permit legacy ZipCrypto at system scope, after which tenant, owner group or user, and campaign rules may only narrow the inherited methods. The same intersection controls the separate channel used to convey a password. Policy records the complete source path and a stable policy hash; malformed configuration fails closed. Policy changes never rewrite old build evidence, while Campaign rejects a queued or sent build whose effective policy is now more restrictive."
             ),
             documentation_types=("admin", "user"),
-            audience=("system_admin", "tenant_admin", "policy_admin", "campaign_manager"),
+            audience=(
+                "system_admin",
+                "tenant_admin",
+                "policy_admin",
+                "campaign_manager",
+            ),
             related_modules=("campaign", "audit", "access"),
             metadata={
                 "kind": "reference",
@@ -558,6 +601,7 @@ manifest = ModuleManifest(
         CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service,
         CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy,
         CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION: _campaign_archive_encryption_policy,
+        POLICY_DSAR_CAPABILITY: _dsar_provider,
     },
     capability_documentation={
         CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS: CapabilityDocumentation(
@@ -587,6 +631,13 @@ manifest = ModuleManifest(
             documentation_types=("admin", "user"),
             audience=("policy_admin", "campaign_manager"),
         ),
+        POLICY_DSAR_CAPABILITY: CapabilityDocumentation(
+            label="Policy data-subject request provider",
+            summary=(
+                "Exports minimized tenant policy-change attribution without policy payloads."
+            ),
+            contract_version="0.1.0",
+        ),
     },
     architecture=declared_module_architecture(
         layer="governance_accountability",

tests/test_dsar_provider.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import json
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.core.dsar import DsarProvider, DsarSubjectRef
+from govoplan_core.db.base import Base
+from govoplan_policy.backend.db.models import PolicyOverride
+from govoplan_policy.backend.dsar_provider import (
+    POLICY_DSAR_CAPABILITY,
+    PolicyDsarProvider,
+)
+from govoplan_policy.backend.manifest import manifest
+
+
+class PolicyDsarProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.session = Session(self.engine)
+        self.provider = PolicyDsarProvider()
+        self.session.add_all(
+            (
+                PolicyOverride(
+                    id="override-1",
+                    policy_family="retention",
+                    target_key="target-secret-do-not-export",
+                    tenant_id="tenant-1",
+                    scope_type="user",
+                    scope_id="scope-subject-do-not-export",
+                    scope_key="scope-key-do-not-export",
+                    policy={"secret": "policy-payload-do-not-export"},
+                    revision=2,
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+                PolicyOverride(
+                    id="override-system",
+                    policy_family="retention",
+                    target_key="system",
+                    tenant_id=None,
+                    scope_type="system",
+                    scope_key="system",
+                    policy={},
+                    revision=1,
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+                PolicyOverride(
+                    id="override-other",
+                    policy_family="retention",
+                    target_key="other",
+                    tenant_id="tenant-2",
+                    scope_type="tenant",
+                    scope_key="tenant-2",
+                    policy={},
+                    revision=1,
+                    created_by="account-1",
+                    updated_by="account-1",
+                ),
+            )
+        )
+        self.session.commit()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_search_is_tenant_safe_and_minimized(self) -> None:
+        self.assertIsInstance(self.provider, DsarProvider)
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(account_id="account-1"),
+        )
+        self.assertEqual(["override-1"], [record.resource_id for record in records])
+        exported = json.dumps([record.to_dict() for record in records])
+        for excluded in (
+            "target-secret-do-not-export",
+            "scope-subject-do-not-export",
+            "scope-key-do-not-export",
+            "policy-payload-do-not-export",
+        ):
+            self.assertNotIn(excluded, exported)
+
+    def test_requires_exact_account_and_supports_narrowing(self) -> None:
+        self.assertEqual(
+            (),
+            self.provider.search_subject(
+                self.session,
+                tenant_id="tenant-1",
+                subject=DsarSubjectRef(email="policy@example.test"),
+            ),
+        )
+        records = self.provider.search_subject(
+            self.session,
+            tenant_id="tenant-1",
+            subject=DsarSubjectRef(
+                account_id="account-1",
+                external_references={"policy.override": "override-1"},
+            ),
+        )
+        self.assertEqual(1, len(records))
+
+    def test_records_are_retained_and_manifest_is_complete(self) -> None:
+        subject = DsarSubjectRef(account_id="account-1")
+        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(all(action.kind == "retain" for action in actions))
+        self.assertIn(POLICY_DSAR_CAPABILITY, manifest.capability_factories)
+        self.assertIn(
+            "policy.data-subject-requests",
+            {topic.id for topic in manifest.documentation},
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_policy_module_contract.py

@@ -18,6 +18,7 @@ from govoplan_core.core.distribution_lists import (
 )
 from govoplan_core.core.reporting import CAPABILITY_POLICY_REPORTING_GOVERNANCE
 from govoplan_policy.backend.manifest import manifest
+from govoplan_policy.backend.dsar_provider import POLICY_DSAR_CAPABILITY
 
 
 ROOT = pathlib.Path(__file__).resolve().parents[1]
@@ -58,6 +59,7 @@ class PolicyModuleContractTests(unittest.TestCase):
                 CAPABILITY_POLICY_VIEW_GOVERNANCE,
                 CAPABILITY_POLICY_ACCESS_EXPLANATION_SUBJECTS,
                 CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION,
+                POLICY_DSAR_CAPABILITY,
             },
             set(manifest.capability_factories),
         )