Create sovereign_ultra_layer.py

sovereign_ultra_layer.py

@@ -0,0 +1,268 @@
+# 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()
+
+
+Sent from my Galaxy