Create sovereign_crypto_seal.py

sovereign_crypto_seal.py

@@ -0,0 +1,323 @@
+
+# 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))