Merge pull request #803 from morluto/agent/centralize-sha256-digest

Centralize SHA-256 digest ownership

Author: morluto

src/jacobian/canonical.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import re
 import unicodedata
@@ -17,6 +18,12 @@
 _DECIMAL_CHUNK_DIGITS = 9
 
 
+def sha256_digest(data: bytes) -> str:
+    """Return the canonical prefixed SHA-256 digest for immutable bytes."""
+
+    return "sha256:" + hashlib.sha256(data).hexdigest()
+
+
 class CanonicalizationError(ValueError):
     """The input cannot be represented by Jacobian's canonical JSON profile."""
 

src/jacobian/contracts/linear.py

@@ -2,12 +2,11 @@
 
 from __future__ import annotations
 
-import hashlib
 from typing import Annotated, Literal, Self
 
 from pydantic import Field, StrictInt, StringConstraints, model_validator
 
-from jacobian.canonical import canonicalize_json
+from jacobian.canonical import canonicalize_json, sha256_digest
 from jacobian.contracts.capabilities import (
     CapabilityProviderAvailability,
     CapabilityProviderRuntime,
@@ -29,14 +28,10 @@
 ]
 
 
-def _sha256(value: bytes) -> str:
-    return f"sha256:{hashlib.sha256(value).hexdigest()}"
-
-
 def linear_variable_order_digest(variables: tuple[str, ...]) -> str:
     """Bind the declared column order without inventing a generic object schema."""
 
-    return _sha256(canonicalize_json({"variables": list(variables)}))
+    return sha256_digest(canonicalize_json({"variables": list(variables)}))
 
 
 def _require_bounded_rationals(values: tuple[CanonicalRational, ...]) -> None:

src/jacobian/contracts/sat.py

@@ -4,7 +4,6 @@
 
 import base64
 import binascii
-import hashlib
 import unicodedata
 from collections.abc import Iterable, Sequence
 from typing import Annotated, Literal, Self
@@ -18,7 +17,7 @@
     model_validator,
 )
 
