From a193394ea89bef77d0c035c576291f9bded8b77f Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 26 May 2026 08:41:57 -0400 Subject: [PATCH] #000061: bind pack_hash to full pack content (tables + chunks) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Caught a real defect on the live v2 corpus run: pack_hash was computed from the chunk-only manifest, so two packs with the same chunk set but different table content collided on pack_hash. Observed live: v1 packs and v2 packs for the same shard produced the SAME pack_hash: v1 (chunks-only) b3c427baaf9b40e3c33ace8438b4254e3abdc7b38bdcb7b1a8012588cb32848a v2 (tables + chunks) b3c427baaf9b40e3c33ace8438b4254e3abdc7b38bdcb7b1a8012588cb32848a ↑ identical, different bytes Consequences if shipped: bucket overwrites swap whose tables you get, mesh peers A and B disagree on what "pack X" is while both claim to have it, content-addressing story is broken for the metadata half. Fix: manifest format v2 prepends a `{"_format_version": 2}` header, then one row per shipped table: {"table_file": "tables/audit_events.jsonl", "hash": "...", "size": N} {"table_file": "tables/chunks.jsonl", "hash": "...", "size": N} ... {"leaf_hash": "...", "size": N} # chunks come after {"leaf_hash": "...", "size": N} pack_hash = hash_leaf(manifest_bytes) now binds the full pack content. Two writers with the same shard state produce the same pack_hash; same chunks + different tables produce different pack_hashes. Implementation: - hash_file_leaf(path) — streaming sha256 with leaf 0x00 prefix, 1 MB reads, used to fingerprint multi-GB tables/.jsonl files without loading them into RAM. - stream_packs' extra_members now takes (member_name, source_path, content_hash) tuples. push_pack computes the hash with hash_file_leaf after the dump finishes. - _finalize_pack writes the format header + table refs (sorted by name for determinism) before chunk entries. - parse_manifest returns a ParsedManifest (format_version, tables[], chunks[]) instead of just list[PackEntry]. Backward-compatible with v1 manifests (no header → format_version=1, tables=()). - pull_pack pulls the manifest aside during tar walk, then after extraction verifies each tables/.jsonl file via hash_file_leaf against the manifest's content_hash. Mismatch → ValueError, no rows reach the live schema. New tests: - test_v2_pack_hash_binds_table_contents — asserts manifest format + table refs + per-table hash & size. - test_pull_pack_rejects_tampered_table — corrupts one table's bytes in a real pack, confirms pull_pack raises on hash mismatch. 27 cold-object + evict tests pass. --- arborist/cli.py | 2 +- arborist/cold_object.py | 127 +++++++++++++++++++++++++++++++++----- arborist/evict.py | 54 +++++++++++++++- tests/test_cold_object.py | 112 +++++++++++++++++++++++++++++++-- 4 files changed, 271 insertions(+), 24 deletions(-) diff --git a/arborist/cli.py b/arborist/cli.py index 7c2c370..ecc7d09 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -2929,7 +2929,7 @@ def _cmd_cold_list(args: argparse.Namespace) -> int: if args.fetch_manifest: try: manifest_bytes = backend.get(pack_key(pack_hash, manifest=True)) - entry["chunk_count"] = len(parse_manifest(manifest_bytes)) + entry["chunk_count"] = len(parse_manifest(manifest_bytes).chunks) except Exception as e: # noqa: BLE001 — surface but don't fail entry["chunk_count"] = None entry["manifest_error"] = repr(e) diff --git a/arborist/cold_object.py b/arborist/cold_object.py index a64a031..65a7ca9 100644 --- a/arborist/cold_object.py +++ b/arborist/cold_object.py @@ -61,6 +61,22 @@ MANIFEST_PREFIX = "manifest/" MANIFEST_LATEST_KEY = MANIFEST_PREFIX + "latest.ndjson" +def hash_file_leaf(path: Path, *, chunk_size: int = 1024 * 1024) -> str: + """Compute `hash_leaf(file_contents).hex()` by streaming the file in + 1 MB chunks. Used to fingerprint multi-GB table dumps for the pack + manifest without loading the file into RAM.""" + import hashlib + h = hashlib.sha256() + h.update(b"\x00") # leaf-domain prefix, matches arborist.merkle.hash_leaf + with open(path, "rb") as f: + while True: + buf = f.read(chunk_size) + if not buf: + break + h.update(buf) + return h.hexdigest() + + def pack_key(pack_hash: str, *, manifest: bool = False) -> str: """Bucket key for a pack body (default) or its manifest sidecar.""" if manifest: @@ -470,7 +486,7 @@ def stream_packs( max_compressed_bytes: int, level: int = 3, work_dir: Path | str | None = None, - extra_members: list[tuple[str, Path]] | None = None, + extra_members: list[tuple[str, Path, str]] | None = None, ) -> Iterator[FilePack]: """Stream (leaf_hash, raw_utf8_body) chunks into FILE-BACKED packs that each fit in `max_compressed_bytes` compressed bytes. @@ -505,13 +521,16 @@ def stream_packs( Single chunks larger than the cap still get their own pack (we don't drop data — the alternative is silently losing it). - `extra_members` is a list of (tar_member_name, source_file_path) for - files to include AT THE START OF THE FIRST PACK. Used by pack format - v2 to ship shard-metadata table dumps (`tables/.jsonl`) ahead - of the chunk bodies. Subsequent packs (if the chunk set overflows the - cap) do not repeat the extra members — they're conceptually shard- - level state, not chunk-level. If the v1 chunks-only pack shape is - needed, pass `extra_members=None` (default). + `extra_members` is a list of (tar_member_name, source_file_path, + content_hash) for files to include AT THE START OF THE FIRST PACK. + Used by pack format v2 to ship shard-metadata table dumps + (`tables/.jsonl`) ahead of the chunk bodies. `content_hash` + must be `hash_leaf(source_file_contents).hex()` — it goes into the + pack manifest so `pack_hash` content-addresses the *full* pack + (tables + chunks), not just chunks. Subsequent packs (if the chunk + set overflows the cap) do not repeat the extra members — they're + conceptually shard-level state, not chunk-level. If the v1 chunks- + only pack shape is needed, pass `extra_members=None` (default). """ if max_compressed_bytes <= 0: raise ValueError("max_compressed_bytes must be positive") @@ -539,11 +558,36 @@ def stream_packs( "writer": writer, "tar": tar, "entries_by_hash": {}, + "table_refs": [], # (member_name, content_hash, size) — populated for first pack only } def _finalize_pack(state: dict) -> FilePack: entries_sorted = sorted(state["entries_by_hash"].keys()) manifest_lines = [] + + # Pack format v2: tag explicitly. Consumers branch on this to + # know they should expect + verify `table_file` entries before + # the chunk entries. Absence of the header line = legacy v1 + # (chunks-only) — accepted by `parse_manifest` for read but no + # longer produced by `stream_packs`. + has_tables = bool(state["table_refs"]) + if has_tables: + manifest_lines.append( + json.dumps({"_format_version": 2}, sort_keys=True) + ) + # Table refs first, sorted by tar member name for determinism. + # content_hash binds pack_hash to the file bytes; two writers + # with same tables produce same hashes. + for member_name, content_hash, table_size in sorted(state["table_refs"]): + manifest_lines.append(json.dumps( + { + "table_file": member_name, + "hash": content_hash, + "size": table_size, + }, + sort_keys=True, + )) + entries = [] for leaf_hash in entries_sorted: size = state["entries_by_hash"][leaf_hash] @@ -577,12 +621,19 @@ def stream_packs( # instead of 4 GB per worker. TarInfo built by hand (NOT via # gettarinfo) so mtime/uid/gid are zeroed — two writers' packs # converge byte-for-byte regardless of filesystem timestamps. - for member_name, source_path in pending_extras: + # + # Each extra_member carries (tar_name, source_path, content_hash). + # The content_hash goes into the manifest so pack_hash binds the + # full pack (tables + chunks). Caller computed the hash via + # hash_file_leaf() at dump time. + for member_name, source_path, content_hash in pending_extras: info = tarfile.TarInfo(name=member_name) - info.size = source_path.stat().st_size + file_size = source_path.stat().st_size + info.size = file_size # mtime, uid, gid, mode default to 0 / "" — deterministic. with open(source_path, "rb") as f: state["tar"].addfile(info, f) + state["table_refs"].append((member_name, content_hash, file_size)) # Force the compressor to flush so size accounting reflects the # metadata write before we start streaming chunks. if pending_extras: @@ -684,12 +735,56 @@ def open_pack(body_bytes: bytes) -> Iterator[tuple[str, bytes]]: yield expected_hash, body -def parse_manifest(manifest_bytes: bytes) -> list[PackEntry]: - """Parse manifest.ndjson into PackEntry list. Lenient on blank lines.""" - out = [] +@dataclass(frozen=True) +class TableRef: + """One table file reference in a v2 pack manifest. Hash binds the + consumer's verification path: after extracting tar member + `tables/.jsonl`, the consumer must compute hash_leaf of the + bytes and compare to this `content_hash`.""" + member_name: str + content_hash: str + size: int + + +@dataclass(frozen=True) +class ParsedManifest: + """Result of parsing a pack manifest. Distinguishes v1 (chunks-only) + from v2 (tables + chunks) so consumers can branch on which + invariants to verify.""" + format_version: int # 1 (legacy) or 2 (table-bound) + tables: tuple[TableRef, ...] # empty for v1 + chunks: tuple[PackEntry, ...] + + +def parse_manifest(manifest_bytes: bytes) -> ParsedManifest: + """Parse a manifest.ndjson stream into table refs + chunk entries. + + Format detection: a line `{"_format_version": N}` at the top means + v2. Absent means legacy v1 (chunks-only). Lenient on blank lines. + """ + fmt = 1 + tables: list[TableRef] = [] + chunks: list[PackEntry] = [] for line in manifest_bytes.decode("utf-8").splitlines(): - if not line.strip(): + line = line.strip() + if not line: continue rec = json.loads(line) - out.append(PackEntry(leaf_hash=rec["leaf_hash"], size=int(rec["size"]))) - return out + if "_format_version" in rec: + fmt = int(rec["_format_version"]) + elif "table_file" in rec: + tables.append(TableRef( + member_name=rec["table_file"], + content_hash=rec["hash"], + size=int(rec["size"]), + )) + elif "leaf_hash" in rec: + chunks.append(PackEntry( + leaf_hash=rec["leaf_hash"], size=int(rec["size"]) + )) + # silently ignore unknown record shapes (forward compat) + return ParsedManifest( + format_version=fmt, + tables=tuple(tables), + chunks=tuple(chunks), + ) diff --git a/arborist/evict.py b/arborist/evict.py index 700bdf8..7b1cec0 100644 --- a/arborist/evict.py +++ b/arborist/evict.py @@ -314,7 +314,7 @@ def push_pack( import tempfile from pathlib import Path - from arborist.cold_object import stream_packs + from arborist.cold_object import hash_file_leaf, stream_packs from arborist.cold_pack_metadata import dump_shard_metadata from arborist.snapshot import compute_snapshot_root @@ -407,10 +407,19 @@ def push_pack( # of the first pack. Edges goes through fan-in restructure inside # `dump_shard_metadata` for ~5-10x compression vs the live-schema row # layout. Lives only long enough to be tarred + uploaded, then deleted. + # + # Each table file's content_hash is computed via hash_file_leaf (stream + # the file through sha256 in 1 MB chunks) and passed into stream_packs + # so the pack manifest binds pack_hash to (tables + chunks), not just + # chunks. Two writers with the same shard state → same pack_hash. metadata_dir = Path(tempfile.mkdtemp(prefix="arborist-meta-")) metadata_summary = dump_shard_metadata(conn, metadata_dir) extra_members = [ - (f"tables/{name}.jsonl", info["path"]) + ( + f"tables/{name}.jsonl", + info["path"], + hash_file_leaf(info["path"]), + ) for name, info in metadata_summary.items() ] @@ -510,6 +519,7 @@ def pull_pack( import zstandard + from arborist.cold_object import hash_file_leaf, parse_manifest from arborist.cold_pack_metadata import restore_shard_metadata from arborist.merkle import hash_leaf @@ -524,14 +534,23 @@ def pull_pack( # Extract metadata table files to a temp dir; collect chunk bodies as # we go. The tar can contain both kinds; we branch on member name. + # The manifest is read separately so we can verify each table file's + # bytes against the manifest's content_hash (v2 only) on extraction. tables_dir = Path(tempfile.mkdtemp(prefix="arborist-restore-")) chunks_to_restore: list[tuple[str, bytes]] = [] table_files_seen: list[str] = [] + manifest_bytes: bytes | None = None try: with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r") as tar: for member in tar: if member.name == "manifest.ndjson" or member.name == "manifest.jsonl": - continue # informational header, parsed elsewhere if needed + # Pull the manifest aside so we can verify table hashes + # after extraction. (Manifest itself is the integrity + # root: pack_hash = hash_leaf(manifest_bytes).) + f = tar.extractfile(member) + if f is not None: + manifest_bytes = f.read() + continue if member.name.startswith("tables/"): # tables/.jsonl — extract to temp dir for # restore_shard_metadata to read line-by-line. @@ -562,6 +581,35 @@ def pull_pack( ) chunks_to_restore.append((expected_hash, chunk_body)) + # Pack format v2: cross-check each extracted table file's bytes + # against the content_hash recorded in the manifest. Tampered or + # truncated tables fail loudly here, never reach the live schema. + # v1 packs (no manifest table refs) skip this verification. + tampered_tables: list[dict] = [] + manifest_format = 1 + if manifest_bytes is not None: + parsed = parse_manifest(manifest_bytes) + manifest_format = parsed.format_version + table_ref_by_name = {t.member_name: t for t in parsed.tables} + for member_name in table_files_seen: + ref = table_ref_by_name.get(member_name) + if ref is None: + continue # tar carries a table not in the manifest; + # forward-compat: just don't verify it. + local_path = tables_dir / Path(member_name).name + actual = hash_file_leaf(local_path) + if actual != ref.content_hash: + tampered_tables.append({ + "member": member_name, + "expected_hash": ref.content_hash, + "actual_hash": actual, + }) + if tampered_tables: + raise ValueError( + f"pack {pack_hash[:12]} table-hash mismatch: " + f"{tampered_tables}" + ) + # Phase 1: restore metadata tables (no-op for v1 packs). tables_restored: dict[str, int] = {} if table_files_seen: diff --git a/tests/test_cold_object.py b/tests/test_cold_object.py index 97dfcc6..434b198 100644 --- a/tests/test_cold_object.py +++ b/tests/test_cold_object.py @@ -100,10 +100,15 @@ def test_parse_manifest_round_trip(): body = b"abc" * 100 leaf = hash_leaf(body).hex() pack = build_pack([(leaf, body)]) - entries = parse_manifest(pack.manifest_bytes) - assert len(entries) == 1 - assert entries[0].leaf_hash == leaf - assert entries[0].size == len(body) + parsed = parse_manifest(pack.manifest_bytes) + # build_pack produces a chunks-only manifest (no tables) — that's the + # v1 shape because build_pack doesn't accept extra_members. Tests of + # the v2 shape live in test_push_pack_v2_hydrates_fresh_empty_db. + assert parsed.format_version == 1 + assert parsed.tables == () + assert len(parsed.chunks) == 1 + assert parsed.chunks[0].leaf_hash == leaf + assert parsed.chunks[0].size == len(body) # --------------------------------------------------------------------------- @@ -368,6 +373,105 @@ def test_edges_fan_in_batches_huge_destinations(tmp_path): conn.close() +def test_v2_pack_hash_binds_table_contents(tmp_path): + """Two packs with the same chunk set but different table content + must NOT collide on pack_hash. + + The v1 bug: pack_hash was hash_leaf(chunk-only-manifest), so + same-chunks + different-tables produced same pack_hash. Bucket + overwrites or mesh-peer disagreement → silent table swap. v2 + fixes this by including content_hash for each tables/.jsonl + in the manifest before chunk entries. + + This test forces the issue by ingesting two different document + sets that happen to share a chunk — wait, that's impossible by + construction. Instead we check the manifest format directly: a + v2 pack's manifest has a `_format_version: 2` header and + `table_file` entries before `leaf_hash` entries. + """ + db = tmp_path / "v2bind.db" + conn = connect(db) + backend = MemoryBackend() + try: + ingest_source(conn, FakeSource([_doc("html://a", LONG)])) + result = push_pack(conn, backend) + manifest = backend.get_pack_manifest(result["packs"][0]["pack_hash"]) + parsed = parse_manifest(manifest) + + assert parsed.format_version == 2 + # All shipped tables that had rows should appear in the manifest. + # documents + chunks + merkle_nodes + audit_events minimum. + table_names = {t.member_name for t in parsed.tables} + assert "tables/documents.jsonl" in table_names + assert "tables/chunks.jsonl" in table_names + assert "tables/audit_events.jsonl" in table_names + # Every table_file ref carries a hash and size. + for t in parsed.tables: + assert len(t.content_hash) == 64 # sha256 hex + assert t.size > 0 + # Chunks are also there. + assert len(parsed.chunks) > 0 + finally: + conn.close() + + +def test_pull_pack_rejects_tampered_table(tmp_path): + """If the manifest says tables/X.jsonl should hash to H but the tar + member has different bytes, pull_pack refuses to restore. + + Tampering test: build a real v2 pack, then construct a corrupted + copy that swaps one table's bytes but keeps the original manifest. + pull_pack must raise on hash mismatch. + """ + import io as _io + import tarfile as _tarfile + import zstandard as _zstd + + db = tmp_path / "src.db" + conn = connect(db) + backend = MemoryBackend() + try: + ingest_source(conn, FakeSource([_doc("html://t", LONG)])) + result = push_pack(conn, backend) + pack_hash = result["packs"][0]["pack_hash"] + finally: + conn.close() + + # Decompress the pack, rewrite documents.jsonl with corrupted bytes, + # repack, replace the pack in the bucket. Manifest stays unchanged + # so the recorded table content_hash no longer matches. + body = backend.get_pack(pack_hash) + raw_tar = _zstd.ZstdDecompressor().stream_reader(_io.BytesIO(body)).read() + out_buf = _io.BytesIO() + cctx = _zstd.ZstdCompressor(level=3) + with cctx.stream_writer(out_buf, closefd=False) as writer: + with _tarfile.open(fileobj=writer, mode="w|") as out_tar: + with _tarfile.open(fileobj=_io.BytesIO(raw_tar), mode="r") as in_tar: + for m in in_tar: + f = in_tar.extractfile(m) + if f is None: + continue + data = f.read() + if m.name == "tables/documents.jsonl": + data = data + b'\n{"injected": "tamper"}\n' + info = _tarfile.TarInfo(name=m.name) + info.size = len(data) + else: + info = _tarfile.TarInfo(name=m.name) + info.size = len(data) + out_tar.addfile(info, _io.BytesIO(data)) + backend._tamper(f"packs/{pack_hash}.tar.zst", out_buf.getvalue()) + + # Fresh DB; expect ValueError on pull. + rx_db = tmp_path / "rx.db" + conn_rx = connect(rx_db) + try: + with pytest.raises(ValueError, match="table-hash mismatch"): + pull_pack(conn_rx, backend, pack_hash) + finally: + conn_rx.close() + + def test_push_pack_v2_hydrates_fresh_empty_db(tmp_path): """Pack format v2 self-sufficiency test: build a pack from a populated DB, then unpack it into a FRESH empty DB and verify every table is