Session-gate viewer auth status and mutation endpoints

strix/viewer/server.py

@@ -238,17 +238,7 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
                 self._send_json(HTTPStatus.OK, {"can_steer": state.steer_handler is not None})
                 return
             if path == "/api/auth/status":
-                # Report verification through is_verified() so an expired record
-                # is advertised as unverified -- otherwise the SPA would suppress
-                # re-verification while history stays locked, stranding the user.
-                record = auth.read_auth()
-                self._send_json(
-                    HTTPStatus.OK,
-                    {
-                        "verified": auth.is_verified(),
-                        "email": record.get("email") if record else None,
-                    },
-                )
+                self._handle_auth_status()
                 return
 
             run_values = query.get("run")
@@ -282,7 +272,29 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             else:
                 self._send_json(HTTPStatus.NOT_FOUND, {"error": "unknown endpoint"})
 
+        def _handle_auth_status(self) -> None:
+            # The cached verified email is only disclosed to a caller holding this
+            # process's session capability, so a cookie-less client on an exposed
+            # --host port cannot read it; everyone else looks unverified.
+            # Verification is reported through is_verified() so an expired record
+            # is advertised as unverified -- otherwise the SPA would suppress
+            # re-verification while history stays locked, stranding the user.
+            if not self._has_session():
+                self._send_json(HTTPStatus.OK, {"verified": False, "email": None})
+                return
+            record = auth.read_auth()
+            self._send_json(
+                HTTPStatus.OK,
+                {
+                    "verified": auth.is_verified(),
+                    "email": record.get("email") if record else None,
+                },
+            )
+
         def _handle_otp_start(self) -> None:
+            if not self._has_session():
+                self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
+                return
             email = str(self._read_body().get("email") or "").strip()
             if not email:
                 self._send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid_email"})
@@ -295,6 +307,9 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             self._send_json(HTTPStatus.OK, {"ok": True})
 
         def _handle_otp_verify(self) -> None:
+            if not self._has_session():
+                self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
+                return
             body = self._read_body()
             email = str(body.get("email") or "").strip()
             code = str(body.get("code") or "").strip()
@@ -315,6 +330,12 @@ def _make_handler(state: _ViewerState) -> type[BaseHTTPRequestHandler]:
             self._send_json(HTTPStatus.OK, {"verified": True, "email": verified_email})
 
         def _handle_forget(self) -> None:
+            # Clearing the cached verification is a state change, so it requires
+            # this process's session capability: a cookie-less caller on an
+            # exposed --host port must not be able to log the operator out.
+            if not self._has_session():
+                self._send_json(HTTPStatus.FORBIDDEN, {"error": "forbidden"})
+                return
             auth.forget()
             self._send_json(HTTPStatus.OK, {"ok": True})
 

tests/test_viewer.py

@@ -86,8 +86,10 @@ def test_build_run_state_from_agents_json(tmp_path: Path) -> None:
     assert state["events"] == []
 
 
-def _get(url: str) -> tuple[int, str, bytes]:
-    with urllib.request.urlopen(url) as resp:  # noqa: S310 - localhost test server
+def _get(url: str, *, cookie: str | None = None) -> tuple[int, str, bytes]:
+    headers = {"Cookie": cookie} if cookie else {}
+    req = urllib.request.Request(url, headers=headers)  # noqa: S310 - localhost test server
+    with urllib.request.urlopen(req) as resp:  # noqa: S310 - localhost test server
         return resp.status, resp.headers.get("Content-Type", ""), resp.read()
 
 
@@ -302,15 +304,38 @@ def test_auth_status_reflects_expiry(tmp_path: Path, monkeypatch: pytest.MonkeyP
     verified = {"value": True}
     monkeypatch.setattr("strix.viewer.auth.is_verified", lambda: verified["value"])
 
-    httpd, url, _ = serve(run_dir, open_browser=False)
+    httpd, url, token = serve(run_dir, open_browser=False)
     try:
-        _, _, body = _get(f"{url}/api/auth/status")
+        cookie = _session_cookie(url, token)
+        _, _, body = _get(f"{url}/api/auth/status", cookie=cookie)
         assert json.loads(body) == {"verified": True, "email": "a@b.com"}
 
         # Once expired, status must advertise unverified so the SPA re-prompts.
         verified["value"] = False
-        _, _, body = _get(f"{url}/api/auth/status")
+        _, _, body = _get(f"{url}/api/auth/status", cookie=cookie)
         assert json.loads(body)["verified"] is False
+
+        # A cookie-less caller never sees the cached email or verified state.
+        verified["value"] = True
+        _, _, body = _get(f"{url}/api/auth/status")
+        assert json.loads(body) == {"verified": False, "email": None}
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def test_auth_mutations_require_session(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    run_dir = _make_run(tmp_path, "authmut", status="running", end_time=None)
+    _bundle(tmp_path, monkeypatch)
+    forgotten = {"value": False}
+    monkeypatch.setattr("strix.viewer.auth.forget", lambda: forgotten.update(value=True))
+
+    httpd, url, _ = serve(run_dir, open_browser=False)
+    try:
+        for path in ("/api/auth/forget", "/api/auth/otp/start", "/api/auth/otp/verify"):
+            status, _ = _post(url, path, {"email": "a@b.com", "code": "123456"})
+            assert status == 403, path
+        assert forgotten["value"] is False
     finally:
         httpd.shutdown()
         httpd.server_close()