sovereign_crypto_seal.py

# sovereign_crypto_seal.py
from __future__ import annotations

import hashlib
import hmac
import json
import os
import secrets
import time
from dataclasses import dataclass, asdict
from typing import Any, Dict, Optional


DEFAULT_SECRET_ENV = "SOVEREIGN_SEAL_SECRET"
DEFAULT_ALGO = "sha256"
DEFAULT_SCHEME = "hmac-sha256-v1"


def _utc_ts() -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def _canonical_json(obj: Any) -> str:
    return json.dumps(
        obj,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    )


def _sha256_hex(data: str) -> str:
    return hashlib.sha256(data.encode("utf-8")).hexdigest()


def _resolve_secret(explicit_secret: Optional[str] = None) -> str:
    if explicit_secret:
        return explicit_secret

    env_secret = os.getenv(DEFAULT_SECRET_ENV, "").strip()
    if env_secret:
        return env_secret

    # deterministic fallback would be dangerous; use generated process-local secret
    # so the system remains functional in dev/test, but caller should set env in prod
    return "dev-only-" + secrets.token_hex(32)


@dataclass
class SovereignSeal:
    scheme: str
    algorithm: str
    created_at: str
    object_type: str
    object_digest: str
    object_size: int
    signer_hint: str
    seal_digest: str
    metadata: Dict[str, Any]

    def as_dict(self) -> Dict[str, Any]:
        return asdict(self)


