dashboards/tests/test_auth_api.py

"""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 `<a href="/logout">` 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:<address>`). 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 `<img src=…>` 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("<!doctype"), \
                f"{path} served a dashboard page to an anonymous caller"


def test_the_login_page_carries_no_id_nothing_reads_and_no_dashboard_nav(anon: TestClient) -> 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 '<a class="brand" href="/">' 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