From 397ac6c04a22a1b6fd1bea87698fe6fa880c2af6 Mon Sep 17 00:00:00 2001 From: Solomon Neas Date: Sat, 1 Aug 2026 00:10:52 -0400 Subject: [PATCH] fix(runs): strip private recovery-checkpoint bodies at export Keep local recovery semantics intact while ensuring verify-archive and other export copies only carry the closed artifact-reference shape. Co-authored-by: Cursor --- src/brigade/run_checkpoint.py | 95 ++++++++++++++++++++++++++++ src/brigade/run_events.py | 10 +++ src/brigade/work_cmd/verification.py | 50 ++++++++++++++- tests/test_run_checkpoint.py | 78 +++++++++++++++++++++++ tests/test_work_cmd_verification.py | 33 ++++++++++ 5 files changed, 263 insertions(+), 3 deletions(-) diff --git a/src/brigade/run_checkpoint.py b/src/brigade/run_checkpoint.py index d50a1f8d..d45fdaf5 100644 --- a/src/brigade/run_checkpoint.py +++ b/src/brigade/run_checkpoint.py @@ -1178,3 +1178,98 @@ def recover_from_checkpoint( except run_projector.ProjectionError as exc: raise CheckpointError(_bound("projection failed"), category="projection") from exc return _restore_run_json_from_checkpoint(run_dir, restore_bytes, run_meta=run_meta) + + +# -- Export boundary (issue #636) --------------------------------------------- + +_ARTIFACT_REFERENCE_KEYS = frozenset({"path", "sha256", "media_type", "byte_size", "privacy_class"}) + + +def checkpoint_artifact_reference(*, sha256: str, byte_size: int) -> dict[str, Any]: + """Return the closed artifact-reference shape used at export boundaries.""" + if not isinstance(sha256, str) or not _HEX64.fullmatch(sha256): + raise CheckpointError(_bound("checkpoint export digest is invalid"), category="export-privacy") + if isinstance(byte_size, bool) or not isinstance(byte_size, int) or byte_size < 0: + raise CheckpointError(_bound("checkpoint export byte size is invalid"), category="export-privacy") + return { + "path": f"events/{CHECKPOINT_DIR_NAME}/{sha256}.json", + "sha256": sha256, + "media_type": CHECKPOINT_MEDIA_TYPE, + "byte_size": byte_size, + "privacy_class": CHECKPOINT_PRIVACY_CLASS, + } + + +def refuse_checkpoint_body_export(*, reason: str = "checkpoint body is private") -> None: + """Refuse an export that would emit checkpoint body content.""" + raise CheckpointError( + _bound(f"{reason} (privacy_class={CHECKPOINT_PRIVACY_CLASS})"), + category="export-privacy", + ) + + +def is_checkpoint_artifact_reference(payload: Mapping[str, Any]) -> bool: + """True when payload is exactly the closed artifact-reference shape.""" + if set(payload) != _ARTIFACT_REFERENCE_KEYS: + return False + sha = payload.get("sha256") + path = payload.get("path") + byte_size = payload.get("byte_size") + return ( + isinstance(sha, str) + and bool(_HEX64.fullmatch(sha)) + and path == f"events/{CHECKPOINT_DIR_NAME}/{sha}.json" + and payload.get("media_type") == CHECKPOINT_MEDIA_TYPE + and payload.get("privacy_class") == CHECKPOINT_PRIVACY_CLASS + and not isinstance(byte_size, bool) + and isinstance(byte_size, int) + and byte_size >= 0 + ) + + +def strip_checkpoint_bodies_for_export(run_dir: Path) -> list[dict[str, Any]]: + """Replace recovery-checkpoint file bodies with artifact references. + + Local recovery continues to read the original private bodies under the run + directory. Call this only on a copy that is about to leave the run tree + (archive, bundle, sync, or similar export boundary). + """ + cp_dir = checkpoint_dir(run_dir) + if not cp_dir.is_dir(): + return [] + replaced: list[dict[str, Any]] = [] + for path in sorted(cp_dir.glob("*.json")): + if not path.is_file() or path.is_symlink(): + refuse_checkpoint_body_export(reason="checkpoint export path is not a regular file") + raw = path.read_bytes() + try: + parsed = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + parsed = None + if isinstance(parsed, dict) and is_checkpoint_artifact_reference(parsed): + replaced.append(dict(parsed)) + continue + sha = hashlib.sha256(raw).hexdigest() + if path.name != f"{sha}.json": + refuse_checkpoint_body_export(reason="checkpoint export filename digest mismatch") + reference = checkpoint_artifact_reference(sha256=sha, byte_size=len(raw)) + path.write_text(json.dumps(reference, indent=2, sort_keys=True) + "\n", encoding="utf-8") + os.chmod(path, 0o600) + replaced.append(reference) + return replaced + + +def assert_export_tree_has_no_checkpoint_bodies(root: Path) -> None: + """Fail closed if any recovery-checkpoint file under root still holds a body.""" + for path in sorted(Path(root).rglob(f"*/{CHECKPOINT_DIR_NAME}/*.json")): + if not path.is_file(): + continue + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise CheckpointError( + _bound("checkpoint export body is unreadable"), + category="export-privacy", + ) from exc + if not isinstance(payload, dict) or not is_checkpoint_artifact_reference(payload): + refuse_checkpoint_body_export(reason="checkpoint body crossed an export boundary") diff --git a/src/brigade/run_events.py b/src/brigade/run_events.py index c5a6d629..5c16e6a0 100644 --- a/src/brigade/run_events.py +++ b/src/brigade/run_events.py @@ -138,6 +138,16 @@ } ) +# Recovery-checkpoint bodies under ``events/recovery-checkpoints/`` are +# ``privacy_class: private`` (see ``run_checkpoint.CHECKPOINT_PRIVACY_CLASS``). +# Journal event payloads only carry the closed artifact-reference shape +# (relative path, sha256, media type, byte size, privacy class). Any exporter +# or collector that copies a run directory across a boundary must strip those +# bodies, replace them with that artifact-reference shape, or refuse with a +# bounded error naming the privacy class. Local recovery may still read the +# private bodies in place. Helpers: ``run_checkpoint.strip_checkpoint_bodies_for_export``, +# ``run_checkpoint.refuse_checkpoint_body_export``. + class CanonicalizationError(ValueError): """Raised when a value cannot be canonicalized under the strict rules.""" diff --git a/src/brigade/work_cmd/verification.py b/src/brigade/work_cmd/verification.py index 7b8af37f..f45f980b 100644 --- a/src/brigade/work_cmd/verification.py +++ b/src/brigade/work_cmd/verification.py @@ -551,24 +551,59 @@ def _verify_archive_tree_manifest(root: Path) -> dict[str, tuple[str, str | None return manifest +def _expected_verify_archive_manifest( + run_dir: Path, source_manifest: dict[str, tuple[str, str | None]] +) -> dict[str, tuple[str, str | None]]: + """Return the tree manifest after recovery-checkpoint export stripping (#636).""" + from brigade import run_checkpoint + + expected = dict(source_manifest) + prefix = f"events/{run_checkpoint.CHECKPOINT_DIR_NAME}/" + for relative, (kind, _digest) in list(expected.items()): + if kind != "file" or not relative.startswith(prefix) or not relative.endswith(".json"): + continue + path = run_dir / relative + raw = path.read_bytes() + try: + parsed = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + parsed = None + if isinstance(parsed, dict) and run_checkpoint.is_checkpoint_artifact_reference(parsed): + continue + sha = hashlib.sha256(raw).hexdigest() + if path.name != f"{sha}.json": + raise OSError(f"verify archive checkpoint export filename digest mismatch: {relative}") + reference = run_checkpoint.checkpoint_artifact_reference(sha256=sha, byte_size=len(raw)) + ref_bytes = (json.dumps(reference, indent=2, sort_keys=True) + "\n").encode("utf-8") + expected[relative] = ("file", hashlib.sha256(ref_bytes).hexdigest()) + return expected + + def _archive_verify_run(run_dir: Path, archive_root: Path) -> dict[str, Any]: """Copy one run dir into the archive and append an index entry. Raises on failure. Integrity is checked twice: the archived receipt bytes must match the source bytes, and a receipt that carries ``digests.receipt_sha256`` must still re-hash to that value after the copy. Callers must treat any exception as - "do not delete the original". + "do not delete the original". Recovery-checkpoint bodies are private and are + replaced with the closed artifact-reference shape before the archive lands + (issue #636); local source run dirs are left unchanged. """ + from brigade import run_checkpoint + run_id = run_dir.name dest = archive_root / run_id source_manifest = _verify_archive_tree_manifest(run_dir) + expected_manifest = _expected_verify_archive_manifest(run_dir, source_manifest) receipt_path = run_dir / "receipt.json" source_receipt_sha = localio.file_sha256(receipt_path) if receipt_path.is_file() else None if dest.is_symlink(): raise OSError(f"verify archive conflict: {dest} is a symlink") if dest.exists(): - # Re-archive of the same run id is only safe when the evidence is identical. - if _verify_archive_tree_manifest(dest) != source_manifest: + # Re-archive is safe when evidence matches the source, or when the + # existing archive is already the privacy-normalized export of source. + dest_manifest = _verify_archive_tree_manifest(dest) + if dest_manifest not in (source_manifest, expected_manifest): raise OSError(f"verify archive conflict: {dest} already exists with different evidence") dest_receipt = dest / "receipt.json" dest_sha = localio.file_sha256(dest_receipt) if dest_receipt.is_file() else None @@ -576,6 +611,8 @@ def _archive_verify_run(run_dir: Path, archive_root: Path) -> dict[str, Any]: raise OSError(f"verify archive conflict: {dest} already exists with different evidence") if dest_sha is not None: _assert_archived_receipt_integrity(dest_receipt) + run_checkpoint.strip_checkpoint_bodies_for_export(dest) + run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(dest) entry = _verify_archive_index_entry(run_dir, dest, dest_sha, already_archived=True) _append_verify_archive_index(archive_root, entry) return entry @@ -590,6 +627,13 @@ def _archive_verify_run(run_dir: Path, archive_root: Path) -> dict[str, Any]: if copied_sha != source_receipt_sha: raise OSError(f"verify archive copy integrity check failed: {run_id}") _assert_archived_receipt_integrity(staging / "receipt.json") + # Export boundary (#636): never archive private recovery-checkpoint + # bodies. Replace them with the closed artifact-reference shape after + # the byte-identical copy check so local recovery semantics stay intact. + run_checkpoint.strip_checkpoint_bodies_for_export(staging) + run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(staging) + if _verify_archive_tree_manifest(staging) != expected_manifest: + raise OSError(f"verify archive export privacy normalize mismatch: {run_id}") os.rename(staging, dest) if _verify_archive_tree_manifest(run_dir) != source_manifest: raise OSError(f"verify archive source changed during copy: {run_id}") diff --git a/tests/test_run_checkpoint.py b/tests/test_run_checkpoint.py index 7325da8d..96d1489c 100644 --- a/tests/test_run_checkpoint.py +++ b/tests/test_run_checkpoint.py @@ -9,6 +9,7 @@ import hashlib import json import os +import shutil import signal import stat import subprocess @@ -3266,3 +3267,80 @@ def test_write_text_atomic_sigkill_during_temp_window_preserves_one_valid_payloa ) for run_dir in attempt_run_dirs: assert _write_text_atomic_temp_paths(run_dir) == [] + + +# -- Issue #636: checkpoint export privacy ----------------------------------- + + +def test_strip_checkpoint_bodies_for_export_replaces_private_body(tmp_path): + run_dir = _run_dir(tmp_path) + private_task = "SECRET_TASK_PROMPT_do_not_export" + private_error = "SECRET_ERROR_TRACE_do_not_export" + body = _writer_bytes( + { + "schema": "brigade.run.v1", + "status": "failed", + "task": private_task, + "error": private_error, + } + ) + placed = _place_checkpoint_file(run_dir, body) + assert private_task in placed.read_text(encoding="utf-8") + + export_copy = tmp_path / "export-copy" + shutil.copytree(run_dir, export_copy) + + refs = run_checkpoint.strip_checkpoint_bodies_for_export(export_copy) + assert len(refs) == 1 + assert run_checkpoint.is_checkpoint_artifact_reference(refs[0]) + assert refs[0]["privacy_class"] == "private" + assert refs[0]["sha256"] == hashlib.sha256(body).hexdigest() + assert refs[0]["byte_size"] == len(body) + + exported = (export_copy / "events" / "recovery-checkpoints" / placed.name).read_text(encoding="utf-8") + assert private_task not in exported + assert private_error not in exported + assert '"task"' not in exported + assert json.loads(exported) == refs[0] + + # Local recovery source is unchanged. + assert placed.read_bytes() == body + run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(export_copy) + + +def test_assert_export_tree_refuses_checkpoint_body(tmp_path): + run_dir = _run_dir(tmp_path) + body = _writer_bytes({"schema": "brigade.run.v1", "status": "planning", "task": "keep-private"}) + _place_checkpoint_file(run_dir, body) + + with pytest.raises(run_checkpoint.CheckpointError, match="privacy_class=private") as exc_info: + run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(run_dir) + assert exc_info.value.category == "export-privacy" + + +def test_refuse_checkpoint_body_export_names_privacy_class(): + with pytest.raises(run_checkpoint.CheckpointError, match="privacy_class=private") as exc_info: + run_checkpoint.refuse_checkpoint_body_export() + assert exc_info.value.category == "export-privacy" + + +def test_local_recovery_round_trips_after_export_strip_of_copy(tmp_path): + """Export stripping a copy must not change local recovery semantics.""" + workspace = _workspace(tmp_path) + run_dir = _run_dir(tmp_path) + run_json_obj = { + "schema": "brigade.run.v1", + "status": "planning", + "task": "SECRET_TASK_for_local_recovery_only", + } + _activated_journal_with_checkpoint(workspace, run_dir, run_json_obj) + + export_copy = tmp_path / "export-copy" + shutil.copytree(run_dir, export_copy) + run_checkpoint.strip_checkpoint_bodies_for_export(export_copy) + run_checkpoint.assert_export_tree_has_no_checkpoint_bodies(export_copy) + + (run_dir / "run.json").unlink() + repaired = run_checkpoint.recover_from_checkpoint(run_dir, None) + assert repaired == run_json_obj + assert (run_dir / "run.json").read_bytes() == _writer_bytes(run_json_obj) diff --git a/tests/test_work_cmd_verification.py b/tests/test_work_cmd_verification.py index cbc5336b..4d2433a7 100644 --- a/tests/test_work_cmd_verification.py +++ b/tests/test_work_cmd_verification.py @@ -2668,3 +2668,36 @@ def test_verify_receipt_identity_preserves_real_index(tmp_target, monkeypatch, c text=True, ).stdout assert after == before + + +def test_archive_verify_run_strips_recovery_checkpoint_bodies(tmp_path): + """Export boundary: verify-archive must not emit private checkpoint bodies (#636).""" + from brigade import run_checkpoint + from brigade.work_cmd import helpers, verification + + root = helpers._verify_runs_root(tmp_path) + root.mkdir(parents=True) + run_dir, _ = _write_verify_run_dir(root, "20260101-000001-a", sign=True) + _write_verify_run_dir(root, "20260101-000002-b") + + private_task = "SECRET_CHECKPOINT_TASK_must_not_archive" + body_obj = {"schema": "brigade.run.v1", "status": "failed", "task": private_task, "error": "boom"} + body = (json.dumps(body_obj, indent=2, sort_keys=True) + "\n").encode("utf-8") + cp_dir = run_dir / "events" / "recovery-checkpoints" + cp_dir.mkdir(parents=True) + sha = hashlib.sha256(body).hexdigest() + (cp_dir / f"{sha}.json").write_bytes(body) + os.chmod(cp_dir / f"{sha}.json", 0o600) + + archive_root = tmp_path / "verify-archive" + removed = verification._prune_verify_runs(tmp_path, keep=1, archive_root=archive_root) + + assert removed == 1 + archived_cp = archive_root / "20260101-000001-a" / "events" / "recovery-checkpoints" / f"{sha}.json" + assert archived_cp.is_file() + payload = json.loads(archived_cp.read_text(encoding="utf-8")) + assert run_checkpoint.is_checkpoint_artifact_reference(payload) + assert private_task not in archived_cp.read_text(encoding="utf-8") + assert '"task"' not in archived_cp.read_text(encoding="utf-8") + # Source verify-run is deleted after archive; the privacy rule is on the export. + assert not (root / "20260101-000001-a").exists()