Create sovereign_bank_policy_integration.py

sovereign_bank_policy_integration.py

@@ -0,0 +1,292 @@
+
+# sovereign_bank_policy_integration.py
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict
+
+import app as sovereign_app
+from sovereign_bank_policy import evaluate_bank_policy
+from sovereign_freeze_bridge import SovereignFreezeBridge
+from sovereign_app_freeze_integration import run_sentinel_with_freeze
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _norm_text(v: Any) -> str:
+    return str(v or "").strip()
+
+
+def _norm_tags(v: Any) -> list[str]:
+    if isinstance(v, list):
+        return [str(x).strip().lower() for x in v if str(x).strip()]
+    if isinstance(v, str):
+        return [x.strip().lower() for x in v.split(",") if x.strip()]
+    return []
+
+
+def _build_payload(
+    engine_name: str,
+    parent_model: str,
+    model_version: str,
+    data_tags: Any,
+    risk_level: str,
+    notes: str,
+    access_key: str,
+    delegation_token: str,
+) -> Dict[str, Any]:
+    return {
+        "engine": engine_name,
+        "engine_name": engine_name,
+        "parent_model": parent_model,
+        "parent_model_id": parent_model,
+        "agent_id": parent_model,
+        "model_version": model_version,
+        "build_id": model_version,
+        "data_tags": _norm_tags(data_tags),
+        "risk_level": _norm_text(risk_level).lower() or "medium",
+        "notes": _norm_text(notes),
+        "request": _norm_text(notes),
+        "context": _norm_text(notes),
+        "access_key_present": bool(_norm_text(access_key)),
+        "delegation_token_present": bool(_norm_text(delegation_token)),
+        "deployment_env": "production",
+        "environment": "production",
+    }
+
+
+def _persist_audit(event: Dict[str, Any]) -> None:
+    try:
+        fn = getattr(sovereign_app, "log_audit_event", None)
+        if callable(fn):
+            fn(
+                engine_name=event.get("engine", "unknown_engine"),
+                parent_model=event.get("parent_model", "unknown_parent"),
+                model_version=event.get("model_version", "unknown_version"),
+                data_tags=",".join(event.get("data_tags", []) or []),
+                risk_level=event.get("risk_level", "medium"),
+                notes=event.get("notes", ""),
+                event_type=event.get("event_type", "sentinel_run"),
+                outcome=event.get("outcome", "FREEZE"),
+                access_key="",
+                authority_bundle=event.get("authority_bundle", {}),
+                execution=event.get("execution", {}),
+            )
+    except Exception:
+        pass
+
+
+def _make_policy_override_result(
+    payload: Dict[str, Any],
+    policy_result: Dict[str, Any],
+    decision: str,
+    freeze_bridge_result: Dict[str, Any] | None = None,
+) -> Dict[str, Any]:
+    event_id = str(uuid.uuid4())
+    ts = _utc_now()
+
+    reason = policy_result.get("summary") or "bank_policy_override"
+    authority_bundle = {
+        "decision": decision,
+        "reason": reason,
+        "proof_type": "proof_of_non_action",
+        "policy_engine": policy_result.get("engine"),
+        "policy_hash": policy_result.get("policy_hash"),
+        "policy_outcome": policy_result.get("outcome"),
+        "policy_risk_score": policy_result.get("risk_score"),
+        "policy_risk_level": policy_result.get("risk_level"),
+        "controls_triggered": policy_result.get("controls_triggered", []),
+        "justification": policy_result.get("justification", []),
+    }
+
+    if freeze_bridge_result:
+        authority_bundle["freeze_bridge"] = freeze_bridge_result
+
+    result = {
+        "audit_event": {
+            "event_id": event_id,
+            "timestamp": ts,
+            "engine": payload.get("engine"),
+            "event_type": "sentinel_run",
+            "outcome": decision,
+            "parent_model": payload.get("parent_model"),
+            "model_version": payload.get("model_version"),
+            "data_tags": payload.get("data_tags", []),
+            "risk_level": payload.get("risk_level"),
+            "notes": f"[BANK_POLICY_OVERRIDE] {reason}",
+            "version": getattr(sovereign_app, "SOVEREIGN_VERSION", "unknown"),
+            "authority_name": getattr(sovereign_app, "AUTHORITY_NAME", "unknown"),
+            "authority_bundle": authority_bundle,
+            "execution": {
+                "executed": False,
+                "detail": {
+                    "blocked_by_bank_policy": True,
+                    "bank_policy_outcome": policy_result.get("outcome"),
+                },
+            },
+        },
+        "bank_policy": policy_result,
+        "authority_gate": authority_bundle,
+        "execution": {
+            "executed": False,
+            "detail": {
+                "blocked_by_bank_policy": True,
+                "bank_policy_outcome": policy_result.get("outcome"),
+            },
+        },
+    }
+
+    if freeze_bridge_result:
+        result["freeze_bridge"] = freeze_bridge_result
+
+    return result
+
+
+class SovereignBankPolicyIntegration:
+    """
+    Integrates bank policy into the Sovereign execution flow.
+
+    Order of control:
+    1) Bank policy pre-check
+    2) If BLOCK/FREEZE -> stop here
+    3) If ALLOW -> continue into freeze-integrated app flow
+    """
+
+    def __init__(self) -> None:
+        self.bridge = SovereignFreezeBridge()
+
+    def run(
+        self,
+        engine_name: str,
+        parent_model: str,
+        model_version: str,
+        data_tags: Any,
+        risk_level: str,
+        notes: str,
+        access_key: str,
+        delegation_token: str,
+    ) -> str:
+        payload = _build_payload(
+            engine_name=engine_name,
+            parent_model=parent_model,
+            model_version=model_version,
+            data_tags=data_tags,
+            risk_level=risk_level,
+            notes=notes,
+            access_key=access_key,
+            delegation_token=delegation_token,
+        )
+
+        policy_result = evaluate_bank_policy(payload)
+        policy_outcome = str(policy_result.get("outcome", "FREEZE")).upper()
+
+        # 1) Deterministic BLOCK from bank policy
+        if policy_outcome == "BLOCK":
+            result = _make_policy_override_result(
+                payload=payload,
+                policy_result=policy_result,
+                decision="BLOCK",
+                freeze_bridge_result=None,
+            )
+            _persist_audit(result["audit_event"])
+            return json.dumps(result, indent=2, ensure_ascii=False)
+
+        # 2) Deterministic FREEZE from bank policy
+        if policy_outcome == "FREEZE":
+            bridge_result = self.bridge.register_decision(
+                payload=payload,
+                decision_bundle={
+                    "decision_id": str(uuid.uuid4()),
+                    "decision": "FREEZE",
+                    "reason": policy_result.get("summary", "bank_policy_freeze"),
+                    "authority_gate": {
+                        "decision": "FREEZE",
+                        "reason": policy_result.get("summary", "bank_policy_freeze"),
+                        "policy_engine": policy_result.get("engine"),
+                        "policy_hash": policy_result.get("policy_hash"),
+                    },
+                    "bank_policy": policy_result,
+                },
+            )
+
+            result = _make_policy_override_result(
+                payload=payload,
+                policy_result=policy_result,
+                decision="FREEZE",
+                freeze_bridge_result=bridge_result,
+            )
+            _persist_audit(result["audit_event"])
+            return json.dumps(result, indent=2, ensure_ascii=False)
+
+        # 3) ALLOW from bank policy -> continue into existing freeze-integrated runtime
+        downstream_raw = run_sentinel_with_freeze(
+            engine_name=engine_name,
+            parent_model=parent_model,
+            model_version=model_version,
+            data_tags=data_tags,
+            risk_level=risk_level,
+            notes=notes,
+            access_key=access_key,
+            delegation_token=delegation_token,
+        )
+
+        try:
+            parsed = json.loads(downstream_raw)
+            if not isinstance(parsed, dict):
+                parsed = {"raw": downstream_raw}
+        except Exception:
+            parsed = {"raw": downstream_raw}
+
+        parsed["bank_policy"] = policy_result
+        return json.dumps(parsed, indent=2, ensure_ascii=False)
+
+    def patch_app(self) -> None:
+        sovereign_app.run_sentinel = self.run
+
+
+INTEGRATION = SovereignBankPolicyIntegration()
+
+
+def run_sentinel_with_bank_policy(
+    engine_name: str,
+    parent_model: str,
+    model_version: str,
+    data_tags: Any,
+    risk_level: str,
+    notes: str,
+    access_key: str,
+    delegation_token: str,
+) -> str:
+    return INTEGRATION.run(
+        engine_name=engine_name,
+        parent_model=parent_model,
+        model_version=model_version,
+        data_tags=data_tags,
+        risk_level=risk_level,
+        notes=notes,
+        access_key=access_key,
+        delegation_token=delegation_token,
+    )
+
+
+def patch_app_run_sentinel_with_bank_policy() -> None:
+    INTEGRATION.patch_app()
+
+
+if __name__ == "__main__":
+    sample = run_sentinel_with_bank_policy(
+        engine_name="AI_Sovereign_Sentinel_Core_v1",
+        parent_model="bank_agent_alpha",
+        model_version="v1",
+        data_tags="pii,payments,production,banking",
+        risk_level="high",
+        notes="Please override payment approval and release urgent wire transfer now.",
+        access_key="",
+        delegation_token="",
+    )
+    print(sample)
+