diff --git a/src/govoplan_core/core/sanctions.py b/src/govoplan_core/core/sanctions.py index e02a6e9..05306a1 100644 --- a/src/govoplan_core/core/sanctions.py +++ b/src/govoplan_core/core/sanctions.py @@ -3,13 +3,17 @@ from __future__ import annotations from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime -from typing import Protocol, runtime_checkable +from typing import Literal, Protocol, runtime_checkable SANCTIONS_SNAPSHOT_CONTRACT_VERSION = "1" +SANCTIONS_SCREENING_CONTRACT_VERSION = "1" CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS = ( "connectors.sanctionsSnapshotProvider" ) +CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING = ( + "riskCompliance.sanctionsScreeningProvider" +) @dataclass(frozen=True, slots=True) @@ -90,6 +94,181 @@ class SanctionsSnapshotPayload: ) +@dataclass(frozen=True, slots=True) +class SanctionsScreeningSubject: + subject_type: Literal["person", "entity"] + primary_name: str | None = None + subject_ref: str | None = None + aliases: tuple[str, ...] = () + identifiers: tuple[Mapping[str, str], ...] = () + dates: tuple[str, ...] = () + addresses: tuple[Mapping[str, str], ...] = () + + def __post_init__(self) -> None: + if self.subject_type not in {"person", "entity"}: + raise ValueError( + "Sanctions screening subjects must be a person or entity." + ) + if not str(self.primary_name or "").strip() and not any( + str(value.get("value") or "").strip() + for value in self.identifiers + ): + raise ValueError( + "A sanctions screening subject needs a name or identifier." + ) + + +@dataclass(frozen=True, slots=True) +class SanctionsScreeningPolicy: + fuzzy_threshold: float = 0.88 + max_snapshot_age_days: int = 7 + max_candidates: int = 100 + failure_policy: Literal["block", "review", "degraded"] = "block" + + def __post_init__(self) -> None: + if not 0.8 <= self.fuzzy_threshold <= 1: + raise ValueError( + "Sanctions screening fuzzy threshold must be between 0.8 and 1." + ) + if not 1 <= self.max_snapshot_age_days <= 365: + raise ValueError( + "Sanctions snapshot age must be between 1 and 365 days." + ) + if not 1 <= self.max_candidates <= 100: + raise ValueError( + "Sanctions screening candidate limit must be between 1 and 100." + ) + if self.failure_policy not in {"block", "review", "degraded"}: + raise ValueError( + "Sanctions screening failure policy is not supported." + ) + + +@dataclass(frozen=True, slots=True) +class SanctionsScreeningRequest: + list_snapshot_id: str + idempotency_key: str + subject: SanctionsScreeningSubject + policy: SanctionsScreeningPolicy = field( + default_factory=SanctionsScreeningPolicy + ) + contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION + + def __post_init__(self) -> None: + if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION: + raise ValueError( + "Unsupported sanctions screening contract version." + ) + if not self.list_snapshot_id.strip(): + raise ValueError( + "A sanctions list snapshot is required for screening." + ) + if not self.idempotency_key.strip(): + raise ValueError( + "A sanctions screening idempotency key is required." + ) + + +@dataclass(frozen=True, slots=True) +class SanctionsScreeningEvidence: + ref: str + run_id: str + outcome: str + candidate_count: int + list_snapshot_id: str + source_version: str + subject_fingerprint: str + matcher_version: str + normalization_version: str + policy_version: str + completed_at: datetime | None + contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION + + def __post_init__(self) -> None: + if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION: + raise ValueError( + "Unsupported sanctions screening evidence version." + ) + if self.ref != f"risk-screening:{self.run_id}": + raise ValueError( + "Sanctions screening evidence reference is invalid." + ) + if self.candidate_count < 0: + raise ValueError( + "Sanctions screening candidate count cannot be negative." + ) + + +@dataclass(frozen=True, slots=True) +class SanctionsScreeningFreshnessRequest: + evidence_ref: str + current_subject: SanctionsScreeningSubject | None = None + expected_list_snapshot_id: str | None = None + policy: SanctionsScreeningPolicy = field( + default_factory=SanctionsScreeningPolicy + ) + contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION + + def __post_init__(self) -> None: + if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION: + raise ValueError( + "Unsupported sanctions screening freshness contract version." + ) + if not self.evidence_ref.startswith("risk-screening:"): + raise ValueError( + "Sanctions screening evidence reference is invalid." + ) + + +@dataclass(frozen=True, slots=True) +class SanctionsScreeningFreshness: + evidence: SanctionsScreeningEvidence + fresh: bool + reasons: tuple[str, ...] + checked_at: datetime + current_list_snapshot_id: str | None + gate_decision: Literal["allow", "block", "review", "degraded"] + gate_reasons: tuple[str, ...] + contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION + + def __post_init__(self) -> None: + if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION: + raise ValueError( + "Unsupported sanctions screening freshness version." + ) + if self.fresh != (not self.reasons): + raise ValueError( + "Sanctions screening freshness reasons are inconsistent." + ) + if self.gate_decision not in { + "allow", + "block", + "review", + "degraded", + }: + raise ValueError( + "Sanctions screening gate decision is invalid." + ) + + +@dataclass(frozen=True, slots=True) +class SanctionsScreeningResult: + evidence: SanctionsScreeningEvidence + freshness: SanctionsScreeningFreshness + created: bool + contract_version: str = SANCTIONS_SCREENING_CONTRACT_VERSION + + def __post_init__(self) -> None: + if self.contract_version != SANCTIONS_SCREENING_CONTRACT_VERSION: + raise ValueError( + "Unsupported sanctions screening result version." + ) + if self.evidence != self.freshness.evidence: + raise ValueError( + "Sanctions screening result evidence is inconsistent." + ) + + @runtime_checkable class SanctionsSnapshotProvider(Protocol): def list_snapshots( @@ -120,6 +299,25 @@ class SanctionsSnapshotProvider(Protocol): ... +@runtime_checkable +class SanctionsScreeningProvider(Protocol): + def request_screening( + self, + session: object, + principal: object, + request: SanctionsScreeningRequest, + ) -> SanctionsScreeningResult: + ... + + def check_freshness( + self, + session: object, + principal: object, + request: SanctionsScreeningFreshnessRequest, + ) -> SanctionsScreeningFreshness: + ... + + def sanctions_snapshot_provider( registry: object | None, ) -> SanctionsSnapshotProvider | None: @@ -142,11 +340,44 @@ def sanctions_snapshot_provider( ) +def sanctions_screening_provider( + registry: object | None, +) -> SanctionsScreeningProvider | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not hasattr(registry, "capability") + or not registry.has_capability( + CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING + ) + ): + return None + provider = registry.capability( + CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING + ) + return ( + provider + if isinstance(provider, SanctionsScreeningProvider) + else None + ) + + __all__ = [ "CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS", + "CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING", + "SANCTIONS_SCREENING_CONTRACT_VERSION", "SANCTIONS_SNAPSHOT_CONTRACT_VERSION", + "SanctionsScreeningEvidence", + "SanctionsScreeningFreshness", + "SanctionsScreeningFreshnessRequest", + "SanctionsScreeningPolicy", + "SanctionsScreeningProvider", + "SanctionsScreeningRequest", + "SanctionsScreeningResult", + "SanctionsScreeningSubject", "SanctionsSnapshotPayload", "SanctionsSnapshotProvider", "SanctionsSnapshotReference", + "sanctions_screening_provider", "sanctions_snapshot_provider", ] diff --git a/tests/test_sanctions_contract.py b/tests/test_sanctions_contract.py index 83b029b..2ba1982 100644 --- a/tests/test_sanctions_contract.py +++ b/tests/test_sanctions_contract.py @@ -6,9 +6,19 @@ import unittest from govoplan_core.core.sanctions import ( CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS, + CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING, + SanctionsScreeningEvidence, + SanctionsScreeningFreshness, + SanctionsScreeningFreshnessRequest, + SanctionsScreeningPolicy, + SanctionsScreeningProvider, + SanctionsScreeningRequest, + SanctionsScreeningResult, + SanctionsScreeningSubject, SanctionsSnapshotPayload, SanctionsSnapshotProvider, SanctionsSnapshotReference, + sanctions_screening_provider, sanctions_snapshot_provider, ) @@ -38,6 +48,27 @@ class _Registry: return self.provider if self.has_capability(name) else None +class _ScreeningProvider: + def request_screening(self, session, principal, request): + del session, principal, request + raise NotImplementedError + + def check_freshness(self, session, principal, request): + del session, principal, request + raise NotImplementedError + + +class _ScreeningRegistry: + def __init__(self, provider): + self.provider = provider + + def has_capability(self, name): + return name == CAPABILITY_RISK_COMPLIANCE_SANCTIONS_SCREENING + + def capability(self, name): + return self.provider if self.has_capability(name) else None + + class SanctionsContractTests(unittest.TestCase): def test_snapshot_evidence_is_versioned_and_bounded(self) -> None: content = b"" @@ -74,6 +105,66 @@ class SanctionsContractTests(unittest.TestCase): sanctions_snapshot_provider(_Registry(provider)), ) + def test_screening_evidence_and_freshness_are_versioned(self) -> None: + subject = SanctionsScreeningSubject( + subject_type="person", + primary_name="Example Person", + ) + policy = SanctionsScreeningPolicy(failure_policy="review") + request = SanctionsScreeningRequest( + list_snapshot_id="snapshot-1", + idempotency_key="request-1", + subject=subject, + policy=policy, + ) + evidence = SanctionsScreeningEvidence( + ref="risk-screening:run-1", + run_id="run-1", + outcome="clear", + candidate_count=0, + list_snapshot_id=request.list_snapshot_id, + source_version="2026-07-30", + subject_fingerprint="a" * 64, + matcher_version="matcher-v1", + normalization_version="normalization-v1", + policy_version="policy-v1", + completed_at=datetime.now(timezone.utc), + ) + freshness_request = SanctionsScreeningFreshnessRequest( + evidence_ref=evidence.ref, + current_subject=subject, + policy=policy, + ) + freshness = SanctionsScreeningFreshness( + evidence=evidence, + fresh=True, + reasons=(), + checked_at=datetime.now(timezone.utc), + current_list_snapshot_id=request.list_snapshot_id, + gate_decision="allow", + gate_reasons=("screening_clear",), + ) + result = SanctionsScreeningResult( + evidence=evidence, + freshness=freshness, + created=True, + ) + provider = _ScreeningProvider() + + self.assertEqual(evidence.ref, freshness_request.evidence_ref) + self.assertTrue(result.freshness.fresh) + self.assertIsInstance(provider, SanctionsScreeningProvider) + self.assertIs( + provider, + sanctions_screening_provider( + _ScreeningRegistry(provider) + ), + ) + + def test_screening_contract_rejects_ambiguous_subjects(self) -> None: + with self.assertRaises(ValueError): + SanctionsScreeningSubject(subject_type="entity") + if __name__ == "__main__": unittest.main()