app.py

import os
import json
import time
import uuid
import hashlib

import gradio as gr

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