fix(web): exchange the launch token for a short-lived session (F-04)

replicant/web/server.py

@@ -19,8 +19,13 @@
 
 Access: the server binds loopback by default but any local address is allowed, so the
 controls do not assume a local-only listener. Every API and websocket call requires
-the persistent token, accepted as a Bearer header, an ``X-Replicant-Token`` header, a
-query parameter, or an httpOnly ``SameSite=Strict`` session cookie. A middleware
+a credential: either the persistent launch token (a Bearer header, an
+``X-Replicant-Token`` header, or a query parameter), or the httpOnly
+``SameSite=Strict`` session cookie the server issues in exchange for it. The cookie
+holds a short-lived random id from :class:`SessionStore`, never the launch token
+itself, so it expires, can be revoked one browser at a time, and is worth nothing
+once it lapses. The browser therefore never puts a credential in a URL, which is
+what kept the launch token out of server logs, history and Referer. A middleware
 rejects any Host that is not the bind address, loopback, or an explicitly allowed
 name (the DNS-rebinding guard). Because the cookie is the only credential a browser
 attaches by itself, a cookie-authenticated write must also carry a matching Origin.
@@ -43,6 +48,7 @@
 import socket
 import sys
 import threading
+import time
 import webbrowser
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
@@ -120,6 +126,67 @@ class DocPage:
 # Set on the first authenticated load so the token does not have to live in the
 # URL bar for the rest of the session.
 SESSION_COOKIE = "replicant_session"
+
+#: How long a browser session is good for. Long enough that an operator running a
+#: four hour plan-paced technique is not logged out mid-run, short enough that a
+#: cookie copied off a shared machine is not a permanent credential.
+SESSION_TTL_S = 12 * 3600
+
+
+class SessionStore:
+    """Short-lived random session ids, exchanged for the persistent launch token.
+
+    F-04: the cookie used to hold the launch token itself, the same value that
+    sits in ``~/.config/replicant/web-token``. That made the cookie the master
+    credential: no expiry, no rotation, and no way to revoke one browser without
+    regenerating the token file and breaking every other client.
+
+    An id here grants the same access while it lives, and nothing once it does
+    not. The launch token remains the bootstrap and the only thing a non-browser
+    client needs, because a script cannot run a cookie jar.
+
+    No lock: every caller is a coroutine on one event loop.
+    """
+
+    def __init__(self, ttl_s: int = SESSION_TTL_S, clock: Any = None) -> None:
+        self.ttl_s = ttl_s
+        self._clock = clock or time.monotonic
+        self._expiry: dict[str, float] = {}
+
+    def __len__(self) -> int:
+        return len(self._expiry)
+
+    def _sweep(self) -> None:
+        """Drop expired ids. Called on issue, so a reconnect loop cannot grow this
+        without bound on a long-lived server."""
+        now = self._clock()
+        for sid in [s for s, exp in self._expiry.items() if exp <= now]:
+            del self._expiry[sid]
+
+    def issue(self) -> str:
+        self._sweep()
+        sid = secrets.token_urlsafe(32)
+        self._expiry[sid] = self._clock() + self.ttl_s
+        return sid
+
+    def validate(self, sid: str) -> bool:
+        if not sid:
+            return False
+        expiry = self._expiry.get(sid)
+        if expiry is None:
+            return False
+        if expiry <= self._clock():
+            del self._expiry[sid]
+            return False
+        return True
+
+    def revoke(self, sid: str) -> None:
+        self._expiry.pop(sid, None)
+
+    def revoke_all(self) -> None:
+        self._expiry.clear()
+
+
 _SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
 
 
@@ -420,6 +487,9 @@ def create_app(
     # Idempotent, so a test that builds several apps does not stack handlers.
     obs_log.install()
     manager = RunManager(catalog, settings)
+    # Per app, not module-global: two apps in one process (the test suite builds
+    # several) must not be able to authenticate each other's browsers.
+    sessions = SessionStore()
     base_orchestrator = Orchestrator(catalog, settings)
 
     def _resolve_vendor(vendor: str | None) -> str:
@@ -466,11 +536,15 @@ def _authenticated_source(request: HTTPConnection) -> str | None:
             ("header", bearer.strip() if scheme.lower() == "bearer" else ""),
             ("header", request.headers.get("x-replicant-token") or ""),
             ("query", request.query_params.get("token") or ""),
-            ("cookie", request.cookies.get(SESSION_COOKIE) or ""),
         )
         for source, supplied in candidates:
             if supplied and secrets.compare_digest(token, supplied):
                 return source
+        # The cookie is checked against the session store, never against the
+        # launch token. It used to hold that token verbatim, which made a value
+        # the browser stores indefinitely into the master credential (F-04).
+        if sessions.validate(request.cookies.get(SESSION_COOKIE) or ""):
+            return "cookie"
         return None
 
     def _origin_ok(connection: HTTPConnection, *, required: bool) -> bool:
