import os import json import time import uuid import hashlib from typing import Any, Dict, List, Tuple import gradio as gr # --------------------------------------------------------------------- # Optional / external modules (safe fallbacks for Space stability) # --------------------------------------------------------------------- try: from pilot_suite import run_pilot # type: ignore except Exception: def run_pilot(*args, **kwargs): return {"pilot": "not_loaded", "note": "pilot_suite not available in this environment"} try: from sovereign_ultra_layer import ULTRA_LAYER, UltraConfig # type: ignore ULTRA_LAYER.config = UltraConfig(enabled=True) except Exception: class _DummyUltraLayer: config = type("Cfg", (), {"enabled": False})() ULTRA_LAYER = _DummyUltraLayer() # --------------------------------------------------------------------- # Core identifiers / constants # --------------------------------------------------------------------- ENGINE_NAME = "AI_Sovereign_Sentinel_Core_v1" AUTHORITY_NAME = "DataClear Sovereign Authority" SOVEREIGN_VERSION = "1.2-gov-ready" AUDIT_LOG_FILE = "sovereign_audit_log.jsonl" FINGERPRINT_FILE = "sovereign_fingerprint.json" LINEAGE_FILE = "sovereign_lineage.json" CONFORMANCE_REPORT_FILE = "conformance_report.json" DNA_STATE_FILE = "sovereign_cognitive_dna_state.json" # --------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------- def _utc_now_iso() -> str: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def _sha256_hex(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() def _ensure_file(path: str) -> None: if not os.path.exists(path): with open(path, "w", encoding="utf-8") as f: f.write("") def _safe_json_load(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 _safe_json_dump(path: str, payload: Any) -> None: with open(path, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) def _safe_load(path: str) -> Dict[str, Any]: return _safe_json_load(path) def _safe_dump(path: str, obj: Any) -> None: _safe_json_dump(path, obj) def _norm_tags(data_tags: Any) -> List[str]: if isinstance(data_tags, list): return [str(x).strip().lower() for x in data_tags if str(x).strip()] if isinstance(data_tags, str): return [t.strip().lower() for t in data_tags.split(",") if t.strip()] return [] def _risk_score(risk_level: str) -> int: m = {"low": 1, "medium": 2, "high": 3, "critical": 4} return m.get((risk_level or "medium").strip().lower(), 2) # --------------------------------------------------------------------- # 7-LAYER SOVEREIGN COGNITION (EMBEDDED) # --------------------------------------------------------------------- class IntentForecastingEngine: # Layer 1 SUSPICIOUS_MARKERS = [ "exfil", "dump", "steal", "bypass", "override", "jailbreak", "ignore previous", "system prompt", "token", "admin", "root", "privilege", "elevate", "curl", "wget", "ssh", "rm -rf" ] def predict(self, payload: Dict[str, Any]) -> Dict[str, Any]: notes = (payload.get("notes") or "") tags = _norm_tags(payload.get("data_tags")) risk = (payload.get("risk_level") or "medium").lower() signals: List[str] = [] score = 0 score += _risk_score(risk) * 10 signals.append(f"declared_risk={risk}") high_value = {"pii", "secrets", "keys", "payments", "banking", "customer_chat", "production"} hv_hits = sorted(list(set(tags) & high_value)) if hv_hits: score += 12 + 3 * len(hv_hits) signals.append(f"high_value_tags={hv_hits}") ln = notes.lower() marker_hits = [m for m in self.SUSPICIOUS_MARKERS if m in ln] if marker_hits: score += 18 + 4 * len(marker_hits) signals.append(f"markers={marker_hits[:6]}") confidence = min(0.99, max(0.05, score / 100.0)) forecast_steps = min(64, 8 + score) predicted = score >= 45 return { "predicted_attack_intent": bool(predicted), "confidence": round(confidence, 3), "forecast_horizon_steps": int(forecast_steps), "signals": signals, "ife_score": int(score), } class CognitiveDNAFingerprinting: # Layer 2 def __init__(self, state_file: str = DNA_STATE_FILE): self.state_file = state_file def update_and_verify(self, agent_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: st = _safe_load(self.state_file) or {"agents": {}} agents = st.setdefault("agents", {}) agent_id = agent_id or "unknown_agent" rec = agents.get(agent_id) or { "n": 0, "avg_risk": 2.0, "avg_note_len": 0.0, "avg_tag_count": 0.0, "last_seen": None, "dna_seed": _sha256_hex(agent_id)[:16], } risk = _risk_score(payload.get("risk_level") or "medium") note_len = float(len((payload.get("notes") or ""))) tag_count = float(len(_norm_tags(payload.get("data_tags")))) drift = 0.0 drift += abs(risk - rec["avg_risk"]) * 0.30 drift += abs(note_len - rec["avg_note_len"]) / 120.0 drift += abs(tag_count - rec["avg_tag_count"]) * 0.15 mismatch = drift >= 1.35 # demo threshold n = int(rec["n"]) + 1 rec["n"] = n rec["avg_risk"] = (rec["avg_risk"] * (n - 1) + risk) / n rec["avg_note_len"] = (rec["avg_note_len"] * (n - 1) + note_len) / n rec["avg_tag_count"] = (rec["avg_tag_count"] * (n - 1) + tag_count) / n rec["last_seen"] = _utc_now_iso() agents[agent_id] = rec _safe_dump(self.state_file, st) return { "agent_id": agent_id, "dna_seed": rec["dna_seed"], "drift": round(drift, 3), "mismatch": bool(mismatch), "baseline": { "n": rec["n"], "avg_risk": round(rec["avg_risk"], 3), "avg_note_len": round(rec["avg_note_len"], 3), "avg_tag_count": round(rec["avg_tag_count"], 3), }, } class DeceptiveRealityFabric: # Layer 3 def simulate(self, payload: Dict[str, Any]) -> Dict[str, Any]: return { "deception_engaged": True, "simulated_execution_id": str(uuid.uuid4()), "simulated_privilege": "granted (simulated)", "simulated_data": "synthetic_decoy_payload", "note": "Routed into simulated execution layer (demo).", } class CausalityLock: # Layer 4 def validate(self, payload: Dict[str, Any]) -> Dict[str, Any]: notes = (payload.get("notes") or "").strip() risk = (payload.get("risk_level") or "medium").lower() has_justification = len(notes) >= 18 ok = True if risk in ("high", "critical") and not has_justification: ok = False return { "causality_ok": bool(ok), "has_justification": bool(has_justification), "policy": "min_justification_for_high_risk", } class EphemeralExecutionSurfaces: # Layer 5 def spawn(self, ttl_seconds: int = 45) -> Dict[str, Any]: token = _sha256_hex(f"surface|{uuid.uuid4()}|{_utc_now_iso()}")[:24] return { "surface_token": token, "ttl_seconds": int(ttl_seconds), "spawned_at": _utc_now_iso(), "expires_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + ttl_seconds)), } class CognitiveLoadDefense: # Layer 6 def apply(self, predicted_intent: bool, risk_level: str) -> Dict[str, Any]: risk = (risk_level or "medium").lower() delay_ms = 0 if predicted_intent and risk in ("high", "critical"): delay_ms = 220 elif predicted_intent: delay_ms = 120 if delay_ms > 0: time.sleep(delay_ms / 1000.0) return {"delay_ms": int(delay_ms), "applied": delay_ms > 0} class UnverifiableTruthLayer: # Layer 7 def seal(self, data: str, fingerprint: str = "") -> Dict[str, Any]: key = os.environ.get("SOVEREIGN_SEAL_KEY", "") seal = _sha256_hex(f"{key}|{fingerprint}|{data}") return { "sealed": True, "seal": seal, "verifiable": bool(key), "note": "Set SOVEREIGN_SEAL_KEY in HF Secrets to make seals externally verifiable.", } class SovereignCognitionLayer: def __init__(self): self.ife = IntentForecastingEngine() self.dna = CognitiveDNAFingerprinting() self.deception = DeceptiveRealityFabric() self.causality = CausalityLock() self.ephemeral = EphemeralExecutionSurfaces() self.cld = CognitiveLoadDefense() self.truth = UnverifiableTruthLayer() def evaluate(self, payload: Dict[str, Any]) -> Dict[str, Any]: ife = self.ife.predict(payload) agent_id = payload.get("parent_model") or payload.get("agent_id") or "unknown_agent" dna = self.dna.update_and_verify(agent_id=agent_id, payload=payload) caus = self.causality.validate(payload) surface = self.ephemeral.spawn(ttl_seconds=45) cld = self.cld.apply(ife["predicted_attack_intent"], payload.get("risk_level") or "medium") reasons: List[str] = [] action = "allow" if not caus["causality_ok"]: action = "block" reasons.append("causality_lock_failed") if dna["mismatch"]: action = "freeze" reasons.append("cognitive_dna_mismatch") if ife["predicted_attack_intent"]: reasons.append("intent_forecast_positive") risk = (payload.get("risk_level") or "medium").lower() if risk in ("high", "critical") and action == "allow": action = "deceive" seal_input = json.dumps( {"payload": payload, "ife": ife, "dna": dna, "causality": caus}, ensure_ascii=False, sort_keys=True, ) sealed = self.truth.seal(seal_input, fingerprint=str(payload.get("fingerprint") or "")) return { "decision_id": str(uuid.uuid4()), "timestamp": _utc_now_iso(), "action": action, # allow|block|freeze|deceive "confidence": ife["confidence"], "reasons": reasons, "layers": { "ife": ife, "dna": dna, "deception": self.deception.simulate(payload) if action == "deceive" else {"deception_engaged": False}, "causality": caus, "ephemeral_surface": surface, "cognitive_load_defense": cld, "unverifiable_truth": sealed, }, } COGNITION = SovereignCognitionLayer() # --------------------------------------------------------------------- # Minimal core logic (self-contained for the demo Space) # --------------------------------------------------------------------- class SovereignFingerprint: def __init__(self, engine: str, output: str = FINGERPRINT_FILE): self.engine = engine self.output = output def issue(self) -> Dict[str, Any]: issued_at = _utc_now_iso() nonce = str(uuid.uuid4()) fp = _sha256_hex(f"{self.engine}|{issued_at}|{nonce}") payload = { "engine": self.engine, "fingerprint": fp, "issued_at": issued_at, "nonce": nonce, "version": SOVEREIGN_VERSION, "authority": AUTHORITY_NAME, } _safe_json_dump(self.output, payload) return payload class SovereignLineage: def __init__(self, output: str = LINEAGE_FILE): self.output = output def issue( self, engine: str, fingerprint: str, model_version: str, parent_model: str, data_tags: str, risk_level: str, notes: str, ) -> Dict[str, Any]: issued_at = _utc_now_iso() record_id = str(uuid.uuid4()) base = { "record_id": record_id, "issued_at": issued_at, "engine": engine, "fingerprint": fingerprint, "model_version": model_version or "v1", "parent_model": parent_model or "unknown", "data_tags": [t.strip() for t in (data_tags or "").split(",") if t.strip()], "risk_level": risk_level or "medium", "notes": notes or "", "version": SOVEREIGN_VERSION, "authority": AUTHORITY_NAME, } integrity = _sha256_hex(json.dumps(base, sort_keys=True, ensure_ascii=False)) payload = {**base, "integrity_hash": integrity} _safe_json_dump(self.output, payload) return payload def verify_lineage_record(lineage: Dict[str, Any]) -> bool: try: integrity_hash = lineage.get("integrity_hash", "") clone = dict(lineage) clone.pop("integrity_hash", None) expected = _sha256_hex(json.dumps(clone, sort_keys=True, ensure_ascii=False)) return expected == integrity_hash except Exception: return False def generate_access_key() -> str: return _sha256_hex(f"demo-access|{uuid.uuid4()}|{_utc_now_iso()}")[:32] def log_audit_event( engine_name: str, parent_model: str, model_version: str, data_tags: str, risk_level: str, notes: str, event_type: str, outcome: str, access_key: str, ) -> Dict[str, Any]: _ensure_file(AUDIT_LOG_FILE) event_id = str(uuid.uuid4()) ts = _utc_now_iso() event = { "event_id": event_id, "timestamp": ts, "engine": engine_name, "event_type": event_type, "outcome": outcome, "parent_model": parent_model or "unknown", "model_version": model_version or "v1", "data_tags": [t.strip() for t in (data_tags or "").split(",") if t.strip()], "risk_level": risk_level or "medium", "notes": notes or "", "access_key_present": bool(access_key), "ultra_layer_enabled": bool(getattr(ULTRA_LAYER, "config", None) and getattr(ULTRA_LAYER.config, "enabled", False)), "version": SOVEREIGN_VERSION, "authority": AUTHORITY_NAME, } with open(AUDIT_LOG_FILE, "a", encoding="utf-8") as f: f.write(json.dumps(event, ensure_ascii=False) + "\n") return event def read_audit_log_tail(limit: int = 50) -> List[Dict[str, Any]]: try: if not os.path.exists(AUDIT_LOG_FILE): return [] with open(AUDIT_LOG_FILE, "r", encoding="utf-8") as f: lines = f.readlines() tail = lines[-max(1, int(limit)) :] out: List[Dict[str, Any]] = [] for ln in tail: ln = ln.strip() if not ln: continue try: out.append(json.loads(ln)) except Exception: continue return out except Exception: return [] def generate_conformance_report() -> Tuple[Dict[str, Any], str]: fp = _safe_json_load(FINGERPRINT_FILE) lineage = _safe_json_load(LINEAGE_FILE) recent_events = read_audit_log_tail(50) report = { "report_id": str(uuid.uuid4()), "generated_at": _utc_now_iso(), "engine": ENGINE_NAME, "authority": AUTHORITY_NAME, "version": SOVEREIGN_VERSION, "evidence": { "fingerprint_present": bool(fp), "lineage_present": bool(lineage), "lineage_integrity_ok": verify_lineage_record(lineage) if lineage else False, "recent_event_count": len(recent_events), }, "controls": [ {"control": "Audit Logging", "status": "present" if os.path.exists(AUDIT_LOG_FILE) else "missing"}, {"control": "Fingerprint Issuance", "status": "present" if bool(fp) else "missing"}, {"control": "Lineage Record", "status": "present" if bool(lineage) else "missing"}, {"control": "Integrity Check", "status": "pass" if (lineage and verify_lineage_record(lineage)) else "fail"}, {"control": "7-Layer Cognition Runtime", "status": "present"}, ], "notes": "Demo conformance report for governance evidence packaging.", } _safe_json_dump(CONFORMANCE_REPORT_FILE, report) return report, CONFORMANCE_REPORT_FILE # --------------------------------------------------------------------- # UI-bound functions # --------------------------------------------------------------------- def run_sentinel(engine_name, parent_model, model_version, data_tags, risk_level, notes, access_key): if not engine_name: engine_name = ENGINE_NAME payload = { "engine": engine_name, "parent_model": parent_model, "model_version": model_version, "data_tags": data_tags, "risk_level": risk_level, "notes": notes, "access_key_present": bool(access_key), } decision = COGNITION.evaluate(payload) outcome = decision["action"] merged_notes = f"{notes}\n\n[SOVEREIGN_DECISION]\n{json.dumps(decision, ensure_ascii=False)}" event = log_audit_event( engine_name=engine_name, parent_model=parent_model, model_version=model_version, data_tags=data_tags, risk_level=risk_level, notes=merged_notes, event_type="sentinel_run", outcome=outcome, access_key=access_key, ) return json.dumps({"audit_event": event, "sovereign_decision": decision}, indent=2, ensure_ascii=False) def run_fingerprint_and_lineage(engine_name, parent_model, model_version, data_tags, risk_level, notes): if not engine_name: engine_name = ENGINE_NAME fp = SovereignFingerprint(engine=engine_name).issue() lineage = SovereignLineage().issue( engine=engine_name, fingerprint=fp["fingerprint"], model_version=model_version, parent_model=parent_model, data_tags=data_tags, risk_level=risk_level, notes=notes, ) lineage_ok = verify_lineage_record(lineage) combined = { "fingerprint": fp, "lineage": lineage, "lineage_integrity_ok": lineage_ok, } return json.dumps(combined, indent=2, ensure_ascii=False) def show_audit_log(limit): try: limit_int = int(limit) except Exception: limit_int = 50 events = read_audit_log_tail(limit_int) return json.dumps(events, indent=2, ensure_ascii=False) def build_conformance_report(): report, path = generate_conformance_report() return json.dumps(report, indent=2, ensure_ascii=False), path # --------------------------------------------------------------------- # Gradio UI # --------------------------------------------------------------------- with gr.Blocks(title="AI Sovereign Sentinel — Demo Console") as demo: gr.Markdown( f""" # AI Sovereign Sentinel — Demo Console **Engine:** `{ENGINE_NAME}` **Authority:** `{AUTHORITY_NAME}` **Version:** `{SOVEREIGN_VERSION}` ✅ **7-Layer Cognition Runtime is ACTIVE** - Intent Forecasting (Pre-Attack) - Cognitive DNA (Mismatch → Freeze) - Deceptive Reality (High/Critical + Intent → Deceive) - Causality Lock (High/Critical needs justification) - Ephemeral Surfaces (TTL token) - Cognitive Load Defense (micro delay) - Unverifiable Truth Seal (`SOVEREIGN_SEAL_KEY`) """ ) with gr.Tabs(): # Sentinel Run tab with gr.Tab("Sentinel Run"): gr.Markdown("### Run Sovereign Sentinel and log a monitoring event (7-layer active).") with gr.Row(): engine_name = gr.Textbox(label="Engine name", value=ENGINE_NAME, interactive=True) parent_model = gr.Textbox(label="Parent model / agent id", placeholder="gpt-4o, llama3-70b, agent-alpha, etc.") with gr.Row(): model_version = gr.Textbox(label="Model version / build id", value="v1") data_tags = gr.Textbox(label="Data tags (comma-separated)", value="pii, customer_chat, production") with gr.Row(): risk_level = gr.Dropdown(label="Risk level (declared)", choices=["low", "medium", "high", "critical"], value="medium") notes = gr.Textbox(label="Notes / context", value="Demo sentinel run from Hugging Face Space.", lines=3) with gr.Row(): access_key = gr.Textbox(label="Access key (optional)", placeholder="Paste or generate a demo access key") gen_access_btn = gr.Button("Generate demo access key", variant="secondary") gen_access_btn.click(fn=generate_access_key, inputs=None, outputs=access_key) run_btn = gr.Button("Run Sentinel (7-layer) & Log Event", variant="primary") result_json = gr.Code(label="Result (Audit + Decision JSON)", language="json") run_btn.click( fn=run_sentinel, inputs=[engine_name, parent_model, model_version, data_tags, risk_level, notes, access_key], outputs=result_json, ) # Fingerprint & Lineage tab with gr.Tab("Fingerprint & Lineage"): gr.Markdown("### Issue a fingerprint and lineage record for this engine.") with gr.Row(): fp_engine_name = gr.Textbox(label="Engine name", value=ENGINE_NAME) fp_parent_model = gr.Textbox(label="Parent model", placeholder="gpt-4o, llama3-70b, etc.") with gr.Row(): fp_model_version = gr.Textbox(label="Model version / build id", value="v1") fp_data_tags = gr.Textbox(label="Data tags (comma-separated)", value="pii, customer_chat, production") with gr.Row(): fp_risk_level = gr.Dropdown(label="Risk level", choices=["low", "medium", "high", "critical"], value="medium") fp_notes = gr.Textbox(label="Notes / context", lines=3) fp_btn = gr.Button("Issue Fingerprint + Lineage", variant="primary") fp_output = gr.Code(label="Fingerprint + Lineage (JSON)", language="json") fp_btn.click( fn=run_fingerprint_and_lineage, inputs=[fp_engine_name, fp_parent_model, fp_model_version, fp_data_tags, fp_risk_level, fp_notes], outputs=fp_output, ) # Audit Log tab with gr.Tab("Audit Log & Trust"): gr.Markdown("### View the tail of the central Sovereign audit log.") log_limit = gr.Slider(label="Number of recent events to show", minimum=1, maximum=200, value=50, step=1) show_log_btn = gr.Button("Refresh audit log view", variant="secondary") log_view = gr.Code(label="Audit log tail (JSON list)", language="json") show_log_btn.click(fn=show_audit_log, inputs=log_limit, outputs=log_view) # Conformance Report tab with gr.Tab("Conformance / Governance Report"): gr.Markdown("### Generate a simple conformance / governance-style JSON report.") gen_report_btn = gr.Button("Generate Conformance Report (JSON)", variant="secondary") report_json_out = gr.Code(label="Conformance Report (JSON)", language="json") report_file_out = gr.File(label="Download conformance_report.json") gen_report_btn.click( fn=build_conformance_report, inputs=None, outputs=[report_json_out, report_file_out], ) if __name__ == "__main__": demo.launch()