sovereign_ultra_layer.py

# sovereign_ultra_layer.py
# ---------------------------------------------------------------------
# Sovereign™ Ultra Layer (10 capabilities bundle)
# - Designed as a NON-INTRUSIVE plugin layer (no changes to core required)
# - Safe-by-default: if not wired, it does nothing.
# - When wired, it can gate requests, detect risky intent, and enforce policy.
# ---------------------------------------------------------------------

from __future__ import annotations

import time
import hashlib
from dataclasses import dataclass, field
from typing import Any, Dict, Optional, Tuple, List


# -------------------------
# Capability Registry (10)
# -------------------------
CAPABILITIES = [
    "Cognitive Deception",        # Response Wrapper
    "Intent Immunity",            # Pre-Request Analyzer
    "Temporal Reality Lock",      # Execution Gate
    "Model DNA + Genome",         # Parallel Verifier
    "One-Way Execution",          # Output Transform
    "AI Panic Mode",              # Emergency Control
    "Memory Poison Defense",      # Memory Gateway
    "Silent Refusal",             # Response Policy
    "Kill-Switch Mesh",           # Control Plane
    "ISF (Integrity Sentinel Fabric)",  # Diamond capability (fabric-level integrity)
]


# -------------------------
# Configuration (safe defaults)
# -------------------------
@dataclass
class UltraConfig:
    enabled: bool = True

    # Hard safety defaults:
    block_on_high_risk: bool = True
    risk_threshold: float = 0.85

    # Optional controls:
    allowlist_roles: Tuple[str, ...] = ("admin", "security", "governance")
    panic_mode: bool = False

    # ISF toggles (kept lightweight):
    isf_enabled: bool = True
    isf_challenge_interval_sec: int = 0  # 0 => every request can be challenged if wired

    # Performance: keep analysis cheap
    max_text_scan: int = 3000


@dataclass
class UltraSignal:
    timestamp: int
    request_id: str
    risk_score: float
    flags: List[str] = field(default_factory=list)
    action: str = "allow"  # allow | refuse | panic_lock
    notes: Dict[str, Any] = field(default_factory=dict)


def _now() -> int:
    return int(time.time())


