From af257df02492e9758d5bac02b0d08201dbb16171 Mon Sep 17 00:00:00 2001 From: Akhilesh Chandra <174138640+mulkakhileshmj@users.noreply.github.com> Date: Thu, 10 Sep 2026 02:09:28 +0530 Subject: [PATCH] fix: validate chunk CRCs when reading canonical episodes after sync Episode._reader opened canonical files with CRC validation off, so the META, relabel and re-check lanes could re-certify an episode that had decayed on disk after sync. The checks stamped fresh measured findings over bytes that no longer match the file's own integrity stamp. Episode._reader now opens with validate_crcs=True. On a damaged episode the data reading checks record a diagnosed CRCValidationError per episode instead of fresh evidence, so the catalog never certifies the file. A healthy episode runs the check lanes unchanged, and the CRC pass rides on decompression that was already happening. The open_reader docstring now states the actual rule, and a regression test flips a canonical chunk's stored CRC and asserts the META lane refuses it with the CRC diagnosis while a healthy control stays green. Closes #474 --- src/hflow/episode.py | 6 +++++- src/hflow/reader.py | 9 ++++---- tests/reuse_test_helpers.py | 24 +++++++++++++++++++++ tests/test_processing_regressions.py | 32 ++++++++++++++++++++++++++++ 4 files changed, 66 insertions(+), 5 deletions(-) diff --git a/src/hflow/episode.py b/src/hflow/episode.py index fbb9fb64..cbcb93f9 100644 --- a/src/hflow/episode.py +++ b/src/hflow/episode.py @@ -308,7 +308,11 @@ def workdir(self) -> Path: @cached_property def _reader(self) -> EpisodeReader: - return open_reader(self.path) + # validate_crcs=True: a content hash proved this file at sync time, + # not at read time. Every post-sync lane (META, relabel, re-check) + # consumes this reader, so a canonical episode that decayed on disk + # must be diagnosed here rather than re-certified (#474). + return open_reader(self.path, validate_crcs=True) def close(self) -> None: if "_reader" in self.__dict__: diff --git a/src/hflow/reader.py b/src/hflow/reader.py index 29f2d77e..196488b9 100644 --- a/src/hflow/reader.py +++ b/src/hflow/reader.py @@ -298,9 +298,10 @@ def open_reader(path: Path | str, *, validate_crcs: bool = False) -> EpisodeRead ``validate_crcs`` checks each chunk's CRC as it is decoded, catching payload damage that magic-byte and summary checks alone cannot see. It - defaults to ``False`` because most callers re-read a canonical file HFlow - already produced and already identifies by content hash; pass ``True`` - only when reading a source that has not been trusted yet (see - ``hflow.transform``). + defaults to ``False`` for reads that stay out of the message data + (summary, metadata, provenance), which chunk CRCs cannot vouch for + anyway. Any read that consumes messages should pass ``True``: a content + hash or receipt proved the file when it was written, not the bytes on + disk at read time (see ``hflow.transform`` and ``hflow.episode``, #474). """ return PythonMcapEpisodeReader(path, validate_crcs=validate_crcs) diff --git a/tests/reuse_test_helpers.py b/tests/reuse_test_helpers.py index a8c69fa9..d0838575 100644 --- a/tests/reuse_test_helpers.py +++ b/tests/reuse_test_helpers.py @@ -44,6 +44,30 @@ def flip_chunk_payload_bytes(episode_path: Path, *, count: int = 4) -> None: episode_path.write_bytes(bytes(data)) +def flip_first_chunk_stored_crc(episode_path: Path) -> None: + """Flip one bit of the first chunk's stored ``uncompressed_crc`` in place. + + The complement of :func:`flip_chunk_payload_bytes` for compressed files: + a canonical episode's chunk records region is compressed, so its payload + bytes are not addressable in place -- but the CRC field in the chunk + header is. The payload still decompresses fine; the bytes simply no + longer match the file's own integrity stamp (real-world header rot), + which only a CRC-validated read can tell. + """ + from mcap.reader import make_reader + + data = bytearray(episode_path.read_bytes()) + summary = make_reader(io.BytesIO(bytes(data))).get_summary() + if summary is None or not summary.chunk_indexes: + raise ValueError(f"{episode_path} has no chunk records to corrupt") + chunk_start = summary.chunk_indexes[0].chunk_start_offset + # Chunk record per the MCAP spec: opcode(1) length(8) then + # message_start_time(8) message_end_time(8) uncompressed_size(8) + # uncompressed_crc(4), so the stored CRC begins at offset 33. + data[chunk_start + 33] ^= 0x01 + episode_path.write_bytes(bytes(data)) + + def content_id_differs_from_delivery_receipt(episode_path: Path, receipt_content_id: str) -> bool: """True when the file on disk no longer matches the recorded content id.""" return content_episode_id(episode_path) != receipt_content_id diff --git a/tests/test_processing_regressions.py b/tests/test_processing_regressions.py index 55e68f47..d95773c4 100644 --- a/tests/test_processing_regressions.py +++ b/tests/test_processing_regressions.py @@ -465,6 +465,38 @@ def _state_only_episode(tmp_path: Path) -> Path: ) +def test_check_lanes_refuse_a_canonical_episode_that_decayed_after_sync(tmp_path: Path) -> None: + """A canonical episode whose stored chunk CRC no longer matches its bytes + must not be re-certified by a post-sync check run (#474): every check that + consumes message data errors with the CRC diagnosis instead of stamping + fresh findings over bytes the file's own integrity stamp disowns. #429 + closed this hole on the LeRobot resume path and #462 on the primary ingest + read; the post-sync ``Episode`` read is the same trust boundary. + """ + from reuse_test_helpers import flip_first_chunk_stored_crc + + source = _state_only_episode(tmp_path) + app = hflow.App("post-sync-crc", data_root=tmp_path / "data") + app.process(source, record=False) + + # The control: a healthy canonical runs the check lane unchanged. + healthy = app.process(source, record=False, stages={hflow.Stage.META}) + assert {run.status for run in healthy.checks} == {hflow.CheckStatus.MEASURED} + + flip_first_chunk_stored_crc(healthy.canonical_path) + + damaged = app.process(source, record=False, stages={hflow.Stage.META}) + digest_run = damaged.check("content_digest") + assert digest_run.status == hflow.CheckStatus.ERROR + assert digest_run.error is not None + assert "CRCValidationError" in digest_run.error + assert damaged.has_errors + # No fresh evidence lands over the damaged bytes: the only checks still + # measuring are the camera ones, which read nothing on a camera-less + # episode and record no keys. + assert all(not run.result.measurements for run in damaged.checks if run.result is not None) + + def test_two_checks_recording_one_measurement_key_are_refused(tmp_path: Path) -> None: """Every step of one run shares its fingerprint and timestamp, so a shared key is a tie the catalog resolves arbitrarily -- one step's value silently