import os import json import time import uuid import hashlib import gradio as gr from pilot_suite import run_pilot # --------------------------------------------------------------------- # 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" # --------------------------------------------------------------------- # Minimal core logic (self-contained for the demo Space) # --------------------------------------------------------------------- class SovereignFingerprint: """ Minimal fingerprint generator for AI Sovereign Sentinel. Creates a JSON file with engine, fingerprint and issued_at. """ def __init__(self, engine: str, output: str = FINGERPRINT_FILE): self.engine = engine self.output = output def _build(self): issued_at = int(time.time()) base = f"{self.engine}|{issued_at}" fp_hash = hashlib.sha256(base.encode()).hexdigest() return { "engine": self.engine, "fingerprint": fp_hash, "issued_at": issued_at, "version": SOVEREIGN_VERSION, "authority": AUTHORITY_NAME, } def issue(self): payload = self._build() with open(self.output, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, indent=2) return payload class SovereignLineage: """ Minimal lineage record, bound to a fingerprint and model metadata. """ def __init__(self, output: str = LINEAGE_FILE): self.output = output def _build( self, engine: str, fingerprint: str, model_version: str, parent_model: str, data_tags: str, risk_level: str, notes: str, extra=None, ): issued_at = int(time.time()) payload_str = "|".join( [ engine, fingerprint, model_version or "", parent_model or "", data_tags or "", risk_level or "", str(issued_at), ] ) lineage_hash = hashlib.sha256(payload_str.encode()).hexdigest() entry = { "engine": engine, "fingerprint": fingerprint, "model_version": model_version, "parent_model": parent_model, "data_tags": data_tags.split(",") if data_tags else [], "risk_level": risk_level, "notes": notes, "issued_at": issued_at, "lineage_hash": lineage_hash, "extra": extra or {}, } return entry def issue( self, engine: str, fingerprint: str, model_version: str, parent_model: str, data_tags: str, risk_level: str, notes: str, extra=None, ): entry = self._build( engine=engine, fingerprint=fingerprint, model_version=model_version, parent_model=parent_model, data_tags=data_tags, risk_level=risk_level, notes=notes, extra=extra, ) with open(self.output, "w", encoding="utf-8") as f: json.dump(entry, f, ensure_ascii=False, indent=2) return entry def verify_lineage_record(entry: dict) -> bool: """ Simple integrity check for a lineage record. Recomputes lineage_hash and compares. """ try: engine = entry.get("engine", "") fingerprint = entry.get("fingerprint", "") model_version = entry.get("model_version", "") parent_model = entry.get("parent_model", "") data_tags = ",".join(entry.get("data_tags", [])) risk_level = entry.get("risk_level", "") issued_at = entry.get("issued_at", "") payload_str = "|".join( [ engine, fingerprint, model_version or "", parent_model or "", data_tags or "", risk_level or "", str(issued_at), ] ) expected_hash = hashlib.sha256(payload_str.encode()).hexdigest() return entry.get("lineage_hash") == expected_hash except Exception: return False 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 = "", extra=None, ): """ Append a JSONL event into the central Sovereign audit log. """ os.makedirs(os.path.dirname(AUDIT_LOG_FILE) or ".", exist_ok=True) entry = { "event_id": str(uuid.uuid4()), "engine": engine_name, "parent_model": parent_model, "model_version": model_version, "data_tags": data_tags.split(",") if data_tags else [], "risk_level": risk_level, "notes": notes, "event_type": event_type, "outcome": outcome, "access_key": access_key, "timestamp": int(time.time()), "extra": extra or {}, } with open(AUDIT_LOG_FILE, "a", encoding="utf-8") as f: f.write(json.dumps(entry) + "\n") return entry def read_audit_log_tail(limit: int = 50): """ Return the last N events from the audit log (if it exists). """ if not os.path.exists(AUDIT_LOG_FILE): return [] with open(AUDIT_LOG_FILE, "r", encoding="utf-8") as f: lines = f.readlines()[-limit:] events = [] for line in lines: line = line.strip() if not line: continue try: events.append(json.loads(line)) except json.JSONDecodeError: continue return events def generate_conformance_report(): """ Minimal conformance-style report summarising the audit log. """ events = read_audit_log_tail(1000) total_events = len(events) by_risk = {} by_parent = {} for e in events: risk = e.get("risk_level", "unknown") parent = e.get("parent_model", "unknown") by_risk[risk] = by_risk.get(risk, 0) + 1 by_parent[parent] = by_parent.get(parent, 0) + 1 report = { "engine": ENGINE_NAME, "version": SOVEREIGN_VERSION, "authority": AUTHORITY_NAME, "generated_at": int(time.time()), "total_events": total_events, "events_by_risk_level": by_risk, "events_by_parent_model": by_parent, } with open(CONFORMANCE_REPORT_FILE, "w", encoding="utf-8") as f: json.dump(report, f, ensure_ascii=False, indent=2) return report, CONFORMANCE_REPORT_FILE def generate_access_key(): """ Simple demo access key generator. """ return "demo-" + uuid.uuid4().hex[:12] # --------------------------------------------------------------------- # Gradio function wrappers # --------------------------------------------------------------------- def run_sentinel( engine_name, parent_model, model_version, data_tags, risk_level, notes, access_key, ): if not engine_name: engine_name = ENGINE_NAME 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=notes, event_type="sentinel_run", outcome="logged", access_key=access_key, ) return json.dumps(event, 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}` This Space is a lightweight console for the Sovereign Sentinel core. You can use it to: - Run a Sentinel monitoring event and write into the central audit log - Issue a cryptographic-style fingerprint for the engine - Issue and verify a minimal lineage record - Inspect the JSONL audit log - Generate a simple conformance / governance-style JSON report """ ) with gr.Tabs(): # ------------------------------------------------------------- # Sentinel Run tab # ------------------------------------------------------------- with gr.Tab("Sentinel Run"): gr.Markdown( "### Run Sovereign Sentinel and log a monitoring event." ) with gr.Row(): engine_name = gr.Textbox( label="Engine name", value=ENGINE_NAME, interactive=True, ) parent_model = gr.Textbox( label="Parent model (e.g. gpt-4o, llama3-70b)", placeholder="gpt-4o, llama3-70b, 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 & Log Event", variant="primary", ) result_json = gr.Code( label="Result (JSON event)", 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 & Trust 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 / Governance 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()