@@ -513,10 +587,16 @@ async def _session_cookie(request: Request, call_next: Any) -> Any:
         if source is not None and source != "cookie" and response.status_code < 400:
             response.set_cookie(
                 SESSION_COOKIE,
-                token,
+                # A fresh short-lived id, not the launch token. See SessionStore.
+                sessions.issue(),
                 httponly=True,
                 samesite="strict",
                 path="/",
+                max_age=sessions.ttl_s,
+                # Only over https, where it means anything. Setting it on the
+                # loopback http the tool serves by default would stop the cookie
+                # being sent at all, which is a worse outcome than not setting it.
+                secure=request.url.scheme == "https",
             )
         return response
 
@@ -805,6 +885,23 @@ def start_run(body: RunBody) -> dict[str, Any]:
             ),
         }
 
+    @app.post("/api/session/logout")
+    def logout(request: Request) -> JSONResponse:
+        """End this browser's session without touching the launch token.
+
+        The point of the exchange: revoking one browser used to mean
+        regenerating the token file, which logged out every other client and
+        every script. Deliberately unauthenticated, because presenting a session
+        id you want destroyed is not something to gate: the worst an attacker can
+        do is end a session they already hold.
+        """
+
+        sid = request.cookies.get(SESSION_COOKIE) or ""
+        sessions.revoke(sid)
+        response = JSONResponse({"ok": True})
+        response.delete_cookie(SESSION_COOKIE, path="/")
+        return response
+
     # Declared before /api/runs/{run_id}: FastAPI matches in registration order,
     # and the parameterised route would otherwise swallow "active" as an id.
     @app.get("/api/runs/active", dependencies=[Depends(require_token)])

tests/test_web_access.py

