from __future__ import annotations import re import json import time import hashlib from dataclasses import dataclass, field from typing import Any, Dict, List, Tuple def _utc_ts() -> str: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def _sha256_text(text: str) -> str: return hashlib.sha256((text or "").encode("utf-8")).hexdigest() def _safe_lower(x: Any) -> str: if x is None: return "" return str(x).strip().lower() def _norm_tags(tags: Any) -> List[str]: if tags is None: return [] if isinstance(tags, list): return [str(t).strip().lower() for t in tags if str(t).strip()] if isinstance(tags, str): return [t.strip().lower() for t in tags.split(",") if t.strip()] return [] def _clip(text: str, n: int = 180) -> str: text = text or "" return text if len(text) <= n else text[:n] + "..." @dataclass class PolicyHit: rule_id: str severity: str title: str reason: str evidence: str recommended_action: str def as_dict(self) -> Dict[str, Any]: return { "rule_id": self.rule_id, "severity": self.severity, "title": self.title, "reason": self.reason, "evidence": self.evidence, "recommended_action": self.recommended_action, } @dataclass class BankPolicyResult: engine: str evaluated_at: str outcome: str risk_score: int risk_level: str summary: str tags: List[str] = field(default_factory=list) hits: List[Dict[str, Any]] = field(default_factory=list) controls_triggered: List[str] = field(default_factory=list) justification: List[str] = field(default_factory=list) policy_hash: str = "" def as_dict(self) -> Dict[str, Any]: return { "engine": self.engine, "evaluated_at": self.evaluated_at, "outcome": self.outcome, "risk_score": self.risk_score, "risk_level": self.risk_level, "summary": self.summary, "tags": self.tags, "hits": self.hits, "controls_triggered": self.controls_triggered, "justification": self.justification, "policy_hash": self.policy_hash, } class SovereignBankPolicy: """ Bank-grade explainable policy engine. Deterministic outputs: - ALLOW - FREEZE - BLOCK """ ENGINE_NAME = "Sovereign_Bank_Policy_v1" PII_PATTERNS: List[Tuple[str, str]] = [ ("iban_like", r"\b[A-Z]{2}\d{2}[A-Z0-9]{10,30}\b"), ("card_like", r"\b(?:\d[ -]*?){13,19}\b"), ("swift_like", r"\b[A-Z]{6}[A-Z0-9]{2}(?:[A-Z0-9]{3})?\b"), ("sort_code_like", r"\b\d{2}-\d{2}-\d{2}\b"), ("account_number_like", r"\b\d{8,12}\b"), ] BLOCK_KEYWORDS = { "credential_theft": [ "steal credentials", "dump credentials", "export credentials", "reveal password", "show password", "admin password", "token dump", "api key", "secret key", "private key", "session token", "root access", "privilege escalation", ], "payment_abuse": [ "override payment", "bypass approval", "skip approval", "force transfer", "unauthorized transfer", "change beneficiary", "modify settlement", "release payment", "approve payment now", "disable payment control", "ignore payment policy", ], "fraud_enablement": [ "synthetic identity fraud", "money mule", "card testing", "transaction laundering", "fraud ring", "bypass aml", "avoid sanctions", "evade kyc", "hide source of funds", "shell customer", "beneficial owner concealment", ], "prompt_attack": [ "ignore previous instructions", "reveal system prompt", "bypass guardrails", "disable security", "disable logging", "override policy", "act as root", "jailbreak", ], "exfiltration": [ "export customer data", "dump database", "extract pii", "download all records", "exfiltrate records", "leak account data", "show full customer list", "show all transactions", ], } FREEZE_KEYWORDS = { "suspicious_payments": [ "urgent wire", "out-of-band payment", "manual transfer", "change destination account", "new beneficiary", "same-day settlement", "high value payment", "override hold", "manual release", "priority transfer", "expedite funds", "temporary approval", ], "aml_kyc_sanctions": [ "sanctions", "pep", "high-risk jurisdiction", "source of funds", "beneficial owner", "aml alert", "transaction monitoring bypass", "screening issue", "name screening", "kyc refresh overdue", ], "access_anomaly": [ "admin override", "temporary admin", "backdoor access", "shared account", "service account misuse", "elevated access", "emergency access", "break glass", ], "privacy_risk": [ "full statement", "all transactions", "customer passport", "national id", "tax id", "date of birth", "full address", "phone and email", "all customer profiles", ], } FINANCIAL_CONTEXT = { "bank", "banking", "payments", "core banking", "swift", "iban", "payment", "ledger", "customer", "retail banking", "private banking", "treasury", "settlement", "compliance", "aml", "kyc", "fraud", "transaction", "beneficiary", "account", "card", "merchant", "sanctions", "financial", "fintech" } HIGH_RISK_TAGS = { "pii", "payments", "payment", "customer_data", "customer_chat", "production", "prod", "aml", "kyc", "sanctions", "ledger", "transactions", "banking", "financial", "secrets", "credentials" } def _collect_text(self, payload: Dict[str, Any]) -> str: parts = [ str(payload.get("engine") or ""), str(payload.get("parent_model") or payload.get("agent_id") or ""), str(payload.get("model_version") or ""), str(payload.get("risk_level") or ""), str(payload.get("notes") or ""), str(payload.get("context") or ""), str(payload.get("request") or ""), str(payload.get("prompt") or ""), str(payload.get("output_text") or ""), " ".join(_norm_tags(payload.get("data_tags"))), ] return "\n".join(p for p in parts if p).strip() def _contains_financial_context(self, text: str, tags: List[str]) -> bool: text_l = _safe_lower(text) if any(tag in self.HIGH_RISK_TAGS for tag in tags): return True return any(tok in text_l for tok in self.FINANCIAL_CONTEXT) def _risk_level_from_score(self, score: int) -> str: if score >= 85: return "CRITICAL" if score >= 65: return "HIGH" if score >= 35: return "MEDIUM" return "LOW" def _recommended_outcome(self, hits: List[PolicyHit], risk_score: int) -> str: if any(h.recommended_action == "BLOCK" for h in hits): return "BLOCK" if any(h.recommended_action == "FREEZE" for h in hits): return "FREEZE" if risk_score >= 85: return "FREEZE" return "ALLOW" def _add_hit( self, hits: List[PolicyHit], rule_id: str, severity: str, title: str, reason: str, evidence: str, recommended_action: str, ) -> None: hits.append( PolicyHit( rule_id=rule_id, severity=severity, title=title, reason=reason, evidence=_clip(evidence, 220), recommended_action=recommended_action, ) ) def _scan_block_keywords(self, text: str, hits: List[PolicyHit]) -> int: score = 0 lower = _safe_lower(text) for category, keywords in self.BLOCK_KEYWORDS.items(): matched = [kw for kw in keywords if kw in lower] if matched: score += 28 + min(12, len(matched) * 4) self._add_hit( hits=hits, rule_id=f"BLOCK_{category.upper()}", severity="CRITICAL", title=f"Bank policy block: {category}", reason=f"Detected explicit prohibited banking/security misuse language in category '{category}'.", evidence=", ".join(matched[:6]), recommended_action="BLOCK", ) return score def _scan_freeze_keywords(self, text: str, hits: List[PolicyHit]) -> int: score = 0 lower = _safe_lower(text) for category, keywords in self.FREEZE_KEYWORDS.items(): matched = [kw for kw in keywords if kw in lower] if matched: score += 14 + min(10, len(matched) * 3) self._add_hit( hits=hits, rule_id=f"FREEZE_{category.upper()}", severity="HIGH", title=f"Bank policy freeze: {category}", reason="Detected ambiguous or high-risk financial activity requiring human or policy review.", evidence=", ".join(matched[:6]), recommended_action="FREEZE", ) return score def _scan_pii(self, text: str, hits: List[PolicyHit]) -> int: score = 0 for name, pattern in self.PII_PATTERNS: matches = re.findall(pattern, text) if not matches: continue evidence = ", ".join(matches[:4]) self._add_hit( hits=hits, rule_id=f"PII_{name.upper()}", severity="HIGH", title=f"Sensitive financial / identity pattern detected: {name}", reason="Potential exposure of customer financial or identifying data.", evidence=evidence, recommended_action="FREEZE", ) score += 12 return score def _scan_high_risk_tags(self, tags: List[str], hits: List[PolicyHit]) -> int: overlap = sorted(set(tags) & self.HIGH_RISK_TAGS) if not overlap: return 0 action = "FREEZE" if any(t in overlap for t in ["pii", "payments", "production", "prod", "sanctions"]) else "ALLOW" self._add_hit( hits=hits, rule_id="TAG_HIGH_RISK_CONTEXT", severity="MEDIUM" if action == "ALLOW" else "HIGH", title="High-risk banking context", reason="Payload indicates sensitive banking or production context.", evidence=", ".join(overlap[:10]), recommended_action=action, ) return 8 + min(12, len(overlap) * 2) def _scan_declared_risk(self, payload: Dict[str, Any], hits: List[PolicyHit]) -> int: level = _safe_lower(payload.get("risk_level") or "medium") if level == "critical": self._add_hit( hits=hits, rule_id="DECLARED_RISK_CRITICAL", severity="CRITICAL", title="Declared critical risk input", reason="Caller declared request as critical-risk.", evidence="risk_level=critical", recommended_action="FREEZE", ) return 25 if level == "high": self._add_hit( hits=hits, rule_id="DECLARED_RISK_HIGH", severity="HIGH", title="Declared high risk input", reason="Caller declared request as high-risk.", evidence="risk_level=high", recommended_action="FREEZE", ) return 16 if level == "medium": return 6 return 0 def _scan_production_payment_combo(self, payload: Dict[str, Any], text: str, tags: List[str], hits: List[PolicyHit]) -> int: lower = _safe_lower(text) prod = any(t in tags for t in ["production", "prod"]) or "production" in lower or "prod" in lower pay = any(t in tags for t in ["payments", "payment", "transactions", "ledger"]) or any( x in lower for x in ["payment", "transfer", "beneficiary", "transaction", "settlement", "ledger"] ) if prod and pay: self._add_hit( hits=hits, rule_id="BANK_PROD_PAYMENT_CONTEXT", severity="HIGH", title="Production payment context", reason="Request touches production-like payment context where proportional control is required.", evidence="production/prod + payment/transaction indicators", recommended_action="FREEZE", ) return 18 return 0 def evaluate(self, payload: Dict[str, Any]) -> Dict[str, Any]: payload = payload or {} text = self._collect_text(payload) tags = _norm_tags(payload.get("data_tags")) hits: List[PolicyHit] = [] controls_triggered: List[str] = [] justification: List[str] = [] score = 0 if not text.strip(): result = BankPolicyResult( engine=self.ENGINE_NAME, evaluated_at=_utc_ts(), outcome="FREEZE", risk_score=50, risk_level="MEDIUM", summary="Empty or insufficient request context for bank-grade evaluation.", tags=tags, hits=[], controls_triggered=["insufficient_context_freeze"], justification=["Bank-grade enforcement requires meaningful context before execution."], ) raw = result.as_dict() raw["policy_hash"] = _sha256_text(json.dumps(raw, sort_keys=True, ensure_ascii=False)) return raw financial_context = self._contains_financial_context(text, tags) if financial_context: controls_triggered.append("financial_context_detected") justification.append("Financial / banking context detected, so bank policy pack was applied.") score += 10 else: controls_triggered.append("generic_context_only") justification.append("No strong banking context markers detected; result remains conservative but lower-confidence.") score += self._scan_declared_risk(payload, hits) score += self._scan_high_risk_tags(tags, hits) score += self._scan_pii(text, hits) score += self._scan_block_keywords(text, hits) score += self._scan_freeze_keywords(text, hits) score += self._scan_production_payment_combo(payload, text, tags, hits) lower = _safe_lower(text) prod = any(t in tags for t in ["production", "prod"]) or "production" in lower or "prod" in lower secrets = any(x in lower for x in ["password", "secret", "token", "private key", "api key", "credential"]) if prod and secrets: self._add_hit( hits=hits, rule_id="PROD_SECRET_COMBINATION", severity="CRITICAL", title="Production secret exposure risk", reason="Production context combined with secret/credential language is not acceptable for autonomous execution.", evidence="production/prod + secret/token/password/key markers", recommended_action="BLOCK", ) score += 35 score = max(0, min(100, score)) risk_level = self._risk_level_from_score(score) outcome = self._recommended_outcome(hits, score) if outcome == "BLOCK": controls_triggered.append("deterministic_block") justification.append("Explicit prohibited misuse indicators were found.") elif outcome == "FREEZE": controls_triggered.append("proportional_freeze") justification.append("High-risk or ambiguous financial activity requires suspension pending validation.") else: controls_triggered.append("allow_with_audit") justification.append("No deterministic block/freeze trigger was found under current bank policy rules.") if financial_context and outcome == "ALLOW" and risk_level in {"HIGH", "CRITICAL"}: outcome = "FREEZE" controls_triggered.append("risk_based_freeze_override") justification.append("Banking context with elevated risk was normalized to FREEZE for proportional control.") summary = { "ALLOW": "No bank-grade deterministic prohibition found; request may proceed under audit.", "FREEZE": "Risky or ambiguous financial context detected; execution should pause for validation.", "BLOCK": "Explicit prohibited financial/security misuse detected; execution should not proceed.", }[outcome] result = BankPolicyResult( engine=self.ENGINE_NAME, evaluated_at=_utc_ts(), outcome=outcome, risk_score=score, risk_level=risk_level, summary=summary, tags=tags, hits=[h.as_dict() for h in hits], controls_triggered=controls_triggered, justification=justification, ) raw = result.as_dict() raw["policy_hash"] = _sha256_text(json.dumps(raw, sort_keys=True, ensure_ascii=False)) return raw def evaluate_bank_policy(payload: Dict[str, Any]) -> Dict[str, Any]: return SovereignBankPolicy().evaluate(payload) if __name__ == "__main__": demo_payload = { "engine": "AI_Sovereign_Sentinel_Core_v1", "parent_model": "bank-agent-alpha", "model_version": "v1", "data_tags": "pii,payments,production,banking", "risk_level": "high", "notes": "User requests to bypass approval and change beneficiary for urgent wire.", "request": "Please override payment approval and release transfer now.", "output_text": "", } print(json.dumps(evaluate_bank_policy(demo_payload), indent=2, ensure_ascii=False))