diff --git a/src/govoplan_risk_compliance/backend/dsar_provider.py b/src/govoplan_risk_compliance/backend/dsar_provider.py new file mode 100644 index 0000000..d5348ea --- /dev/null +++ b/src/govoplan_risk_compliance/backend/dsar_provider.py @@ -0,0 +1,695 @@ +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +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_risk_compliance.backend.db.models import ( + RiskAssuranceEdge, + RiskAssuranceNode, + RiskSanctionsListSnapshot, + RiskScreeningCandidate, + RiskScreeningDisposition, + RiskScreeningException, + RiskScreeningRun, + RiskScreeningSubjectSnapshot, +) + + +RISK_COMPLIANCE_DSAR_CAPABILITY = dsar_capability_name("risk_compliance") +_MAX_RECORDS = 5_000 +_MAX_SUBJECT_ITEMS = 100 +_MAX_SUBJECT_BYTES = 256 * 1024 +_CONFLICT = object() + + +@dataclass(frozen=True, slots=True) +class _SubjectSelectors: + account_id: str | None + membership_id: str | None + subject_ref: str | None + screening_id: str | None + assurance_node_id: str | None + assurance_edge_id: str | None + + @property + def actor_ids(self) -> tuple[str, ...]: + return tuple( + dict.fromkeys( + value for value in (self.account_id, self.membership_id) if value + ) + ) + + @property + def narrowed(self) -> bool: + return bool( + self.screening_id or self.assurance_node_id or self.assurance_edge_id + ) + + +class RiskComplianceDsarProvider: + provider_id = "risk_compliance" + module_id = "risk_compliance" + + 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 selectors.subject_ref and (not selectors.narrowed or selectors.screening_id): + query = ( + db.query(RiskScreeningRun, RiskScreeningSubjectSnapshot) + .join( + RiskScreeningSubjectSnapshot, + RiskScreeningSubjectSnapshot.id + == RiskScreeningRun.subject_snapshot_id, + ) + .filter( + RiskScreeningRun.tenant_id == tenant_id, + RiskScreeningSubjectSnapshot.tenant_id == tenant_id, + RiskScreeningSubjectSnapshot.subject_ref == selectors.subject_ref, + ) + ) + if selectors.screening_id: + query = query.filter(RiskScreeningRun.id == selectors.screening_id) + records.extend( + _subject_screening_record(run, snapshot) + for run, snapshot in _limited( + query, + RiskScreeningRun.started_at, + RiskScreeningRun.id, + label="subject screening", + ) + ) + + actor_ids = selectors.actor_ids + if actor_ids and (not selectors.narrowed or selectors.screening_id): + run_query = ( + db.query(RiskScreeningRun, RiskScreeningSubjectSnapshot) + .join( + RiskScreeningSubjectSnapshot, + RiskScreeningSubjectSnapshot.id + == RiskScreeningRun.subject_snapshot_id, + ) + .filter( + RiskScreeningRun.tenant_id == tenant_id, + RiskScreeningSubjectSnapshot.tenant_id == tenant_id, + or_( + RiskScreeningRun.created_by.in_(actor_ids), + RiskScreeningSubjectSnapshot.submitted_by.in_(actor_ids), + ), + ) + ) + if selectors.screening_id: + run_query = run_query.filter( + RiskScreeningRun.id == selectors.screening_id + ) + records.extend( + _screening_actor_record(run, snapshot, actor_ids) + for run, snapshot in _limited( + run_query, + RiskScreeningRun.started_at, + RiskScreeningRun.id, + label="screening actor attribution", + ) + ) + + disposition_query = db.query(RiskScreeningDisposition).filter( + RiskScreeningDisposition.tenant_id == tenant_id, + or_( + RiskScreeningDisposition.actor_account_id.in_(actor_ids), + RiskScreeningDisposition.actor_membership_id.in_(actor_ids), + ), + ) + if selectors.screening_id: + disposition_query = ( + db.query(RiskScreeningDisposition) + .join( + RiskScreeningCandidate, + RiskScreeningCandidate.id + == RiskScreeningDisposition.candidate_id, + ) + .filter( + RiskScreeningDisposition.tenant_id == tenant_id, + RiskScreeningCandidate.run_id == selectors.screening_id, + or_( + RiskScreeningDisposition.actor_account_id.in_(actor_ids), + RiskScreeningDisposition.actor_membership_id.in_(actor_ids), + ), + ) + ) + records.extend( + _disposition_actor_record(row) + for row in _limited( + disposition_query, + RiskScreeningDisposition.created_at, + RiskScreeningDisposition.id, + label="disposition attribution", + ) + ) + + if actor_ids and not selectors.narrowed: + records.extend(self._unnarrowed_actor_records(db, tenant_id, actor_ids)) + + if actor_ids and selectors.assurance_node_id: + query = db.query(RiskAssuranceNode).filter( + RiskAssuranceNode.tenant_id == tenant_id, + RiskAssuranceNode.id == selectors.assurance_node_id, + RiskAssuranceNode.created_by.in_(actor_ids), + ) + records.extend( + _assurance_node_actor_record(row) + for row in _limited( + query, + RiskAssuranceNode.recorded_at, + RiskAssuranceNode.id, + label="assurance-node attribution", + ) + ) + + if actor_ids and selectors.assurance_edge_id: + query = db.query(RiskAssuranceEdge).filter( + RiskAssuranceEdge.tenant_id == tenant_id, + RiskAssuranceEdge.id == selectors.assurance_edge_id, + RiskAssuranceEdge.created_by.in_(actor_ids), + ) + records.extend( + _assurance_edge_actor_record(row) + for row in _limited( + query, + RiskAssuranceEdge.recorded_at, + RiskAssuranceEdge.id, + label="assurance-edge attribution", + ) + ) + + if len(records) > _MAX_RECORDS: + raise ValueError( + "Risk Compliance DSAR result limit exceeded; narrow selectors." + ) + return tuple( + sorted(records, key=lambda item: (item.resource_type, item.resource_id)) + ) + + @staticmethod + def _unnarrowed_actor_records( + db: Session, + tenant_id: str, + actor_ids: tuple[str, ...], + ) -> list[DsarRecordRef]: + records: list[DsarRecordRef] = [] + imports = db.query(RiskSanctionsListSnapshot).filter( + RiskSanctionsListSnapshot.tenant_id == tenant_id, + RiskSanctionsListSnapshot.imported_by.in_(actor_ids), + ) + records.extend( + _snapshot_import_actor_record(row) + for row in _limited( + imports, + RiskSanctionsListSnapshot.imported_at, + RiskSanctionsListSnapshot.id, + label="snapshot-import attribution", + ) + ) + exceptions = db.query(RiskScreeningException).filter( + RiskScreeningException.tenant_id == tenant_id, + RiskScreeningException.created_by.in_(actor_ids), + ) + records.extend( + _exception_actor_record(row) + for row in _limited( + exceptions, + RiskScreeningException.created_at, + RiskScreeningException.id, + label="exception attribution", + ) + ) + nodes = db.query(RiskAssuranceNode).filter( + RiskAssuranceNode.tenant_id == tenant_id, + RiskAssuranceNode.created_by.in_(actor_ids), + ) + records.extend( + _assurance_node_actor_record(row) + for row in _limited( + nodes, + RiskAssuranceNode.recorded_at, + RiskAssuranceNode.id, + label="assurance-node attribution", + ) + ) + edges = db.query(RiskAssuranceEdge).filter( + RiskAssuranceEdge.tenant_id == tenant_id, + RiskAssuranceEdge.created_by.in_(actor_ids), + ) + records.extend( + _assurance_edge_actor_record(row) + for row in _limited( + edges, + RiskAssuranceEdge.recorded_at, + RiskAssuranceEdge.id, + label="assurance-edge attribution", + ) + ) + return records + + 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("Risk Compliance DSAR subject selectors conflict.") + actions: list[DsarErasureActionRef] = [] + for record in records: + _validate_record(record) + actions.append( + DsarErasureActionRef( + action_id=( + f"risk_compliance:retain:{record.resource_type}:" + f"{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 "Risk and compliance evidence remains 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("Risk Compliance DSAR subject selectors conflict.") + results: list[DsarExecutionResultRef] = [] + for action in actions: + _validate_action(action) + if action.executable or action.kind != "retain": + raise ValueError("Risk Compliance DSAR publishes retain actions only.") + results.append( + DsarExecutionResultRef( + action_id=action.action_id, + status="blocked", + summary=( + "Risk and compliance evidence remains unchanged under its " + "legal, audit, and accountability obligations." + ), + 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("risk_compliance.account"), + references.get("access.account"), + ), + "membership_id": _coalesce( + subject.membership_id, + references.get("risk_compliance.membership"), + references.get("tenancy.membership"), + ), + "subject_ref": _coalesce( + references.get("risk_compliance.subject"), + references.get("risk_compliance.subject_ref"), + ), + "screening_id": _coalesce( + references.get("risk_compliance.screening"), + references.get("risk_compliance.screening_id"), + ), + "assurance_node_id": _coalesce( + references.get("risk_compliance.assurance_node"), + references.get("risk_compliance.assurance_node_id"), + ), + "assurance_edge_id": _coalesce( + references.get("risk_compliance.assurance_edge"), + references.get("risk_compliance.assurance_edge_id"), + ), + } + if any(value is _CONFLICT for value in values.values()): + return None + selectors = _SubjectSelectors( + account_id=_optional_string(values["account_id"]), + membership_id=_optional_string(values["membership_id"]), + subject_ref=_optional_string(values["subject_ref"]), + screening_id=_optional_string(values["screening_id"]), + assurance_node_id=_optional_string(values["assurance_node_id"]), + assurance_edge_id=_optional_string(values["assurance_edge_id"]), + ) + if not selectors.actor_ids and not selectors.subject_ref: + return None + return selectors + + +def _subject_screening_record( + run: RiskScreeningRun, snapshot: RiskScreeningSubjectSnapshot +) -> DsarRecordRef: + data = { + "screening_id": run.id, + "subject_snapshot_id": snapshot.id, + "subject_ref": snapshot.subject_ref, + "subject_type": snapshot.subject_type, + "primary_name": (snapshot.primary_name or "")[:1_000] or None, + "aliases": _string_list(snapshot.aliases, 1_000), + "identifiers": _mapping_list( + snapshot.identifiers, + allowed=("type", "value"), + limits={"type": 100, "value": 1_000}, + ), + "dates": _string_list(snapshot.dates, 100), + "addresses": _mapping_list( + snapshot.addresses, + allowed=("street", "city", "region", "postal_code", "country"), + limits={ + "street": 1_000, + "city": 500, + "region": 500, + "postal_code": 100, + "country": 255, + }, + ), + "status": run.status, + "outcome": run.outcome, + "candidate_count": run.candidate_count, + "started_at": _iso(run.started_at), + "completed_at": _iso(run.completed_at), + } + _bounded_json(data) + return _record( + resource_type="screening_subject_submission", + resource_id=run.id, + category="sanctions_screening_subject_data", + title="Sanctions screening subject submission", + data=data, + observed_at=run.completed_at or run.started_at, + retention_reason=( + "Version-pinned screening inputs and outcomes are retained as legal and " + "compliance evidence." + ), + ) + + +def _screening_actor_record( + run: RiskScreeningRun, + snapshot: RiskScreeningSubjectSnapshot, + actor_ids: tuple[str, ...], +) -> DsarRecordRef: + activities = [] + if run.created_by in actor_ids: + activities.append("created_screening") + if snapshot.submitted_by in actor_ids: + activities.append("submitted_screening_subject") + return _record( + resource_type="screening_actor_attribution", + resource_id=run.id, + category="risk_compliance_actor_attribution", + title="Screening actor attribution", + data={ + "screening_id": run.id, + "status": run.status, + "outcome": run.outcome, + "candidate_count": run.candidate_count, + "activities": activities, + "started_at": _iso(run.started_at), + "completed_at": _iso(run.completed_at), + }, + observed_at=run.completed_at or run.started_at, + ) + + +def _snapshot_import_actor_record(row: RiskSanctionsListSnapshot) -> DsarRecordRef: + return _record( + resource_type="snapshot_import_actor_attribution", + resource_id=row.id, + category="risk_compliance_actor_attribution", + title="Sanctions snapshot import attribution", + data={ + "snapshot_id": row.id, + "provider_id": row.provider_id, + "source_id": row.source_id, + "source_version": row.source_version, + "status": row.status, + "entry_count": row.entry_count, + "activity": "imported_sanctions_snapshot", + "imported_at": _iso(row.imported_at), + }, + observed_at=row.imported_at, + ) + + +def _disposition_actor_record(row: RiskScreeningDisposition) -> DsarRecordRef: + return _record( + resource_type="disposition_actor_attribution", + resource_id=row.id, + category="risk_compliance_actor_attribution", + title="Screening disposition actor attribution", + data={ + "disposition_id": row.id, + "candidate_id": row.candidate_id, + "decision": row.decision, + "scope": row.scope, + "separation_status": row.separation_status, + "expires_at": _iso(row.expires_at), + "review_at": _iso(row.review_at), + "activity": "recorded_screening_disposition", + "created_at": _iso(row.created_at), + }, + observed_at=row.created_at, + ) + + +def _exception_actor_record(row: RiskScreeningException) -> DsarRecordRef: + return _record( + resource_type="exception_actor_attribution", + resource_id=row.id, + category="risk_compliance_actor_attribution", + title="Screening exception actor attribution", + data={ + "exception_id": row.id, + "scope": row.scope, + "status": row.status, + "starts_at": _iso(row.starts_at), + "expires_at": _iso(row.expires_at), + "review_at": _iso(row.review_at), + "activity": "created_screening_exception", + }, + observed_at=row.created_at, + ) + + +def _assurance_node_actor_record(row: RiskAssuranceNode) -> DsarRecordRef: + return _record( + resource_type="assurance_node_actor_attribution", + resource_id=row.id, + category="risk_compliance_actor_attribution", + title="Assurance-object actor attribution", + data={ + "assurance_node_id": row.id, + "stable_id": row.stable_id, + "kind": row.kind, + "revision": row.revision, + "state": row.state, + "valid_from": _iso(row.valid_from), + "valid_to": _iso(row.valid_to), + "recorded_at": _iso(row.recorded_at), + "superseded_at": _iso(row.superseded_at), + "activity": "created_assurance_object_revision", + }, + observed_at=row.recorded_at, + ) + + +def _assurance_edge_actor_record(row: RiskAssuranceEdge) -> DsarRecordRef: + return _record( + resource_type="assurance_edge_actor_attribution", + resource_id=row.id, + category="risk_compliance_actor_attribution", + title="Assurance-relation actor attribution", + data={ + "assurance_edge_id": row.id, + "stable_id": row.stable_id, + "revision": row.revision, + "relation": row.relation, + "state": row.state, + "valid_from": _iso(row.valid_from), + "valid_to": _iso(row.valid_to), + "recorded_at": _iso(row.recorded_at), + "superseded_at": _iso(row.superseded_at), + "activity": "created_assurance_relation_revision", + }, + observed_at=row.recorded_at, + ) + + +def _record( + *, + resource_type: str, + resource_id: str, + category: str, + title: str, + data: Mapping[str, object], + observed_at: datetime | None, + retention_reason: str | None = None, +) -> DsarRecordRef: + return DsarRecordRef( + provider_id="risk_compliance", + module_id="risk_compliance", + resource_type=resource_type, + resource_id=resource_id, + category=category, + title=title, + data=data, + observed_at=_aware(observed_at), + immutable_evidence=True, + retention_reason=( + retention_reason + or "Risk and compliance attribution is retained for legal, audit, and accountability evidence." + ), + ) + + +def _string_list(value: object, item_limit: int) -> list[str]: + if not isinstance(value, list) or len(value) > _MAX_SUBJECT_ITEMS: + raise ValueError("Risk Compliance DSAR subject list exceeds its bound.") + return [str(item)[:item_limit] for item in value] + + +def _mapping_list( + value: object, + *, + allowed: tuple[str, ...], + limits: Mapping[str, int], +) -> list[dict[str, str]]: + if not isinstance(value, list) or len(value) > _MAX_SUBJECT_ITEMS: + raise ValueError("Risk Compliance DSAR subject mapping list exceeds its bound.") + result: list[dict[str, str]] = [] + for item in value: + if not isinstance(item, Mapping): + raise ValueError("Risk Compliance DSAR subject mapping is invalid.") + result.append( + { + key: str(item[key])[: limits[key]] + for key in allowed + if item.get(key) is not None + } + ) + return result + + +def _bounded_json(value: object) -> None: + try: + encoded = json.dumps(value, ensure_ascii=False, sort_keys=True).encode("utf-8") + except (TypeError, ValueError) as exc: + raise ValueError("Risk Compliance DSAR subject data is invalid.") from exc + if len(encoded) > _MAX_SUBJECT_BYTES: + raise ValueError("Risk Compliance DSAR subject data exceeds its byte bound.") + + +def _limited(query, first, second, *, label: str): + rows = query.order_by(first, second).limit(_MAX_RECORDS + 1).all() + if len(rows) > _MAX_RECORDS: + raise ValueError( + f"Risk Compliance DSAR {label} limit exceeded; narrow selectors." + ) + return rows + + +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 _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("Risk Compliance DSAR requires a SQLAlchemy Session.") + return value + + +_RESOURCE_TYPES = { + "screening_subject_submission", + "screening_actor_attribution", + "snapshot_import_actor_attribution", + "disposition_actor_attribution", + "exception_actor_attribution", + "assurance_node_actor_attribution", + "assurance_edge_actor_attribution", +} + + +def _validate_record(record: DsarRecordRef) -> None: + if record.provider_id != "risk_compliance" or record.module_id != "risk_compliance": + raise ValueError("Risk Compliance DSAR cannot plan a foreign provider record.") + if record.resource_type not in _RESOURCE_TYPES or not record.resource_id: + raise ValueError("Risk Compliance DSAR record identity is invalid.") + + +def _validate_action(action: DsarErasureActionRef) -> None: + if action.provider_id != "risk_compliance" or action.module_id != "risk_compliance": + raise ValueError( + "Risk Compliance DSAR cannot execute a foreign provider action." + ) + if not action.action_id.startswith("risk_compliance:retain:"): + raise ValueError("Risk Compliance DSAR action identity is invalid.") + + +__all__ = ["RISK_COMPLIANCE_DSAR_CAPABILITY", "RiskComplianceDsarProvider"] diff --git a/src/govoplan_risk_compliance/backend/manifest.py b/src/govoplan_risk_compliance/backend/manifest.py index 9d80fd2..3f6510f 100644 --- a/src/govoplan_risk_compliance/backend/manifest.py +++ b/src/govoplan_risk_compliance/backend/manifest.py @@ -11,6 +11,7 @@ from govoplan_core.core.module_guards import ( persistent_table_uninstall_guard, ) from govoplan_core.core.modules import ( + CapabilityDocumentation, DocumentationLink, DocumentationTopic, FrontendModule, @@ -50,6 +51,10 @@ from govoplan_risk_compliance.backend.db.models import ( RiskScreeningRun, RiskScreeningSubjectSnapshot, ) +from govoplan_risk_compliance.backend.dsar_provider import ( + RISK_COMPLIANCE_DSAR_CAPABILITY, + RiskComplianceDsarProvider, +) from govoplan_risk_compliance.backend.permissions import ( ADMIN_SCOPE, READ_SCOPE, @@ -269,6 +274,10 @@ def _assurance_search_source(context): return create_risk_assurance_search_source(context) +def _dsar_provider(_context) -> RiskComplianceDsarProvider: + return RiskComplianceDsarProvider() + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "risk_sanctions_list_snapshots": ( @@ -339,9 +348,7 @@ DOCUMENTATION = ( ), DocumentationLink( label="Interface pattern migration", - href=( - "govoplan-risk-compliance/docs/INTERFACE_PATTERN_MIGRATION.md" - ), + href=("govoplan-risk-compliance/docs/INTERFACE_PATTERN_MIGRATION.md"), kind="repository", ), ), @@ -412,6 +419,10 @@ manifest = ModuleManifest( name="risk_compliance.sanctions_screening", version="1.0.0", ), + ModuleInterfaceProvider( + name=RISK_COMPLIANCE_DSAR_CAPABILITY, + version="0.1.0", + ), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -426,6 +437,17 @@ manifest = ModuleManifest( route_factory=_route_factory, capability_factories={ CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING: (_sanctions_screening_provider), + RISK_COMPLIANCE_DSAR_CAPABILITY: _dsar_provider, + }, + capability_documentation={ + RISK_COMPLIANCE_DSAR_CAPABILITY: CapabilityDocumentation( + label="Risk Compliance data-subject request provider", + summary=( + "Exports verified screening-subject data and minimized compliance " + "attribution while protecting third-party and legal-review evidence." + ), + contract_version="0.1.0", + ), }, search_sources=( SearchSourceProviderRegistration( @@ -565,7 +587,53 @@ manifest = ModuleManifest( label="Risk Compliance", ), ), - documentation=DOCUMENTATION, + documentation=( + DocumentationTopic( + id="risk_compliance.data-subject-requests", + title="Risk and compliance data-subject requests", + summary=( + "Export verified screening-subject data and accountable activity " + "without disclosing third-party sanctions or review evidence." + ), + body=( + "Risk Compliance correlates screening subject data only through an " + "exact, separately verified subject reference. An account or membership " + "identifier independently locates the subject's own operator, reviewer, " + "import, exception, and assurance-graph attribution. Searches can narrow " + "to a screening or assurance revision, but an object identifier alone " + "never establishes identity. Subject exports include bounded submitted " + "names, aliases, identifiers, dates, addresses, and the screening " + "lifecycle outcome. They exclude sanctions-entry data about third " + "parties, candidate matching evidence, fingerprints, hashes, policy " + "snapshots, reviewer reasons, authority context, provenance, and evidence " + "references. Version-pinned screenings, dispositions, exceptions, and " + "assurance revisions remain retained legal and accountability evidence." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("user", "operator", "module_admin", "auditor"), + related_modules=("core", "access", "audit", "records", "policy"), + order=90, + metadata={ + "help_contexts": [ + "risk_compliance.sanctions.screening", + "privacy.data-subject-requests", + ], + "consequence_classes": { + "export_screening_subject": ( + "Returns bounded subject input and lifecycle data for an exact reference." + ), + "exclude_third_party_evidence": ( + "Never returns sanctions entries, match evidence, or protected review payloads." + ), + "retain_compliance_evidence": ( + "Preserves version-pinned legal, audit, and accountability history." + ), + }, + }, + ), + *DOCUMENTATION, + ), ) diff --git a/tests/test_dsar_provider.py b/tests/test_dsar_provider.py new file mode 100644 index 0000000..64adb19 --- /dev/null +++ b/tests/test_dsar_provider.py @@ -0,0 +1,468 @@ +from __future__ import annotations + +import json +import unittest +from datetime import UTC, datetime, timedelta + +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_core.privacy.dsar_workflow import ( + create_data_subject_request, + search_data_subject_request, +) +from govoplan_risk_compliance.backend.db.models import ( + RiskAssuranceEdge, + RiskAssuranceNode, + RiskSanctionsEntry, + RiskSanctionsListSnapshot, + RiskScreeningCandidate, + RiskScreeningDisposition, + RiskScreeningException, + RiskScreeningRun, + RiskScreeningSubjectSnapshot, +) +from govoplan_risk_compliance.backend.dsar_provider import ( + RISK_COMPLIANCE_DSAR_CAPABILITY, + RiskComplianceDsarProvider, +) +from govoplan_risk_compliance.backend.manifest import manifest + + +NOW = datetime(2026, 8, 22, 10, 0, tzinfo=UTC) + + +class _Registry: + def __init__(self, provider: RiskComplianceDsarProvider) -> None: + self.provider = provider + + def capability_names(self): + return (RISK_COMPLIANCE_DSAR_CAPABILITY,) + + def capability_owner(self, name): + if name != RISK_COMPLIANCE_DSAR_CAPABILITY: + raise KeyError(name) + return "risk_compliance" + + def tenant_entitlement_resolver(self): + class _Resolver: + @staticmethod + def resolve(session, tenant_id): + del session, tenant_id + return type("State", (), {"effective_modules": ("risk_compliance",)})() + + return _Resolver() + + def require_tenant_capability(self, name, session, **kwargs): + del session, kwargs + if name != RISK_COMPLIANCE_DSAR_CAPABILITY: + raise KeyError(name) + return self.provider + + def manifests(self): + return (type("Manifest", (), {"id": "risk_compliance"})(),) + + +class RiskComplianceDsarProviderTests(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 = RiskComplianceDsarProvider() + self.assertIsInstance(self.provider, DsarProvider) + self._seed() + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def _seed(self) -> None: + list_snapshot = RiskSanctionsListSnapshot( + id="list-1", + tenant_id="tenant-1", + visibility="tenant", + connector_snapshot_ref="connector:snapshot:secret-do-not-export", + provider_id="un", + publisher="United Nations", + jurisdiction="global", + list_type="sanctions", + source_id="consolidated", + source_version="2026-08-22", + publication_at=NOW, + effective_at=NOW, + acquired_at=NOW, + sha256="list-sha-do-not-export", + connector_run_id="connector-run-do-not-export", + raw_evidence_ref="evidence-ref-do-not-export", + source_parser_version="parser-v1", + normalization_version="normalizer-v1", + signature_evidence={"secret": "signature-do-not-export"}, + provenance={"secret": "provenance-do-not-export"}, + entry_count=1, + status="active", + imported_by="account-1", + imported_at=NOW, + created_at=NOW, + updated_at=NOW, + ) + subject_snapshot = RiskScreeningSubjectSnapshot( + id="subject-snapshot-1", + tenant_id="tenant-1", + subject_ref="party:person-1", + subject_type="person", + primary_name="Ada Example", + normalized_name="ada example normalized-do-not-export", + aliases=["Ada E."], + identifiers=[ + { + "type": "resident-number", + "value": "resident-123", + "unexpected": "identifier-extra-do-not-export", + } + ], + dates=["1990-01-01"], + addresses=[ + { + "street": "Example Street 1", + "city": "Exampletown", + "country": "DE", + "unexpected": "address-extra-do-not-export", + } + ], + fingerprint="subject-fingerprint-do-not-export", + submitted_by="account-1", + created_at=NOW, + updated_at=NOW, + ) + run = RiskScreeningRun( + id="screening-1", + tenant_id="tenant-1", + subject_snapshot_id="subject-snapshot-1", + list_snapshot_id="list-1", + idempotency_key="screening-idempotency-do-not-export", + request_hash="screening-request-hash-do-not-export", + matcher_version="matcher-v1", + normalization_version="normalizer-v1", + policy_version="policy-v1", + policy_snapshot={"secret": "policy-snapshot-do-not-export"}, + status="complete", + outcome="review", + candidate_count=1, + started_at=NOW, + completed_at=NOW, + created_by="account-1", + created_at=NOW, + updated_at=NOW, + ) + sanctions_entry = RiskSanctionsEntry( + id="entry-1", + snapshot_id="list-1", + source_entry_id="third-party-entry-do-not-export", + subject_type="person", + primary_name="Third Party Name Do Not Export", + normalized_name="third party", + original_script_name=None, + reference_number="third-party-reference-do-not-export", + listed_on=None, + programmes=[], + measures=[], + raw_evidence_locator="third-party-evidence-do-not-export", + details={"secret": "third-party-details-do-not-export"}, + created_at=NOW, + updated_at=NOW, + ) + candidate = RiskScreeningCandidate( + id="candidate-1", + tenant_id="tenant-1", + run_id="screening-1", + entry_id="entry-1", + score=91, + match_kind="fuzzy", + evidence=[{"secret": "candidate-evidence-do-not-export"}], + review_status="confirmed", + current_disposition_id="disposition-1", + created_at=NOW, + updated_at=NOW, + ) + disposition = RiskScreeningDisposition( + id="disposition-1", + tenant_id="tenant-1", + candidate_id="candidate-1", + decision="false_positive", + reason="review-reason-do-not-export", + evidence_refs=["review-evidence-do-not-export"], + scope="subject_entry", + expires_at=NOW + timedelta(days=30), + review_at=NOW + timedelta(days=15), + actor_account_id="account-1", + actor_membership_id="membership-1", + actor_authority={"secret": "authority-do-not-export"}, + separation_status="independent", + override_reason="override-reason-do-not-export", + created_at=NOW, + ) + exception = RiskScreeningException( + id="exception-1", + tenant_id="tenant-1", + subject_fingerprint="exception-fingerprint-do-not-export", + source_entry_ref="exception-source-entry-do-not-export", + scope="subject_entry", + status="active", + reason="exception-reason-do-not-export", + evidence_refs=["exception-evidence-do-not-export"], + starts_at=NOW, + expires_at=NOW + timedelta(days=30), + review_at=NOW + timedelta(days=15), + originating_disposition_id="disposition-1", + created_by="account-1", + created_at=NOW, + updated_at=NOW, + ) + node = RiskAssuranceNode( + id="node-1", + tenant_id="tenant-1", + stable_id="control-1", + kind="control", + revision=1, + label="Sensitive control label do not export", + description="Sensitive control description do not export", + state="active", + owner_ref="owner-secret-do-not-export", + scope_ref="scope-secret-do-not-export", + governed_object_ref="object-secret-do-not-export", + valid_from=NOW, + recorded_at=NOW, + provenance={"secret": "node-provenance-do-not-export"}, + legal_basis_refs=["legal-secret-do-not-export"], + policy_refs=["policy-secret-do-not-export"], + evidence_refs=["node-evidence-do-not-export"], + classification="restricted", + created_by="account-1", + created_at=NOW, + updated_at=NOW, + ) + edge = RiskAssuranceEdge( + id="edge-1", + tenant_id="tenant-1", + stable_id="relation-1", + revision=1, + source_node_ref="control-1", + target_node_ref="risk-1", + relation="mitigates", + state="active", + owner_ref="edge-owner-secret-do-not-export", + scope_ref="edge-scope-secret-do-not-export", + valid_from=NOW, + recorded_at=NOW, + provenance={"secret": "edge-provenance-do-not-export"}, + legal_basis_refs=["edge-legal-secret-do-not-export"], + policy_refs=["edge-policy-secret-do-not-export"], + evidence_refs=["edge-evidence-do-not-export"], + created_by="account-1", + created_at=NOW, + updated_at=NOW, + ) + other_tenant_node = RiskAssuranceNode( + id="node-other", + tenant_id="tenant-2", + stable_id="other-control", + kind="control", + revision=1, + label="Other tenant", + state="active", + owner_ref="owner", + valid_from=NOW, + recorded_at=NOW, + provenance={}, + legal_basis_refs=[], + policy_refs=[], + evidence_refs=[], + classification="internal", + created_by="account-1", + ) + self.session.add_all( + ( + list_snapshot, + subject_snapshot, + run, + sanctions_entry, + candidate, + disposition, + exception, + node, + edge, + other_tenant_node, + ) + ) + + @staticmethod + def _subject() -> DsarSubjectRef: + return DsarSubjectRef( + account_id="account-1", + membership_id="membership-1", + external_references={ + "risk_compliance.subject": "party:person-1", + }, + ) + + def test_search_exports_subject_data_and_minimized_attribution(self) -> None: + records = self.provider.search_subject( + self.session, tenant_id="tenant-1", subject=self._subject() + ) + self.assertEqual( + { + "screening_subject_submission", + "screening_actor_attribution", + "snapshot_import_actor_attribution", + "disposition_actor_attribution", + "exception_actor_attribution", + "assurance_node_actor_attribution", + "assurance_edge_actor_attribution", + }, + {record.resource_type for record in records}, + ) + exported = json.dumps([record.to_dict() for record in records]) + self.assertIn("Ada Example", exported) + self.assertIn("resident-123", exported) + for excluded in ( + "normalized-do-not-export", + "identifier-extra-do-not-export", + "address-extra-do-not-export", + "subject-fingerprint-do-not-export", + "screening-idempotency-do-not-export", + "screening-request-hash-do-not-export", + "policy-snapshot-do-not-export", + "Third Party Name Do Not Export", + "third-party-reference-do-not-export", + "candidate-evidence-do-not-export", + "review-reason-do-not-export", + "review-evidence-do-not-export", + "authority-do-not-export", + "override-reason-do-not-export", + "exception-fingerprint-do-not-export", + "exception-source-entry-do-not-export", + "exception-reason-do-not-export", + "Sensitive control label do not export", + "node-provenance-do-not-export", + "edge-owner-secret-do-not-export", + "edge-evidence-do-not-export", + "list-sha-do-not-export", + "signature-do-not-export", + ): + self.assertNotIn(excluded, exported) + + def test_subject_data_requires_exact_module_reference(self) -> None: + account_only = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef(account_id="account-1"), + ) + reference_only = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + external_references={ + "risk_compliance.subject": "party:person-1", + } + ), + ) + self.assertNotIn( + "screening_subject_submission", + {record.resource_type for record in account_only}, + ) + self.assertEqual( + {"screening_subject_submission"}, + {record.resource_type for record in reference_only}, + ) + + def test_conflicts_narrowing_and_tenant_boundaries(self) -> None: + conflict = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={ + "risk_compliance.account": "account-other", + }, + ), + ) + narrowed = self.provider.search_subject( + self.session, + tenant_id="tenant-1", + subject=DsarSubjectRef( + account_id="account-1", + external_references={ + "risk_compliance.subject": "party:person-1", + "risk_compliance.screening": "screening-1", + }, + ), + ) + full = self.provider.search_subject( + self.session, tenant_id="tenant-1", subject=self._subject() + ) + self.assertEqual((), conflict) + self.assertEqual( + { + "screening_subject_submission", + "screening_actor_attribution", + "disposition_actor_attribution", + }, + {record.resource_type for record in narrowed}, + ) + self.assertNotIn("node-other", {record.resource_id for record in full}) + + def test_erasure_retains_legal_and_accountability_evidence(self) -> None: + records = self.provider.search_subject( + self.session, tenant_id="tenant-1", subject=self._subject() + ) + actions = self.provider.plan_erasure( + self.session, + tenant_id="tenant-1", + subject=self._subject(), + records=records, + ) + self.assertTrue(actions) + self.assertTrue( + all(action.kind == "retain" and not action.executable for action in actions) + ) + results = self.provider.execute_erasure( + self.session, + tenant_id="tenant-1", + subject=self._subject(), + actions=actions, + request_id="dsar-risk-1", + ) + self.assertTrue(all(result.status == "blocked" for result in results)) + + def test_manifest_and_core_workflow_discover_provider(self) -> None: + self.assertIn(RISK_COMPLIANCE_DSAR_CAPABILITY, manifest.capability_factories) + self.assertIn( + "risk_compliance.data-subject-requests", + {topic.id for topic in manifest.documentation}, + ) + row = create_data_subject_request( + self.session, + tenant_id="tenant-1", + reference="DSAR-RISK-1", + request_kind="access", + subject=self._subject(), + purpose="Risk screening access request", + legal_basis=None, + due_at=None, + requested_by_account_id="operator-1", + ) + search_data_subject_request( + self.session, + registry=_Registry(self.provider), + row=row, + expected_revision=row.resource_revision, + ) + self.assertEqual("searched", row.status) + self.assertEqual(7, row.search_result["record_count"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index e7f0667..c271df3 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -6,6 +6,9 @@ from govoplan_core.core.sanctions import ( CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING, SanctionsScreeningProvider, ) +from govoplan_risk_compliance.backend.dsar_provider import ( + RISK_COMPLIANCE_DSAR_CAPABILITY, +) from govoplan_risk_compliance.backend.manifest import ( ADMIN_SCOPE, READ_SCOPE, @@ -27,10 +30,7 @@ class ManifestTests(unittest.TestCase): self.assertEqual(manifest.dependencies, ("access",)) self.assertIn("connectors", manifest.optional_dependencies) self.assertEqual( - { - permission.scope - for permission in manifest.permissions - }, + {permission.scope for permission in manifest.permissions}, { READ_SCOPE, WRITE_SCOPE, @@ -59,11 +59,11 @@ class ManifestTests(unittest.TestCase): ) self.assertTrue(manifest.requires_interfaces[0].optional) self.assertEqual( - {"risk_compliance.sanctions_screening"}, { - item.name - for item in manifest.provides_interfaces + "risk_compliance.sanctions_screening", + RISK_COMPLIANCE_DSAR_CAPABILITY, }, + {item.name for item in manifest.provides_interfaces}, ) capability = manifest.capability_factories[ CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING