| """Canonical JSON and content-addressed objects for the TOM world kernel.""" |
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| from pathlib import Path |
| from typing import Any, Mapping |
|
|
| HASH_PREFIX = "sha256:" |
|
|
|
|
| def canonical_bytes(value: Any) -> bytes: |
| """Return the profile's deterministic UTF-8 JSON representation.""" |
|
|
| return json.dumps( |
| value, |
| sort_keys=True, |
| ensure_ascii=False, |
| separators=(",", ":"), |
| allow_nan=False, |
| ).encode("utf-8") |
|
|
|
|
| def digest_bytes(data: bytes, *, prefix: bool = True) -> str: |
| digest = hashlib.sha256(data).hexdigest() |
| return HASH_PREFIX + digest if prefix else digest |
|
|
|
|
| def digest_file(path: str | Path, *, prefix: bool = True) -> str: |
| h = hashlib.sha256() |
| with Path(path).open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| h.update(chunk) |
| digest = h.hexdigest() |
| return HASH_PREFIX + digest if prefix else digest |
|
|
|
|
| def content_hash(record: Mapping[str, Any]) -> str: |
| body = {key: value for key, value in record.items() if key != "content_hash"} |
| return digest_bytes(canonical_bytes(body)) |
|
|
|
|
| def attach_hash(record: Mapping[str, Any]) -> dict[str, Any]: |
| result = dict(record) |
| result["content_hash"] = content_hash(result) |
| return result |
|
|
|
|
| def verify_hash(record: Mapping[str, Any]) -> bool: |
| return isinstance(record.get("content_hash"), str) and record["content_hash"] == content_hash(record) |
|
|
|
|
| def write_canonical(path: str | Path, value: Any, *, terminal_newline: bool = True) -> None: |
| data = canonical_bytes(value) + (b"\n" if terminal_newline else b"") |
| destination = Path(path) |
| destination.parent.mkdir(parents=True, exist_ok=True) |
| destination.write_bytes(data) |
|
|
