Merge pull request #78 from QWED-AI/feat/phase-19-trust

feat(guards): implement SovereigntyGuard for PII data residency and local routing

Author: rahuldass19

qwed_sdk/guards/sovereignty_guard.py

@@ -0,0 +1,44 @@
+import re
+
+class SovereigntyGuard:
+    """
+    Enforces Data Residency and Sovereignty.
+    Source: Alignment with privacy-sensitive legal deployments [Source 60, 61].
+    """
+    def __init__(self, required_local_providers=None):
+        if required_local_providers is None:
+            required_local_providers = ["ollama", "vllm_local"]
+        self.local_providers = list(required_local_providers)
+        self._local_providers_casefold = {p.casefold() for p in self.local_providers}
+        
+        # Regex for SSN variants and Confidentiality markers
+        self.sensitive_patterns = [
+            re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),      # SSN: dash-separated
+            re.compile(r"\b\d{3}\s\d{2}\s\d{4}\b"),     # SSN: space-separated
+            re.compile(r"\b\d{9}\b"),                    # SSN: contiguous (use with care — broad)
+            re.compile(r"CONFIDENTIAL", re.IGNORECASE),
+        ]
+
+    def verify_routing(self, prompt: str, target_provider: str) -> dict:
+        if not prompt:
+            raise ValueError("prompt must be a non-empty string.")
+        if not target_provider:
+            raise ValueError("target_provider must be a non-empty string.")
+
+        # ISSUE: Does this prompt contain sensitive data destined for an external provider?
+        is_sensitive = any(p.search(prompt) for p in self.sensitive_patterns)
+        
+        # RULE: Sensitive data must never be routed to non-local infrastructure.
+        # [Source: MSLR, data residency compliance]
+        
+        # APPLICATION: Evaluate whether target_provider is an approved local provider.
+        if is_sensitive and target_provider.casefold() not in self._local_providers_casefold:
+            # CONCLUSION: Violation — block routing.
+            return {
+                "verified": False,
+                "risk": "DATA_SOVEREIGNTY_VIOLATION",
+                "message": f"Sensitive data detected. Routing to external provider '{target_provider}' is blocked. Must use local infrastructure."
+            }
+            
+        # CONCLUSION: No violation detected — permit routing.
+        return {"verified": True, "risk": None, "message": "Routing permitted."}

tests/test_phase19_sovereignty.py

@@ -0,0 +1,106 @@
+from qwed_sdk.guards.sovereignty_guard import SovereigntyGuard
+
+def test_sovereignty_guard_allows_safe_data_to_external():
+    guard = SovereigntyGuard()
+    result = guard.verify_routing(prompt="What is the capital of France?", target_provider="openai")
+    
+    assert result["verified"] is True
+    assert result["risk"] is None
+    assert "message" in result
+
+def test_sovereignty_guard_blocks_sensitive_data_to_external():
+    guard = SovereigntyGuard()
+    prompt = "Here is the CONFIDENTIAL contract for our new client."
+    result = guard.verify_routing(prompt=prompt, target_provider="openai")
+    
+    assert result["verified"] is False
+    assert result["risk"] == "DATA_SOVEREIGNTY_VIOLATION"
+    assert "message" in result
+
+def test_sovereignty_guard_blocks_ssn_to_external():
+    guard = SovereigntyGuard()
+    prompt = "My SSN is 123-45-6789. Please process my application."
+    result = guard.verify_routing(prompt=prompt, target_provider="anthropic")
+    
+    assert result["verified"] is False
+    assert result["risk"] == "DATA_SOVEREIGNTY_VIOLATION"
+    assert "message" in result
+
+def test_sovereignty_guard_allows_sensitive_data_to_local():
+    guard = SovereigntyGuard()
+    prompt = "Here is the CONFIDENTIAL contract. SSN: 123-45-6789."
+    # Ollama is in the default local_providers list
+    result = guard.verify_routing(prompt=prompt, target_provider="ollama")
+    
+    assert result["verified"] is True
+    assert result["risk"] is None
+    assert "message" in result
+
+def test_sovereignty_guard_blocks_space_separated_ssn_to_external():
+    guard = SovereigntyGuard()
+    prompt = "My SSN is 123 45 6789. Please process my application."
+    result = guard.verify_routing(prompt=prompt, target_provider="anthropic")
+    
+    assert result["verified"] is False
+    assert result["risk"] == "DATA_SOVEREIGNTY_VIOLATION"
+    assert "message" in result
+
+def test_sovereignty_guard_blocks_contiguous_ssn_to_external():
+    guard = SovereigntyGuard()
+    prompt = "SSN: 123456789"
+    result = guard.verify_routing(prompt=prompt, target_provider="openai")
+    
+    assert result["verified"] is False
+    assert result["risk"] == "DATA_SOVEREIGNTY_VIOLATION"
+    assert "message" in result
+
+def test_sovereignty_guard_allows_vllm_local():
+    guard = SovereigntyGuard()
+    prompt = "Here is the CONFIDENTIAL contract. SSN: 123-45-6789."
+    result = guard.verify_routing(prompt=prompt, target_provider="vllm_local")
+    
+    assert result["verified"] is True
+    assert result["risk"] is None
+    assert "message" in result
+
+def test_sovereignty_guard_case_insensitive_provider():
+    guard = SovereigntyGuard()
+    prompt = "Here is the CONFIDENTIAL contract. SSN: 123-45-6789."
+    result = guard.verify_routing(prompt=prompt, target_provider="OLLAMA")
+    
+    assert result["verified"] is True
+    assert result["risk"] is None
+    assert "message" in result
+
+def test_sovereignty_guard_custom_local_provider():
+    guard = SovereigntyGuard(required_local_providers=["my_local_llm"])
+    prompt = "Here is the CONFIDENTIAL contract. SSN: 123-45-6789."
+    
+    # Allowed local provider
+    result_local = guard.verify_routing(prompt=prompt, target_provider="my_local_llm")
+    assert result_local["verified"] is True
+    assert "message" in result_local
+    
+    # Default 'ollama' is now external since required_local_providers overrode it
+    result_external = guard.verify_routing(prompt=prompt, target_provider="ollama")
+    assert result_external["verified"] is False
+    assert result_external["risk"] == "DATA_SOVEREIGNTY_VIOLATION"
+    assert "message" in result_external
+
+import pytest
+
+def test_sovereignty_guard_raises_on_empty_prompt():
+    guard = SovereigntyGuard()
+    with pytest.raises(ValueError, match="prompt must be a non-empty string"):
+        guard.verify_routing(prompt="", target_provider="ollama")
+        
+    with pytest.raises(ValueError, match="prompt must be a non-empty string"):
+        guard.verify_routing(prompt=None, target_provider="ollama")
+
+def test_sovereignty_guard_raises_on_empty_provider():
+    guard = SovereigntyGuard()
+    with pytest.raises(ValueError, match="target_provider must be a non-empty string"):
+        guard.verify_routing(prompt="Valid prompt", target_provider="")
+        
+    with pytest.raises(ValueError, match="target_provider must be a non-empty string"):
+        guard.verify_routing(prompt="Valid prompt", target_provider=None)