sovereign_evidence_packager.py

# 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