Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 95 additions & 0 deletions src/brigade/run_checkpoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
10 changes: 10 additions & 0 deletions src/brigade/run_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
50 changes: 47 additions & 3 deletions src/brigade/work_cmd/verification.py
Original file line number Diff line number Diff line change
Expand Up @@ -551,31 +551,68 @@ 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
if dest_sha != source_receipt_sha:
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
Expand All @@ -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}")
Expand Down
78 changes: 78 additions & 0 deletions tests/test_run_checkpoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import hashlib
import json
import os
import shutil
import signal
import stat
import subprocess
Expand Down Expand Up @@ -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)
33 changes: 33 additions & 0 deletions tests/test_work_cmd_verification.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Loading