From e5af67893d1bb373a2e0301a2f2d6cc6114e8d62 Mon Sep 17 00:00:00 2001 From: VARUN3WARE Date: Fri, 4 Sep 2026 20:16:46 +0530 Subject: [PATCH 1/3] feat(snapshot): record table and copied-asset integrity in format.json (#397) Keep format version 1 and add path/size_bytes/sha256 receipts for required Parquet tables and copy-mode assets, plus a content_id over the normalized inventory so missing members are detectable without shipping a verifier yet. --- docs/how-to/export-dataset-snapshot.md | 18 ++++ src/hflow/snapshot.py | 102 ++++++++++++++++-- tests/test_dataset_snapshot.py | 144 +++++++++++++++++++++++++ 3 files changed, 254 insertions(+), 10 deletions(-) diff --git a/docs/how-to/export-dataset-snapshot.md b/docs/how-to/export-dataset-snapshot.md index 9dc99e73..b0f444cf 100644 --- a/docs/how-to/export-dataset-snapshot.md +++ b/docs/how-to/export-dataset-snapshot.md @@ -62,6 +62,24 @@ each table, records the media mode, and states whether media paths are relative to the export directory. The JSON marker is written last and the completed directory is activated atomically. +Under the same format version `1`, the marker also records delivery integrity +so a later verifier (not shipped here) can tell whether the published bytes +are still intact: + +| Field | Meaning | +| --- | --- | +| `tables..path` | Required Parquet file relative to the export root | +| `tables..size_bytes` | Size of that file at export time | +| `tables..sha256` | Full SHA-256 of that file's bytes | +| `assets[]` | Same `path` / `size_bytes` / `sha256` for every regular file under `assets/` in copy mode; empty in references mode (remote media are not fetched) | +| `content_id` | 16-hex digest of the normalized inventory (all table and asset receipts, sorted by path), so a deleted member is visible even when every remaining file still matches | + +This is a receipt, not a verify command: export does not re-read the +destination after transfer, and there is no public `verify_dataset_snapshot` +API or CLI yet. Older HFlow overwrite checks only `format` and +`format_version`, so these added keys stay backward compatible without a +format-version bump. + The Parquet tables form one snapshot: | File | Grain and purpose | diff --git a/src/hflow/snapshot.py b/src/hflow/snapshot.py index 5806490e..50082f8f 100644 --- a/src/hflow/snapshot.py +++ b/src/hflow/snapshot.py @@ -35,6 +35,18 @@ _TAGS_TABLE_FILE_NAME = "tags.parquet" _INTERVALS_TABLE_FILE_NAME = "intervals.parquet" _FORMAT_MARKER_FILE_NAME = "format.json" +_REQUIRED_TABLE_FILES: dict[str, str] = { + "samples": _SAMPLES_TABLE_FILE_NAME, + "measurements": _MEASUREMENTS_TABLE_FILE_NAME, + "observations": _OBSERVATIONS_TABLE_FILE_NAME, + "media": _MEDIA_TABLE_FILE_NAME, + "check_runs": _CHECK_RUNS_TABLE_FILE_NAME, + "tags": _TAGS_TABLE_FILE_NAME, + "intervals": _INTERVALS_TABLE_FILE_NAME, +} +_COPIED_ASSETS_DIRECTORY_NAME = "assets" +# Whole-inventory digest width matches prepared-manifest episode content_id (#389). +_INVENTORY_CONTENT_ID_HEX_CHARS = 16 class SnapshotMediaMode(StrEnum): @@ -103,6 +115,74 @@ def _quote_sql_string(value: str) -> str: return "'" + value.replace("'", "''") + "'" +def _sha256_hex(path: Path) -> str: + """Full SHA-256 hex digest of a file's bytes.""" + digest = hashlib.sha256() + with path.open("rb") as stream: + while chunk := stream.read(8 * 1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +def _file_integrity_record(relative_path: str, absolute_path: Path) -> dict[str, str | int]: + """Receipt for one delivered snapshot file (table or copied asset).""" + return { + "path": relative_path, + "size_bytes": absolute_path.stat().st_size, + "sha256": _sha256_hex(absolute_path), + } + + +def _inventory_content_id(entries: list[dict[str, str | int]]) -> str: + """Digest of the normalized integrity inventory (catches missing members). + + Entries are sorted by ``path`` and serialized with stable separators so the + digest depends only on the delivered set, not write order. Width matches + LeRobot prepared-manifest ``content_id`` (#389). + """ + normalized = sorted(entries, key=lambda entry: str(entry["path"])) + payload = json.dumps(normalized, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode()).hexdigest()[:_INVENTORY_CONTENT_ID_HEX_CHARS] + + +def _build_snapshot_integrity_marker_fields( + staging_directory: Path, +) -> dict[str, object]: + """Integrity fields recorded in ``format.json`` under format version 1. + + Required Parquet tables and every regular file under ``assets/`` (copy mode) + get ``path`` / ``size_bytes`` / ``sha256``. ``content_id`` digests that + normalized inventory so a deleted member is detectable without a verifier + product yet. References mode leaves ``assets`` empty: remote media are not + fetched for hashing. + """ + tables: dict[str, dict[str, str | int]] = {} + for table_name, file_name in _REQUIRED_TABLE_FILES.items(): + absolute_path = staging_directory / file_name + if not absolute_path.is_file(): + raise FileNotFoundError( + f"snapshot staging is missing required table {file_name!r} " + f"under {staging_directory}" + ) + tables[table_name] = _file_integrity_record(file_name, absolute_path) + + assets: list[dict[str, str | int]] = [] + assets_directory = staging_directory / _COPIED_ASSETS_DIRECTORY_NAME + if assets_directory.is_dir(): + for absolute_path in sorted(assets_directory.rglob("*")): + if not absolute_path.is_file(): + continue + relative_path = absolute_path.relative_to(staging_directory).as_posix() + assets.append(_file_integrity_record(relative_path, absolute_path)) + + inventory = [*tables.values(), *assets] + return { + "tables": tables, + "assets": assets, + "content_id": _inventory_content_id(inventory), + } + + def _copy_query_to_parquet( connection: duckdb.DuckDBPyConnection, query: str, destination: Path ) -> int: @@ -335,7 +415,11 @@ def _copied_media_relative_path(*, episode_id: str, artifact_name: str, artifact safe_source_file_name = _safe_file_name(source_file_name) identity_digest = hashlib.sha256(f"{artifact_name}\0{artifact_uri}".encode()).hexdigest()[:12] safe_episode_id = _safe_file_name(episode_id) - return Path("assets") / safe_episode_id / f"{identity_digest}-{safe_source_file_name}" + return ( + Path(_COPIED_ASSETS_DIRECTORY_NAME) + / safe_episode_id + / f"{identity_digest}-{safe_source_file_name}" + ) def _write_media_table( @@ -579,6 +663,12 @@ def export_dataset_snapshot( mode every recorded artifact is materialized below ``assets/`` and the media table stores a path relative to the export directory. + ``format.json`` stays format version ``1`` and records per-file integrity + (``path``, ``size_bytes``, ``sha256``) for every required table and every + copied asset, plus a ``content_id`` over that normalized inventory. That + is a delivery receipt only: this export does not verify the destination + after transfer. + The completed directory appears atomically. Existing destinations are refused unless ``overwrite=True`` and ``format.json`` identifies a supported HFlow dataset snapshot; even then, the prior export remains in @@ -670,15 +760,7 @@ def export_dataset_snapshot( "media_uri_base": ( "export_directory" if resolved_media_mode is SnapshotMediaMode.COPY else None ), - "tables": { - "samples": _SAMPLES_TABLE_FILE_NAME, - "measurements": _MEASUREMENTS_TABLE_FILE_NAME, - "observations": _OBSERVATIONS_TABLE_FILE_NAME, - "media": _MEDIA_TABLE_FILE_NAME, - "check_runs": _CHECK_RUNS_TABLE_FILE_NAME, - "tags": _TAGS_TABLE_FILE_NAME, - "intervals": _INTERVALS_TABLE_FILE_NAME, - }, + **_build_snapshot_integrity_marker_fields(staging_directory), } (staging_directory / _FORMAT_MARKER_FILE_NAME).write_text( json.dumps(format_marker, indent=2, sort_keys=True) + "\n" diff --git a/tests/test_dataset_snapshot.py b/tests/test_dataset_snapshot.py index 4ab176e2..0f346821 100644 --- a/tests/test_dataset_snapshot.py +++ b/tests/test_dataset_snapshot.py @@ -175,6 +175,23 @@ def test_dataset_snapshot_is_tool_neutral_and_selected_by_manifest(tmp_path: Pat assert format_marker["format"] == "hflow-dataset-snapshot" assert format_marker["format_version"] == "1" assert format_marker["media_mode"] == "references" + assert format_marker["assets"] == [] + assert set(format_marker["tables"]) == { + "samples", + "measurements", + "observations", + "media", + "check_runs", + "tags", + "intervals", + } + for table_name, receipt in format_marker["tables"].items(): + assert receipt["path"] == f"{table_name}.parquet" + assert receipt["size_bytes"] == (output_directory / receipt["path"]).stat().st_size + assert receipt["sha256"] == snapshot_module._sha256_hex(output_directory / receipt["path"]) + inventory = [*format_marker["tables"].values(), *format_marker["assets"]] + assert format_marker["content_id"] == snapshot_module._inventory_content_id(inventory) + assert len(format_marker["content_id"]) == 16 sample_row = duckdb.execute( """ @@ -647,3 +664,130 @@ def test_snapshot_samples_report_unverified_for_a_crashed_critical_check( [str(output_directory / "samples.parquet")], ).fetchone() assert sample_row == (append_result.episode_id, "unverified") + + +def test_dataset_snapshot_copy_mode_records_asset_integrity(tmp_path: Path) -> None: + catalog = Catalog(tmp_path / "catalog") + _append_snapshot_episode( + catalog, + tmp_path, + name="copy-integrity", + score=0.5, + with_media=True, + ) + output_directory = tmp_path / "dataset-snapshot" + + report = hflow.export_dataset_snapshot( + catalog.location, + output_directory, + media_mode=hflow.SnapshotMediaMode.COPY, + ) + + assert report.copied_media_count == 1 + format_marker = json.loads((output_directory / "format.json").read_text()) + assert format_marker["format_version"] == "1" + assert format_marker["media_mode"] == "copy" + assert len(format_marker["assets"]) == 1 + asset_receipt = format_marker["assets"][0] + asset_path = output_directory / asset_receipt["path"] + assert asset_path.is_file() + assert asset_receipt["path"].startswith("assets/") + assert asset_receipt["size_bytes"] == asset_path.stat().st_size + assert asset_receipt["sha256"] == snapshot_module._sha256_hex(asset_path) + inventory = [*format_marker["tables"].values(), *format_marker["assets"]] + assert format_marker["content_id"] == snapshot_module._inventory_content_id(inventory) + + +def test_dataset_snapshot_marker_integrity_detects_post_export_mutations( + tmp_path: Path, +) -> None: + catalog = Catalog(tmp_path / "catalog") + _append_snapshot_episode( + catalog, + tmp_path, + name="mutation-source", + score=0.9, + with_media=True, + ) + output_directory = tmp_path / "dataset-snapshot" + hflow.export_dataset_snapshot( + catalog.location, + output_directory, + media_mode=hflow.SnapshotMediaMode.COPY, + ) + original_marker = json.loads((output_directory / "format.json").read_text()) + assert original_marker["format_version"] == "1" + + media_copy = tmp_path / "media-mutated" + tags_copy = tmp_path / "tags-deleted" + samples_copy = tmp_path / "samples-truncated" + for destination in (media_copy, tags_copy, samples_copy): + snapshot_module.shutil.copytree(output_directory, destination) + + # Replacing copied media bytes leaves tables readable but mismatches the receipt. + media_receipt = original_marker["assets"][0] + media_path = media_copy / media_receipt["path"] + media_path.write_bytes(b"replaced media bytes") + assert snapshot_module._sha256_hex(media_path) != media_receipt["sha256"] + assert media_path.stat().st_size != media_receipt["size_bytes"] + media_inventory = [ + *[ + snapshot_module._file_integrity_record(receipt["path"], media_copy / receipt["path"]) + for receipt in original_marker["tables"].values() + ], + snapshot_module._file_integrity_record(media_receipt["path"], media_path), + ] + assert snapshot_module._inventory_content_id(media_inventory) != original_marker["content_id"] + + # Deleting a required table is invisible to format/format_version checks, + # but the inventory digest no longer matches the marker. + (tags_copy / "tags.parquet").unlink() + assert not (tags_copy / "tags.parquet").exists() + remaining_entries = [ + snapshot_module._file_integrity_record(receipt["path"], tags_copy / receipt["path"]) + for receipt in original_marker["tables"].values() + if (tags_copy / receipt["path"]).is_file() + ] + [ + snapshot_module._file_integrity_record(receipt["path"], tags_copy / receipt["path"]) + for receipt in original_marker["assets"] + ] + assert snapshot_module._inventory_content_id(remaining_entries) != original_marker["content_id"] + overwrite_marker = json.loads((tags_copy / "format.json").read_text()) + assert overwrite_marker["format"] == "hflow-dataset-snapshot" + assert overwrite_marker["format_version"] == "1" + + # Truncation changes size and hash while the destination still looks like a snapshot. + samples_path = samples_copy / "samples.parquet" + samples_path.write_bytes(samples_path.read_bytes()[:64]) + samples_receipt = original_marker["tables"]["samples"] + assert samples_path.stat().st_size != samples_receipt["size_bytes"] + assert snapshot_module._sha256_hex(samples_path) != samples_receipt["sha256"] + + +def test_dataset_snapshot_overwrite_still_accepts_integrity_enriched_v1_marker( + tmp_path: Path, +) -> None: + catalog = Catalog(tmp_path / "catalog") + _append_snapshot_episode( + catalog, + tmp_path, + name="overwrite-v1", + score=0.2, + with_media=False, + ) + output_directory = tmp_path / "dataset-snapshot" + hflow.export_dataset_snapshot(catalog.location, output_directory) + first_marker = json.loads((output_directory / "format.json").read_text()) + assert "content_id" in first_marker + assert first_marker["format_version"] == "1" + + report = hflow.export_dataset_snapshot( + catalog.location, + output_directory, + overwrite=True, + ) + assert report.retained_backup is None + second_marker = json.loads((output_directory / "format.json").read_text()) + assert second_marker["format_version"] == "1" + assert "content_id" in second_marker + assert set(second_marker["tables"]) == set(first_marker["tables"]) From c6bc02a9f23187e695410a76bce6e32bc96e0bae Mon Sep 17 00:00:00 2001 From: VARUN3WARE Date: Sat, 5 Sep 2026 11:02:33 +0530 Subject: [PATCH 2/3] fix(snapshot): keep string tables; nest integrity under its own key Leave the published name-to-filename tables map unchanged under format v1, move receipts to integrity.tables/assets, and use a full-length inventory content_id. Documents the copy-mode second media read for hashing. --- docs/how-to/export-dataset-snapshot.md | 31 ++++++----- src/hflow/snapshot.py | 39 +++++++------ tests/test_dataset_snapshot.py | 76 ++++++++++++++++---------- 3 files changed, 87 insertions(+), 59 deletions(-) diff --git a/docs/how-to/export-dataset-snapshot.md b/docs/how-to/export-dataset-snapshot.md index b0f444cf..89604593 100644 --- a/docs/how-to/export-dataset-snapshot.md +++ b/docs/how-to/export-dataset-snapshot.md @@ -58,27 +58,32 @@ dataset-snapshot/ ## Format contract `format.json` identifies `hflow-dataset-snapshot` format version `1`, names -each table, records the media mode, and states whether media paths are relative -to the export directory. The JSON marker is written last and the completed -directory is activated atomically. +each table (a name-to-filename map), records the media mode, and states whether +media paths are relative to the export directory. The JSON marker is written +last and the completed directory is activated atomically. -Under the same format version `1`, the marker also records delivery integrity -so a later verifier (not shipped here) can tell whether the published bytes -are still intact: +Under the same format version `1`, an additive `integrity` block records +delivery integrity so a later verifier (not shipped here) can tell whether the +published bytes are still intact. The original `tables` map is unchanged for +external readers; receipts live only under `integrity`: | Field | Meaning | | --- | --- | -| `tables..path` | Required Parquet file relative to the export root | -| `tables..size_bytes` | Size of that file at export time | -| `tables..sha256` | Full SHA-256 of that file's bytes | -| `assets[]` | Same `path` / `size_bytes` / `sha256` for every regular file under `assets/` in copy mode; empty in references mode (remote media are not fetched) | -| `content_id` | 16-hex digest of the normalized inventory (all table and asset receipts, sorted by path), so a deleted member is visible even when every remaining file still matches | +| `tables.` | Required Parquet filename (unchanged string map) | +| `integrity.tables..path` | Same file relative to the export root | +| `integrity.tables..size_bytes` | Size of that file at export time | +| `integrity.tables..sha256` | Full SHA-256 of that file's bytes | +| `integrity.assets[]` | Same `path` / `size_bytes` / `sha256` for every regular file under `assets/` in copy mode; empty in references mode (remote media are not fetched) | +| `integrity.content_id` | Full SHA-256 of the normalized inventory (all table and asset receipts, sorted by path), so a deleted member is visible even when every remaining file still matches | + +Copy mode re-reads each copied asset once after the copy to compute its hash +(a second full read of media bytes on export). This is a receipt, not a verify command: export does not re-read the destination after transfer, and there is no public `verify_dataset_snapshot` API or CLI yet. Older HFlow overwrite checks only `format` and -`format_version`, so these added keys stay backward compatible without a -format-version bump. +`format_version`, and readers that only consume the string `tables` map keep +working; the `integrity` key is purely additive under format version `1`. The Parquet tables form one snapshot: diff --git a/src/hflow/snapshot.py b/src/hflow/snapshot.py index 50082f8f..5eca7596 100644 --- a/src/hflow/snapshot.py +++ b/src/hflow/snapshot.py @@ -45,8 +45,6 @@ "intervals": _INTERVALS_TABLE_FILE_NAME, } _COPIED_ASSETS_DIRECTORY_NAME = "assets" -# Whole-inventory digest width matches prepared-manifest episode content_id (#389). -_INVENTORY_CONTENT_ID_HEX_CHARS = 16 class SnapshotMediaMode(StrEnum): @@ -134,27 +132,31 @@ def _file_integrity_record(relative_path: str, absolute_path: Path) -> dict[str, def _inventory_content_id(entries: list[dict[str, str | int]]) -> str: - """Digest of the normalized integrity inventory (catches missing members). + """Full SHA-256 of the normalized integrity inventory. Entries are sorted by ``path`` and serialized with stable separators so the - digest depends only on the delivered set, not write order. Width matches - LeRobot prepared-manifest ``content_id`` (#389). + digest depends only on the delivered set, not write order. This is the + delivery's integrity digest (corruption and deliberate set tampering), not + an episode identity; the field name ``content_id`` matches prepared-manifest + receipts (#389) but the value is full-length like the per-file hashes and + Croissant's SHA-256 recommendation. """ normalized = sorted(entries, key=lambda entry: str(entry["path"])) payload = json.dumps(normalized, sort_keys=True, separators=(",", ":")) - return hashlib.sha256(payload.encode()).hexdigest()[:_INVENTORY_CONTENT_ID_HEX_CHARS] + return hashlib.sha256(payload.encode()).hexdigest() def _build_snapshot_integrity_marker_fields( staging_directory: Path, ) -> dict[str, object]: - """Integrity fields recorded in ``format.json`` under format version 1. + """Additive ``integrity`` block for ``format.json`` under format version 1. Required Parquet tables and every regular file under ``assets/`` (copy mode) get ``path`` / ``size_bytes`` / ``sha256``. ``content_id`` digests that normalized inventory so a deleted member is detectable without a verifier product yet. References mode leaves ``assets`` empty: remote media are not - fetched for hashing. + fetched for hashing. Copy mode re-reads each copied asset once after the + copy to compute its hash. """ tables: dict[str, dict[str, str | int]] = {} for table_name, file_name in _REQUIRED_TABLE_FILES.items(): @@ -177,9 +179,11 @@ def _build_snapshot_integrity_marker_fields( inventory = [*tables.values(), *assets] return { - "tables": tables, - "assets": assets, - "content_id": _inventory_content_id(inventory), + "integrity": { + "tables": tables, + "assets": assets, + "content_id": _inventory_content_id(inventory), + } } @@ -663,11 +667,13 @@ def export_dataset_snapshot( mode every recorded artifact is materialized below ``assets/`` and the media table stores a path relative to the export directory. - ``format.json`` stays format version ``1`` and records per-file integrity - (``path``, ``size_bytes``, ``sha256``) for every required table and every - copied asset, plus a ``content_id`` over that normalized inventory. That - is a delivery receipt only: this export does not verify the destination - after transfer. + ``format.json`` stays format version ``1``. The published ``tables`` map + remains a name-to-filename contract. An additive ``integrity`` block + records per-file ``path`` / ``size_bytes`` / ``sha256`` for every required + table and every copied asset, plus a full-length ``content_id`` over that + normalized inventory. Copy mode re-reads each copied asset once after the + copy to hash it. This is a delivery receipt only: this export does not + verify the destination after transfer. The completed directory appears atomically. Existing destinations are refused unless ``overwrite=True`` and ``format.json`` identifies a @@ -760,6 +766,7 @@ def export_dataset_snapshot( "media_uri_base": ( "export_directory" if resolved_media_mode is SnapshotMediaMode.COPY else None ), + "tables": dict(_REQUIRED_TABLE_FILES), **_build_snapshot_integrity_marker_fields(staging_directory), } (staging_directory / _FORMAT_MARKER_FILE_NAME).write_text( diff --git a/tests/test_dataset_snapshot.py b/tests/test_dataset_snapshot.py index 0f346821..69b201bc 100644 --- a/tests/test_dataset_snapshot.py +++ b/tests/test_dataset_snapshot.py @@ -175,23 +175,26 @@ def test_dataset_snapshot_is_tool_neutral_and_selected_by_manifest(tmp_path: Pat assert format_marker["format"] == "hflow-dataset-snapshot" assert format_marker["format_version"] == "1" assert format_marker["media_mode"] == "references" - assert format_marker["assets"] == [] - assert set(format_marker["tables"]) == { - "samples", - "measurements", - "observations", - "media", - "check_runs", - "tags", - "intervals", + assert format_marker["tables"] == { + "samples": "samples.parquet", + "measurements": "measurements.parquet", + "observations": "observations.parquet", + "media": "media.parquet", + "check_runs": "check_runs.parquet", + "tags": "tags.parquet", + "intervals": "intervals.parquet", } - for table_name, receipt in format_marker["tables"].items(): - assert receipt["path"] == f"{table_name}.parquet" - assert receipt["size_bytes"] == (output_directory / receipt["path"]).stat().st_size - assert receipt["sha256"] == snapshot_module._sha256_hex(output_directory / receipt["path"]) - inventory = [*format_marker["tables"].values(), *format_marker["assets"]] - assert format_marker["content_id"] == snapshot_module._inventory_content_id(inventory) - assert len(format_marker["content_id"]) == 16 + integrity = format_marker["integrity"] + assert integrity["assets"] == [] + assert set(integrity["tables"]) == set(format_marker["tables"]) + for table_name, file_name in format_marker["tables"].items(): + receipt = integrity["tables"][table_name] + assert receipt["path"] == file_name + assert receipt["size_bytes"] == (output_directory / file_name).stat().st_size + assert receipt["sha256"] == snapshot_module._sha256_hex(output_directory / file_name) + inventory = [*integrity["tables"].values(), *integrity["assets"]] + assert integrity["content_id"] == snapshot_module._inventory_content_id(inventory) + assert len(integrity["content_id"]) == 64 sample_row = duckdb.execute( """ @@ -687,15 +690,18 @@ def test_dataset_snapshot_copy_mode_records_asset_integrity(tmp_path: Path) -> N format_marker = json.loads((output_directory / "format.json").read_text()) assert format_marker["format_version"] == "1" assert format_marker["media_mode"] == "copy" - assert len(format_marker["assets"]) == 1 - asset_receipt = format_marker["assets"][0] + assert format_marker["tables"]["media"] == "media.parquet" + integrity = format_marker["integrity"] + assert len(integrity["assets"]) == 1 + asset_receipt = integrity["assets"][0] asset_path = output_directory / asset_receipt["path"] assert asset_path.is_file() assert asset_receipt["path"].startswith("assets/") assert asset_receipt["size_bytes"] == asset_path.stat().st_size assert asset_receipt["sha256"] == snapshot_module._sha256_hex(asset_path) - inventory = [*format_marker["tables"].values(), *format_marker["assets"]] - assert format_marker["content_id"] == snapshot_module._inventory_content_id(inventory) + inventory = [*integrity["tables"].values(), *integrity["assets"]] + assert integrity["content_id"] == snapshot_module._inventory_content_id(inventory) + assert len(integrity["content_id"]) == 64 def test_dataset_snapshot_marker_integrity_detects_post_export_mutations( @@ -717,6 +723,7 @@ def test_dataset_snapshot_marker_integrity_detects_post_export_mutations( ) original_marker = json.loads((output_directory / "format.json").read_text()) assert original_marker["format_version"] == "1" + original_integrity = original_marker["integrity"] media_copy = tmp_path / "media-mutated" tags_copy = tmp_path / "tags-deleted" @@ -725,7 +732,7 @@ def test_dataset_snapshot_marker_integrity_detects_post_export_mutations( snapshot_module.shutil.copytree(output_directory, destination) # Replacing copied media bytes leaves tables readable but mismatches the receipt. - media_receipt = original_marker["assets"][0] + media_receipt = original_integrity["assets"][0] media_path = media_copy / media_receipt["path"] media_path.write_bytes(b"replaced media bytes") assert snapshot_module._sha256_hex(media_path) != media_receipt["sha256"] @@ -733,11 +740,13 @@ def test_dataset_snapshot_marker_integrity_detects_post_export_mutations( media_inventory = [ *[ snapshot_module._file_integrity_record(receipt["path"], media_copy / receipt["path"]) - for receipt in original_marker["tables"].values() + for receipt in original_integrity["tables"].values() ], snapshot_module._file_integrity_record(media_receipt["path"], media_path), ] - assert snapshot_module._inventory_content_id(media_inventory) != original_marker["content_id"] + assert ( + snapshot_module._inventory_content_id(media_inventory) != original_integrity["content_id"] + ) # Deleting a required table is invisible to format/format_version checks, # but the inventory digest no longer matches the marker. @@ -745,21 +754,24 @@ def test_dataset_snapshot_marker_integrity_detects_post_export_mutations( assert not (tags_copy / "tags.parquet").exists() remaining_entries = [ snapshot_module._file_integrity_record(receipt["path"], tags_copy / receipt["path"]) - for receipt in original_marker["tables"].values() + for receipt in original_integrity["tables"].values() if (tags_copy / receipt["path"]).is_file() ] + [ snapshot_module._file_integrity_record(receipt["path"], tags_copy / receipt["path"]) - for receipt in original_marker["assets"] + for receipt in original_integrity["assets"] ] - assert snapshot_module._inventory_content_id(remaining_entries) != original_marker["content_id"] + assert ( + snapshot_module._inventory_content_id(remaining_entries) != original_integrity["content_id"] + ) overwrite_marker = json.loads((tags_copy / "format.json").read_text()) assert overwrite_marker["format"] == "hflow-dataset-snapshot" assert overwrite_marker["format_version"] == "1" + assert overwrite_marker["tables"]["tags"] == "tags.parquet" # Truncation changes size and hash while the destination still looks like a snapshot. samples_path = samples_copy / "samples.parquet" samples_path.write_bytes(samples_path.read_bytes()[:64]) - samples_receipt = original_marker["tables"]["samples"] + samples_receipt = original_integrity["tables"]["samples"] assert samples_path.stat().st_size != samples_receipt["size_bytes"] assert snapshot_module._sha256_hex(samples_path) != samples_receipt["sha256"] @@ -778,8 +790,10 @@ def test_dataset_snapshot_overwrite_still_accepts_integrity_enriched_v1_marker( output_directory = tmp_path / "dataset-snapshot" hflow.export_dataset_snapshot(catalog.location, output_directory) first_marker = json.loads((output_directory / "format.json").read_text()) - assert "content_id" in first_marker + assert "integrity" in first_marker + assert "content_id" in first_marker["integrity"] assert first_marker["format_version"] == "1" + assert first_marker["tables"]["samples"] == "samples.parquet" report = hflow.export_dataset_snapshot( catalog.location, @@ -789,5 +803,7 @@ def test_dataset_snapshot_overwrite_still_accepts_integrity_enriched_v1_marker( assert report.retained_backup is None second_marker = json.loads((output_directory / "format.json").read_text()) assert second_marker["format_version"] == "1" - assert "content_id" in second_marker - assert set(second_marker["tables"]) == set(first_marker["tables"]) + assert "integrity" in second_marker + assert "content_id" in second_marker["integrity"] + assert second_marker["tables"] == first_marker["tables"] + assert set(second_marker["integrity"]["tables"]) == set(first_marker["tables"]) From e2446d4935ae8779bfc029fa87faf83c8b891bdc Mon Sep 17 00:00:00 2001 From: Kingston Date: Sun, 6 Sep 2026 00:59:37 -0700 Subject: [PATCH 3/3] docs(snapshot): say the receipt is not a tamper defence --- docs/how-to/export-dataset-snapshot.md | 5 +++++ src/hflow/snapshot.py | 15 ++++++++++----- tests/test_dataset_snapshot.py | 3 ++- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/docs/how-to/export-dataset-snapshot.md b/docs/how-to/export-dataset-snapshot.md index 89604593..c3f814d9 100644 --- a/docs/how-to/export-dataset-snapshot.md +++ b/docs/how-to/export-dataset-snapshot.md @@ -85,6 +85,11 @@ API or CLI yet. Older HFlow overwrite checks only `format` and `format_version`, and readers that only consume the string `tables` map keep working; the `integrity` key is purely additive under format version `1`. +The receipt travels unsigned inside the `format.json` it describes, so it +catches corruption and accidental loss, not tampering: anyone who can edit a +table can recompute the hashes to match. Use a signature or an out-of-band +checksum if you need to detect a deliberate change. + The Parquet tables form one snapshot: | File | Grain and purpose | diff --git a/src/hflow/snapshot.py b/src/hflow/snapshot.py index 5eca7596..19f8fd94 100644 --- a/src/hflow/snapshot.py +++ b/src/hflow/snapshot.py @@ -135,11 +135,16 @@ def _inventory_content_id(entries: list[dict[str, str | int]]) -> str: """Full SHA-256 of the normalized integrity inventory. Entries are sorted by ``path`` and serialized with stable separators so the - digest depends only on the delivered set, not write order. This is the - delivery's integrity digest (corruption and deliberate set tampering), not - an episode identity; the field name ``content_id`` matches prepared-manifest - receipts (#389) but the value is full-length like the per-file hashes and - Croissant's SHA-256 recommendation. + digest depends only on the delivered set, not write order. It catches + corruption and accidental loss: a truncated table, a partial transfer, a + member that never arrived. It is not a tamper defence, because it travels + unsigned inside the same ``format.json`` it describes, so anyone who edits + a table can recompute it. + + This is a delivery receipt, not an episode identity. The field name + ``content_id`` matches prepared-manifest receipts (#389) but the value is + full-length like the per-file hashes and Croissant's SHA-256 + recommendation. """ normalized = sorted(entries, key=lambda entry: str(entry["path"])) payload = json.dumps(normalized, sort_keys=True, separators=(",", ":")) diff --git a/tests/test_dataset_snapshot.py b/tests/test_dataset_snapshot.py index 69b201bc..2a5de97c 100644 --- a/tests/test_dataset_snapshot.py +++ b/tests/test_dataset_snapshot.py @@ -1,4 +1,5 @@ import json +import shutil from pathlib import Path import duckdb @@ -729,7 +730,7 @@ def test_dataset_snapshot_marker_integrity_detects_post_export_mutations( tags_copy = tmp_path / "tags-deleted" samples_copy = tmp_path / "samples-truncated" for destination in (media_copy, tags_copy, samples_copy): - snapshot_module.shutil.copytree(output_directory, destination) + shutil.copytree(output_directory, destination) # Replacing copied media bytes leaves tables readable but mismatches the receipt. media_receipt = original_integrity["assets"][0]