"""The session gate: what opens a route, what refuses, and what each refusal says. Phase 11 tasks 03 and 04. `auth.py` decides in isolation (`tests/test_auth.py`); this file runs those decisions through a real ASGI request, because every bug in a gate lives in the plumbing — a path that skipped the middleware, a redirect a `curl` cannot use, a write that reached the UPS before the CSRF check, a `no-store` left off a 303 that then parked a browser on the login page after it had signed in. Task 04's section is the same instrument on the one route that *re-credentials* a caller: what a wrong current password must leave untouched, what a rotated signing secret does to the other browsers, and which of the two throttle buckets each form spends. Two habits make these tests worth keeping: * **Anonymous requests use `follow_redirects=False`.** `TestClient` follows redirects by default, so an unsigned `GET /api/config` would fetch the login page and report 200 — which is how a gated suite ends up passing while testing nothing. * **A refusal is compared against the answer the route would have given.** The app under test has no UPS, no hosts and no engines configured, so a write that got past the gate answers `404 {"configured": false}` instead of `403`. Asserting 403 therefore proves the *gate* refused, not the route. """ from __future__ import annotations import logging import math import re import sqlite3 import time from collections.abc import Callable, Iterator from dataclasses import replace from pathlib import Path from typing import Any, cast from urllib.parse import unquote import pytest from fastapi import HTTPException, Request from fastapi.testclient import TestClient from auth_client import ADMIN_PASSWORD, sign_in from auth_contract import FAST_PBKDF2 from dashboards import auth from dashboards.api import MIN_PASSWORD_LENGTH, AppContext from dashboards.app import PUBLIC_PATHS, PUBLIC_PREFIXES, create_app, is_public from dashboards.config import Settings from dashboards.poller import Hub from dashboards.sources import nut from dashboards.store import SampleStore #: A browser-shaped `Accept`, the way the page scripts send it through `Dash.fetchJSON`. JSON_ACCEPT = {"accept": "application/json"} #: A browser typing the URL into the address bar. BROWSER_HEADERS = {"accept": "text/html,application/xhtml+xml", "user-agent": "Mozilla/5.0"} CURL_HEADERS = {"accept": "*/*", "user-agent": "curl/8.5.0"} PROTECTED_API = ( "/api/config", "/api/events", "/api/metrics", "/api/history", "/api/summary", "/api/system/snapshot", "/api/system/history", "/api/disks/snapshot", "/api/disks/history", "/api/llm/snapshot", "/api/llm/history", "/api/ups/snapshot", "/api/ups/history", "/api/ups/commands", "/api/auth/status", ) #: Every HTML page behind the gate. `/settings` is a **page** and not one of the four #: dashboards (it carries no `aria-current` tab), which is exactly why it is in this list: the #: one thing that must never regress is a new route answering anonymously. PROTECTED_PAGES = ("/", "/system", "/disks", "/llm", "/ups", "/settings") #: The one write this app has, with the fields the route needs (the UPS is unconfigured, so a #: request that reaches the route answers 404 — that is the point). WRITE = {"ups": "apc", "command": "beeper.disable", "confirm": "yes"} def context(client: TestClient) -> Any: """The app's `AppContext`. `TestClient.app` is typed as the ASGI callable it wraps, so every read through it needs one cast here instead of one `# type: ignore` per assertion — the per-line version of what the rest of the suite already does (`tests/test_api.py`). """ return cast(Any, client.app).state.ctx def served_routes(client: TestClient) -> list[Any]: """Every route object the app serves, top level (the login audit needs no flattening).""" return cast(Any, client.app).routes def walk_routes(routes: list[Any]) -> list[Any]: """Every routable object the app serves, flattened. Three shapes, because `app.routes` is not one flat list in the FastAPI this project pins: a page route has `.path`, `include_router` leaves a wrapper whose `original_router` holds the real (already-prefixed) routes, and a `Mount` is `/static` and its prefix. An unrecognised shape raises instead of skipping — the point of the list built on this is that nothing hides in it, so a new kind of route object has to be a conscious edit here rather than a quiet hole in the audit. """ out: list[Any] = [] for route in routes: if isinstance(getattr(route, "path", None), str): out.append(route) continue inner = getattr(route, "original_router", None) if inner is not None: out.extend(walk_routes(list(getattr(inner, "routes", ())))) continue raise AssertionError(f"unroutable route object in the table: {route!r}") return out def route_paths(routes: list[Any]) -> list[tuple[str, set[str]]]: """The ``(path, methods)`` pairs of `walk_routes`, for the surface audit.""" return [(str(route.path), {str(m) for m in getattr(route, "methods", None) or ()}) for route in walk_routes(routes)] def endpoint_for(client: TestClient, path: str) -> Callable[..., Any]: """The function behind one path, for the defensive branch HTTP cannot reach. `POST /api/auth/login` checks `ctx.auth` itself, and the gate answers 503 for a public path before the route runs — so the only way to prove the route's own refusal is to call it. """ for route in walk_routes(served_routes(client)): if getattr(route, "path", None) == path: handler = getattr(route, "endpoint", None) if callable(handler): return cast(Callable[..., Any], handler) raise AssertionError(f"nothing serves {path}") def settings_for(db: Path) -> Settings: return Settings(db_path=str(db), admin_password=ADMIN_PASSWORD, pbkdf2_iterations=FAST_PBKDF2) @pytest.fixture def store(tmp_path: Path) -> Iterator[SampleStore]: db = SampleStore(str(tmp_path / "gate.sqlite3")) try: yield db finally: db.close() @pytest.fixture def app(store: SampleStore) -> Any: """An app with **no** sources configured, so a route that runs is distinguishable. No targets, no engines, no NUT: every protected answer that got as far as its handler says `404 {"configured": false}`, and nothing this file does can touch the fleet. The login POST and the UPS POST are the two writes on the app, and only one of them dials anything. """ return create_app(settings_for(store.path), store) @pytest.fixture def anon(app: Any) -> Iterator[TestClient]: """A client that has never signed in — and follows no redirects.""" with TestClient(app, follow_redirects=False) as client: yield client @pytest.fixture def authed(anon: TestClient) -> TestClient: """The same client after the real login round-trip: cookie jar and CSRF header.""" sign_in(anon) return anon # --------------------------------------------------------------------------- # # The public list, and what it costs # --------------------------------------------------------------------------- # def test_the_public_list_is_the_documented_three() -> None: """`PUBLIC_PATHS` is the whole attack surface; this line makes adding to it a decision.""" assert set(PUBLIC_PATHS) == {"/login", "/api/auth/login", "/api/health"} assert tuple(PUBLIC_PREFIXES) == ("/static/",) @pytest.mark.parametrize("path", [ "/login", "/api/auth/login", "/api/health", "/static", "/static/", "/static/app.js", "/static/dash.css", "/static/page-login.js", ]) def test_public_paths_are_public_in_the_function(path: str) -> None: assert is_public(path) is True @pytest.mark.parametrize("path", [ "/api/config", "/api/auth/status", "/api/auth/logout", "/api/ups/command", "/api/health/past", "/login/again", "/", "/system", "/staticx/app.js", ]) def test_everything_else_is_not_public(path: str) -> None: """/static is public as a *tree*, never as a prefix match on a longer path.""" assert is_public(path) is False def test_health_stays_readable_the_way_the_healthcheck_parses_it(anon: TestClient) -> None: """`./scripts/dev-server.sh status` and the Containerfile HEALTHCHECK read this unsigned.""" response = anon.get("/api/health") assert response.status_code == 200 body = response.json() # The keys the two healthcheckers read and the one answer they act on. `sources` is the # per-job list `/api/health` grew in phase 03; the script and the Containerfile grep `ok` # out of this body, so its name and shape are a contract with deployment, not here. assert set(body) == {"ok", "status", "sources", "db", "uptime_seconds"} assert body["ok"] is True and body["status"] == "empty" assert "no-store" in response.headers["cache-control"] def test_static_assets_answer_without_a_session(anon: TestClient) -> None: """The login page is styled by the same sheet as the dashboards, so `/static` must be open.""" for path in ("/static/app.js", "/static/dash.css", "/static/charts.js", "/static/page-login.js"): response = anon.get(path) assert response.status_code == 200, path assert len(response.text) > 100, path def test_the_login_page_serves_anonymously(anon: TestClient) -> None: response = anon.get("/login") assert response.status_code == 200 assert "text/html" in response.headers["content-type"] assert "Homelab · sign in" in response.text assert 'id="login-form"' in response.text assert "/static/page-login.js" in response.text assert "no-store" in response.headers["cache-control"] def test_the_login_page_is_one_password_field_and_nothing_else(anon: TestClient) -> None: """One admin: a username field that is always the same word is theatre, and the server would ignore it. The field is a password field with the browser's *current*-password hint (`autocomplete="current-password"` — the one that fills in what you type today without offering to save a new one), and the refusal line is announced. """ html = anon.get("/login").text assert 'id="admin-password"' in html assert 'type="password"' in html assert 'autocomplete="current-password"' in html assert 'for="admin-password"' in html assert 'id="sign-in"' in html assert 'id="login-error"' in html and 'role="alert"' in html assert 'id="login-note"' in html for username_field in ("name=\"username\"", "name=\"user\"", 'name="login"', 'type="text"'): assert username_field not in html, f"/login grew a second field: {username_field}" def test_the_login_page_says_what_it_protects_without_naming_the_fleet(anon: TestClient) -> None: """The one sentence of copy on the page has to be true and useless to a stranger. "Including the UPS controls" is the fact an operator acts on — this password can cut power — while the hostnames, the NUT variables and the target list stay out of a page served to whoever finds the domain. """ html = anon.get("/login").text assert "UPS" in html and "Sign in" in html def _strip(source: str) -> str: """A script or page with its comments gone, so an assertion tests code and not prose. The house helper (`tests/test_api.py::_code`): a comment that *names* the thing a script must not do is documentation, and this file has several — about `Dash.fetchJSON`, about following a `next`, about markup in a refusal. """ source = re.sub(r"/\*.*?\*/", " ", source, flags=re.S) return re.sub(r"^\s*//.*$", " ", source, flags=re.M) def test_the_login_page_loads_nothing_from_the_internet(anon: TestClient) -> None: """Rule 5 applies to the page in front of the dashboards most of all: no CDN, no webfont. The `/login` half of `tests/test_api.py::test_no_page_loads_anything_from_the_internet`, run unsigned: that one's client is signed in, and a signed-in `GET /login` is a `303`, so following redirects there would audit `/`'s bytes and call it `/login`. Both the fragment list and the absolute-URL pattern are checked, because the scheme-relative spelling (`//cdn.example/…`) is the one a `https?://` regex misses. """ text = anon.get("/login").text assert not re.search(r"(src|href)\s*=\s*['\"]https?://", text, re.I) # the same four fragments `tests/test_api.py` checks on the signed-in pages, so the login # page is held to the same rule rather than to a looser one written for it for fragment in ("//cdn.", "https://fonts.", "//unpkg.com", "//cdnjs"): assert fragment not in text # …and it must not describe what it is protecting. for word in ("driveripper", "deskwork", "gateway", "upsd", "NUT_", "HOST_TARGETS"): assert word not in text # --------------------------------------------------------------------------- # # Anonymous callers: the shape of "no" # --------------------------------------------------------------------------- # @pytest.mark.parametrize("path", PROTECTED_API) def test_an_anonymous_api_call_gets_json_401_and_never_a_redirect( anon: TestClient, path: str) -> None: response = anon.get(path, headers=JSON_ACCEPT) assert response.status_code == 401, path assert "application/json" in response.headers["content-type"] assert response.json() == {"ok": False, "detail": "sign in required", "login": "/login"} assert "no-store" in response.headers["cache-control"] assert "location" not in response.headers, "a data caller was bounced to an HTML page" @pytest.mark.parametrize("path", PROTECTED_PAGES) def test_a_browser_typing_a_page_url_is_redirected_and_keeps_its_target( anon: TestClient, path: str) -> None: response = anon.get(path, headers=BROWSER_HEADERS) assert response.status_code == 303, path assert response.headers["location"] == f"/login?next={path}" assert "no-store" in response.headers["cache-control"] def test_the_next_parameter_survives_a_query_string(anon: TestClient) -> None: """Range chips live in the URL; losing them loses the chart the operator was reading.""" location = anon.get("/llm?range=1h", headers=BROWSER_HEADERS).headers["location"] # The `?` and `=` are encoded, so the value comes back as one field and not two — which is # what the login page's reader sees — while the path itself stays readable in the header. assert location == "/login?next=/llm%3Frange%3D1h" assert unquote(location) == "/login?next=/llm?range=1h" def test_curl_asking_for_a_page_is_told_where_to_sign_in(anon: TestClient) -> None: """`curl` follows redirects into a form, so a page request gets one; a data request does not, and that distinction is what keeps both callers usable without a config knob.""" page = anon.get("/ups", headers=CURL_HEADERS) assert page.status_code == 303 assert page.headers["location"] == "/login?next=/ups" assert anon.get("/api/ups/snapshot", headers=CURL_HEADERS).status_code == 401 @pytest.mark.parametrize("verb", ["post", "put", "patch", "delete"]) def test_an_unauthenticated_write_is_answered_in_json(anon: TestClient, verb: str) -> None: """A 303 on a POST is a login page pretending to accept a write. Everything that is not a page `GET` gets JSON, so the caller's error handling stays one shape.""" response = anon.request(verb.upper(), "/api/ups/command", json=WRITE) assert response.status_code == 401 assert "application/json" in response.headers["content-type"] def test_a_forged_cookie_is_answered_exactly_like_no_session(anon: TestClient) -> None: """The answer must not explain *why*: the reason a token failed is a secret about the key.""" anon.cookies.set(auth.SESSION_COOKIE, "not-a-session") response = anon.get("/api/config", headers=JSON_ACCEPT) assert response.status_code == 401 assert response.json()["detail"] == "sign in required" def test_an_expired_session_is_answered_exactly_like_no_session(anon: TestClient) -> None: state = context(anon).auth assert state is not None anon.cookies.set(auth.SESSION_COOKIE, auth.mint_session(state.secret, hours=-1)) response = anon.get("/api/config", headers=JSON_ACCEPT) assert response.status_code == 401 def test_an_unknown_path_keeps_its_404_with_and_without_a_session(anon: TestClient) -> None: """The gate must not turn a typo into a sign-in prompt, and must not answer 404 before it. Otherwise `/api/typo` redirects a script to a login page and "did I spell that route right" stops being an answerable question — and a path that exists for another verb would be hidden behind a 401 for anyone not already in. """ missing = anon.get("/api/nope", headers=JSON_ACCEPT) assert missing.status_code == 404 assert missing.json()["ok"] is False assert missing.json()["detail"].startswith("no such route") assert "application/json" in missing.headers["content-type"] assert anon.get("/nope", headers=BROWSER_HEADERS).status_code == 404 # …and the same answer once signed in: the 404 is the router's, not a concession the gate # grants or withholds depending on who is asking. sign_in(anon) assert anon.get("/api/nope", headers=JSON_ACCEPT).status_code == 404 assert anon.get("/nope", headers=BROWSER_HEADERS).status_code == 404 def test_the_gate_answers_503_when_startup_never_completed(tmp_path: Path) -> None: """No `ctx` means no `AuthState`: there is nothing to check a token against. Not a 401 — an operator who gets 401 here types a password forever. A half-started process is a different failure from an unauthenticated caller and says so. **Including the public paths**, which is one deliberate step past the task text: a process that never built its context cannot answer `/api/health` with `ok` either (that is the "frozen source that looks healthy" failure `AGENTS.md` invariant 4 calls the worst one), and a `/login` form that cannot verify a password is worse than no form. """ client = TestClient(create_app(settings_for(tmp_path / "cold.sqlite3")), follow_redirects=False) # lifespan deliberately not entered response = client.get("/api/config", headers=JSON_ACCEPT) assert response.status_code == 503 assert response.json() == {"ok": False, "detail": "startup did not complete"} assert "no-store" in response.headers["cache-control"] for path, headers in (("/api/health", JSON_ACCEPT), ("/login", BROWSER_HEADERS), ("/api/auth/login", JSON_ACCEPT)): assert client.request("GET", path, headers=headers).status_code == 503, path # --------------------------------------------------------------------------- # # The login endpoint # --------------------------------------------------------------------------- # def test_a_correct_password_mints_the_session_the_gate_accepts(anon: TestClient) -> None: body = sign_in(anon) assert body["ok"] is True assert body["username"] == auth.ADMIN_USER assert body["expires_at"] > time.time() + 3600 assert body["next"] == "/" assert anon.get("/api/config", headers=JSON_ACCEPT).status_code == 200 def test_the_session_cookie_carries_the_flags_a_session_cookie_needs(anon: TestClient) -> None: response = anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) assert response.status_code == 200 cookie = response.headers["set-cookie"] assert f"{auth.SESSION_COOKIE}=" in cookie for flag in ("httponly", "samesite=lax", "path=/"): assert flag in cookie.lower(), cookie # …and no `Secure` on a plain-HTTP request, because dev is http://127.0.0.1:8090. assert "secure" not in cookie.lower() def test_a_proxied_https_login_marks_the_cookie_secure(anon: TestClient) -> None: """nginx terminates TLS, so this is the only way the deployed cookie ever gets `Secure`.""" response = anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}, headers={"x-forwarded-proto": "https"}) assert response.status_code == 200 assert "secure" in response.headers["set-cookie"].lower() def test_the_password_never_appears_in_anything_the_client_gets(anon: TestClient) -> None: response = anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) assert ADMIN_PASSWORD not in response.text assert ADMIN_PASSWORD not in "".join(response.headers.values()) assert ADMIN_PASSWORD not in response.cookies[auth.SESSION_COOKIE] assert ADMIN_PASSWORD not in anon.get("/api/auth/status").text verifier = context(anon).auth.verifier assert verifier.startswith("pbkdf2_sha256$") and ADMIN_PASSWORD not in verifier def test_a_json_body_that_is_not_an_object_is_a_422(anon: TestClient) -> None: """`["testpass"]` is JSON, and it is not a login. The parser reads a field, so anything without fields is the same refusal as an absent password — never an exception on the way to the hash comparison. """ for body in (["testpass"], "testpass", 42, None): response = anon.post("/api/auth/login", json=body) assert response.status_code == 422, body assert "password" in response.json()["detail"] @pytest.mark.parametrize("body", [ {"password": ""}, {"password": None}, {}, {"password": 42}, {"other": "x"}, ]) def test_a_login_without_a_password_string_is_refused_before_any_hashing( anon: TestClient, body: dict[str, Any]) -> None: """`{"password": ""}` is a bad request, not a wrong password. The distinction is not tidiness: an empty string is the input that could turn a mis-seeded store into an oracle, so it never reaches `verify_password`, and it must not be counted as a guess against the throttle either. """ response = anon.post("/api/auth/login", json=body) assert response.status_code == 422 assert "password" in response.json()["detail"] assert context(anon).login_throttle.attempts("testclient") == 0 def test_a_non_json_login_body_is_a_422(anon: TestClient) -> None: response = anon.post("/api/auth/login", content="hunter2", headers={"content-type": "text/plain"}) assert response.status_code == 422 assert response.json()["ok"] is False def test_a_wrong_password_is_one_sentence_that_names_no_account(anon: TestClient) -> None: response = anon.post("/api/auth/login", json={"password": "not-it"}) assert response.status_code == 401 assert response.json() == {"ok": False, "detail": "wrong password"} assert "admin" not in response.text def test_the_login_answers_are_never_cacheable(anon: TestClient) -> None: good = anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) bad = anon.post("/api/auth/login", json={"password": "not-it"}) assert "no-store" in good.headers["cache-control"] assert "no-store" in bad.headers["cache-control"] def test_login_is_post_only(anon: TestClient) -> None: response = anon.get("/api/auth/login") assert response.status_code == 405 assert "application/json" in response.headers["content-type"] def test_logging_in_clears_the_failure_count(anon: TestClient) -> None: throttle = context(anon).login_throttle for _ in range(throttle.max_attempts - 1): assert anon.post("/api/auth/login", json={"password": "not-it"}).status_code == 401 assert throttle.attempts("testclient") == throttle.max_attempts - 1 sign_in(anon) assert throttle.attempts("testclient") == 0 def test_a_blocked_client_is_refused_before_the_password_is_hashed(anon: TestClient) -> None: """The proof the hash never ran: the **correct** password is refused while blocked.""" throttle = context(anon).login_throttle for _ in range(throttle.max_attempts): assert anon.post("/api/auth/login", json={"password": "not-it"}).status_code == 401 blocked = anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) assert blocked.status_code == 429 seconds = int(blocked.headers["retry-after"]) assert seconds == math.ceil(throttle.retry_after("testclient")) >= 1 detail = blocked.json()["detail"] # the only fact the response may add is how long to wait — never an account, never a reason assert ADMIN_PASSWORD not in detail and auth.ADMIN_USER not in detail def test_the_first_block_is_a_minute_not_a_second(anon: TestClient) -> None: """What a hammering client actually pays: `base_seconds`, not a poll-friendly 1 s.""" throttle = context(anon).login_throttle assert throttle.base_seconds >= 30 assert throttle.max_seconds <= 3600 def test_the_throttle_is_keyed_per_client(app: Any) -> None: """One noisy machine must not lock out the operator standing next to it.""" with TestClient(app, follow_redirects=False, client=("203.0.113.7", 1)) as noisy: throttle = context(noisy).login_throttle for _ in range(throttle.max_attempts): noisy.post("/api/auth/login", json={"password": "not-it"}) assert noisy.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 429 # A second peer address on the same process, same instant, same wrong-or-right # password: the buckets are per key, so this one still gets as far as the hash. quiet = TestClient(app, follow_redirects=False, client=("203.0.113.9", 1)) assert quiet.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 200 def test_no_login_response_or_event_names_the_client_address(anon: TestClient, store: SampleStore) -> None: """`auth.py` logs no IP, and the event row must not either — `/api/events` is readable.""" for _ in range(3): response = anon.post("/api/auth/login", json={"password": "not-it"}) assert "testclient" not in response.text denials = [event for event in store.recent_events(20) if event.kind == "login-denied"] assert denials, "a refused sign-in must leave a trace somewhere" for event in denials: assert event.source == "auth" assert "testclient" not in event.message and "203.0" not in event.message assert ADMIN_PASSWORD not in event.message assert auth.ADMIN_USER not in event.message def test_a_refused_sign_in_is_visible_in_the_feed_and_quiet_on_the_overview( anon: TestClient) -> None: """`source: "auth"` is the deliberate half of the design: searchable, never blinking.""" anon.post("/api/auth/login", json={"password": "not-it"}) sign_in(anon) filtered = anon.get("/api/events", params={"source": "auth"}).json()["events"] assert filtered and filtered[0]["kind"] == "login-denied" everything = anon.get("/api/events").json()["events"] assert any(e["kind"] == "login-denied" for e in everything) alerts = anon.get("/api/summary").json()["alerts"] assert not [a for a in alerts if a["source"] == "auth"], \ "a denied login became an overview alert — it stays a searchable event" # --------------------------------------------------------------------------- # # Session, bearer token, status, logout # --------------------------------------------------------------------------- # def test_status_names_the_session_it_belongs_to(authed: TestClient) -> None: response = authed.get("/api/auth/status") body = response.json() assert body["ok"] is True and body["authenticated"] is True assert body["username"] == auth.ADMIN_USER assert body["seconds_left"] > 3600 assert body["session_hours"] == context(authed).settings.session_hours state = context(authed).auth assert state is not None token = authed.cookies[auth.SESSION_COOKIE] assert auth.verify_csrf(state.secret, token, body["csrf"]) is True # the status body is the only place the CSRF token is published, and it is not a session assert body["csrf"] != token assert ADMIN_PASSWORD not in response.text def test_the_bearer_header_opens_the_api_for_a_script(anon: TestClient) -> None: """`curl -H "Authorization: Bearer …"` is the documented scripted caller.""" anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) token = anon.cookies[auth.SESSION_COOKIE] del anon.cookies[auth.SESSION_COOKIE] assert anon.get("/api/config", headers=JSON_ACCEPT).status_code == 401 assert anon.get("/api/config", headers={"authorization": f"Bearer {token}"}).status_code == 200 def test_a_bearer_caller_needs_no_csrf_token_and_a_cookie_caller_does( anon: TestClient) -> None: """The asymmetry is the CSRF rule stated twice, and both halves are pinned here. A cross-site page can make a browser send its cookie; it cannot read a token out of the operator's shell. So the bearer write goes through and the cookie write without the header does not — and the refusal comes from the gate, because this app's UPS route would have answered `404 {"configured": false}`. """ anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) token = anon.cookies[auth.SESSION_COOKIE] cookieless = anon.post("/api/ups/command", json=WRITE) assert cookieless.status_code == 403 assert "csrf" in cookieless.json()["detail"].lower() del anon.cookies[auth.SESSION_COOKIE] bearer = anon.post("/api/ups/command", json=WRITE, headers={"authorization": f"Bearer {token}"}) assert bearer.status_code == 404, "the gate stopped a bearer write it should have passed" assert bearer.json().get("configured") is False @pytest.mark.parametrize("header", ["Token abc", "Basic dGVzdDp0ZXN0", "Bearer", "bearer", "Bearerx abc"]) def test_an_authorization_header_that_is_not_a_bearer_token_is_no_session( anon: TestClient, header: str) -> None: response = anon.get("/api/config", headers={"authorization": header}) assert response.status_code == 401 def test_a_session_signed_by_another_install_is_no_session(tmp_path: Path) -> None: """The signature is checked against *this* install's secret. Same password, different `auth.session_secret` — a second deployment, or this one after a task-04 password change — and its cookie opens nothing here. """ with TestClient(create_app(settings_for(tmp_path / "other.sqlite3")), follow_redirects=False) as other: other.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) stolen = other.cookies[auth.SESSION_COOKIE] assert stolen with TestClient(create_app(settings_for(tmp_path / "here.sqlite3")), follow_redirects=False) as here: here.cookies.set(auth.SESSION_COOKIE, stolen) assert here.get("/api/config", headers=JSON_ACCEPT).status_code == 401 def test_rotating_the_signing_secret_evicts_every_live_session(authed: TestClient) -> None: """That is what "changing the password logs every other browser out" is made of (task 04).""" assert authed.get("/api/config", headers=JSON_ACCEPT).status_code == 200 state = context(authed).auth assert state is not None context(authed).auth = replace(state, secret=auth.new_secret()) assert authed.get("/api/config", headers=JSON_ACCEPT).status_code == 401 def test_logout_ends_the_cookie_and_needs_the_csrf_header(authed: TestClient) -> None: """A plain `` is a CSRF: a planted link could sign the operator out of a dashboard they were mid-outage in, so `/api/auth/logout` is gated like any other write.""" csrf = authed.headers.get(auth.CSRF_HEADER) assert csrf del authed.headers[auth.CSRF_HEADER] refused = authed.post("/api/auth/logout") assert refused.status_code == 403 authed.headers[auth.CSRF_HEADER] = str(csrf) assert authed.post("/api/auth/logout").status_code == 200 assert authed.get("/api/config", headers=JSON_ACCEPT).status_code == 401 def test_the_csrf_token_is_bound_to_the_session_that_received_it(tmp_path: Path) -> None: """One token per session: a token minted for another install's session is refused. The gate checks the header against *the session the request authenticated as*, which is the only binding that makes a leaked token worthless on its own. """ first = TestClient(create_app(settings_for(tmp_path / "first.sqlite3")), follow_redirects=False) with first: sign_in(first) token_for_first = str(first.headers[auth.CSRF_HEADER]) second = TestClient(create_app(settings_for(tmp_path / "second.sqlite3")), follow_redirects=False) with second: second.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) # Assigned, not deleted-then-set: this client never sent a token of its own, and an # `httpx` header that is absent cannot be removed. second.headers[auth.CSRF_HEADER] = token_for_first assert second.post("/api/auth/logout").status_code == 403 # --------------------------------------------------------------------------- # # Changing the password: one form, every browser # --------------------------------------------------------------------------- # #: Longer than `api.MIN_PASSWORD_LENGTH` and nothing like the seed, so "the old password no #: longer works" cannot pass by accident on a similar string. NEW_PASSWORD = "a-quite-different-new-password" def meta_rows(store: SampleStore) -> dict[str, str | None]: """Both credential rows as they sit in SQLite — the state a refusal must not disturb.""" return {auth.META_VERIFIER: store.get_meta(auth.META_VERIFIER), auth.META_SESSION_SECRET: store.get_meta(auth.META_SESSION_SECRET)} def on_disk(store: SampleStore) -> bytes: """Every byte the database owns, WAL included: what a `grep` on the file would find.""" raw = Path(store.path).read_bytes() for suffix in ("-wal", "-shm"): side = Path(f"{store.path}{suffix}") if side.exists(): raw += side.read_bytes() return raw def change(client: TestClient, *, current: str = ADMIN_PASSWORD, new: str = NEW_PASSWORD, **headers: str) -> Any: """The change-password POST, with the client's own CSRF token already on it. `headers` is for the per-request cases (a proxy header, a replayed token); the token lives on the client because `sign_in` put it there, exactly like the page's `Dash.fetchJSON`. """ return client.post("/api/auth/password", json={"current": current, "new": new}, headers=headers or None) def change_and_adopt(client: TestClient, **payload: str) -> dict[str, Any]: """Change the password and put the returned CSRF token on the client. A successful change rotates the signing secret, so the token `sign_in` installed is dead the moment the response arrives. Leaving it there would 403 every later write in a test for a reason that has nothing to do with what that test is about. """ response = change(client, **payload) assert response.status_code == 200, response.text body: dict[str, Any] = response.json() client.headers[auth.CSRF_HEADER] = str(body["csrf"]) return body def test_a_wrong_current_password_changes_nothing(authed: TestClient, store: SampleStore) -> None: """`403`, and the stored rows **byte-identical** to before. The failure this rules out is a route that writes the new verifier and only then discovers the current one was wrong: the admin password would change for someone who does not know it, and the 403 would still claim otherwise. """ before = meta_rows(store) response = change(authed, current="not-the-password") assert response.status_code == 403 assert response.json() == {"ok": False, "detail": "current password is wrong"} assert "no-store" in response.headers["cache-control"] assert meta_rows(store) == before, "a refused change wrote to the stored credential" # A refusal is a refusal, not a lockout: the session the caller already held still works. assert authed.get("/api/summary", headers=JSON_ACCEPT).status_code == 200 assert ADMIN_PASSWORD not in response.text and NEW_PASSWORD not in response.text def test_a_refused_change_is_throttled_on_its_own_key_and_the_sixth_attempt_is_late( authed: TestClient, store: SampleStore) -> None: """Five wrong *current* passwords, and the sixth — the **correct** one — answers `429`. Same arithmetic as the login form, on a separate key (`password:
`). The key has to exist because a stolen cookie plus this form is an oracle for the old password — the right answer lands visibly, the wrong one is a 403 — and it has to be **separate** because hammering one form must not lock the operator out of the other, and clearing one must not unlock the other. """ throttle = context(authed).login_throttle key = "password:testclient" before = meta_rows(store) for _ in range(throttle.max_attempts): assert change(authed, current="not-the-password").status_code == 403 assert throttle.attempts(key) == throttle.max_attempts assert throttle.attempts("testclient") == 0, "the change form spent the login bucket" blocked = change(authed) assert blocked.status_code == 429 assert int(blocked.headers["retry-after"]) == math.ceil(throttle.retry_after(key)) >= 1 assert "password" in blocked.json()["detail"].lower() assert meta_rows(store) == before # The login form is untouched by all of it: one throttle object, the other bucket. assert authed.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 200 assert throttle.attempts("testclient") == 0 def test_a_refused_change_is_an_event_and_never_an_alert(authed: TestClient) -> None: """`password-denied` and `password-changed` join `login-denied` in the feed, quietly. The overview is for outages: `alerts.build` promotes only `kind: "unreachable"`, so the two new kinds reach `/api/events?source=auth` and nowhere else on `/`. A credential event that blinked a red panel would train the operator to ignore the panels that matter. """ change(authed, current="not-the-password") change_and_adopt(authed) feed = authed.get("/api/events", params={"source": "auth"}).json()["events"] kinds = [event["kind"] for event in feed] assert "password-denied" in kinds and "password-changed" in kinds for event in feed: assert event["severity"] == "warn" assert ADMIN_PASSWORD not in event["message"] and NEW_PASSWORD not in event["message"] assert "testclient" not in event["message"], "an event row quoted the peer address" alerts = authed.get("/api/summary").json()["alerts"] assert not [alert for alert in alerts if alert["source"] == "auth"], \ "a password change became an overview alert" @pytest.mark.parametrize("new,fragment", [ ("", "empty"), (" ", "empty"), ("\t\n ", "empty"), ("a" * (MIN_PASSWORD_LENGTH - 1), "at least"), (ADMIN_PASSWORD, "differ"), ], ids=["empty", "spaces", "whitespace", "one-char-short", "same-as-current"]) def test_a_new_password_that_is_refused_names_the_reason(authed: TestClient, store: SampleStore, new: str, fragment: str) -> None: """`422` that says *which* rule, the rows untouched, and nothing recorded. Length is the whole composition policy — no symbol or digit mandate, which produces sticky notes rather than passwords. A rejected composition is not a guess at anything, so it must not spend a throttle slot or write a `password-denied` row; the empty/short/same cases would otherwise be a way to burn an operator's backoff with a keyboard stuck on Enter. """ before = meta_rows(store) response = change(authed, new=new) assert response.status_code == 422 assert response.json()["ok"] is False assert fragment in response.json()["detail"] assert meta_rows(store) == before assert context(authed).login_throttle.attempts("password:testclient") == 0 assert store.recent_events(20, source="auth") == [] assert ADMIN_PASSWORD not in response.text @pytest.mark.parametrize("raw,detail", [ ("not json at all", "send a JSON body with `current` and `new`"), ('"a string"', "send a JSON body with `current` and `new`"), ("[1,2]", "send a JSON body with `current` and `new`"), ("{}", "`current` is required"), ('{"current": ""}', "`current` is required"), ('{"current": 42}', "`current` is required"), ('{"current": "whatever"}', "`new` is required"), ('{"current": "whatever", "new": 42}', "`new` is required"), ]) def test_a_malformed_change_request_is_refused_before_the_password_is_read( authed: TestClient, store: SampleStore, raw: str, detail: str) -> None: """The same bargain as `POST /api/auth/login`: a body that never asked to authenticate is a `422`, not a wrong password. It must not reach `verify_password` (an absent `current` is not an incorrect one) and must not charge the throttle, or anyone with a malformed template could spend the backoff. """ before = meta_rows(store) response = authed.post("/api/auth/password", content=raw, headers={"content-type": "application/json"}) assert response.status_code == 422 assert response.json()["detail"] == detail assert meta_rows(store) == before assert context(authed).login_throttle.attempts("password:testclient") == 0 def test_a_password_change_replaces_the_verifier_and_the_secret(authed: TestClient, store: SampleStore) -> None: """The order the task calls out, pinned: the response is built from the **new** state. Both rows change, the old password stops verifying against the stored string, and the caller walks away with a session that works — because the cookie in the response was minted from the secret the change just installed. Minting it from `ctx.auth` before the write is the bug this test exists for: the operator would be signed out by their own form. """ before = meta_rows(store) old_cookie = authed.cookies[auth.SESSION_COOKIE] response = change(authed) assert response.status_code == 200, response.text body = response.json() assert body["ok"] is True and body["changed"] is True assert body["hint"] == "other browsers are now signed out" assert body["expires_at"] > time.time() assert body["csrf"] assert "no-store" in response.headers["cache-control"] after = meta_rows(store) assert after[auth.META_VERIFIER] != before[auth.META_VERIFIER] assert after[auth.META_SESSION_SECRET] != before[auth.META_SESSION_SECRET] stored = str(after[auth.META_VERIFIER]) assert stored.startswith(f"{auth.PBKDF2_ALGO}$"), "the row is not an encoded verifier" assert auth.verify_password(NEW_PASSWORD, stored) is True assert auth.verify_password(ADMIN_PASSWORD, stored) is False assert NEW_PASSWORD not in stored and ADMIN_PASSWORD not in stored assert NEW_PASSWORD not in response.text and ADMIN_PASSWORD not in response.text # A good change clears its own backoff, so a confused operator who mistyped twice is back to # a clean slate rather than one mistake away from a lockout on the form they just succeeded at. assert context(authed).login_throttle.attempts("password:testclient") == 0 # The caller survives because the response carried a cookie from the new secret … authed.headers[auth.CSRF_HEADER] = str(body["csrf"]) assert authed.get("/api/summary", headers=JSON_ACCEPT).status_code == 200 assert authed.get("/api/auth/status").json()["csrf"] == body["csrf"] # …and the cookie that was in the jar before the change is worthless now, on the page # and as a bearer token: it is the same signed value either way. stale = TestClient(cast(Any, authed.app), follow_redirects=False) stale.cookies.set(auth.SESSION_COOKIE, old_cookie) assert stale.get("/api/summary", headers=JSON_ACCEPT).status_code == 401 assert stale.get("/api/config", headers=JSON_ACCEPT).status_code == 401 def test_a_password_change_signs_out_every_other_browser(authed: TestClient) -> None: """The second browser captured *before* the change is out; the one that made it stays in. This is the whole reason a password change rotates the signing secret. Sessions are stateless — there is no row to delete and no revocation list to consult — so without the rotation "change the password" would leave a cookie an attacker already holds valid for another `DASH_SESSION_HOURS` (30 days by default) while the UI claimed the opposite. """ other = TestClient(cast(Any, authed.app), follow_redirects=False, client=("203.0.113.44", 1)) sign_in(other) assert other.get("/api/summary", headers=JSON_ACCEPT).status_code == 200 change_and_adopt(authed) assert authed.get("/api/summary", headers=JSON_ACCEPT).status_code == 200 refused = other.get("/api/summary", headers=JSON_ACCEPT) assert refused.status_code == 401 assert refused.json() == {"ok": False, "detail": "sign in required", "login": "/login"} # …and that browser gets back in with the new password and only the new one. assert other.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 401 assert other.post("/api/auth/login", json={"password": NEW_PASSWORD}).status_code == 200 def test_the_session_a_password_change_returns_still_operates_the_ups( nut_app: TestClient) -> None: """The rotated secret re-credentials the caller and does not stop them, measured on the wire. `nut_app` is the only fixture here with a driver to reach, so it is where the end-to-end promise gets checked: the new cookie with the new CSRF token authorises the same instant command, while the token in use before the change is refused with nothing sent. A rotation that broke the caller would be the bug; a refusal that reached `instcmd` first would be worse. """ old_csrf = str(nut_app.headers[auth.CSRF_HEADER]) body = change_and_adopt(nut_app) assert body["csrf"] != old_csrf replayed = nut_app.post("/api/ups/command", json={"ups": "apc", "command": "test.panel.start"}, headers={auth.CSRF_HEADER: old_csrf}) assert replayed.status_code == 403 assert "csrf" in replayed.json()["detail"].lower() assert WireRecorder.sent == [], "a token the new session does not know reached the driver" accepted = nut_app.post("/api/ups/command", json={"ups": "apc", "command": "test.panel.start"}) assert accepted.status_code == 200, accepted.text assert WireRecorder.sent == [("apc", "test.panel.start")] def test_the_changed_password_beats_the_env_seed_on_the_next_start(tmp_path: Path) -> None: """Restart semantics: the row is the credential, `DASH_ADMIN_PASSWORD` is a seed. The env var still says `testpass` on the second start — which is exactly what a container with an unedited env file looks like — and the changed password wins both ways. The other order (the seed overwriting the row) is the bug this pins: a redeploy would roll every operator's change back without saying so. """ db = tmp_path / "restart.sqlite3" store = SampleStore(str(db)) try: with TestClient(create_app(settings_for(db), store), follow_redirects=False) as first: sign_in(first) assert change_and_adopt(first)["changed"] is True with TestClient(create_app(settings_for(db), store), follow_redirects=False) as second: state = context(second).auth assert state is not None and state.env_ignored is True assert second.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 401 assert sign_in(second, NEW_PASSWORD)["ok"] is True assert second.get("/api/summary", headers=JSON_ACCEPT).status_code == 200 finally: store.close() def test_a_login_upgrades_a_verifier_minted_at_an_older_cost(tmp_path: Path) -> None: """`DASH_PBKDF2_ITERATIONS` has to actually buy something. A row minted at 1 000 rounds verifies forever, so without this the operator raises the setting and the stored cost never moves. One login at the new cost replaces the row and keeps the signing secret: the same password works, no browser is signed out, and a future phase that raises the default gets the upgrade on the next sign-in rather than a migration script. """ db = tmp_path / "rehash.sqlite3" store = SampleStore(str(db)) try: cheap = auth.hash_password(ADMIN_PASSWORD, iterations=1_000) store.set_meta(auth.META_VERIFIER, cheap) app = create_app(Settings(db_path=str(db), admin_password=ADMIN_PASSWORD, pbkdf2_iterations=2_000), store) with TestClient(app, follow_redirects=False) as client: secret_before = store.get_meta(auth.META_SESSION_SECRET) assert client.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 200 stored = store.get_meta(auth.META_VERIFIER) assert stored is not None and stored.startswith(f"{auth.PBKDF2_ALGO}$2000$") assert stored != cheap assert auth.verify_password(ADMIN_PASSWORD, stored) is True assert store.get_meta(auth.META_SESSION_SECRET) == secret_before, \ "the rehash rotated the secret — that is a logout, not an upgrade" assert context(client).auth.verifier == stored assert client.get("/api/config", headers=JSON_ACCEPT).status_code == 200 # and a second login against the new row still gets in assert client.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 200 finally: store.close() def test_a_refused_rehash_write_does_not_cost_the_operator_their_login( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None: """A `database is locked` on the upgrade row must not turn a correct password into a 500. The dev store is known to refuse writes now and then (`AGENTS.md` → Known gaps), and a login route that let that escape would lock the admin out until the DB recovered — and then keep doing it, because the row is still behind the configured cost. The warning is asserted to carry no credential: it is fixed text, the same rule `auth.py` holds itself to. """ db = tmp_path / "rehash-refused.sqlite3" store = SampleStore(str(db)) try: cheap = auth.hash_password(ADMIN_PASSWORD, iterations=1_000) store.set_meta(auth.META_VERIFIER, cheap) app = create_app(Settings(db_path=str(db), admin_password=ADMIN_PASSWORD, pbkdf2_iterations=2_000), store) with TestClient(app, follow_redirects=False) as client: def refuse(*_args: object) -> None: raise sqlite3.OperationalError("database is locked") monkeypatch.setattr(store, "set_meta", refuse) with caplog.at_level(logging.WARNING, logger="dashboards.api"): assert client.post("/api/auth/login", json={"password": ADMIN_PASSWORD}).status_code == 200 assert client.get("/api/config", headers=JSON_ACCEPT).status_code == 200 assert store.get_meta(auth.META_VERIFIER) == cheap recorded = "\n".join(record.getMessage() for record in caplog.records) assert "try again" in recorded assert ADMIN_PASSWORD not in recorded and cheap not in recorded finally: store.close() def test_the_re_credentialed_cookie_carries_the_login_attributes(anon: TestClient) -> None: """One cookie, one set of attributes — whichever route handed it out. A `Secure` flag on the login path and not on the change path is a session that half-works behind TLS, and `HttpOnly` is the one attribute nobody notices missing until there is an XSS on the page. """ login = anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) assert login.status_code == 200 changed = anon.post("/api/auth/password", json={"current": ADMIN_PASSWORD, "new": NEW_PASSWORD}, headers={auth.CSRF_HEADER: str(login.json()["csrf"])}) assert changed.status_code == 200 def attributes(header: str) -> set[str]: return {part.strip().lower() for part in header.split(";")[1:]} assert attributes(changed.headers["set-cookie"]) == attributes(login.headers["set-cookie"]) assert {"httponly", "samesite=lax", "path=/"} <= attributes(changed.headers["set-cookie"]) assert any(part.startswith("max-age=") for part in attributes(changed.headers["set-cookie"])) # The `X-Forwarded-Proto` rule is shared too, because it is the same helper. proxied = anon.post("/api/auth/password", json={"current": NEW_PASSWORD, "new": "yet-another-long-one"}, headers={auth.CSRF_HEADER: str(changed.json()["csrf"]), "x-forwarded-proto": "https"}) assert "secure" in attributes(proxied.headers["set-cookie"]) def test_the_password_route_is_post_only_and_needs_the_csrf_token(authed: TestClient, store: SampleStore) -> None: """It is a write like any other: the gate's CSRF rule applies, and `GET` is a 405. A change-password form reachable by a `` would be the quietest account takeover there is — the attacker sets a password they know and the operator notices at the next login. """ assert authed.get("/api/auth/password").status_code == 405 before = meta_rows(store) csrf = str(authed.headers[auth.CSRF_HEADER]) del authed.headers[auth.CSRF_HEADER] refused = change(authed) assert refused.status_code == 403 assert "csrf" in refused.json()["detail"].lower() assert meta_rows(store) == before authed.headers[auth.CSRF_HEADER] = csrf assert change(authed).status_code == 200 def test_no_probe_sees_the_password_change(anon: TestClient) -> None: """`/api/health` is public and `/api/config` is the browser's map of the app. Neither is a place where "the admin password was changed at T", a verifier, or a session detail belongs: the first is answered to an anonymous probe by design (the Containerfile `HEALTHCHECK`), and the second is the payload a page renders. The fact lives in the event feed, behind the session, and on `/settings`. """ sign_in(anon) change_and_adopt(anon) for path in "/api/health", "/api/config": text = anon.get(path, headers=JSON_ACCEPT).text.lower() for word in ("password", "verifier", "pbkdf2", "csrf", "session", "dash_session"): assert word not in text, f"{path} grew auth detail: {word!r}" def test_the_database_never_holds_either_password_as_text(authed: TestClient, store: SampleStore) -> None: """The completion check — `SELECT value FROM meta` prints an encoded verifier and nothing else. Run on the raw bytes including the WAL, because a plaintext that only lives in a sidecar file is still a plaintext someone will find with `strings`. """ change_and_adopt(authed) raw = on_disk(store) assert ADMIN_PASSWORD.encode() not in raw assert NEW_PASSWORD.encode() not in raw assert auth.META_VERIFIER.encode() in raw, "the row this test is about is not in the file" # --------------------------------------------------------------------------- # # Origin and `Sec-Fetch-Site`, on every protected write # --------------------------------------------------------------------------- # class WireRecorder: """`nut.NutClient` with the socket left out, so "did the command reach the driver?" is a question this file can answer instead of infer from a status code. `tests/test_api.py` has the equivalent (`ScriptedNutClient`) for UPS behaviour; the gate's promise is proven here rather than by importing another test module's fixture graph, because the failure this rules out is the one where a 403 is answered *after* `instcmd` — a power cut with an error message attached. """ #: Both are set by the `nut_app` fixture before the app starts, which is also what resets #: them between tests — the same shape `tests/test_api.py::ScriptedNutClient` uses. sent: list[tuple[str, str]] logged_in: list[str] def __init__(self, host: str, port: int, *, timeout: float = 4.0, username: str | None = None, password: str | None = None) -> None: self.credentials = (username, password) async def __aenter__(self) -> WireRecorder: return self async def __aexit__(self, *_exc: object) -> None: return None async def list_ups(self) -> list[dict[str, str]]: return [{"name": "apc", "description": "APC Back-UPS RS 1500MS"}] async def list_vars(self, ups: str) -> dict[str, str]: return {"ups.mfr": "APC", "ups.model": "Back-UPS RS 1500MS", "ups.status": "OL", "battery.charge": "100", "ups.load": "18", "battery.voltage": "13.6", "input.voltage": "230", "battery.mfr.date": "2022-04", "ups.test.result": "No test initiated"} async def list_cmds(self, ups: str) -> list[str]: return ["test.panel.start", "beeper.mute", "driver.killpower"] async def list_rw_vars(self, ups: str) -> list[str]: return [] async def login(self) -> None: assert self.credentials[0], "a command send must authenticate as a named user" WireRecorder.logged_in.append(str(self.credentials[0])) async def instcmd(self, ups: str, command: str) -> str: WireRecorder.sent.append((ups, command)) return "OK" @pytest.fixture def nut_app(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]: """A signed-in client on an app whose upsd is `WireRecorder`. Configured, actions allowed, **not** simulated: the simulator has no wire to reach, so only this branch can show that a refusal happened before it. The UPS card is waited for rather than injected because `/api/ups/command` reads the name and the advertised commands out of the last poll — the same route the page uses, unedited. """ WireRecorder.sent = [] WireRecorder.logged_in = [] monkeypatch.setattr(nut, "NutClient", WireRecorder) db = tmp_path / "ups-gate.sqlite3" store = SampleStore(str(db)) try: app = create_app(Settings(db_path=str(db), admin_password=ADMIN_PASSWORD, pbkdf2_iterations=FAST_PBKDF2, nut_host="127.0.0.1", nut_user="dashboards", nut_password="not-the-real-one", nut_allow_actions=True, nut_interval=0.3), store) with TestClient(app, follow_redirects=False) as client: sign_in(client) deadline = time.time() + 10 while time.time() < deadline: if client.get("/api/ups/snapshot", headers=JSON_ACCEPT).json().get("ups"): break time.sleep(0.05) else: # pragma: no cover - the recorder always answers pytest.fail("the scripted upsd never produced a payload") yield client finally: store.close() def test_a_write_without_the_csrf_token_never_reaches_the_wire(nut_app: TestClient) -> None: """The CSRF gate measured on the wire, not on a status code. Same session, same body, same route: the only difference is the header, and the difference in what the UPS hears is everything. Proven both ways, because a gate that refused every write would pass a test that only asserted the refusal. """ csrf = str(nut_app.headers.get(auth.CSRF_HEADER)) del nut_app.headers[auth.CSRF_HEADER] refused = nut_app.post("/api/ups/command", json={"ups": "apc", "command": "test.panel.start"}) assert refused.status_code == 403 assert "csrf" in refused.json()["detail"].lower() assert WireRecorder.sent == [], "the driver got a command the gate had refused" assert WireRecorder.logged_in == [] nut_app.headers[auth.CSRF_HEADER] = csrf accepted = nut_app.post("/api/ups/command", json={"ups": "apc", "command": "test.panel.start"}) assert accepted.status_code == 200, accepted.text assert WireRecorder.sent == [("apc", "test.panel.start")] def test_the_danger_word_is_still_what_cuts_power_after_the_gate(nut_app: TestClient) -> None: """The gate is additive: the session and CSRF checks run first and change nothing behind them. `driver.killpower` still needs the typed `DANGER` (`AGENTS.md` invariant 1), still refuses without it, and still reaches the driver with it — a phase that added a fifth gate and lost the fourth would be a regression this app would not notice on its own. """ without = nut_app.post("/api/ups/command", json={"ups": "apc", "command": "driver.killpower"}) assert without.status_code == 422 assert "DANGER" in without.json()["detail"] assert WireRecorder.sent == [] with_danger = nut_app.post("/api/ups/command", json={"ups": "apc", "command": "driver.killpower", "confirm": "DANGER"}) assert with_danger.status_code == 200, with_danger.text assert with_danger.json()["risk"] == nut.RISK_DANGEROUS assert WireRecorder.sent == [("apc", "driver.killpower")] def test_a_cross_origin_write_is_refused_even_with_a_session_and_a_csrf_token( authed: TestClient) -> None: response = authed.post("/api/ups/command", json=WRITE, headers={"origin": "https://elsewhere.example"}) assert response.status_code == 403 assert "cross-origin" in response.json()["detail"] def test_a_same_origin_write_is_accepted(authed: TestClient) -> None: """`http://testserver` is what this client's own origin is: proof of a comparison, not a ban on the header being present.""" response = authed.post("/api/ups/command", json=WRITE, headers={"origin": "http://testserver"}) assert response.status_code == 404, "the origin check refused its own origin" assert response.json().get("configured") is False def test_behind_a_tls_proxy_the_origin_still_matches(authed: TestClient) -> None: """nginx forwards `Host: dash.example.com` and says `X-Forwarded-Proto: https`. Without reading that header, every write from the deployed dashboard would be a cross-origin refusal — the kind of bug that only appears in production, which is why it is tested here and not discovered during an outage. """ response = authed.post("/api/ups/command", json=WRITE, headers={"origin": "https://dash.example.com", "host": "dash.example.com", "x-forwarded-proto": "https"}) assert response.status_code == 404 assert response.json().get("configured") is False @pytest.mark.parametrize("site,expected", [ ("cross-site", 403), ("unknown-token", 403), ("same-origin", 404), ("none", 404), ]) def test_sec_fetch_site_is_believed_when_present(authed: TestClient, site: str, expected: int) -> None: """An unrecognised value is a refusal: the safe default for a browser token nobody has seen yet is the one that costs a frustrated operator, not a compromised UPS.""" response = authed.post("/api/ups/command", json=WRITE, headers={"sec-fetch-site": site}) assert response.status_code == expected def test_a_get_needs_neither_origin_nor_csrf(authed: TestClient) -> None: """`auth.SAFE_METHODS` is the whole rule: reads carry no CSRF risk, so they carry no gate.""" for site in ("cross-site", "same-origin", "nonsense"): response = authed.get("/api/config", headers={"sec-fetch-site": site, "origin": "https://elsewhere.example"}) assert response.status_code == 200, site def test_options_reaches_the_router_because_it_cannot_change_anything( authed: TestClient) -> None: """`OPTIONS` **is** in `auth.SAFE_METHODS` (`tests/test_auth.py` pins the constant), so the gate does not demand a CSRF token for it — and the router answers 405, because this app has no CORS middleware and sends no preflight answer by design. The asymmetry with a write is deliberate: `OPTIONS` cannot mutate anything, so gating it would protect nothing while making every browser preflight a 403 nobody can explain. If a CORS handler is ever added, it is added here and in `auth.SAFE_METHODS` together, not by letting the middleware decide on its own. """ del authed.headers[auth.CSRF_HEADER] response = authed.request("OPTIONS", "/api/ups/command") assert response.status_code == 405 assert "application/json" in response.headers["content-type"] # --------------------------------------------------------------------------- # # The browser half, on the bytes a browser gets # --------------------------------------------------------------------------- # def test_the_shared_fetch_attaches_the_token_only_to_writes(anon: TestClient) -> None: app_js = anon.get("/static/app.js").text assert "function ensureAuth" in app_js assert "/api/auth/status" in app_js # the memoised promise, not just a flag: five page scripts load at once on every dashboard assert re.search(r"if \(!authPromise\)", app_js) fetch = app_js[app_js.index("async function fetchJSON"):] assert "READ_METHODS.indexOf(method) < 0" in fetch assert "await ensureAuth()" in fetch assert "headers[CSRF_HEADER] = state.csrf" in fetch assert "response.status === 401" in fetch and "bounceToLogin()" in fetch def test_the_csrf_header_name_matches_the_server(anon: TestClient) -> None: """There is no build step to generate one from the other, so the literal *is* the contract.""" match = re.search(r"const CSRF_HEADER = '([^']+)';", anon.get("/static/app.js").text) assert match, "app.js no longer names the CSRF header" assert match.group(1) == auth.CSRF_HEADER def test_the_login_bounce_happens_once_per_page(anon: TestClient) -> None: """A 2-second poll loop whose session expired would otherwise stack history entries and keep re-navigating; one guard, one navigation.""" bounce = anon.get("/static/app.js").text bounce = bounce[bounce.index("function bounceToLogin"):] assert re.search(r"if \(bouncedToLogin\) return;", bounce) assert re.search(r"bouncedToLogin = true;", bounce) assert "encodeURIComponent(window.location.pathname)" in bounce def test_app_js_exports_the_session_for_the_page_that_shows_it(anon: TestClient) -> None: """`authStatus` is the live object task 05 prints "signed in · expires …" from.""" exports = anon.get("/static/app.js").text.rpartition("return {")[2] assert "authStatus" in exports and "ensureAuth" in exports def test_the_login_page_posts_through_the_shared_fetch(anon: TestClient) -> None: """One request path for the whole app, the login page included. Task 03 could not use `Dash.fetchJSON` here: it bounces every `401` to `/login`, which on this page is a reload, and a reload deletes the "Wrong password." before it can be read. Task 05 fixed the helper instead of forking it (`bounceToLogin()` answers "already here" by doing nothing — see `test_a_refusal_on_the_login_page_never_navigates`), so the form gets the `accept` header, the `ensureAuth` memo and a thrown `Error` carrying `status` like every other request in the app, and there is no second fetch implementation to drift. """ login_js = _strip(anon.get("/static/page-login.js").text) assert "Dash.fetchJSON('/api/auth/login'" in login_js assert "method: 'POST'" in login_js # and the refusal is text, never markup: the message comes from the server assert "textContent" in login_js and "innerHTML" not in login_js def test_a_refusal_on_the_login_page_never_navigates(anon: TestClient) -> None: """The guard that makes the above true, and the reason it is not a special case. `bounceToLogin()` exists so a poll loop whose session expired navigates once. On `/login` itself there is nowhere to bounce to, and the navigation is the bug: it would swallow the inline refusal. The path compared is `auth.LOGIN_PATH`, pinned as a literal because there is no build step to generate one from the other. """ app_js = anon.get("/static/app.js").text match = re.search(r"const LOGIN_PATH = '([^']+)';", app_js) assert match, "app.js no longer names the login path" assert match.group(1) == auth.LOGIN_PATH bounce = app_js[app_js.index("function bounceToLogin"):] assert re.search(r"if \(window\.location\.pathname === LOGIN_PATH\) return;", bounce) # and it still bounces everywhere else, once assert re.search(r"bouncedToLogin = true;", bounce) def test_the_refused_sign_in_renders_the_servers_own_words(anon: TestClient) -> None: """`401` is the one sentence; `429` is the server's `detail` and its `Retry-After`. Branching on `err.status` rather than on the message is the point: the body's prose is operator-facing and free to change, the status is the contract. And the countdown reads `Retry-After` because a wait the page made up would be an invented number (rule 2). """ login_js = _strip(anon.get("/static/page-login.js").text) assert "err.status === 401" in login_js assert "Wrong password." in login_js assert "err.status === 429" in login_js assert "err.headers.get('retry-after')" in login_js or "retry-after" in login_js # the helper carries the headers on the error, or the form could not read them fetch = anon.get("/static/app.js").text assert "error.headers = response.headers" in fetch def test_the_lockout_takes_the_button_out_until_the_wait_is_over(anon: TestClient) -> None: """A pressable button during a lockout is a button that spends more throttle for nothing.""" login_js = _strip(anon.get("/static/page-login.js").text) countdown = login_js[login_js.index("function countDown"):] assert "button.disabled = true" in countdown assert "button.disabled = false" in countdown assert "setInterval(" in countdown and "clearInterval(timer)" in login_js # a `429` that named no wait is not counted down, and so never disables anything forever assert re.search(r"if \(!Number\.isFinite\(seconds\) \|\| seconds <= 0\) return;", countdown) # and when it ends, the refusal goes with it: "try again in 59s" over a pressable button # is a number that no longer describes anything clear = countdown[countdown.index("if (left <= 0)"):] assert "write(noteLine, '');" in clear and "write(errorLine, '');" in clear assert "button.disabled = false;" in clear def test_the_login_page_judges_next_before_following_it(anon: TestClient) -> None: """The client copy of `auth.safe_next`, because one redirect is followed in the browser. `POST /api/auth/login` answers with a `next` the server already judged — but when the browser already holds a session the page redirects on its own, straight from the query string. That is the one place `?next=` is followed on client strength, so the same rule is restated here: an absolute path that is not scheme-relative, or nothing. """ login_js = _strip(anon.get("/static/page-login.js").text) judge = login_js[login_js.index("function target"):login_js.index("const form")] assert "next.startsWith('/')" in judge assert "next.startsWith('//')" in judge assert "next.startsWith('/\\\\')" in judge assert "return '/'" in judge # and both redirects go through it — no `location.*` with a raw query-string value for call in re.finditer(r"window\.location\.(assign|replace)\(([^)]*)\)", login_js): assert "target(" in call.group(2) or "requested(" in call.group(2), call.group(0) def test_nothing_stores_a_credential_where_a_page_can_read_it(anon: TestClient) -> None: """`localStorage` holds the theme and nothing else: no `sessionStorage`, no session in a URL. A session in `localStorage` is readable by any XSS on the page and is sent nowhere useful; a session in a URL lands in history, `Referer`, and screenshots. """ app_js = anon.get("/static/app.js").text theme = re.search(r"localStorage\.setItem\('([^']+)', root\.dataset\.theme\)", app_js) assert theme, "app.js no longer stores the theme" assert theme.group(1) == "dash-theme" assert "setItem('dash_session'" not in app_js assert "localStorage.setItem(auth" not in app_js assert "sessionStorage" not in app_js login_js = anon.get("/static/page-login.js").text assert "localStorage" not in login_js and "sessionStorage" not in login_js # the bounce puts the *path* in the query string — never a token assert auth.SESSION_COOKIE not in app_js def test_every_refusal_carries_no_store(anon: TestClient) -> None: """Whichever middleware wrote it: a cached 303 parks a browser on the login page after it has signed in, and a cached 401 will not come back without a hard reload.""" answers = [ anon.get("/system", headers=BROWSER_HEADERS), # 303 anon.get("/api/config", headers=JSON_ACCEPT), # 401 anon.post("/api/ups/command", json=WRITE), # 401 anon.post("/api/auth/login", json={"password": ""}), # 422 anon.post("/api/auth/login", json={"password": "x"}), # 401 anon.get("/api/nope", headers=JSON_ACCEPT), # 404 ] for response in answers: assert "no-store" in response.headers.get("cache-control", ""), response.url def test_the_served_route_table_is_the_surface_the_gate_describes(anon: TestClient) -> None: """Every HTML route this app serves, checked unsigned: pages redirect, `/login` answers. Reading the served app is the only honest way to notice a route added outside the gate — a new page that forgot to be protected would answer 200 here with fleet data on it. """ pages = sorted(str(route.path) for route in served_routes(anon) if isinstance(getattr(route, "path", None), str) and not str(route.path).startswith(("/api", "/static"))) assert pages == ["/", "/disks", "/llm", "/login", "/settings", "/system", "/ups"] for path in pages: response = anon.get(path, headers=BROWSER_HEADERS) if path == "/login": assert response.status_code == 200 else: assert response.status_code == 303, path assert response.headers["location"].startswith("/login?next=") def test_every_route_that_is_not_public_requires_a_session(anon: TestClient) -> None: """The default-deny proof, taken from the **served** route table and not from a list here. `PROTECTED_API`/`PROTECTED_PAGES` above are hand-written, so they can go stale in exactly the direction that matters: a future phase adds a route, assumes the gate is a decorator rather than a middleware, and every hand-written list in this file still passes. Walking `app.routes` is the one check that cannot be forgotten, which makes it the test that protects the next phase rather than this one. A `Mount` counts as its own prefix (`/static`), because that is what it serves: the asset tree and nothing else. A path the router answers 404 for is not in the table at all, so the typo case is `test_an_unknown_path_keeps_its_404_with_and_without_a_session`'s, not this one's. """ served = route_paths(served_routes(anon)) paths = sorted({path for path, _methods in served}) assert paths, "the app serves no routes at all — this test would pass on an empty table" # The known surface has to be *in* the table: a renamed route would otherwise drop out of # `gated` quietly and take its coverage with it. assert set(PROTECTED_API) | set(PROTECTED_PAGES) <= set(paths) # …and the public half of the served table is exactly what `PUBLIC_PATHS` says it is. assert {path for path in paths if is_public(path)} == { "/login", "/api/auth/login", "/api/health", "/static"} gated = [path for path in paths if not is_public(path)] assert len(gated) == len(paths) - 4, "the public surface grew without this test being read" assert {"/api/ups/command", "/api/auth/logout"} <= { # the two writes, both gated path for path, methods in served if "POST" in methods} for path in gated: page = anon.get(path, headers=BROWSER_HEADERS) data = anon.get(path, headers=JSON_ACCEPT) write = anon.post(path, json=WRITE) assert page.status_code in (303, 401), f"{path} answered {page.status_code} anonymously" assert data.status_code == 401, f"{path} answered {data.status_code} anonymously" assert data.json() == {"ok": False, "detail": "sign in required", "login": "/login"} assert write.status_code == 401, f"{path} took an anonymous {write.status_code}" for response in (page, data, write): assert "no-store" in response.headers["cache-control"], path assert not response.text.lstrip().startswith(" None: """The orphan-`id` audit every dashboard page gets, run on the login page too. It lives here rather than in `tests/test_api.py::PAGE_SCRIPTS` because that helper's client is signed in, and a signed-in `GET /login` is a 303 to the dashboard: fetched with redirects followed it would audit `/`'s markup under the name `/login` and pass on the wrong bytes. Unsigned, there is no redirect, and the same question gets asked — is any element here dead markup? The chrome assertion is the other half, and it is about the *nav*, not the header: the masthead's brand link and theme toggle belong on this page (same app, same tokens, and a dark room is a dark room before you sign in), but the tabs do not. Every tab leads to a page that bounces straight back here, and a nav of links that all redirect is a lie about what is available. The status dots and the clock go for the same reason in the other direction: there is nothing signed in for them to say yet. """ html = _strip(anon.get("/login").text) corpus = "".join(_strip(anon.get(f"/static/{name}").text) for name in ("app.js", "charts.js", "page-login.js", "dash.css")) anchors = set(re.findall( r'(?:aria-labelledby|aria-describedby|aria-controls|for)="([^"]+)"', html)) ids = set(re.findall(r'id="([^"]+)"', html)) assert ids, "the login page has nothing to reference, so this test would pass on an empty file" for element_id in sorted(ids): assert (f'"{element_id}"' in corpus or f"'{element_id}'" in corpus or f"#{element_id}" in corpus or element_id in anchors), \ f"/login has id=\"{element_id}\" that nothing reads" for dead in ('class="tabs"', 'aria-label="Dashboards"', 'id="nav-status"', 'id="clock"'): assert dead not in html, f"/login renders the {dead} of a page nobody is signed in to" # …and the brand link is there, because this is the same app and the operator should be # able to tell before they type a password into it. assert '' in html # --------------------------------------------------------------------------- # # `?next=` is judged, never followed # --------------------------------------------------------------------------- # @pytest.mark.parametrize("raw,expected", [ ("/llm", "/llm"), ("/llm?range=1h", "/llm?range=1h"), ("/", "/"), ]) def test_an_in_app_next_comes_back_as_the_landing_place(anon: TestClient, raw: str, expected: str) -> None: body = sign_in(anon, next=raw) assert body["next"] == expected @pytest.mark.parametrize("raw", [ "http://evil.example", "//evil.example", "/\\evil.example", "https://evil.example/x", "javascript:alert(1)", "/ok /not", "", "/a\nb", "/%2F%2Fevil.example", ]) def test_a_next_that_leaves_the_site_is_dropped_to_the_dashboard(anon: TestClient, raw: str) -> None: """The login page hands `next` straight to `location.replace`, so this is the open-redirect gate — and `/ok /not` is refused because the value goes into a `Location` header, where a space is not a path but a smuggled second one. """ body = sign_in(anon, next=raw) assert body["next"] == "/" assert "evil" not in str(body) def test_a_signed_in_visit_to_the_login_page_lands_on_its_next(authed: TestClient) -> None: """/login is not a dead end for someone already in: it 303s to `?next=`, or to `/`.""" response = authed.get("/login?next=%2Fdisks", headers=BROWSER_HEADERS) assert response.status_code == 303 assert response.headers["location"] == "/disks" assert "no-store" in response.headers["cache-control"] def test_a_signed_in_visit_to_the_login_page_drops_a_foreign_next(authed: TestClient) -> None: response = authed.get("/login?next=%2F%2Fevil.example", headers=BROWSER_HEADERS) assert response.status_code == 303 assert response.headers["location"] == "/" def test_a_signed_in_visit_to_the_login_page_without_a_next_goes_home( authed: TestClient) -> None: response = authed.get("/login", headers=BROWSER_HEADERS) assert response.status_code == 303 assert response.headers["location"] == "/" def test_a_cross_origin_login_post_is_refused_before_the_throttle(anon: TestClient) -> None: """A public write is still a write: the origin gate applies with no session behind it. The cost of not refusing is concrete — any page on the internet could spend this browser's login throttle, locking the operator out of their own dashboard while they read the news. """ before = context(anon).login_throttle.attempts("testclient") response = anon.post("/api/auth/login", json={"password": ADMIN_PASSWORD}, headers={"origin": "https://elsewhere.example"}) assert response.status_code == 403 assert "cross-origin" in response.json()["detail"] assert context(anon).login_throttle.attempts("testclient") == before def test_a_cross_origin_get_of_the_login_page_is_fine(anon: TestClient) -> None: """Only writes carry the origin requirement; a form is not a mutation.""" response = anon.get("/login", headers={"origin": "https://elsewhere.example"}) assert response.status_code == 200 # --------------------------------------------------------------------------- # # The two defensive dependencies, driven directly # --------------------------------------------------------------------------- # def _request(with_session: auth.Session | None = None) -> Request: scope: dict[str, Any] = {"type": "http", "method": "GET", "path": "/api/auth/status", "headers": [], "query_string": b"", "state": {}} request = Request(scope) if with_session is not None: request.state.session = with_session return request def test_get_session_answers_the_gate_or_returns_the_session() -> None: """`get_session` is a route's last line of defence, and its refusal is the caller's 401. Reached only if a protected route is added outside the gate — exactly the mistake worth a test, because the alternative is a 500 with a traceback or, worse, a route that reads a `None` session and serves a half-built payload. Driven directly: the HTTP layer above it cannot produce the missing-state case without deleting the gate. """ from dashboards.api import get_session # noqa: PLC0415 with pytest.raises(HTTPException) as raised: get_session(_request()) assert raised.value.status_code == 401 assert raised.value.detail == {"ok": False, "detail": "sign in required", "login": auth.LOGIN_PATH} session = auth.Session(username="admin", expires_at=time.time() + 60, token="t") assert get_session(_request(session)) is session def test_get_auth_state_answers_503_without_a_startup(store: SampleStore) -> None: """Same shape as the gate's answer, for a route that reached its dependencies.""" from dashboards.api import get_auth_state # noqa: PLC0415 bare = AppContext(settings=settings_for(store.path), store=store, hub=Hub(store), started_at=0.0) assert bare.auth is None with pytest.raises(HTTPException) as raised: get_auth_state(bare) assert raised.value.status_code == 503 assert raised.value.detail == {"ok": False, "detail": "startup did not complete"} def test_login_answers_503_when_the_context_has_no_credential(authed: TestClient) -> None: """A half-started app answers "startup did not complete" rather than "wrong password", so the operator's fix is not to type harder. The **gate** is what answers here: `state is None` is checked before the public list, so a process with no credential serves nothing at all (`test_the_gate_answers_503_when_startup_never_completed _pins`). The route's own equivalent check is driven directly below, because no request can reach it while the gate stands. """ context(authed).auth = None response = authed.post("/api/auth/login", json={"password": ADMIN_PASSWORD}) assert response.status_code == 503 assert response.json() == {"ok": False, "detail": "startup did not complete"} assert "no-store" in response.headers["cache-control"] async def test_the_login_route_refuses_without_a_credential(store: SampleStore) -> None: """The route's own `ctx.auth is None` branch, called past the gate. `AppContext.auth` is typed `AuthState | None`, so the login handler must not dereference it blindly: the alternative to this refusal is a 500 with a traceback on the one endpoint whose caller is already unable to do anything else. """ handler = endpoint_for( TestClient(create_app(settings_for(store.path), store), follow_redirects=False), "/api/auth/login") scope: dict[str, Any] = { "type": "http", "http_version": "1.1", "method": "POST", "scheme": "http", "path": "/api/auth/login", "raw_path": b"/api/auth/login", "query_string": b"", "root_path": "", "headers": [(b"content-type", b"application/json")], "client": ("198.51.100.4", 51_234), "server": ("testserver", 80), } async def receive() -> dict[str, Any]: return {"type": "http.request", "body": b'{"password": "whatever"}', "more_body": False} bare = AppContext(settings=settings_for(store.path), store=store, hub=Hub(store), started_at=0.0) assert bare.auth is None response = await handler(Request(scope, receive), bare) assert response.status_code == 503 assert "startup" in response.body.decode() assert ADMIN_PASSWORD not in response.body.decode() assert bare.login_throttle.attempts("198.51.100.4") == 0 # --------------------------------------------------------------------------- # # `auth.safe_next`, the function the two blocks above are built on # --------------------------------------------------------------------------- # @pytest.mark.parametrize("raw", [ "/", "/llm", "/disks", "/system?range=6h", "/llm?engine=gufo&range=15m", "/ups?host=apc#batteries", "/api/config", "/a%20path", "/disks?host=gateway-ops", ]) def test_safe_next_keeps_a_location_on_this_app(raw: str) -> None: """Kept verbatim, query and all: the chips live in the query string (task 03).""" assert auth.safe_next(raw) == raw @pytest.mark.parametrize("raw", [ None, "", "llm", "./llm", "http://evil.example", "https://evil.example/x", "//evil.example", "/\\evil.example", "/%2F%2Fevil.example", "/%2f%2fevil.example", "http:/evil.example", "javascript:alert(1)", "/x?u=http://evil.example", "/a\nb", "/a\rb", "/a\tb", "/a\x7fb", " /llm", ]) def test_safe_next_drops_every_location_that_is_not(raw: str | None) -> None: """Each of these leaves the site the login form belongs to, by one of four tricks. Scheme-prefixed, scheme-relative (with the old backslash spelling), the same trick percent-encoded for a reader that decodes before resolving the authority, and a control character — which is not a path at all but a second `Location` header. `None` means "no opinion", and every caller falls back to `/`. """ assert auth.safe_next(raw) is None def test_safe_next_is_idempotent_and_never_widens(anon: TestClient) -> None: """Running the keeper twice cannot let through what the first pass refused. The value is read once by the gate and once by the login route, and a `next` that grew a scheme on the second pass would be a redirect nobody re-checked. """ for raw in ("/llm", "/llm?range=1h", "//evil.example", "/%2F%2Fevil.example", "", "x"): once = auth.safe_next(raw) assert auth.safe_next(once) == once del anon