From da1baede36c4af5db7c3a75520f385897b3db66b Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 30 Jul 2026 01:55:03 +0200 Subject: [PATCH] feat: reconcile sanctions screening freshness --- AGENTS.md | 2 +- README.md | 9 +- docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md | 43 +- .../backend/capabilities.py | 137 ++++++ .../backend/manifest.py | 23 + .../backend/router.py | 134 +++++- .../backend/schemas.py | 39 ++ .../backend/screening.py | 437 ++++++++++++++++++ tests/test_manifest.py | 15 + tests/test_sanctions_screening.py | 123 +++++ 10 files changed, 926 insertions(+), 36 deletions(-) create mode 100644 src/govoplan_risk_compliance/backend/capabilities.py diff --git a/AGENTS.md b/AGENTS.md index 6e79538..d31def9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,5 +20,5 @@ Use Gitea issues as the canonical backlog and state log. The shared workflow is Focused verification: ```bash -PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src /mnt/DATA/git/govoplan-core/.venv/bin/python -m unittest discover -s tests +PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src /mnt/DATA/git/govoplan/.venv/bin/python -m unittest discover -s tests ``` diff --git a/README.md b/README.md index 2d5f041..c3cd5aa 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,11 @@ Its runtime module ID is `risk_compliance`; the repository and Python distributi The first complete vertical is sanctions screening. Connectors acquires immutable source evidence; Risk Compliance imports and normalizes exact list versions, runs deterministic version-pinned screening, and presents potential matches to an independent reviewer. Fuzzy matching only creates review candidates and never confirms a legal match. -The module includes database migrations, tenant-isolated APIs, an operational WebUI, append-only dispositions, time-bounded false-positive exceptions, and audit events. Queue and audit summaries deliberately retain only the minimum subject data needed for the workflow. +The module includes database migrations, tenant-isolated APIs, an operational +WebUI, append-only dispositions, time-bounded false-positive exceptions, +freshness reconciliation, explicit consumer gates, and audit events. Queue and +audit summaries deliberately retain only the minimum subject data needed for +the workflow. ## Initial Ownership @@ -22,6 +26,7 @@ The module includes database migrations, tenant-isolated APIs, an operational We - internal-control evidence - immutable sanctions list catalogues - sanctions screening and reviewer dispositions +- stable screening evidence references and freshness gates ## Boundaries @@ -59,7 +64,7 @@ Focused manifest verification: ```bash cd /mnt/DATA/git/govoplan-risk-compliance -PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src /mnt/DATA/git/govoplan-core/.venv/bin/python -m unittest discover -s tests +PYTHONPATH=src:/mnt/DATA/git/govoplan-core/src /mnt/DATA/git/govoplan/.venv/bin/python -m unittest discover -s tests ``` ## Gitea Workflow diff --git a/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md b/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md index e9771a0..db67094 100644 --- a/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md +++ b/docs/RISK_COMPLIANCE_DOMAIN_BOUNDARY.md @@ -12,6 +12,10 @@ Risk and compliance workflows for data protection incidents, DPIAs, compliance c - data protection incident records - audit measures - internal-control evidence +- immutable normalized sanctions-list snapshots +- version-pinned screening runs and candidate explanations +- reviewer dispositions, exceptions, and screening freshness +- block, review, and degraded screening-gate decisions ## Does Not Own @@ -28,20 +32,37 @@ Risk and compliance workflows for data protection incidents, DPIAs, compliance c - files - tasks - notifications +- connectors -## Seed State +## Sanctions Screening Contract -The current repository state is intentionally small: +Connectors owns acquisition and raw source evidence. Risk Compliance imports an +exact connector snapshot, preserves its provenance, normalizes the list, and +owns every legal screening and review decision made from it. -- module manifest and entry point -- tenant-level permission definitions -- manager and viewer role templates -- documentation topic describing the module boundary -- Gitea issue workflow templates -- manifest contract test +The module provides the versioned +`risk_compliance.sanctions_screening` interface and +`riskCompliance.sanctionsScreeningProvider` capability. A consumer explicitly +submits: -No runtime API, database model, migration, WebUI route, or navigation item is registered yet. The first implementation slice should preserve the boundary above and only add user-visible surfaces once the workflow model is clear. +- an immutable list snapshot +- an idempotency key +- the minimum subject data required for comparison +- matching limits and its `block`, `review`, or `degraded` failure policy -## First Implementation Slice +The response contains a stable `risk-screening:` evidence reference and +a gate decision. Consumers can later check that evidence against a current +subject, an expected or latest source snapshot, and a current policy without +importing this module's internals. -Define risk register, control, evidence, DPIA, incident, measure, and review-cycle concepts. +Freshness reasons explicitly distinguish source replacement or age, subject +changes, matcher/normalization/policy changes, incomplete outcomes, and expired +or review-due dispositions. The reconciliation API lists the latest known +screening per subject that can already be proven stale. Subject changes remain +the consuming system's responsibility because Risk Compliance deliberately +stores immutable screening-time snapshots rather than owning each source +record. + +Fuzzy candidates always require review. A confirmed match blocks; a set of +independently cleared false positives allows; source or execution uncertainty +uses the consuming module's configured failure policy. diff --git a/src/govoplan_risk_compliance/backend/capabilities.py b/src/govoplan_risk_compliance/backend/capabilities.py new file mode 100644 index 0000000..ceea245 --- /dev/null +++ b/src/govoplan_risk_compliance/backend/capabilities.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +from govoplan_core.core.sanctions import ( + SanctionsScreeningEvidence, + SanctionsScreeningFreshness, + SanctionsScreeningFreshnessRequest, + SanctionsScreeningPolicy, + SanctionsScreeningRequest, + SanctionsScreeningResult, + SanctionsScreeningSubject, +) +from govoplan_risk_compliance.backend.screening import ( + ScreeningFreshnessAssessment, + ScreeningPolicy, + ScreeningSubject, + assess_screening_freshness, + run_screening, + screening_evidence_ref, +) + + +class RiskComplianceSanctionsScreeningProvider: + def request_screening( + self, + session: object, + principal: object, + request: SanctionsScreeningRequest, + ) -> SanctionsScreeningResult: + policy = _local_policy(request.policy) + subject = _local_subject(request.subject) + run, created = run_screening( + session, + principal, + list_snapshot_id=request.list_snapshot_id, + idempotency_key=request.idempotency_key, + subject=subject, + policy=policy, + ) + assessment = assess_screening_freshness( + session, + principal, + evidence_ref=screening_evidence_ref(run.id), + current_subject=subject, + expected_list_snapshot_id=request.list_snapshot_id, + policy=policy, + failure_policy=request.policy.failure_policy, + ) + freshness = _freshness_response(assessment) + return SanctionsScreeningResult( + evidence=freshness.evidence, + freshness=freshness, + created=created, + ) + + def check_freshness( + self, + session: object, + principal: object, + request: SanctionsScreeningFreshnessRequest, + ) -> SanctionsScreeningFreshness: + assessment = assess_screening_freshness( + session, + principal, + evidence_ref=request.evidence_ref, + current_subject=( + _local_subject(request.current_subject) + if request.current_subject is not None + else None + ), + expected_list_snapshot_id=( + request.expected_list_snapshot_id + ), + policy=_local_policy(request.policy), + failure_policy=request.policy.failure_policy, + ) + return _freshness_response(assessment) + + +def _local_subject( + value: SanctionsScreeningSubject, +) -> ScreeningSubject: + return ScreeningSubject( + subject_type=value.subject_type, + primary_name=value.primary_name, + subject_ref=value.subject_ref, + aliases=tuple(value.aliases), + identifiers=tuple( + dict(item) + for item in value.identifiers + ), + dates=tuple(value.dates), + addresses=tuple( + dict(item) + for item in value.addresses + ), + ) + + +def _local_policy( + value: SanctionsScreeningPolicy, +) -> ScreeningPolicy: + return ScreeningPolicy( + fuzzy_threshold=value.fuzzy_threshold, + max_snapshot_age_days=value.max_snapshot_age_days, + max_candidates=value.max_candidates, + ) + + +def _freshness_response( + value: ScreeningFreshnessAssessment, +) -> SanctionsScreeningFreshness: + run = value.run + evidence = SanctionsScreeningEvidence( + ref=value.evidence_ref, + run_id=run.id, + outcome=run.outcome, + candidate_count=run.candidate_count, + list_snapshot_id=run.list_snapshot_id, + source_version=run.list_snapshot.source_version, + subject_fingerprint=run.subject_snapshot.fingerprint, + matcher_version=run.matcher_version, + normalization_version=run.normalization_version, + policy_version=run.policy_version, + completed_at=run.completed_at, + ) + return SanctionsScreeningFreshness( + evidence=evidence, + fresh=value.fresh, + reasons=value.reasons, + checked_at=value.checked_at, + current_list_snapshot_id=value.current_list_snapshot_id, + gate_decision=value.gate_decision, + gate_reasons=value.gate_reasons, + ) + + +__all__ = ["RiskComplianceSanctionsScreeningProvider"] diff --git a/src/govoplan_risk_compliance/backend/manifest.py b/src/govoplan_risk_compliance/backend/manifest.py index 9ffd8c4..82f439f 100644 --- a/src/govoplan_risk_compliance/backend/manifest.py +++ b/src/govoplan_risk_compliance/backend/manifest.py @@ -16,12 +16,16 @@ from govoplan_core.core.modules import ( FrontendModule, FrontendRoute, MigrationSpec, + ModuleInterfaceProvider, ModuleInterfaceRequirement, ModuleManifest, NavItem, PermissionDefinition, RoleTemplate, ) +from govoplan_core.core.sanctions import ( + CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING, +) from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base from govoplan_risk_compliance.backend.db.models import ( @@ -176,6 +180,14 @@ def _route_factory(_context): return router +def _sanctions_screening_provider(_context): + from govoplan_risk_compliance.backend.capabilities import ( + RiskComplianceSanctionsScreeningProvider, + ) + + return RiskComplianceSanctionsScreeningProvider() + + def _tenant_summary(session, tenant_id: str) -> dict[str, int]: return { "risk_sanctions_list_snapshots": ( @@ -262,6 +274,12 @@ manifest = ModuleManifest( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR, ), + provides_interfaces=( + ModuleInterfaceProvider( + name="risk_compliance.sanctions_screening", + version="1.0.0", + ), + ), requires_interfaces=( ModuleInterfaceRequirement( name="connectors.sanctions_snapshots", @@ -273,6 +291,11 @@ manifest = ModuleManifest( permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_route_factory, + capability_factories={ + CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING: ( + _sanctions_screening_provider + ), + }, frontend=FrontendModule( module_id=MODULE_ID, package_name="@govoplan/risk-compliance-webui", diff --git a/src/govoplan_risk_compliance/backend/router.py b/src/govoplan_risk_compliance/backend/router.py index 82eaa89..3f523fd 100644 --- a/src/govoplan_risk_compliance/backend/router.py +++ b/src/govoplan_risk_compliance/backend/router.py @@ -35,8 +35,11 @@ from govoplan_risk_compliance.backend.schemas import ( ListSnapshotImportResponse, ListSnapshotListResponse, ListSnapshotResponse, + RescreeningRequirementListResponse, ReviewQueueItemResponse, ReviewQueueResponse, + ScreeningFreshnessCheckRequest, + ScreeningFreshnessResponse, ScreeningRunCreateRequest, ScreeningRunCreateResponse, ScreeningRunListResponse, @@ -45,11 +48,15 @@ from govoplan_risk_compliance.backend.schemas import ( SubjectSnapshotResponse, ) from govoplan_risk_compliance.backend.screening import ( + ScreeningFreshnessAssessment, ScreeningPolicy, ScreeningSubject, + assess_screening_freshness, get_screening_run, + list_rescreening_requirements, list_screening_runs, run_screening, + screening_evidence_ref, ) @@ -217,28 +224,8 @@ def api_run_screening( principal, list_snapshot_id=payload.list_snapshot_id, idempotency_key=payload.idempotency_key, - subject=ScreeningSubject( - subject_type=payload.subject.subject_type, - subject_ref=payload.subject.subject_ref, - primary_name=payload.subject.primary_name, - aliases=tuple(payload.subject.aliases), - identifiers=tuple( - value.model_dump() - for value in payload.subject.identifiers - ), - dates=tuple(payload.subject.dates), - addresses=tuple( - value.model_dump(exclude_none=True) - for value in payload.subject.addresses - ), - ), - policy=ScreeningPolicy( - fuzzy_threshold=payload.policy.fuzzy_threshold, - max_snapshot_age_days=( - payload.policy.max_snapshot_age_days - ), - max_candidates=payload.policy.max_candidates, - ), + subject=_screening_subject(payload.subject), + policy=_screening_policy(payload.policy), ) except RiskSanctionsError as exc: raise _http_error(exc) from exc @@ -308,6 +295,66 @@ def api_get_screening( return _run_response(item) +@router.post( + "/sanctions/screenings/{run_id}/freshness", + response_model=ScreeningFreshnessResponse, +) +def api_check_screening_freshness( + run_id: str, + payload: ScreeningFreshnessCheckRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> ScreeningFreshnessResponse: + try: + item = assess_screening_freshness( + session, + principal, + evidence_ref=screening_evidence_ref(run_id), + current_subject=( + _screening_subject(payload.current_subject) + if payload.current_subject is not None + else None + ), + expected_list_snapshot_id=( + payload.expected_list_snapshot_id + ), + policy=( + _screening_policy(payload.policy) + if payload.policy is not None + else None + ), + failure_policy=payload.failure_policy, + ) + except RiskSanctionsError as exc: + raise _http_error(exc) from exc + return _freshness_response(item) + + +@router.get( + "/sanctions/rescreening-requirements", + response_model=RescreeningRequirementListResponse, +) +def api_list_rescreening_requirements( + limit: int = Query(default=100, ge=1, le=500), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> RescreeningRequirementListResponse: + try: + items = list_rescreening_requirements( + session, + principal, + limit=limit, + ) + except RiskSanctionsError as exc: + raise _http_error(exc) from exc + return RescreeningRequirementListResponse( + requirements=[ + _freshness_response(item) + for item in items + ] + ) + + @router.get( "/sanctions/review-queue", response_model=ReviewQueueResponse, @@ -456,6 +503,32 @@ def _list_snapshot_response(item) -> ListSnapshotResponse: ) +def _screening_subject(item) -> ScreeningSubject: + return ScreeningSubject( + subject_type=item.subject_type, + subject_ref=item.subject_ref, + primary_name=item.primary_name, + aliases=tuple(item.aliases), + identifiers=tuple( + value.model_dump() + for value in item.identifiers + ), + dates=tuple(item.dates), + addresses=tuple( + value.model_dump(exclude_none=True) + for value in item.addresses + ), + ) + + +def _screening_policy(item) -> ScreeningPolicy: + return ScreeningPolicy( + fuzzy_threshold=item.fuzzy_threshold, + max_snapshot_age_days=item.max_snapshot_age_days, + max_candidates=item.max_candidates, + ) + + def _subject_response(item) -> SubjectSnapshotResponse: return SubjectSnapshotResponse.model_validate( item, @@ -534,6 +607,7 @@ def _candidate_response(item) -> CandidateResponse: def _run_summary(item) -> ScreeningRunSummaryResponse: return ScreeningRunSummaryResponse( id=item.id, + evidence_ref=screening_evidence_ref(item.id), subject_name=item.subject_snapshot.primary_name, subject_type=item.subject_snapshot.subject_type, list_snapshot_id=item.list_snapshot_id, @@ -553,6 +627,7 @@ def _run_summary(item) -> ScreeningRunSummaryResponse: def _run_response(item) -> ScreeningRunResponse: return ScreeningRunResponse( id=item.id, + evidence_ref=screening_evidence_ref(item.id), idempotency_key=item.idempotency_key, request_hash=item.request_hash, matcher_version=item.matcher_version, @@ -581,6 +656,21 @@ def _run_response(item) -> ScreeningRunResponse: ) +def _freshness_response( + item: ScreeningFreshnessAssessment, +) -> ScreeningFreshnessResponse: + return ScreeningFreshnessResponse( + evidence_ref=item.evidence_ref, + run=_run_summary(item.run), + fresh=item.fresh, + reasons=list(item.reasons), + checked_at=item.checked_at, + current_list_snapshot_id=item.current_list_snapshot_id, + gate_decision=item.gate_decision, + gate_reasons=list(item.gate_reasons), + ) + + def _candidate_detail_response(item) -> CandidateDetailResponse: return CandidateDetailResponse( candidate=_candidate_response(item), diff --git a/src/govoplan_risk_compliance/backend/schemas.py b/src/govoplan_risk_compliance/backend/schemas.py index 1bff8fd..1950907 100644 --- a/src/govoplan_risk_compliance/backend/schemas.py +++ b/src/govoplan_risk_compliance/backend/schemas.py @@ -122,6 +122,20 @@ class ScreeningRunCreateRequest(BaseModel): ) +class ScreeningFreshnessCheckRequest(BaseModel): + current_subject: ScreeningSubjectInput | None = None + expected_list_snapshot_id: str | None = Field( + default=None, + max_length=36, + ) + policy: ScreeningPolicyInput | None = None + failure_policy: Literal[ + "block", + "review", + "degraded", + ] = "block" + + class SubjectSnapshotResponse(BaseModel): id: str subject_ref: str | None @@ -196,6 +210,7 @@ class CandidateResponse(BaseModel): class ScreeningRunSummaryResponse(BaseModel): id: str + evidence_ref: str subject_name: str | None subject_type: str list_snapshot_id: str @@ -213,6 +228,7 @@ class ScreeningRunSummaryResponse(BaseModel): class ScreeningRunResponse(BaseModel): id: str + evidence_ref: str idempotency_key: str request_hash: str matcher_version: str @@ -240,6 +256,26 @@ class ScreeningRunListResponse(BaseModel): runs: list[ScreeningRunSummaryResponse] +class ScreeningFreshnessResponse(BaseModel): + evidence_ref: str + run: ScreeningRunSummaryResponse + fresh: bool + reasons: list[str] + checked_at: datetime + current_list_snapshot_id: str | None + gate_decision: Literal[ + "allow", + "block", + "review", + "degraded", + ] + gate_reasons: list[str] + + +class RescreeningRequirementListResponse(BaseModel): + requirements: list[ScreeningFreshnessResponse] + + class ReviewQueueItemResponse(BaseModel): id: str run_id: str @@ -331,6 +367,9 @@ __all__ = [ "ListSnapshotResponse", "ReviewQueueItemResponse", "ReviewQueueResponse", + "RescreeningRequirementListResponse", + "ScreeningFreshnessCheckRequest", + "ScreeningFreshnessResponse", "ScreeningRunCreateRequest", "ScreeningRunCreateResponse", "ScreeningRunListResponse", diff --git a/src/govoplan_risk_compliance/backend/screening.py b/src/govoplan_risk_compliance/backend/screening.py index 0e55d01..1801e55 100644 --- a/src/govoplan_risk_compliance/backend/screening.py +++ b/src/govoplan_risk_compliance/backend/screening.py @@ -16,6 +16,7 @@ from govoplan_risk_compliance.backend.db.models import ( RiskSanctionsEntry, RiskSanctionsListSnapshot, RiskScreeningCandidate, + RiskScreeningDisposition, RiskScreeningException, RiskScreeningRun, RiskScreeningSubjectSnapshot, @@ -44,6 +45,10 @@ MAX_MATCH_ENTRIES = 5_000 MAX_CANDIDATES = 100 DEFAULT_FUZZY_THRESHOLD = 0.88 DEFAULT_MAX_SNAPSHOT_AGE_DAYS = 7 +SCREENING_EVIDENCE_PREFIX = "risk-screening:" +SCREENING_FAILURE_POLICIES = frozenset( + {"block", "review", "degraded"} +) @dataclass(frozen=True, slots=True) @@ -99,6 +104,21 @@ class ScreeningPolicy: } +@dataclass(frozen=True, slots=True) +class ScreeningFreshnessAssessment: + run: RiskScreeningRun + fresh: bool + reasons: tuple[str, ...] + checked_at: datetime + current_list_snapshot_id: str | None + gate_decision: str + gate_reasons: tuple[str, ...] + + @property + def evidence_ref(self) -> str: + return screening_evidence_ref(self.run.id) + + def run_screening( session: Session, principal: ApiPrincipal, @@ -284,6 +304,183 @@ def run_screening( ), True +def screening_evidence_ref(run_id: str) -> str: + clean_run_id = str(run_id or "").strip() + if not clean_run_id: + raise RiskSanctionsConflictError( + "A screening run is required for evidence." + ) + return f"{SCREENING_EVIDENCE_PREFIX}{clean_run_id}" + + +def assess_screening_freshness( + session: Session, + principal: ApiPrincipal, + *, + evidence_ref: str, + current_subject: ScreeningSubject | None = None, + expected_list_snapshot_id: str | None = None, + policy: ScreeningPolicy | None = None, + failure_policy: str = "block", + now: datetime | None = None, +) -> ScreeningFreshnessAssessment: + if failure_policy not in SCREENING_FAILURE_POLICIES: + raise RiskSanctionsConflictError( + "Failure policy must be block, review, or degraded." + ) + run = get_screening_run( + session, + principal, + run_id=_run_id_from_evidence_ref(evidence_ref), + ) + checked_at = now or utcnow() + effective_policy = policy or _policy_from_run(run) + reasons: list[str] = [] + + if expected_list_snapshot_id: + current_snapshot = _visible_list_snapshot( + session, + principal, + snapshot_id=expected_list_snapshot_id, + ) + else: + current_snapshot = _latest_visible_list_snapshot( + session, + principal, + baseline=run.list_snapshot, + ) + reasons.extend( + _snapshot_freshness_reasons( + run, + current_snapshot=current_snapshot, + checked_at=checked_at, + policy=effective_policy, + ) + ) + + if current_subject is not None: + current_fingerprint = subject_fingerprint( + _canonical_subject(current_subject) + ) + if current_fingerprint != run.subject_snapshot.fingerprint: + reasons.append("subject_changed") + reasons.extend( + _record_freshness_reasons( + session, + run=run, + checked_at=checked_at, + ) + ) + if policy is not None and not _run_policy_matches(run, policy): + reasons.append("policy_changed") + + unique_reasons = tuple(dict.fromkeys(reasons)) + gate_decision, gate_reasons = _screening_gate( + run, + freshness_reasons=unique_reasons, + failure_policy=failure_policy, + ) + return ScreeningFreshnessAssessment( + run=run, + fresh=not unique_reasons, + reasons=unique_reasons, + checked_at=checked_at, + current_list_snapshot_id=( + current_snapshot.id + if current_snapshot is not None + else None + ), + gate_decision=gate_decision, + gate_reasons=gate_reasons, + ) + + +def list_rescreening_requirements( + session: Session, + principal: ApiPrincipal, + *, + limit: int = 100, + now: datetime | None = None, +) -> tuple[ScreeningFreshnessAssessment, ...]: + _require_scope(principal, SANCTIONS_READ_SCOPE) + bounded_limit = max(1, min(limit, 500)) + items = tuple( + session.scalars( + select(RiskScreeningRun) + .where(RiskScreeningRun.tenant_id == principal.tenant_id) + .options( + selectinload(RiskScreeningRun.subject_snapshot), + selectinload(RiskScreeningRun.list_snapshot), + selectinload(RiskScreeningRun.candidates), + ) + .order_by( + RiskScreeningRun.created_at.desc(), + RiskScreeningRun.id.desc(), + ) + .limit(bounded_limit * 5) + ).unique() + ) + latest_runs: list[RiskScreeningRun] = [] + seen_subjects: set[str] = set() + for item in items: + subject_key = ( + item.subject_snapshot.subject_ref + or item.subject_snapshot.fingerprint + ) + if subject_key in seen_subjects: + continue + seen_subjects.add(subject_key) + latest_runs.append(item) + + disposition_ids = { + candidate.current_disposition_id + for item in latest_runs + for candidate in item.candidates + if candidate.current_disposition_id + } + dispositions = { + item.id: item + for item in session.scalars( + select(RiskScreeningDisposition).where( + RiskScreeningDisposition.tenant_id + == principal.tenant_id, + RiskScreeningDisposition.id.in_(disposition_ids), + ) + ) + } if disposition_ids else {} + latest_snapshots: dict[ + tuple[str, str], + RiskSanctionsListSnapshot | None, + ] = {} + requirements: list[ScreeningFreshnessAssessment] = [] + checked_at = now or utcnow() + for item in latest_runs: + source_key = ( + item.list_snapshot.provider_id, + item.list_snapshot.source_id, + ) + if source_key not in latest_snapshots: + latest_snapshots[source_key] = ( + _latest_visible_list_snapshot( + session, + principal, + baseline=item.list_snapshot, + ) + ) + assessment = _assess_loaded_run( + session, + run=item, + checked_at=checked_at, + current_snapshot=latest_snapshots[source_key], + dispositions=dispositions, + ) + if not assessment.fresh: + requirements.append(assessment) + if len(requirements) >= bounded_limit: + break + return tuple(requirements) + + def get_screening_run( session: Session, principal: ApiPrincipal, @@ -364,6 +561,234 @@ def _visible_list_snapshot( return item +def _latest_visible_list_snapshot( + session: Session, + principal: ApiPrincipal, + *, + baseline: RiskSanctionsListSnapshot, +) -> RiskSanctionsListSnapshot | None: + return session.scalar( + select(RiskSanctionsListSnapshot) + .where( + RiskSanctionsListSnapshot.provider_id + == baseline.provider_id, + RiskSanctionsListSnapshot.source_id == baseline.source_id, + or_( + RiskSanctionsListSnapshot.tenant_id + == principal.tenant_id, + RiskSanctionsListSnapshot.visibility == "global", + ), + ) + .order_by( + RiskSanctionsListSnapshot.acquired_at.desc(), + RiskSanctionsListSnapshot.imported_at.desc(), + RiskSanctionsListSnapshot.id.desc(), + ) + .limit(1) + ) + + +def _assess_loaded_run( + session: Session, + *, + run: RiskScreeningRun, + checked_at: datetime, + current_snapshot: RiskSanctionsListSnapshot | None, + dispositions: dict[str, RiskScreeningDisposition], +) -> ScreeningFreshnessAssessment: + policy = _policy_from_run(run) + reasons = list( + _snapshot_freshness_reasons( + run, + current_snapshot=current_snapshot, + checked_at=checked_at, + policy=policy, + ) + ) + reasons.extend( + _record_freshness_reasons( + session, + run=run, + checked_at=checked_at, + dispositions=dispositions, + ) + ) + unique_reasons = tuple(dict.fromkeys(reasons)) + gate_decision, gate_reasons = _screening_gate( + run, + freshness_reasons=unique_reasons, + failure_policy="block", + ) + return ScreeningFreshnessAssessment( + run=run, + fresh=not unique_reasons, + reasons=unique_reasons, + checked_at=checked_at, + current_list_snapshot_id=( + current_snapshot.id + if current_snapshot is not None + else None + ), + gate_decision=gate_decision, + gate_reasons=gate_reasons, + ) + + +def _snapshot_freshness_reasons( + run: RiskScreeningRun, + *, + current_snapshot: RiskSanctionsListSnapshot | None, + checked_at: datetime, + policy: ScreeningPolicy, +) -> tuple[str, ...]: + if current_snapshot is None: + return ("source_unavailable",) + reasons: list[str] = [] + if current_snapshot.id != run.list_snapshot_id: + reasons.append("source_snapshot_changed") + state = _list_state( + current_snapshot, + now=checked_at, + max_age_days=policy.max_snapshot_age_days, + ) + if state == "stale": + reasons.append("source_snapshot_stale") + elif state == "unavailable": + reasons.append("source_unavailable") + return tuple(reasons) + + +def _record_freshness_reasons( + session: Session, + *, + run: RiskScreeningRun, + checked_at: datetime, + dispositions: dict[str, RiskScreeningDisposition] | None = None, +) -> tuple[str, ...]: + reasons: list[str] = [] + if run.matcher_version != MATCHER_VERSION: + reasons.append("matcher_version_changed") + if run.normalization_version != NORMALIZATION_VERSION: + reasons.append("normalization_version_changed") + if run.policy_version != POLICY_VERSION: + reasons.append("policy_version_changed") + if run.status != "completed": + reasons.append("screening_incomplete") + if run.outcome in {"stale", "unavailable", "insufficient"}: + reasons.append(f"screening_outcome_{run.outcome}") + + disposition_ids = { + candidate.current_disposition_id + for candidate in run.candidates + if candidate.current_disposition_id + } + if not disposition_ids: + return tuple(reasons) + current_dispositions = dispositions + if current_dispositions is None: + current_dispositions = { + item.id: item + for item in session.scalars( + select(RiskScreeningDisposition).where( + RiskScreeningDisposition.tenant_id == run.tenant_id, + RiskScreeningDisposition.id.in_(disposition_ids), + ) + ) + } + for disposition_id in sorted(disposition_ids): + item = current_dispositions.get(disposition_id) + if item is None: + reasons.append("disposition_evidence_unavailable") + continue + if ( + item.expires_at is not None + and _aware_datetime(item.expires_at) + <= _aware_datetime(checked_at) + ): + reasons.append("disposition_expired") + if ( + item.review_at is not None + and _aware_datetime(item.review_at) + <= _aware_datetime(checked_at) + ): + reasons.append("disposition_review_due") + return tuple(dict.fromkeys(reasons)) + + +def _policy_from_run(run: RiskScreeningRun) -> ScreeningPolicy: + values = dict(run.policy_snapshot) + return ScreeningPolicy( + fuzzy_threshold=float( + values.get("fuzzy_threshold", DEFAULT_FUZZY_THRESHOLD) + ), + max_snapshot_age_days=int( + values.get( + "max_snapshot_age_days", + DEFAULT_MAX_SNAPSHOT_AGE_DAYS, + ) + ), + max_candidates=int( + values.get("max_candidates", MAX_CANDIDATES) + ), + ) + + +def _run_policy_matches( + run: RiskScreeningRun, + policy: ScreeningPolicy, +) -> bool: + recorded = dict(run.policy_snapshot) + expected = policy.to_dict() + return all( + recorded.get(key) == value + for key, value in expected.items() + ) + + +def _screening_gate( + run: RiskScreeningRun, + *, + freshness_reasons: tuple[str, ...], + failure_policy: str, +) -> tuple[str, tuple[str, ...]]: + if freshness_reasons: + return ( + failure_policy, + ("rescreening_required", *freshness_reasons), + ) + if run.outcome == "clear": + return "allow", ("screening_clear",) + if run.outcome == "potential": + statuses = { + candidate.review_status + for candidate in run.candidates + } + if "confirmed" in statuses: + return "block", ("confirmed_match",) + unresolved = statuses - {"false_positive"} + if unresolved or not statuses: + return "review", ("candidate_review_required",) + return "allow", ("all_candidates_cleared",) + return ( + failure_policy, + (f"screening_outcome_{run.outcome}",), + ) + + +def _run_id_from_evidence_ref(evidence_ref: str) -> str: + clean_ref = str(evidence_ref or "").strip() + if not clean_ref.startswith(SCREENING_EVIDENCE_PREFIX): + raise RiskSanctionsConflictError( + "Screening evidence reference is invalid." + ) + run_id = clean_ref.removeprefix(SCREENING_EVIDENCE_PREFIX) + if not run_id or ":" in run_id: + raise RiskSanctionsConflictError( + "Screening evidence reference is invalid." + ) + return run_id + + def _canonical_subject(subject: ScreeningSubject) -> dict[str, Any]: aliases = sorted( { @@ -606,6 +1031,12 @@ def _list_state( return "active" +def _aware_datetime(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + def _bounded(value: object, length: int) -> str | None: clean = " ".join(str(value or "").split()) return clean[:length] or None @@ -640,9 +1071,15 @@ __all__ = [ "MATCHER_VERSION", "MAX_CANDIDATES", "POLICY_VERSION", + "SCREENING_EVIDENCE_PREFIX", + "SCREENING_FAILURE_POLICIES", + "ScreeningFreshnessAssessment", "ScreeningPolicy", "ScreeningSubject", + "assess_screening_freshness", "get_screening_run", + "list_rescreening_requirements", "list_screening_runs", "run_screening", + "screening_evidence_ref", ] diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 5dfd679..1184401 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -2,6 +2,10 @@ from __future__ import annotations import unittest +from govoplan_core.core.sanctions import ( + CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING, + SanctionsScreeningProvider, +) from govoplan_risk_compliance.backend.manifest import ( ADMIN_SCOPE, READ_SCOPE, @@ -49,6 +53,17 @@ class ManifestTests(unittest.TestCase): manifest.requires_interfaces[0].name, ) self.assertTrue(manifest.requires_interfaces[0].optional) + self.assertEqual( + {"risk_compliance.sanctions_screening"}, + { + item.name + for item in manifest.provides_interfaces + }, + ) + capability = manifest.capability_factories[ + CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING + ](None) + self.assertIsInstance(capability, SanctionsScreeningProvider) if __name__ == "__main__": diff --git a/tests/test_sanctions_screening.py b/tests/test_sanctions_screening.py index bbc7811..2d45e83 100644 --- a/tests/test_sanctions_screening.py +++ b/tests/test_sanctions_screening.py @@ -1,5 +1,6 @@ from __future__ import annotations +from dataclasses import replace from datetime import timedelta from types import SimpleNamespace import hashlib @@ -12,6 +13,10 @@ from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.core.sanctions import ( CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS, + SanctionsScreeningFreshnessRequest, + SanctionsScreeningPolicy, + SanctionsScreeningRequest, + SanctionsScreeningSubject, SanctionsSnapshotPayload, SanctionsSnapshotReference, ) @@ -29,6 +34,9 @@ from govoplan_risk_compliance.backend.db.models import ( RiskScreeningRun, RiskScreeningSubjectSnapshot, ) +from govoplan_risk_compliance.backend.capabilities import ( + RiskComplianceSanctionsScreeningProvider, +) from govoplan_risk_compliance.backend.permissions import ( SANCTIONS_ADMIN_SCOPE, SANCTIONS_READ_SCOPE, @@ -48,7 +56,10 @@ from govoplan_risk_compliance.backend.screening import ( MATCHER_VERSION, ScreeningPolicy, ScreeningSubject, + assess_screening_freshness, + list_rescreening_requirements, run_screening, + screening_evidence_ref, ) @@ -306,6 +317,108 @@ class SanctionsScreeningTests(unittest.TestCase): self.assertEqual("stale", item.outcome) self.assertEqual(0, item.candidate_count) + def test_freshness_reconciles_source_subject_matcher_and_policy(self) -> None: + item, _ = run_screening( + self.session, + principal(), + list_snapshot_id=self.list_snapshot.id, + idempotency_key="freshness-1", + subject=ScreeningSubject( + subject_type="person", + primary_name="No Match", + subject_ref="person-1", + ), + ) + item.matcher_version = "sanctions-matcher-legacy" + self.provider.snapshot = replace( + self.provider.snapshot, + ref="sanctions-snapshot:fixture-v2", + source_version="fixture-v2", + acquired_at=utcnow(), + connector_run_id="run-fixture-v2", + ) + current_snapshot, created = import_connector_snapshot( + self.session, + principal(), + registry=self.registry, + connector_snapshot_ref=self.provider.snapshot.ref, + ) + self.assertTrue(created) + + assessment = assess_screening_freshness( + self.session, + principal(), + evidence_ref=screening_evidence_ref(item.id), + current_subject=ScreeningSubject( + subject_type="person", + primary_name="Changed Subject", + subject_ref="person-1", + ), + policy=ScreeningPolicy(fuzzy_threshold=0.9), + failure_policy="degraded", + ) + + self.assertFalse(assessment.fresh) + self.assertEqual(current_snapshot.id, assessment.current_list_snapshot_id) + self.assertEqual("degraded", assessment.gate_decision) + self.assertIn("source_snapshot_changed", assessment.reasons) + self.assertIn("subject_changed", assessment.reasons) + self.assertIn("matcher_version_changed", assessment.reasons) + self.assertIn("policy_changed", assessment.reasons) + requirements = list_rescreening_requirements( + self.session, + principal(), + ) + self.assertEqual([item.id], [value.run.id for value in requirements]) + + def test_versioned_capability_returns_stable_gate_evidence(self) -> None: + provider = RiskComplianceSanctionsScreeningProvider() + request = SanctionsScreeningRequest( + list_snapshot_id=self.list_snapshot.id, + idempotency_key="capability-1", + subject=SanctionsScreeningSubject( + subject_type="entity", + primary_name="No Match Company", + subject_ref="entity-1", + ), + policy=SanctionsScreeningPolicy( + failure_policy="review", + ), + ) + + result = provider.request_screening( + self.session, + principal(), + request, + ) + replay = provider.request_screening( + self.session, + principal(), + request, + ) + changed = provider.check_freshness( + self.session, + principal(), + SanctionsScreeningFreshnessRequest( + evidence_ref=result.evidence.ref, + current_subject=SanctionsScreeningSubject( + subject_type="entity", + primary_name="Changed Company", + subject_ref="entity-1", + ), + expected_list_snapshot_id=self.list_snapshot.id, + policy=request.policy, + ), + ) + + self.assertTrue(result.created) + self.assertFalse(replay.created) + self.assertEqual(result.evidence.ref, replay.evidence.ref) + self.assertEqual("allow", result.freshness.gate_decision) + self.assertFalse(changed.fresh) + self.assertEqual("review", changed.gate_decision) + self.assertIn("subject_changed", changed.reasons) + def test_review_is_separated_and_exception_requires_review(self) -> None: submitter = principal( scopes=( @@ -358,6 +471,16 @@ class SanctionsScreeningTests(unittest.TestCase): 1, self.session.query(RiskScreeningException).count(), ) + expired = assess_screening_freshness( + self.session, + principal(), + evidence_ref=screening_evidence_ref(item.id), + expected_list_snapshot_id=self.list_snapshot.id, + failure_policy="review", + now=disposition.expires_at + timedelta(seconds=1), + ) + self.assertIn("disposition_expired", expired.reasons) + self.assertEqual("review", expired.gate_decision) repeated, _ = run_screening( self.session,