sovereign_bank_policy.py

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))