class SovereignCryptoSeal:
    """
    Lightweight cryptographic sealing utility for Sovereign artifacts.

    Capabilities:
    - canonical digest of arbitrary JSON-serializable objects
    - HMAC-based sealing
    - verification of digest + seal
    - manifest generation for evidence packages

    Intended use:
    - audit artifacts
    - benchmark reports
    - freeze cases
    - runtime outputs
    - technical identity packages
    """

    def __init__(
        self,
        secret: Optional[str] = None,
        signer_hint: str = "sovereign-local-seal",
    ) -> None:
        self.secret = _resolve_secret(secret)
        self.signer_hint = signer_hint

    # ---------------------------------------------------------
    # Core digest functions
    # ---------------------------------------------------------
    def canonicalize(self, obj: Any) -> str:
        return _canonical_json(obj)

    def object_digest(self, obj: Any) -> str:
        canonical = self.canonicalize(obj)
        return _sha256_hex(canonical)

    def _seal_digest(self, object_digest: str, object_type: str, metadata: Optional[Dict[str, Any]] = None) -> str:
        payload = {
            "scheme": DEFAULT_SCHEME,
            "algorithm": DEFAULT_ALGO,
            "object_type": object_type,
            "object_digest": object_digest,
            "metadata": metadata or {},
        }
        msg = _canonical_json(payload).encode("utf-8")
        return hmac.new(self.secret.encode("utf-8"), msg, hashlib.sha256).hexdigest()

    # ---------------------------------------------------------
    # Public seal / verify
    # ---------------------------------------------------------
    def seal_object(
        self,
        obj: Any,
        object_type: str,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        canonical = self.canonicalize(obj)
        digest = _sha256_hex(canonical)

        seal = SovereignSeal(
            scheme=DEFAULT_SCHEME,
            algorithm=DEFAULT_ALGO,
            created_at=_utc_ts(),
            object_type=str(object_type).strip() or "unknown_object",
            object_digest=digest,
            object_size=len(canonical.encode("utf-8")),
            signer_hint=self.signer_hint,
            seal_digest=self._seal_digest(
                object_digest=digest,
                object_type=object_type,
                metadata=metadata,
            ),
            metadata=metadata or {},
        )

        return {
            "ok": True,
            "sealed": True,
            "object_type": object_type,
            "canonical": canonical,
            "seal": seal.as_dict(),
        }

    def verify_object(
        self,
        obj: Any,
        seal: Dict[str, Any],
    ) -> Dict[str, Any]:
        if not isinstance(seal, dict):
            return {
                "ok": False,
                "verified": False,
                "error": "invalid_seal_format",
            }

        object_type = str(seal.get("object_type", "unknown_object"))
        metadata = seal.get("metadata") if isinstance(seal.get("metadata"), dict) else {}

        canonical = self.canonicalize(obj)
        digest = _sha256_hex(canonical)

        if digest != seal.get("object_digest"):
            return {
                "ok": False,
                "verified": False,
                "error": "object_digest_mismatch",
                "expected_digest": seal.get("object_digest"),
                "actual_digest": digest,
            }

        expected_seal_digest = self._seal_digest(
            object_digest=digest,
            object_type=object_type,
            metadata=metadata,
        )

        if expected_seal_digest != seal.get("seal_digest"):
            return {
                "ok": False,
                "verified": False,
                "error": "seal_digest_mismatch",
                "expected_seal_digest": expected_seal_digest,
                "actual_seal_digest": seal.get("seal_digest"),
            }

        return {
            "ok": True,
            "verified": True,
            "object_type": object_type,
            "object_digest": digest,
            "seal_digest": expected_seal_digest,
        }

    # ---------------------------------------------------------
    # Convenience helpers
    # ---------------------------------------------------------
    def build_manifest(
        self,
        *,
        manifest_type: str,
        subject_name: str,
        subject_version: str,
        artifacts: Dict[str, Any],
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        artifact_entries = {}
        for name, obj in artifacts.items():
            artifact_entries[name] = {
                "digest": self.object_digest(obj),
                "type": type(obj).__name__,
            }

        manifest = {
            "manifest_type": manifest_type,
            "subject_name": subject_name,
            "subject_version": subject_version,
            "generated_at": _utc_ts(),
            "artifacts": artifact_entries,
            "metadata": metadata or {},
        }

        sealed = self.seal_object(
            manifest,
            object_type=f"manifest:{manifest_type}",
            metadata={
                "subject_name": subject_name,
                "subject_version": subject_version,
            },
        )

        return {
            "manifest": manifest,
            "seal": sealed["seal"],
        }

    def seal_runtime_result(
        self,
        result: Dict[str, Any],
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        return self.seal_object(
            result,
            object_type="runtime_result",
            metadata=metadata or {},
        )

    def seal_freeze_case(
        self,
        freeze_case: Dict[str, Any],
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        return self.seal_object(
            freeze_case,
            object_type="freeze_case",
            metadata=metadata or {},
        )

    def seal_benchmark_report(
        self,
        report: Dict[str, Any],
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        return self.seal_object(
            report,
            object_type="benchmark_report",
            metadata=metadata or {},
        )


DEFAULT_SEALER = SovereignCryptoSeal()


def seal_object(
    obj: Any,
    object_type: str,
    metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    return DEFAULT_SEALER.seal_object(obj, object_type, metadata)


def verify_object(obj: Any, seal: Dict[str, Any]) -> Dict[str, Any]:
    return DEFAULT_SEALER.verify_object(obj, seal)


def build_manifest(
    *,
    manifest_type: str,
    subject_name: str,
    subject_version: str,
    artifacts: Dict[str, Any],
    metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    return DEFAULT_SEALER.build_manifest(
        manifest_type=manifest_type,
        subject_name=subject_name,
        subject_version=subject_version,
        artifacts=artifacts,
        metadata=metadata,
    )


if __name__ == "__main__":
    sample = {
        "decision": "FREEZE",
        "reason": "high-risk payment context",
        "risk_score": 91,
    }

    sealed = seal_object(
        sample,
        "runtime_result",
        metadata={"environment": "banking", "stage": "demo"},
    )
    print(json.dumps(sealed, indent=2, ensure_ascii=False))

    verified = verify_object(sample, sealed["seal"])
    print(json.dumps(verified, indent=2, ensure_ascii=False))