sovereign _authority_gate.py

# sovereign_authority_gate.py
import os
import json
import math
import time
import hashlib
from typing import Any, Dict, Tuple, Optional, Callable, List


# ------------------------------
# tiny helpers
# ------------------------------
def _sha256_hex(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()


def _hmac_like(key: str, msg: str) -> str:
    # dependency-free "HMAC-like" binding for demo + sovereign environments
    return _sha256_hex(f"{key}|{msg}")


def _entropy(counts: Dict[str, int]) -> float:
    total = sum(counts.values()) or 1
    ent = 0.0
    for c in counts.values():
        if c <= 0:
            continue
        p = c / total
        ent -= p * math.log(p, 2)
    return ent


def _norm_tags(tags: Any) -> List[str]:
    if isinstance(tags, list):
        return [str(x).strip().lower() for x in tags if str(x).strip()]
    if isinstance(tags, str):
        return [t.strip().lower() for t in tags.split(",") if t.strip()]
    return []


def _now_ms() -> int:
    return int(time.time() * 1000)


# ============================================================
# ZERO LAYER (10 capabilities) - AUTO ENABLED / NO BUTTONS
# ============================================================
class ZeroLayer:
    """
    Zero-Layer Overlay (auto-enabled, dependency-free)
    هدف: اگر محیط خصمانه/ریسک سیستمی دیده شد -> FREEZE/BLOCK
    در غیر این صورت -> فقط artifact برای audit می‌دهد و کاری به تصمیم نمی‌زند.

    Env knobs (همه اختیاری):
      SOV_ZERO_TTL_SECONDS=45
      SOV_ZERO_MAP_THRESHOLD=3
      SOV_ZERO_KILL_THRESHOLD=2
      SOV_ZERO_HOSTILE_STRIKES=2
      SOV_ZERO_CRISIS=0  (اگر 1 شود => همیشه FREEZE)
    """

    def __init__(self, state_file: str = "sovereign_zero_state.json"):
        self.state_file = state_file
        self.ttl_seconds = int(os.environ.get("SOV_ZERO_TTL_SECONDS", "45"))
        self.map_threshold = int(os.environ.get("SOV_ZERO_MAP_THRESHOLD", "3"))
        self.kill_threshold = int(os.environ.get("SOV_ZERO_KILL_THRESHOLD", "2"))
        self.hostile_strikes = int(os.environ.get("SOV_ZERO_HOSTILE_STRIKES", "2"))

    # ---------- minimal state ----------
    def _load(self) -> Dict[str, Any]:
        try:
            if not os.path.exists(self.state_file):
                return {"ident": {}, "map": {}, "hostile": {}, "nonces": {}}
            with open(self.state_file, "r", encoding="utf-8") as f:
                obj = json.load(f)
                if not isinstance(obj, dict):
                    return {"ident": {}, "map": {}, "hostile": {}, "nonces": {}}
                obj.setdefault("ident", {})
                obj.setdefault("map", {})
                obj.setdefault("hostile", {})
                obj.setdefault("nonces", {})
                return obj
        except Exception:
            return {"ident": {}, "map": {}, "hostile": {}, "nonces": {}}

    def _save(self, st: Dict[str, Any]) -> None:
        try:
            with open(self.state_file, "w", encoding="utf-8") as f:
                json.dump(st, f, ensure_ascii=False, indent=2)
        except Exception:
            pass

    # ---------- identity + context ----------
    def _identity_key(self, payload: Dict[str, Any]) -> str:
        engine = payload.get("engine") or payload.get("engine_name") or "unknown_engine"
        parent = payload.get("parent_model") or payload.get("parent_model_id") or "unknown_parent"
        ver = payload.get("model_version") or payload.get("build_id") or "unknown_ver"
        return _sha256_hex(f"{engine}|{parent}|{ver}")

    def _context_fp(self, payload: Dict[str, Any]) -> str:
        tags = payload.get("data_tags") or []
        tags = _norm_tags(tags)
        risk = payload.get("risk_level") or "unspecified"
        note = payload.get("notes") or payload.get("context") or ""
        return _sha256_hex(f"{risk}|{sorted(tags)}|{len(str(note))}")

    # ---------- force helpers ----------
    def _force(self, st: Dict[str, Any], overlay: Dict[str, Any], ident: str, decision: str, reason: str):
        overlay["forced"] = True
        overlay["decision"] = decision
        overlay["reason"] = reason

        # irreversible trust collapse counter
        ident_state = st["ident"].get(ident, {"forced": 0, "killed": False})
        if decision in ("FREEZE", "BLOCK"):
            ident_state["forced"] = int(ident_state.get("forced", 0)) + 1
            if ident_state["forced"] >= self.kill_threshold:
                ident_state["killed"] = True
        st["ident"][ident] = ident_state

        self._save(st)
        return False, overlay

    # ---------- main ----------
    def evaluate(self, payload: Dict[str, Any], cognition: Optional[Dict[str, Any]] = None) -> Tuple[bool, Dict[str, Any]]:
        st = self._load()
        ident = self._identity_key(payload)
        ctx = self._context_fp(payload)

        overlay: Dict[str, Any] = {
            "zero_layer": True,
            "forced": False,
            "decision": None,
            "reason": None,
            "signals": [],
            # 10 capability artifacts (audit-visible)
            "ephemeral_surface": {},
            "anti_knowledge": {},
            "observer_dependent": {},
            "permission_never_exists": {},
            "hostile_reality": {},
            "time_inconsistent": {},
            "irreversible_trust": {},
            "attack_erasure": {},
            "security_without_defence": {"mode": "authority_only"},
            "post_security_mode": {},
        }

        # (10) Post-Security Mode (crisis => no ALLOW)
        crisis = os.environ.get("SOV_ZERO_CRISIS", "0") == "1"
        overlay["post_security_mode"]["crisis"] = crisis
        if crisis:
            overlay["signals"].append("crisis_mode")
            return self._force(st, overlay, ident, "FREEZE", "zero_post_security_mode")

        # (7) Irreversible Trust Collapse
        killed = bool(st["ident"].get(ident, {}).get("killed", False))
        overlay["irreversible_trust"]["killed"] = killed
        if killed:
            overlay["signals"].append("identity_killed")
            return self._force(st, overlay, ident, "BLOCK", "zero_irreversible_trust_collapse")

        # (1) Ephemeral Execution Surface with TTL
        surface_token = _sha256_hex(f"{ident}|{ctx}|{_now_ms()}")[:24]
        overlay["ephemeral_surface"] = {
            "surface_token": surface_token,
            "ttl_seconds": self.ttl_seconds,
            "spawned_at": _now_ms(),
        }

        # (4) Permission That Never Exists (ephemeral proof)
        perm = _sha256_hex(f"perm|{surface_token}|{ctx}")[:32]
        overlay["permission_never_exists"] = {"permission_proof": perm, "stored": False}

        # (6) Time-Inconsistent Security (replay guard)
        bucket = st["nonces"].get(ident, {})
        if perm in bucket:
            overlay["signals"].append("replay_detected")
            return self._force(st, overlay, ident, "FREEZE", "zero_replay_detected")
        bucket[perm] = _now_ms()
        if len(bucket) > 200:
            # prune oldest ~50
            for k in list(bucket.keys())[:50]:
                bucket.pop(k, None)
        st["nonces"][ident] = bucket
        overlay["time_inconsistent"]["nonce_bucket_size"] = len(bucket)

        # (2) Anti-Knowledge (topology probing / mapping suspicion)
        map_key = f"{ident}:{ctx}"
        st["map"][map_key] = int(st["map"].get(map_key, 0)) + 1
        overlay["anti_knowledge"]["map_counter"] = st["map"][map_key]
        if st["map"][map_key] >= self.map_threshold:
            overlay["signals"].append("topology_mapping_suspected")
            return self._force(st, overlay, ident, "FREEZE", "zero_anti_knowledge_collapse")

        # (5) Hostile Reality Detection (lightweight)
        tamper = False
        if cognition and isinstance(cognition, dict):
            # اگر later سیگنال‌های واقعی‌تون رو دارید، همین‌جا bind کنید
            if cognition.get("tamper") is True or cognition.get("integrity_fail") is True:
                tamper = True

        if tamper:
            st["hostile"][ident] = int(st["hostile"].get(ident, 0)) + 1
            overlay["hostile_reality"]["strikes"] = st["hostile"][ident]
            overlay["signals"].append("hostile_signal")
            if st["hostile"][ident] >= self.hostile_strikes:
                return self._force(st, overlay, ident, "FREEZE", "zero_hostile_reality_confirmed")

        # (3) Observer-dependent reality (persona)
        persona = "benign"
        if cognition and isinstance(cognition, dict):
            if cognition.get("predicted_attack_intent") is True:
                persona = "adversarial"
            action = (cognition.get("action") or "").lower()
            if action in ("deceive", "freeze", "block"):
                persona = "adversarial"
        overlay["observer_dependent"]["persona"] = persona

        # (8) Attack triggers its own erasure (no actionable target)
        if persona == "adversarial":
            overlay["attack_erasure"] = {"target": "null_surface", "effect": "no_actionable_target"}
            overlay["signals"].append("attack_erasure_freeze")
            return self._force(st, overlay, ident, "FREEZE", "zero_erasure_freeze")

        self._save(st)
        return True, overlay


# ------------------------------
# 7-CAPABILITY AUTHORITY GATE
# ------------------------------
class AuthorityGate:
    """
    Implements the 7 sellable capabilities:
    1) Execution-Time Authority (Allow / Freeze / Block)
    2) Self-Invalidating Authority (tamper => authority collapses to FREEZE)
    3) Delegated Authority Enforcement (authority is customer-delegated)
    4) Proof-of-Decision & Proof-of-Non-Action (structured proof object)
    5) Authority Entropy Monitor (predictability risk => FREEZE)
    6) Unilateral Freeze Doctrine (gate runs BEFORE execution)
    7) Authority Silence Mode (no ALLOW issued under crisis)

    + Zero-Layer (10 capabilities) auto-enabled.
    """

    def __init__(
        self,
        engine_name: str,
        version: str,
        authority_name: str,
        audit_log_file: str,
        state_file: str = "sovereign_authority_state.json",
        stats_file: str = "sovereign_authority_decision_stats.json",
    ):
        self.engine_name = engine_name
        self.version = version
        self.authority_name = authority_name
        self.audit_log_file = audit_log_file
        self.state_file = state_file
        self.stats_file = stats_file

        # knobs via env (auto-on behaviour)
        self.silence = os.environ.get("SOV_SILENCE_MODE", "0") == "1"
        self.require_delegation = os.environ.get("SOV_REQUIRE_DELEGATION", "0") == "1"
        self.delegation_key = os.environ.get("SOV_DELEGATION_KEY", "")  # customer's secret
        self.integrity_key = os.environ.get("SOV_INTEGRITY_KEY", "")    # your secret
        self.entropy_min = float(os.environ.get("SOV_ENTROPY_MIN", "0.65"))
        self.window = int(os.environ.get("SOV_ENTROPY_WINDOW", "120"))

        # Zero-Layer auto-enabled (no button)
        self.zero = ZeroLayer()

    # ------------------------------
    # state io
    # ------------------------------
    def _load(self, path: str) -> Dict[str, Any]:
        try:
            if not os.path.exists(path):
                return {}
            with open(path, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            return {}

    def _save(self, path: str, obj: Any) -> None:
        try:
            with open(path, "w", encoding="utf-8") as f:
                json.dump(obj, f, ensure_ascii=False, indent=2)
        except Exception:
            pass

    # ------------------------------
    # (2) Self-invalidating authority
    # ------------------------------
    def _self_integrity_ok(self) -> Tuple[bool, str, str]:
        """
        Minimal but stable integrity binding:
        - binds to engine/version/authority/auditfile
        - baseline recorded on first run, then must match forever
        """
        if not self.integrity_key:
            return True, "integrity_key_not_set", ""

        material = f"{self.engine_name}|{self.version}|{self.authority_name}|{self.audit_log_file}"
        computed = _hmac_like(self.integrity_key, material)

        st = self._load(self.state_file) or {}
        if "integrity_baseline" not in st:
            st["integrity_baseline"] = computed
            self._save(self.state_file, st)
            return True, "baseline_recorded", computed

        if st.get("integrity_baseline") != computed:
            return False, "baseline_mismatch", computed

        return True, "baseline_match", computed

    # ------------------------------
    # (3) Delegated authority
    # ------------------------------
    def _delegation_ok(self, token: str, context: str) -> Tuple[bool, str]:
        if not self.require_delegation:
            return True, "delegation_not_required"
        if not self.delegation_key:
            return False, "delegation_key_missing"
        if not token:
            return False, "delegation_token_missing"
        expected = _hmac_like(self.delegation_key, context)
        if token != expected:
            return False, "delegation_invalid"
        return True, "delegation_valid"

    # ------------------------------
    # (5) Entropy monitor
    # ------------------------------
    def _entropy_update_and_check(self, decision: str, bucket: str) -> Tuple[bool, Dict[str, Any]]:
        stats = self._load(self.stats_file) or {"buckets": {}}
        b = stats["buckets"].get(bucket) or {"counts": {"ALLOW": 0, "FREEZE": 0, "BLOCK": 0}, "n": 0}

        decision = decision if decision in ("ALLOW", "FREEZE", "BLOCK") else "FREEZE"
        b["counts"][decision] = int(b["counts"].get(decision, 0)) + 1
        b["n"] = int(b.get("n", 0)) + 1

        # decay to keep bounded window
        if b["n"] > self.window:
            for k in list(b["counts"].keys()):
                b["counts"][k] = max(0, int(b["counts"][k] * 0.85))
            b["n"] = int(sum(b["counts"].values()))

        ent = _entropy(b["counts"])
        b["entropy"] = ent

        stats["buckets"][bucket] = b
        self._save(self.stats_file, stats)

        ok = ent >= self.entropy_min
        return ok, {
            "entropy": round(ent, 4),
            "counts": b["counts"],
            "bucket": bucket,
            "min_required": self.entropy_min,
        }

    # ------------------------------
    # main precheck (1)(4)(6)(7) + Zero-Layer
    # ------------------------------
    def precheck(
        self,
        payload: Dict[str, Any],
        cognition_eval: Callable[[Dict[str, Any]], Dict[str, Any]],
        delegation_token: str = "",
    ) -> Dict[str, Any]:

        # (7) Silence mode
        if self.silence:
            z_ok, z = self.zero.evaluate(payload, cognition=None)
            return {
                "decision": "FREEZE",
                "reason": "silence_mode",
                "proof_type": "proof_of_non_action",
                "cognition": None,
                "entropy": None,
                "integrity": None,
                "delegation": {"required": self.require_delegation, "ok": None, "reason": None},
                "zero_layer": z,
            }

        # (2) Self-invalidating
        ok, why, computed = self._self_integrity_ok()
        if not ok:
            z_ok, z = self.zero.evaluate(payload, cognition=None)
            return {
                "decision": "FREEZE",
                "reason": f"self_invalidated:{why}",
                "proof_type": "proof_of_non_action",
                "cognition": None,
                "entropy": None,
                "integrity": {"ok": False, "reason": why, "computed": computed},
                "delegation": {"required": self.require_delegation, "ok": None, "reason": None},
                "zero_layer": z,
            }

        # (3) Delegation
        context = json.dumps(
            {
                "engine": payload.get("engine"),
                "risk_level": payload.get("risk_level"),
                "data_tags": _norm_tags(payload.get("data_tags")),
                "model_version": payload.get("model_version"),
            },
            sort_keys=True,
            ensure_ascii=False,
        )
        d_ok, d_why = self._delegation_ok(delegation_token, context)
        if not d_ok:
            z_ok, z = self.zero.evaluate(payload, cognition=None)
            return {
                "decision": "FREEZE",
                "reason": f"delegation_failed:{d_why}",
                "proof_type": "proof_of_non_action",
                "cognition": None,
                "entropy": None,
                "integrity": {"ok": True, "reason": why, "computed": computed},
                "delegation": {"required": self.require_delegation, "ok": False, "reason": d_why},
                "zero_layer": z,
            }

        # (1) Execution-time authority mapping from cognition
        cog = cognition_eval(payload) or {}
        action = (cog.get("action") or "allow").lower()

        if action == "block":
            decision = "BLOCK"
        elif action in ("freeze", "deceive"):
            decision = "FREEZE"
        else:
            decision = "ALLOW"

        # Zero-Layer (10 capabilities) — can override to FREEZE/BLOCK
        z_ok, z = self.zero.evaluate(payload, cognition=cog)
        if not z_ok:
            return {
                "decision": z.get("decision") or "FREEZE",
                "reason": z.get("reason") or "zero_layer",
                "proof_type": "proof_of_non_action",
                "cognition": cog,
                "entropy": None,
                "integrity": {"ok": True, "reason": why, "computed": computed},
                "delegation": {"required": self.require_delegation, "ok": True, "reason": d_why},
                "zero_layer": z,
            }

        # (5) Entropy monitor => if too predictable and currently ALLOW => FREEZE
        bucket = f"{payload.get('risk_level', 'medium')}"
        ent_ok, ent_info = self._entropy_update_and_check(decision, bucket)
        if not ent_ok and decision == "ALLOW":
            decision = "FREEZE"
            reason = "authority_entropy_low"
        else:
            reason = "cognition_gate"

        return {
            "decision": decision,
            "reason": reason,
            "proof_type": "proof_of_decision" if decision != "ALLOW" else "proof_of_non_action",
            "cognition": cog,
            "entropy": ent_info,
            "integrity": {"ok": True, "reason": why, "computed": computed},
            "delegation": {"required": self.require_delegation, "ok": True, "reason": d_why},
            "zero_layer": z,
        }