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