fix: fail closed python sanctions screening

Author: ramtamilselvan

sdk/python/tests/test_compliance_engine.py

@@ -3,6 +3,18 @@
 import pytest
 
 from zeroid.compliance.engine import ComplianceEngine, ComplianceCheckResult, CrossBorderCheckResult
+from zeroid.compliance.screening import SanctionsScreener, ScreeningEntry, ScreeningListType
+
+
+def configured_screener() -> SanctionsScreener:
+    return SanctionsScreener([
+        ScreeningEntry(
+            name="Lazarus Group",
+            list_type=ScreeningListType.SANCTIONS,
+            jurisdiction="KP",
+            identifiers=["0x" + "de" * 20],
+        ),
+    ])
 
 
 class TestComplianceEngine:
@@ -45,27 +57,33 @@ def test_check_credentials_high_risk_warning(self) -> None:
         assert any("risk" in w.lower() for w in result.warnings)
 
     def test_screen_entity_by_name(self) -> None:
-        engine = ComplianceEngine()
+        engine = ComplianceEngine(screener=configured_screener())
         result = engine.screen_entity(name="Lazarus Group")
         assert result.matched is True
 
     def test_screen_entity_by_identifier(self) -> None:
-        engine = ComplianceEngine()
+        engine = ComplianceEngine(screener=configured_screener())
         result = engine.screen_entity(identifier="0x" + "de" * 20)
         assert result.matched is True
 
     def test_screen_entity_no_match(self) -> None:
-        engine = ComplianceEngine()
+        engine = ComplianceEngine(screener=configured_screener())
         result = engine.screen_entity(name="Good Actor")
         assert result.matched is False
 
+    def test_screen_entity_default_screener_fails_closed(self) -> None:
+        engine = ComplianceEngine()
+        result = engine.screen_entity(name="Good Actor")
+        assert result.matched is True
+        assert result.error
+
     def test_screen_entity_no_args_raises(self) -> None:
         engine = ComplianceEngine()
         with pytest.raises(ValueError, match="At least one"):
             engine.screen_entity()
 
     def test_screen_entity_name_no_match_falls_to_identifier(self) -> None:
-        engine = ComplianceEngine()
+        engine = ComplianceEngine(screener=configured_screener())
         result = engine.screen_entity(
             name="Not Found", identifier="0x" + "de" * 20
         )

sdk/python/tests/test_compliance_screening.py

@@ -8,6 +8,32 @@
 )
 
 
+def sample_entries() -> list[ScreeningEntry]:
+    return [
+        ScreeningEntry(
+            name="Lazarus Group",
+            list_type=ScreeningListType.SANCTIONS,
+            jurisdiction="KP",
+            identifiers=["0x" + "de" * 20, "lazarus.kp"],
+            reason="State-sponsored cyber operations",
+        ),
+        ScreeningEntry(
+            name="Tornado Cash",
+            list_type=ScreeningListType.SANCTIONS,
+            jurisdiction="GLOBAL",
+            identifiers=["0x" + "ca" * 20],
+            reason="OFAC SDN listing - mixer service",
+        ),
+        ScreeningEntry(
+            name="Test PEP Entity",
+            list_type=ScreeningListType.PEP,
+            jurisdiction="XX",
+            identifiers=["pep-test-001"],
+            reason="Politically exposed person - test entry",
+        ),
+    ]
+
+
 class TestScreeningListType:
     def test_values(self) -> None:
         assert ScreeningListType.SANCTIONS.value == "sanctions"
@@ -34,54 +60,55 @@ def test_creation(self) -> None:
 
 
 class TestSanctionsScreener:
-    def test_defaults_loaded(self) -> None:
+    def test_default_unconfigured_fails_closed(self) -> None:
         screener = SanctionsScreener()
         result = screener.screen_name("Lazarus")
         assert result.matched is True
