diff --git a/docs/command-inventory.md b/docs/command-inventory.md index 768a849e..189e29b5 100644 --- a/docs/command-inventory.md +++ b/docs/command-inventory.md @@ -56,7 +56,7 @@ enabled: run `brigade extras on` once, or set `BRIGADE_EXTRAS=1`. - `brigade route`: 1 command path(s) - `brigade run`: 1 command path(s) - `brigade runbook` (extras): 5 command path(s) -- `brigade runs`: 8 command path(s) +- `brigade runs`: 9 command path(s) - `brigade scrub`: 1 command path(s) - `brigade search`: 6 command path(s) - `brigade security`: 15 command path(s) @@ -443,6 +443,7 @@ enabled: run `brigade extras on` once, or set `BRIGADE_EXTRAS=1`. - `brigade runs latest` - `brigade runs list` - `brigade runs recover` +- `brigade runs redact` - `brigade runs resume` - `brigade runs show` - `brigade runs steer` diff --git a/src/brigade/cli/runs.py b/src/brigade/cli/runs.py index 9659412a..475ef973 100644 --- a/src/brigade/cli/runs.py +++ b/src/brigade/cli/runs.py @@ -110,6 +110,42 @@ def register(sub: argparse._SubParsersAction) -> None: default=None, help="Explicit runs directory for run ids. Defaults to .brigade/runs under --cwd.", ) + p_runs_redact = runs_sub.add_parser( + "redact", + help="Run the explicit operator procedure for lifecycle journal redaction.", + ) + p_runs_redact.add_argument("run", help="Run directory path, run id under --runs-dir, or 'latest'.") + p_runs_redact.add_argument( + "--cwd", + type=Path, + default=Path("."), + help="Workspace whose default .brigade/runs directory should be used for run ids.", + ) + p_runs_redact.add_argument( + "--runs-dir", + type=Path, + default=None, + help="Explicit runs directory for run ids. Defaults to .brigade/runs under --cwd.", + ) + p_runs_redact.add_argument("--from-sequence", dest="sequence_start", type=int, default=None) + p_runs_redact.add_argument("--to-sequence", dest="sequence_end", type=int, default=None) + p_runs_redact.add_argument( + "--reason", + default=None, + help="Closed incident reason code; never include the private value.", + ) + p_runs_redact.add_argument( + "--cleanup-quarantine", + dest="cleanup_operation", + default=None, + metavar="OPERATION_ID", + help="Explicitly remove a previously verified quarantine.", + ) + p_runs_redact.add_argument( + "--operator-confirm", + action="store_true", + help="Confirm this operator-only incident procedure.", + ) p_runs_resume = runs_sub.add_parser( "resume", help="Re-attach interrupted app-server workers from a run and re-synthesize." ) @@ -148,6 +184,17 @@ def dispatch(args) -> int: return _control_request(args.run, cwd=args.cwd, runs_dir=args.runs_dir, payload=payload) if args.runs_command == "recover": return runs_cmd.recover(args.run, cwd=args.cwd, runs_dir=args.runs_dir) + if args.runs_command == "redact": + return runs_cmd.redact( + args.run, + cwd=args.cwd, + runs_dir=args.runs_dir, + sequence_start=args.sequence_start, + sequence_end=args.sequence_end, + reason=args.reason, + operator_confirmed=args.operator_confirm, + cleanup_operation=args.cleanup_operation, + ) if args.runs_command == "resume": return runs_cmd.resume(args.run_dir) args._brigade_parser.error(f"unknown runs command: {args.runs_command}") diff --git a/src/brigade/run_redaction.py b/src/brigade/run_redaction.py new file mode 100644 index 00000000..897d7a71 --- /dev/null +++ b/src/brigade/run_redaction.py @@ -0,0 +1,2221 @@ +"""Operator-only redaction for authoritative lifecycle journals. + +Normal lifecycle writes remain append-only. This module is the exceptional +incident procedure for removing exposed payload bytes from a closed run: + +1. verify the authoritative journal, projection, and absent run lock; +2. durably quarantine the original journal under a deterministic transaction; +3. replace affected payloads and idempotency keys, then re-chain the journal; +4. atomically replace the active journal and compatibility projection; +5. verify both again before writing a bounded redaction record. + +The quarantine is retained by default. Deleting it requires a separate, +explicit cleanup call which repeats chain and projection verification first. +Transaction state makes retries safe across crashes before or after journal +replacement. Standard library only. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import stat +import threading +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping, Sequence +from uuid import uuid4 + +from brigade import localio, run_checkpoint, run_events, run_journal, run_projector, run_shadow, runguard + +REDACTION_SCHEMA = "brigade.run_redaction.v1" +REDACTION_SCHEMA_VERSION = 1 +MAX_RUN_JSON_BYTES = 1024 * 1024 +REDACTED_VALUE = "[REDACTED]" +REASON_CODES = frozenset( + { + "credential-exposure", + "personal-data-exposure", + "policy-removal", + "other-sensitive-data", + } +) +TERMINAL_STATUSES = frozenset({"ok", "dry-run", "failed", "timeout", "incomplete", "canceled"}) + +_FILE_MODE = 0o600 +_DIR_MODE = 0o700 +_OPERATION_RE = re.compile(r"^redact-[0-9a-f]{16}$") +_QUARANTINE_TEMP_RE = re.compile(r"^\.original\.jsonl\.[0-9a-f]{32}\.tmp$") +_STATE_TEMP_RE = re.compile(r"^\.state\.json\.[0-9a-f]{32}\.tmp$") +_RECORD_TEMP_RE = re.compile(r"^\.record\.json\.[0-9a-f]{32}\.tmp$") +_PROCESS_LOCK = threading.Lock() +_PROJECTION_DIGEST_FIELD = "journal_last_event_digest" +_PLATFORM_NAME = os.name +_UNSUPPORTED_PLATFORM = "unsupported platform for safe redaction transactions" +_REQUIRED_DIR_FD_OPERATIONS = (os.open, os.mkdir, os.rename, os.unlink, os.link) +_LINK_OPERATION = os.link +_LISTDIR_OPERATION = os.listdir + + +class RedactionError(RuntimeError): + """A bounded redaction failure which never carries journal payload values.""" + + def __init__(self, diagnostic: str) -> None: + bounded = _bound(diagnostic) + super().__init__(bounded) + self.diagnostic = bounded + + +@dataclass(frozen=True) +class RedactionReport: + """Paths and state for one redaction transaction.""" + + operation_id: str + sequence_start: int + sequence_end: int + quarantine_path: Path + record_path: Path + cleaned: bool = False + + +def _require_secure_transaction_platform() -> None: + if ( + _PLATFORM_NAME != "posix" + or any(operation not in os.supports_dir_fd for operation in _REQUIRED_DIR_FD_OPERATIONS) + or _LISTDIR_OPERATION not in os.supports_fd + or _LINK_OPERATION not in os.supports_follow_symlinks + or run_journal._O_NOFOLLOW == 0 + or run_journal._O_DIRECTORY == 0 + or not run_journal._HAS_FCHMOD + or not callable(getattr(os, "fsync", None)) + ): + raise RedactionError(_UNSUPPORTED_PLATFORM) + + +def _probe_secure_transaction_directory(run_dir: Path) -> None: + path = run_dir / "events" + try: + info = os.lstat(path) + except OSError as exc: + raise RedactionError("events path validation failed") from exc + if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): + raise RedactionError("events path is not a safe directory") + if getattr(info, "st_file_attributes", 0) & 0x400: + raise RedactionError("events path is a reparse point") + try: + if path.resolve(strict=True) != path.absolute(): + raise RedactionError("events path traverses a link") + except (OSError, RuntimeError) as exc: + raise RedactionError("events path validation failed") from exc + try: + fd = run_journal._open_nofollow( + path, + os.O_RDONLY | run_journal._O_DIRECTORY, + ) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(_UNSUPPORTED_PLATFORM) from exc + + primary: BaseException | None = None + try: + try: + opened = os.fstat(fd) + if not stat.S_ISDIR(opened.st_mode) or opened.st_dev != info.st_dev or opened.st_ino != info.st_ino: + raise RedactionError("events directory identity changed") + os.fsync(fd) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(_UNSUPPORTED_PLATFORM) from exc + except BaseException as exc: + primary = exc + raise + finally: + try: + os.close(fd) + except BaseException as exc: + if primary is None: + raise RedactionError(_UNSUPPORTED_PLATFORM) from exc + + +def _bound(message: str) -> str: + if len(message) <= run_events.MAX_DIAGNOSTIC_LEN: + return message + return message[: run_events.MAX_DIAGNOSTIC_LEN - 1] + "…" + + +def _operation_id(run_id: str, sequence_start: int, sequence_end: int, reason: str) -> str: + request = { + "run_id": run_id, + "sequence_start": sequence_start, + "sequence_end": sequence_end, + "reason_code": reason, + } + encoded = json.dumps(request, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + return f"redact-{hashlib.sha256(encoded).hexdigest()[:16]}" + + +def _operation_paths(run_dir: Path, operation_id: str) -> tuple[Path, Path, Path]: + operation_dir = run_dir / "events" / "redactions" / operation_id + return operation_dir, operation_dir / "original.jsonl", operation_dir / "record.json" + + +def _validate_request( + sequence_start: object, + sequence_end: object, + reason: object, + *, + operator_confirmed: bool, +) -> tuple[int, int, str]: + if operator_confirmed is not True: + raise RedactionError("operator confirmation is required") + if ( + isinstance(sequence_start, bool) + or not isinstance(sequence_start, int) + or isinstance(sequence_end, bool) + or not isinstance(sequence_end, int) + or sequence_start < 1 + or sequence_end < sequence_start + ): + raise RedactionError("invalid sequence range") + if not isinstance(reason, str) or reason not in REASON_CODES: + raise RedactionError(f"reason code must be one of: {', '.join(sorted(REASON_CODES))}") + return sequence_start, sequence_end, reason + + +def _resolve_run_dir(run_dir: Path) -> Path: + path = Path(run_dir).expanduser() + try: + if stat.S_ISLNK(os.lstat(path).st_mode): + raise RedactionError("run directory must not be a symlink") + resolved = path.resolve(strict=True) + except RedactionError: + raise + except (OSError, RuntimeError) as exc: + raise RedactionError("run directory is not resolvable") from exc + if not resolved.is_dir(): + raise RedactionError("run directory is not a directory") + return resolved + + +def _read_bounded_regular(path: Path, *, limit: int, category: str) -> bytes: + try: + fd = run_journal._open_nofollow(path, os.O_RDONLY) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(f"{category} read failed") from exc + try: + try: + info = os.fstat(fd) + except OSError as exc: + raise RedactionError(f"{category} stat failed") from exc + if not stat.S_ISREG(info.st_mode): + raise RedactionError(f"{category} is not a regular file") + if info.st_nlink != 1: + raise RedactionError(f"{category} link count is not one") + if stat.S_IMODE(info.st_mode) != _FILE_MODE: + try: + run_journal._chmod_fd_or_path(fd, path, _FILE_MODE) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(f"{category} mode normalization failed") from exc + if info.st_size < 0 or info.st_size > limit: + raise RedactionError(f"{category} exceeds size bound") + data = bytearray() + while len(data) < info.st_size: + try: + chunk = os.read(fd, min(run_journal._READ_CHUNK, info.st_size - len(data))) + except OSError as exc: + raise RedactionError(f"{category} read failed") from exc + if not chunk: + break + data.extend(chunk) + if len(data) != info.st_size: + raise RedactionError(f"{category} changed during read") + try: + extra = os.read(fd, 1) + except OSError as exc: + raise RedactionError(f"{category} read failed") from exc + if extra: + raise RedactionError(f"{category} changed during read") + return bytes(data) + finally: + os.close(fd) + + +def _parse_json_object(raw: bytes, *, category: str) -> dict[str, Any]: + def no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ValueError("duplicate JSON key") + result[key] = value + return result + + try: + payload = json.loads(raw.decode("utf-8"), object_pairs_hook=no_duplicate_keys) + except (UnicodeDecodeError, json.JSONDecodeError, RecursionError, ValueError) as exc: + raise RedactionError(f"{category} is malformed") from exc + if not isinstance(payload, dict): + raise RedactionError(f"{category} is not an object") + return payload + + +def _load_json_object(path: Path, *, limit: int, category: str) -> dict[str, Any]: + return _parse_json_object( + _read_bounded_regular(path, limit=limit, category=category), + category=category, + ) + + +def _snapshot_gate(run_dir: Path) -> tuple[dict[str, Any], Path]: + snapshot = _load_json_object( + run_dir / "run.json", + limit=MAX_RUN_JSON_BYTES, + category="run projection", + ) + if snapshot.get("run_journal_authority_requested") is not True: + raise RedactionError("run is not an authoritative journal run") + if snapshot.get("status") not in TERMINAL_STATUSES: + raise RedactionError("redaction requires a closed terminal run") + try: + workspace = runguard.resolve_run_lock_workspace(snapshot, run_dir) + except (OSError, RuntimeError, runguard.RunGuardError) as exc: + raise RedactionError("run lock workspace is ambiguous") from exc + if workspace is None: + raise RedactionError("run lock workspace is ambiguous") + return snapshot, workspace + + +def _run_snapshot_and_lock_gate(run_dir: Path) -> tuple[dict[str, Any], Path]: + snapshot, workspace = _snapshot_gate(run_dir) + try: + lock_path = runguard.lock_path(workspace) + except (OSError, RuntimeError, runguard.RunGuardError) as exc: + raise RedactionError("run lock workspace is ambiguous") from exc + state = runguard.run_lock_state(workspace, run_dir) + if state == "absent" and os.path.lexists(lock_path): + raise RedactionError("run lock state is ambiguous") + if state != "absent": + raise RedactionError(f"run lock state is {state}; redaction requires an absent lock") + return snapshot, workspace + + +@contextmanager +def _exclusive_redaction_lock(run_dir: Path): + _, workspace = _run_snapshot_and_lock_gate(run_dir) + try: + with runguard.run_lock(workspace, run_dir=run_dir): + if not runguard.is_active_run_owner(workspace, run_dir): + raise RedactionError("redaction could not prove exclusive run lock ownership") + snapshot, locked_workspace = _snapshot_gate(run_dir) + if locked_workspace != workspace: + raise RedactionError("run lock workspace changed during redaction") + yield snapshot, workspace + except RedactionError: + raise + except runguard.RunGuardError as exc: + raise RedactionError("redaction could not acquire exclusive run lock") from exc + + +def _assert_active_owner(workspace: Path, run_dir: Path) -> None: + if not runguard.is_active_run_owner(workspace, run_dir): + raise RedactionError("redaction lost exclusive run lock ownership") + + +def _verified_events(journal_path: Path, *, category: str) -> list[run_journal.RunEvent]: + try: + report = run_journal.read_journal_bounded(journal_path) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(f"{category} verification failed") from exc + if report.partial_tail is not None or report.chain_errors or not report.events: + raise RedactionError(f"{category} verification failed") + return report.events + + +def _projection(snapshot: Mapping[str, Any], events: Sequence[run_journal.RunEvent]) -> run_projector.RunProjection: + try: + return run_projector.project_run_snapshot(snapshot, events, journal_present=True) + except run_projector.ProjectionError as exc: + raise RedactionError("projection verification failed") from exc + + +def _projection_semantics(snapshot: Mapping[str, Any]) -> dict[str, Any]: + return {key: value for key, value in snapshot.items() if key != _PROJECTION_DIGEST_FIELD} + + +def _verify_current_projection( + snapshot: Mapping[str, Any], + events: Sequence[run_journal.RunEvent], +) -> run_projector.RunProjection: + projected = _projection(snapshot, events) + if dict(snapshot) != projected.snapshot: + raise RedactionError("run projection is stale or inconsistent") + return projected + + +def _validate_checkpoint_artifacts( + run_dir: Path, + snapshot: Mapping[str, Any], + events: Sequence[run_journal.RunEvent], +) -> None: + checkpoint_events = [event for event in events if event.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE] + latest = run_checkpoint.latest_checkpoint_event(list(events)) + if latest is None or not checkpoint_events: + raise RedactionError("checkpoint verification failed") + latest_bytes: bytes | None = None + try: + for event in checkpoint_events: + checkpoint_bytes = run_checkpoint.validate_checkpoint(run_dir, event) + if event.sequence == latest.sequence: + latest_bytes = checkpoint_bytes + if latest_bytes is None: + raise RedactionError("checkpoint verification failed") + checkpoint_obj = run_checkpoint._parse_checkpoint_object(latest_bytes) + run_checkpoint._verify_coverage(list(events), latest, checkpoint_obj) + except RedactionError: + raise + except run_checkpoint.CheckpointError as exc: + raise RedactionError("checkpoint verification failed") from exc + if ( + latest.payload.get("body_kind") != "base-stripped" + or checkpoint_obj.get("run_journal_authority_requested") is not True + ): + raise RedactionError("checkpoint authority verification failed") + projected = _projection(checkpoint_obj, events) + if dict(snapshot) != projected.snapshot: + raise RedactionError("checkpoint authority projection mismatch") + + +def _redacted_payload(event: run_journal.RunEvent) -> dict[str, Any]: + # Checkpoint payload values are a closed structural reference verified by + # run_checkpoint. Replacing them would make recovery unverifiable. They + # cannot contain arbitrary lifecycle detail under the checkpoint schema. + if event.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE: + try: + run_checkpoint._validate_payload(event.payload) + except run_checkpoint.CheckpointError as exc: + raise RedactionError("checkpoint payload validation failed") from exc + return dict(event.payload) + + redacted: dict[str, Any] = {} + for key, value in event.payload.items(): + # Status is a projector input with a closed allowlist. Retaining the + # allowed status token preserves the observable run state. + if key == "status": + redacted[key] = value + elif value is None: + redacted[key] = None + elif isinstance(value, str): + redacted[key] = REDACTED_VALUE + elif isinstance(value, int) and not isinstance(value, bool): + redacted[key] = 0 + else: + raise RedactionError("affected payload contains an unsupported value") + return redacted + + +def _redaction_idempotency_key(operation_id: str, sequence: int) -> str: + return f"redaction:{operation_id}:{sequence}" + + +def _rewrite_events( + events: Sequence[run_journal.RunEvent], + *, + sequence_start: int, + sequence_end: int, + operation_id: str, +) -> tuple[list[run_journal.RunEvent], bytes]: + rewritten: list[run_journal.RunEvent] = [] + previous_digest: str | None = None + total = bytearray() + idempotency_keys: set[str] = set() + for event in events: + affected = sequence_start <= event.sequence <= sequence_end + payload = _redacted_payload(event) if affected else dict(event.payload) + idempotency_key = ( + _redaction_idempotency_key(operation_id, event.sequence) if affected else event.idempotency_key + ) + if idempotency_key in idempotency_keys: + raise RedactionError("rewritten journal idempotency key collision") + idempotency_keys.add(idempotency_key) + try: + envelope = run_events.build_event( + run_id=event.run_id, + sequence=event.sequence, + event_type=event.event_type, + payload=payload, + idempotency_key=idempotency_key, + recorded_at=event.recorded_at, + previous_digest=previous_digest, + ) + line = run_events.canonical_bytes(envelope) + b"\n" + except (run_events.CanonicalizationError, ValueError) as exc: + raise RedactionError("rewritten journal canonicalization failed") from exc + if len(line) > run_events.MAX_LINE_BYTES: + raise RedactionError("rewritten journal line exceeds size bound") + total.extend(line) + if len(total) > run_checkpoint.MAX_JOURNAL_BYTES: + raise RedactionError("rewritten journal exceeds size bound") + try: + rewritten_event = run_journal.RunEvent( + schema=envelope["schema"], + schema_version=envelope["schema_version"], + event_id=envelope["event_id"], + run_id=envelope["run_id"], + sequence=envelope["sequence"], + event_type=envelope["event_type"], + recorded_at=envelope["recorded_at"], + idempotency_key=envelope["idempotency_key"], + request_digest=envelope["request_digest"], + previous_digest=envelope["previous_digest"], + event_digest=envelope["event_digest"], + payload=dict(envelope["payload"]), + ) + except (KeyError, TypeError) as exc: + raise RedactionError("rewritten journal envelope construction failed") from exc + rewritten.append(rewritten_event) + previous_digest = rewritten_event.event_digest + return rewritten, bytes(total) + + +def _write_all(fd: int, data: bytes, *, category: str) -> None: + position = 0 + while position < len(data): + try: + written = os.write(fd, data[position:]) + except OSError as exc: + raise RedactionError(f"{category} write failed") from exc + if written <= 0: + raise RedactionError(f"{category} write failed") + position += written + + +def _fsync_file(fd: int, *, category: str) -> None: + try: + os.fsync(fd) + except OSError as exc: + raise RedactionError(f"{category} durability failed") from exc + + +def _open_directory_handle(path: Path, *, category: str) -> int: + try: + info = os.lstat(path) + except OSError as exc: + raise RedactionError(f"{category} path validation failed") from exc + if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): + raise RedactionError(f"{category} path is not a safe directory") + if getattr(info, "st_file_attributes", 0) & 0x400: + raise RedactionError(f"{category} path is a reparse point") + try: + if path.resolve(strict=True) != path.absolute(): + raise RedactionError(f"{category} path traverses a link") + except (OSError, RuntimeError) as exc: + raise RedactionError(f"{category} path validation failed") from exc + flags = os.O_RDONLY | run_journal._O_DIRECTORY + try: + fd = run_journal._open_nofollow(path, flags) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(f"{category} path validation failed") from exc + try: + opened = os.fstat(fd) + if not stat.S_ISDIR(opened.st_mode) or opened.st_dev != info.st_dev or opened.st_ino != info.st_ino: + raise RedactionError(f"{category} directory identity changed") + if stat.S_IMODE(opened.st_mode) != _DIR_MODE: + try: + run_journal._chmod_fd_or_path(fd, path, _DIR_MODE) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(f"{category} mode normalization failed") from exc + return fd + except BaseException: + try: + os.close(fd) + except BaseException: + pass + raise + + +def _assert_directory_identity(path: Path, fd: int, *, category: str) -> None: + try: + current = os.lstat(path) + opened = os.fstat(fd) + except OSError as exc: + raise RedactionError(f"{category} directory identity check failed") from exc + if ( + stat.S_ISLNK(current.st_mode) + or not stat.S_ISDIR(current.st_mode) + or current.st_dev != opened.st_dev + or current.st_ino != opened.st_ino + ): + raise RedactionError(f"{category} directory identity changed") + + +def _fsync_directory_handle(path: Path, fd: int, *, category: str) -> None: + _assert_directory_identity(path, fd, category=category) + try: + os.fsync(fd) + except OSError as exc: + raise RedactionError(f"{category} directory durability failed") from exc + + +def _mkdir_private_durable(path: Path, *, category: str) -> None: + parent = path.parent + parent_fd = _open_directory_handle(parent, category=f"{category} parent") + created = False + try: + try: + os.mkdir(path.name, mode=_DIR_MODE, dir_fd=parent_fd) + created = True + except FileExistsError: + pass + except OSError as exc: + raise RedactionError(f"{category} creation failed") from exc + _assert_directory_identity(parent, parent_fd, category=f"{category} parent") + child_fd = _open_directory_handle(path, category=category) + try: + if created: + _fsync_directory_handle(parent, parent_fd, category=f"{category} parent") + finally: + os.close(child_fd) + finally: + os.close(parent_fd) + + +def _prepare_operation_dirs(run_dir: Path, operation_id: str) -> Path: + events_dir = run_dir / "events" + events_fd = _open_directory_handle(events_dir, category="events") + os.close(events_fd) + redactions_dir = events_dir / "redactions" + _mkdir_private_durable(redactions_dir, category="redactions") + operation_dir = redactions_dir / operation_id + _mkdir_private_durable(operation_dir, category="redaction operation") + return operation_dir + + +def _existing_operation_dir(run_dir: Path, operation_id: str) -> bool: + redactions_dir = run_dir / "events" / "redactions" + if not os.path.lexists(redactions_dir): + return False + redactions_fd = _open_directory_handle(redactions_dir, category="redactions") + os.close(redactions_fd) + operation_dir = redactions_dir / operation_id + if not os.path.lexists(operation_dir): + return False + operation_fd = _open_directory_handle(operation_dir, category="redaction operation") + os.close(operation_fd) + return True + + +def _open_relative( + directory: Path, + directory_fd: int, + name: str, + flags: int, + mode: int = 0o666, + *, + category: str = "redaction operation", +) -> int: + _assert_directory_identity(directory, directory_fd, category=category) + open_flags = flags | run_journal._O_NOFOLLOW + try: + return os.open(name, open_flags, mode, dir_fd=directory_fd) + except OSError as exc: + if exc.errno in {getattr(os, "ELOOP", 40), getattr(os, "ENOTDIR", 20)}: + raise RedactionError(f"{category} path refused a link") from exc + raise + + +def _publish_no_replace( + directory: Path, + directory_fd: int, + temporary_name: str, + final_name: str, +) -> None: + _assert_directory_identity(directory, directory_fd, category="redaction operation") + os.link( + temporary_name, + final_name, + src_dir_fd=directory_fd, + dst_dir_fd=directory_fd, + follow_symlinks=False, + ) + + +def _unlink_relative(directory: Path, directory_fd: int, name: str) -> bool: + _assert_directory_identity(directory, directory_fd, category="redaction operation") + try: + os.unlink(name, dir_fd=directory_fd) + except FileNotFoundError: + return False + return True + + +def _remove_operation_temps( + operation_dir: Path, + operation_fd: int, + *, + patterns: Sequence[re.Pattern[str]], + category: str, +) -> None: + _assert_directory_identity( + operation_dir, + operation_fd, + category="redaction operation", + ) + try: + names = os.listdir(operation_fd) + except OSError as exc: + raise RedactionError(f"{category} listing failed") from exc + removed = False + for name in names: + if any(pattern.fullmatch(name) for pattern in patterns): + try: + removed = _unlink_relative(operation_dir, operation_fd, name) or removed + except OSError as exc: + raise RedactionError(f"{category} removal failed") from exc + if removed: + _fsync_directory_handle(operation_dir, operation_fd, category=category) + + +def _replace_relative( + directory: Path, + directory_fd: int, + temporary_name: str, + final_name: str, + *, + category: str, +) -> None: + _assert_directory_identity(directory, directory_fd, category=category) + os.rename( + temporary_name, + final_name, + src_dir_fd=directory_fd, + dst_dir_fd=directory_fd, + ) + + +def _verify_open_regular( + fd: int, + path: Path, + *, + category: str, + expected: bytes | None = None, + limit: int = run_checkpoint.MAX_JOURNAL_BYTES, +) -> bytes: + try: + info = os.fstat(fd) + except OSError as exc: + raise RedactionError(f"{category} stat failed") from exc + if not stat.S_ISREG(info.st_mode): + raise RedactionError(f"{category} is not a regular file") + if info.st_nlink != 1: + raise RedactionError(f"{category} link count is not one") + if stat.S_IMODE(info.st_mode) != _FILE_MODE: + try: + run_journal._chmod_fd_or_path(fd, path, _FILE_MODE) + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(f"{category} mode normalization failed") from exc + size_limit = len(expected) if expected is not None else limit + if info.st_size > size_limit or (expected is not None and info.st_size != len(expected)): + raise RedactionError(f"{category} conflicts with transaction data") + data = bytearray() + while len(data) < info.st_size: + try: + chunk = os.read(fd, min(run_journal._READ_CHUNK, info.st_size - len(data))) + except OSError as exc: + raise RedactionError(f"{category} read failed") from exc + if not chunk: + break + data.extend(chunk) + try: + extra = os.read(fd, 1) + except OSError as exc: + raise RedactionError(f"{category} read failed") from exc + if len(data) != info.st_size or extra or (expected is not None and bytes(data) != expected): + raise RedactionError(f"{category} conflicts with transaction data") + return bytes(data) + + +def _read_relative_bounded_regular_from_handle( + directory: Path, + directory_fd: int, + name: str, + *, + limit: int, + category: str, +) -> bytes: + fd = _open_relative( + directory, + directory_fd, + name, + os.O_RDONLY, + category=f"{category} parent", + ) + try: + return _verify_open_regular( + fd, + directory / name, + category=category, + limit=limit, + ) + finally: + os.close(fd) + + +def _read_relative_bounded_regular( + directory: Path, + name: str, + *, + limit: int, + category: str, +) -> bytes: + directory_fd = _open_directory_handle(directory, category=f"{category} parent") + try: + try: + return _read_relative_bounded_regular_from_handle( + directory, + directory_fd, + name, + limit=limit, + category=category, + ) + except OSError as exc: + raise RedactionError(f"{category} read failed") from exc + finally: + os.close(directory_fd) + + +def _load_json_object_relative( + directory: Path, + name: str, + *, + limit: int, + category: str, +) -> dict[str, Any]: + return _parse_json_object( + _read_relative_bounded_regular( + directory, + name, + limit=limit, + category=category, + ), + category=category, + ) + + +def _publish_quarantine(operation_dir: Path, quarantine_path: Path, data: bytes) -> None: + directory_fd = _open_directory_handle(operation_dir, category="redaction operation") + temporary_name = f".original.jsonl.{uuid4().hex}.tmp" + temporary_fd: int | None = None + primary: RedactionError | None = None + try: + _remove_operation_temps( + operation_dir, + directory_fd, + patterns=(_QUARANTINE_TEMP_RE,), + category="redaction quarantine cleanup", + ) + try: + existing_fd = _open_relative( + operation_dir, + directory_fd, + quarantine_path.name, + os.O_RDONLY, + ) + except FileNotFoundError: + existing_fd = None + except OSError as exc: + raise RedactionError("redaction quarantine open failed") from exc + if existing_fd is not None: + try: + _verify_open_regular( + existing_fd, + quarantine_path, + category="redaction quarantine", + expected=data, + ) + _fsync_file(existing_fd, category="redaction quarantine") + finally: + os.close(existing_fd) + _fsync_directory_handle( + operation_dir, + directory_fd, + category="redaction quarantine", + ) + return + + try: + temporary_fd = _open_relative( + operation_dir, + directory_fd, + temporary_name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + _FILE_MODE, + ) + run_journal._chmod_fd_or_path( + temporary_fd, + operation_dir / temporary_name, + _FILE_MODE, + ) + _write_all(temporary_fd, data, category="redaction quarantine") + _fsync_file(temporary_fd, category="redaction quarantine") + os.close(temporary_fd) + temporary_fd = None + _publish_no_replace( + operation_dir, + directory_fd, + temporary_name, + quarantine_path.name, + ) + _fsync_directory_handle( + operation_dir, + directory_fd, + category="redaction quarantine", + ) + except RedactionError: + raise + except OSError as exc: + raise RedactionError("redaction quarantine durability failed") from exc + except RedactionError as exc: + primary = exc + raise + finally: + if temporary_fd is not None: + try: + os.close(temporary_fd) + except OSError: + pass + try: + unlinked = _unlink_relative(operation_dir, directory_fd, temporary_name) + if unlinked: + _fsync_directory_handle( + operation_dir, + directory_fd, + category="redaction quarantine cleanup", + ) + except (OSError, RedactionError): + if primary is None: + raise RedactionError("redaction quarantine cleanup failed") from None + finally: + os.close(directory_fd) + + +def _atomic_write( + path: Path, + data: bytes, + *, + mode: int, + category: str, + parent_fd: int | None = None, +) -> None: + parent = path.parent + owns_parent_fd = parent_fd is None + if parent_fd is None: + parent_fd = _open_directory_handle(parent, category=f"{category} parent") + temporary_name = f".{path.name}.{uuid4().hex}.tmp" + temporary_fd: int | None = None + primary: RedactionError | None = None + try: + try: + existing_fd = _open_relative( + parent, + parent_fd, + path.name, + os.O_RDONLY, + category=f"{category} parent", + ) + except FileNotFoundError: + existing_fd = None + except OSError as exc: + raise RedactionError(f"{category} path validation failed") from exc + if existing_fd is not None: + try: + _verify_open_regular(existing_fd, path, category=category) + finally: + os.close(existing_fd) + try: + temporary_fd = _open_relative( + parent, + parent_fd, + temporary_name, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + mode, + category=f"{category} parent", + ) + run_journal._chmod_fd_or_path( + temporary_fd, + parent / temporary_name, + mode, + ) + _write_all(temporary_fd, data, category=category) + _fsync_file(temporary_fd, category=category) + os.close(temporary_fd) + temporary_fd = None + _replace_relative( + parent, + parent_fd, + temporary_name, + path.name, + category=f"{category} parent", + ) + _fsync_directory_handle( + parent, + parent_fd, + category=category, + ) + except RedactionError: + raise + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError(f"{category} replace failed") from exc + except RedactionError as exc: + primary = exc + raise + finally: + if temporary_fd is not None: + try: + os.close(temporary_fd) + except OSError: + pass + try: + unlinked = _unlink_relative(parent, parent_fd, temporary_name) + if unlinked: + _fsync_directory_handle( + parent, + parent_fd, + category=f"{category} cleanup", + ) + except (OSError, RedactionError): + if primary is None: + raise RedactionError(f"{category} cleanup failed") from None + finally: + if owns_parent_fd: + os.close(parent_fd) + + +def _state_bytes( + *, + operation_id: str, + run_id: str, + sequence_start: int, + sequence_end: int, + reason: str, + original_sha256: str | None, + rewritten_sha256: str | None, + phase: str, + parent_operation_id: str | None, + rewritten_digest_retired_by: str | None = None, +) -> bytes: + if (rewritten_sha256 is None) == (rewritten_digest_retired_by is None): + raise RedactionError("redaction transaction digest is invalid") + payload = { + "schema": REDACTION_SCHEMA, + "schema_version": REDACTION_SCHEMA_VERSION, + "operation_id": operation_id, + "run_id": run_id, + "sequence_start": sequence_start, + "sequence_end": sequence_end, + "reason_code": reason, + "parent_operation_id": parent_operation_id, + "phase": phase, + } + if rewritten_sha256 is not None: + payload["rewritten_sha256"] = rewritten_sha256 + if rewritten_digest_retired_by is not None: + payload["rewritten_digest_retired_by"] = rewritten_digest_retired_by + if original_sha256 is not None: + payload["original_sha256"] = original_sha256 + return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8") + + +def _load_and_validate_state( + state_path: Path, + *, + operation_id: str, + run_id: str, + sequence_start: int | None = None, + sequence_end: int | None = None, + reason: str | None = None, +) -> dict[str, Any]: + state = _load_json_object_relative( + state_path.parent, + state_path.name, + limit=16 * 1024, + category="redaction transaction", + ) + if ( + state.get("schema") != REDACTION_SCHEMA + or state.get("schema_version") != REDACTION_SCHEMA_VERSION + or state.get("operation_id") != operation_id + or state.get("run_id") != run_id + ): + raise RedactionError("redaction transaction metadata mismatch") + if sequence_start is not None and state.get("sequence_start") != sequence_start: + raise RedactionError("redaction transaction sequence mismatch") + if sequence_end is not None and state.get("sequence_end") != sequence_end: + raise RedactionError("redaction transaction sequence mismatch") + if reason is not None and state.get("reason_code") != reason: + raise RedactionError("redaction transaction reason mismatch") + if state.get("phase") not in {"quarantined", "replaced", "verified", "cleanup-authorized", "cleaned"}: + raise RedactionError("redaction transaction phase is invalid") + rewritten_digest = state.get("rewritten_sha256") + rewritten_digest_retired_by = state.get("rewritten_digest_retired_by") + valid_rewritten_digest = isinstance(rewritten_digest, str) and bool(re.fullmatch(r"[0-9a-f]{64}", rewritten_digest)) + valid_retirement = ( + isinstance(rewritten_digest_retired_by, str) + and bool(_OPERATION_RE.fullmatch(rewritten_digest_retired_by)) + and rewritten_digest_retired_by != operation_id + ) + if valid_rewritten_digest == valid_retirement: + raise RedactionError("redaction transaction digest is invalid") + original_digest = state.get("original_sha256") + if state.get("phase") == "cleaned": + if original_digest is not None: + raise RedactionError("cleaned redaction transaction retains original digest") + elif not isinstance(original_digest, str) or not re.fullmatch(r"[0-9a-f]{64}", original_digest): + raise RedactionError("redaction transaction digest is invalid") + parent_operation_id = state.get("parent_operation_id") + if parent_operation_id is not None and ( + not isinstance(parent_operation_id, str) or not _OPERATION_RE.fullmatch(parent_operation_id) + ): + raise RedactionError("redaction transaction parent is invalid") + return state + + +def _write_state( + state_path: Path, + *, + operation_id: str, + run_id: str, + sequence_start: int, + sequence_end: int, + reason: str, + original_sha256: str | None, + rewritten_sha256: str | None, + phase: str, + parent_operation_id: str | None = None, + rewritten_digest_retired_by: str | None = None, +) -> None: + _atomic_write( + state_path, + _state_bytes( + operation_id=operation_id, + run_id=run_id, + sequence_start=sequence_start, + sequence_end=sequence_end, + reason=reason, + original_sha256=original_sha256, + rewritten_sha256=rewritten_sha256, + phase=phase, + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ), + mode=_FILE_MODE, + category="redaction transaction", + ) + + +def _record_bytes( + *, + operation_id: str, + run_id: str, + sequence_start: int, + sequence_end: int, + reason: str, + rewritten_sha256: str | None, + quarantine_retained: bool, + parent_operation_id: str | None, + rewritten_digest_retired_by: str | None = None, +) -> bytes: + if (rewritten_sha256 is None) == (rewritten_digest_retired_by is None): + raise RedactionError("redaction record digest is invalid") + payload = { + "schema": REDACTION_SCHEMA, + "schema_version": REDACTION_SCHEMA_VERSION, + "operation_id": operation_id, + "run_id": run_id, + "sequence_start": sequence_start, + "sequence_end": sequence_end, + "reason_code": reason, + "parent_operation_id": parent_operation_id, + "chain_verified": True, + "projection_verified": True, + "quarantine_retained": quarantine_retained, + } + if rewritten_sha256 is not None: + payload["rewritten_journal_sha256"] = rewritten_sha256 + if rewritten_digest_retired_by is not None: + payload["rewritten_digest_retired_by"] = rewritten_digest_retired_by + return (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8") + + +def _write_redaction_record( + record_path: Path, + *, + operation_id: str, + run_id: str, + sequence_start: int, + sequence_end: int, + reason: str, + rewritten_sha256: str | None, + quarantine_retained: bool, + parent_operation_id: str | None = None, + rewritten_digest_retired_by: str | None = None, +) -> None: + _atomic_write( + record_path, + _record_bytes( + operation_id=operation_id, + run_id=run_id, + sequence_start=sequence_start, + sequence_end=sequence_end, + reason=reason, + rewritten_sha256=rewritten_sha256, + quarantine_retained=quarantine_retained, + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ), + mode=_FILE_MODE, + category="redaction record", + ) + + +def _validate_redaction_record( + record_path: Path, + *, + operation_id: str, + run_id: str, + sequence_start: int, + sequence_end: int, + reason: str, + rewritten_sha256: str | None, + parent_operation_id: str | None = None, + rewritten_digest_retired_by: str | None = None, +) -> dict[str, Any]: + record = _load_json_object_relative( + record_path.parent, + record_path.name, + limit=16 * 1024, + category="redaction record", + ) + if ( + record.get("schema") != REDACTION_SCHEMA + or record.get("schema_version") != REDACTION_SCHEMA_VERSION + or record.get("operation_id") != operation_id + or record.get("run_id") != run_id + or record.get("sequence_start") != sequence_start + or record.get("sequence_end") != sequence_end + or record.get("reason_code") != reason + or record.get("parent_operation_id") != parent_operation_id + or record.get("chain_verified") is not True + or record.get("projection_verified") is not True + or not isinstance(record.get("quarantine_retained"), bool) + ): + raise RedactionError("redaction record verification failed") + if rewritten_sha256 is not None: + if ( + not re.fullmatch(r"[0-9a-f]{64}", rewritten_sha256) + or record.get("rewritten_journal_sha256") != rewritten_sha256 + or record.get("rewritten_digest_retired_by") is not None + ): + raise RedactionError("redaction record verification failed") + elif ( + rewritten_digest_retired_by is None + or not _OPERATION_RE.fullmatch(rewritten_digest_retired_by) + or rewritten_digest_retired_by == operation_id + or record.get("rewritten_journal_sha256") is not None + or record.get("rewritten_digest_retired_by") != rewritten_digest_retired_by + ): + raise RedactionError("redaction record verification failed") + return record + + +def _parse_lineage_record(raw_record: bytes, *, operation_id: str, run_id: str) -> dict[str, Any]: + record = _parse_json_object(raw_record, category="redaction record") + rewritten = record.get("rewritten_journal_sha256") + rewritten_digest_retired_by = record.get("rewritten_digest_retired_by") + valid_rewritten = isinstance(rewritten, str) and bool(re.fullmatch(r"[0-9a-f]{64}", rewritten)) + valid_retirement = ( + isinstance(rewritten_digest_retired_by, str) + and bool(_OPERATION_RE.fullmatch(rewritten_digest_retired_by)) + and rewritten_digest_retired_by != operation_id + ) + parent = record.get("parent_operation_id") + if ( + record.get("schema") != REDACTION_SCHEMA + or record.get("schema_version") != REDACTION_SCHEMA_VERSION + or record.get("operation_id") != operation_id + or record.get("run_id") != run_id + or record.get("reason_code") not in REASON_CODES + or isinstance(record.get("sequence_start"), bool) + or not isinstance(record.get("sequence_start"), int) + or isinstance(record.get("sequence_end"), bool) + or not isinstance(record.get("sequence_end"), int) + or valid_rewritten == valid_retirement + or (parent is not None and (not isinstance(parent, str) or not _OPERATION_RE.fullmatch(parent))) + or record.get("chain_verified") is not True + or record.get("projection_verified") is not True + or not isinstance(record.get("quarantine_retained"), bool) + ): + raise RedactionError("redaction lineage record verification failed") + return record + + +def _validate_lineage_graph(records: Mapping[str, Mapping[str, Any]]) -> None: + if not records: + return + children: dict[str, list[str]] = {} + for operation_id, record in records.items(): + parent = record.get("parent_operation_id") + if isinstance(parent, str): + if parent not in records: + raise RedactionError("redaction lineage is incomplete") + children.setdefault(parent, []).append(operation_id) + retired_by = record.get("rewritten_digest_retired_by") + if isinstance(retired_by, str): + child = records.get(retired_by) + if child is None or child.get("parent_operation_id") != operation_id: + raise RedactionError("redaction lineage retirement mismatch") + + if any(len(descendants) != 1 for descendants in children.values()): + raise RedactionError("redaction lineage contains a fork") + for parent, descendants in children.items(): + retired_by = records[parent].get("rewritten_digest_retired_by") + if retired_by is not None and retired_by != descendants[0]: + raise RedactionError("redaction lineage retirement mismatch") + + tips = set(records) - set(children) + if len(tips) != 1: + raise RedactionError("redaction lineage contains multiple tips") + for operation_id in records: + visited: set[str] = set() + current: str | None = operation_id + while current is not None: + if current in visited: + raise RedactionError("redaction lineage contains a cycle") + visited.add(current) + parent = records[current].get("parent_operation_id") + current = parent if isinstance(parent, str) else None + + +def _load_operation_inventory( + redactions_dir: Path, + run_id: str, + *, + resumable_operation_id: str | None, +) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]]]: + if not os.path.lexists(redactions_dir): + return {}, {} + root_fd = _open_directory_handle(redactions_dir, category="redactions") + records: dict[str, dict[str, Any]] = {} + states: dict[str, dict[str, Any]] = {} + incomplete: list[str] = [] + try: + try: + names = sorted(os.listdir(root_fd)) + except OSError as exc: + raise RedactionError("redaction transaction listing failed") from exc + for operation_id in names: + if not _OPERATION_RE.fullmatch(operation_id): + continue + operation_dir = redactions_dir / operation_id + operation_fd = _open_directory_handle(operation_dir, category="redaction operation") + try: + try: + _read_relative_bounded_regular_from_handle( + operation_dir, + operation_fd, + "state.json", + limit=16 * 1024, + category="redaction transaction", + ) + except FileNotFoundError: + try: + _read_relative_bounded_regular_from_handle( + operation_dir, + operation_fd, + "record.json", + limit=16 * 1024, + category="redaction record", + ) + except FileNotFoundError: + if operation_id != resumable_operation_id: + raise RedactionError( + "incomplete redaction transaction exists; retry its original request" + ) from None + incomplete.append(operation_id) + continue + except OSError as exc: + raise RedactionError("redaction record read failed") from exc + raise RedactionError("redaction transaction state/record mismatch") from None + except OSError as exc: + raise RedactionError("redaction transaction read failed") from exc + finally: + os.close(operation_fd) + + state = _load_and_validate_state( + operation_dir / "state.json", + operation_id=operation_id, + run_id=run_id, + ) + states[operation_id] = state + operation_fd = _open_directory_handle(operation_dir, category="redaction operation") + try: + try: + raw_record = _read_relative_bounded_regular_from_handle( + operation_dir, + operation_fd, + "record.json", + limit=16 * 1024, + category="redaction record", + ) + except FileNotFoundError: + if operation_id != resumable_operation_id or state.get("phase") not in { + "quarantined", + "replaced", + }: + raise RedactionError( + "incomplete redaction transaction exists; retry its original request" + ) from None + incomplete.append(operation_id) + continue + except OSError as exc: + raise RedactionError("redaction record read failed") from exc + finally: + os.close(operation_fd) + + record = _parse_lineage_record(raw_record, operation_id=operation_id, run_id=run_id) + if state.get("phase") in {"quarantined", "replaced", "cleanup-authorized"}: + if operation_id != resumable_operation_id: + raise RedactionError("incomplete redaction transaction exists; retry its original request") + incomplete.append(operation_id) + records[operation_id] = record + finally: + os.close(root_fd) + + if len(set(incomplete)) > 1: + raise RedactionError("multiple incomplete redaction transactions exist") + for operation_id, record in records.items(): + state = states[operation_id] + if ( + state.get("sequence_start") != record.get("sequence_start") + or state.get("sequence_end") != record.get("sequence_end") + or state.get("reason_code") != record.get("reason_code") + or state.get("parent_operation_id") != record.get("parent_operation_id") + ): + raise RedactionError("redaction transaction state/record mismatch") + state_digest = state.get("rewritten_sha256") + record_digest = record.get("rewritten_journal_sha256") + state_retired_by = state.get("rewritten_digest_retired_by") + record_retired_by = record.get("rewritten_digest_retired_by") + if state_digest != record_digest or state_retired_by != record_retired_by: + if record_retired_by is not None and state_digest is not None: + split_retired_by = record_retired_by + split_digest = state_digest + elif state_retired_by is not None and record_digest is not None: + split_retired_by = state_retired_by + split_digest = record_digest + else: + raise RedactionError("redaction transaction state/record mismatch") + child_state = states.get(split_retired_by) + child_record = records.get(split_retired_by) + if ( + split_retired_by != resumable_operation_id + or child_state is None + or child_record is None + or child_state.get("phase") != "cleanup-authorized" + or child_state.get("parent_operation_id") != operation_id + or child_record.get("parent_operation_id") != operation_id + or child_state.get("original_sha256") != split_digest + ): + raise RedactionError("redaction transaction state/record mismatch") + phase = state.get("phase") + quarantine_retained = record.get("quarantine_retained") + if (phase == "verified" and quarantine_retained is not True) or ( + phase == "cleaned" and quarantine_retained is not False + ): + raise RedactionError("redaction transaction state/record mismatch") + _validate_lineage_graph(records) + return records, states + + +def _retire_rewritten_digest_aliases( + redactions_dir: Path, + *, + run_id: str, + sensitive_digest: str, + retired_by_operation_id: str, + records: Mapping[str, Mapping[str, Any]], +) -> None: + for operation_id, record in records.items(): + state_path = redactions_dir / operation_id / "state.json" + state = _load_and_validate_state( + state_path, + operation_id=operation_id, + run_id=run_id, + sequence_start=record["sequence_start"], + sequence_end=record["sequence_end"], + reason=record["reason_code"], + ) + if state.get("parent_operation_id") != record.get("parent_operation_id"): + raise RedactionError("redaction lineage metadata mismatch") + + record_digest = record.get("rewritten_journal_sha256") + record_retired_by = record.get("rewritten_digest_retired_by") + state_digest = state.get("rewritten_sha256") + state_retired_by = state.get("rewritten_digest_retired_by") + record_matches = record_digest == sensitive_digest or record_retired_by == retired_by_operation_id + state_matches = state_digest == sensitive_digest or state_retired_by == retired_by_operation_id + if not record_matches and not state_matches: + continue + if (record_digest != sensitive_digest and record_retired_by != retired_by_operation_id) or ( + state_digest != sensitive_digest and state_retired_by != retired_by_operation_id + ): + raise RedactionError("redaction lineage digest mismatch") + operation_dir = redactions_dir / operation_id + record_path = operation_dir / "record.json" + if not (record_retired_by == retired_by_operation_id and state_retired_by == retired_by_operation_id): + _write_redaction_record( + record_path, + operation_id=operation_id, + run_id=run_id, + sequence_start=record["sequence_start"], + sequence_end=record["sequence_end"], + reason=record["reason_code"], + rewritten_sha256=None, + quarantine_retained=record["quarantine_retained"], + parent_operation_id=record.get("parent_operation_id"), + rewritten_digest_retired_by=retired_by_operation_id, + ) + _write_state( + state_path, + operation_id=operation_id, + run_id=run_id, + sequence_start=record["sequence_start"], + sequence_end=record["sequence_end"], + reason=record["reason_code"], + original_sha256=state.get("original_sha256"), + rewritten_sha256=None, + phase=state["phase"], + parent_operation_id=state.get("parent_operation_id"), + rewritten_digest_retired_by=retired_by_operation_id, + ) + operation_fd = _open_directory_handle(operation_dir, category="redaction operation") + try: + _remove_operation_temps( + operation_dir, + operation_fd, + patterns=(_STATE_TEMP_RE, _RECORD_TEMP_RE), + category="redaction lineage cleanup", + ) + finally: + os.close(operation_fd) + _validate_redaction_record( + record_path, + operation_id=operation_id, + run_id=run_id, + sequence_start=record["sequence_start"], + sequence_end=record["sequence_end"], + reason=record["reason_code"], + rewritten_sha256=None, + parent_operation_id=record.get("parent_operation_id"), + rewritten_digest_retired_by=retired_by_operation_id, + ) + + +def _lineage_parent_for_digest( + records: Mapping[str, Mapping[str, Any]], + journal_digest: str, +) -> str | None: + matches = [ + operation_id + for operation_id, record in records.items() + if record.get("rewritten_journal_sha256") == journal_digest + ] + if len(matches) > 1: + raise RedactionError("redaction lineage is ambiguous") + return matches[0] if matches else None + + +def _lineage_contains( + records: Mapping[str, Mapping[str, Any]], + *, + ancestor_operation_id: str, + active_journal_digest: str, +) -> bool: + current = _lineage_parent_for_digest(records, active_journal_digest) + visited: set[str] = set() + while current is not None: + if current == ancestor_operation_id: + return True + if current in visited: + raise RedactionError("redaction lineage contains a cycle") + visited.add(current) + record = records.get(current) + if record is None: + raise RedactionError("redaction lineage is incomplete") + parent = record.get("parent_operation_id") + current = parent if isinstance(parent, str) else None + return False + + +def _digest(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _canonical_event_bytes(events: Sequence[run_journal.RunEvent]) -> bytes: + try: + return b"".join(run_events.canonical_bytes(event.to_dict()) + b"\n" for event in events) + except run_events.CanonicalizationError as exc: + raise RedactionError("journal canonicalization verification failed") from exc + + +def _verify_quarantine(quarantine_path: Path, expected_digest: str) -> None: + original = _read_relative_bounded_regular( + quarantine_path.parent, + quarantine_path.name, + limit=run_checkpoint.MAX_JOURNAL_BYTES, + category="redaction quarantine", + ) + if _digest(original) != expected_digest: + raise RedactionError("redaction quarantine verification failed") + + +def _assert_affected_values_removed( + original: Sequence[run_journal.RunEvent], + rewritten: Sequence[run_journal.RunEvent], + start: int, + end: int, +) -> None: + for prior, current in zip(original, rewritten, strict=True): + if not start <= prior.sequence <= end or prior.event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE: + continue + for key, value in prior.payload.items(): + if key == "status" or value is None: + continue + if value == REDACTED_VALUE or value == 0: + continue + if current.payload.get(key) == value: + raise RedactionError("rewritten journal still contains affected private values") + + +def _replace_journal(journal_path: Path, rewritten: bytes) -> None: + try: + info = os.lstat(journal_path) + except OSError as exc: + raise RedactionError("journal replace preflight failed") from exc + if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode) or info.st_nlink != 1: + raise RedactionError("journal replace preflight failed") + _atomic_write( + journal_path, + rewritten, + mode=_FILE_MODE, + category="journal", + ) + + +def _replace_projection(run_dir: Path, projection: run_projector.RunProjection) -> None: + path = run_dir / "run.json" + try: + info = os.lstat(path) + except OSError as exc: + raise RedactionError("projection replace preflight failed") from exc + if not stat.S_ISREG(info.st_mode) or stat.S_ISLNK(info.st_mode) or info.st_nlink != 1: + raise RedactionError("projection replace preflight failed") + _atomic_write( + path, + projection.to_bytes(), + mode=_FILE_MODE, + category="projection", + ) + + +def _remove_shadow_history_siblings(events_dir: Path, events_fd: int) -> None: + _assert_directory_identity(events_dir, events_fd, category="events") + try: + names = os.listdir(events_fd) + except OSError as exc: + raise RedactionError("shadow projection history listing failed") from exc + artifact_name = run_shadow.SHADOW_ARTIFACT_NAME + removed = False + for name in names: + if not (name.startswith(f"{artifact_name}.corrupt-") or name.startswith(".stale-projector-")): + continue + try: + removed = _unlink_relative(events_dir, events_fd, name) or removed + except OSError as exc: + raise RedactionError("shadow projection history removal failed") from exc + if removed: + _fsync_directory_handle(events_dir, events_fd, category="shadow projection history") + + +def _refresh_shadow_artifact( + run_dir: Path, + snapshot: Mapping[str, Any], + events: Sequence[run_journal.RunEvent], +) -> dict[str, Any]: + events_dir = run_dir / "events" + events_fd = _open_directory_handle(events_dir, category="events") + artifact_path = run_shadow.shadow_artifact_path(run_dir) + try: + _remove_shadow_history_siblings(events_dir, events_fd) + # Redaction changes every digest downstream of the affected range. + # Rebase shadow evidence to the verified rewritten journal instead of + # retaining comparison hashes that can serve as candidate-value oracles. + artifact = run_shadow._empty_artifact(run_dir.name) + tail = events[-1] + projected = _projection(snapshot, events) + projected_digest = _digest(projected.to_bytes()) + artifact["comparisons"] = 1 + artifact["matches"] = 1 + artifact["last_compared_sequence"] = tail.sequence + artifact["last_compared_event_digest"] = tail.event_digest + artifact["last_shadow_digest"] = projected_digest + artifact["last_projected_digest"] = projected_digest + artifact["last_differing_fields"] = [] + artifact["last_outcome"] = run_shadow.OUTCOME_MATCH + artifact["last_error_category"] = None + recorded_at = localio.utc_now_iso() + artifact["last_recorded_at"] = recorded_at + artifact["recent_records"] = [ + { + "outcome": run_shadow.OUTCOME_MATCH, + "category": None, + "sequence": tail.sequence, + "event_digest": tail.event_digest, + "shadow_digest": projected_digest, + "projected_digest": projected_digest, + "differing_fields": [], + "recorded_at": recorded_at, + } + ] + artifact_bytes = (json.dumps(artifact, indent=2, sort_keys=True) + "\n").encode("utf-8") + _atomic_write( + artifact_path, + artifact_bytes, + mode=_FILE_MODE, + category="shadow projection", + parent_fd=events_fd, + ) + return _parse_json_object(artifact_bytes, category="shadow projection") + finally: + os.close(events_fd) + + +def _post_replace_verify( + run_dir: Path, + *, + expected_digest: str | None, +) -> tuple[dict[str, Any], list[run_journal.RunEvent], str]: + journal_path = run_dir / "events" / "lifecycle.jsonl" + active = _read_bounded_regular( + journal_path, + limit=run_checkpoint.MAX_JOURNAL_BYTES, + category="journal", + ) + active_digest = _digest(active) + if expected_digest is not None and active_digest != expected_digest: + raise RedactionError("post-rewrite verification failed") + events = _verified_events(journal_path, category="post-rewrite journal") + snapshot = _load_json_object( + run_dir / "run.json", + limit=MAX_RUN_JSON_BYTES, + category="run projection", + ) + _verify_current_projection(snapshot, events) + _validate_checkpoint_artifacts(run_dir, snapshot, events) + shadow = _refresh_shadow_artifact(run_dir, snapshot, events) + tail = events[-1] + if ( + shadow.get("schema") != run_shadow.SHADOW_SCHEMA + or shadow.get("schema_version") != run_shadow.SHADOW_SCHEMA_VERSION + or shadow.get("projector_version") != run_projector.PROJECTOR_VERSION + or shadow.get("run_id") != run_dir.name + or shadow.get("last_outcome") != run_shadow.OUTCOME_MATCH + or shadow.get("last_compared_sequence") != tail.sequence + or shadow.get("last_compared_event_digest") != tail.event_digest + or shadow.get("last_shadow_digest") != shadow.get("last_projected_digest") + or shadow.get("last_differing_fields") != [] + ): + raise RedactionError("shadow projection verification failed") + return snapshot, events, active_digest + + +def _resume_projection_after_rewrite( + run_dir: Path, + events: Sequence[run_journal.RunEvent], +) -> None: + snapshot = _load_json_object( + run_dir / "run.json", + limit=MAX_RUN_JSON_BYTES, + category="run projection", + ) + projected = _projection(snapshot, events) + if snapshot == projected.snapshot: + return + if _projection_semantics(snapshot) != _projection_semantics(projected.snapshot): + raise RedactionError("post-rewrite projection semantics changed") + _replace_projection(run_dir, projected) + + +def redact_journal( + run_dir: Path, + *, + sequence_start: int, + sequence_end: int, + reason: str, + operator_confirmed: bool = False, +) -> RedactionReport: + """Redact an inclusive sequence range and retain the source quarantine. + + The operation is deterministic for ``run_id + range + reason``. Repeating + the same call resumes an interrupted transaction or returns the already + verified result without creating another sensitive copy. + """ + start, end, bounded_reason = _validate_request( + sequence_start, + sequence_end, + reason, + operator_confirmed=operator_confirmed, + ) + _require_secure_transaction_platform() + resolved_run_dir = _resolve_run_dir(run_dir) + _probe_secure_transaction_directory(resolved_run_dir) + + with _PROCESS_LOCK: + with _exclusive_redaction_lock(resolved_run_dir) as (snapshot, workspace): + journal_path = resolved_run_dir / "events" / "lifecycle.jsonl" + events = _verified_events(journal_path, category="journal") + if end > len(events): + raise RedactionError("invalid sequence range") + if any(event.run_id != resolved_run_dir.name for event in events): + raise RedactionError("journal run identity mismatch") + + operation_id = _operation_id(resolved_run_dir.name, start, end, bounded_reason) + operation_dir, quarantine_path, record_path = _operation_paths( + resolved_run_dir, + operation_id, + ) + state_path = operation_dir / "state.json" + redactions_dir = operation_dir.parent + original_bytes = _read_bounded_regular( + journal_path, + limit=run_checkpoint.MAX_JOURNAL_BYTES, + category="journal", + ) + if original_bytes != _canonical_event_bytes(events): + raise RedactionError("journal changed during redaction preflight") + active_digest = _digest(original_bytes) + lineage_records, inventory_states = _load_operation_inventory( + redactions_dir, + resolved_run_dir.name, + resumable_operation_id=operation_id, + ) + prior_state = inventory_states.get(operation_id) + + if prior_state is not None: + if ( + prior_state.get("sequence_start") != start + or prior_state.get("sequence_end") != end + or prior_state.get("reason_code") != bounded_reason + ): + raise RedactionError("redaction transaction metadata mismatch") + prior_rewritten_sha256 = prior_state.get("rewritten_sha256") + if isinstance(prior_rewritten_sha256, str) and active_digest == prior_rewritten_sha256: + _resume_projection_after_rewrite(resolved_run_dir, events) + snapshot = _load_json_object( + resolved_run_dir / "run.json", + limit=MAX_RUN_JSON_BYTES, + category="run projection", + ) + + before_projection = _verify_current_projection(snapshot, events) + _validate_checkpoint_artifacts(resolved_run_dir, snapshot, events) + + if prior_state is not None: + parent_operation_id = prior_state.get("parent_operation_id") + rewritten_sha256 = prior_state.get("rewritten_sha256") + rewritten_digest_retired_by = prior_state.get("rewritten_digest_retired_by") + is_current_or_descendant = ( + isinstance(rewritten_sha256, str) and active_digest == rewritten_sha256 + ) or _lineage_contains( + lineage_records, + ancestor_operation_id=operation_id, + active_journal_digest=active_digest, + ) + if is_current_or_descendant: + if prior_state["phase"] == "cleanup-authorized": + raise RedactionError("redaction cleanup is incomplete; retry explicit cleanup") + cleaned = prior_state["phase"] == "cleaned" + if cleaned: + if os.path.lexists(quarantine_path): + raise RedactionError("cleanup state conflicts with retained quarantine") + else: + original_sha256 = prior_state.get("original_sha256") + if not isinstance(original_sha256, str): + raise RedactionError("redaction transaction digest is invalid") + _verify_quarantine(quarantine_path, original_sha256) + if isinstance(rewritten_sha256, str) and active_digest == rewritten_sha256: + _resume_projection_after_rewrite(resolved_run_dir, events) + _post_replace_verify(resolved_run_dir, expected_digest=None) + if not os.path.lexists(record_path): + if not isinstance(rewritten_sha256, str) or active_digest != rewritten_sha256 or cleaned: + raise RedactionError("redaction lineage record is missing") + try: + _write_redaction_record( + record_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + rewritten_sha256=rewritten_sha256, + quarantine_retained=True, + parent_operation_id=parent_operation_id, + ) + except (OSError, RedactionError) as exc: + raise RedactionError("redaction record write failed") from exc + _write_state( + state_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + original_sha256=prior_state.get("original_sha256"), + rewritten_sha256=rewritten_sha256, + phase="verified", + parent_operation_id=parent_operation_id, + ) + _validate_redaction_record( + record_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + rewritten_sha256=rewritten_sha256, + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ) + return RedactionReport( + operation_id, + start, + end, + quarantine_path, + record_path, + cleaned=cleaned, + ) + if active_digest != prior_state.get("original_sha256") or prior_state["phase"] != "quarantined": + raise RedactionError("redaction transaction does not match the active journal") + + parent_operation_id = ( + prior_state.get("parent_operation_id") + if prior_state is not None + else _lineage_parent_for_digest(lineage_records, active_digest) + ) + rewritten_events, rewritten_bytes = _rewrite_events( + events, + sequence_start=start, + sequence_end=end, + operation_id=operation_id, + ) + _assert_affected_values_removed(events, rewritten_events, start, end) + after_projection = _projection(snapshot, rewritten_events) + if _projection_semantics(before_projection.snapshot) != _projection_semantics(after_projection.snapshot): + raise RedactionError("redaction changes the observable run projection") + original_sha256 = active_digest + rewritten_sha256 = _digest(rewritten_bytes) + if prior_state is not None and ( + prior_state.get("original_sha256") != original_sha256 + or prior_state.get("rewritten_sha256") != rewritten_sha256 + or prior_state.get("rewritten_digest_retired_by") is not None + ): + raise RedactionError("redaction transaction digest mismatch") + + _prepare_operation_dirs(resolved_run_dir, operation_id) + try: + _publish_quarantine(operation_dir, quarantine_path, original_bytes) + _write_state( + state_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + original_sha256=original_sha256, + rewritten_sha256=rewritten_sha256, + phase="quarantined", + parent_operation_id=parent_operation_id, + ) + except RedactionError: + raise + except (OSError, run_journal.RunJournalError) as exc: + raise RedactionError("redaction quarantine durability failed") from exc + + _assert_active_owner(workspace, resolved_run_dir) + _replace_journal(journal_path, rewritten_bytes) + _write_state( + state_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + original_sha256=original_sha256, + rewritten_sha256=rewritten_sha256, + phase="replaced", + parent_operation_id=parent_operation_id, + ) + _replace_projection(resolved_run_dir, after_projection) + _post_replace_verify(resolved_run_dir, expected_digest=rewritten_sha256) + try: + _write_redaction_record( + record_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + rewritten_sha256=rewritten_sha256, + quarantine_retained=True, + parent_operation_id=parent_operation_id, + ) + except (OSError, RedactionError) as exc: + raise RedactionError("redaction record write failed") from exc + _write_state( + state_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + original_sha256=original_sha256, + rewritten_sha256=rewritten_sha256, + phase="verified", + parent_operation_id=parent_operation_id, + ) + _validate_redaction_record( + record_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=bounded_reason, + rewritten_sha256=rewritten_sha256, + parent_operation_id=parent_operation_id, + ) + return RedactionReport(operation_id, start, end, quarantine_path, record_path) + + +def cleanup_redaction_quarantine( + run_dir: Path, + *, + operation_id: str, + operator_confirmed: bool = False, +) -> RedactionReport: + """Remove one verified quarantine after an explicit incident-procedure step.""" + if operator_confirmed is not True: + raise RedactionError("operator confirmation is required") + if not isinstance(operation_id, str) or not _OPERATION_RE.fullmatch(operation_id): + raise RedactionError("invalid redaction operation id") + _require_secure_transaction_platform() + resolved_run_dir = _resolve_run_dir(run_dir) + _probe_secure_transaction_directory(resolved_run_dir) + + with _PROCESS_LOCK: + with _exclusive_redaction_lock(resolved_run_dir) as (snapshot, workspace): + operation_dir, quarantine_path, record_path = _operation_paths( + resolved_run_dir, + operation_id, + ) + if not _existing_operation_dir(resolved_run_dir, operation_id): + raise RedactionError("redaction transaction does not exist") + state_path = operation_dir / "state.json" + lineage_records, inventory_states = _load_operation_inventory( + operation_dir.parent, + resolved_run_dir.name, + resumable_operation_id=operation_id, + ) + state = inventory_states.get(operation_id) + if state is None: + raise RedactionError("redaction transaction metadata mismatch") + start = state.get("sequence_start") + end = state.get("sequence_end") + reason = state.get("reason_code") + parent_operation_id = state.get("parent_operation_id") + rewritten_sha256 = state.get("rewritten_sha256") + rewritten_digest_retired_by = state.get("rewritten_digest_retired_by") + if ( + isinstance(start, bool) + or not isinstance(start, int) + or isinstance(end, bool) + or not isinstance(end, int) + or not isinstance(reason, str) + or reason not in REASON_CODES + ): + raise RedactionError("redaction transaction metadata mismatch") + + journal_path = resolved_run_dir / "events" / "lifecycle.jsonl" + events = _verified_events(journal_path, category="journal") + _verify_current_projection(snapshot, events) + _validate_checkpoint_artifacts(resolved_run_dir, snapshot, events) + try: + _, _, active_digest = _post_replace_verify( + resolved_run_dir, + expected_digest=None, + ) + except RedactionError as exc: + raise RedactionError("cleanup verification failed") from exc + if not _lineage_contains( + lineage_records, + ancestor_operation_id=operation_id, + active_journal_digest=active_digest, + ): + raise RedactionError("cleanup verification failed") + try: + _validate_redaction_record( + record_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=reason, + rewritten_sha256=rewritten_sha256, + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ) + except RedactionError as exc: + raise RedactionError("cleanup verification failed") from exc + + if state["phase"] == "cleaned": + if os.path.lexists(quarantine_path): + raise RedactionError("cleanup state conflicts with retained quarantine") + return RedactionReport( + operation_id, + start, + end, + quarantine_path, + record_path, + cleaned=True, + ) + + original_sha256 = state.get("original_sha256") + if not isinstance(original_sha256, str): + raise RedactionError("cleanup verification failed") + if os.path.lexists(quarantine_path): + try: + _verify_quarantine(quarantine_path, original_sha256) + except RedactionError as exc: + raise RedactionError("cleanup verification failed") from exc + elif state["phase"] != "cleanup-authorized": + raise RedactionError("cleanup verification failed") + + # Authorization is durable before deletion. A crash in this + # window leaves enough state to revalidate and complete cleanup. + _write_redaction_record( + record_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=reason, + rewritten_sha256=rewritten_sha256, + quarantine_retained=True, + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ) + _write_state( + state_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=reason, + original_sha256=original_sha256, + rewritten_sha256=rewritten_sha256, + phase="cleanup-authorized", + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ) + + operation_fd = _open_directory_handle( + operation_dir, + category="redaction operation", + ) + try: + _assert_active_owner(workspace, resolved_run_dir) + try: + removed_quarantine = _unlink_relative( + operation_dir, + operation_fd, + quarantine_path.name, + ) + if removed_quarantine: + _fsync_directory_handle( + operation_dir, + operation_fd, + category="quarantine cleanup", + ) + _remove_operation_temps( + operation_dir, + operation_fd, + patterns=(_QUARANTINE_TEMP_RE, _STATE_TEMP_RE), + category="redaction cleanup", + ) + except (OSError, RedactionError) as exc: + raise RedactionError("quarantine cleanup failed") from exc + finally: + os.close(operation_fd) + + _retire_rewritten_digest_aliases( + operation_dir.parent, + run_id=resolved_run_dir.name, + sensitive_digest=original_sha256, + retired_by_operation_id=operation_id, + records=lineage_records, + ) + _post_replace_verify( + resolved_run_dir, + expected_digest=active_digest, + ) + _write_redaction_record( + record_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=reason, + rewritten_sha256=rewritten_sha256, + quarantine_retained=False, + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ) + _write_state( + state_path, + operation_id=operation_id, + run_id=resolved_run_dir.name, + sequence_start=start, + sequence_end=end, + reason=reason, + original_sha256=None, + rewritten_sha256=rewritten_sha256, + phase="cleaned", + parent_operation_id=parent_operation_id, + rewritten_digest_retired_by=rewritten_digest_retired_by, + ) + return RedactionReport( + operation_id, + start, + end, + quarantine_path, + record_path, + cleaned=True, + ) + + +__all__ = [ + "REDACTED_VALUE", + "REASON_CODES", + "RedactionError", + "RedactionReport", + "cleanup_redaction_quarantine", + "redact_journal", +] diff --git a/src/brigade/runs_cmd.py b/src/brigade/runs_cmd.py index c4dea122..1e5b4685 100644 --- a/src/brigade/runs_cmd.py +++ b/src/brigade/runs_cmd.py @@ -1741,6 +1741,60 @@ def recover(run: str | Path, *, cwd: Path, runs_dir: Path | None = None) -> int: return _recover_legacy(run_dir, workspace, parseable, run_meta, read_error) +def redact( + run: str | Path, + *, + cwd: Path, + runs_dir: Path | None = None, + sequence_start: int | None = None, + sequence_end: int | None = None, + reason: str | None = None, + operator_confirmed: bool = False, + cleanup_operation: str | None = None, +) -> int: + """Run the explicit operator procedure for lifecycle journal redaction.""" + run_dir, error = _resolve_run_dir(run, cwd=cwd, runs_dir=runs_dir) + if error is not None: + print(error, file=sys.stderr) + return 2 + assert run_dir is not None + + from . import run_redaction + + try: + if cleanup_operation is not None: + if sequence_start is not None or sequence_end is not None or reason is not None: + raise run_redaction.RedactionError("cleanup cannot include a sequence range or reason code") + report = run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=cleanup_operation, + operator_confirmed=operator_confirmed, + ) + print(f"redaction cleanup: {report.operation_id}") + print("quarantine: removed") + return 0 + if sequence_start is None or sequence_end is None or reason is None: + raise run_redaction.RedactionError("redaction requires a sequence range and reason code") + report = run_redaction.redact_journal( + run_dir, + sequence_start=sequence_start, + sequence_end=sequence_end, + reason=reason, + operator_confirmed=operator_confirmed, + ) + except run_redaction.RedactionError as exc: + print(f"error: redaction failed: {exc.diagnostic}", file=sys.stderr) + return 2 + print(f"redaction: {report.operation_id}") + print(f"sequences: {report.sequence_start}-{report.sequence_end}") + print(f"record: {report.record_path}") + if report.cleaned: + print("quarantine: removed") + else: + print(f"quarantine: retained at {report.quarantine_path}") + return 0 + + def watch( run: str | Path, *, diff --git a/tests/test_run_redaction.py b/tests/test_run_redaction.py new file mode 100644 index 00000000..c18a5805 --- /dev/null +++ b/tests/test_run_redaction.py @@ -0,0 +1,1855 @@ +"""Tests for the operator-only lifecycle journal redaction procedure.""" + +from __future__ import annotations + +import json +import os +import subprocess +import stat +import sys +import textwrap +import time +from pathlib import Path + +import pytest + +from brigade import cli, run_checkpoint, run_events, run_journal, run_projector, run_redaction, run_shadow + +RUN_ID = "20260730-190000-redact" +SECRET = "secret-value-that-must-not-survive" +REASON_CODE = "credential-exposure" + + +def _journal_path(run_dir: Path) -> Path: + return run_dir / "events" / "lifecycle.jsonl" + + +def _append( + journal: Path, + *, + event_type: str, + payload: dict, + key: str, + prior: int, + second: int, +) -> run_journal.RunEvent: + return run_journal.append_event( + journal, + run_id=RUN_ID, + event_type=event_type, + payload=payload, + idempotency_key=key, + expected_previous_sequence=prior, + recorded_at=f"2026-07-30T19:00:{second:02d}.000000Z", + ) + + +def _authority_run( + tmp_path: Path, + *, + first_idempotency_key: str = "created", +) -> tuple[Path, dict, list[run_journal.RunEvent]]: + run_dir = tmp_path / "workspace" / ".brigade" / "runs" / RUN_ID + journal = _journal_path(run_dir) + base = { + "schema": "brigade.run.v1", + "schema_version": 1, + "task": "redaction fixture", + "status": "ok", + "cwd": str(tmp_path / "workspace"), + "lock_workspace": str(tmp_path / "workspace"), + "lifecycle_journal_requested": True, + "run_journal_authority_requested": True, + } + checkpoint_bytes = run_projector.encode_snapshot_bytes(base) + run_checkpoint.publish_checkpoint_file(run_dir, checkpoint_bytes) + events = [ + _append( + journal, + event_type="run.created", + payload={"status": "started"}, + key=first_idempotency_key, + prior=0, + second=0, + ), + _append( + journal, + event_type="run.planning.started", + payload={"detail": SECRET}, + key="planning", + prior=1, + second=1, + ), + _append( + journal, + event_type=run_checkpoint.CHECKPOINT_EVENT_TYPE, + payload=run_checkpoint._checkpoint_payload( + checkpoint_bytes, + paired_event_type="run.completed", + body_kind="base-stripped", + ), + key=run_checkpoint._checkpoint_idempotency_key( + run_checkpoint._checkpoint_payload( + checkpoint_bytes, + paired_event_type="run.completed", + body_kind="base-stripped", + )["sha256"], + paired_event_type="run.completed", + body_kind="base-stripped", + ), + prior=2, + second=2, + ), + _append( + journal, + event_type="run.completed", + payload={"status": "ok", "detail": "complete"}, + key="completed", + prior=3, + second=3, + ), + ] + projection = run_projector.project_run_snapshot(base, events, journal_present=True) + (run_dir / "run.json").write_bytes(projection.to_bytes()) + return run_dir, projection.snapshot, events + + +def _without_tail_digest(snapshot: dict) -> dict: + return {key: value for key, value in snapshot.items() if key != "journal_last_event_digest"} + + +def _latest_checkpoint_path(run_dir: Path) -> Path: + report = run_journal.read_journal_bounded(_journal_path(run_dir)) + event = run_checkpoint.latest_checkpoint_event(report.events) + assert event is not None + return run_checkpoint.checkpoint_path(run_dir, event.payload["sha256"]) + + +def _append_uncovered_terminal_event(run_dir: Path) -> None: + journal = _journal_path(run_dir) + report = run_journal.read_journal_bounded(journal) + _append( + journal, + event_type="run.completed", + payload={"status": "ok", "detail": "late terminal event"}, + key="late-completed", + prior=report.events[-1].sequence, + second=4, + ) + events = run_journal.read_journal_bounded(journal).events + snapshot = json.loads((run_dir / "run.json").read_text()) + projection = run_projector.project_run_snapshot(snapshot, events, journal_present=True) + (run_dir / "run.json").write_bytes(projection.to_bytes()) + + +def _artifact_snapshot(run_dir: Path) -> dict[str, bytes]: + return {str(path.relative_to(run_dir)): path.read_bytes() for path in sorted(run_dir.rglob("*")) if path.is_file()} + + +@pytest.mark.parametrize("platform_name", ["nt", "java"]) +def test_redaction_refuses_unsupported_platform_before_lock_or_mutation( + tmp_path, + monkeypatch, + platform_name, +): + run_dir, _, _ = _authority_run(tmp_path) + before = _artifact_snapshot(run_dir) + external = tmp_path / "external" + external.mkdir() + original_events = run_dir / "events" + parked_events = run_dir / "events-parked" + monkeypatch.setattr(run_redaction, "_PLATFORM_NAME", platform_name, raising=False) + + def raceable_parent_replacement(*args, **kwargs): + original_events.rename(parked_events) + original_events.symlink_to(external, target_is_directory=True) + (external / "escaped").write_text("unsafe fallback reached") + raise AssertionError("platform gate ran after the write-capable lock path") + + monkeypatch.setattr( + run_redaction, + "_exclusive_redaction_lock", + raceable_parent_replacement, + ) + + with pytest.raises(run_redaction.RedactionError, match="unsupported platform"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _artifact_snapshot(run_dir) == before + assert original_events.is_dir() + assert not original_events.is_symlink() + assert list(external.iterdir()) == [] + assert not (tmp_path / "workspace" / ".brigade" / "run.lock").exists() + + +@pytest.mark.parametrize("missing_operation", [os.open, os.mkdir, os.rename, os.unlink, os.link]) +def test_redaction_refuses_missing_dirfd_operation_before_any_write( + tmp_path, + monkeypatch, + missing_operation, +): + run_dir, _, _ = _authority_run(tmp_path) + before = _artifact_snapshot(run_dir) + supported = set(os.supports_dir_fd) + supported.discard(missing_operation) + monkeypatch.setattr(run_redaction.os, "supports_dir_fd", supported) + monkeypatch.setattr( + run_redaction, + "_exclusive_redaction_lock", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("unsupported capability reached lock acquisition") + ), + ) + + with pytest.raises(run_redaction.RedactionError, match="unsupported platform"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _artifact_snapshot(run_dir) == before + assert not (tmp_path / "workspace" / ".brigade" / "run.lock").exists() + + +def test_redaction_refuses_missing_fd_listdir_before_any_write(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + before = _artifact_snapshot(run_dir) + supported = set(os.supports_fd) + supported.discard(os.listdir) + monkeypatch.setattr(run_redaction.os, "supports_fd", supported) + monkeypatch.setattr( + run_redaction, + "_exclusive_redaction_lock", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("unsupported capability reached lock acquisition") + ), + ) + + with pytest.raises(run_redaction.RedactionError, match="unsupported platform"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _artifact_snapshot(run_dir) == before + + +@pytest.mark.parametrize( + "missing_capability", + ["directory-fsync", "nofollow", "directory-open", "fchmod", "link-nofollow"], +) +def test_redaction_refuses_missing_durability_or_containment_before_any_write( + tmp_path, + monkeypatch, + missing_capability, +): + run_dir, _, _ = _authority_run(tmp_path) + before = _artifact_snapshot(run_dir) + if missing_capability == "directory-fsync": + monkeypatch.setattr(run_redaction.os, "fsync", None) + elif missing_capability == "nofollow": + monkeypatch.setattr(run_journal, "_O_NOFOLLOW", 0) + elif missing_capability == "directory-open": + monkeypatch.setattr(run_journal, "_O_DIRECTORY", 0) + elif missing_capability == "fchmod": + monkeypatch.setattr(run_journal, "_HAS_FCHMOD", False) + else: + supported = set(os.supports_follow_symlinks) + supported.discard(os.link) + monkeypatch.setattr(run_redaction.os, "supports_follow_symlinks", supported) + monkeypatch.setattr( + run_redaction, + "_exclusive_redaction_lock", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("unsupported durability or containment reached lock acquisition") + ), + ) + + with pytest.raises(run_redaction.RedactionError, match="unsupported platform"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _artifact_snapshot(run_dir) == before + + +def test_redaction_refuses_directory_fsync_probe_before_lock_or_mutation( + tmp_path, + monkeypatch, +): + run_dir, _, _ = _authority_run(tmp_path) + before = _artifact_snapshot(run_dir) + monkeypatch.setattr( + run_redaction.os, + "fsync", + lambda fd: (_ for _ in ()).throw(OSError("directory fsync unsupported")), + ) + monkeypatch.setattr( + run_redaction, + "_exclusive_redaction_lock", + lambda *args, **kwargs: (_ for _ in ()).throw( + AssertionError("failed directory fsync probe reached lock acquisition") + ), + ) + + with pytest.raises(run_redaction.RedactionError, match="unsupported platform"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _artifact_snapshot(run_dir) == before + assert not (tmp_path / "workspace" / ".brigade" / "run.lock").exists() + + +def test_unsupported_platform_does_not_break_redaction_cli_help(monkeypatch, capsys): + monkeypatch.setattr(run_redaction, "_PLATFORM_NAME", "nt") + + with pytest.raises(SystemExit) as raised: + cli.main(["runs", "redact", "--help"]) + + assert raised.value.code == 0 + help_text = capsys.readouterr().out + assert "usage: brigade runs redact" in help_text + assert "Closed incident reason code" in help_text + + +def test_cleanup_refuses_unsupported_platform_before_lock_or_mutation(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + before = _artifact_snapshot(run_dir) + monkeypatch.setattr(run_redaction, "_PLATFORM_NAME", "nt", raising=False) + monkeypatch.setattr( + run_redaction, + "_exclusive_redaction_lock", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("unsupported cleanup reached lock acquisition")), + ) + + with pytest.raises(run_redaction.RedactionError, match="unsupported platform"): + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + + assert _artifact_snapshot(run_dir) == before + assert report.quarantine_path.is_file() + assert not (tmp_path / "workspace" / ".brigade" / "run.lock").exists() + + +@pytest.mark.parametrize( + ("failure_point", "exception_type"), + [ + ("fstat", KeyboardInterrupt), + ("identity", SystemExit), + ("mode", KeyboardInterrupt), + ], +) +def test_open_directory_handle_closes_fd_once_on_baseexception( + tmp_path, + monkeypatch, + failure_point, + exception_type, +): + directory = tmp_path / "directory" + directory.mkdir(mode=0o755) + real_open = run_journal._open_nofollow + real_close = os.close + real_fstat = os.fstat + opened: list[int] = [] + closed: list[int] = [] + + def capture_open(*args, **kwargs): + fd = real_open(*args, **kwargs) + opened.append(fd) + return fd + + def track_close(fd): + if opened and fd == opened[0]: + closed.append(fd) + return real_close(fd) + + monkeypatch.setattr(run_journal, "_open_nofollow", capture_open) + monkeypatch.setattr(run_redaction.os, "close", track_close) + if failure_point == "fstat": + monkeypatch.setattr( + run_redaction.os, + "fstat", + lambda fd: (_ for _ in ()).throw(exception_type()), + ) + elif failure_point == "identity": + + class ExplodingIdentity: + def __init__(self, fd): + info = real_fstat(fd) + self.st_mode = info.st_mode + self.st_ino = info.st_ino + + @property + def st_dev(self): + raise exception_type() + + monkeypatch.setattr( + run_redaction.os, + "fstat", + ExplodingIdentity, + ) + else: + monkeypatch.setattr( + run_journal, + "_chmod_fd_or_path", + lambda *args, **kwargs: (_ for _ in ()).throw(exception_type()), + ) + + with pytest.raises(exception_type): + run_redaction._open_directory_handle(directory, category="test directory") + + assert len(opened) == 1 + assert closed == opened + with pytest.raises(OSError): + real_fstat(opened[0]) + + +def test_open_directory_handle_preserves_baseexception_when_close_raises( + tmp_path, + monkeypatch, +): + directory = tmp_path / "directory" + directory.mkdir() + real_open = run_journal._open_nofollow + real_close = os.close + opened: list[int] = [] + closed: list[int] = [] + + def capture_open(*args, **kwargs): + fd = real_open(*args, **kwargs) + opened.append(fd) + return fd + + def close_then_raise(fd): + closed.append(fd) + real_close(fd) + raise OSError("simulated close failure") + + monkeypatch.setattr(run_journal, "_open_nofollow", capture_open) + monkeypatch.setattr( + run_redaction.os, + "fstat", + lambda fd: (_ for _ in ()).throw(KeyboardInterrupt()), + ) + monkeypatch.setattr(run_redaction.os, "close", close_then_raise) + + with pytest.raises(KeyboardInterrupt): + run_redaction._open_directory_handle(directory, category="test directory") + + assert closed == opened + + +def test_redaction_quarantines_rewrites_rechains_and_reprojects(tmp_path): + run_dir, before_projection, original_events = _authority_run(tmp_path) + + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + active = _journal_path(run_dir).read_bytes() + record = report.record_path.read_bytes() + assert SECRET.encode() not in active + assert SECRET.encode() not in record + assert b'"reason_code": "credential-exposure"' in record + assert b'"sequence_start": 2' in record + assert b'"sequence_end": 2' in record + assert report.quarantine_path.is_file() + assert SECRET.encode() in report.quarantine_path.read_bytes() + assert stat.S_IMODE(report.quarantine_path.stat().st_mode) == 0o600 + assert stat.S_IMODE(report.quarantine_path.parent.stat().st_mode) == 0o700 + + verified = run_journal.read_journal_bounded(_journal_path(run_dir)) + assert verified.chain_errors == [] + assert verified.partial_tail is None + assert len(verified.events) == len(original_events) + assert verified.events[1].payload == {"detail": "[REDACTED]"} + assert verified.events[0].event_digest == original_events[0].event_digest + assert verified.events[1].event_digest != original_events[1].event_digest + assert verified.events[2].previous_digest == verified.events[1].event_digest + + current = json.loads((run_dir / "run.json").read_text()) + after_projection = run_projector.project_run_snapshot(current, verified.events, journal_present=True).snapshot + assert current == after_projection + assert _without_tail_digest(current) == _without_tail_digest(before_projection) + assert current["journal_last_event_digest"] == verified.events[-1].event_digest + + +def test_redaction_refuses_live_lock_without_mutation(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + workspace = tmp_path / "workspace" + lock = workspace / ".brigade" / "run.lock" + lock.mkdir(parents=True) + (lock / "pid").write_text(f"{os.getpid()}\n") + (lock / "owner.json").write_text( + json.dumps( + { + "schema": "brigade.run_lock.v1", + "owner_token": "active-owner", + "pid": os.getpid(), + "run_dir": str(run_dir.resolve()), + "acquired_at": "2026-07-30T19:00:00+00:00", + } + ) + ) + before = _journal_path(run_dir).read_bytes() + + with pytest.raises(run_redaction.RedactionError, match="run lock state is live"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _journal_path(run_dir).read_bytes() == before + assert not (run_dir / "events" / "redactions").exists() + + +@pytest.mark.parametrize("lock_kind", ["malformed", "stale", "foreign"]) +def test_redaction_fails_closed_on_ambiguous_lock_state(tmp_path, lock_kind): + run_dir, _, _ = _authority_run(tmp_path) + workspace = tmp_path / "workspace" + lock = workspace / ".brigade" / "run.lock" + if lock_kind == "malformed": + lock.parent.mkdir(parents=True, exist_ok=True) + lock.write_text("not a directory") + else: + lock.mkdir(parents=True) + (lock / "pid").write_text("99999999\n") + owner_run = run_dir if lock_kind == "stale" else tmp_path / "other-run" + (lock / "owner.json").write_text( + json.dumps( + { + "schema": "brigade.run_lock.v1", + "owner_token": "dead-owner", + "pid": 99999999, + "run_dir": str(owner_run.resolve()), + "acquired_at": "2026-07-30T19:00:00+00:00", + } + ) + ) + before = _journal_path(run_dir).read_bytes() + + with pytest.raises(run_redaction.RedactionError, match="run lock state"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _journal_path(run_dir).read_bytes() == before + + +def test_redaction_refuses_malformed_journal_before_quarantine(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + journal = _journal_path(run_dir) + journal.write_bytes(journal.read_bytes() + b'{"partial":') + before = journal.read_bytes() + + with pytest.raises(run_redaction.RedactionError, match="journal"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert journal.read_bytes() == before + assert not (run_dir / "events" / "redactions").exists() + + +@pytest.mark.parametrize( + ("start", "end"), + [(0, 1), (1, 0), (2, 5), (True, 2), (2, False)], +) +def test_redaction_rejects_invalid_sequence_range_without_mutation(tmp_path, start, end): + run_dir, _, _ = _authority_run(tmp_path) + before = _journal_path(run_dir).read_bytes() + + with pytest.raises(run_redaction.RedactionError, match="sequence range"): + run_redaction.redact_journal( + run_dir, + sequence_start=start, + sequence_end=end, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _journal_path(run_dir).read_bytes() == before + + +def test_redaction_requires_explicit_operator_confirmation(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + before = _journal_path(run_dir).read_bytes() + + with pytest.raises(run_redaction.RedactionError, match="operator confirmation"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + ) + + assert _journal_path(run_dir).read_bytes() == before + + +@pytest.mark.parametrize("reason", ["", " ", "x" * 241, "line one\nline two"]) +def test_redaction_rejects_unbounded_or_multiline_reason(tmp_path, reason): + run_dir, _, _ = _authority_run(tmp_path) + before = _journal_path(run_dir).read_bytes() + + with pytest.raises(run_redaction.RedactionError, match="reason"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=reason, + operator_confirmed=True, + ) + + assert _journal_path(run_dir).read_bytes() == before + + +@pytest.mark.parametrize("reason", [SECRET, "planning"]) +def test_redaction_rejects_reason_that_copies_affected_private_value(tmp_path, reason): + run_dir, _, _ = _authority_run(tmp_path) + before = _journal_path(run_dir).read_bytes() + + with pytest.raises(run_redaction.RedactionError, match="reason"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=reason, + operator_confirmed=True, + ) + + assert _journal_path(run_dir).read_bytes() == before + + +def test_redaction_retry_is_idempotent(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + active = _journal_path(run_dir).read_bytes() + + replay = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert replay.operation_id == first.operation_id + assert replay.quarantine_path == first.quarantine_path + assert replay.record_path == first.record_path + assert _journal_path(run_dir).read_bytes() == active + assert len(list((run_dir / "events" / "redactions").glob("*/original.jsonl"))) == 1 + + +def test_redaction_retry_refuses_tampered_quarantine(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + first.quarantine_path.write_bytes(b"tampered") + + with pytest.raises(run_redaction.RedactionError, match="quarantine verification"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + +def test_redaction_retry_refuses_symlinked_record(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + outside = tmp_path / "outside-record.json" + outside.write_text("{}") + first.record_path.unlink() + first.record_path.symlink_to(outside) + + with pytest.raises(run_redaction.RedactionError, match="redaction record"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert outside.read_text() == "{}" + + +def test_redaction_replace_failure_retains_original_and_durable_quarantine(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + before = _journal_path(run_dir).read_bytes() + + def fail_replace(*args, **kwargs): + raise OSError("simulated replace failure") + + monkeypatch.setattr(run_redaction, "_replace_relative", fail_replace) + + with pytest.raises(run_redaction.RedactionError, match="replace"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert _journal_path(run_dir).read_bytes() == before + quarantines = list((run_dir / "events" / "redactions").glob("*/original.jsonl")) + assert len(quarantines) == 1 + assert quarantines[0].read_bytes() == before + + +def test_redaction_retry_completes_after_record_crash_window(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + real_write_record = run_redaction._write_redaction_record + calls = 0 + + def fail_once(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + raise OSError("simulated record failure") + return real_write_record(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_write_redaction_record", fail_once) + + with pytest.raises(run_redaction.RedactionError, match="record"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert SECRET.encode() not in _journal_path(run_dir).read_bytes() + assert list((run_dir / "events" / "redactions").glob("*/original.jsonl")) + + replay = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert replay.record_path.is_file() + assert SECRET.encode() not in replay.record_path.read_bytes() + + +def test_different_request_refuses_incomplete_recordless_operation(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + real_write_record = run_redaction._write_redaction_record + + def fail_record(*args, **kwargs): + raise OSError("simulated record failure") + + monkeypatch.setattr(run_redaction, "_write_redaction_record", fail_record) + with pytest.raises(run_redaction.RedactionError, match="record"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + first_operation = run_redaction._operation_id(RUN_ID, 2, 2, REASON_CODE) + first_dir, first_quarantine, first_record = run_redaction._operation_paths(run_dir, first_operation) + assert first_quarantine.is_file() + assert not first_record.exists() + assert json.loads((first_dir / "state.json").read_text())["phase"] == "replaced" + + monkeypatch.setattr(run_redaction, "_write_redaction_record", real_write_record) + with pytest.raises(run_redaction.RedactionError, match="incomplete redaction transaction"): + run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + + quarantines = list((run_dir / "events" / "redactions").glob("*/original.jsonl")) + assert quarantines == [first_quarantine] + recovered = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert recovered.record_path == first_record + assert recovered.record_path.is_file() + + +def test_redaction_retry_repairs_projection_after_replace_crash_window(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + real_replace_projection = run_redaction._replace_projection + calls = 0 + + def fail_once(*args, **kwargs): + nonlocal calls + calls += 1 + if calls == 1: + raise run_redaction.RedactionError("simulated projection replace failure") + return real_replace_projection(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_replace_projection", fail_once) + + with pytest.raises(run_redaction.RedactionError, match="projection replace"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert SECRET.encode() not in _journal_path(run_dir).read_bytes() + stale = json.loads((run_dir / "run.json").read_text()) + + replay = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + repaired = json.loads((run_dir / "run.json").read_text()) + assert stale["journal_last_event_digest"] != repaired["journal_last_event_digest"] + assert replay.record_path.is_file() + + +def test_cleanup_is_explicit_and_gated_by_post_rewrite_verification(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert report.quarantine_path.is_file() + + active = _journal_path(run_dir) + active.write_bytes(active.read_bytes() + b'{"partial":') + with pytest.raises(run_redaction.RedactionError, match="verification"): + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + assert report.quarantine_path.is_file() + + +def test_cleanup_removes_only_verified_quarantine(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + cleaned = run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + + assert cleaned.quarantine_path == report.quarantine_path + assert not report.quarantine_path.exists() + assert report.record_path.is_file() + assert SECRET.encode() not in _journal_path(run_dir).read_bytes() + assert SECRET.encode() not in report.record_path.read_bytes() + + +def test_cleanup_rebases_shadow_history_without_pre_redaction_digest_oracles(tmp_path): + run_dir, snapshot, events = _authority_run(tmp_path) + prior_projection = run_projector.project_run_snapshot(snapshot, events, journal_present=True) + prior_projection_digest = run_redaction._digest(prior_projection.to_bytes()) + prior_tail = events[-1] + run_shadow._record_match( + run_dir, + RUN_ID, + prior_tail.sequence, + prior_tail.event_digest, + prior_projection_digest, + prior_projection_digest, + ) + prior_shadow = json.loads(run_shadow.shadow_artifact_path(run_dir).read_text()) + forbidden_digests = { + prior_shadow["last_compared_event_digest"], + prior_shadow["last_shadow_digest"], + prior_shadow["last_projected_digest"], + } + artifact_path = run_shadow.shadow_artifact_path(run_dir) + corrupt_sibling = artifact_path.with_name(f"{artifact_path.name}.corrupt-20260730T190000000000Z") + stale_sibling = artifact_path.with_name(".stale-projector-v2-20260730T190000000000Z") + corrupt_sibling.write_text(json.dumps(prior_shadow)) + stale_sibling.write_text(json.dumps(prior_shadow)) + + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + + refreshed = json.loads(run_shadow.shadow_artifact_path(run_dir).read_text()) + encoded = json.dumps(refreshed, sort_keys=True) + assert refreshed["comparisons"] == 1 + assert refreshed["matches"] == 1 + assert len(refreshed["recent_records"]) == 1 + assert not corrupt_sibling.exists() + assert not stale_sibling.exists() + for digest in forbidden_digests: + assert digest not in encoded + + +def test_cleanup_retry_reverifies_retained_quarantine_after_authorization_crash(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + real_unlink = run_redaction.os.unlink + + def fail_unlink(path, *args, **kwargs): + if Path(path).name == "original.jsonl": + raise OSError("simulated cleanup failure") + return real_unlink(path, *args, **kwargs) + + monkeypatch.setattr(run_redaction.os, "unlink", fail_unlink) + with pytest.raises(run_redaction.RedactionError, match="quarantine cleanup"): + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + + report.quarantine_path.write_bytes(b"tampered") + monkeypatch.setattr(run_redaction.os, "unlink", real_unlink) + with pytest.raises(run_redaction.RedactionError, match="cleanup verification"): + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + assert report.quarantine_path.is_file() + + +def test_redaction_refuses_symlinked_journal(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + journal = _journal_path(run_dir) + target = tmp_path / "outside.jsonl" + target.write_bytes(journal.read_bytes()) + journal.unlink() + journal.symlink_to(target) + + with pytest.raises(run_redaction.RedactionError, match="journal"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert SECRET.encode() in target.read_bytes() + assert not (run_dir / "events" / "redactions").exists() + + +def test_redacted_payloads_remain_valid_for_projection_sensitive_status_fields(tmp_path): + run_dir, before_projection, _ = _authority_run(tmp_path) + + run_redaction.redact_journal( + run_dir, + sequence_start=1, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + + report = run_journal.read_journal_bounded(_journal_path(run_dir)) + assert report.chain_errors == [] + assert report.events[0].payload == {"status": "started"} + assert report.events[1].payload == {"detail": "[REDACTED]"} + assert report.events[2].event_type == run_checkpoint.CHECKPOINT_EVENT_TYPE + assert report.events[3].payload == {"status": "ok", "detail": "[REDACTED]"} + current = json.loads((run_dir / "run.json").read_text()) + assert current["status"] == before_projection["status"] == "ok" + assert current["journal_last_sequence"] == 4 + for event in report.events: + assert run_events.validate_event(event.to_dict()) == [] + + +def test_two_operator_processes_cannot_publish_concurrent_rewrites(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + marker = tmp_path / "first-operator-inside-transaction" + script = textwrap.dedent( + """ + import sys + import time + from pathlib import Path + from brigade import run_redaction + + run_dir = Path(sys.argv[1]) + marker = Path(sys.argv[2]) + reason = sys.argv[3] + should_pause = sys.argv[4] == "pause" + original = run_redaction._rewrite_events + + def paused_rewrite(*args, **kwargs): + marker.write_text("inside") + time.sleep(1.0) + return original(*args, **kwargs) + + if should_pause: + run_redaction._rewrite_events = paused_rewrite + try: + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=reason, + operator_confirmed=True, + ) + except run_redaction.RedactionError as exc: + print(exc.diagnostic) + raise SystemExit(2) + print("ok") + """ + ) + first = subprocess.Popen( + [sys.executable, "-c", script, str(run_dir), str(marker), REASON_CODE, "pause"], + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + for _ in range(100): + if marker.exists(): + break + time.sleep(0.02) + assert marker.exists() + + second = subprocess.run( + [ + sys.executable, + "-c", + script, + str(run_dir), + str(marker), + "personal-data-exposure", + "no-pause", + ], + text=True, + capture_output=True, + check=False, + ) + first_stdout, first_stderr = first.communicate(timeout=10) + + assert first.returncode == 0, first_stderr + assert first_stdout.strip() == "ok" + assert second.returncode == 2 + assert "run lock state is live" in second.stdout + assert run_journal.read_journal_bounded(_journal_path(run_dir)).chain_errors == [] + + +@pytest.mark.parametrize("symlink_level", ["redactions", "operation"]) +def test_redaction_rejects_symlinked_transaction_parent_without_external_write(tmp_path, symlink_level): + run_dir, _, _ = _authority_run(tmp_path) + external = tmp_path / "external" + external.mkdir() + redactions = run_dir / "events" / "redactions" + if symlink_level == "redactions": + redactions.symlink_to(external, target_is_directory=True) + else: + redactions.mkdir(mode=0o700) + operation_id = run_redaction._operation_id(RUN_ID, 2, 2, REASON_CODE) + (redactions / operation_id).symlink_to(external, target_is_directory=True) + + with pytest.raises(run_redaction.RedactionError, match="path|symlink"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert list(external.iterdir()) == [] + + +def test_redaction_rejects_raced_operation_parent_without_external_write(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + external = tmp_path / "external" + external.mkdir() + original_publish = run_redaction._publish_quarantine + + def race_parent(*args, **kwargs): + operation_dir = args[0] + operation_dir.rmdir() + operation_dir.symlink_to(external, target_is_directory=True) + return original_publish(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_publish_quarantine", race_parent) + with pytest.raises(run_redaction.RedactionError, match="path|symlink|quarantine"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + assert list(external.iterdir()) == [] + + +def test_redaction_fsyncs_created_parents_before_journal_replace(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + order: list[str] = [] + original_fsync = run_redaction._fsync_directory_handle + original_replace = run_redaction._replace_journal + + def record_fsync(path, fd, *, category): + order.append(f"fsync:{Path(path).name}") + return original_fsync(path, fd, category=category) + + def record_replace(*args, **kwargs): + order.append("replace:journal") + return original_replace(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_fsync_directory_handle", record_fsync) + monkeypatch.setattr(run_redaction, "_replace_journal", record_replace) + + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + replace_index = order.index("replace:journal") + assert order.index("fsync:events") < replace_index + assert order.index("fsync:redactions") < replace_index + + +def test_redaction_retries_after_partial_quarantine_write(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + original_write_all = run_redaction._write_all + + def partial_write(fd, data, *, category): + if category == "redaction quarantine": + os.write(fd, data[: max(1, len(data) // 2)]) + raise run_redaction.RedactionError("redaction quarantine write failed") + return original_write_all(fd, data, category=category) + + monkeypatch.setattr(run_redaction, "_write_all", partial_write) + with pytest.raises(run_redaction.RedactionError, match="quarantine"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + monkeypatch.setattr(run_redaction, "_write_all", original_write_all) + + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert report.quarantine_path.is_file() + assert not list(report.quarantine_path.parent.glob(".original.jsonl.*.tmp")) + + +def test_redaction_refsyncs_exact_quarantine_after_file_fsync_failure(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + original_fsync = run_redaction._fsync_file + failed = False + + def fail_first_fsync(fd, *, category): + nonlocal failed + if category == "redaction quarantine" and not failed: + failed = True + raise run_redaction.RedactionError("simulated quarantine fsync failure") + return original_fsync(fd, category=category) + + monkeypatch.setattr(run_redaction, "_fsync_file", fail_first_fsync) + with pytest.raises(run_redaction.RedactionError, match="quarantine"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + operation_id = run_redaction._operation_id(RUN_ID, 2, 2, REASON_CODE) + operation_dir, quarantine, _ = run_redaction._operation_paths(run_dir, operation_id) + assert operation_dir.is_dir() + quarantine.write_bytes(_journal_path(run_dir).read_bytes()) + quarantine.chmod(0o644) + quarantine_inode = quarantine.stat().st_ino + resynced = False + + def track_fsync(fd, *, category): + nonlocal resynced + if os.fstat(fd).st_ino == quarantine_inode: + resynced = True + return original_fsync(fd, category=category) + + monkeypatch.setattr(run_redaction, "_fsync_file", track_fsync) + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert resynced is True + + +def test_redaction_refsyncs_exact_quarantine_parent_before_state(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + original_fsync_dir = run_redaction._fsync_directory_handle + failed = False + + def fail_quarantine_parent(path, fd, *, category): + nonlocal failed + path = Path(path) + if path.name.startswith("redact-") and (path / "original.jsonl").exists() and not failed: + failed = True + raise run_redaction.RedactionError("simulated quarantine directory fsync failure") + return original_fsync_dir(path, fd, category=category) + + monkeypatch.setattr(run_redaction, "_fsync_directory_handle", fail_quarantine_parent) + with pytest.raises(run_redaction.RedactionError, match="quarantine"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + parent_resynced = False + original_write_state = run_redaction._write_state + + def track_fsync_dir(path, fd, *, category): + nonlocal parent_resynced + path = Path(path) + if path.name.startswith("redact-"): + parent_resynced = True + return original_fsync_dir(path, fd, category=category) + + def require_resync_before_state(*args, **kwargs): + assert parent_resynced is True + return original_write_state(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_fsync_directory_handle", track_fsync_dir) + monkeypatch.setattr(run_redaction, "_write_state", require_resync_before_state) + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + +@pytest.mark.parametrize("failure_name", ["publish"]) +def test_redaction_retries_after_quarantine_publish_failure(tmp_path, monkeypatch, failure_name): + run_dir, _, _ = _authority_run(tmp_path) + original_publish = run_redaction._publish_no_replace + failed = False + + def fail_once(*args, **kwargs): + nonlocal failed + if not failed: + failed = True + raise OSError("simulated quarantine publish failure") + return original_publish(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_publish_no_replace", fail_once) + with pytest.raises(run_redaction.RedactionError, match="quarantine"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + monkeypatch.setattr(run_redaction, "_publish_no_replace", original_publish) + + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert report.quarantine_path.is_file() + + +@pytest.mark.parametrize("checkpoint_fault", ["missing", "tampered", "hardlinked", "oversize"]) +def test_redaction_refuses_invalid_checkpoint_artifact_before_quarantine(tmp_path, checkpoint_fault): + run_dir, _, _ = _authority_run(tmp_path) + checkpoint = _latest_checkpoint_path(run_dir) + if checkpoint_fault == "missing": + checkpoint.unlink() + elif checkpoint_fault == "tampered": + checkpoint.write_bytes(b"{}") + elif checkpoint_fault == "hardlinked": + os.link(checkpoint, tmp_path / "checkpoint-copy") + else: + checkpoint.write_bytes(b"x" * (run_checkpoint.MAX_CHECKPOINT_BYTES + 1)) + + with pytest.raises(run_redaction.RedactionError, match="checkpoint"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert not (run_dir / "events" / "redactions").exists() + + +def test_redaction_refuses_uncovered_checkpoint_tail_before_quarantine(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + _append_uncovered_terminal_event(run_dir) + + with pytest.raises(run_redaction.RedactionError, match="checkpoint"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert not (run_dir / "events" / "redactions").exists() + + +def test_cleanup_refuses_missing_checkpoint_and_preserves_quarantine(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + _latest_checkpoint_path(run_dir).unlink() + + with pytest.raises(run_redaction.RedactionError, match="checkpoint"): + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + assert report.quarantine_path.is_file() + + +def test_redaction_rejects_generated_idempotency_collision(tmp_path): + operation_id = run_redaction._operation_id(RUN_ID, 2, 2, REASON_CODE) + collision = f"redaction:{operation_id}:2" + run_dir, _, _ = _authority_run(tmp_path, first_idempotency_key=collision) + + with pytest.raises(run_redaction.RedactionError, match="idempotency"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + +@pytest.mark.parametrize("status", ["started", "planning", "dispatching", "paused", "unknown"]) +def test_redaction_refuses_nonterminal_or_ambiguous_projection_status(tmp_path, status): + run_dir, _, _ = _authority_run(tmp_path) + snapshot = json.loads((run_dir / "run.json").read_text()) + snapshot["status"] = status + (run_dir / "run.json").write_text(json.dumps(snapshot, indent=2, sort_keys=True) + "\n") + + with pytest.raises(run_redaction.RedactionError, match="terminal"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert not (run_dir / "events" / "redactions").exists() + + +def test_overlapping_redactions_are_deterministic_and_idempotent(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + second = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason="policy-removal", + operator_confirmed=True, + ) + active = _journal_path(run_dir).read_bytes() + + replay = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason="policy-removal", + operator_confirmed=True, + ) + assert first.operation_id != second.operation_id + assert replay.operation_id == second.operation_id + assert _journal_path(run_dir).read_bytes() == active + + +def test_sequential_redaction_lineage_allows_each_quarantine_cleanup(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + second = run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + + first_cleaned = run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=first.operation_id, + operator_confirmed=True, + ) + second_cleaned = run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=second.operation_id, + operator_confirmed=True, + ) + assert first_cleaned.cleaned is True + assert second_cleaned.cleaned is True + assert not first.quarantine_path.exists() + assert not second.quarantine_path.exists() + + +def test_sequential_cleanup_removes_prior_rewritten_digest_alias_oracle(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + first_state_path = first.record_path.parent / "state.json" + first_state = json.loads(first_state_path.read_text()) + first_rewritten_digest = first_state["rewritten_sha256"] + assert first_rewritten_digest in first.record_path.read_text() + + second = run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + second_state = json.loads((second.record_path.parent / "state.json").read_text()) + assert second_state["original_sha256"] == first_rewritten_digest + + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=second.operation_id, + operator_confirmed=True, + ) + + for path in sorted((run_dir / "events" / "redactions").rglob("*")): + if path.is_file(): + assert first_rewritten_digest.encode() not in path.read_bytes() + + first_cleaned = run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=first.operation_id, + operator_confirmed=True, + ) + assert first_cleaned.cleaned is True + + +def test_cleanup_retry_converges_parent_record_state_retirement_split(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + second = run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + first_state_path = first.record_path.parent / "state.json" + first_digest = json.loads(first_state_path.read_text())["rewritten_sha256"] + real_write_state = run_redaction._write_state + failed = False + + def fail_parent_state(*args, **kwargs): + nonlocal failed + if ( + kwargs.get("operation_id") == first.operation_id + and kwargs.get("rewritten_digest_retired_by") == second.operation_id + and not failed + ): + failed = True + raise run_redaction.RedactionError("simulated parent retirement state failure") + return real_write_state(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_write_state", fail_parent_state) + with pytest.raises(run_redaction.RedactionError, match="parent retirement state"): + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=second.operation_id, + operator_confirmed=True, + ) + + assert not second.quarantine_path.exists() + assert json.loads((second.record_path.parent / "state.json").read_text())["phase"] == "cleanup-authorized" + assert json.loads(first.record_path.read_text())["rewritten_digest_retired_by"] == second.operation_id + assert json.loads(first_state_path.read_text())["rewritten_sha256"] == first_digest + + monkeypatch.setattr(run_redaction, "_write_state", real_write_state) + cleaned = run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=second.operation_id, + operator_confirmed=True, + ) + assert cleaned.cleaned is True + assert first_digest.encode() not in first.record_path.read_bytes() + assert first_digest.encode() not in first_state_path.read_bytes() + + +def test_cleanup_retry_finishes_after_child_record_before_cleaned_state(tmp_path, monkeypatch): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + second = run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + second_state_path = second.record_path.parent / "state.json" + second_original_digest = json.loads(second_state_path.read_text())["original_sha256"] + real_write_state = run_redaction._write_state + failed = False + + def fail_child_cleaned_state(*args, **kwargs): + nonlocal failed + if kwargs.get("operation_id") == second.operation_id and kwargs.get("phase") == "cleaned" and not failed: + failed = True + raise run_redaction.RedactionError("simulated child cleaned state failure") + return real_write_state(*args, **kwargs) + + monkeypatch.setattr(run_redaction, "_write_state", fail_child_cleaned_state) + with pytest.raises(run_redaction.RedactionError, match="child cleaned state"): + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=second.operation_id, + operator_confirmed=True, + ) + + assert not second.quarantine_path.exists() + assert json.loads(second_state_path.read_text())["phase"] == "cleanup-authorized" + assert json.loads(second.record_path.read_text())["quarantine_retained"] is False + assert json.loads(first.record_path.read_text())["rewritten_digest_retired_by"] == second.operation_id + stale_parent_state = first.record_path.parent / f".state.json.{'a' * 32}.tmp" + stale_parent_record = first.record_path.parent / f".record.json.{'b' * 32}.tmp" + stale_parent_state.write_text(second_original_digest) + stale_parent_record.write_text(second_original_digest) + + monkeypatch.setattr(run_redaction, "_write_state", real_write_state) + cleaned = run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=second.operation_id, + operator_confirmed=True, + ) + assert cleaned.cleaned is True + assert not stale_parent_state.exists() + assert not stale_parent_record.exists() + for path in sorted((run_dir / "events" / "redactions").rglob("*")): + if path.is_file(): + assert second_original_digest.encode() not in path.read_bytes() + + +def test_redaction_inventory_rejects_record_without_state(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + (first.record_path.parent / "state.json").unlink() + + with pytest.raises(run_redaction.RedactionError, match="state/record mismatch"): + run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + assert len(list((run_dir / "events" / "redactions").glob("*/original.jsonl"))) == 1 + + +def test_redaction_inventory_rejects_forged_multiple_tip_graph(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + first = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + second = run_redaction.redact_journal( + run_dir, + sequence_start=4, + sequence_end=4, + reason="personal-data-exposure", + operator_confirmed=True, + ) + second_state_path = second.record_path.parent / "state.json" + second_state = json.loads(second_state_path.read_text()) + second_record = json.loads(second.record_path.read_text()) + second_state["parent_operation_id"] = None + second_record["parent_operation_id"] = None + second_state_path.write_text(json.dumps(second_state)) + second.record_path.write_text(json.dumps(second_record)) + + with pytest.raises(run_redaction.RedactionError, match="multiple tips"): + run_redaction.redact_journal( + run_dir, + sequence_start=3, + sequence_end=3, + reason="other-sensitive-data", + operator_confirmed=True, + ) + assert first.quarantine_path.exists() + assert second.quarantine_path.exists() + + +@pytest.mark.parametrize("artifact", ["journal", "quarantine", "state"]) +def test_redaction_refuses_hardlinked_sensitive_artifact(tmp_path, artifact): + run_dir, _, _ = _authority_run(tmp_path) + if artifact == "journal": + os.link(_journal_path(run_dir), tmp_path / "journal-copy") + with pytest.raises(run_redaction.RedactionError, match="link"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + return + + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + target = report.quarantine_path if artifact == "quarantine" else report.record_path.parent / "state.json" + os.link(target, tmp_path / f"{artifact}-copy") + with pytest.raises(run_redaction.RedactionError, match="link"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + + +def test_redaction_normalizes_private_file_modes(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + journal = _journal_path(run_dir) + journal.chmod(0o644) + + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert stat.S_IMODE(journal.stat().st_mode) == 0o600 + assert stat.S_IMODE(report.quarantine_path.stat().st_mode) == 0o600 + assert stat.S_IMODE((report.record_path.parent / "state.json").stat().st_mode) == 0o600 + assert stat.S_IMODE(report.record_path.stat().st_mode) == 0o600 + + report.quarantine_path.chmod(0o644) + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + assert stat.S_IMODE(report.quarantine_path.stat().st_mode) == 0o600 + + +@pytest.mark.parametrize( + "reason", + [ + "credential exposure", + "Credential-Exposure", + "secret", + "credential-exposure-extra", + "x" * 241, + ], +) +def test_redaction_accepts_only_closed_reason_codes(tmp_path, reason): + run_dir, _, _ = _authority_run(tmp_path) + + with pytest.raises(run_redaction.RedactionError, match="reason code"): + run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=reason, + operator_confirmed=True, + ) + + +def test_cleanup_removes_original_digest_oracle_from_durable_metadata(tmp_path): + run_dir, _, _ = _authority_run(tmp_path) + report = run_redaction.redact_journal( + run_dir, + sequence_start=2, + sequence_end=2, + reason=REASON_CODE, + operator_confirmed=True, + ) + operation_dir = report.record_path.parent + stale_state = operation_dir / f".state.json.{'a' * 32}.tmp" + stale_quarantine = operation_dir / f".original.jsonl.{'b' * 32}.tmp" + state_before = json.loads((operation_dir / "state.json").read_text()) + stale_state.write_text(json.dumps({"original_sha256": state_before["original_sha256"]})) + stale_quarantine.write_bytes(report.quarantine_path.read_bytes()) + run_redaction.cleanup_redaction_quarantine( + run_dir, + operation_id=report.operation_id, + operator_confirmed=True, + ) + + state = json.loads((report.record_path.parent / "state.json").read_text()) + record = json.loads(report.record_path.read_text()) + assert "original_sha256" not in state + assert "original_sha256" not in record + assert "original_journal_sha256" not in state + assert "original_journal_sha256" not in record + assert not stale_state.exists() + assert not stale_quarantine.exists() diff --git a/tests/test_runs_cmd.py b/tests/test_runs_cmd.py index b80dee47..f819f784 100644 --- a/tests/test_runs_cmd.py +++ b/tests/test_runs_cmd.py @@ -17,6 +17,7 @@ from brigade import run_journal from brigade import run_lifecycle from brigade import run_resume +from brigade import run_redaction from brigade import runguard from brigade import runs_cmd from brigade import tools_cmd @@ -899,6 +900,115 @@ def test_runs_recover_cli_dispatches_resolved_run(tmp_path, monkeypatch): assert seen == {"run": str(run_dir), "cwd": tmp_path, "runs_dir": None} +def test_runs_redact_cli_dispatches_operator_procedure(tmp_path, monkeypatch): + run_dir = tmp_path / "run" + run_dir.mkdir() + seen = {} + + def fake_redact(run, **kwargs): + seen.update(run=run, **kwargs) + return 0 + + monkeypatch.setattr(runs_cmd, "redact", fake_redact, raising=False) + + rc = cli.main( + [ + "runs", + "redact", + str(run_dir), + "--from-sequence", + "2", + "--to-sequence", + "4", + "--reason", + "credential-exposure", + "--operator-confirm", + "--cwd", + str(tmp_path), + ] + ) + + assert rc == 0 + assert seen == { + "run": str(run_dir), + "cwd": tmp_path, + "runs_dir": None, + "sequence_start": 2, + "sequence_end": 4, + "reason": "credential-exposure", + "operator_confirmed": True, + "cleanup_operation": None, + } + + +def test_runs_redact_cli_dispatches_explicit_cleanup(tmp_path, monkeypatch): + run_dir = tmp_path / "run" + run_dir.mkdir() + seen = {} + monkeypatch.setattr( + runs_cmd, + "redact", + lambda run, **kwargs: seen.update(run=run, **kwargs) or 0, + raising=False, + ) + + rc = cli.main( + [ + "runs", + "redact", + str(run_dir), + "--cleanup-quarantine", + "redact-abc123", + "--operator-confirm", + "--cwd", + str(tmp_path), + ] + ) + + assert rc == 0 + assert seen == { + "run": str(run_dir), + "cwd": tmp_path, + "runs_dir": None, + "sequence_start": None, + "sequence_end": None, + "reason": None, + "operator_confirmed": True, + "cleanup_operation": "redact-abc123", + } + + +def test_runs_redact_cli_reports_removed_quarantine_on_cleaned_replay(tmp_path, monkeypatch, capsys): + run_dir = tmp_path / "run" + run_dir.mkdir() + monkeypatch.setattr( + run_redaction, + "redact_journal", + lambda *args, **kwargs: run_redaction.RedactionReport( + operation_id="redact-0123456789abcdef", + sequence_start=2, + sequence_end=2, + quarantine_path=run_dir / "events" / "redactions" / "redact-0123456789abcdef" / "original.jsonl", + record_path=run_dir / "events" / "redactions" / "redact-0123456789abcdef" / "record.json", + cleaned=True, + ), + ) + + rc = runs_cmd.redact( + run_dir, + cwd=tmp_path, + sequence_start=2, + sequence_end=2, + reason="credential-exposure", + operator_confirmed=True, + ) + + assert rc == 0 + output = capsys.readouterr().out + assert "quarantine: removed" in output + assert "quarantine: retained" not in output + + def test_runs_recover_marks_dead_owner_run_terminal(tmp_path, capsys): workspace = tmp_path / "workspace" run_dir = workspace / ".brigade" / "runs" / "orphan"