@@ -66,6 +66,21 @@ def make_client(
     return TestClient(app, base_url=base_url)
 
 
+def establish_session(client: TestClient) -> None:
+    """Authenticate once with the launch token so the server issues a session.
+
+    These tests used to write the launch token straight into the cookie, because
+    that is literally what the cookie held. F-04 replaced it with a short-lived
+    id the server mints, so the only way to get a valid cookie is to be given
+    one. That is the point of the change, and it is why this helper exists rather
+    than a constant.
+    """
+
+    resp = client.get("/api/health", params={"token": TOKEN})
+    assert resp.status_code == 200
+    assert client.cookies.get(SESSION_COOKIE)
+
+
 @pytest.fixture()
 def config_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
     """Point the config dir at a temp path so tests never touch ~/.config."""
@@ -193,7 +208,7 @@ def test_bearer_header_authenticates(tmp_path: Path) -> None:
 
 def test_session_cookie_authenticates(tmp_path: Path) -> None:
     client = make_client(tmp_path)
-    client.cookies.set(SESSION_COOKIE, TOKEN)
+    establish_session(client)
 
     assert client.get("/api/catalog").status_code == 200
 
@@ -264,7 +279,7 @@ def test_cookie_authenticated_write_requires_a_matching_origin(tmp_path: Path) -
     # cross-site POST automatically, which a token in a header or query string
     # never was.
     client = make_client(tmp_path)
-    client.cookies.set(SESSION_COOKIE, TOKEN)
+    establish_session(client)
 
     resp = client.post(
         "/api/runs",
@@ -277,7 +292,7 @@ def test_cookie_authenticated_write_requires_a_matching_origin(tmp_path: Path) -
 
 def test_cookie_authenticated_write_accepts_a_same_origin_request(tmp_path: Path) -> None:
     client = make_client(tmp_path)
-    client.cookies.set(SESSION_COOKIE, TOKEN)
+    establish_session(client)
 
     resp = client.post(
         "/api/runs",
@@ -292,7 +307,7 @@ def test_cookie_authenticated_write_without_an_origin_is_refused(tmp_path: Path)
     # Browsers send Origin on every non-GET. A cookie-authenticated write with no
     # Origin at all is not something the SPA produces, so it fails closed.
     client = make_client(tmp_path)
-    client.cookies.set(SESSION_COOKIE, TOKEN)
+    establish_session(client)
 
     resp = client.post("/api/runs", json={"technique_id": "REP-001", "no_send": True})
 
@@ -316,7 +331,7 @@ def test_header_authenticated_write_needs_no_origin(tmp_path: Path) -> None:
 
 def test_cookie_authenticated_read_needs_no_origin(tmp_path: Path) -> None:
     client = make_client(tmp_path)
-    client.cookies.set(SESSION_COOKIE, TOKEN)
+    establish_session(client)
 
     assert client.get("/api/catalog").status_code == 200
 

tests/test_web_sessions.py

@@ -0,0 +1,163 @@
+# Copyright 2026 Imran Hafeez (RZA)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""The browser cookie must not be the master token.
+
+F-04 of the 2026-08 security review. The launch token is persistent, lives in
+``~/.config/replicant/web-token``, and was ALSO the value written into the
+session cookie. So the cookie was the master credential itself: it never
+expired, could not be rotated, could not be revoked without regenerating the
+token file, and anything that read it held permanent access.
+
+The same token was additionally placed in EventSource and WebSocket query
+strings, where URLs are the least private part of a request: they reach server
+logs, browser history, and the Referer header.
+
+This replaces it with an exchange. The launch token still bootstraps, but what
+the browser then holds is a short-lived random session id that the server can
+expire, rotate and revoke, and that grants nothing if it leaks after expiry.
+
+The rule this review adopted applies: each guard here was run against the
+unfixed code and observed to fail.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("fastapi")
+
+from fastapi.testclient import TestClient  # noqa: E402
+
+from replicant.config.settings import Settings  # noqa: E402
+from replicant.core.models import load_catalog  # noqa: E402
+from replicant.resources import TECHNIQUE_CATALOG  # noqa: E402
+from replicant.web.server import SESSION_COOKIE, SessionStore, create_app  # noqa: E402
+
+TOKEN = "launch-token-value"
+CATALOG = load_catalog(TECHNIQUE_CATALOG)
+
+
+@pytest.fixture()
+def client(tmp_path: Path) -> TestClient:
+    app = create_app(CATALOG, Settings(manifest_dir=str(tmp_path)), token=TOKEN)
+    return TestClient(app, base_url="http://localhost")
+
+
+class TestSessionStore:
+    def test_issued_ids_are_not_the_launch_token(self) -> None:
+        store = SessionStore(ttl_s=60)
+
+        assert store.issue() != TOKEN
+
+    def test_issued_ids_are_unique(self) -> None:
+        store = SessionStore(ttl_s=60)
+
+        assert len({store.issue() for _ in range(50)}) == 50
+
+    def test_a_fresh_id_validates(self) -> None:
+        store = SessionStore(ttl_s=60)
+
+        assert store.validate(store.issue()) is True
+
+    def test_an_unknown_id_does_not(self) -> None:
+        store = SessionStore(ttl_s=60)
+
+        assert store.validate("not-a-session") is False
+        assert store.validate("") is False
+
+    def test_an_expired_id_stops_working(self) -> None:
+        """The whole point: a leaked cookie has a shelf life."""
+        clock = {"now": 1000.0}
+        store = SessionStore(ttl_s=60, clock=lambda: clock["now"])
+        sid = store.issue()
+
+        clock["now"] += 61
+
+        assert store.validate(sid) is False
+
+    def test_revoking_one_id_leaves_the_others(self) -> None:
+        store = SessionStore(ttl_s=60)
+        keep, drop = store.issue(), store.issue()
+
+        store.revoke(drop)
+
+        assert store.validate(drop) is False
+        assert store.validate(keep) is True
+
+    def test_revoke_all_ends_every_session(self) -> None:
+        store = SessionStore(ttl_s=60)
+        ids = [store.issue() for _ in range(3)]
+
+        store.revoke_all()
+
+        assert not any(store.validate(i) for i in ids)
+
+    def test_expired_entries_do_not_accumulate(self) -> None:
+        """Otherwise a reconnect loop is an unbounded dict on a long-lived server."""
+        clock = {"now": 1000.0}
+        store = SessionStore(ttl_s=10, clock=lambda: clock["now"])
+        for _ in range(100):
+            store.issue()
+            clock["now"] += 1
+
+        store.issue()
+
+        assert len(store) < 100
+
+
+class TestExchange:
+    def test_the_cookie_is_not_the_launch_token(self, client: TestClient) -> None:
+        """The defect, stated directly."""
+        resp = client.get("/api/health", params={"token": TOKEN})
+
+        cookie = resp.cookies.get(SESSION_COOKIE)
+        assert cookie
+        assert cookie != TOKEN
+
+    def test_the_cookie_authenticates_on_its_own(self, client: TestClient) -> None:
+        client.get("/api/config", params={"token": TOKEN})
+
+        # No token in this one: the cookie the client kept must carry it.
+        assert client.get("/api/config").status_code == 200
+
+    def test_a_forged_cookie_is_refused(self, client: TestClient) -> None:
+        client.cookies.set(SESSION_COOKIE, "forged-session-id")
+
+        assert client.get("/api/config").status_code == 401
+
+    def test_the_launch_token_in_a_cookie_is_refused(self, client: TestClient) -> None:
+        """It used to be exactly this value, so this is the regression that matters."""
+        client.cookies.set(SESSION_COOKIE, TOKEN)
+
+        assert client.get("/api/config").status_code == 401
+
+    def test_logging_out_revokes_the_session(self, client: TestClient) -> None:
+        client.get("/api/config", params={"token": TOKEN})
+        assert client.get("/api/config").status_code == 200
+
+        client.post("/api/session/logout")
+
+        assert client.get("/api/config").status_code == 401
+
+    def test_the_launch_token_still_works_for_a_non_browser_client(
+        self, client: TestClient
+    ) -> None:
+        """Scripts hold the token from the file and cannot run a cookie jar."""
+        client.cookies.clear()
+
+        resp = client.get("/api/config", headers={"x-replicant-token": TOKEN})
+
+        assert resp.status_code == 200