-        assert len(result.matches) == 1
+        assert result.matches == []
+        assert result.error
 
     def test_screen_name_no_match(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_name("Innocent Corp")
         assert result.matched is False
 
     def test_screen_name_case_insensitive(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_name("lazarus group")
         assert result.matched is True
 
     def test_screen_name_substring(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_name("Tornado")
         assert result.matched is True
 
     def test_screen_identifier_match(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_identifier("0x" + "de" * 20)
         assert result.matched is True
 
     def test_screen_identifier_no_match(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_identifier("0x" + "00" * 20)
         assert result.matched is False
 
     def test_screen_identifier_case_insensitive(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_identifier("0x" + "DE" * 20)
         assert result.matched is True
 
     def test_screen_jurisdiction_match(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_jurisdiction("KP")
         assert result.matched is True
 
     def test_screen_jurisdiction_no_match(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_jurisdiction("US")
         assert result.matched is False
 
     def test_screen_jurisdiction_case_insensitive(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_jurisdiction("kp")
         assert result.matched is True
 
@@ -100,6 +127,6 @@ def test_add_entry(self) -> None:
         assert result2.matched is True
 
     def test_query_preserved(self) -> None:
-        screener = SanctionsScreener()
+        screener = SanctionsScreener(sample_entries())
         result = screener.screen_name("TestQuery")
         assert result.query == "TestQuery"

sdk/python/zeroid/compliance/screening.py

@@ -1,7 +1,7 @@
 """Sanctions and PEP screening.
 
-Provides screening against mock sanctions/PEP watchlists for
-compliance checking.
+Callers must provide current, authoritative watchlist entries before using this
+module for compliance decisions. An unconfigured screener fails closed.
 """
 
 from __future__ import annotations
@@ -50,41 +50,15 @@ class ScreeningResult:
     matched: bool
     matches: list[ScreeningEntry] = field(default_factory=list)
     query: str = ""
+    error: str = ""
 
 
 class SanctionsScreener:
     """Screens entities against sanctions and PEP watchlists."""
 
-    def __init__(self) -> None:
-        """Initialize the screener with a default mock watchlist."""
-        self._entries: list[ScreeningEntry] = []
-        self._load_defaults()
-
-    def _load_defaults(self) -> None:
-        """Load default mock watchlist entries."""
-        self._entries = [
-            ScreeningEntry(
-                name="Lazarus Group",
-                list_type=ScreeningListType.SANCTIONS,
-                jurisdiction="KP",
-                identifiers=["0x" + "de" * 20, "lazarus.kp"],
-                reason="State-sponsored cyber operations",
-            ),
-            ScreeningEntry(
-                name="Tornado Cash",
-                list_type=ScreeningListType.SANCTIONS,
-                jurisdiction="GLOBAL",
-                identifiers=["0x" + "ca" * 20],
-                reason="OFAC SDN listing — mixer service",
-            ),
-            ScreeningEntry(
-                name="Test PEP Entity",
-                list_type=ScreeningListType.PEP,
-                jurisdiction="XX",
-                identifiers=["pep-test-001"],
-                reason="Politically exposed person — test entry",
-            ),
-        ]
+    def __init__(self, entries: list[ScreeningEntry] | None = None) -> None:
+        """Initialize the screener with caller-supplied watchlist entries."""
+        self._entries = list(entries or [])
 
     def add_entry(self, entry: ScreeningEntry) -> None:
         """Add an entry to the watchlist.
@@ -105,6 +79,9 @@ def screen_name(self, name: str) -> ScreeningResult:
         Returns:
             ScreeningResult with any matches.
         """
+        if not self._entries:
+            return self._unconfigured_result(name)
+
         matches = [
             e for e in self._entries
             if name.lower() in e.name.lower() or e.name.lower() in name.lower()
@@ -120,6 +97,9 @@ def screen_identifier(self, identifier: str) -> ScreeningResult:
         Returns:
             ScreeningResult with any matches.
         """
+        if not self._entries:
+            return self._unconfigured_result(identifier)
+
         identifier_lower = identifier.lower()
         matches = [
             e for e in self._entries
@@ -138,6 +118,9 @@ def screen_jurisdiction(self, jurisdiction_code: str) -> ScreeningResult:
         Returns:
             ScreeningResult with any matches.
         """
+        if not self._entries:
+            return self._unconfigured_result(jurisdiction_code)
+
         code_upper = jurisdiction_code.upper()
         matches = [
             e for e in self._entries
@@ -146,3 +129,11 @@ def screen_jurisdiction(self, jurisdiction_code: str) -> ScreeningResult:
         return ScreeningResult(
             matched=len(matches) > 0, matches=matches, query=jurisdiction_code
         )
+
+    def _unconfigured_result(self, query: str) -> ScreeningResult:
+        return ScreeningResult(
+            matched=True,
+            matches=[],
+            query=query,
+            error="screening watchlist entries are not configured",
+        )