diff --git a/strix/viewer/server.py b/strix/viewer/server.py index 9ef2927b..1a4e647e 100644 --- a/strix/viewer/server.py +++ b/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}) diff --git a/tests/test_viewer.py b/tests/test_viewer.py index 6a8409a2..923edb4e 100644 --- a/tests/test_viewer.py +++ b/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()