-from jacobian.canonical import canonicalize_json
+from jacobian.canonical import canonicalize_json, sha256_digest
 from jacobian.contracts.capabilities import (
     CapabilityProviderAvailability,
     CapabilityProviderRuntime,
@@ -52,10 +51,6 @@
 _PROOF_FORMAT_VERSION: Literal["drat-text/v1"] = "drat-text/v1"
 
 
-def _sha256(value: bytes) -> str:
-    return f"sha256:{hashlib.sha256(value).hexdigest()}"
-
-
 class SatVariableBinding(ContractModel):
     """One deterministic symbolic-name to DIMACS-ID binding."""
 
@@ -126,7 +121,7 @@ def require_canonical_instance(self) -> Self:
             raise ValueError("clauses must be unique and canonically ordered")
         if self.variable_map_digest != sat_variable_map_digest(self.variables):
             raise ValueError("variable-map digest does not match the canonical map")
-        if self.dimacs_digest != _sha256(self.to_dimacs_bytes()):
+        if self.dimacs_digest != sha256_digest(self.to_dimacs_bytes()):
             raise ValueError(
                 "DIMACS digest does not match the deterministic projection"
             )
@@ -378,7 +373,7 @@ def require_exact_raw_proof(self) -> Self:
         raw = _decode_base64(self.proof_base64)
         if self.proof_base64 != base64.b64encode(raw).decode("ascii"):
             raise ValueError("proof bytes must use canonical base64")
-        if self.proof_digest != _sha256(raw):
+        if self.proof_digest != sha256_digest(raw):
             raise ValueError("raw proof digest does not match the preserved bytes")
         _require_available_producer(self.producer)
         return self
@@ -397,7 +392,7 @@ def from_bytes(
         return cls(
             cnf=cnf,
             proof_base64=base64.b64encode(proof).decode("ascii"),
-            proof_digest=_sha256(proof),
+            proof_digest=sha256_digest(proof),
             producer=producer,
             resource_budget=resource_budget,
         )
@@ -473,7 +468,7 @@ def bind_exact_bytes(self) -> Self:
         raw = _decode_base64(self.proof_base64)
         if self.proof_base64 != base64.b64encode(raw).decode("ascii"):
             raise ValueError("LRAT proof must use canonical base64")
-        if self.proof_digest != _sha256(raw) or self.proof_byte_count != len(raw):
+        if self.proof_digest != sha256_digest(raw) or self.proof_byte_count != len(raw):
             raise ValueError("LRAT proof digest or byte count does not match")
         if len(raw) > self.limits.max_proof_bytes:
             raise ValueError("LRAT proof exceeds its declared byte limit")
@@ -490,7 +485,7 @@ def from_bytes(
         return cls(
             cnf=cnf,
             proof_base64=base64.b64encode(proof).decode("ascii"),
-            proof_digest=_sha256(proof),
+            proof_digest=sha256_digest(proof),
             proof_byte_count=len(proof),
             limits=limits,
         )
@@ -613,7 +608,7 @@ def canonicalize_cnf(
         variables=variables,
         clauses=clause_models,
         variable_map_digest=sat_variable_map_digest(variables),
-        dimacs_digest=_sha256(_dimacs_bytes(len(variables), clause_models)),
+        dimacs_digest=sha256_digest(_dimacs_bytes(len(variables), clause_models)),
     )
 
 
@@ -626,7 +621,7 @@ def sat_variable_map_digest(
         "variable_map_format": "jacobian.sat.variable-map/v1",
         "variables": [variable.model_dump(mode="json") for variable in variables],
     }
-    return _sha256(canonicalize_json(payload))
+    return sha256_digest(canonicalize_json(payload))
 
 
 def _literal_sort_key(literal: int) -> tuple[int, bool]:

src/jacobian/contracts/smt.py

@@ -4,7 +4,6 @@
 
 import base64
 import binascii
-import hashlib
 from typing import Annotated, Literal, Self
 
 from pydantic import (
@@ -15,6 +14,7 @@
     model_validator,
 )
 
+from jacobian.canonical import sha256_digest
 from jacobian.contracts.capabilities import (
     CapabilityInstallTier,
     CapabilityProviderAvailability,
@@ -54,10 +54,6 @@
 _ALETHE_HOLE_MARKER = b":rule hole"
 
 
-def _sha256(value: bytes) -> str:
-    return f"sha256:{hashlib.sha256(value).hexdigest()}"
-
-
 def _decode_base64(value: str) -> bytes:
     try:
         return base64.b64decode(value, validate=True)
@@ -270,7 +266,7 @@ class SmtProblemArtifact(ContractModel):
     @model_validator(mode="after")
     def require_exact_profile_input(self) -> Self:
         _validate_single_query_profile(self.smtlib_text, self.logic)
-        if self.smtlib_digest != _sha256(self.smtlib_text.encode("ascii")):
+        if self.smtlib_digest != sha256_digest(self.smtlib_text.encode("ascii")):
             raise ValueError("SMT-LIB digest does not match the exact input bytes")
         return self
 
@@ -284,7 +280,7 @@ def from_text(
         return cls(
             logic=logic,
             smtlib_text=smtlib_text,
-            smtlib_digest=_sha256(smtlib_text.encode("ascii")),
+            smtlib_digest=sha256_digest(smtlib_text.encode("ascii")),
         )
 
     def raw_bytes(self) -> bytes:
@@ -325,7 +321,7 @@ def require_exact_raw_proof(self) -> Self:
         proof = _decode_base64(self.proof_base64)
         if self.proof_base64 != base64.b64encode(proof).decode("ascii"):
             raise ValueError("proof bytes must use canonical base64")
-        if self.proof_digest != _sha256(proof):
+        if self.proof_digest != sha256_digest(proof):
             raise ValueError("Alethe proof digest does not match the preserved bytes")
         holes = proof.count(_ALETHE_HOLE_MARKER)
         if self.alethe_hole_count != holes or self.contains_holes != (holes > 0):
@@ -359,7 +355,7 @@ def from_bytes(
         return cls(
             problem=problem,
             proof_base64=base64.b64encode(proof).decode("ascii"),
-            proof_digest=_sha256(proof),
+            proof_digest=sha256_digest(proof),
             alethe_hole_count=holes,
             contains_holes=holes > 0,
             producer=producer,

src/jacobian/storage/blobs.py

@@ -15,13 +15,13 @@
 from contextlib import contextmanager
 from pathlib import Path
 
+from jacobian.canonical import sha256_digest
 from jacobian.storage.errors import (
     ArtifactIntegrityError,
     ArtifactNotFoundError,
     StorageError,
     StorageLimitError,
 )
-from jacobian.storage.identity import sha256_digest
 
 _LOGGER = logging.getLogger(__name__)
 

src/jacobian/storage/identity.py

@@ -6,7 +6,7 @@
 from dataclasses import dataclass
 from typing import Final
 
-from jacobian.canonical import CanonicalLimits, canonicalize_json
+from jacobian.canonical import CanonicalLimits, canonicalize_json, sha256_digest
 from jacobian.contracts.artifacts import ArtifactManifest
 from jacobian.storage.errors import ArtifactNotFoundError
 
@@ -35,10 +35,6 @@ class _ArtifactIdentity:
     manifest_bytes: bytes
 
 
-def sha256_digest(data: bytes) -> str:
-    return "sha256:" + hashlib.sha256(data).hexdigest()
-
-
 def uri_from_digest(digest: str) -> str:
     return "artifact://sha256/" + digest.removeprefix("sha256:")
 
@@ -143,6 +139,5 @@ def artifact_identity(
     "artifact_identity",
     "digest_from_uri",
     "framed_digest",
-    "sha256_digest",
     "uri_from_digest",
 ]

tests/unit/contracts/test_canonical_json.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import hashlib
+
 import pytest
 from hypothesis import given, settings
 from hypothesis import strategies as st
@@ -9,10 +11,16 @@
     CanonicalizationError,
     CanonicalLimits,
     canonicalize_json,
+    sha256_digest,
 )
 from jacobian.contracts.exact import CanonicalRational, require_bounded_rational
 
 
+def test_sha256_digest_uses_the_canonical_prefixed_format() -> None:
+    for value in (b"", b"\x00", b"jacobian", b"\xde\xad\xbe\xef" * 100):
+        assert sha256_digest(value) == "sha256:" + hashlib.sha256(value).hexdigest()
+
+
 def test_equivalent_rationals_have_identical_canonical_bytes() -> None:
     first = canonicalize_json({"weight": {"num": "2", "den": "4"}})
     second = canonicalize_json({"weight": {"num": "1", "den": "2"}})