From d2e56f52bdf1f83f568d73899db7333660668d4b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 08:41:27 +0800 Subject: [PATCH] Validate checkpoint capture routes before quota admission Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/checkpoint_wake.py | 2 + benchmark/tests/test_checkpoint_handoff.py | 55 +++++++++++++++++++ loopx/cli_commands/quota_capture.py | 9 ++- loopx/control_plane/heartbeat/host.py | 2 +- .../turn_driver/decision_capture.py | 42 +++++++++++++- 5 files changed, 106 insertions(+), 4 deletions(-) diff --git a/benchmark/runtime/checkpoint_wake.py b/benchmark/runtime/checkpoint_wake.py index 6ba8788b6c..dd62124ac9 100644 --- a/benchmark/runtime/checkpoint_wake.py +++ b/benchmark/runtime/checkpoint_wake.py @@ -18,6 +18,7 @@ from pathlib import Path from loopx.control_plane.effect_runtime import effect_runtime_result +from loopx.control_plane.turn_driver.decision_capture import bind_handoff_capture_root from .checkpoint import _mark_replan_attempt, _participant_id, held_quota_decision, read_checkpoint_request from .checkpoint_handoff import CHECKPOINT_CAPTURE_SECONDS @@ -55,6 +56,7 @@ def __enter__(self): self.listener = self.stack.enter_context(socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)) self.listener.bind(self.socket_path) self.listener.listen(1) + bind_handoff_capture_root(self.socket_path, self.capture_root) return self except BaseException: self.stack.close() diff --git a/benchmark/tests/test_checkpoint_handoff.py b/benchmark/tests/test_checkpoint_handoff.py index e16e0d0cc5..8849b06046 100644 --- a/benchmark/tests/test_checkpoint_handoff.py +++ b/benchmark/tests/test_checkpoint_handoff.py @@ -295,6 +295,61 @@ def controller(root, components, metadata, probe, deadline, **kwargs): assert code.read_text() == "retained source" +@pytest.mark.skipif(os.name != "posix", reason="Requires the Host Unix rendezvous") +def test_foreign_capture_root_does_not_commit_quota_guard(tmp_path): + from pathlib import Path + import subprocess + + from benchmark.runtime.checkpoint import held_quota_decision + from benchmark.runtime.checkpoint_wake import CheckpointWake + from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, TURN_ID, _configure_autonomous_replan_fixture, + _heartbeat_receipt_count, _initialize_fixture_authority, _write_fixture, + ) + + project, runtime, registry = _write_fixture(tmp_path / "authority") + _configure_autonomous_replan_fixture(project, runtime, registry) + _initialize_fixture_authority(project, runtime, provider="sqlite") + request = tmp_path / "checkpoint-request.json" + request.write_text(json.dumps({ + "destination": str(tmp_path / "checkpoints"), "trigger": "fresh_replan", + "point_metadata": {}, "components": [{"id": "workspace", "source_path": str(project)}], + })) + wake_env = { + "LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"), + "LOOPX_CHECKPOINT_REQUEST": str(request), + } + with CheckpointWake(wake_env, {"mode": "heartbeat", "stage": "execute", "turn_id": TURN_ID}) as host: + assert host.listener is not None + host.capture_root.mkdir(parents=True, mode=0o700) + native_context = tmp_path / "native-context" + native_context.mkdir(mode=0o700) + command = [sys.executable, "-m", "loopx.entrypoint", "--format", "json", + "--registry", str(registry), "--runtime-root", str(runtime), "quota", "should-run", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", TURN_ID, + "--runtime-profile", "generic_cli", "--context-delivery", "off", + "--decision-handoff-socket", host.socket_path, "--decision-output-root"] + env = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[2])} + refused = subprocess.run([*command, str(native_context)], cwd=project, env=env, + capture_output=True, text=True, timeout=15, check=False) + assert refused.returncode == 1 + assert "capture root differs from the current Host wake" in refused.stdout + refused.stderr + assert not list(native_context.iterdir()) + assert _heartbeat_receipt_count(runtime, TURN_ID) == 0 + + host.listener.settimeout(15) + with subprocess.Popen([*command, str(host.capture_root)], cwd=project, env=env, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as process: + with held_quota_decision(host.listener, host.capture_root, goal_id=GOAL_ID, + agent_id=AGENT_ID, turn_instance_id=TURN_ID, + interrupt_when=lambda _: False) as held: + assert held["quota"]["ok"] is True + assert held["path"].parent.parent == host.capture_root + stdout, stderr = process.communicate(timeout=15) + assert process.returncode == 0, (stdout, stderr) + assert _heartbeat_receipt_count(runtime, TURN_ID) == 1 + + @pytest.mark.skipif(sys.platform != "linux", reason="Qualifies the Linux writer process group") @pytest.mark.parametrize("use_selector,closeout,settlement_only", [ (False, False, False), (True, False, False), (True, True, False), (True, False, True), diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 512335a3e5..9488ec93f8 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -510,8 +510,13 @@ def prepare_decision_capture(args: argparse.Namespace) -> Path | None: raise QuotaCommandValidationError( "native Codex context delivery needs JSON and a capture destination") if handoff is not None: - validate_handoff_socket(handoff) - if requested is None and root is None: + handoff_root = None + if root is not None: + handoff_root = Path(root).expanduser().absolute() + elif requested is not None: + handoff_root = Path(requested).expanduser().absolute().parent + validate_handoff_socket(handoff, capture_root=handoff_root) + if handoff_root is None: raise QuotaCommandValidationError("decision handoff requires an explicit decision capture destination") if requested is None and root is None: return None diff --git a/loopx/control_plane/heartbeat/host.py b/loopx/control_plane/heartbeat/host.py index 0900fe27e0..ddf5dffab5 100644 --- a/loopx/control_plane/heartbeat/host.py +++ b/loopx/control_plane/heartbeat/host.py @@ -58,7 +58,7 @@ def resolve_heartbeat_capture_root( if decision_handoff_socket is not None: from ..turn_driver.decision_capture import validate_handoff_socket - validate_handoff_socket(decision_handoff_socket) + validate_handoff_socket(decision_handoff_socket, capture_root=directory) return directory diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index cf6c63a147..898a4022c3 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -229,7 +229,45 @@ def prepare_turn_decision_capture( decision=decision, existing_journal=existing_journal).publish() -def validate_handoff_socket(value: str) -> Path: +_HANDOFF_CAPTURE_ROOT_LIMIT = 4096 + + +def _handoff_capture_root_file(endpoint: Path) -> Path: + return endpoint.with_name(endpoint.name + ".capture-root") + + +def bind_handoff_capture_root(socket_path: str, capture_root: Path) -> None: + """Publish the current Host's transport destination beside its private socket.""" + endpoint = validate_handoff_socket(socket_path) + raw = os.fsencode(capture_root) + if capture_root.resolve() != capture_root or len(raw) > _HANDOFF_CAPTURE_ROOT_LIMIT: + raise ValueError("decision handoff requires a physical absolute capture root") + binding = _handoff_capture_root_file(endpoint) + with os.fdopen(os.open(binding, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as stream: + stream.write(raw) + + +def _validate_handoff_capture_root(endpoint: Path, requested_root: Path) -> None: + try: + with os.fdopen(open_regular_capture(_handoff_capture_root_file(endpoint)), "rb") as stream: + metadata = os.fstat(stream.fileno()) + if metadata.st_uid != os.getuid() or metadata.st_mode & 0o077: + raise ValueError("capture root binding must be private to the Host user") + bound_root = stream.read(_HANDOFF_CAPTURE_ROOT_LIMIT + 1) + if not bound_root or len(bound_root) > _HANDOFF_CAPTURE_ROOT_LIMIT: + raise ValueError("invalid capture root binding size") + except FileNotFoundError: + return # Earlier Hosts still validate the capture when accepting it. + except (OSError, ValueError) as exc: + raise QuotaCommandValidationError(f"invalid decision handoff binding; no guard was run: {exc}") from exc + if bound_root != os.fsencode(requested_root): + raise QuotaCommandValidationError( + "decision handoff capture root differs from the current Host wake; no guard was run. " + "Use the --decision-output-root named by the current wake." + ) + + +def validate_handoff_socket(value: str, *, capture_root: Path | None = None) -> Path: """A caller-owned local transport, not an authority or admission endpoint.""" path = Path(value) try: @@ -250,6 +288,8 @@ def validate_handoff_socket(value: str) -> Path: ) from exc except OSError as exc: raise QuotaCommandValidationError(f"invalid decision handoff socket: {exc}") from exc + if capture_root is not None: + _validate_handoff_capture_root(path, capture_root) return path