Create sovereign_evidence_packager.py

sovereign_evidence_packager.py

@@ -0,0 +1,221 @@
+
+# sovereign_evidence_packager.py
+import os
+import json
+import hashlib
+import zipfile
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional, List, Tuple
+
+
+BUILD_FINGERPRINT_FILE = os.getenv("SOVEREIGN_BUILD_FINGERPRINT_FILE", "build_fingerprint.json")
+SBOM_LITE_FILE = os.getenv("SOVEREIGN_SBOM_LITE_FILE", "sbom_lite.json")
+POLICY_MANIFEST_DIR = os.getenv("SOVEREIGN_POLICY_MANIFEST_DIR", "policy_manifests")
+EVIDENCE_PACKAGES_DIR = os.getenv("SOVEREIGN_EVIDENCE_PACKAGES_DIR", "evidence_packages")
+
+AUDIT_LOG_FILE = os.getenv("SOVEREIGN_AUDIT_LOG_FILE", "sovereign_audit_log.jsonl")
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _sha256_bytes(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()
+
+
+def _stable_json_bytes(obj: Any) -> bytes:
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+
+
+def _file_sha256(path: str) -> Optional[str]:
+    try:
+        h = hashlib.sha256()
+        with open(path, "rb") as f:
+            for chunk in iter(lambda: f.read(1024 * 1024), b""):
+                h.update(chunk)
+        return h.hexdigest()
+    except Exception:
+        return None
+
+
+def _safe_write_json(path: str, data: Any) -> None:
+    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
+    tmp = path + ".tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2, sort_keys=True)
+    os.replace(tmp, path)
+
+
+def ensure_build_fingerprint() -> None:
+    """
+    Capability #9:
+      - Build/Release fingerprint (hashes of key files)
+      - SBOM-lite (requirements snapshot)
+    Runs once at startup (safe to call multiple times).
+    """
+    # If already exists, don't overwrite unless forced
+    if os.path.exists(BUILD_FINGERPRINT_FILE) and os.path.exists(SBOM_LITE_FILE):
+        return
+
+    # Collect hashes of key project files if present
+    candidates = [
+        "app.py",
+        "sovereign_core.py",
+        "sovereign_ultra_layer.py",
+        "sovereign_infinity_layer.py",
+        "pilot_suite.py",
+        "report_generator.py",
+        "dockerfile",
+        "Dockerfile",
+        "requirements.txt",
+    ]
+
+    files_hashed = []
+    for name in candidates:
+        if os.path.exists(name) and os.path.isfile(name):
+            files_hashed.append({
+                "file": name,
+                "sha256": _file_sha256(name),
+            })
+
+    build_fp = {
+        "issued_at": _utc_now_iso(),
+        "build_id": os.getenv("SOVEREIGN_BUILD_ID", "UNSPECIFIED"),
+        "version": os.getenv("SOVEREIGN_VERSION", "UNSPECIFIED"),
+        "files": files_hashed,
+    }
+    _safe_write_json(BUILD_FINGERPRINT_FILE, build_fp)
+
+    # SBOM-lite: requirements snapshot
+    req_lines: List[str] = []
+    if os.path.exists("requirements.txt"):
+        try:
+            with open("requirements.txt", "r", encoding="utf-8") as f:
+                req_lines = [ln.strip() for ln in f.readlines() if ln.strip() and not ln.strip().startswith("#")]
+        except Exception:
+            req_lines = []
+
+    sbom_lite = {
+        "issued_at": _utc_now_iso(),
+        "type": "sbom-lite",
+        "requirements": req_lines,
+        "notes": "Lightweight dependency snapshot for procurement and assurance packs.",
+    }
+    _safe_write_json(SBOM_LITE_FILE, sbom_lite)
+
+
+def write_policy_manifest(result: Dict[str, Any]) -> str:
+    """
+    Capability #8:
+      - Policy manifest snapshot (sealed hash)
+    """
+    os.makedirs(POLICY_MANIFEST_DIR, exist_ok=True)
+
+    decision_id = str(result.get("decision_id") or "unknown")
+    ts = str(result.get("timestamp") or _utc_now_iso()).replace(":", "-")
+    path = os.path.join(POLICY_MANIFEST_DIR, f"policy_manifest__{ts}__{decision_id}.json")
+
+    # Keep it minimal & procurement-friendly
+    manifest = {
+        "issued_at": _utc_now_iso(),
+        "decision_id": decision_id,
+        "authority_id": (result.get("authority") or {}).get("authority_id", "UNKNOWN"),
+        "root_authority_id": (result.get("authority") or {}).get("root_authority_id", "UNKNOWN"),
+        "jurisdiction_id": (result.get("authority") or {}).get("jurisdiction_id", "UNKNOWN"),
+        "license_id": (result.get("authority") or {}).get("license_id", "UNKNOWN"),
+        "enforcement": {
+            "authority_enforce_boundaries": os.getenv("SOVEREIGN_AUTHORITY_ENFORCE", "0") == "1",
+        },
+        "environment_policy": {
+            "allowed_envs": ["dev", "staging", "production", "restricted"],
+            "prod_pii_highrisk_action": "freeze_if_enforced",
+        },
+        "governance_states": result.get("governance_states", ["allow", "freeze", "block"]),
+    }
+
+    manifest_hash = _sha256_bytes(_stable_json_bytes(manifest))
+    manifest["manifest_hash_sha256"] = manifest_hash
+
+    _safe_write_json(path, manifest)
+    return path
+
+
+def _read_audit_tail(max_lines: int = 200) -> Optional[str]:
+    """
+    Best-effort: include a small audit tail for the evidence pack.
+    """
+    if not os.path.exists(AUDIT_LOG_FILE):
+        return None
+    try:
+        with open(AUDIT_LOG_FILE, "r", encoding="utf-8") as f:
+            lines = f.readlines()[-max_lines:]
+        return "".join(lines)
+    except Exception:
+        return None
+
+
+def export_evidence_package(result: Dict[str, Any], policy_manifest_path: Optional[str] = None) -> None:
+    """
+    Capability #10:
+      - Evidence package export (folder + zip)
+    """
+    os.makedirs(EVIDENCE_PACKAGES_DIR, exist_ok=True)
+
+    decision_id = str(result.get("decision_id") or "unknown")
+    ts = str(result.get("timestamp") or _utc_now_iso()).replace(":", "-")
+    pack_dir = os.path.join(EVIDENCE_PACKAGES_DIR, f"evidence__{ts}__{decision_id}")
+    os.makedirs(pack_dir, exist_ok=True)
+
+    # Write core artifacts
+    result_path = os.path.join(pack_dir, "result.json")
+    _safe_write_json(result_path, result)
+
+    # Copy policy manifest if provided
+    if policy_manifest_path and os.path.exists(policy_manifest_path):
+        try:
+            with open(policy_manifest_path, "r", encoding="utf-8") as f:
+                manifest = json.load(f)
+            _safe_write_json(os.path.join(pack_dir, "policy_manifest.json"), manifest)
+        except Exception:
+            pass
+
+    # Include build fingerprint and sbom-lite (if exists)
+    if os.path.exists(BUILD_FINGERPRINT_FILE):
+        try:
+            with open(BUILD_FINGERPRINT_FILE, "r", encoding="utf-8") as f:
+                build_fp = json.load(f)
+            _safe_write_json(os.path.join(pack_dir, "build_fingerprint.json"), build_fp)
+        except Exception:
+            pass
+
+    if os.path.exists(SBOM_LITE_FILE):
+        try:
+            with open(SBOM_LITE_FILE, "r", encoding="utf-8") as f:
+                sbom = json.load(f)
+            _safe_write_json(os.path.join(pack_dir, "sbom_lite.json"), sbom)
+        except Exception:
+            pass
+
+    # Best-effort include tail of audit log
+    audit_tail = _read_audit_tail()
+    if audit_tail:
+        try:
+            with open(os.path.join(pack_dir, "audit_tail.jsonl"), "w", encoding="utf-8") as f:
+                f.write(audit_tail)
+        except Exception:
+            pass
+
+    # Create zip
+    zip_path = pack_dir + ".zip"
+    try:
+        with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as z:
+            for root, _, files in os.walk(pack_dir):
+                for fn in files:
+                    full = os.path.join(root, fn)
+                    rel = os.path.relpath(full, pack_dir)
+                    z.write(full, arcname=rel)
+    except Exception:
+        # If zip fails, folder still exists
+        pass
+