fix(harbor): require an admin key on the capture proxy control plane

src/openenv/core/harness/capture/server.py

@@ -572,7 +572,7 @@ def create_app(
             from this app is an eval rollout.
         admin_key (`str`, *optional*):
             Required by the session-management routes when set. Leave unset for a private port; set it
-            whenever this app is reachable from outside, which `serve` does automatically.
+            whenever this app is reachable from outside, which `serve` and `rollout` do automatically.
         max_model_calls (`int`, *optional*, defaults to `0`):
             Default ceiling on model calls per session; `0` is unlimited, and a session may name its
             own. Once a rollout reaches it the proxy answers a terminal completion itself, which ends
@@ -1348,6 +1348,14 @@ def _ingest(
     return app
 
 
+def _is_loopback_host(host: str) -> bool:
+    """Whether `host` is a loopback-only bind (private local port).
+
+    `0.0.0.0` / `::` are intentionally not loopback: they publish on every interface.
+    """
+    return host in {"127.0.0.1", "::1", "localhost"}
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(
         description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
@@ -1366,7 +1374,12 @@ def main() -> None:
         "upstream may be a hosted provider, and reporting one of these two when it is not says "
         "something untrue about capture.",
     )
-    parser.add_argument("--host", default="0.0.0.0")
+    parser.add_argument(
+        "--host",
+        default="0.0.0.0",
+        help="bind address (default: 0.0.0.0). Non-loopback binds mint an admin key when none is "
+        "set so /sessions* is never left open on a reachable interface.",
+    )
     parser.add_argument("--port", type=int, default=8100)
     parser.add_argument(
         "--max-output-tokens",
@@ -1398,6 +1411,13 @@ def main() -> None:
         help="what the upstream can return. Probed from the endpoint when omitted, which is the "
         "recommended path; pass it only to force a level.",
     )
+    parser.add_argument(
+        "--admin-key",
+        default=os.environ.get("OPENENV_CAPTURE_ADMIN_KEY", ""),
+        help="key the session-management routes (/sessions*) require (defaults to "
+        "$OPENENV_CAPTURE_ADMIN_KEY). Required for non-loopback binds: if unset there, a random "
+        "key is minted and printed. Loopback-only binds may leave it unset.",
+    )
     args = parser.parse_args()
 
     import uvicorn
@@ -1423,6 +1443,17 @@ def main() -> None:
         for fix in report.param_fixes:
             print(f"  upstream compat: {fix}")
 
+    # Flag/env win; otherwise mint when the bind is reachable from outside. Leaving the key unset
+    # with --host 0.0.0.0 (the CLI default) would publish /sessions* ungated — the same open
+    # control plane `run_batch` already refuses. Loopback stays fail-open for private local use.
+    admin_key = args.admin_key or None
+    if not admin_key and not _is_loopback_host(args.host):
+        admin_key = secrets.token_urlsafe(32)
+        print(
+            f"capture admin key (minted for --host={args.host}; "
+            f"set --admin-key or $OPENENV_CAPTURE_ADMIN_KEY to pin): {admin_key}"
+        )
+
     uvicorn.run(
         create_app(
             llm_url=args.llm_url,
@@ -1433,6 +1464,7 @@ def main() -> None:
             api_key=args.api_key or None,
             auth_header=args.auth_header,
             capture_level=level,
+            admin_key=admin_key,
         ),
         host=args.host,
         port=args.port,

src/openenv/harbor/runner.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import os
+import secrets
 from pathlib import Path
 
 from openenv.core.harness.capture import CaptureServer
@@ -42,6 +43,7 @@ async def run_batch(
     api_key: str | None = None,
     auth_header: str = "Authorization",
     provider: str = "openai",
+    admin_key: str | None = None,
 ) -> list[HarborRolloutResult]:
     """Run `task_indices` from `dataset` and print a per-rollout report.
 
@@ -58,6 +60,10 @@ async def run_batch(
             Harbor environment type.
         expose (`str`, *optional*, defaults to `"gradio"`):
             How the sandbox reaches the capture proxy: `gradio`, `cloudflare` or `direct`.
+        admin_key (`str`, *optional*):
+            Key the capture proxy's session-management routes require. Defaults to
+            `$OPENENV_CAPTURE_ADMIN_KEY`, else a random one minted for this batch. The proxy is
+            published on a public URL, so these routes are never left open.
 
     Returns:
         `list[HarborRolloutResult]`: One per index, in order.
@@ -94,6 +100,17 @@ async def run_batch(
     trials_dir = trials_dir or Path("/tmp/openenv-harbor-trials")
     trials_dir.mkdir(parents=True, exist_ok=True)
 
+    # The forwarder below puts the proxy on a public URL, and `_admin_ok` waves every caller through
+    # when no key is set — so an unset key here is an open control plane: anyone with the URL could
+    # list rollouts, read their tokens, delete them, or mint a session key the proxy then honours.
+    # Rollouts never go through those routes (they use the in-process registry), so the key is only
+    # for an operator, and a random one costs nothing. Same resolution as `HarborService`.
+    admin_key = (
+        admin_key
+        or os.environ.get("OPENENV_CAPTURE_ADMIN_KEY")
+        or secrets.token_urlsafe(32)
+    )
+
     capture = CaptureServer(
         llm_url=llm_url,
         model=model,
@@ -102,6 +119,7 @@ async def run_batch(
         auth_header=auth_header,
         provider=provider,
         capture_level=capture_level,
+        admin_key=admin_key,
     )
     capture.start()
     # The capture proxy is already listening on a bound port in a background thread, so an exception
@@ -116,6 +134,9 @@ async def run_batch(
         capture.stop()
         raise
     print(f"\ncapture  :{port} -> {public_url}  ({forwarder.name})")
+    print(
+        "         session routes are gated; set OPENENV_CAPTURE_ADMIN_KEY to call them yourself"
+    )
     print(f"trials   {trials_dir}\n")
 
     results: list[HarborRolloutResult] = []

tests/envs/test_harbor_capture_server.py

@@ -13,6 +13,10 @@
 `start()` refuses rather than proceeds.
 
 No credentials and no engine are needed: `start()` binds a socket and never contacts `llm_url`.
+
+The last tests cover the standalone CLI (`python -m openenv.core.harness.capture.server`): it had no
+way to set the admin key, so a port published from it served the session-management routes to
+anyone who could reach it.
 """
 
 from __future__ import annotations
@@ -22,6 +26,7 @@
 import pytest
 
 harbor_runner = pytest.importorskip("openenv.harbor.runner")
+capture_server = pytest.importorskip("openenv.core.harness.capture.server")
 
 CaptureServer = harbor_runner.CaptureServer
 
@@ -123,3 +128,72 @@ def test_stop_releases_the_port(capture):
     successor = capture(port)
     successor.start()
     assert successor.app.state.instance_id != server.app.state.instance_id
+
+
+def run_cli(monkeypatch, *argv: str) -> dict:
+    """Run the CLI entry point with `argv`, returning the kwargs it built the app with.
+
+    `--capture-level` is forced so the CLI never probes `UNUSED_ENGINE`; `uvicorn.run` is stubbed so
+    nothing binds a port.
+    """
+    import uvicorn
+
+    seen: dict = {}
+
+    def fake_create_app(**kwargs):
+        seen.update(kwargs)
+        return object()
+
+    monkeypatch.setattr(capture_server, "create_app", fake_create_app)
+    monkeypatch.setattr(uvicorn, "run", lambda *_a, **_k: None)
+    monkeypatch.setattr(
+        "sys.argv",
+        ["server", "--llm-url", UNUSED_ENGINE, "--capture-level", "tokens", *argv],
+    )
+    capture_server.main()
+    return seen
+
+
+def test_cli_passes_admin_key_flag_to_the_app(monkeypatch):
+    monkeypatch.delenv("OPENENV_CAPTURE_ADMIN_KEY", raising=False)
+
+    seen = run_cli(monkeypatch, "--admin-key", "from-flag")
+
+    assert seen["admin_key"] == "from-flag"
+
+
+def test_cli_admin_key_defaults_to_the_env_var(monkeypatch):
+    """Same variable `serve` and `rollout` read, so one setting covers every entry point."""
+    monkeypatch.setenv("OPENENV_CAPTURE_ADMIN_KEY", "from-env")
+
+    seen = run_cli(monkeypatch)
+
+    assert seen["admin_key"] == "from-env"
+
+
+def test_cli_mints_admin_key_for_default_non_loopback_host(monkeypatch):
+    """Default `--host 0.0.0.0` must never leave `/sessions*` ungated."""
+    monkeypatch.delenv("OPENENV_CAPTURE_ADMIN_KEY", raising=False)
+
+    seen = run_cli(monkeypatch)
+
+    assert seen["admin_key"], "default non-loopback bind must mint an admin key"
+    assert len(seen["admin_key"]) >= 32
+
+
+def test_cli_mints_admin_key_for_explicit_non_loopback_host(monkeypatch):
+    monkeypatch.delenv("OPENENV_CAPTURE_ADMIN_KEY", raising=False)
+
+    seen = run_cli(monkeypatch, "--host", "0.0.0.0")
+
+    assert seen["admin_key"]
+    assert len(seen["admin_key"]) >= 32
+
+
+def test_cli_leaves_admin_key_unset_on_loopback_without_flag_or_env(monkeypatch):
+    """A private loopback port stays as convenient as before; no key is minted."""
+    monkeypatch.delenv("OPENENV_CAPTURE_ADMIN_KEY", raising=False)
+
+    seen = run_cli(monkeypatch, "--host", "127.0.0.1")
+
+    assert seen["admin_key"] is None

tests/envs/test_harbor_runner_lifecycle.py

@@ -12,11 +12,18 @@
 that named nothing about the actual cause. Teardown had the mirror of the same gap: `forwarder.stop()`
 ran before `capture.stop()` without a guard, so a tunnel that failed to shut down took the port with it.
 
+The same entry point also publishes the proxy through a tunnel, so its session-management routes are
+reachable by anyone with the URL. `_admin_ok` admits every caller while no admin key is set, which
+made `run_batch` an open control plane: the tests at the bottom pin that it always hands
+`CaptureServer` a key.
+
 These tests stub every external dependency; no engine, no sandbox and no network are involved.
 """
 
 from __future__ import annotations
 
+import asyncio
+
 import pytest
 
 runner = pytest.importorskip("openenv.harbor.runner")
@@ -29,8 +36,12 @@ class FakeCaps:
 
 
 def wire(monkeypatch, tmp_path, forwarder):
-    """Replace startup, the capture server and the dataset with local stubs."""
+    """Replace startup, the capture server and the dataset with local stubs.
+
+    Returns the teardown log and the list of kwargs each `CaptureServer` was built with.
+    """
     stopped: list[str] = []
+    built: list[dict] = []
 
     monkeypatch.setattr(runner, "resolve_task_dirs", lambda _d: [tmp_path])
     monkeypatch.setattr(
@@ -42,6 +53,7 @@ class FakeCapture:
         inference = None
 
         def __init__(self, **kwargs):
+            built.append(kwargs)
             self.port = kwargs.get("port", 8100)
 
         def start(self):
@@ -56,7 +68,22 @@ def stop(self):
         lambda _kind: forwarder(stopped),
         raising=False,
     )
-    return stopped
+    return stopped, built
+
+
+class Quiet:
+    """A forwarder that starts and stops without incident."""
+
+    name = "gradio"
+
+    def __init__(self, _stopped):
+        pass
+
+    def start(self, _port):
+        return "https://tunnel.invalid"
+
+    def stop(self):
+        pass
 
 
 def test_a_forwarder_that_cannot_start_releases_the_port(monkeypatch, tmp_path):
@@ -69,10 +96,10 @@ def __init__(self, _stopped):
         def start(self, _port):
             raise RuntimeError("cloudflared is not installed")
 
-    stopped = wire(monkeypatch, tmp_path, Exploding)
+    stopped, _ = wire(monkeypatch, tmp_path, Exploding)
 
     with pytest.raises(RuntimeError, match="cloudflared"):
-        __import__("asyncio").run(
+        asyncio.run(
             runner.run_batch(llm_url="http://x/v1", dataset="d", task_indices=[0])
         )
 
@@ -95,16 +122,72 @@ def stop(self):
             self._stopped.append("forwarder")
             raise RuntimeError("tunnel already gone")
 
-    stopped = wire(monkeypatch, tmp_path, BadTeardown)
+    stopped, _ = wire(monkeypatch, tmp_path, BadTeardown)
     monkeypatch.setattr(
         runner, "run_rollout", None
     )  # never reached: no indices are in range
 
     with pytest.raises(RuntimeError, match="tunnel already gone"):
-        __import__("asyncio").run(
+        asyncio.run(
             runner.run_batch(llm_url="http://x/v1", dataset="d", task_indices=[99])
         )
 
     assert stopped == ["forwarder", "capture"], (
         "capture.stop() must run even when the forwarder's teardown raises"
     )
+
+
+def run_empty_batch(**kwargs) -> None:
+    """Drive `run_batch` through capture + forwarder setup with no rollout in range."""
+    asyncio.run(
+        runner.run_batch(
+            llm_url="http://x/v1", dataset="d", task_indices=[99], **kwargs
+        )
+    )
+
+
+def test_run_batch_always_gates_the_control_plane(monkeypatch, tmp_path):
+    """The regression: the proxy went out on a public tunnel with `admin_key` unset.
+
+    `_admin_ok` returns True for every caller while the key is unset, so `POST /sessions` minted keys
+    for anyone (an open relay to the upstream) and `GET /sessions/{id}/rollout` served token-level
+    training data to anyone who found the URL.
+    """
+    monkeypatch.delenv("OPENENV_CAPTURE_ADMIN_KEY", raising=False)
+    _, built = wire(monkeypatch, tmp_path, Quiet)
+
+    run_empty_batch()
+
+    (kwargs,) = built
+    assert kwargs.get("admin_key"), "CaptureServer was built without an admin key"
+    assert len(kwargs["admin_key"]) >= 32
+
+
+def test_run_batch_mints_a_fresh_key_per_batch(monkeypatch, tmp_path):
+    monkeypatch.delenv("OPENENV_CAPTURE_ADMIN_KEY", raising=False)
+    _, built = wire(monkeypatch, tmp_path, Quiet)
+
+    run_empty_batch()
+    run_empty_batch()
+
+    first, second = (k["admin_key"] for k in built)
+    assert first != second
+
+
+def test_run_batch_forwards_an_explicit_admin_key(monkeypatch, tmp_path):
+    monkeypatch.setenv("OPENENV_CAPTURE_ADMIN_KEY", "from-env")
+    _, built = wire(monkeypatch, tmp_path, Quiet)
+
+    run_empty_batch(admin_key="explicit")
+
+    assert built[0]["admin_key"] == "explicit", "the argument must beat the env var"
+
+
+def test_run_batch_honours_the_admin_key_env_var(monkeypatch, tmp_path):
+    """Same variable `HarborService` reads, so one setting covers `serve` and `rollout`."""
+    monkeypatch.setenv("OPENENV_CAPTURE_ADMIN_KEY", "from-env")
+    _, built = wire(monkeypatch, tmp_path, Quiet)
+
+    run_empty_batch()
+
+    assert built[0]["admin_key"] == "from-env"