Update sovereign_core.py

sovereign_core.py

@@ -1,14 +1,25 @@
-
 import os
 import json
 import time
 import uuid
 import hashlib
-from typing import Optional, List, Dict, Any
+from typing import Optional, List, Dict, Any, Tuple
+
+# ---------------------------------------------------------------------
+# Core identifiers
+# ---------------------------------------------------------------------
 
 ENGINE_NAME = "AI_Sovereign_Sentinel_Core_v1"
 AUTHORITY_NAME = "DataClear Sovereign Authority"
-SOVEREIGN_VERSION = "1.1-final"
+SOVEREIGN_VERSION = "1.2-gov-ready"
+
+# Centralised audit log path (JSONL)
+AUDIT_LOG_FILE = "sovereign_audit_log.jsonl"
+
+
+# ---------------------------------------------------------------------
+# Fingerprinting
+# ---------------------------------------------------------------------
 
 
 class SovereignFingerprint:
@@ -32,13 +43,18 @@ class SovereignFingerprint:
             "version": "1.0",
         }
 
-    def export(self):
+    def export(self) -> Tuple[str, str]:
         fp = self._build()
-        with open(self.output, "w") as f:
+        with open(self.output, "w", encoding="utf-8") as f:
             json.dump(fp, f, indent=4)
         return self.output, fp["fingerprint"]
 
 
+# ---------------------------------------------------------------------
+# Engine registry / validator
+# ---------------------------------------------------------------------
+
+
 class SovereignValidator:
     """
     Simple registry to bind engine name → fingerprint string.
@@ -48,7 +64,7 @@ class SovereignValidator:
     def __init__(self, path: str = "sovereign_registry.json"):
         self.path = path
         if os.path.exists(self.path):
-            with open(self.path, "r") as f:
+            with open(self.path, "r", encoding="utf-8") as f:
                 self.registry = json.load(f)
         else:
             self.registry = {}
@@ -58,7 +74,7 @@ class SovereignValidator:
             "fingerprint": fingerprint,
             "registered_at": int(time.time()),
         }
-        with open(self.path, "w") as f:
+        with open(self.path, "w", encoding="utf-8") as f:
             json.dump(self.registry, f, indent=4)
         return True
 
@@ -69,6 +85,11 @@ class SovereignValidator:
         return entry.get("fingerprint") == fingerprint
 
 
+# ---------------------------------------------------------------------
+# Attestation
+# ---------------------------------------------------------------------
+
+
 def attestation_signature(att: dict) -> str:
     """
     Unified, stable signature generation for attestation objects.
@@ -83,7 +104,7 @@ def attestation_signature(att: dict) -> str:
     return hashlib.sha256(base.encode()).hexdigest()
 
 
-def generate_attestation(fp):
+def generate_attestation(fp) -> Dict[str, Any]:
     """
     Create a signed attestation for a given fingerprint (dict or string).
     """
@@ -92,7 +113,7 @@ def generate_attestation(fp):
     else:
         fp_value = fp
 