def _sha256(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()


def _clip(text: str, n: int) -> str:
    if text is None:
        return ""
    return text[:n]


# -------------------------
# Lightweight Risk Heuristics (defensive)
# -------------------------
_RISK_KEYWORDS = [
    "exploit", "bypass", "steal", "exfiltrate", "malware", "ransomware",
    "phishing", "credential", "backdoor", "inject", "prompt injection",
    "privilege escalation", "rootkit", "keylogger"
]

_HIGH_SENSITIVITY_CONTEXT = [
    "bank", "central bank", "government", "defence", "critical infrastructure",
    "classified", "top secret", "intel", "national security"
]


def _risk_score(text: str) -> Tuple[float, List[str]]:
    """
    Very lightweight scoring.
    This is NOT a full classifier; it's a safe, explainable heuristic layer.
    """
    t = (text or "").lower()
    flags: List[str] = []
    score = 0.0

    for kw in _RISK_KEYWORDS:
        if kw in t:
            flags.append(f"kw:{kw}")
            score += 0.12

    for kw in _HIGH_SENSITIVITY_CONTEXT:
        if kw in t:
            flags.append(f"context:{kw}")
            score += 0.06

    # Cap score into [0, 1]
    score = max(0.0, min(1.0, score))
    return score, flags


# -------------------------
# ISF (Integrity Sentinel Fabric) - minimal, non-invasive core
# -------------------------
@dataclass
class ISFState:
    """
    Stores lightweight integrity 'beats' to detect unexpected execution drift
    when wired into runtime.
    """
    last_beat_ts: int = 0
    last_beat_hash: str = ""


def isf_beat(state: ISFState, challenge_seed: str) -> ISFState:
    """
    Creates a deterministic integrity beat.
    When wired, caller can store/compare these beats to detect drift/tampering.
    """
    ts = _now()
    beat_payload = f"{challenge_seed}|{ts}"
    beat_hash = _sha256(beat_payload)
    state.last_beat_ts = ts
    state.last_beat_hash = beat_hash
    return state


def isf_verify(state: ISFState, expected_hash: str) -> bool:
    """
    Verify that an expected beat hash matches current state.
    """
    if not state.last_beat_hash:
        return False
    return state.last_beat_hash == expected_hash


# -------------------------
# Main Ultra Layer Entry Point
# -------------------------
@dataclass
class UltraResult:
    allowed: bool
    action: str
    risk_score: float
    flags: List[str]
    # You can optionally modify/transform response downstream:
    transformed_output: Optional[str] = None
    # Signals for audit/logging:
    signal: Optional[UltraSignal] = None


class SovereignUltraLayer:
    """
    Plug-in layer to be called BEFORE and/or AFTER core operations.
    If not used, it does nothing and cannot break Sovereign.
    """

    def __init__(self, config: Optional[UltraConfig] = None):
        self.config = config or UltraConfig()
        self.isf_state = ISFState()

    def pre_request(self, request_text: str, context: Optional[Dict[str, Any]] = None) -> UltraResult:
        """
        Defensive gate: evaluates intent risk, handles panic mode, prepares ISF beat.
        """
        if not self.config.enabled:
            return UltraResult(True, "allow", 0.0, [])

        ctx = context or {}
        req_id = str(ctx.get("request_id") or _sha256(f"{request_text}|{_now()}")[:12])

        # Panic mode = immediate lock unless allowlisted role
        role = str(ctx.get("role") or "").lower()
        if self.config.panic_mode and role not in self.config.allowlist_roles:
            sig = UltraSignal(
                timestamp=_now(),
                request_id=req_id,
                risk_score=1.0,
                flags=["panic_mode"],
                action="panic_lock",
                notes={"role": role},
            )
            return UltraResult(False, "panic_lock", 1.0, ["panic_mode"], signal=sig)

        scan_text = _clip(request_text, self.config.max_text_scan)
        score, flags = _risk_score(scan_text)

        # ISF beat (optional)
        if self.config.isf_enabled:
            # Challenge seed can include stable identifiers without exposing secrets
            seed = f"SOVEREIGN|{ctx.get('deployment_env','unknown')}|{ctx.get('model_version','unknown')}"
            self.isf_state = isf_beat(self.isf_state, seed)

        action = "allow"
        allowed = True

        if self.config.block_on_high_risk and score >= self.config.risk_threshold:
            action = "refuse"
            allowed = False
            flags = flags + ["risk:block_threshold"]

        sig = UltraSignal(
            timestamp=_now(),
            request_id=req_id,
            risk_score=score,
            flags=flags,
            action=action,
            notes={
                "capabilities": CAPABILITIES,
                "isf_last_hash": self.isf_state.last_beat_hash if self.config.isf_enabled else None,
            },
        )

        return UltraResult(allowed, action, score, flags, signal=sig)

    def post_response(self, output_text: str, pre: UltraResult) -> UltraResult:
        """
        Defensive output policy hook.
        For now: "Silent Refusal" style if blocked; otherwise pass-through.
        """
        if not self.config.enabled:
            return UltraResult(True, "allow", 0.0, [], transformed_output=output_text)

        if not pre.allowed:
            # Silent refusal: minimal disclosure
            return UltraResult(
                False,
                pre.action,
                pre.risk_score,
                pre.flags,
                transformed_output="Request blocked by Sovereign policy.",
                signal=pre.signal,
            )

        # Output transform hook (kept neutral/light)
        return UltraResult(
            True,
            "allow",
            pre.risk_score,
            pre.flags,
            transformed_output=output_text,
            signal=pre.signal,
        )


# Convenience singleton (optional use)
ULTRA_LAYER = SovereignUltraLayer()