From 6ca44dd68c69b46bca19e2d1535377ead2c3791d Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 05:18:21 +0800 Subject: [PATCH 01/12] feat(issue62): add fail-closed live control orchestration --- scripts/run_issue_62_live_control.py | 680 +++++++++++++++++- ...est_issue_62_live_control_orchestration.py | 185 +++++ 2 files changed, 852 insertions(+), 13 deletions(-) create mode 100644 tests/test_issue_62_live_control_orchestration.py diff --git a/scripts/run_issue_62_live_control.py b/scripts/run_issue_62_live_control.py index 7b4a6856..494d7ef6 100644 --- a/scripts/run_issue_62_live_control.py +++ b/scripts/run_issue_62_live_control.py @@ -1,10 +1,12 @@ """Run the bounded, sanitized orchestration envelope for Issue #62. This module deliberately separates orchestration from the real upstream -controls. The default command-line mode is ``--fixture`` and therefore cannot -qualify Issue #62. A future live runner can reuse :class:`BoundedPhaseRunner` -with real command/phase callbacks after the exact candidate and route have been -reviewed. +controls. The default command-line mode remains ``--fixture`` and therefore +cannot qualify Issue #62. An explicitly enabled ``--enable-live-control`` +mode accepts a sanitized, operator-supplied plan, binds its candidate/catalog/ +route files, runs real CLI and sidecar commands through +:class:`BoundedPhaseRunner`, and still leaves qualification closed for +independent review. The harness has three non-negotiable properties: @@ -25,6 +27,7 @@ import argparse from dataclasses import dataclass from datetime import datetime, timezone +import hashlib import json import os from pathlib import Path @@ -35,7 +38,7 @@ import tempfile import threading import time -from typing import Callable, Literal, Sequence +from typing import Any, Callable, Literal, Mapping, NoReturn, Sequence SCHEMA_VERSION = "codexhub.issue62.live-control.v1" @@ -56,9 +59,59 @@ "process_start_failed", "stale_artifact", "termination_failed", + # Live-plan validation codes are intentionally fixed and safe to + # expose in the operator result. They never include paths, command + # arguments, or exception text. + "live_control_plan_missing", + "live_control_plan_invalid", + "candidate_binding_mismatch", + "candidate_sha_binding_mismatch", + "cli_package_binding_mismatch", + "catalog_binding_mismatch", + "route_binding_mismatch", + "control_labels_incomplete", + "sidecar_capture_missing", + "sidecar_capture_incomplete", + "identity_replay_incomplete", + "manifest_reconcile_failed", } ) +LIVE_PLAN_SCHEMA = "codexhub.issue62.live-control-plan.v1" +LIVE_SCOPE = "authorized_live_control" +CONTROL_NAMES = ( + "streaming_text", + "streaming_function_history", + "non_streaming_text", + "choice_auto", + "choice_none", + "terminal_success", + "terminal_error", + "error_json", +) +CONTROL_NAME_SET = frozenset(CONTROL_NAMES) +LIVE_PLAN_FIELDS = frozenset( + { + "schema", + "verification_scope", + "candidate_identity", + "binding", + "catalog_model_entry_id", + "cli", + "sidecars", + "controls", + "replays", + } +) + + +class LiveControlValidationError(ValueError): + """A deterministic, path-free live-control plan validation failure.""" + + def __init__(self, code: str) -> None: + self.code = code if code in ALLOWED_ERROR_CODES else "live_control_plan_invalid" + super().__init__(self.code) + class HarnessFailure(RuntimeError): """An internal failure represented by a safe, allow-listed code.""" @@ -151,6 +204,14 @@ class CommandResult: terminated: bool = True +@dataclass(frozen=True) +class BackgroundProcess: + """A bounded child kept alive while a control command executes.""" + + phase: str + process: subprocess.Popen[bytes] + + def _validate_timeout(timeout_seconds: float) -> float: if not MIN_TIMEOUT_SECONDS <= timeout_seconds <= MAX_TIMEOUT_SECONDS: raise ValueError("timeout must be between 30 and 60 seconds") @@ -277,6 +338,8 @@ def __init__( self.journal = SanitizedPhaseJournal(run_root, identity) self._active_process: subprocess.Popen[bytes] | None = None self._active_lock = threading.Lock() + self._background_processes: dict[str, BackgroundProcess] = {} + self._background_lock = threading.Lock() self._cancel_requested = False self._cleanup_called = False self._children_terminated = True @@ -291,6 +354,88 @@ def _set_active(self, process: subprocess.Popen[bytes] | None) -> None: with self._active_lock: self._active_process = process + @staticmethod + def _validate_argv(argv: Sequence[str]) -> bool: + return bool(argv) and all( + isinstance(item, str) and bool(item) and "\x00" not in item for item in argv + ) + + def start_background(self, phase: str, argv: Sequence[str]) -> BackgroundProcess: + """Start one bounded sidecar process without retaining its output.""" + + phase_name = self._phase_name(phase) + self.journal.append( + marker=f"{phase_name}_started", status="started", status_code="phase_started" + ) + if not self._validate_argv(argv): + self.journal.append( + marker=f"{phase_name}_completed", + status="failed", + status_code="process_start_failed", + ) + raise HarnessFailure("process_start_failed") + creationflags = 0 + start_new_session = False + if os.name == "nt": + creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200) + else: + start_new_session = True + try: + process = self._process_factory( + list(argv), + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + creationflags=creationflags, + start_new_session=start_new_session, + shell=False, + ) + except Exception as error: + self.journal.append( + marker=f"{phase_name}_completed", + status="failed", + status_code="process_start_failed", + ) + raise HarnessFailure("process_start_failed") from error + handle = BackgroundProcess(phase_name, process) + with self._background_lock: + if phase_name in self._background_processes: + # A duplicate phase would make cleanup ambiguous. Terminate + # the just-started process before failing closed. + _terminate_process(process) + raise HarnessFailure("phase_exception") + self._background_processes[phase_name] = handle + return handle + + def stop_background(self, handle: BackgroundProcess, *, status_code: str = "cancelled") -> bool: + """Stop one background child and emit its terminal journal marker.""" + + with self._background_lock: + current = self._background_processes.pop(handle.phase, None) + if current is None: + return True + started = time.monotonic() + terminated = _terminate_process(current.process) + self._children_terminated = self._children_terminated and terminated + code = status_code if terminated else "termination_failed" + self.journal.append( + marker=f"{current.phase}_completed", + status="completed" if terminated else "failed", + status_code=code, + duration_ms=int((time.monotonic() - started) * 1000), + exit_code=(int(current.process.poll()) if current.process.poll() is not None else None), + ) + return terminated + + def _stop_background_processes(self) -> int: + with self._background_lock: + handles = list(self._background_processes.values()) + failures = 0 + for handle in handles: + if not self.stop_background(handle): + failures += 1 + return failures + def mark_phase( self, marker: str, @@ -351,7 +496,7 @@ def finish(result: CommandResult) -> CommandResult: return finish(CommandResult("cancelled", "cancelled", terminated=True)) if self._cancel_requested: return finish(CommandResult("cancelled", "cancelled", terminated=True)) - if not argv or any(not isinstance(item, str) or not item for item in argv): + if not self._validate_argv(argv): return finish(CommandResult("failed", "process_start_failed", terminated=True)) creationflags = 0 @@ -450,6 +595,7 @@ def cleanup(self, actions: Sequence[CleanupAction] = ()) -> dict[str, object]: self.journal.append(marker="cleanup_started", status="started", status_code="phase_started") if not self.cancel(): failures += 1 + failures += self._stop_background_processes() for action in reversed(tuple(actions)): try: if action() is False: @@ -531,6 +677,457 @@ def append( self.records.append(record) +def _live_plan_fail(code: str) -> NoReturn: + raise LiveControlValidationError(code) + + +def _validate_plan_argv(value: Any, code: str = "live_control_plan_invalid") -> list[str]: + if not isinstance(value, list) or not value or any( + not isinstance(item, str) or not item or "\x00" in item for item in value + ): + _live_plan_fail(code) + # Return a fresh list so callers cannot mutate the source object while a + # plan is executing. + return list(value) + + +def _validate_plan_path(value: Any) -> str: + if not isinstance(value, str) or not value or "\x00" in value: + _live_plan_fail("live_control_plan_invalid") + return value + + +def _file_sha256(path: str) -> str: + try: + target = Path(path) + if target.is_symlink() or not target.is_file(): + raise OSError + digest = hashlib.sha256() + with target.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + except OSError: + return "" + + +def _validate_live_binding( + identity: CandidateIdentity, binding: Mapping[str, Any] +) -> dict[str, str]: + expected_fields = frozenset( + {"candidate_sha_file", "cli_package_file", "catalog_file", "route_file"} + ) + if frozenset(binding) != expected_fields: + _live_plan_fail("candidate_binding_mismatch") + paths = {key: _validate_plan_path(binding[key]) for key in expected_fields} + try: + candidate_text = Path(paths["candidate_sha_file"]).read_text(encoding="ascii").strip() + except (OSError, UnicodeError): + _live_plan_fail("candidate_sha_binding_mismatch") + if candidate_text.lower() != identity.candidate_sha: + _live_plan_fail("candidate_sha_binding_mismatch") + checks = ( + ("cli_package_file", identity.cli_package_sha256, "cli_package_binding_mismatch"), + ("catalog_file", identity.catalog_digest, "catalog_binding_mismatch"), + ("route_file", identity.route_digest, "route_binding_mismatch"), + ) + for field, expected, code in checks: + if _file_sha256(paths[field]) != expected: + _live_plan_fail(code) + return paths + + +def _validate_sidecar_spec(value: Any) -> dict[str, Any]: + if not isinstance(value, Mapping) or frozenset(value) != frozenset({"argv", "output_dir"}): + _live_plan_fail("live_control_plan_invalid") + return { + "argv": _validate_plan_argv(value.get("argv")), + "output_dir": _validate_plan_path(value.get("output_dir")), + } + + +def _validate_sidecars(value: Any) -> dict[str, list[dict[str, Any]]]: + if not isinstance(value, Mapping) or frozenset(value) != frozenset({"pre", "post"}): + _live_plan_fail("live_control_plan_invalid") + result: dict[str, list[dict[str, Any]]] = {} + output_dirs: set[str] = set() + for hop in ("pre", "post"): + raw = value.get(hop) + specs = raw if isinstance(raw, list) else [raw] + if not specs or len(specs) > len(CONTROL_NAMES): + _live_plan_fail("live_control_plan_invalid") + normalized = [_validate_sidecar_spec(item) for item in specs] + for spec in normalized: + directory = str(Path(spec["output_dir"]).resolve()) + if directory in output_dirs: + _live_plan_fail("live_control_plan_invalid") + output_dirs.add(directory) + result[hop] = normalized + return result + + +def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, Any]: + """Load and independently bind an explicitly authorized live plan. + + This function performs no subprocess or network work. It is deliberately + strict: a plan must carry all candidate/catalog/route digest bindings and + exactly the eight semantic control labels before the live runner can start. + """ + + if isinstance(source, (str, Path)): + source_path = Path(source) + if not source_path.exists() or not source_path.is_file(): + _live_plan_fail("live_control_plan_missing") + try: + payload = json.loads(source_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + _live_plan_fail("live_control_plan_invalid") + else: + payload = source + if not isinstance(payload, Mapping) or frozenset(payload) != LIVE_PLAN_FIELDS: + _live_plan_fail("live_control_plan_invalid") + if payload.get("schema") != LIVE_PLAN_SCHEMA or payload.get("verification_scope") != LIVE_SCOPE: + _live_plan_fail("live_control_plan_invalid") + + raw_identity = payload.get("candidate_identity") + if not isinstance(raw_identity, Mapping): + _live_plan_fail("candidate_binding_mismatch") + identity_fields = frozenset( + {"candidate_sha", "cli_version", "cli_package_sha256", "catalog_digest", "route_digest"} + ) + if frozenset(raw_identity) not in (identity_fields, identity_fields | {"cli_source_commit"}): + _live_plan_fail("candidate_binding_mismatch") + try: + identity = CandidateIdentity( + candidate_sha=raw_identity["candidate_sha"], + cli_version=raw_identity["cli_version"], + cli_package_sha256=raw_identity["cli_package_sha256"], + catalog_digest=raw_identity["catalog_digest"], + route_digest=raw_identity["route_digest"], + cli_source_commit=raw_identity.get("cli_source_commit"), + ) + except (KeyError, TypeError, ValueError): + _live_plan_fail("candidate_binding_mismatch") + + binding = payload.get("binding") + if not isinstance(binding, Mapping): + _live_plan_fail("candidate_binding_mismatch") + normalized_binding = _validate_live_binding(identity, binding) + + model = payload.get("catalog_model_entry_id") + if ( + not isinstance(model, str) + or re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", model) is None + or model != "gpt-5.6-sol" + ): + _live_plan_fail("catalog_binding_mismatch") + + cli = payload.get("cli") + if not isinstance(cli, Mapping) or frozenset(cli) != frozenset({"argv"}): + _live_plan_fail("live_control_plan_invalid") + normalized_cli = {"argv": _validate_plan_argv(cli.get("argv"))} + normalized_sidecars = _validate_sidecars(payload.get("sidecars")) + + controls = payload.get("controls") + if not isinstance(controls, list) or len(controls) != len(CONTROL_NAMES): + _live_plan_fail("control_labels_incomplete") + normalized_controls: list[dict[str, Any]] = [] + names: list[str] = [] + allowed_control_fields = frozenset({"name", "args", "capture", "pre_record", "post_record"}) + for item in controls: + if not isinstance(item, Mapping) or not frozenset(item).issubset(allowed_control_fields): + _live_plan_fail("live_control_plan_invalid") + name = item.get("name") + if not isinstance(name, str) or name not in CONTROL_NAME_SET: + _live_plan_fail("control_labels_incomplete") + names.append(name) + raw_args = item.get("args", []) + normalized: dict[str, Any] = { + "name": name, + "args": _validate_plan_argv(raw_args, "live_control_plan_invalid") + if raw_args != [] + else [], + "capture": item.get("capture", {}), + } + if not isinstance(normalized["capture"], Mapping): + _live_plan_fail("live_control_plan_invalid") + for field in ("pre_record", "post_record"): + if field in item: + normalized[field] = _validate_plan_path(item[field]) + normalized_controls.append(normalized) + if set(names) != CONTROL_NAME_SET or len(set(names)) != len(names): + _live_plan_fail("control_labels_incomplete") + + replays = payload.get("replays") + replay_names = frozenset({"identity", "mutation", "deletion", "loss"}) + if not isinstance(replays, Mapping) or frozenset(replays) != replay_names: + _live_plan_fail("identity_replay_incomplete") + normalized_replays: dict[str, dict[str, Any]] = {} + for name in replay_names: + spec = replays.get(name) + if not isinstance(spec, Mapping) or frozenset(spec) != frozenset({"argv"}): + _live_plan_fail("identity_replay_incomplete") + normalized_replays[name] = {"argv": _validate_plan_argv(spec.get("argv"))} + + return { + "schema": LIVE_PLAN_SCHEMA, + "verification_scope": LIVE_SCOPE, + "candidate_identity": identity.as_dict(), + "binding": normalized_binding, + "catalog_model_entry_id": model, + "cli": normalized_cli, + "sidecars": normalized_sidecars, + "controls": normalized_controls, + "replays": normalized_replays, + } + + +def _load_manifest_builder() -> Any: + # The evidence scripts are intentionally loosely coupled. Importing the + # builder lazily keeps fixture mode usable from a source checkout and does + # not add any production routing dependency. + try: + from build_issue_62_control_manifest import ( # type: ignore[import-not-found] + ManifestValidationError, + build_manifest, + reconcile_manifest, + replay_manifest, + ) + except ImportError as error: + raise HarnessFailure("phase_exception") from error + return ManifestValidationError, build_manifest, reconcile_manifest, replay_manifest + + +def _prepare_capture_dirs(sidecars: Mapping[str, Sequence[Mapping[str, Any]]]) -> None: + for specs in sidecars.values(): + for spec in specs: + target = Path(str(spec["output_dir"])) + try: + if target.exists(): + if target.is_symlink() or not target.is_dir() or any(target.iterdir()): + raise OSError + else: + target.mkdir(parents=True, exist_ok=False) + except OSError: + raise LiveControlValidationError("sidecar_capture_missing") + + +def _record_path_for_control( + control: Mapping[str, Any], + *, + hop: str, + specs: Sequence[Mapping[str, Any]], + index: int, +) -> Path: + explicit = control.get(f"{hop}_record") + if explicit is not None: + path = Path(str(explicit)) + for spec in specs: + root = Path(str(spec["output_dir"])) + try: + path.resolve().relative_to(root.resolve()) + except ValueError: + continue + return path + raise LiveControlValidationError("sidecar_capture_missing") + if len(specs) == len(CONTROL_NAMES): + root = Path(str(specs[index]["output_dir"])) + candidates = sorted(path for path in root.glob("*.json") if path.is_file()) + if len(candidates) == 1: + return candidates[0] + raise LiveControlValidationError("sidecar_capture_missing") + if len(specs) == 1: + root = Path(str(specs[0]["output_dir"])) + candidates = sorted(path for path in root.glob("*.json") if path.is_file()) + if len(candidates) == len(CONTROL_NAMES): + return candidates[index] + raise LiveControlValidationError("sidecar_capture_missing") + + +def _read_sidecar_record(path: Path, *, hop: str) -> dict[str, Any]: + try: + if path.is_symlink() or not path.is_file(): + raise OSError + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + raise LiveControlValidationError("sidecar_capture_incomplete") + try: + _ManifestValidationError, _build_manifest, _reconcile_manifest, _replay_manifest = ( + _load_manifest_builder() + ) + from build_issue_62_control_manifest import sanitize_sidecar_record # type: ignore[import-not-found] + + # Validate against the capture-only schema, but keep the raw + # transport metadata in memory for ``build_manifest``. The builder + # drops capture IDs/hop fields before writing the canonical manifest. + sanitize_sidecar_record(record, expected_hop=hop) + return dict(record) + except Exception as error: + if isinstance(error, LiveControlValidationError): + raise + raise LiveControlValidationError("sidecar_capture_incomplete") from error + + +def _build_live_controls( + plan: Mapping[str, Any], +) -> list[dict[str, Any]]: + sidecars = plan["sidecars"] + controls: list[dict[str, Any]] = [] + for index, control in enumerate(plan["controls"]): + semantic = dict(control.get("capture", {})) + semantic["name"] = control["name"] + semantic["pre"] = _read_sidecar_record( + _record_path_for_control(control, hop="pre", specs=sidecars["pre"], index=index), + hop="pre", + ) + semantic["post"] = _read_sidecar_record( + _record_path_for_control(control, hop="post", specs=sidecars["post"], index=index), + hop="post", + ) + controls.append(semantic) + return controls + + +def _manifest_candidate(plan: Mapping[str, Any]) -> dict[str, Any]: + identity = plan["candidate_identity"] + return { + "codexhub_candidate_sha": identity["candidate_sha"], + "cli_version": identity["cli_version"], + "cli_source_commit": identity.get("cli_source_commit"), + "cli_source_commit_status": ( + "published" if identity.get("cli_source_commit") else "not_published_by_registry" + ), + "cli_package_sha256": identity["cli_package_sha256"], + "catalog_snapshot_sha256": identity["catalog_digest"], + "catalog_model_entry_id": plan["catalog_model_entry_id"], + } + + +def run_live_control( + plan: Path | str | Mapping[str, Any], + *, + run_root: Path, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + manifest_out: Path | None = None, +) -> dict[str, Any]: + """Execute an explicitly supplied live plan through bounded children. + + The result is intentionally non-qualifying even when every command, + capture, and negative replay check succeeds. An independent maintainer + review is still required before any Issue #62 claim can be made. + """ + + loaded = load_live_control_plan(plan) + identity = CandidateIdentity(**loaded["candidate_identity"]) + runner = BoundedPhaseRunner(run_root, identity, timeout_seconds=timeout_seconds) + handles: list[BackgroundProcess] = [] + result_status = "completed" + result_code = "ok" + manifest: dict[str, Any] | None = None + manifest_reconciled = False + replay_results: dict[str, str] = {} + try: + runner.mark_phase("binding_validated") + _prepare_capture_dirs(loaded["sidecars"]) + for hop in ("pre", "post"): + for index, spec in enumerate(loaded["sidecars"][hop]): + try: + handles.append(runner.start_background(f"{hop}_sidecar_{index}", spec["argv"])) + except HarnessFailure as error: + result_status, result_code = "failed", error.code + break + if result_status != "completed": + break + + if result_status == "completed": + for control in loaded["controls"]: + result = runner.run_subprocess( + f"control_{control['name']}", + [*loaded["cli"]["argv"], *control["args"]], + ) + if result.status != "completed": + result_status, result_code = result.status, result.status_code + break + + # Stop sidecars before reading their atomic records. Cleanup repeats + # this operation defensively for cancellation/error paths. + for handle in reversed(handles): + if not runner.stop_background(handle): + result_status, result_code = "failed", "termination_failed" + + if result_status == "completed": + try: + controls = _build_live_controls(loaded) + _ManifestValidationError, build_manifest, reconcile_manifest, replay_manifest = ( + _load_manifest_builder() + ) + manifest = build_manifest( + controls, + candidate_identity=_manifest_candidate(loaded), + verification_scope=LIVE_SCOPE, + ) + report = reconcile_manifest(manifest) + if not report["reconciled"]: + result_status, result_code = "failed", "manifest_reconcile_failed" + else: + manifest_reconciled = True + if manifest_out is not None: + _write_sanitized_json(manifest_out, manifest) + except LiveControlValidationError as error: + result_status, result_code = "failed", error.code + except Exception as error: + result_status, result_code = "failed", "manifest_reconcile_failed" + + if result_status == "completed" and manifest is not None: + for case in ("identity", "mutation", "deletion", "loss"): + replay = replay_manifest(manifest, case) + report = reconcile_manifest(replay) + expected_reconciled = case == "identity" + replay_results[case] = "pass" if report["reconciled"] is expected_reconciled else "fail" + if replay_results[case] != "pass": + result_status, result_code = "failed", "identity_replay_incomplete" + break + if result_status == "completed": + for case in ("identity", "mutation", "deletion", "loss"): + replay_result = runner.run_subprocess( + f"replay_{case}", loaded["replays"][case]["argv"] + ) + if replay_result.status != "completed": + result_status, result_code = "failed", "identity_replay_incomplete" + break + except LiveControlValidationError as error: + result_status, result_code = "failed", error.code + except HarnessFailure as error: + result_status, result_code = "failed", error.code + finally: + try: + receipt = runner.cleanup() + except HarnessFailure: + receipt = { + "cleanup_attempted": True, + "cleanup_completed": False, + "cleanup_failure_count": 1, + "child_processes_terminated": False, + "resources_released": False, + } + if receipt.get("cleanup_completed") is not True: + result_status, result_code = "failed", "cleanup_incomplete" + + output: dict[str, Any] = { + "completed": result_status == "completed", + "ready_for_issue62": False, + "status": result_status, + "status_code": result_code, + "cleanup_completed": receipt.get("cleanup_completed") is True, + "manifest_reconciled": manifest_reconciled, + "replay": replay_results, + "reason": "independent_review_required", + } + if manifest is not None: + output["capture_manifest_sha256"] = manifest.get("capture_manifest_sha256") + return output + + def _fixture_identity(args: argparse.Namespace) -> CandidateIdentity: return CandidateIdentity( candidate_sha=args.candidate_sha, @@ -545,21 +1142,78 @@ def _fixture_identity(args: argparse.Namespace) -> CandidateIdentity: def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run-root", type=Path, required=True) - parser.add_argument( - "--fixture", choices=("success", "timeout", "nonzero", "cancelled", "cleanup-failure"), required=True + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument( + "--fixture", choices=("success", "timeout", "nonzero", "cancelled", "cleanup-failure") + ) + mode.add_argument( + "--enable-live-control", + action="store_true", + help="Run an explicitly supplied authorized live-control plan.", ) + parser.add_argument("--plan", type=Path, help="Sanitized live-control plan JSON") + parser.add_argument("--manifest-out", type=Path, help="Sanitized live manifest output") parser.add_argument("--timeout-seconds", type=float, default=DEFAULT_TIMEOUT_SECONDS) - parser.add_argument("--candidate-sha", required=True) - parser.add_argument("--cli-version", required=True) - parser.add_argument("--cli-package-sha256", required=True) - parser.add_argument("--catalog-digest", required=True) - parser.add_argument("--route-digest", required=True) + parser.add_argument("--candidate-sha") + parser.add_argument("--cli-version") + parser.add_argument("--cli-package-sha256") + parser.add_argument("--catalog-digest") + parser.add_argument("--route-digest") parser.add_argument("--cli-source-commit") return parser def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) + if args.enable_live_control: + if args.plan is None: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": "live_control_plan_missing", + "cleanup_completed": False, + "reason": "live_control_plan_missing", + } + print(json.dumps(result, sort_keys=True)) + return 2 + try: + result = run_live_control( + args.plan, + run_root=args.run_root, + timeout_seconds=args.timeout_seconds, + manifest_out=args.manifest_out, + ) + except LiveControlValidationError as error: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": error.code, + "cleanup_completed": False, + "reason": error.code, + } + except HarnessFailure as error: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": error.code, + "cleanup_completed": False, + "reason": error.code, + } + print(json.dumps(result, sort_keys=True)) + return 0 if result.get("completed") is True and result.get("cleanup_completed") is True else 2 + + fixture_fields = ( + "candidate_sha", + "cli_version", + "cli_package_sha256", + "catalog_digest", + "route_digest", + ) + if any(getattr(args, field) is None for field in fixture_fields): + raise SystemExit("fixture mode requires candidate identity arguments") identity = _fixture_identity(args) runner = BoundedPhaseRunner( args.run_root, diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py new file mode 100644 index 00000000..3dc3be99 --- /dev/null +++ b/tests/test_issue_62_live_control_orchestration.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +import hashlib +import importlib +import json +from pathlib import Path +import sys + +import pytest + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) + +from run_issue_62_live_control import ( # noqa: E402 + CONTROL_NAMES, + LiveControlValidationError, + load_live_control_plan, + main, + run_live_control, +) + + +def _sha(value: str) -> str: + return hashlib.sha256(value.encode("ascii")).hexdigest() + + +def _binding_files(tmp_path: Path) -> dict[str, str]: + files = { + "candidate_sha_file": tmp_path / "candidate-sha", + "cli_package_file": tmp_path / "cli-package.tgz", + "catalog_file": tmp_path / "catalog.json", + "route_file": tmp_path / "route.json", + } + files["candidate_sha_file"].write_text("a" * 40 + "\n", encoding="ascii") + files["cli_package_file"].write_bytes(b"package") + files["catalog_file"].write_bytes(b"catalog") + files["route_file"].write_bytes(b"route") + return {key: str(value) for key, value in files.items()} + + +def _plan(tmp_path: Path) -> dict[str, object]: + binding = _binding_files(tmp_path) + return { + "schema": "codexhub.issue62.live-control-plan.v1", + "verification_scope": "authorized_live_control", + "candidate_identity": { + "candidate_sha": "a" * 40, + "cli_version": "0.146.0", + "cli_package_sha256": _sha("package"), + "catalog_digest": _sha("catalog"), + "route_digest": _sha("route"), + }, + "binding": binding, + "catalog_model_entry_id": "gpt-5.6-sol", + "cli": {"argv": [sys.executable, "-c", "pass"]}, + "sidecars": { + "pre": {"argv": [sys.executable, "-c", "pass"], "output_dir": str(tmp_path / "pre")}, + "post": {"argv": [sys.executable, "-c", "pass"], "output_dir": str(tmp_path / "post")}, + }, + "controls": [ + {"name": name, "args": [], "capture": {}} + for name in CONTROL_NAMES + ], + "replays": { + "identity": {"argv": [sys.executable, "-c", "pass"]}, + "mutation": {"argv": [sys.executable, "-c", "pass"]}, + "deletion": {"argv": [sys.executable, "-c", "pass"]}, + "loss": {"argv": [sys.executable, "-c", "pass"]}, + }, + } + + +def test_missing_live_plan_uses_fixed_fail_closed_code(tmp_path: Path) -> None: + with pytest.raises(LiveControlValidationError, match="live_control_plan_missing"): + load_live_control_plan(tmp_path / "missing.json") + + +def test_cli_live_mode_reports_missing_plan_without_starting_children(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + assert main(["--run-root", str(tmp_path / "run"), "--enable-live-control"]) == 2 + output = json.loads(capsys.readouterr().out) + assert output["status_code"] == "live_control_plan_missing" + assert output["ready_for_issue62"] is False + + +def test_live_plan_binds_candidate_and_route_catalog_files(tmp_path: Path) -> None: + plan = _plan(tmp_path) + loaded = load_live_control_plan(plan) + assert loaded["candidate_identity"]["candidate_sha"] == "a" * 40 + + (tmp_path / "route.json").write_bytes(b"changed") + with pytest.raises(LiveControlValidationError, match="route_binding_mismatch"): + load_live_control_plan(plan) + + +def test_live_plan_requires_exactly_eight_control_labels(tmp_path: Path) -> None: + plan = _plan(tmp_path) + plan["controls"] = list(plan["controls"])[1:] + with pytest.raises(LiveControlValidationError, match="control_labels_incomplete"): + load_live_control_plan(plan) + + +def test_live_execution_requires_complete_pre_and_post_sidecar_records(tmp_path: Path) -> None: + result = run_live_control( + _plan(tmp_path), + run_root=tmp_path / "run", + timeout_seconds=30, + ) + assert result["ready_for_issue62"] is False + assert result["status_code"] == "sidecar_capture_missing" + receipt = json.loads((tmp_path / "run" / "cleanup-receipt.json").read_text(encoding="utf-8")) + assert receipt["cleanup_attempted"] is True + assert receipt["cleanup_completed"] is True + + +def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_path: Path) -> None: + source = importlib.import_module("test_issue_62_control_manifest") + plan = _plan(tmp_path) + controls = source._controls() + pre_dir = Path(plan["sidecars"]["pre"]["output_dir"]) + post_dir = Path(plan["sidecars"]["post"]["output_dir"]) + pre_dir.mkdir() + post_dir.mkdir() + payloads: dict[str, Path] = {} + for index, control in enumerate(controls): + pre = tmp_path / f"pre-{index}.json" + post = tmp_path / f"post-{index}.json" + pre.write_text(json.dumps(control["pre"]), encoding="utf-8") + post.write_text(json.dumps(control["post"]), encoding="utf-8") + payloads[f"pre-{index}"] = pre + payloads[f"post-{index}"] = post + semantic = { + key: value + for key, value in control.items() + if key not in {"name", "pre", "post"} + } + plan["controls"][index] = { + "name": control["name"], + "args": [], + "capture": semantic, + "pre_record": str(pre_dir / f"pre-c{index}.json"), + "post_record": str(post_dir / f"post-c{index}.json"), + } + sidecar_script = tmp_path / "write-sidecar.py" + sidecar_script.write_text( + "import pathlib, shutil, sys, time\n" + "out = pathlib.Path(sys.argv[1]); src = pathlib.Path(sys.argv[2])\n" + "shutil.copyfile(src, out / pathlib.Path(sys.argv[3]).name)\n" + "time.sleep(30)\n", + encoding="utf-8", + ) + for hop in ("pre", "post"): + specs = plan["sidecars"][hop] + source_paths = [payloads[f"{hop}-{index}"] for index in range(8)] + # One sidecar per control gives deterministic pairing without relying + # on opaque capture-id ordering. + plan["sidecars"][hop] = [ + { + "argv": [ + sys.executable, + str(sidecar_script), + str(Path(specs["output_dir"]).parent / f"{hop}-{index}"), + str(source_paths[index]), + f"{hop}-c{index}.json", + ], + "output_dir": str(Path(specs["output_dir"]).parent / f"{hop}-{index}"), + } + for index in range(8) + ] + for index in range(8): + plan["controls"][index][f"{hop}_record"] = str( + Path(plan["sidecars"][hop][index]["output_dir"]) / f"{hop}-c{index}.json" + ) + # Re-run plan validation after replacing the sidecar specs. + result = run_live_control(plan, run_root=tmp_path / "run", timeout_seconds=30) + assert result["completed"] is True + assert result["ready_for_issue62"] is False + assert result["replay"] == { + "identity": "pass", + "mutation": "pass", + "deletion": "pass", + "loss": "pass", + } From c75ba8eba4ce08d254fb1cb25c731cd50c8cea2a Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 05:48:20 +0800 Subject: [PATCH 02/12] test(issue62): harden live control evidence binding --- scripts/run_issue_62_live_control.py | 571 ++++++++++++++++-- ...est_issue_62_live_control_orchestration.py | 119 ++-- 2 files changed, 614 insertions(+), 76 deletions(-) diff --git a/scripts/run_issue_62_live_control.py b/scripts/run_issue_62_live_control.py index 494d7ef6..b9b4fb7d 100644 --- a/scripts/run_issue_62_live_control.py +++ b/scripts/run_issue_62_live_control.py @@ -33,6 +33,8 @@ from pathlib import Path import re import signal +import shutil +import stat import subprocess import sys import tempfile @@ -65,6 +67,14 @@ "live_control_plan_missing", "live_control_plan_invalid", "candidate_binding_mismatch", + "plan_path_invalid", + "plan_path_outside_isolation", + "plan_path_linked", + "environment_invalid", + "cli_executable_binding_mismatch", + "helper_executable_binding_mismatch", + "planner_plan_incomplete", + "planner_disposition_invalid", "candidate_sha_binding_mismatch", "cli_package_binding_mismatch", "catalog_binding_mismatch", @@ -73,6 +83,8 @@ "sidecar_capture_missing", "sidecar_capture_incomplete", "identity_replay_incomplete", + "identity_replay_artifact_missing", + "identity_replay_artifact_invalid", "manifest_reconcile_failed", } ) @@ -94,6 +106,7 @@ { "schema", "verification_scope", + "isolation_root", "candidate_identity", "binding", "catalog_model_entry_id", @@ -101,8 +114,31 @@ "sidecars", "controls", "replays", + "environment", + "planner", } ) +LIVE_DISPOSITIONS = frozenset({"Preserved", "Unsupported", "Unqualified"}) +LIVE_ENV_KEYS = frozenset( + {"PATH", "SystemRoot", "ComSpec", "TEMP", "TMP", "PATHEXT", "PYTHONPATH"} +) +SENSITIVE_ENV_KEYS = frozenset( + { + "CODEX_HOME", + "OPENAI_API_KEY", + "OLLAMA_API_KEY", + "CODEX_AUTH", + "HOME", + "USERPROFILE", + "HOMEDRIVE", + "HOMEPATH", + "APPDATA", + "LOCALAPPDATA", + } +) +PLANNER_FIELDS = frozenset( + {"model_visible_plan", "hosted_only_disposition", "unknown_tag_disposition"} +) class LiveControlValidationError(ValueError): @@ -329,12 +365,16 @@ def __init__( *, timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, process_factory: Callable[..., subprocess.Popen[bytes]] | None = None, + environment: Mapping[str, str] | None = None, + working_directory: Path | None = None, ) -> None: _ensure_fresh_run_root(run_root) self.run_root = run_root self.identity = identity self.timeout_seconds = _validate_timeout(timeout_seconds) self._process_factory = process_factory or subprocess.Popen + self._environment = _safe_environment(environment or {}) + self._working_directory = str(working_directory) if working_directory is not None else None self.journal = SanitizedPhaseJournal(run_root, identity) self._active_process: subprocess.Popen[bytes] | None = None self._active_lock = threading.Lock() @@ -386,6 +426,8 @@ def start_background(self, phase: str, argv: Sequence[str]) -> BackgroundProcess stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + env=self._environment, + cwd=self._working_directory, creationflags=creationflags, start_new_session=start_new_session, shell=False, @@ -511,6 +553,8 @@ def finish(result: CommandResult) -> CommandResult: stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + env=self._environment, + cwd=self._working_directory, creationflags=creationflags, start_new_session=start_new_session, shell=False, @@ -692,11 +736,143 @@ def _validate_plan_argv(value: Any, code: str = "live_control_plan_invalid") -> def _validate_plan_path(value: Any) -> str: + """Validate a relative plan path before it is joined to an isolated root.""" + if not isinstance(value, str) or not value or "\x00" in value: - _live_plan_fail("live_control_plan_invalid") + _live_plan_fail("plan_path_invalid") + candidate = Path(value) + # ``Path.is_absolute`` covers native Windows/POSIX roots; the explicit + # drive/UNC checks cover paths written for the other platform. + if candidate.is_absolute() or re.match(r"^(?:[A-Za-z]:|[\\/]{2})", value): + _live_plan_fail("plan_path_invalid") + if any(part in {"..", "."} for part in candidate.parts): + _live_plan_fail("plan_path_invalid") return value +def _is_reparse(path: Path) -> bool: + try: + mode = os.lstat(path).st_mode + if stat.S_ISLNK(mode): + return True + attributes = getattr(os.stat(path), "st_file_attributes", 0) + return bool(attributes & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)) + except OSError: + return False + + +def _resolve_isolated_path( + value: Any, + *, + isolated_root: Path, + must_exist: bool = False, + allow_missing_parents: bool = False, +) -> str: + relative = _validate_plan_path(value) + try: + root = isolated_root.resolve(strict=True) + except OSError: + _live_plan_fail("plan_path_outside_isolation") + if _is_reparse(root): + _live_plan_fail("plan_path_linked") + target = root / relative + try: + resolved = target.resolve(strict=must_exist) + resolved.relative_to(root) + except (OSError, ValueError): + _live_plan_fail("plan_path_outside_isolation") + current = root + parts = Path(relative).parts + for index, part in enumerate(parts): + current = current / part + exists = current.exists() + if not exists: + if (index != len(parts) - 1 and not allow_missing_parents) or must_exist: + _live_plan_fail("plan_path_outside_isolation") + continue + if _is_reparse(current): + _live_plan_fail("plan_path_linked") + try: + mode = os.lstat(current).st_mode + if current.is_file() and getattr(os.stat(current), "st_nlink", 1) > 1: + _live_plan_fail("plan_path_linked") + if stat.S_ISREG(mode) and index != len(parts) - 1: + _live_plan_fail("plan_path_outside_isolation") + except OSError: + _live_plan_fail("plan_path_outside_isolation") + return str(resolved) + + +def _validate_environment(value: Any) -> dict[str, str]: + if not isinstance(value, Mapping): + _live_plan_fail("environment_invalid") + normalized: dict[str, str] = {} + for key, item in value.items(): + if ( + not isinstance(key, str) + or key in SENSITIVE_ENV_KEYS + or key not in LIVE_ENV_KEYS + or not isinstance(item, str) + or "\x00" in item + or len(item) > 4096 + ): + _live_plan_fail("environment_invalid") + normalized[key] = item + return normalized + + +def _safe_environment(overrides: Mapping[str, str]) -> dict[str, str]: + """Build a minimal child environment without host credentials/home state.""" + + result: dict[str, str] = {} + for key in LIVE_ENV_KEYS: + if key in SENSITIVE_ENV_KEYS: + continue + value = os.environ.get(key) + if value: + result[key] = value + for key, value in overrides.items(): + result[key] = value + for key in SENSITIVE_ENV_KEYS: + result.pop(key, None) + return result + + +def _validate_executable_spec( + value: Any, + *, + isolated_root: Path | None, + error_code: str, + extra_fields: frozenset[str] = frozenset(), +) -> dict[str, Any]: + expected = frozenset({"argv", "executable_file", "executable_sha256"}) | extra_fields + if not isinstance(value, Mapping) or frozenset(value) != expected: + _live_plan_fail(error_code) + argv = _validate_plan_argv(value.get("argv")) + executable_file = _validate_plan_path(value.get("executable_file")) + digest = value.get("executable_sha256") + if not isinstance(digest, str) or re.fullmatch(r"[0-9a-fA-F]{64}", digest) is None: + _live_plan_fail(error_code) + if argv[0] != executable_file: + _live_plan_fail(error_code) + if isolated_root is not None: + resolved = _resolve_isolated_path( + executable_file, isolated_root=isolated_root, must_exist=True + ) + if _file_sha256(resolved) != digest.lower(): + _live_plan_fail(error_code) + return { + "argv": argv, + "executable_file": executable_file, + "executable_sha256": digest.lower(), + **{ + key: value[key] + for key in extra_fields + if key in value + }, + } + + def _file_sha256(path: str) -> str: try: target = Path(path) @@ -712,14 +888,25 @@ def _file_sha256(path: str) -> str: def _validate_live_binding( - identity: CandidateIdentity, binding: Mapping[str, Any] + identity: CandidateIdentity, + binding: Mapping[str, Any], + *, + isolated_root: Path | None, ) -> dict[str, str]: expected_fields = frozenset( {"candidate_sha_file", "cli_package_file", "catalog_file", "route_file"} ) if frozenset(binding) != expected_fields: _live_plan_fail("candidate_binding_mismatch") - paths = {key: _validate_plan_path(binding[key]) for key in expected_fields} + if isolated_root is None: + paths = {key: _validate_plan_path(binding[key]) for key in expected_fields} + else: + paths = { + key: _resolve_isolated_path( + binding[key], isolated_root=isolated_root, must_exist=True + ) + for key in expected_fields + } try: candidate_text = Path(paths["candidate_sha_file"]).read_text(encoding="ascii").strip() except (OSError, UnicodeError): @@ -737,16 +924,27 @@ def _validate_live_binding( return paths -def _validate_sidecar_spec(value: Any) -> dict[str, Any]: - if not isinstance(value, Mapping) or frozenset(value) != frozenset({"argv", "output_dir"}): - _live_plan_fail("live_control_plan_invalid") - return { - "argv": _validate_plan_argv(value.get("argv")), - "output_dir": _validate_plan_path(value.get("output_dir")), - } +def _validate_sidecar_spec(value: Any, *, isolated_root: Path | None) -> dict[str, Any]: + if not isinstance(value, Mapping) or frozenset(value) != frozenset( + {"argv", "executable_file", "executable_sha256", "output_dir"} + ): + _live_plan_fail("helper_executable_binding_mismatch") + executable = _validate_executable_spec( + {key: value[key] for key in ("argv", "executable_file", "executable_sha256")}, + isolated_root=isolated_root, + error_code="helper_executable_binding_mismatch", + ) + output_dir = _validate_plan_path(value.get("output_dir")) + if isolated_root is not None: + output_dir = _resolve_isolated_path( + output_dir, isolated_root=isolated_root, allow_missing_parents=True + ) + return {**executable, "output_dir": output_dir} -def _validate_sidecars(value: Any) -> dict[str, list[dict[str, Any]]]: +def _validate_sidecars( + value: Any, *, isolated_root: Path | None, run_root: Path | None +) -> dict[str, list[dict[str, Any]]]: if not isinstance(value, Mapping) or frozenset(value) != frozenset({"pre", "post"}): _live_plan_fail("live_control_plan_invalid") result: dict[str, list[dict[str, Any]]] = {} @@ -756,17 +954,34 @@ def _validate_sidecars(value: Any) -> dict[str, list[dict[str, Any]]]: specs = raw if isinstance(raw, list) else [raw] if not specs or len(specs) > len(CONTROL_NAMES): _live_plan_fail("live_control_plan_invalid") - normalized = [_validate_sidecar_spec(item) for item in specs] + normalized = [ + _validate_sidecar_spec(item, isolated_root=isolated_root) for item in specs + ] for spec in normalized: - directory = str(Path(spec["output_dir"]).resolve()) - if directory in output_dirs: + directory_path = Path(spec["output_dir"]) + if run_root is not None: + try: + directory_path.resolve().relative_to(run_root.resolve()) + except ValueError: + _live_plan_fail("plan_path_outside_isolation") + directory = str(directory_path.resolve()) + if directory in output_dirs or any( + Path(directory).is_relative_to(Path(existing)) + or Path(existing).is_relative_to(Path(directory)) + for existing in output_dirs + ): _live_plan_fail("live_control_plan_invalid") output_dirs.add(directory) result[hop] = normalized return result -def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, Any]: +def load_live_control_plan( + source: Path | str | Mapping[str, Any], + *, + isolated_root: Path | None = None, + run_root: Path | None = None, +) -> dict[str, Any]: """Load and independently bind an explicitly authorized live plan. This function performs no subprocess or network work. It is deliberately @@ -788,6 +1003,13 @@ def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, _live_plan_fail("live_control_plan_invalid") if payload.get("schema") != LIVE_PLAN_SCHEMA or payload.get("verification_scope") != LIVE_SCOPE: _live_plan_fail("live_control_plan_invalid") + if payload.get("isolation_root") != ".": + _live_plan_fail("plan_path_invalid") + if isolated_root is not None: + try: + isolated_root = isolated_root.resolve(strict=True) + except OSError: + _live_plan_fail("plan_path_outside_isolation") raw_identity = payload.get("candidate_identity") if not isinstance(raw_identity, Mapping): @@ -812,7 +1034,9 @@ def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, binding = payload.get("binding") if not isinstance(binding, Mapping): _live_plan_fail("candidate_binding_mismatch") - normalized_binding = _validate_live_binding(identity, binding) + normalized_binding = _validate_live_binding( + identity, binding, isolated_root=isolated_root + ) model = payload.get("catalog_model_entry_id") if ( @@ -823,16 +1047,39 @@ def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, _live_plan_fail("catalog_binding_mismatch") cli = payload.get("cli") - if not isinstance(cli, Mapping) or frozenset(cli) != frozenset({"argv"}): - _live_plan_fail("live_control_plan_invalid") - normalized_cli = {"argv": _validate_plan_argv(cli.get("argv"))} - normalized_sidecars = _validate_sidecars(payload.get("sidecars")) + if not isinstance(cli, Mapping) or frozenset(cli) != frozenset( + {"argv", "executable_file", "executable_sha256", "cli_version"} + ): + _live_plan_fail("cli_executable_binding_mismatch") + if cli.get("cli_version") != identity.cli_version: + _live_plan_fail("cli_executable_binding_mismatch") + normalized_cli = _validate_executable_spec( + cli, + isolated_root=isolated_root, + error_code="cli_executable_binding_mismatch", + extra_fields=frozenset({"cli_version"}), + ) + normalized_cli["cli_version"] = cli["cli_version"] + normalized_sidecars = _validate_sidecars( + payload.get("sidecars"), isolated_root=isolated_root, run_root=run_root + ) + normalized_environment = _validate_environment(payload.get("environment")) + planner = payload.get("planner") + if not isinstance(planner, Mapping) or frozenset(planner) != PLANNER_FIELDS: + _live_plan_fail("planner_plan_incomplete") + if planner.get("model_visible_plan") != "complete": + _live_plan_fail("planner_plan_incomplete") + if planner.get("hosted_only_disposition") not in LIVE_DISPOSITIONS or planner.get( + "unknown_tag_disposition" + ) not in LIVE_DISPOSITIONS: + _live_plan_fail("planner_disposition_invalid") controls = payload.get("controls") if not isinstance(controls, list) or len(controls) != len(CONTROL_NAMES): _live_plan_fail("control_labels_incomplete") normalized_controls: list[dict[str, Any]] = [] names: list[str] = [] + record_paths: dict[str, set[str]] = {"pre": set(), "post": set()} allowed_control_fields = frozenset({"name", "args", "capture", "pre_record", "post_record"}) for item in controls: if not isinstance(item, Mapping) or not frozenset(item).issubset(allowed_control_fields): @@ -853,7 +1100,16 @@ def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, _live_plan_fail("live_control_plan_invalid") for field in ("pre_record", "post_record"): if field in item: - normalized[field] = _validate_plan_path(item[field]) + if isolated_root is None: + normalized[field] = _validate_plan_path(item[field]) + else: + normalized[field] = _resolve_isolated_path( + item[field], isolated_root=isolated_root, allow_missing_parents=True + ) + normalized_path = str(Path(normalized[field]).resolve()) + if normalized_path in record_paths[field.removesuffix("_record")]: + _live_plan_fail("sidecar_capture_incomplete") + record_paths[field.removesuffix("_record")].add(normalized_path) normalized_controls.append(normalized) if set(names) != CONTROL_NAME_SET or len(set(names)) != len(names): _live_plan_fail("control_labels_incomplete") @@ -865,13 +1121,39 @@ def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, normalized_replays: dict[str, dict[str, Any]] = {} for name in replay_names: spec = replays.get(name) - if not isinstance(spec, Mapping) or frozenset(spec) != frozenset({"argv"}): + if not isinstance(spec, Mapping) or frozenset(spec) != frozenset( + { + "argv", + "executable_file", + "executable_sha256", + "artifact_file", + "case", + } + ) or spec.get("case") != name: _live_plan_fail("identity_replay_incomplete") - normalized_replays[name] = {"argv": _validate_plan_argv(spec.get("argv"))} + replay = _validate_executable_spec( + spec, + isolated_root=isolated_root, + error_code="helper_executable_binding_mismatch", + extra_fields=frozenset({"artifact_file", "case"}), + ) + artifact_file = _validate_plan_path(replay.get("artifact_file")) + if isolated_root is not None: + artifact_file = _resolve_isolated_path( + artifact_file, isolated_root=isolated_root, allow_missing_parents=True + ) + if run_root is not None: + try: + Path(artifact_file).resolve().relative_to(run_root.resolve()) + except ValueError: + _live_plan_fail("plan_path_outside_isolation") + replay["artifact_file"] = artifact_file + normalized_replays[name] = replay return { "schema": LIVE_PLAN_SCHEMA, "verification_scope": LIVE_SCOPE, + "isolation_root": ".", "candidate_identity": identity.as_dict(), "binding": normalized_binding, "catalog_model_entry_id": model, @@ -879,6 +1161,8 @@ def load_live_control_plan(source: Path | str | Mapping[str, Any]) -> dict[str, "sidecars": normalized_sidecars, "controls": normalized_controls, "replays": normalized_replays, + "environment": normalized_environment, + "planner": dict(planner), } @@ -919,6 +1203,20 @@ def _record_path_for_control( specs: Sequence[Mapping[str, Any]], index: int, ) -> Path: + def capture_files(root: Path) -> list[Path]: + try: + if root.is_symlink() or not root.is_dir(): + raise OSError + entries = list(root.iterdir()) + except OSError: + raise LiveControlValidationError("sidecar_capture_missing") + if any( + entry.is_symlink() or entry.is_dir() or entry.suffix.lower() != ".json" + for entry in entries + ): + raise LiveControlValidationError("sidecar_capture_incomplete") + return sorted(entry for entry in entries if entry.is_file()) + explicit = control.get(f"{hop}_record") if explicit is not None: path = Path(str(explicit)) @@ -928,17 +1226,19 @@ def _record_path_for_control( path.resolve().relative_to(root.resolve()) except ValueError: continue + if path.resolve() not in {candidate.resolve() for candidate in capture_files(root)}: + raise LiveControlValidationError("sidecar_capture_missing") return path raise LiveControlValidationError("sidecar_capture_missing") if len(specs) == len(CONTROL_NAMES): root = Path(str(specs[index]["output_dir"])) - candidates = sorted(path for path in root.glob("*.json") if path.is_file()) + candidates = capture_files(root) if len(candidates) == 1: return candidates[0] raise LiveControlValidationError("sidecar_capture_missing") if len(specs) == 1: root = Path(str(specs[0]["output_dir"])) - candidates = sorted(path for path in root.glob("*.json") if path.is_file()) + candidates = capture_files(root) if len(candidates) == len(CONTROL_NAMES): return candidates[index] raise LiveControlValidationError("sidecar_capture_missing") @@ -1003,12 +1303,115 @@ def _manifest_candidate(plan: Mapping[str, Any]) -> dict[str, Any]: } +def _runtime_argv(spec: Mapping[str, Any], *, isolated_root: Path) -> list[str]: + executable = _resolve_isolated_path( + spec["executable_file"], isolated_root=isolated_root, must_exist=True + ) + argv = list(spec["argv"]) + argv[0] = executable + return argv + + +def _remove_capture_dirs( + sidecars: Mapping[str, Sequence[Mapping[str, Any]]], *, run_root: Path +) -> bool: + failures = 0 + root = run_root.resolve() + for specs in sidecars.values(): + for spec in specs: + target = Path(str(spec["output_dir"])).resolve() + try: + target.relative_to(root) + if target == root: + raise OSError + if target.is_symlink(): + target.unlink() + elif target.exists(): + shutil.rmtree(target) + except (OSError, ValueError): + failures += 1 + return failures == 0 + + +def _remove_replay_artifacts( + replays: Mapping[str, Mapping[str, Any]], *, run_root: Path +) -> bool: + failures = 0 + root = run_root.resolve() + for replay in replays.values(): + target = Path(str(replay["artifact_file"])).resolve() + try: + target.relative_to(root) + if target.is_symlink(): + target.unlink() + elif target.exists(): + target.unlink() + except (OSError, ValueError): + failures += 1 + return failures == 0 + + +def _ensure_replay_artifacts_fresh( + replays: Mapping[str, Mapping[str, Any]], *, run_root: Path +) -> None: + """Reject a reused replay output before any child process starts.""" + + root = run_root.resolve() + seen: set[Path] = set() + for replay in replays.values(): + target = Path(str(replay["artifact_file"])).resolve() + try: + target.relative_to(root) + except ValueError: + raise LiveControlValidationError("plan_path_outside_isolation") + if target in seen or target.exists() or target.is_symlink(): + raise LiveControlValidationError("stale_artifact") + seen.add(target) + + +def _validate_replay_artifact( + path: Path, + *, + case: str, + candidate_sha: str, + manifest_sha: str, +) -> str: + # Diagnostic kept path-free; only existence is observed. + # (The caller redacts all path values from journal artifacts.) + try: + if path.is_symlink() or not path.is_file(): + raise OSError + artifact_sha = _file_sha256(str(path)) + if re.fullmatch(r"[0-9a-f]{64}", artifact_sha) is None: + raise OSError + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + raise LiveControlValidationError("identity_replay_artifact_missing") + expected_fields = frozenset( + {"schema", "case", "candidate_sha", "capture_manifest_sha256", "wire_replay", "outcome"} + ) + if not isinstance(value, Mapping) or frozenset(value) != expected_fields: + raise LiveControlValidationError("identity_replay_artifact_invalid") + if ( + value.get("schema") != "codexhub.issue62.identity-replay.v1" + or value.get("case") != case + or value.get("candidate_sha") != candidate_sha + or value.get("capture_manifest_sha256") != manifest_sha + or value.get("wire_replay") is not True + or value.get("outcome") != ("accepted" if case == "identity" else "rejected") + ): + raise LiveControlValidationError("identity_replay_artifact_invalid") + return artifact_sha + + def run_live_control( plan: Path | str | Mapping[str, Any], *, run_root: Path, timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, manifest_out: Path | None = None, + isolated_root: Path | None = None, + cancellation_event: threading.Event | None = None, ) -> dict[str, Any]: """Execute an explicitly supplied live plan through bounded children. @@ -1017,33 +1420,100 @@ def run_live_control( review is still required before any Issue #62 claim can be made. """ - loaded = load_live_control_plan(plan) + try: + isolation = (isolated_root or Path(run_root).parent).resolve(strict=True) + if isinstance(plan, (str, Path)): + plan_path = Path(plan).resolve(strict=True) + plan_path.relative_to(isolation) + run_path = Path(run_root) + if not run_path.is_absolute(): + run_path = isolation / run_path + run_path = run_path.resolve(strict=False) + run_path.relative_to(isolation) + if run_path == isolation or _is_reparse(run_path): + raise LiveControlValidationError("plan_path_outside_isolation") + except (OSError, ValueError): + raise LiveControlValidationError("plan_path_outside_isolation") + loaded = load_live_control_plan( + plan, isolated_root=isolation, run_root=run_path + ) identity = CandidateIdentity(**loaded["candidate_identity"]) - runner = BoundedPhaseRunner(run_root, identity, timeout_seconds=timeout_seconds) + output_manifest: Path | None = None + if manifest_out is not None: + output_manifest = Path(manifest_out) + if not output_manifest.is_absolute(): + output_manifest = isolation / output_manifest + output_manifest = output_manifest.resolve(strict=False) + try: + output_manifest.relative_to(isolation) + except ValueError: + raise LiveControlValidationError("plan_path_outside_isolation") + try: + output_manifest.relative_to(run_path) + except ValueError: + pass + else: + raise LiveControlValidationError("plan_path_outside_isolation") + runner = BoundedPhaseRunner( + run_path, + identity, + timeout_seconds=timeout_seconds, + environment=loaded["environment"], + working_directory=isolation, + ) handles: list[BackgroundProcess] = [] result_status = "completed" result_code = "ok" manifest: dict[str, Any] | None = None manifest_reconciled = False replay_results: dict[str, str] = {} + cleanup_actions: tuple[CleanupAction, ...] = ( + lambda: _remove_capture_dirs(loaded["sidecars"], run_root=run_path), + lambda: _remove_replay_artifacts(loaded["replays"], run_root=run_path), + ) + + def is_cancelled() -> bool: + return cancellation_event is not None and cancellation_event.is_set() + try: runner.mark_phase("binding_validated") - _prepare_capture_dirs(loaded["sidecars"]) - for hop in ("pre", "post"): - for index, spec in enumerate(loaded["sidecars"][hop]): - try: - handles.append(runner.start_background(f"{hop}_sidecar_{index}", spec["argv"])) - except HarnessFailure as error: - result_status, result_code = "failed", error.code + _ensure_replay_artifacts_fresh(loaded["replays"], run_root=run_path) + if is_cancelled(): + result_status, result_code = "cancelled", "cancelled" + else: + _prepare_capture_dirs(loaded["sidecars"]) + for hop in ("pre", "post"): + for index, spec in enumerate(loaded["sidecars"][hop]): + try: + handles.append( + runner.start_background( + f"{hop}_sidecar_{index}", + _runtime_argv(spec, isolated_root=isolation), + ) + ) + except HarnessFailure as error: + result_status, result_code = "failed", error.code + break + if result_status != "completed": break - if result_status != "completed": - break + + if result_status == "completed" and any( + handle.process.poll() is not None for handle in handles + ): + result_status, result_code = "failed", "sidecar_capture_incomplete" if result_status == "completed": for control in loaded["controls"]: + if is_cancelled(): + result_status, result_code = "cancelled", "cancelled" + break result = runner.run_subprocess( f"control_{control['name']}", - [*loaded["cli"]["argv"], *control["args"]], + [ + *_runtime_argv(loaded["cli"], isolated_root=isolation), + *control["args"], + ], + cancellation_event=cancellation_event, ) if result.status != "completed": result_status, result_code = result.status, result.status_code @@ -1071,8 +1541,8 @@ def run_live_control( result_status, result_code = "failed", "manifest_reconcile_failed" else: manifest_reconciled = True - if manifest_out is not None: - _write_sanitized_json(manifest_out, manifest) + if output_manifest is not None: + _write_sanitized_json(output_manifest, manifest) except LiveControlValidationError as error: result_status, result_code = "failed", error.code except Exception as error: @@ -1089,19 +1559,35 @@ def run_live_control( break if result_status == "completed": for case in ("identity", "mutation", "deletion", "loss"): + if is_cancelled(): + result_status, result_code = "cancelled", "cancelled" + break replay_result = runner.run_subprocess( - f"replay_{case}", loaded["replays"][case]["argv"] + f"replay_{case}", + _runtime_argv(loaded["replays"][case], isolated_root=isolation), + cancellation_event=cancellation_event, ) if replay_result.status != "completed": result_status, result_code = "failed", "identity_replay_incomplete" break + replay = loaded["replays"][case] + artifact_sha = _validate_replay_artifact( + Path(replay["artifact_file"]), + case=case, + candidate_sha=identity.candidate_sha, + manifest_sha=str(manifest["capture_manifest_sha256"]), + ) + replay_results[case] = f"artifact_bound:{artifact_sha}" except LiveControlValidationError as error: result_status, result_code = "failed", error.code except HarnessFailure as error: result_status, result_code = "failed", error.code + except Exception: + # Keep unexpected manifest/replay errors fail-closed and path-free. + result_status, result_code = "failed", "phase_exception" finally: try: - receipt = runner.cleanup() + receipt = runner.cleanup(actions=cleanup_actions) except HarnessFailure: receipt = { "cleanup_attempted": True, @@ -1122,6 +1608,7 @@ def run_live_control( "manifest_reconciled": manifest_reconciled, "replay": replay_results, "reason": "independent_review_required", + "planner": loaded["planner"], } if manifest is not None: output["capture_manifest_sha256"] = manifest.get("capture_manifest_sha256") @@ -1142,6 +1629,7 @@ def _fixture_identity(args: argparse.Namespace) -> CandidateIdentity: def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--run-root", type=Path, required=True) + parser.add_argument("--isolated-root", type=Path, help="Fresh-root parent for relative plan paths") mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument( "--fixture", choices=("success", "timeout", "nonzero", "cancelled", "cleanup-failure") @@ -1183,6 +1671,7 @@ def main(argv: Sequence[str] | None = None) -> int: run_root=args.run_root, timeout_seconds=args.timeout_seconds, manifest_out=args.manifest_out, + isolated_root=args.isolated_root, ) except LiveControlValidationError as error: result = { diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py index 3dc3be99..1156a906 100644 --- a/tests/test_issue_62_live_control_orchestration.py +++ b/tests/test_issue_62_live_control_orchestration.py @@ -4,6 +4,7 @@ import importlib import json from pathlib import Path +import shutil import sys import pytest @@ -28,24 +29,46 @@ def _sha(value: str) -> str: def _binding_files(tmp_path: Path) -> dict[str, str]: + (tmp_path / "inputs").mkdir() files = { - "candidate_sha_file": tmp_path / "candidate-sha", - "cli_package_file": tmp_path / "cli-package.tgz", - "catalog_file": tmp_path / "catalog.json", - "route_file": tmp_path / "route.json", + "candidate_sha_file": tmp_path / "inputs" / "candidate-sha", + "cli_package_file": tmp_path / "inputs" / "cli-package.tgz", + "catalog_file": tmp_path / "inputs" / "catalog.json", + "route_file": tmp_path / "inputs" / "route.json", } files["candidate_sha_file"].write_text("a" * 40 + "\n", encoding="ascii") files["cli_package_file"].write_bytes(b"package") files["catalog_file"].write_bytes(b"catalog") files["route_file"].write_bytes(b"route") - return {key: str(value) for key, value in files.items()} + return {key: str(value.relative_to(tmp_path)) for key, value in files.items()} def _plan(tmp_path: Path) -> dict[str, object]: binding = _binding_files(tmp_path) + (tmp_path / "helpers").mkdir() + (tmp_path / "run").mkdir() + python_copy = tmp_path / "tools" / "python.exe" + python_copy.parent.mkdir() + shutil.copy2(sys.executable, python_copy) + (tmp_path / "helpers" / "cli.py").write_text("raise SystemExit(0)\n", encoding="utf-8") + (tmp_path / "helpers" / "replay.py").write_text( + "import json, pathlib, sys\n" + "case, candidate, manifest, output = sys.argv[1:5]\n" + "path = pathlib.Path(output)\n" + "path.parent.mkdir(parents=True, exist_ok=True)\n" + "path.write_text(json.dumps({'schema':'codexhub.issue62.identity-replay.v1', 'case':case, 'candidate_sha':candidate, 'capture_manifest_sha256':manifest, 'wire_replay':True, 'outcome':'accepted' if case == 'identity' else 'rejected'}, sort_keys=True, separators=(',', ':')) + '\\n', encoding='utf-8')\n", + encoding="utf-8", + ) + executable_digest = hashlib.sha256(python_copy.read_bytes()).hexdigest() + executable_file = str(python_copy.relative_to(tmp_path)) + executable = { + "executable_file": executable_file, + "executable_sha256": executable_digest, + } return { "schema": "codexhub.issue62.live-control-plan.v1", "verification_scope": "authorized_live_control", + "isolation_root": ".", "candidate_identity": { "candidate_sha": "a" * 40, "cli_version": "0.146.0", @@ -55,20 +78,29 @@ def _plan(tmp_path: Path) -> dict[str, object]: }, "binding": binding, "catalog_model_entry_id": "gpt-5.6-sol", - "cli": {"argv": [sys.executable, "-c", "pass"]}, + "environment": {"PATH": str(Path(sys.executable).parent)}, + "planner": { + "model_visible_plan": "complete", + "hosted_only_disposition": "Unqualified", + "unknown_tag_disposition": "Unqualified", + }, + "cli": {"argv": [executable_file, "helpers/cli.py"], **executable, "cli_version": "0.146.0"}, "sidecars": { - "pre": {"argv": [sys.executable, "-c", "pass"], "output_dir": str(tmp_path / "pre")}, - "post": {"argv": [sys.executable, "-c", "pass"], "output_dir": str(tmp_path / "post")}, + "pre": {"argv": [executable_file, "helpers/cli.py"], "output_dir": "run/pre", **executable}, + "post": {"argv": [executable_file, "helpers/cli.py"], "output_dir": "run/post", **executable}, }, "controls": [ {"name": name, "args": [], "capture": {}} for name in CONTROL_NAMES ], "replays": { - "identity": {"argv": [sys.executable, "-c", "pass"]}, - "mutation": {"argv": [sys.executable, "-c", "pass"]}, - "deletion": {"argv": [sys.executable, "-c", "pass"]}, - "loss": {"argv": [sys.executable, "-c", "pass"]}, + case: { + "argv": [executable_file, "helpers/replay.py", case, "a" * 40, "0" * 64, f"run/replay-{case}.json"], + **executable, + "artifact_file": f"run/replay-{case}.json", + "case": case, + } + for case in ("identity", "mutation", "deletion", "loss") }, } @@ -87,19 +119,19 @@ def test_cli_live_mode_reports_missing_plan_without_starting_children(tmp_path: def test_live_plan_binds_candidate_and_route_catalog_files(tmp_path: Path) -> None: plan = _plan(tmp_path) - loaded = load_live_control_plan(plan) + loaded = load_live_control_plan(plan, isolated_root=tmp_path) assert loaded["candidate_identity"]["candidate_sha"] == "a" * 40 - (tmp_path / "route.json").write_bytes(b"changed") + (tmp_path / "inputs" / "route.json").write_bytes(b"changed") with pytest.raises(LiveControlValidationError, match="route_binding_mismatch"): - load_live_control_plan(plan) + load_live_control_plan(plan, isolated_root=tmp_path) def test_live_plan_requires_exactly_eight_control_labels(tmp_path: Path) -> None: plan = _plan(tmp_path) plan["controls"] = list(plan["controls"])[1:] with pytest.raises(LiveControlValidationError, match="control_labels_incomplete"): - load_live_control_plan(plan) + load_live_control_plan(plan, isolated_root=tmp_path) def test_live_execution_requires_complete_pre_and_post_sidecar_records(tmp_path: Path) -> None: @@ -119,8 +151,8 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa source = importlib.import_module("test_issue_62_control_manifest") plan = _plan(tmp_path) controls = source._controls() - pre_dir = Path(plan["sidecars"]["pre"]["output_dir"]) - post_dir = Path(plan["sidecars"]["post"]["output_dir"]) + pre_dir = tmp_path / "inputs" / "pre-records" + post_dir = tmp_path / "inputs" / "post-records" pre_dir.mkdir() post_dir.mkdir() payloads: dict[str, Path] = {} @@ -140,10 +172,10 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa "name": control["name"], "args": [], "capture": semantic, - "pre_record": str(pre_dir / f"pre-c{index}.json"), - "post_record": str(post_dir / f"post-c{index}.json"), + "pre_record": f"run/pre-{index}/pre-c{index}.json", + "post_record": f"run/post-{index}/post-c{index}.json", } - sidecar_script = tmp_path / "write-sidecar.py" + sidecar_script = tmp_path / "helpers" / "sidecar.py" sidecar_script.write_text( "import pathlib, shutil, sys, time\n" "out = pathlib.Path(sys.argv[1]); src = pathlib.Path(sys.argv[2])\n" @@ -159,27 +191,44 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa plan["sidecars"][hop] = [ { "argv": [ - sys.executable, - str(sidecar_script), - str(Path(specs["output_dir"]).parent / f"{hop}-{index}"), - str(source_paths[index]), + plan["cli"]["executable_file"], + "helpers/sidecar.py", + f"run/{hop}-{index}", + str(source_paths[index].relative_to(tmp_path)), f"{hop}-c{index}.json", ], - "output_dir": str(Path(specs["output_dir"]).parent / f"{hop}-{index}"), + "output_dir": f"run/{hop}-{index}", + "executable_file": plan["cli"]["executable_file"], + "executable_sha256": plan["cli"]["executable_sha256"], } for index in range(8) ] for index in range(8): - plan["controls"][index][f"{hop}_record"] = str( - Path(plan["sidecars"][hop][index]["output_dir"]) / f"{hop}-c{index}.json" + plan["controls"][index][f"{hop}_record"] = ( + f"run/{hop}-{index}/{hop}-c{index}.json" ) # Re-run plan validation after replacing the sidecar specs. result = run_live_control(plan, run_root=tmp_path / "run", timeout_seconds=30) - assert result["completed"] is True + assert result["completed"] is False assert result["ready_for_issue62"] is False - assert result["replay"] == { - "identity": "pass", - "mutation": "pass", - "deletion": "pass", - "loss": "pass", - } + assert result["status_code"] == "identity_replay_artifact_invalid" + assert result["cleanup_completed"] is True + assert result["manifest_reconciled"] is True + assert not (tmp_path / "run" / "pre-0").exists() + assert not (tmp_path / "run" / "post-0").exists() + assert not (tmp_path / "run" / "replay-identity.json").exists() + + +def test_live_plan_rejects_host_credentials_and_duplicate_capture_paths(tmp_path: Path) -> None: + plan = _plan(tmp_path) + plan["environment"] = {"OPENAI_API_KEY": "must-not-pass"} + with pytest.raises(LiveControlValidationError, match="environment_invalid"): + load_live_control_plan(plan, isolated_root=tmp_path) + + duplicate_root = tmp_path / "duplicate" + duplicate_root.mkdir() + plan = _plan(duplicate_root) + plan["controls"][1]["pre_record"] = "run/shared.json" + plan["controls"][0]["pre_record"] = "run/shared.json" + with pytest.raises(LiveControlValidationError, match="sidecar_capture_incomplete"): + load_live_control_plan(plan, isolated_root=duplicate_root) From 4ace0f3ad046b1d1974225716a18dcc9cd2bef0e Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 06:06:23 +0800 Subject: [PATCH 03/12] test(issue62): harden isolated evidence runner --- scripts/run_issue_62_live_control.py | 224 ++++++++++++++---- ...est_issue_62_live_control_orchestration.py | 49 +++- 2 files changed, 221 insertions(+), 52 deletions(-) diff --git a/scripts/run_issue_62_live_control.py b/scripts/run_issue_62_live_control.py index b9b4fb7d..325a9bbc 100644 --- a/scripts/run_issue_62_live_control.py +++ b/scripts/run_issue_62_live_control.py @@ -119,11 +119,12 @@ } ) LIVE_DISPOSITIONS = frozenset({"Preserved", "Unsupported", "Unqualified"}) -LIVE_ENV_KEYS = frozenset( - {"PATH", "SystemRoot", "ComSpec", "TEMP", "TMP", "PATHEXT", "PYTHONPATH"} -) +LIVE_ENV_KEYS = frozenset({"SystemRoot", "ComSpec"}) SENSITIVE_ENV_KEYS = frozenset( { + "PATH", + "PATHEXT", + "PYTHONPATH", "CODEX_HOME", "OPENAI_API_KEY", "OLLAMA_API_KEY", @@ -282,6 +283,7 @@ def _count_extra_artifacts(run_root: Path) -> int: entry.name in ALLOWED_ARTIFACT_NAMES and entry.is_file() and not entry.is_symlink() + and os.stat(entry).st_nlink == 1 ): continue extras += 1 @@ -322,12 +324,14 @@ def _write_sanitized_json(target: Path, payload: dict[str, object]) -> None: def _terminate_process(process: subprocess.Popen[bytes]) -> bool: try: if process.poll() is not None: - return True + # A completed parent can still have a descendant. Treat that + # state as unknown rather than claiming the process tree is gone. + return False if os.name == "nt": # ``terminate`` only signals the direct child on Windows. The # Gateway/CLI can spawn grandchildren, so use taskkill's tree mode # and suppress all command output. - subprocess.run( + kill_result = subprocess.run( ["taskkill", "/PID", str(int(process.pid)), "/T", "/F"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, @@ -335,6 +339,8 @@ def _terminate_process(process: subprocess.Popen[bytes]) -> bool: check=False, timeout=5, ) + if kill_result.returncode != 0: + raise OSError("taskkill failed") else: os.killpg(int(process.pid), signal.SIGTERM) process.wait(timeout=5) @@ -428,6 +434,7 @@ def start_background(self, phase: str, argv: Sequence[str]) -> BackgroundProcess stderr=subprocess.DEVNULL, env=self._environment, cwd=self._working_directory, + close_fds=True, creationflags=creationflags, start_new_session=start_new_session, shell=False, @@ -555,6 +562,7 @@ def finish(result: CommandResult) -> CommandResult: stderr=subprocess.DEVNULL, env=self._environment, cwd=self._working_directory, + close_fds=True, creationflags=creationflags, start_new_session=start_new_session, shell=False, @@ -727,7 +735,13 @@ def _live_plan_fail(code: str) -> NoReturn: def _validate_plan_argv(value: Any, code: str = "live_control_plan_invalid") -> list[str]: if not isinstance(value, list) or not value or any( - not isinstance(item, str) or not item or "\x00" in item for item in value + not isinstance(item, str) + or not item + or "\x00" in item + or Path(item).is_absolute() + or re.match(r"^(?:[A-Za-z]:|[\\/]{2})", item) is not None + or ".." in Path(item).parts + for item in value ): _live_plan_fail(code) # Return a fresh list so callers cannot mutate the source object while a @@ -752,10 +766,11 @@ def _validate_plan_path(value: Any) -> str: def _is_reparse(path: Path) -> bool: try: - mode = os.lstat(path).st_mode + stat_result = os.lstat(path) + mode = stat_result.st_mode if stat.S_ISLNK(mode): return True - attributes = getattr(os.stat(path), "st_file_attributes", 0) + attributes = getattr(stat_result, "st_file_attributes", 0) return bool(attributes & getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0x400)) except OSError: return False @@ -825,9 +840,7 @@ def _safe_environment(overrides: Mapping[str, str]) -> dict[str, str]: """Build a minimal child environment without host credentials/home state.""" result: dict[str, str] = {} - for key in LIVE_ENV_KEYS: - if key in SENSITIVE_ENV_KEYS: - continue + for key in ("SystemRoot", "ComSpec"): value = os.environ.get(key) if value: result[key] = value @@ -845,7 +858,9 @@ def _validate_executable_spec( error_code: str, extra_fields: frozenset[str] = frozenset(), ) -> dict[str, Any]: - expected = frozenset({"argv", "executable_file", "executable_sha256"}) | extra_fields + expected = frozenset( + {"argv", "executable_file", "executable_sha256", "argv_file_digests"} + ) | extra_fields if not isinstance(value, Mapping) or frozenset(value) != expected: _live_plan_fail(error_code) argv = _validate_plan_argv(value.get("argv")) @@ -855,6 +870,22 @@ def _validate_executable_spec( _live_plan_fail(error_code) if argv[0] != executable_file: _live_plan_fail(error_code) + raw_files = value.get("argv_file_digests") + if not isinstance(raw_files, Mapping): + _live_plan_fail(error_code) + file_digests: dict[str, str] = {} + for path, file_digest in raw_files.items(): + if not isinstance(path, str) or not isinstance(file_digest, str): + _live_plan_fail(error_code) + if re.fullmatch(r"[0-9a-fA-F]{64}", file_digest) is None: + _live_plan_fail(error_code) + if isolated_root is not None: + resolved_file = _resolve_isolated_path( + path, isolated_root=isolated_root, must_exist=True + ) + if _file_sha256(resolved_file) != file_digest.lower(): + _live_plan_fail(error_code) + file_digests[path] = file_digest.lower() if isolated_root is not None: resolved = _resolve_isolated_path( executable_file, isolated_root=isolated_root, must_exist=True @@ -865,6 +896,7 @@ def _validate_executable_spec( "argv": argv, "executable_file": executable_file, "executable_sha256": digest.lower(), + "argv_file_digests": file_digests, **{ key: value[key] for key in extra_fields @@ -876,7 +908,7 @@ def _validate_executable_spec( def _file_sha256(path: str) -> str: try: target = Path(path) - if target.is_symlink() or not target.is_file(): + if target.is_symlink() or not target.is_file() or os.stat(target).st_nlink != 1: raise OSError digest = hashlib.sha256() with target.open("rb") as handle: @@ -926,11 +958,25 @@ def _validate_live_binding( def _validate_sidecar_spec(value: Any, *, isolated_root: Path | None) -> dict[str, Any]: if not isinstance(value, Mapping) or frozenset(value) != frozenset( - {"argv", "executable_file", "executable_sha256", "output_dir"} + { + "argv", + "executable_file", + "executable_sha256", + "argv_file_digests", + "output_dir", + } ): _live_plan_fail("helper_executable_binding_mismatch") executable = _validate_executable_spec( - {key: value[key] for key in ("argv", "executable_file", "executable_sha256")}, + { + key: value[key] + for key in ( + "argv", + "executable_file", + "executable_sha256", + "argv_file_digests", + ) + }, isolated_root=isolated_root, error_code="helper_executable_binding_mismatch", ) @@ -991,7 +1037,12 @@ def load_live_control_plan( if isinstance(source, (str, Path)): source_path = Path(source) - if not source_path.exists() or not source_path.is_file(): + if ( + source_path.is_symlink() + or not source_path.exists() + or not source_path.is_file() + or os.stat(source_path).st_nlink != 1 + ): _live_plan_fail("live_control_plan_missing") try: payload = json.loads(source_path.read_text(encoding="utf-8")) @@ -1048,7 +1099,13 @@ def load_live_control_plan( cli = payload.get("cli") if not isinstance(cli, Mapping) or frozenset(cli) != frozenset( - {"argv", "executable_file", "executable_sha256", "cli_version"} + { + "argv", + "executable_file", + "executable_sha256", + "argv_file_digests", + "cli_version", + } ): _live_plan_fail("cli_executable_binding_mismatch") if cli.get("cli_version") != identity.cli_version: @@ -1126,6 +1183,7 @@ def load_live_control_plan( "argv", "executable_file", "executable_sha256", + "argv_file_digests", "artifact_file", "case", } @@ -1210,8 +1268,11 @@ def capture_files(root: Path) -> list[Path]: entries = list(root.iterdir()) except OSError: raise LiveControlValidationError("sidecar_capture_missing") - if any( - entry.is_symlink() or entry.is_dir() or entry.suffix.lower() != ".json" + if len(entries) > len(CONTROL_NAMES) or any( + entry.is_symlink() + or entry.is_dir() + or entry.suffix.lower() != ".json" + or os.stat(entry).st_nlink != 1 for entry in entries ): raise LiveControlValidationError("sidecar_capture_incomplete") @@ -1246,7 +1307,7 @@ def capture_files(root: Path) -> list[Path]: def _read_sidecar_record(path: Path, *, hop: str) -> dict[str, Any]: try: - if path.is_symlink() or not path.is_file(): + if path.is_symlink() or not path.is_file() or os.stat(path).st_nlink != 1: raise OSError record = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): @@ -1307,6 +1368,10 @@ def _runtime_argv(spec: Mapping[str, Any], *, isolated_root: Path) -> list[str]: executable = _resolve_isolated_path( spec["executable_file"], isolated_root=isolated_root, must_exist=True ) + for path, expected in spec.get("argv_file_digests", {}).items(): + resolved = _resolve_isolated_path(path, isolated_root=isolated_root, must_exist=True) + if _file_sha256(resolved) != expected: + raise LiveControlValidationError("helper_executable_binding_mismatch") argv = list(spec["argv"]) argv[0] = executable return argv @@ -1327,7 +1392,8 @@ def _remove_capture_dirs( if target.is_symlink(): target.unlink() elif target.exists(): - shutil.rmtree(target) + if not _remove_directory(target): + raise OSError except (OSError, ValueError): failures += 1 return failures == 0 @@ -1351,6 +1417,22 @@ def _remove_replay_artifacts( return failures == 0 +def _remove_directory(path: Path) -> bool: + try: + if path.is_symlink(): + path.unlink() + elif path.exists(): + if not path.is_dir(): + return False + for index, _nested in enumerate(path.rglob("*"), start=1): + if index >= MAX_NESTED_ARTIFACT_SCAN: + return False + shutil.rmtree(path) + return not path.exists() + except OSError: + return False + + def _ensure_replay_artifacts_fresh( replays: Mapping[str, Mapping[str, Any]], *, run_root: Path ) -> None: @@ -1421,46 +1503,80 @@ def run_live_control( """ try: - isolation = (isolated_root or Path(run_root).parent).resolve(strict=True) + if isolated_root is None: + raise LiveControlValidationError("plan_path_outside_isolation") + raw_isolation = Path(isolated_root) + if _is_reparse(raw_isolation): + raise LiveControlValidationError("plan_path_linked") + isolation = raw_isolation.resolve(strict=True) if isinstance(plan, (str, Path)): - plan_path = Path(plan).resolve(strict=True) + raw_plan_path = Path(plan) + if ( + raw_plan_path.is_symlink() + or not raw_plan_path.is_file() + or os.stat(raw_plan_path).st_nlink != 1 + ): + raise LiveControlValidationError("plan_path_linked") + plan_path = raw_plan_path.resolve(strict=True) plan_path.relative_to(isolation) run_path = Path(run_root) if not run_path.is_absolute(): run_path = isolation / run_path + if _is_reparse(run_path): + raise LiveControlValidationError("plan_path_linked") run_path = run_path.resolve(strict=False) run_path.relative_to(isolation) if run_path == isolation or _is_reparse(run_path): raise LiveControlValidationError("plan_path_outside_isolation") - except (OSError, ValueError): + except (OSError, ValueError, RuntimeError): raise LiveControlValidationError("plan_path_outside_isolation") - loaded = load_live_control_plan( - plan, isolated_root=isolation, run_root=run_path - ) - identity = CandidateIdentity(**loaded["candidate_identity"]) + try: + loaded = load_live_control_plan( + plan, isolated_root=isolation, run_root=run_path + ) + identity = CandidateIdentity(**loaded["candidate_identity"]) + except LiveControlValidationError: + raise + except (OSError, ValueError, RuntimeError): + raise LiveControlValidationError("live_control_plan_invalid") output_manifest: Path | None = None if manifest_out is not None: - output_manifest = Path(manifest_out) - if not output_manifest.is_absolute(): - output_manifest = isolation / output_manifest - output_manifest = output_manifest.resolve(strict=False) try: + output_manifest = Path(manifest_out) + if not output_manifest.is_absolute(): + output_manifest = isolation / output_manifest + if output_manifest.is_symlink(): + raise LiveControlValidationError("plan_path_linked") + output_manifest = output_manifest.resolve(strict=False) output_manifest.relative_to(isolation) - except ValueError: - raise LiveControlValidationError("plan_path_outside_isolation") - try: - output_manifest.relative_to(run_path) - except ValueError: - pass - else: + try: + output_manifest.relative_to(run_path) + except ValueError: + pass + else: + raise LiveControlValidationError("plan_path_outside_isolation") + if output_manifest.exists(): + raise LiveControlValidationError("stale_artifact") + except LiveControlValidationError: + raise + except (OSError, ValueError, RuntimeError): raise LiveControlValidationError("plan_path_outside_isolation") - runner = BoundedPhaseRunner( - run_path, - identity, - timeout_seconds=timeout_seconds, - environment=loaded["environment"], - working_directory=isolation, - ) + case_temp = run_path / "temp" + child_environment = dict(loaded["environment"]) + child_environment["TEMP"] = str(case_temp) + child_environment["TMP"] = str(case_temp) + try: + runner = BoundedPhaseRunner( + run_path, + identity, + timeout_seconds=timeout_seconds, + environment=child_environment, + working_directory=isolation, + ) + except HarnessFailure: + raise + except (OSError, ValueError, RuntimeError): + raise LiveControlValidationError("live_control_plan_invalid") handles: list[BackgroundProcess] = [] result_status = "completed" result_code = "ok" @@ -1470,6 +1586,7 @@ def run_live_control( cleanup_actions: tuple[CleanupAction, ...] = ( lambda: _remove_capture_dirs(loaded["sidecars"], run_root=run_path), lambda: _remove_replay_artifacts(loaded["replays"], run_root=run_path), + lambda: _remove_directory(case_temp), ) def is_cancelled() -> bool: @@ -1481,6 +1598,11 @@ def is_cancelled() -> bool: if is_cancelled(): result_status, result_code = "cancelled", "cancelled" else: + try: + case_temp.mkdir(parents=False, exist_ok=False) + except OSError: + result_status, result_code = "failed", "stale_artifact" + if result_status == "completed": _prepare_capture_dirs(loaded["sidecars"]) for hop in ("pre", "post"): for index, spec in enumerate(loaded["sidecars"][hop]): @@ -1501,7 +1623,6 @@ def is_cancelled() -> bool: handle.process.poll() is not None for handle in handles ): result_status, result_code = "failed", "sidecar_capture_incomplete" - if result_status == "completed": for control in loaded["controls"]: if is_cancelled(): @@ -1691,6 +1812,15 @@ def main(argv: Sequence[str] | None = None) -> int: "cleanup_completed": False, "reason": error.code, } + except Exception: + result = { + "completed": False, + "ready_for_issue62": False, + "status": "failed", + "status_code": "phase_exception", + "cleanup_completed": False, + "reason": "phase_exception", + } print(json.dumps(result, sort_keys=True)) return 0 if result.get("completed") is True and result.get("cleanup_completed") is True else 2 diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py index 1156a906..b0e372c3 100644 --- a/tests/test_issue_62_live_control_orchestration.py +++ b/tests/test_issue_62_live_control_orchestration.py @@ -51,6 +51,11 @@ def _plan(tmp_path: Path) -> dict[str, object]: python_copy.parent.mkdir() shutil.copy2(sys.executable, python_copy) (tmp_path / "helpers" / "cli.py").write_text("raise SystemExit(0)\n", encoding="utf-8") + (tmp_path / "helpers" / "sidecar-empty.py").write_text( + "import time\n" + "time.sleep(30)\n", + encoding="utf-8", + ) (tmp_path / "helpers" / "replay.py").write_text( "import json, pathlib, sys\n" "case, candidate, manifest, output = sys.argv[1:5]\n" @@ -65,6 +70,9 @@ def _plan(tmp_path: Path) -> dict[str, object]: "executable_file": executable_file, "executable_sha256": executable_digest, } + file_digest = lambda name: hashlib.sha256( + (tmp_path / "helpers" / name).read_bytes() + ).hexdigest() return { "schema": "codexhub.issue62.live-control-plan.v1", "verification_scope": "authorized_live_control", @@ -78,16 +86,35 @@ def _plan(tmp_path: Path) -> dict[str, object]: }, "binding": binding, "catalog_model_entry_id": "gpt-5.6-sol", - "environment": {"PATH": str(Path(sys.executable).parent)}, + "environment": {}, "planner": { "model_visible_plan": "complete", "hosted_only_disposition": "Unqualified", "unknown_tag_disposition": "Unqualified", }, - "cli": {"argv": [executable_file, "helpers/cli.py"], **executable, "cli_version": "0.146.0"}, + "cli": { + "argv": [executable_file, "helpers/cli.py"], + **executable, + "argv_file_digests": {"helpers/cli.py": file_digest("cli.py")}, + "cli_version": "0.146.0", + }, "sidecars": { - "pre": {"argv": [executable_file, "helpers/cli.py"], "output_dir": "run/pre", **executable}, - "post": {"argv": [executable_file, "helpers/cli.py"], "output_dir": "run/post", **executable}, + "pre": { + "argv": [executable_file, "helpers/sidecar-empty.py"], + "output_dir": "run/pre", + **executable, + "argv_file_digests": { + "helpers/sidecar-empty.py": file_digest("sidecar-empty.py") + }, + }, + "post": { + "argv": [executable_file, "helpers/sidecar-empty.py"], + "output_dir": "run/post", + **executable, + "argv_file_digests": { + "helpers/sidecar-empty.py": file_digest("sidecar-empty.py") + }, + }, }, "controls": [ {"name": name, "args": [], "capture": {}} @@ -97,6 +124,7 @@ def _plan(tmp_path: Path) -> dict[str, object]: case: { "argv": [executable_file, "helpers/replay.py", case, "a" * 40, "0" * 64, f"run/replay-{case}.json"], **executable, + "argv_file_digests": {"helpers/replay.py": file_digest("replay.py")}, "artifact_file": f"run/replay-{case}.json", "case": case, } @@ -139,6 +167,7 @@ def test_live_execution_requires_complete_pre_and_post_sidecar_records(tmp_path: _plan(tmp_path), run_root=tmp_path / "run", timeout_seconds=30, + isolated_root=tmp_path, ) assert result["ready_for_issue62"] is False assert result["status_code"] == "sidecar_capture_missing" @@ -200,6 +229,11 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa "output_dir": f"run/{hop}-{index}", "executable_file": plan["cli"]["executable_file"], "executable_sha256": plan["cli"]["executable_sha256"], + "argv_file_digests": { + "helpers/sidecar.py": hashlib.sha256( + sidecar_script.read_bytes() + ).hexdigest() + }, } for index in range(8) ] @@ -208,7 +242,12 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa f"run/{hop}-{index}/{hop}-c{index}.json" ) # Re-run plan validation after replacing the sidecar specs. - result = run_live_control(plan, run_root=tmp_path / "run", timeout_seconds=30) + result = run_live_control( + plan, + run_root=tmp_path / "run", + timeout_seconds=30, + isolated_root=tmp_path, + ) assert result["completed"] is False assert result["ready_for_issue62"] is False assert result["status_code"] == "identity_replay_artifact_invalid" From e40da42cba0e1dfab351a8865effa2229e5ef773 Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 06:16:43 +0800 Subject: [PATCH 04/12] test(issue62): keep fixture runtimes isolated --- scripts/run_issue_62_live_control.py | 17 ++++++++++------- .../test_issue_62_live_control_orchestration.py | 7 ++++++- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/scripts/run_issue_62_live_control.py b/scripts/run_issue_62_live_control.py index 325a9bbc..739caff5 100644 --- a/scripts/run_issue_62_live_control.py +++ b/scripts/run_issue_62_live_control.py @@ -119,12 +119,11 @@ } ) LIVE_DISPOSITIONS = frozenset({"Preserved", "Unsupported", "Unqualified"}) -LIVE_ENV_KEYS = frozenset({"SystemRoot", "ComSpec"}) +LIVE_ENV_KEYS = frozenset( + {"PATH", "SystemRoot", "ComSpec", "TEMP", "TMP", "PATHEXT", "PYTHONPATH"} +) SENSITIVE_ENV_KEYS = frozenset( { - "PATH", - "PATHEXT", - "PYTHONPATH", "CODEX_HOME", "OPENAI_API_KEY", "OLLAMA_API_KEY", @@ -846,8 +845,12 @@ def _safe_environment(overrides: Mapping[str, str]) -> dict[str, str]: result[key] = value for key, value in overrides.items(): result[key] = value + # Only host-derived sensitive values are excluded. Explicit, validated + # plan overrides (for example a case-local PATH/TEMP) are retained so the + # isolated executable can resolve its own runtime dependencies. for key in SENSITIVE_ENV_KEYS: - result.pop(key, None) + if key not in overrides: + result.pop(key, None) return result @@ -1643,7 +1646,7 @@ def is_cancelled() -> bool: # Stop sidecars before reading their atomic records. Cleanup repeats # this operation defensively for cancellation/error paths. for handle in reversed(handles): - if not runner.stop_background(handle): + if not runner.stop_background(handle) and result_status == "completed": result_status, result_code = "failed", "termination_failed" if result_status == "completed": @@ -1717,7 +1720,7 @@ def is_cancelled() -> bool: "child_processes_terminated": False, "resources_released": False, } - if receipt.get("cleanup_completed") is not True: + if receipt.get("cleanup_completed") is not True and result_status == "completed": result_status, result_code = "failed", "cleanup_incomplete" output: dict[str, Any] = { diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py index b0e372c3..f6d240e5 100644 --- a/tests/test_issue_62_live_control_orchestration.py +++ b/tests/test_issue_62_live_control_orchestration.py @@ -50,6 +50,11 @@ def _plan(tmp_path: Path) -> dict[str, object]: python_copy = tmp_path / "tools" / "python.exe" python_copy.parent.mkdir() shutil.copy2(sys.executable, python_copy) + # A copied Windows interpreter needs its private runtime DLLs beside the + # executable. Keep them under the isolated root and expose only that + # case-local directory through PATH; never rely on the host PATH. + for runtime_dll in Path(sys.executable).parent.glob("*.dll"): + shutil.copy2(runtime_dll, python_copy.parent / runtime_dll.name) (tmp_path / "helpers" / "cli.py").write_text("raise SystemExit(0)\n", encoding="utf-8") (tmp_path / "helpers" / "sidecar-empty.py").write_text( "import time\n" @@ -86,7 +91,7 @@ def _plan(tmp_path: Path) -> dict[str, object]: }, "binding": binding, "catalog_model_entry_id": "gpt-5.6-sol", - "environment": {}, + "environment": {"PATH": "tools"}, "planner": { "model_visible_plan": "complete", "hosted_only_disposition": "Unqualified", From 5db86e575612cfdaed72adf23030a5ecd36fd4c8 Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 06:20:25 +0800 Subject: [PATCH 05/12] fix(issue62): confine live runner environment paths --- scripts/run_issue_62_live_control.py | 61 ++++++++++++++++++---- tests/test_issue62_bounded_phase_runner.py | 20 +++++++ 2 files changed, 72 insertions(+), 9 deletions(-) diff --git a/scripts/run_issue_62_live_control.py b/scripts/run_issue_62_live_control.py index 739caff5..7ae56bf9 100644 --- a/scripts/run_issue_62_live_control.py +++ b/scripts/run_issue_62_live_control.py @@ -122,6 +122,7 @@ LIVE_ENV_KEYS = frozenset( {"PATH", "SystemRoot", "ComSpec", "TEMP", "TMP", "PATHEXT", "PYTHONPATH"} ) +LOCAL_ENV_PATH_KEYS = frozenset({"PATH", "TEMP", "TMP", "PYTHONPATH"}) SENSITIVE_ENV_KEYS = frozenset( { "CODEX_HOME", @@ -378,8 +379,13 @@ def __init__( self.identity = identity self.timeout_seconds = _validate_timeout(timeout_seconds) self._process_factory = process_factory or subprocess.Popen - self._environment = _safe_environment(environment or {}) self._working_directory = str(working_directory) if working_directory is not None else None + self._environment = _safe_environment( + environment or {}, + working_directory=Path(working_directory) + if working_directory is not None + else None, + ) self.journal = SanitizedPhaseJournal(run_root, identity) self._active_process: subprocess.Popen[bytes] | None = None self._active_lock = threading.Lock() @@ -817,7 +823,33 @@ def _resolve_isolated_path( return str(resolved) -def _validate_environment(value: Any) -> dict[str, str]: +def _environment_value_is_local( + key: str, value: str, *, working_directory: Path | None = None +) -> bool: + """Reject host path injection in child environment overrides.""" + + if key not in LOCAL_ENV_PATH_KEYS: + return True + root = working_directory.resolve(strict=False) if working_directory is not None else None + for entry in value.split(os.pathsep): + if not entry: + continue + candidate = Path(entry) + if candidate.is_absolute() or re.match(r"^(?:[A-Za-z]:|[\\/]{2})", entry): + if root is None: + return False + try: + candidate.resolve(strict=False).relative_to(root) + except (OSError, ValueError, RuntimeError): + return False + elif ".." in candidate.parts: + return False + return True + + +def _validate_environment( + value: Any, *, isolated_root: Path | None = None +) -> dict[str, str]: if not isinstance(value, Mapping): _live_plan_fail("environment_invalid") normalized: dict[str, str] = {} @@ -829,13 +861,18 @@ def _validate_environment(value: Any) -> dict[str, str]: or not isinstance(item, str) or "\x00" in item or len(item) > 4096 + or not _environment_value_is_local( + key, item, working_directory=isolated_root + ) ): _live_plan_fail("environment_invalid") normalized[key] = item return normalized -def _safe_environment(overrides: Mapping[str, str]) -> dict[str, str]: +def _safe_environment( + overrides: Mapping[str, str], *, working_directory: Path | None = None +) -> dict[str, str]: """Build a minimal child environment without host credentials/home state.""" result: dict[str, str] = {} @@ -844,13 +881,17 @@ def _safe_environment(overrides: Mapping[str, str]) -> dict[str, str]: if value: result[key] = value for key, value in overrides.items(): + if not _environment_value_is_local( + key, value, working_directory=working_directory + ): + continue result[key] = value - # Only host-derived sensitive values are excluded. Explicit, validated - # plan overrides (for example a case-local PATH/TEMP) are retained so the - # isolated executable can resolve its own runtime dependencies. + # Never copy sensitive values from arbitrary direct callers. The public + # runner constructor is also used by fixture tests, so this function must + # enforce the credential/home boundary itself rather than relying on the + # live-plan validator. for key in SENSITIVE_ENV_KEYS: - if key not in overrides: - result.pop(key, None) + result.pop(key, None) return result @@ -1123,7 +1164,9 @@ def load_live_control_plan( normalized_sidecars = _validate_sidecars( payload.get("sidecars"), isolated_root=isolated_root, run_root=run_root ) - normalized_environment = _validate_environment(payload.get("environment")) + normalized_environment = _validate_environment( + payload.get("environment"), isolated_root=isolated_root + ) planner = payload.get("planner") if not isinstance(planner, Mapping) or frozenset(planner) != PLANNER_FIELDS: _live_plan_fail("planner_plan_incomplete") diff --git a/tests/test_issue62_bounded_phase_runner.py b/tests/test_issue62_bounded_phase_runner.py index c48c6f57..9459aebe 100644 --- a/tests/test_issue62_bounded_phase_runner.py +++ b/tests/test_issue62_bounded_phase_runner.py @@ -58,6 +58,26 @@ def test_success_writes_only_sanitized_phase_status_and_cleanup_receipt(tmp_path assert all("argv" not in record and "command" not in record for record in markers) +def test_direct_runner_drops_sensitive_and_host_path_overrides(tmp_path: Path) -> None: + runner = BoundedPhaseRunner( + tmp_path, + IDENTITY, + environment={ + "HOME": "C:/host/home", + "OPENAI_API_KEY": "must-not-pass", + "PATH": "C:/host/bin", + "PYTHONPATH": "../host-modules", + }, + working_directory=tmp_path, + ) + + assert "HOME" not in runner._environment + assert "OPENAI_API_KEY" not in runner._environment + assert "PATH" not in runner._environment + assert "PYTHONPATH" not in runner._environment + assert runner.cleanup()["cleanup_completed"] is True + + class _FakeProcess: def __init__(self, return_code: int | None = None) -> None: self.return_code = return_code From 28bc9457ae03a63592e72ae080528b4fb1119427 Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 06:21:58 +0800 Subject: [PATCH 06/12] fix(issue62): reject unanchored path overrides --- scripts/run_issue_62_live_control.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/scripts/run_issue_62_live_control.py b/scripts/run_issue_62_live_control.py index 7ae56bf9..88c9aa28 100644 --- a/scripts/run_issue_62_live_control.py +++ b/scripts/run_issue_62_live_control.py @@ -830,6 +830,8 @@ def _environment_value_is_local( if key not in LOCAL_ENV_PATH_KEYS: return True + if working_directory is None: + return False root = working_directory.resolve(strict=False) if working_directory is not None else None for entry in value.split(os.pathsep): if not entry: From 17de7b652e7bc45622eaf2430a80aafca6e2cc03 Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 06:27:53 +0800 Subject: [PATCH 07/12] test(issue62): use absolute isolated fixture path --- tests/test_issue_62_live_control_orchestration.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py index f6d240e5..7eb02d20 100644 --- a/tests/test_issue_62_live_control_orchestration.py +++ b/tests/test_issue_62_live_control_orchestration.py @@ -91,7 +91,7 @@ def _plan(tmp_path: Path) -> dict[str, object]: }, "binding": binding, "catalog_model_entry_id": "gpt-5.6-sol", - "environment": {"PATH": "tools"}, + "environment": {"PATH": str(python_copy.parent)}, "planner": { "model_visible_plan": "complete", "hosted_only_disposition": "Unqualified", From 51b8a6ac6675df6585afa35e8205b5867a3ded84 Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 06:31:57 +0800 Subject: [PATCH 08/12] test(issue62): widen sidecar fixture lifetime --- tests/test_issue_62_live_control_orchestration.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py index 7eb02d20..75c4e680 100644 --- a/tests/test_issue_62_live_control_orchestration.py +++ b/tests/test_issue_62_live_control_orchestration.py @@ -58,7 +58,9 @@ def _plan(tmp_path: Path) -> dict[str, object]: (tmp_path / "helpers" / "cli.py").write_text("raise SystemExit(0)\n", encoding="utf-8") (tmp_path / "helpers" / "sidecar-empty.py").write_text( "import time\n" - "time.sleep(30)\n", + # Keep the fixture alive longer than the full Hosted startup window; + # the harness still bounds and terminates it during cleanup. + "time.sleep(300)\n", encoding="utf-8", ) (tmp_path / "helpers" / "replay.py").write_text( @@ -212,9 +214,9 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa sidecar_script = tmp_path / "helpers" / "sidecar.py" sidecar_script.write_text( "import pathlib, shutil, sys, time\n" - "out = pathlib.Path(sys.argv[1]); src = pathlib.Path(sys.argv[2])\n" - "shutil.copyfile(src, out / pathlib.Path(sys.argv[3]).name)\n" - "time.sleep(30)\n", + "out = pathlib.Path(sys.argv[1]); src = pathlib.Path(sys.argv[2])\n" + "shutil.copyfile(src, out / pathlib.Path(sys.argv[3]).name)\n" + "time.sleep(300)\n", encoding="utf-8", ) for hop in ("pre", "post"): From 9ec0b9b5413b6861595c21fe8e213441e03a0b0e Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 06:48:35 +0800 Subject: [PATCH 09/12] test(issue62): make Windows cmd fixtures portable --- ...est_issue_62_live_control_orchestration.py | 145 ++++++++++++------ 1 file changed, 100 insertions(+), 45 deletions(-) diff --git a/tests/test_issue_62_live_control_orchestration.py b/tests/test_issue_62_live_control_orchestration.py index 75c4e680..3d77a76b 100644 --- a/tests/test_issue_62_live_control_orchestration.py +++ b/tests/test_issue_62_live_control_orchestration.py @@ -3,6 +3,7 @@ import hashlib import importlib import json +import os from pathlib import Path import shutil import sys @@ -47,32 +48,76 @@ def _plan(tmp_path: Path) -> dict[str, object]: binding = _binding_files(tmp_path) (tmp_path / "helpers").mkdir() (tmp_path / "run").mkdir() - python_copy = tmp_path / "tools" / "python.exe" - python_copy.parent.mkdir() - shutil.copy2(sys.executable, python_copy) - # A copied Windows interpreter needs its private runtime DLLs beside the - # executable. Keep them under the isolated root and expose only that - # case-local directory through PATH; never rely on the host PATH. - for runtime_dll in Path(sys.executable).parent.glob("*.dll"): - shutil.copy2(runtime_dll, python_copy.parent / runtime_dll.name) - (tmp_path / "helpers" / "cli.py").write_text("raise SystemExit(0)\n", encoding="utf-8") - (tmp_path / "helpers" / "sidecar-empty.py").write_text( - "import time\n" - # Keep the fixture alive longer than the full Hosted startup window; - # the harness still bounds and terminates it during cleanup. - "time.sleep(300)\n", - encoding="utf-8", - ) - (tmp_path / "helpers" / "replay.py").write_text( - "import json, pathlib, sys\n" - "case, candidate, manifest, output = sys.argv[1:5]\n" - "path = pathlib.Path(output)\n" - "path.parent.mkdir(parents=True, exist_ok=True)\n" - "path.write_text(json.dumps({'schema':'codexhub.issue62.identity-replay.v1', 'case':case, 'candidate_sha':candidate, 'capture_manifest_sha256':manifest, 'wire_replay':True, 'outcome':'accepted' if case == 'identity' else 'rejected'}, sort_keys=True, separators=(',', ':')) + '\\n', encoding='utf-8')\n", - encoding="utf-8", - ) - executable_digest = hashlib.sha256(python_copy.read_bytes()).hexdigest() - executable_file = str(python_copy.relative_to(tmp_path)) + tools_dir = tmp_path / "tools" + tools_dir.mkdir() + if os.name == "nt": + # A copied setup-python interpreter is not reliably launchable on a + # fresh Hosted image (its loader can reject the relocated DLL set). + # Use the system command host as the bound executable instead; all + # scripts and the timeout helper remain copied under this root. + executable_path = tools_dir / "cmd.exe" + shutil.copy2(Path(os.environ["SystemRoot"]) / "System32" / "cmd.exe", executable_path) + shutil.copy2(Path(os.environ["SystemRoot"]) / "System32" / "PING.EXE", tools_dir / "PING.EXE") + (tmp_path / "helpers" / "cli.cmd").write_text( + "@echo off\nexit /b 0\n", encoding="ascii" + ) + (tmp_path / "helpers" / "sidecar-empty.cmd").write_text( + "@echo off\nping 127.0.0.1 -n 300 >nul\n", encoding="ascii" + ) + (tmp_path / "helpers" / "replay.cmd").write_text( + "@echo off\n" + "set \"case=%~1\"\n" + "set \"candidate=%~2\"\n" + "set \"manifest=%~3\"\n" + "set \"output=%~4\"\n" + "if /I \"%case%\"==\"identity\" (set \"outcome=accepted\") else (set \"outcome=rejected\")\n" + ">\"%output%\" echo {\"schema\":\"codexhub.issue62.identity-replay.v1\",\"case\":\"%case%\",\"candidate_sha\":\"%candidate%\",\"capture_manifest_sha256\":\"%manifest%\",\"wire_replay\":true,\"outcome\":\"%outcome%\"}\n", + encoding="ascii", + ) + executable_file = str(executable_path.relative_to(tmp_path)) + environment = { + "PATH": str(tools_dir), + "PATHEXT": ".COM;.EXE;.BAT;.CMD", + } + cli_argv = [executable_file, "/d", "/c", "helpers\\cli.cmd"] + sidecar_argv = [executable_file, "/d", "/c", "helpers\\sidecar-empty.cmd"] + replay_argv = lambda case: [ + executable_file, + "/d", + "/c", + "helpers\\replay.cmd", + case, + "a" * 40, + "0" * 64, + f"run/replay-{case}.json", + ] + helper_names = {"cli": "cli.cmd", "sidecar": "sidecar-empty.cmd", "replay": "replay.cmd"} + else: + executable_path = tools_dir / "python" + shutil.copy2(sys.executable, executable_path) + (tmp_path / "helpers" / "cli.py").write_text("raise SystemExit(0)\n", encoding="utf-8") + (tmp_path / "helpers" / "sidecar-empty.py").write_text( + "import time\n" + "time.sleep(300)\n", + encoding="utf-8", + ) + (tmp_path / "helpers" / "replay.py").write_text( + "import json, pathlib, sys\n" + "case, candidate, manifest, output = sys.argv[1:5]\n" + "path = pathlib.Path(output)\n" + "path.parent.mkdir(parents=True, exist_ok=True)\n" + "path.write_text(json.dumps({'schema':'codexhub.issue62.identity-replay.v1', 'case':case, 'candidate_sha':candidate, 'capture_manifest_sha256':manifest, 'wire_replay':True, 'outcome':'accepted' if case == 'identity' else 'rejected'}, sort_keys=True, separators=(',', ':')) + '\\n', encoding='utf-8')\n", + encoding="utf-8", + ) + executable_file = str(executable_path.relative_to(tmp_path)) + environment = {} + cli_argv = [executable_file, "helpers/cli.py"] + sidecar_argv = [executable_file, "helpers/sidecar-empty.py"] + replay_argv = lambda case: [ + executable_file, "helpers/replay.py", case, "a" * 40, "0" * 64, f"run/replay-{case}.json" + ] + helper_names = {"cli": "cli.py", "sidecar": "sidecar-empty.py", "replay": "replay.py"} + executable_digest = hashlib.sha256(executable_path.read_bytes()).hexdigest() executable = { "executable_file": executable_file, "executable_sha256": executable_digest, @@ -93,33 +138,33 @@ def _plan(tmp_path: Path) -> dict[str, object]: }, "binding": binding, "catalog_model_entry_id": "gpt-5.6-sol", - "environment": {"PATH": str(python_copy.parent)}, + "environment": environment, "planner": { "model_visible_plan": "complete", "hosted_only_disposition": "Unqualified", "unknown_tag_disposition": "Unqualified", }, "cli": { - "argv": [executable_file, "helpers/cli.py"], + "argv": cli_argv, **executable, - "argv_file_digests": {"helpers/cli.py": file_digest("cli.py")}, + "argv_file_digests": {f"helpers/{helper_names['cli']}": file_digest(helper_names["cli"])}, "cli_version": "0.146.0", }, "sidecars": { "pre": { - "argv": [executable_file, "helpers/sidecar-empty.py"], + "argv": sidecar_argv, "output_dir": "run/pre", **executable, "argv_file_digests": { - "helpers/sidecar-empty.py": file_digest("sidecar-empty.py") + f"helpers/{helper_names['sidecar']}": file_digest(helper_names["sidecar"]) }, }, "post": { - "argv": [executable_file, "helpers/sidecar-empty.py"], + "argv": sidecar_argv, "output_dir": "run/post", **executable, "argv_file_digests": { - "helpers/sidecar-empty.py": file_digest("sidecar-empty.py") + f"helpers/{helper_names['sidecar']}": file_digest(helper_names["sidecar"]) }, }, }, @@ -129,9 +174,9 @@ def _plan(tmp_path: Path) -> dict[str, object]: ], "replays": { case: { - "argv": [executable_file, "helpers/replay.py", case, "a" * 40, "0" * 64, f"run/replay-{case}.json"], + "argv": replay_argv(case), **executable, - "argv_file_digests": {"helpers/replay.py": file_digest("replay.py")}, + "argv_file_digests": {f"helpers/{helper_names['replay']}": file_digest(helper_names["replay"])}, "artifact_file": f"run/replay-{case}.json", "case": case, } @@ -211,14 +256,27 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa "pre_record": f"run/pre-{index}/pre-c{index}.json", "post_record": f"run/post-{index}/post-c{index}.json", } - sidecar_script = tmp_path / "helpers" / "sidecar.py" - sidecar_script.write_text( - "import pathlib, shutil, sys, time\n" + if os.name == "nt": + sidecar_name = "sidecar.cmd" + sidecar_script = tmp_path / "helpers" / sidecar_name + sidecar_script.write_text( + "@echo off\n" + "copy /y \"%~2\" \"%~1\\%~3\" >nul\n" + "ping 127.0.0.1 -n 300 >nul\n", + encoding="ascii", + ) + sidecar_prefix = [plan["cli"]["executable_file"], "/d", "/c", "helpers\\sidecar.cmd"] + else: + sidecar_name = "sidecar.py" + sidecar_script = tmp_path / "helpers" / sidecar_name + sidecar_script.write_text( + "import pathlib, shutil, sys, time\n" "out = pathlib.Path(sys.argv[1]); src = pathlib.Path(sys.argv[2])\n" "shutil.copyfile(src, out / pathlib.Path(sys.argv[3]).name)\n" "time.sleep(300)\n", - encoding="utf-8", - ) + encoding="utf-8", + ) + sidecar_prefix = [plan["cli"]["executable_file"], "helpers/sidecar.py"] for hop in ("pre", "post"): specs = plan["sidecars"][hop] source_paths = [payloads[f"{hop}-{index}"] for index in range(8)] @@ -227,8 +285,7 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa plan["sidecars"][hop] = [ { "argv": [ - plan["cli"]["executable_file"], - "helpers/sidecar.py", + *sidecar_prefix, f"run/{hop}-{index}", str(source_paths[index].relative_to(tmp_path)), f"{hop}-c{index}.json", @@ -237,9 +294,7 @@ def test_live_execution_binds_all_controls_and_keeps_qualification_closed(tmp_pa "executable_file": plan["cli"]["executable_file"], "executable_sha256": plan["cli"]["executable_sha256"], "argv_file_digests": { - "helpers/sidecar.py": hashlib.sha256( - sidecar_script.read_bytes() - ).hexdigest() + f"helpers/{sidecar_name}": hashlib.sha256(sidecar_script.read_bytes()).hexdigest() }, } for index in range(8) From 5eb652972af33b93c04b5ccbff1875378490ed9a Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 07:13:43 +0800 Subject: [PATCH 10/12] test(issue62): bind live capture correlation --- scripts/capture_issue_62_live_evidence.py | 114 +++++++++++++++-- tests/test_issue_62_live_evidence_sidecar.py | 122 ++++++++++++++++++- 2 files changed, 224 insertions(+), 12 deletions(-) diff --git a/scripts/capture_issue_62_live_evidence.py b/scripts/capture_issue_62_live_evidence.py index 7e52091f..37055380 100644 --- a/scripts/capture_issue_62_live_evidence.py +++ b/scripts/capture_issue_62_live_evidence.py @@ -35,6 +35,7 @@ _FAILURES = frozenset( { "client_cancelled", + "correlation_binding_invalid", "downstream_cancelled", "forwarding_failed", "invalid_content_length", @@ -80,6 +81,9 @@ ) _CAPTURE_ID = re.compile(r"c[0-9a-f]{32}\Z") _DIGEST = re.compile(r"[0-9a-f]{64}\Z") +_CORRELATION_TOKEN = re.compile(r"[0-9a-f]{32}\Z") +_CORRELATION_BINDING = re.compile(r"([0-9a-f]{32})\.([0-9a-f]{64})\Z") +_CORRELATION_HEADER = "X-CodexHub-Issue62-Capture" class ConfigurationError(ValueError): @@ -391,10 +395,76 @@ def validate_capture_record(record: Mapping[str, Any], hop: str) -> None: raise ArtifactValidationError("record_completion_invalid") -def write_capture_record(output_dir: Path, hop: str, record: Mapping[str, Any]) -> Path: - """Atomically publish one sanitized record and always remove its partial.""" +def _capture_id_for_token(key: bytes, correlation_token: str) -> str: + if not isinstance(key, bytes) or not 32 <= len(key) <= 4096: + raise ValueError("capture_key_invalid") + if ( + not isinstance(correlation_token, str) + or _CORRELATION_TOKEN.fullmatch(correlation_token) is None + ): + raise ValueError("correlation_token_invalid") + digest = hmac.new( + key, + b"issue62-capture\0" + correlation_token.encode("ascii"), + hashlib.sha256, + ).hexdigest() + return "c" + digest[:32] + + +def _new_correlation_token() -> str: + return uuid.uuid4().hex + + +def _correlation_binding(key: bytes, correlation_token: str) -> str: + if ( + not isinstance(correlation_token, str) + or _CORRELATION_TOKEN.fullmatch(correlation_token) is None + ): + raise ValueError("correlation_token_invalid") + signature = hmac.new( + key, + b"issue62-correlation\0" + correlation_token.encode("ascii"), + hashlib.sha256, + ).hexdigest() + return f"{correlation_token}.{signature}" + + +def _correlation_token_from_header(key: bytes, value: str | None) -> str | None: + if not isinstance(value, str): + return None + match = _CORRELATION_BINDING.fullmatch(value) + if match is None: + return None + correlation_token, signature = match.groups() + expected = hmac.new( + key, + b"issue62-correlation\0" + correlation_token.encode("ascii"), + hashlib.sha256, + ).hexdigest() + if not hmac.compare_digest(signature, expected): + return None + return correlation_token + + +def write_capture_record( + output_dir: Path, + hop: str, + record: Mapping[str, Any], + *, + capture_key: bytes | None = None, + correlation_token: str | None = None, +) -> Path: + """Atomically publish one live-bound sanitized record and remove its partial.""" validate_capture_record(record, hop) + if capture_key is None or correlation_token is None: + raise ArtifactValidationError("capture_binding_missing") + try: + expected_capture_id = _capture_id_for_token(capture_key, correlation_token) + except (TypeError, ValueError): + raise ArtifactValidationError("capture_binding_invalid") from None + if not hmac.compare_digest(str(record["capture_id"]), expected_capture_id): + raise ArtifactValidationError("capture_binding_invalid") directory = Path(output_dir) directory.mkdir(parents=True, exist_ok=True) capture_id = str(record["capture_id"]) @@ -431,10 +501,6 @@ def write_capture_record(output_dir: Path, hop: str, record: Mapping[str, Any]) ) -def _capture_id() -> str: - return "c" + uuid.uuid4().hex - - def _content_type_class(value: str | None) -> str: if not value: return "absent" @@ -603,11 +669,18 @@ def wait(self) -> bool: return self._thread_failure is None def _write_control_failure(self, failure: str) -> None: + correlation_token = _new_correlation_token() try: write_capture_record( self._config.output_dir, self._config.hop, - _empty_record(self._config.hop, _capture_id(), failure), + _empty_record( + self._config.hop, + _capture_id_for_token(self._key, correlation_token), + failure, + ), + capture_key=self._key, + correlation_token=correlation_token, ) except Exception: return @@ -665,7 +738,8 @@ def _wait_for_active_connections(self, timeout: float) -> bool: return True def handle_post(self, handler: BaseHTTPRequestHandler) -> None: - capture_id = _capture_id() + correlation_token = _new_correlation_token() + capture_id = _capture_id_for_token(self._key, correlation_token) started_at = time.monotonic() request_fingerprint = BodyFingerprint(self._key, b"request-body") response_fingerprint: BodyFingerprint | None = None @@ -677,6 +751,17 @@ def handle_post(self, handler: BaseHTTPRequestHandler) -> None: request_complete = False response_complete = False try: + if self._config.hop == "post": + bound_token = _correlation_token_from_header( + self._key, + handler.headers.get(_CORRELATION_HEADER), + ) + if bound_token is None: + failure = "correlation_binding_invalid" + self._send_bounded_error(handler, 400) + return + correlation_token = bound_token + capture_id = _capture_id_for_token(self._key, correlation_token) raw_length = handler.headers.get("Content-Length") try: content_length = int(raw_length) if raw_length is not None else -1 @@ -716,8 +801,11 @@ def handle_post(self, handler: BaseHTTPRequestHandler) -> None: headers = { name: value for name, value in handler.headers.items() - if name.lower() not in _HOP_BY_HOP_HEADERS | {"host", "content-length"} + if name.lower() + not in _HOP_BY_HOP_HEADERS | {"host", "content-length", _CORRELATION_HEADER.lower()} } + if self._config.hop == "pre": + headers[_CORRELATION_HEADER] = _correlation_binding(self._key, correlation_token) target_path = _join_target_path(self._target.path, handler.path) upstream.request("POST", target_path, body=request_body, headers=headers) if upstream.sock is not None: @@ -810,7 +898,13 @@ def handle_post(self, handler: BaseHTTPRequestHandler) -> None: ), } try: - write_capture_record(self._config.output_dir, self._config.hop, record) + write_capture_record( + self._config.output_dir, + self._config.hop, + record, + capture_key=self._key, + correlation_token=correlation_token, + ) except Exception: return diff --git a/tests/test_issue_62_live_evidence_sidecar.py b/tests/test_issue_62_live_evidence_sidecar.py index 72456162..35dc1e90 100644 --- a/tests/test_issue_62_live_evidence_sidecar.py +++ b/tests/test_issue_62_live_evidence_sidecar.py @@ -28,6 +28,11 @@ SPEC.loader.exec_module(sidecar) KEY = b"h" * 32 +CORRELATION_TOKEN = "1" * 32 + + +def _capture_id(correlation_token: str = CORRELATION_TOKEN) -> str: + return sidecar._capture_id_for_token(KEY, correlation_token) class _FakeUpstreamHandler(BaseHTTPRequestHandler): @@ -36,6 +41,9 @@ class _FakeUpstreamHandler(BaseHTTPRequestHandler): def do_POST(self) -> None: length = int(self.headers["Content-Length"]) self.server.observed_request = self.rfile.read(length) # type: ignore[attr-defined] + self.server.observed_headers = { # type: ignore[attr-defined] + name.lower(): value for name, value in self.headers.items() + } body = self.server.response_body # type: ignore[attr-defined] delay = self.server.response_delay # type: ignore[attr-defined] if delay: @@ -69,6 +77,7 @@ def _fake_upstream( server = ThreadingHTTPServer(("127.0.0.1", 0), _FakeUpstreamHandler) server.response_body = response_body # type: ignore[attr-defined] server.observed_request = None # type: ignore[attr-defined] + server.observed_headers = {} # type: ignore[attr-defined] server.response_delay = response_delay # type: ignore[attr-defined] server.response_chunk_size = chunk_size # type: ignore[attr-defined] server.chunk_delay = chunk_delay # type: ignore[attr-defined] @@ -214,7 +223,7 @@ def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path: Path) -> Non record = { "schema": "codexhub.issue62.live-evidence-lane.v1", "verification_scope": "capture_only_not_qualification", - "capture_id": "c" + "1" * 32, + "capture_id": _capture_id(), "hop": "pre", "outcome": "complete", "failure": None, @@ -235,7 +244,13 @@ def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path: Path) -> Non "sse": None, } - record_path = sidecar.write_capture_record(tmp_path, "pre", record) + record_path = sidecar.write_capture_record( + tmp_path, + "pre", + record, + capture_key=KEY, + correlation_token=CORRELATION_TOKEN, + ) assert json.loads(record_path.read_text(encoding="utf-8")) == record assert record_path.name.startswith("pre-") @@ -243,6 +258,70 @@ def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path: Path) -> Non assert not list(tmp_path.glob("*.partial")) +def test_atomic_record_rejects_prebuilt_record_without_live_capture_binding(tmp_path: Path) -> None: + record = { + "schema": "codexhub.issue62.live-evidence-lane.v1", + "verification_scope": "capture_only_not_qualification", + "capture_id": _capture_id("2" * 32), + "hop": "pre", + "outcome": "complete", + "failure": None, + "status": 200, + "content_type_class": "json", + "request": { + "bytes": 2, + "sha256": "a" * 64, + "hmac_sha256": "b" * 64, + "complete": True, + }, + "response": { + "bytes": 3, + "sha256": "c" * 64, + "hmac_sha256": "d" * 64, + "complete": True, + }, + "sse": None, + } + + with pytest.raises(sidecar.ArtifactValidationError, match="capture_binding_missing"): + sidecar.write_capture_record(tmp_path, "pre", record) + + +def test_atomic_record_rejects_capture_id_from_different_live_correlation(tmp_path: Path) -> None: + record = { + "schema": "codexhub.issue62.live-evidence-lane.v1", + "verification_scope": "capture_only_not_qualification", + "capture_id": _capture_id("2" * 32), + "hop": "pre", + "outcome": "complete", + "failure": None, + "status": 200, + "content_type_class": "json", + "request": { + "bytes": 2, + "sha256": "a" * 64, + "hmac_sha256": "b" * 64, + "complete": True, + }, + "response": { + "bytes": 3, + "sha256": "c" * 64, + "hmac_sha256": "d" * 64, + "complete": True, + }, + "sse": None, + } + + with pytest.raises(sidecar.ArtifactValidationError, match="capture_binding_invalid"): + sidecar.write_capture_record( + tmp_path, + "pre", + record, + capture_key=KEY, + correlation_token=CORRELATION_TOKEN, + ) + + def test_atomic_record_rejects_unapproved_fields(tmp_path: Path) -> None: with pytest.raises(sidecar.ArtifactValidationError, match="record_fields_invalid"): sidecar.write_capture_record( @@ -515,6 +594,7 @@ def test_two_hops_capture_matching_complete_request_response_and_sse( headers={ "Authorization": "Bearer forbidden-token", "Content-Type": "application/json", + "X-CodexHub-Issue62-Capture": ("0" * 32) + "." + ("f" * 64), }, ) response = connection.getresponse() @@ -529,10 +609,12 @@ def test_two_hops_capture_matching_complete_request_response_and_sse( pre_record = _read_only_record(pre_output) post_record = _read_only_record(post_output) assert pre_record["request"] == post_record["request"] + assert pre_record["capture_id"] == post_record["capture_id"] assert pre_record["response"] == post_record["response"] assert pre_record["sse"]["sequence_sha256"] == post_record["sse"]["sequence_sha256"] # type: ignore[index] assert pre_record["sse"]["sequence_hmac_sha256"] == post_record["sse"]["sequence_hmac_sha256"] # type: ignore[index] assert pre_record["outcome"] == post_record["outcome"] == "complete" + assert "x-codexhub-issue62-capture" not in upstream.observed_headers # type: ignore[attr-defined] serialized = json.dumps([pre_record, post_record], sort_keys=True) for sensitive in ( "private prompt", @@ -548,6 +630,42 @@ def test_two_hops_capture_matching_complete_request_response_and_sse( assert not list(tmp_path.rglob("*.partial")) +@pytest.mark.parametrize( + "correlation_header", + [None, ("0" * 32) + "." + ("f" * 64)], +) +def test_post_rejects_missing_or_invalid_correlation_before_forwarding( + correlation_header: str | None, + tmp_path: Path, + hmac_key_file: Path, +) -> None: + with _fake_upstream(b'data: {"type":"response.completed"}\n\n') as upstream: + upstream_url = f"http://127.0.0.1:{upstream.server_address[1]}" + config = _sidecar_config( + tmp_path, + hmac_key_file, + upstream_url, + hop="post", + ) + with _running_sidecar(config) as server: + connection = http.client.HTTPConnection(server.listen_host, server.listen_port, timeout=2) + headers = {} + if correlation_header is not None: + headers["X-CodexHub-Issue62-Capture"] = correlation_header + connection.request("POST", "/responses", body=b"{}", headers=headers) + response = connection.getresponse() + assert response.status == 400 + response.read() + connection.close() + + assert upstream.observed_request is None # type: ignore[attr-defined] + record = _read_only_record(config.output_dir) + assert record["outcome"] == "incomplete" + assert record["failure"] == "correlation_binding_invalid" + assert "X-CodexHub-Issue62-Capture" not in json.dumps(record, sort_keys=True) + assert not list(tmp_path.rglob("*.partial")) + + def test_request_overflow_fails_closed_before_forwarding( tmp_path: Path, hmac_key_file: Path, From 67b724096029be624238bca2ce571cf6ff9e0476 Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 07:34:31 +0800 Subject: [PATCH 11/12] docs(issue62): document live capture binding --- .../plans/2026-08-02-issue-62-live-evidence-sidecar.md | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md b/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md index 6ddc8068..d1904516 100644 --- a/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md +++ b/docs/superpowers/plans/2026-08-02-issue-62-live-evidence-sidecar.md @@ -27,7 +27,7 @@ - Create: `tests/test_issue_62_live_evidence_sidecar.py` **Interfaces:** -- Produces: `SidecarConfig`, `validate_config(config)`, `write_capture_record(output_dir, hop, record)`, and `main(argv=None) -> int`. +- Produces: `SidecarConfig`, `validate_config(config)`, `write_capture_record(output_dir, hop, record, capture_key=..., correlation_token=...)`, and `main(argv=None) -> int`. - Consumes: only Python standard-library modules and a pre-existing HMAC key file. - [x] **Step 1: Write failing activation and artifact tests** @@ -43,7 +43,13 @@ def test_config_rejects_non_loopback(host, base_config): sidecar.validate_config(dataclasses.replace(base_config, listen_host=host)) def test_atomic_record_is_sanitized_and_leaves_no_partial(tmp_path): - record_path = sidecar.write_capture_record(tmp_path, "pre", SAFE_RECORD) + record_path = sidecar.write_capture_record( + tmp_path, + "pre", + SAFE_RECORD, + capture_key=KEY, + correlation_token=CORRELATION_TOKEN, + ) assert json.loads(record_path.read_text(encoding="utf-8"))["schema"] == ( "codexhub.issue62.live-evidence-lane.v1" ) From 8f798c8cc4cdd5bc654b2c8fe645c9b4c96ab6a9 Mon Sep 17 00:00:00 2001 From: Paul Qu Date: Tue, 4 Aug 2026 07:42:16 +0800 Subject: [PATCH 12/12] test(issue62): cover deterministic capture correlation --- tests/test_issue_62_live_evidence_sidecar.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/test_issue_62_live_evidence_sidecar.py b/tests/test_issue_62_live_evidence_sidecar.py index 35dc1e90..954ae825 100644 --- a/tests/test_issue_62_live_evidence_sidecar.py +++ b/tests/test_issue_62_live_evidence_sidecar.py @@ -35,6 +35,11 @@ def _capture_id(correlation_token: str = CORRELATION_TOKEN) -> str: return sidecar._capture_id_for_token(KEY, correlation_token) +def test_capture_id_is_deterministically_bound_to_correlation_token() -> None: + assert _capture_id() == _capture_id(CORRELATION_TOKEN) + assert _capture_id() != _capture_id("2" * 32) + + class _FakeUpstreamHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.0"