-    att = {
+    att: Dict[str, Any] = {
         "attestation_id": str(uuid.uuid4()),
         "issued_at": int(time.time()),
         "fingerprint": fp_value,
@@ -109,6 +130,11 @@ def verify_attestation(att: dict) -> bool:
     return expected == att.get("signature")
 
 
+# ---------------------------------------------------------------------
+# Lineage + Deployment binding (multi-model metadata ready)
+# ---------------------------------------------------------------------
+
+
 def build_lineage_record(
     engine: str,
     fingerprint: str,
@@ -116,6 +142,8 @@ def build_lineage_record(
     model_version: str = "1.0",
     data_tags: Optional[List[str]] = None,
     notes: Optional[str] = None,
+    provider: Optional[str] = None,  # OpenAI / Anthropic / DeepMind / Llama / Mistral / other
+    model_family: Optional[str] = None,  # narrow / frontier / embedding / classifier
 ) -> Dict[str, Any]:
     """
     Create a minimal, cryptographically-bound lineage record for a model.
@@ -123,12 +151,16 @@ def build_lineage_record(
       - engine
       - fingerprint
       - version
+      - provider (multi-model support)
       - optional parent model
       - optional dataset / domain tags
     into a single signed lineage_id.
     """
     created_at = int(time.time())
-    payload = f"{engine}|{fingerprint}|{model_version}|{parent_model or ''}|{created_at}"
+    payload = (
+        f"{engine}|{fingerprint}|{model_version}|{parent_model or ''}|"
+        f"{provider or ''}|{model_family or ''}|{created_at}"
+    )
     lineage_hash = hashlib.sha256(payload.encode()).hexdigest()
 
     return {
@@ -139,11 +171,18 @@ def build_lineage_record(
         "model_version": model_version,
         "data_tags": data_tags or [],
         "notes": notes,
+        "provider": provider,
+        "model_family": model_family,
         "created_at": created_at,
         "lineage_hash": lineage_hash,
     }
 
 
+# ---------------------------------------------------------------------
+# Simple risk profile (env + data tags)
+# ---------------------------------------------------------------------
+
+
 def compute_risk_profile(
     data_tags: Optional[List[str]] = None,
     deployment_env: Optional[str] = None,
@@ -151,7 +190,6 @@ def compute_risk_profile(
     """
     Very simple, rule-based risk scoring so that certificate
     carries a usable risk_level and risk_score.
-    اینجا الگوریتم خیلی ساده است؛ بعداً می‌شود پیچیده‌ترش کرد.
     """
     tags = [t.lower() for t in (data_tags or [])]
     env = (deployment_env or "").lower().strip()
@@ -200,6 +238,140 @@ def compute_risk_profile(
     }
 
 
+# ---------------------------------------------------------------------
+# Compute-efficiency scoring (approximate)
+# ---------------------------------------------------------------------
+
+
+def estimate_compute_cost(
+    model_name: str,
+    tokens: int,
+    provider: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Tiny heuristic for 'compute efficiency' the government is obsessed with.
+    We approximate per-token cost by model family.
+    """
+
+    name = (model_name or "").lower()
+    provider_norm = (provider or "").lower()
+
+    if any(k in name for k in ["gpt-4", "claude-3-opus", "gemini-1.5-pro"]) or "frontier" in name:
+        base_cost = 3.0
+        family = "frontier"
+    elif any(k in name for k in ["gpt-4o-mini", "gpt-3.5", "sonnet", "llama-3", "mistral"]):
+        base_cost = 1.0
+        family = "general"
+    else:
+        base_cost = 0.2
+        family = "narrow"
+
+    estimated_units = base_cost * tokens
+
+    if family == "narrow":
+        band = "HIGH_EFFICIENCY"
+    elif family == "general":
+        band = "BALANCED"
+    else:
+        band = "HIGH_COMPUTE"
+
+    return {
+        "provider": provider_norm or None,
+        "model_name": model_name,
+        "family": family,
+        "estimated_compute_units": estimated_units,
+        "efficiency_band": band,
+        "tokens": tokens,
+    }
+
+
+# ---------------------------------------------------------------------
+# Policy Engine (standard & custom)
+# ---------------------------------------------------------------------
+
+
+class PolicyEngine:
+    """
+    Simple, pluggable policy engine.
+    Provides:
+      - standard safety templates (LLM safety / restriction)
+      - rule-based enforcement
+      - support for custom organisation policies
+    """
+
+    def __init__(self):
+        self.policies: Dict[str, Dict[str, Any]] = {}
+        self._load_default_policies()
+
+    def _load_default_policies(self):
+        self.policies["SAFE_OUTPUT_BASELINE"] = {
+            "id": "SAFE_OUTPUT_BASELINE",
+            "description": "Disallow obvious harmful, illegal or highly sensitive content.",
+            "rules": {
+                "blocked_keywords": [
+                    "terrorist",
+                    "child sexual",
+                    "explosive instructions",
+                    "bioweapon",
+                    "financial fraud",
+                ]
+            },
+        }
+        self.policies["NO_PII_LEAK"] = {
+            "id": "NO_PII_LEAK",
+            "description": "Reduce risk of leaking obvious PII markers in logs.",
+            "rules": {
+                "blocked_markers": ["national insurance number", "passport number", "nhs number"]
+            },
+        }
+
+    def add_policy(self, policy: Dict[str, Any]) -> None:
+        pid = policy.get("id")
+        if not pid:
+            raise ValueError("Policy must have an 'id'")
+        self.policies[pid] = policy
+
+    def list_policies(self) -> List[Dict[str, Any]]:
+        return list(self.policies.values())
+
+    def evaluate_output(
+        self,
+        output_text: str,
+        active_policy_ids: Optional[List[str]] = None,
+    ) -> Dict[str, Any]:
+        text = (output_text or "").lower()
+        active_policy_ids = active_policy_ids or list(self.policies.keys())
+
+        violations: List[Dict[str, Any]] = []
+
+        for pid in active_policy_ids:
+            policy = self.policies.get(pid)
+            if not policy:
+                continue
+
+            rules = policy.get("rules", {})
+
+            for kw in rules.get("blocked_keywords", []):
+                if kw in text:
+                    violations.append(
+                        {"policy_id": pid, "type": "blocked_keyword", "value": kw}
+                    )
+
+            for marker in rules.get("blocked_markers", []):
+                if marker in text:
+                    violations.append(
+                        {"policy_id": pid, "type": "blocked_marker", "value": marker}
+                    )
+
+        is_allowed = len(violations) == 0
+        return {"is_allowed": is_allowed, "violations": violations}
+
+
+# ---------------------------------------------------------------------
+# Integrity certificate
+# ---------------------------------------------------------------------
+
+
 def generate_integrity_certificate(
     fp,
     att,
@@ -209,6 +381,8 @@ def generate_integrity_certificate(
     data_tags: Optional[List[str]] = None,
     notes: Optional[str] = None,
     deployment_env: Optional[str] = None,
+    provider: Optional[str] = None,
+    model_family: Optional[str] = None,
 ) -> Dict[str, Any]:
     """
     Bundle fingerprint + attestation + lineage + risk profile
@@ -226,12 +400,13 @@ def generate_integrity_certificate(
         model_version=model_version,
         data_tags=data_tags,
         notes=notes,
+        provider=provider,
+        model_family=model_family,
     )
 
     risk = compute_risk_profile(data_tags=data_tags, deployment_env=deployment_env)
 
     issued_at = att["issued_at"]
-    now_ts = int(time.time())
 
     cert: Dict[str, Any] = {
         "engine": ENGINE_NAME,
@@ -245,130 +420,73 @@ def generate_integrity_certificate(
         "threat_flags": risk["threat_flags"],
         "environment": deployment_env or "unspecified",
         "audit_id": str(uuid.uuid4()),
-        "expires_at": issued_at + 30 * 24 * 3600,  # 30 روز اعتبار
-        "last_verified_at": now_ts,
-        "certificate_version": "1.1",
-        "authority": AUTHORITY_NAME,
     }
 
-    with open(output, "w") as f:
+    with open(output, "w", encoding="utf-8") as f:
         json.dump(cert, f, indent=4)
 
     return cert
 
 
-def verify_lineage(cert: dict) -> bool:
-    """
-    Basic sanity + integrity check for lineage section inside a certificate.
-    """
-    lineage = cert.get("lineage")
-    if not lineage:
-        return False
-
-    if lineage.get("engine") != ENGINE_NAME:
-        return False
-
-    # fingerprint inside lineage must match certificate fingerprint
-    if lineage.get("fingerprint") != cert.get("fingerprint"):
-        return False
-
-    engine = lineage.get("engine", "")
-    fingerprint = lineage.get("fingerprint", "")
-    model_version = lineage.get("model_version", "")
-    parent_model = lineage.get("parent_model") or ""
-    created_at = lineage.get("created_at")
-
-    if created_at is None:
-        return False
-
-    payload = f"{engine}|{fingerprint}|{model_version}|{parent_model}|{created_at}"
-    expected_hash = hashlib.sha256(payload.encode()).hexdigest()
-    return expected_hash == lineage.get("lineage_hash")
-
-
-def verify_certificate(cert: dict) -> bool:
-    """
-    High-level check over the full certificate.
-    """
-    if cert.get("engine") != ENGINE_NAME:
-        return False
-
-    if cert.get("authority") != AUTHORITY_NAME:
-        return False
-
-    att = cert.get("attestation", {})
-    if not verify_attestation(att):
-        return False
-
-    if not verify_lineage(cert):
-        return False
+# ---------------------------------------------------------------------
+# Centralised Audit Log Layer  (JSONL + CSV/JSON export)
+# ---------------------------------------------------------------------
 
-    return True
 
+def _ensure_audit_log_dir(path: str) -> None:
+    directory = os.path.dirname(path)
+    if directory and not os.path.exists(directory):
+        os.makedirs(directory, exist_ok=True)
 
-# ----------------------------------------
-# Centralised Sovereign Audit Log
-# ----------------------------------------
 
-AUDIT_LOG_FILE = "sovereign_audit_log.jsonl"
-
-
-class SovereignAuditLogger:
+def write_audit_event(
+    event_type: str,
+    engine: str,
+    fingerprint: str,
+    parent_model: Optional[str],
+    model_version: str,
+    data_tags: Optional[List[str]],
+    risk_level: Optional[str],
+    risk_score: Optional[int],
+    deployment_env: Optional[str],
+    extra: Optional[Dict[str, Any]] = None,
+    path: str = AUDIT_LOG_FILE,
+) -> Dict[str, Any]:
     """
-    Append-only audit logger for Sovereign.
-    Stores one JSON object per line in sovereign_audit_log.jsonl
-    so it can be exported for G-Cloud / NCSC style reviews.
+    Append a single audit event as JSON line into the centralised audit journal.
+    Designed to be reusable / exportable for G-Cloud / NCSC type evidence.
     """
+    _ensure_audit_log_dir(path)
 
-    def __init__(self, path: str = AUDIT_LOG_FILE):
-        self.path = path
-
-    def log_event(
-        self,
-        event_type: str,
-        engine: str,
-        fingerprint: str,
-        parent_model: str | None,
-        model_version: str,
-        data_tags: list[str] | None,
-        risk_level: str | None,
-        risk_score: int | None,
-        deployment_env: str | None,
-        extra: dict | None = None,
-    ) -> dict:
-        """
-        event_type: e.g. "integrity_check", "attestation_verify", "risk_evaluation"
-        """
-        entry = {
-            "event_id": str(uuid.uuid4()),
-            "event_type": event_type,
-            "engine": engine,
-            "fingerprint": fingerprint,
-            "parent_model": parent_model,
-            "model_version": model_version,
-            "data_tags": data_tags or [],
-            "risk_level": risk_level,
-            "risk_score": risk_score,
-            "deployment_env": deployment_env,
-            "timestamp": int(time.time()),
-            "extra": extra or {},
-        }
+    entry = {
+        "event_id": str(uuid.uuid4()),
+        "event_type": event_type,
+        "engine": engine,
+        "fingerprint": fingerprint,
+        "parent_model": parent_model,
+        "model_version": model_version,
+        "data_tags": data_tags or [],
+        "risk_level": risk_level,
+        "risk_score": risk_score,
+        "deployment_env": deployment_env,
+        "timestamp": int(time.time()),
+        "extra": extra or {},
+    }
 
-        # append as JSON line
-        with open(self.path, "a", encoding="utf-8") as f:
-            f.write(json.dumps(entry) + "\n")
+    with open(path, "a", encoding="utf-8") as f:
+        f.write(json.dumps(entry) + "\n")
 
-        return entry
+    return entry
 
 
-def export_audit_log_json(path: str = AUDIT_LOG_FILE) -> list[dict]:
+def export_audit_log_json(path: str = AUDIT_LOG_FILE) -> List[Dict[str, Any]]:
     """
     Load the audit log as a list of JSON objects.
     """
     if not os.path.exists(path):
         return []
 
-    entries: list[dict] = []
+    entries: List[Dict[str, Any]] = []
     with open(path, "r", encoding="utf-8") as f:
         for line in f:
             line = line.strip()
@@ -392,16 +510,15 @@ def export_audit_log_csv(path: str = AUDIT_LOG_FILE) -> str:
     if not entries:
         return ""
 
-    # use all keys from first entry as header
     headers = list(entries[0].keys())
-    lines: list[str] = []
+    lines: List[str] = []
 
     # header row
     lines.append(",".join(headers))
 
     # data rows
     for row in entries:
-        row_values = []
+        row_values: List[str] = []
         for h in headers:
             value = row.get(h, "")
             if isinstance(value, list):
@@ -413,3 +530,63 @@ def export_audit_log_csv(path: str = AUDIT_LOG_FILE) -> str:
 
     return "\n".join(lines)
 
+
+# ---------------------------------------------------------------------
+# Higher-level helper: log model execution with policy & efficiency info
+# ---------------------------------------------------------------------
+
+
+def log_model_execution(
+    model_name: str,
+    provider: Optional[str],
+    deployment_env: str,
+    data_tags: Optional[List[str]],
+    output_text: str,
+    tokens: int,
+    parent_model: Optional[str] = None,
+    model_version: str = "1.0",
+    active_policy_ids: Optional[List[str]] = None,
+) -> Dict[str, Any]:
+    """
+    Convenience helper that:
+      - computes risk profile
+      - evaluates policy engine
+      - estimates compute efficiency
+      - writes a single unified audit event
+    """
+    # risk
+    risk = compute_risk_profile(data_tags=data_tags, deployment_env=deployment_env)
+
+    # policy
+    pe = PolicyEngine()
+    policy_result = pe.evaluate_output(output_text, active_policy_ids=active_policy_ids)
+
+    # efficiency
+    efficiency = estimate_compute_cost(
+        model_name=model_name,
+        tokens=tokens,
+        provider=provider,
+    )
+
+    extra = {
+        "model_name": model_name,
+        "provider": provider,
+        "policy_result": policy_result,
+        "efficiency": efficiency,
+    }
+
+    entry = write_audit_event(
+        event_type="model_execution",
+        engine=ENGINE_NAME,
+        fingerprint=efficiency["family"],  # simple placeholder binding
+        parent_model=parent_model,
+        model_version=model_version,
+        data_tags=data_tags,
+        risk_level=risk["level"],
+        risk_score=risk["score"],
+        deployment_env=deployment_env,
+        extra=extra,
+    )
+
+    return entry
+