From 0ea81a06dc51211e5d91587d818fd05cd2039703 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 23 Oct 2026 06:26:41 -0500 Subject: [PATCH] wip(local-api): prove session identity and serialize launches Co-Authored-By: Claude Opus 5.5 --- backend/scripts/local_capture_api.py | 80 +++++++++++++--- backend/tests/test_local_capture_api.py | 121 +++++++++++++++++++++++- 2 files changed, 187 insertions(+), 14 deletions(-) diff --git a/backend/scripts/local_capture_api.py b/backend/scripts/local_capture_api.py index a710e7ec..7bbf12c6 100644 --- a/backend/scripts/local_capture_api.py +++ b/backend/scripts/local_capture_api.py @@ -33,6 +33,8 @@ from __future__ import annotations import argparse +import fcntl +import hashlib import json import os import secrets @@ -157,10 +159,9 @@ def _pid_alive(pid: object) -> bool: def _live_session(session_path: Path) -> dict | None: """The session at `session_path` if the server that wrote it still serves. - Live means its pid is alive and its loopback port accepts a connection. - Either alone misleads: a crashed server's pid can be reused, and its port - can be taken by another process. A missing or unreadable file is not live; - no client could use its credentials either. + A live pid and an accepting port can both belong to unrelated processes + after a crash. The local-only identity endpoint must also return the hash + of this session's reviewer token. The probe never sends that token. """ try: session = json.loads(session_path.read_text()) @@ -175,11 +176,18 @@ def _live_session(session_path: Path) -> dict | None: return None if host not in LOOPBACK_HOSTS or port is None: return None - try: - with socket.create_connection((host, port), timeout=1): - return session - except OSError: + reviewer_token = session.get("reviewerToken") + if not isinstance(reviewer_token, str) or not reviewer_token: return None + identity = hashlib.sha256(reviewer_token.encode()).hexdigest() + try: + with _DIRECT.open(f"{_public_url(host, port)}/v1/local-session", timeout=1) as response: + proof = json.loads(response.read(256)) + if response.status == 200 and isinstance(proof, dict) and proof.get("identity") == identity: + return session + except (OSError, ValueError): + pass + return None # The readiness probe must reach the listener itself. urlopen honors @@ -307,8 +315,52 @@ def main(argv: list[str] | None = None) -> int: return 2 session_dir = args.session_dir - volume_root = session_dir / "volume" session_path = session_dir / "session.json" + # Keep this lock file in place: unlinking it could let another launcher lock + # a different inode while this server still owns the original one. The OS + # releases flock on exit, including a crash, so no stale-lock recovery is needed. + session_dir.mkdir(parents=True, exist_ok=True) + fd = os.open(session_dir / "session.lock", os.O_RDWR | os.O_CREAT, 0o600) + with os.fdopen(fd, "a") as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + print( + f"refusing to start: {session_path} is locked by a starting or live local capture API. " + "Stop it first, or pass another --session-dir.", + file=sys.stderr, + ) + return 1 + return _serve(args, session_path) + + +def _local_app(reviewer_token: str): + """Add identity proof only to the loopback launcher's app, not deployed APIs.""" + from fastapi.responses import JSONResponse + + from server.api.app import create_app + + identity = hashlib.sha256(reviewer_token.encode()).hexdigest() + + class _Identity: + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope["type"] == "http" and scope["method"] == "GET" and scope["path"] == "/v1/local-session": + await JSONResponse({"identity": identity})(scope, receive, send) + else: + await self.app(scope, receive, send) + + app = create_app() + # Wrap the API's bearer gate: the probe reads a hash, never a credential, + # and must not send the session's token to an unverified listener. + app.add_middleware(_Identity) + return app + + +def _serve(args: argparse.Namespace, session_path: Path) -> int: + volume_root = session_path.parent / "volume" # A second launch on a free port would bind and replace a running server's # credentials, so the failed-bind guard below never fires; refuse instead. live = _live_session(session_path) @@ -371,14 +423,16 @@ def main(argv: list[str] | None = None) -> int: # HSS_LOCAL_OUTCOME and HSS_LOCAL_STAGE_DELAY_S pass through untouched. config = uvicorn.Config( - "server.api.app:create_app", - factory=True, + _local_app(reviewer_token), host=args.host, port=port, log_level="info", ) - _announce(base, session_path, args.health_timeout) - uvicorn.Server(config).run(sockets=[listener]) + try: + _announce(base, session_path, args.health_timeout) + uvicorn.Server(config).run(sockets=[listener]) + finally: + listener.close() return 0 diff --git a/backend/tests/test_local_capture_api.py b/backend/tests/test_local_capture_api.py index bd8e80d2..0d1428bb 100644 --- a/backend/tests/test_local_capture_api.py +++ b/backend/tests/test_local_capture_api.py @@ -492,6 +492,125 @@ def test_second_launch_refuses_a_live_session(tmp_path: Path) -> None: _stop(first) +def test_reused_pid_and_unrelated_listener_allow_startup(tmp_path: Path) -> None: + launcher = _launcher_module() + listener = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _AnswerEverything) + listener.hits = [] + thread = threading.Thread(target=listener.serve_forever, daemon=True) + thread.start() + proc = None + try: + path = tmp_path / "session.json" + path.write_text( + json.dumps( + { + "baseUrl": f"http://127.0.0.1:{listener.server_port}/v1", + "reviewerToken": "session-only-token", + "pid": os.getpid(), + } + ) + ) + assert launcher._live_session(path) is None + proc = _start_server(tmp_path, tmp_path / "new.log") + session = json.loads(path.read_text()) + assert session["pid"] == proc.pid + assert launcher._live_session(path) == session + finally: + _stop(proc) + listener.shutdown() + listener.server_close() + thread.join(timeout=5) + + +def test_live_session_requires_the_running_servers_identity(tmp_path: Path) -> None: + launcher = _launcher_module() + session_dir = tmp_path / "real" + proc = _start_server(session_dir, tmp_path / "real.log") + try: + session = json.loads((session_dir / "session.json").read_text()) + copied_path = tmp_path / "session.json" + copied_path.write_text(json.dumps(session)) + assert launcher._live_session(copied_path) == session + session["reviewerToken"] = "not-the-running-servers-token" + copied_path.write_text(json.dumps(session)) + assert launcher._live_session(copied_path) is None + finally: + _stop(proc) + + +def test_concurrent_launches_allow_exactly_one(tmp_path: Path) -> None: + # Pause the first launch immediately after its liveness check. Without the + # lifetime lock, the second launch can also pass the same check before + # either publishes a session. With the lock, it exits before the check. + wrapper = """ +import importlib.util +import os +from pathlib import Path +import sys +import time +spec = importlib.util.spec_from_file_location("launcher", sys.argv[1]) +launcher = importlib.util.module_from_spec(spec) +spec.loader.exec_module(launcher) +original = launcher._live_session +root = Path(sys.argv[2]) +def paused(path): + result = original(path) + (root / f"checked-{os.getpid()}").touch() + deadline = time.monotonic() + 20 + while not (root / "release").exists(): + if time.monotonic() >= deadline: + raise RuntimeError("test did not release launch barrier") + time.sleep(0.01) + return result +launcher._live_session = paused +raise SystemExit(launcher.main(["--port", "0", "--session-dir", str(root), "--health-timeout", "20"])) +""" + processes = [] + logs = [] + try: + for index in range(2): + log_path = tmp_path / f"launch-{index}.log" + logs.append(log_path) + with log_path.open("w") as log: + proc = subprocess.Popen( + [sys.executable, "-c", wrapper, str(SCRIPT), str(tmp_path)], + cwd=ROOT, + stdout=log, + stderr=subprocess.STDOUT, + text=True, + ) + processes.append(proc) + deadline = time.monotonic() + 20 + while not (tmp_path / f"checked-{proc.pid}").exists() and proc.poll() is None: + assert time.monotonic() < deadline, log_path.read_text() + time.sleep(0.01) + (tmp_path / "release").touch() + deadline = time.monotonic() + 30 + session_path = tmp_path / "session.json" + launcher = _launcher_module() + while True: + live = launcher._live_session(session_path) + if live is not None: + break + assert time.monotonic() < deadline, [path.read_text() for path in logs] + time.sleep(0.05) + winners = [proc for proc in processes if proc.poll() is None] + assert len(winners) == 1, [path.read_text() for path in logs] + assert live["pid"] == winners[0].pid + loser = next(proc for proc in processes if proc is not winners[0]) + assert loser.wait(timeout=5) == 1 + loser_log = logs[processes.index(loser)].read_text() + assert "refusing to start" in loser_log + # A dead launch must not leave an exclusive lock behind. + _stop(winners[0]) + restarted = _start_server(tmp_path, tmp_path / "restart.log") + processes.append(restarted) + finally: + (tmp_path / "release").touch() + for proc in processes: + _stop(proc) + + def test_live_session_needs_a_live_pid_and_an_answering_port(tmp_path: Path) -> None: launcher = _launcher_module() path = tmp_path / "session.json" @@ -508,7 +627,7 @@ def test_live_session_needs_a_live_pid_and_an_answering_port(tmp_path: Path) -> exited.wait() try: cases = [ - (os.getpid(), f"127.0.0.1:{open_port}", True), + (os.getpid(), f"127.0.0.1:{open_port}", False), # A crashed server whose pid was reused: nothing listens. (os.getpid(), f"127.0.0.1:{refusing.getsockname()[1]}", False), # The pid is gone; whatever owns the port now is not this session.