From 4aae6c433d0df410ed37b41a63f61a5d5748493c Mon Sep 17 00:00:00 2001 From: mptyl Date: Sun, 12 Jul 2026 07:31:46 +0200 Subject: [PATCH] fix(dwh): bind snapshots to validated bytes --- .superpowers/sdd/evidence-task-6-report.md | 10 +++ harness/tests/test_dwh_preprocess_job.py | 84 ++++++++++++++++++++++ harness/tht/jobs/dwh_pipeline.py | 57 +++++++++++---- 3 files changed, 136 insertions(+), 15 deletions(-) diff --git a/.superpowers/sdd/evidence-task-6-report.md b/.superpowers/sdd/evidence-task-6-report.md index 91ccc3b9..8dcb9d6b 100644 --- a/.superpowers/sdd/evidence-task-6-report.md +++ b/.superpowers/sdd/evidence-task-6-report.md @@ -18,3 +18,13 @@ ACTIVE publication, and retention cleanup. Each test proves the replacement tree written, or deleted; the descriptor-pinned original either completes consistently or fails closed. Existing owner binding, legacy rejection, crash reconciliation, resume, atomic rollback, retention, and reader/writer exclusion behavior remains covered. + +## Final review correction + +Snapshot materialization now reads the manifest and every owned artifact exactly once through the +already-open generation descriptor, validates each hash against those exact bytes, and writes the +same byte objects to the private snapshot. A deterministic second-read mutation test proves hostile +pickle bytes can neither pass validation nor enter the snapshot. Reconciliation closes the ACTIVE +generation descriptor in a `finally` block on matches, mismatches, and exceptions. Pipeline-owned +snapshot directories are removed and deregistered after `run_job` on both successful and failed +runs, preventing repeated pipeline use from accumulating temporary directories or registry entries. diff --git a/harness/tests/test_dwh_preprocess_job.py b/harness/tests/test_dwh_preprocess_job.py index fe1aa8e9..8bef80ba 100644 --- a/harness/tests/test_dwh_preprocess_job.py +++ b/harness/tests/test_dwh_preprocess_job.py @@ -524,6 +524,90 @@ def test_snapshot_root_swap_after_lease_never_reads_replacement(monkeypatch, tmp assert (replacement / "sentinel").read_text() == "replacement-secret" +def test_snapshot_copies_each_validated_artifact_once_without_reopen(monkeypatch, tmp_path): + import tht.jobs.dwh_pipeline as module + + pipeline = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("trusted"), + build_lsh=lambda physical, output: [ + (output / name).write_text("trusted") + for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json") + ], + ) + assert pipeline.run().status == "succeeded" + real_read = module._read_owned_at + reads = {} + + def mutate_on_reopen(directory_fd, name, *, readonly): + reads[name] = reads.get(name, 0) + 1 + if name.endswith(".pkl") and reads[name] > 1: + return b"MALICIOUS_PICKLE" + return real_read(directory_fd, name, readonly=readonly) + + monkeypatch.setattr(module, "_read_owned_at", mutate_on_reopen) + with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot: + assert (snapshot.lsh_dir / "demo_lsh.pkl").read_text() == "trusted" + assert "MALICIOUS" not in (snapshot.lsh_dir / "demo_lsh.pkl").read_text() + assert all(count == 1 for count in reads.values()) + + +def test_reconcile_mismatch_closes_active_generation_fd(monkeypatch, tmp_path): + import os + from types import SimpleNamespace + import pytest + import tht.jobs.dwh_pipeline as module + + pipeline = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("catalog"), + build_lsh=lambda physical, output: _write_lsh([], physical, output), + ) + report = pipeline.run() + run_dir = tmp_path / ".tht-jobs" / "dwh" / "runs" / report.run_id + real_active = module._active_generation_fd + + def mismatched_active(root_fd, binding): + generation, generation_fd = real_active(root_fd, binding) + return "f" * 32, generation_fd + + monkeypatch.setattr(module, "_active_generation_fd", mismatched_active) + source = SimpleNamespace( + run_id=report.run_id, + stages=(SimpleNamespace( + status="running", effect_state="intent", name="lsh", + artifact_files=("physical.yaml", "demo_lsh.pkl", "demo_minhashes.pkl", + "demo_meta.json"), + ),), + ) + before = len(os.listdir("/dev/fd")) + with pytest.raises(Exception, match="not ACTIVE"): + pipeline._reconcile_effects(source, run_dir) + assert len(os.listdir("/dev/fd")) == before + + +def test_pipeline_releases_materialized_snapshot_after_every_run(tmp_path): + import tht.jobs.dwh_pipeline as module + + pipeline = DwhPreprocessPipeline( + workspace_id="demo", workspace_root=tmp_path, + config_fingerprint=FP, input_fingerprint=FP, + introspect=lambda output: output.write_text("catalog"), + build_lsh=lambda physical, output: _write_lsh([], physical, output), + ) + baseline = set(module._SNAPSHOT_DIRS) + for _ in range(3): + assert pipeline.run().status == "succeeded" + assert set(module._SNAPSHOT_DIRS) == baseline + assert pipeline._snapshot_holder is None + pipeline.introspect = lambda output: (_ for _ in ()).throw(RuntimeError("injected")) + assert pipeline.run().status == "failed" + assert set(module._SNAPSHOT_DIRS) == baseline + assert pipeline._snapshot_holder is None + + def test_publish_root_swap_after_lease_never_writes_replacement(monkeypatch, tmp_path): import tht.jobs.dwh_pipeline as module diff --git a/harness/tht/jobs/dwh_pipeline.py b/harness/tht/jobs/dwh_pipeline.py index bda6eacf..87783fba 100644 --- a/harness/tht/jobs/dwh_pipeline.py +++ b/harness/tht/jobs/dwh_pipeline.py @@ -366,7 +366,7 @@ def _open_generation_fd(generations_fd: int, generation: str) -> int: def _active_generation_fd( - root_fd: int, expected_binding: dict[str, str] + root_fd: int, expected_binding: dict[str, str], *, validate: bool = True ) -> tuple[str, int] | None: _validate_root_binding_fd(root_fd, expected_binding) try: @@ -394,7 +394,8 @@ def _active_generation_fd( finally: os.close(generations_fd) try: - validate_generation_fd(generation_fd, generation, expected_binding) + if validate: + validate_generation_fd(generation_fd, generation, expected_binding) return generation, generation_fd except BaseException: os.close(generation_fd) @@ -408,10 +409,10 @@ def _materialize_generation_fd( _SNAPSHOT_DIRS.add(holder) snapshot_root = holder try: - manifest = validate_generation_fd(generation_fd, generation, binding) - for name in manifest["files"]: + _, payloads = _read_validated_generation_fd(generation_fd, generation, binding) + for name, payload in payloads.items(): destination = snapshot_root / name - destination.write_bytes(_read_owned_at(generation_fd, name, readonly=True)) + destination.write_bytes(payload) destination.chmod(0o400) return holder, snapshot_root except BaseException: @@ -423,6 +424,15 @@ def _materialize_generation_fd( def validate_generation_fd( directory_fd: int, generation: str, expected_binding: dict[str, str] | None = None, ) -> dict: + manifest, _ = _read_validated_generation_fd( + directory_fd, generation, expected_binding + ) + return manifest + + +def _read_validated_generation_fd( + directory_fd: int, generation: str, expected_binding: dict[str, str] | None = None, +) -> tuple[dict, dict[str, bytes]]: try: directory_info = os.fstat(directory_fd) if ( @@ -453,13 +463,15 @@ def validate_generation_fd( ) if set(os.listdir(directory_fd)) != set(files) | {GENERATION_MANIFEST}: raise ValueError + payloads = {} for name, expected in files.items(): if name != "physical.yaml" and not _SAFE_FILE.fullmatch(name): raise ValueError payload = _read_owned_at(directory_fd, name, readonly=True) if hashlib.sha256(payload).hexdigest() != expected: raise ValueError - return manifest + payloads[name] = payload + return manifest, payloads except (OSError, KeyError, TypeError, ValueError, UnicodeDecodeError, json.JSONDecodeError) as error: raise CorruptCheckpointError("published DWH generation is invalid") from error @@ -478,7 +490,7 @@ def resolve_dwh_snapshot(cfg) -> DwhArtifactSnapshot: def _resolve_dwh_snapshot_locked( cfg, binding: dict[str, str], root_fd: int, ) -> DwhArtifactSnapshot: - active = _active_generation_fd(root_fd, binding) + active = _active_generation_fd(root_fd, binding, validate=False) if active is None: return DwhArtifactSnapshot( None, cfg.paths.artifacts / "mschema" / "physical.yaml", cfg.paths.indexes / "lsh" @@ -576,7 +588,7 @@ class DwhPreprocessPipeline: _claim_or_validate_root_binding(lease.root_fd, self.binding) lease.assert_root_identity(self.workspace_root / ".tht-dwh") self._assert_active_binding(lease.root_fd) - active = _active_generation_fd(lease.root_fd, self.binding) + active = _active_generation_fd(lease.root_fd, self.binding, validate=False) if active is not None: generation, generation_fd = active try: @@ -633,10 +645,22 @@ class DwhPreprocessPipeline: return self._publish_stage(context, "lsh", spec) implementations = {"introspect": introspect_stage, "lsh": lsh_stage} - return run_job( - spec, tuple(implementations[step] for step in steps), - reconcile_effects=self._reconcile_effects, - ) + try: + return run_job( + spec, tuple(implementations[step] for step in steps), + reconcile_effects=self._reconcile_effects, + ) + finally: + holder, self._snapshot_holder = self._snapshot_holder, None + if isinstance(holder, Path): + shutil.rmtree(holder, ignore_errors=True) + _SNAPSHOT_DIRS.discard(holder) + if self.current_physical is not None and holder in self.current_physical.parents: + self.current_physical = None + if self.current_lsh_dir is not None and ( + self.current_lsh_dir == holder or holder in self.current_lsh_dir.parents + ): + self.current_lsh_dir = None def _reconcile_effects(self, source, run_dir: Path) -> set[str]: running = next( @@ -654,9 +678,12 @@ class DwhPreprocessPipeline: run_dir / "artifacts", running.artifact_files, ) active = _active_generation_fd(lease.root_fd, self.binding) - if active is None or active[0] != source.run_id: - raise CorruptCheckpointError("sealed DWH publication is not ACTIVE") - os.close(active[1]) + try: + if active is None or active[0] != source.run_id: + raise CorruptCheckpointError("sealed DWH publication is not ACTIVE") + finally: + if active is not None: + os.close(active[1]) finally: os.close(generations_fd) finally: