gabriel / musehub public
hash_utils.py python
135 lines 5.0 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago
1 """Deterministic contract hashing for execution lineage verification.
2
3 Rules:
4 - Structural fields participate in hashes.
5 - Advisory / meta fields are excluded.
6 - Serialization is canonical: sorted keys, no whitespace, json.dumps.
7 - Hash is SHA-256 with ``sha256:`` prefix — full 256-bit collision resistance.
8 - No MD5, no pickle, no repr().
9
10 Excluded fields (advisory / meta / visual / runtime):
11 contract_version, contract_hash, parent_contract_hash, execution_hash,
12 l2_generate_prompt, region_name, gm_guidance,
13 assigned_color, existing_track_id.
14 """
15
16 import dataclasses
17 import json
18 from typing import TYPE_CHECKING
19
20 from muse.core.types import blob_id
21 from musehub.types.json_types import JSONObject, JSONValue
22
23 if TYPE_CHECKING:
24 from _typeshed import DataclassInstance
25
26 _HASH_EXCLUDED_FIELDS = frozenset({
27 "contract_version",
28 "contract_hash",
29 "parent_contract_hash",
30 "execution_hash",
31 "l2_generate_prompt",
32 "region_name",
33 "gm_guidance",
34 "assigned_color",
35 "existing_track_id",
36 })
37
38 def _normalize_value(value: DataclassInstance | JSONValue) -> JSONValue:
39 """Recursively normalize a value for canonical serialization."""
40 if dataclasses.is_dataclass(value) and not isinstance(value, type):
41 return canonical_contract_dict(value)
42 if isinstance(value, dict):
43 return {k: _normalize_value(v) for k, v in sorted(value.items())}
44 if isinstance(value, (list, tuple)):
45 return [_normalize_value(item) for item in value]
46 if isinstance(value, (int, float, str, bool, type(None))):
47 return value
48 return str(value)
49
50 def canonical_contract_dict(obj: DataclassInstance) -> JSONObject:
51 """Convert a frozen dataclass to a canonical ordered dict for hashing.
52
53 Excludes advisory/meta fields defined in ``_HASH_EXCLUDED_FIELDS``.
54 Recursively normalizes nested dataclasses and collections.
55 Keys are sorted for deterministic serialization.
56
57 Special case: ``CompositionContract.sections`` is serialized as a
58 sorted list of section contract hashes (not full objects), keeping
59 the root hash compact and order-independent.
60 """
61 if not dataclasses.is_dataclass(obj):
62 raise TypeError(f"Expected a dataclass, got {type(obj).__name__}")
63
64 _is_composition = type(obj).__name__ == "CompositionContract"
65
66 result: JSONObject = {}
67 for f in dataclasses.fields(obj):
68 if f.name in _HASH_EXCLUDED_FIELDS:
69 continue
70 value = getattr(obj, f.name)
71 if _is_composition and f.name == "sections":
72 result["sections"] = sorted(
73 getattr(s, "contract_hash", "") for s in value
74 )
75 else:
76 result[f.name] = _normalize_value(value)
77
78 return dict(sorted(result.items()))
79
80 def compute_contract_hash(obj: DataclassInstance) -> str:
81 """Compute a deterministic ``sha256:``-prefixed hash of structural contract fields."""
82 canonical = canonical_contract_dict(obj)
83 serialized = json.dumps(canonical, separators=(",", ":"), sort_keys=True)
84 return blob_id(serialized.encode("utf-8"))
85
86 def seal_contract(obj: DataclassInstance, parent_hash: str = "") -> None:
87 """Compute and set ``contract_hash`` on a frozen dataclass.
88
89 Uses ``object.__setattr__`` to bypass frozen enforcement.
90 Optionally sets ``parent_contract_hash`` if provided.
91 """
92 if parent_hash:
93 object.__setattr__(obj, "parent_contract_hash", parent_hash)
94 h = compute_contract_hash(obj)
95 object.__setattr__(obj, "contract_hash", h)
96
97 def set_parent_hash(obj: DataclassInstance, parent_hash: str) -> None:
98 """Set ``parent_contract_hash`` on a frozen dataclass without unfreezing it.
99
100 Uses ``object.__setattr__`` to bypass the ``frozen=True`` restriction.
101 Call this when linking a child contract to its parent before sealing the
102 child with ``seal_contract``.
103 """
104 object.__setattr__(obj, "parent_contract_hash", parent_hash)
105
106 def verify_contract_hash(obj: DataclassInstance) -> bool:
107 """Recompute hash and compare to the stored ``contract_hash``.
108
109 Returns ``True`` if the stored hash matches the recomputed hash.
110 """
111 stored = getattr(obj, "contract_hash", "")
112 if not stored:
113 return False
114 return compute_contract_hash(obj) == stored
115
116 def hash_list_canonical(items: list[str]) -> str:
117 """Collision-proof parent hash from a list of child hashes.
118
119 Sorts lexicographically, JSON-encodes the sorted list, then
120 SHA-256 hashes the result. Returns a ``sha256:``-prefixed canonical ID.
121
122 This replaces the old ``SHA256("hashA:hashB")`` pattern which
123 was vulnerable to delimiter collisions.
124 """
125 serialized = json.dumps(sorted(items))
126 return blob_id(serialized.encode("utf-8"))
127
128 def compute_execution_hash(contract_hash: str, trace_id: str) -> str:
129 """Bind an execution to a specific contract + session.
130
131 Prevents replay attacks: same contract in a different session
132 produces a different execution_hash. Returns a ``sha256:``-prefixed ID.
133 """
134 payload = (contract_hash + trace_id).encode("utf-8")
135 return blob_id(payload)
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 121 days ago