From a206e307928bb59c9ee0b8b479e67bf29f3167ea Mon Sep 17 00:00:00 2001 From: Ben Date: Thu, 4 Jun 2026 17:26:18 +1000 Subject: [PATCH 1/4] feat(dashboard-auth): add pluggable password (non-redirect) login MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dashboard auth gate was OAuth-only: a DashboardAuthProvider could authenticate only via a redirect to an IDP (start_login -> /auth/callback -> complete_login). There was no first-class path for username/password auth, so self-hosters who just want a password on their dashboard had no clean option short of an external OAuth IDP. Extend the provider framework with a parallel, non-redirect front door that converges on the same Session + cookie + refresh machinery: - base.py: add the optional supports_password flag and complete_password_login(username, password) -> Session (default raises NotImplementedError so an OAuth-only provider that forgets the flag fails loudly). Add InvalidCredentialsError. OAuth providers are unaffected (flag defaults False; the method is never called). - routes.py: add POST /auth/password-login, mirroring the cookie-minting tail of /auth/callback but skipping PKCE/state/code. Returns JSON {ok, next} (the form POSTs via fetch). Generic 401 for both unknown user and wrong password (no enumeration oracle); 404 hides whether a provider exists or supports passwords; per-IP sliding-window rate limit (10/min -> 429). /api/auth/providers now reports supports_password so the login page can branch. - middleware.py: allowlist /auth/password-login (a bootstrap route). verify/refresh/revoke/ws-tickets/logout need zero changes — a password session is just a Session with provider-minted opaque tokens. - login_page.py: render a credential form (instead of a redirect button) for supports_password providers, wired by a small inline script that POSTs to /auth/password-login and navigates on success. OAuth-only pages stay script-free. --- hermes_cli/dashboard_auth/__init__.py | 2 + hermes_cli/dashboard_auth/base.py | 62 +++++++++ hermes_cli/dashboard_auth/login_page.py | 162 +++++++++++++++++++++++- hermes_cli/dashboard_auth/middleware.py | 1 + hermes_cli/dashboard_auth/routes.py | 160 ++++++++++++++++++++++- 5 files changed, 379 insertions(+), 8 deletions(-) diff --git a/hermes_cli/dashboard_auth/__init__.py b/hermes_cli/dashboard_auth/__init__.py index 4a5c68b6e4e26..faba37610384d 100644 --- a/hermes_cli/dashboard_auth/__init__.py +++ b/hermes_cli/dashboard_auth/__init__.py @@ -14,6 +14,7 @@ Session, LoginStart, InvalidCodeError, + InvalidCredentialsError, ProviderError, RefreshExpiredError, assert_protocol_compliance, @@ -30,6 +31,7 @@ "Session", "LoginStart", "InvalidCodeError", + "InvalidCredentialsError", "ProviderError", "RefreshExpiredError", "assert_protocol_compliance", diff --git a/hermes_cli/dashboard_auth/base.py b/hermes_cli/dashboard_auth/base.py index 207c7c602d4ad..06dab5dd5a416 100644 --- a/hermes_cli/dashboard_auth/base.py +++ b/hermes_cli/dashboard_auth/base.py @@ -55,6 +55,16 @@ class InvalidCodeError(Exception): """ +class InvalidCredentialsError(Exception): + """A username/password pair was rejected by a password provider. + + Raised by :meth:`DashboardAuthProvider.complete_password_login`. The + ``/auth/password-login`` route translates this to HTTP 401 with a + deliberately generic detail (never distinguishing "unknown user" from + "wrong password") so the endpoint can't be used as a username oracle. + """ + + class RefreshExpiredError(Exception): """The refresh token is dead. @@ -94,11 +104,33 @@ class DashboardAuthProvider(ABC): Subclasses MUST set ``name`` (lowercase identifier, stable forever) and ``display_name`` (user-facing label on the login page). + + Password (non-redirect) providers: + A provider that authenticates with a username + password instead of + an OAuth redirect sets ``supports_password = True`` and implements + ``complete_password_login``. The login page then renders a + credential form (POSTing to ``/auth/password-login``) instead of a + "Log in with X" redirect button. Everything downstream of login — + ``verify_session`` / ``refresh_session`` / ``revoke_session``, the + session cookies, the WS-ticket mint — is identical to the OAuth + path, because a password session is just a :class:`Session` with + provider-minted opaque tokens. The OAuth methods (``start_login`` / + ``complete_login``) remain abstract; a pure-password provider that + will never be reached via the redirect flow may implement them as + stubs that raise ``NotImplementedError``. """ name: str = "" display_name: str = "" + # When True, this provider authenticates via username + password + # (``complete_password_login``) rather than (or in addition to) the + # OAuth redirect flow. The login page renders a credential form for + # such providers; the ``/auth/password-login`` route dispatches to + # ``complete_password_login``. OAuth-only providers leave this False + # and are completely unaffected. + supports_password: bool = False + @abstractmethod def start_login(self, *, redirect_uri: str) -> LoginStart: ... @@ -121,6 +153,36 @@ def refresh_session(self, *, refresh_token: str) -> Session: ... @abstractmethod def revoke_session(self, *, refresh_token: str) -> None: ... + def complete_password_login( + self, *, username: str, password: str + ) -> "Session": + """Verify a username/password pair and mint a :class:`Session`. + + Only called when ``supports_password`` is True (the + ``/auth/password-login`` route guards on the flag). The default + raises ``NotImplementedError`` so an OAuth-only provider that + forgets to set the flag fails loudly rather than silently + accepting credentials. + + The returned ``Session`` carries provider-minted opaque + ``access_token`` / ``refresh_token`` exactly like the OAuth path, + so all downstream session handling (cookies, verify, refresh, + ws-tickets, logout) is identical. + + Failure semantics: + * ``InvalidCredentialsError`` — username/password rejected. The + route surfaces a generic 401 (no user-vs-password + distinction). Implementations SHOULD spend constant time on + unknown users (dummy hash verify) to avoid a timing oracle. + * ``ProviderError`` — the backing credential store is + unreachable (LDAP/DB down); the route surfaces 503. + """ + raise NotImplementedError( + f"{type(self).__name__} does not support password login " + "(set supports_password = True and override " + "complete_password_login)" + ) + def assert_protocol_compliance(cls: type) -> None: """Raise ``TypeError`` if ``cls`` doesn't fully implement the provider protocol. diff --git a/hermes_cli/dashboard_auth/login_page.py b/hermes_cli/dashboard_auth/login_page.py index 74da4dbe2f029..6459445486b7b 100644 --- a/hermes_cli/dashboard_auth/login_page.py +++ b/hermes_cli/dashboard_auth/login_page.py @@ -225,6 +225,56 @@ class name MUST NOT change without updating outline-offset: 3px; }} + /* Password provider form — same visual language as the OAuth buttons: + squared inputs, hairline borders, amber focus ring. */ + .provider-form {{ + display: grid; + gap: 0.75rem; + text-align: left; + }} + .form-title {{ + font-family: 'Rules Compressed', 'Collapse', sans-serif; + font-weight: 600; + font-size: 0.72rem; + letter-spacing: 0.18em; + text-transform: uppercase; + color: color-mix(in srgb, var(--foreground) 70%, transparent); + }} + .field {{ + display: grid; + gap: 0.3rem; + }} + .field-label {{ + font-size: 0.72rem; + letter-spacing: 0.12em; + text-transform: uppercase; + color: color-mix(in srgb, var(--foreground) 55%, transparent); + }} + .field-input {{ + width: 100%; + box-sizing: border-box; + padding: 0.7rem 0.8rem; + background: color-mix(in srgb, #000000 25%, var(--background-base)); + color: var(--foreground); + border: 1px solid var(--hairline-strong); + border-radius: 0; + font-family: 'Collapse', sans-serif; + font-size: 0.95rem; + }} + .field-input:focus-visible {{ + outline: none; + border-color: var(--midground); + box-shadow: 0 0 0 1px var(--midground); + }} + .form-error {{ + color: #ff6b6b; + font-size: 0.82rem; + letter-spacing: 0.02em; + }} + .provider-form .provider-btn {{ + margin-top: 0.25rem; + }} + footer {{ margin-top: 1.75rem; text-align: center; @@ -264,6 +314,7 @@ class name MUST NOT change without updating Public bind · Auth required +{password_script} """ @@ -350,6 +401,60 @@ class name MUST NOT change without updating """ +# Inline script that wires every password provider form to POST JSON to +# ``/auth/password-login`` and navigate on success. Emitted ONLY when at +# least one ``supports_password`` provider is listed (OAuth-only login +# pages stay script-free, preserving the no-JS contract for that case). +# +# Plain string (NOT run through ``str.format``), so braces are literal — +# do not double them. A single delegated submit handler covers all forms; +# the provider name is read from the form's ``data-provider`` attribute. +_PASSWORD_FORM_SCRIPT = """\ + +""" + + def render_login_html(*, next_path: str = "") -> str: """Return the full HTML for ``GET /login``. @@ -375,10 +480,55 @@ def render_login_html(*, next_path: str = "") -> str: next_qs = "" buttons = [] + needs_password_script = False for p in providers: - buttons.append( - f' ' - f'Sign in with {html.escape(p.display_name)}' - ) - return _LOGIN_HTML_TEMPLATE.format(provider_buttons="\n".join(buttons)) + if getattr(p, "supports_password", False): + needs_password_script = True + buttons.append(_render_password_form(p, next_path)) + else: + buttons.append( + f' ' + f'Sign in with {html.escape(p.display_name)}' + ) + script = _PASSWORD_FORM_SCRIPT if needs_password_script else "" + return _LOGIN_HTML_TEMPLATE.format( + provider_buttons="\n".join(buttons), + password_script=script, + ) + + +def _render_password_form(provider, next_path: str) -> str: + """Render a username/password form for a ``supports_password`` provider. + + The form is wired by :data:`_PASSWORD_FORM_SCRIPT` (a single delegated + submit handler) to POST JSON to ``/auth/password-login`` and navigate + on success. ``next_path`` is carried in a hidden field; it has already + been validated same-origin by the caller and is HTML-escaped here as + defence in depth. The provider ``name`` is emitted in a ``data-`` + attribute (not a hidden input) so the script reads it without trusting + form-field ordering. + """ + pname = html.escape(provider.name, quote=True) + plabel = html.escape(provider.display_name) + safe_next = html.escape(next_path, quote=True) if next_path else "" + return ( + f'
\n' + f'
Sign in with {plabel}
\n' + f' \n' + f' \n' + f' \n' + f' \n' + f' \n' + f'
' + ) diff --git a/hermes_cli/dashboard_auth/middleware.py b/hermes_cli/dashboard_auth/middleware.py index 8c6216e9ed6a9..0e25c1033eabd 100644 --- a/hermes_cli/dashboard_auth/middleware.py +++ b/hermes_cli/dashboard_auth/middleware.py @@ -38,6 +38,7 @@ _GATE_PUBLIC_PREFIXES: tuple[str, ...] = ( "/auth/login", "/auth/callback", + "/auth/password-login", "/auth/logout", "/login", "/api/auth/providers", diff --git a/hermes_cli/dashboard_auth/routes.py b/hermes_cli/dashboard_auth/routes.py index 637cb7048d06a..68ca1886ca8ba 100644 --- a/hermes_cli/dashboard_auth/routes.py +++ b/hermes_cli/dashboard_auth/routes.py @@ -16,11 +16,14 @@ from __future__ import annotations import logging +import threading import time -from typing import Any +from collections import defaultdict, deque +from typing import Any, Deque, Dict, Tuple from fastapi import APIRouter, HTTPException, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse +from pydantic import BaseModel from hermes_cli.dashboard_auth import ( get_provider, @@ -29,6 +32,7 @@ from hermes_cli.dashboard_auth.audit import AuditEvent, audit_log from hermes_cli.dashboard_auth.base import ( InvalidCodeError, + InvalidCredentialsError, ProviderError, ) from hermes_cli.dashboard_auth.cookies import ( @@ -154,7 +158,13 @@ async def api_auth_providers() -> Any: ) return { "providers": [ - {"name": p.name, "display_name": p.display_name} + { + "name": p.name, + "display_name": p.display_name, + "supports_password": bool( + getattr(p, "supports_password", False) + ), + } for p in providers ], } @@ -377,6 +387,152 @@ def _validate_post_login_target(raw: str) -> str: return decoded +# --------------------------------------------------------------------------- +# Public: password (non-redirect) login +# --------------------------------------------------------------------------- +# +# Brute-force throttle. The OAuth flow has no guessable secret on our side +# (the IDP owns credentials), but ``/auth/password-login`` accepts a +# password we verify locally, so it's a credential-stuffing target. A +# simple in-process sliding-window limiter per client IP raises the cost +# of online guessing without any external dependency. It is intentionally +# best-effort: process-local (resets on restart), and behind a trusting +# proxy the IP is the proxy's unless X-Forwarded-For is set — which is why +# this is defence-in-depth on top of the provider's own constant-time +# verify, not the only line of defence. + +_PW_RATE_MAX_ATTEMPTS = 10 +_PW_RATE_WINDOW_SEC = 60.0 +_pw_attempts: Dict[str, Deque[float]] = defaultdict(deque) +_pw_attempts_lock = threading.Lock() + + +def _password_rate_limited(ip: str) -> bool: + """True if ``ip`` has exceeded the password-login attempt budget. + + Sliding window: prune attempts older than the window, then check the + count. Records the attempt timestamp when allowed. An empty IP (no + discernible client) shares a single bucket — fail-safe toward + throttling rather than letting unattributable traffic through + unmetered. + """ + now = time.monotonic() + cutoff = now - _PW_RATE_WINDOW_SEC + key = ip or "_unknown_" + with _pw_attempts_lock: + bucket = _pw_attempts[key] + while bucket and bucket[0] < cutoff: + bucket.popleft() + if len(bucket) >= _PW_RATE_MAX_ATTEMPTS: + return True + bucket.append(now) + return False + + +def _reset_password_rate_limit() -> None: + """Test-only: clear all rate-limit buckets.""" + with _pw_attempts_lock: + _pw_attempts.clear() + + +class _PasswordLoginBody(BaseModel): + provider: str + username: str + password: str + next: str = "" + + +@router.post("/auth/password-login", name="auth_password_login") +async def auth_password_login(request: Request, body: _PasswordLoginBody): + """Authenticate a username/password against a password provider. + + Mirrors the cookie-minting tail of ``/auth/callback`` but skips the + PKCE/state/code machinery (those are OAuth-only). On success sets the + session cookies and returns JSON ``{"ok": true, "next": }`` — + the credential form POSTs via fetch and navigates client-side, so a + 302 (which fetch follows opaquely) is the wrong shape here. + + Failure modes, all deliberately generic so the endpoint can't be used + as a username oracle or a provider-enumeration oracle: + * unknown provider / provider lacks password support → 404 + * bad credentials → 401 ("Invalid credentials") + * backing store unreachable → 503 + * too many attempts from this IP → 429 + """ + ip = _client_ip(request) + if _password_rate_limited(ip): + audit_log( + AuditEvent.LOGIN_FAILURE, + provider=body.provider, + reason="rate_limited", + ip=ip, + ) + raise HTTPException( + status_code=429, + detail="Too many login attempts. Try again shortly.", + ) + + p = get_provider(body.provider) + if p is None or not getattr(p, "supports_password", False): + # Don't leak which providers exist or which support passwords — + # same 404 whether the provider is unknown or OAuth-only. + audit_log( + AuditEvent.LOGIN_FAILURE, + provider=body.provider, + reason="unknown_password_provider", + ip=ip, + ) + raise HTTPException(status_code=404, detail="Unknown provider") + + try: + session = p.complete_password_login( + username=body.username, password=body.password + ) + except InvalidCredentialsError: + audit_log( + AuditEvent.LOGIN_FAILURE, + provider=body.provider, + reason="invalid_credentials", + ip=ip, + ) + # Generic message — never distinguish unknown-user from wrong-password. + raise HTTPException(status_code=401, detail="Invalid credentials") + except NotImplementedError: + # supports_password was True but the method isn't actually + # implemented — a provider bug, not a client error. + raise HTTPException(status_code=500, detail="Provider misconfigured") + except ProviderError as e: + audit_log( + AuditEvent.LOGIN_FAILURE, + provider=body.provider, + reason="provider_unreachable", + ip=ip, + ) + raise HTTPException(status_code=503, detail=f"Provider unreachable: {e}") + + audit_log( + AuditEvent.LOGIN_SUCCESS, + provider=body.provider, + user_id=session.user_id, + email=session.email, + org_id=session.org_id, + ip=ip, + ) + + expires_in = max(60, session.expires_at - int(time.time())) + landing = _validate_post_login_target(body.next) or "/" + resp = JSONResponse({"ok": True, "next": landing}) + set_session_cookies( + resp, + access_token=session.access_token, + refresh_token=session.refresh_token, + access_token_expires_in=expires_in, + use_https=detect_https(request), + prefix=_prefix(request), + ) + return resp + + @router.post("/auth/logout", name="auth_logout") async def auth_logout(request: Request): _at, rt = read_session_cookies(request) From 960d62ae3d8ebdfaad4f9f5830ce4aa80e4d2064 Mon Sep 17 00:00:00 2001 From: Ben Date: Thu, 4 Jun 2026 17:26:32 +1000 Subject: [PATCH 2/4] feat(dashboard-auth): add BasicAuthProvider username/password plugin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A bundled, zero-infrastructure 'just put a password on my dashboard' provider that uses the supports_password extension point. No external IDP, no database: sessions are stateless HMAC-signed tokens the provider mints and verifies itself, and passwords are hashed with stdlib scrypt (no third-party dependency — deliberately avoids bcrypt to keep the dep surface unchanged). - plugins/dashboard_auth/basic: BasicAuthProvider (scrypt verify with a constant-time dummy-hash path for unknown users so the endpoint is not a username-timing oracle; access/refresh tokens carry a 'kind' claim that verify/refresh enforce; cross-secret tokens are rejected). The register() entry point mirrors the Nous plugin's config/env precedence (env wins; empty treated as unset) and LAST_SKIP_REASON channel. - config.py: document the canonical dashboard.basic_auth.* surface (username / password_hash / password / secret / session_ttl_seconds). Activates only when username + (password or password_hash) are set, so OAuth users and loopback/--insecure operators are unaffected. Without an explicit secret a random per-process key is generated (logged): fine for a single process, but sessions then don't survive restart or span workers. --- hermes_cli/config.py | 28 ++ plugins/dashboard_auth/basic/__init__.py | 491 +++++++++++++++++++++++ plugins/dashboard_auth/basic/plugin.yaml | 7 + 3 files changed, 526 insertions(+) create mode 100644 plugins/dashboard_auth/basic/__init__.py create mode 100644 plugins/dashboard_auth/basic/plugin.yaml diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 1b72c31527321..a6948c23f2f2e 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1482,6 +1482,34 @@ def _ensure_hermes_home_managed(home: Path): "client_id": "", # agent:{instance_id} — Portal provisions this "portal_url": "", # blank → use plugin default (production Portal) }, + # Username/password gate configuration — read by the bundled + # ``dashboard_auth/basic`` plugin (a self-hosted "just put a + # password on my dashboard" provider that needs no OAuth IDP). + # The plugin registers a password provider when ``username`` plus + # either ``password_hash`` (preferred — no plaintext at rest) or + # ``password`` (plaintext, hashed in-memory at load) are set. Each + # key is overridable by an env var + # (``HERMES_DASHBOARD_BASIC_AUTH_USERNAME`` / + # ``_PASSWORD_HASH`` / ``_PASSWORD`` / ``_SECRET`` / + # ``_TTL_SECONDS``), env winning when non-empty. Leave ``username`` + # empty (the default) to keep the plugin a no-op — loopback / + # ``--insecure`` operators and OAuth users are unaffected. + # + # ``secret`` is the HMAC key used to sign the stateless session + # tokens this provider mints. When empty, a random per-process key + # is generated — fine for a single process, but sessions then + # don't survive a restart or span multiple workers. Set an + # explicit ``secret`` (32+ random bytes, base64/hex/raw) for + # stable multi-worker / restart-surviving sessions. Compute a + # ``password_hash`` with + # ``python -c "from plugins.dashboard_auth.basic import hash_password; print(hash_password('PW'))"``. + "basic_auth": { + "username": "", # blank → plugin no-op (no password provider) + "password_hash": "", # scrypt$... (preferred — no plaintext at rest) + "password": "", # plaintext fallback (hashed in-memory at load) + "secret": "", # token-signing key; blank → random per-process + "session_ttl_seconds": 0, # 0 → plugin default (12h) + }, # Public URL override (env: ``HERMES_DASHBOARD_PUBLIC_URL``). # When set, this is the complete authority — scheme + host + # optional path prefix (e.g. ``https://example.com/hermes``) — diff --git a/plugins/dashboard_auth/basic/__init__.py b/plugins/dashboard_auth/basic/__init__.py new file mode 100644 index 0000000000000..12ec0fe51355c --- /dev/null +++ b/plugins/dashboard_auth/basic/__init__.py @@ -0,0 +1,491 @@ +"""BasicAuthProvider — username/password dashboard auth (no OAuth IDP). + +A self-hosted "just put a password on my dashboard" provider. It plugs +into the same ``DashboardAuthProvider`` framework as the Nous OAuth +provider, but authenticates with a username + password instead of an +OAuth redirect: it sets ``supports_password = True`` and implements +``complete_password_login``. The login page renders a credential form for +it; everything downstream of login (session cookies, verify, refresh, +ws-tickets, logout) is identical to the OAuth path because a password +session is just a :class:`Session` with provider-minted opaque tokens. + +This provider has **no external IDP and no database**. Credentials are +configured up front; sessions are stateless HMAC-signed tokens this +provider mints and verifies itself. That keeps it zero-infrastructure — +appropriate for a single-box self-hosted dashboard. + +Configuration surfaces (env wins over config.yaml when set non-empty), +mirroring the Nous provider's precedence convention: + + ``config.yaml`` — canonical surface:: + + dashboard: + basic_auth: + username: admin # required + # Provide EITHER a precomputed scrypt hash (preferred — no + # plaintext at rest) ... + password_hash: "scrypt$..." # see hash_password() + # ... OR a plaintext password (hashed in-memory at load). + password: "s3cret" + secret: "<32+ random bytes, base64 or hex>" # optional; token-signing key + session_ttl_seconds: 43200 # optional; access-token lifetime (default 12h) + + Environment overrides:: + + HERMES_DASHBOARD_BASIC_AUTH_USERNAME + HERMES_DASHBOARD_BASIC_AUTH_PASSWORD_HASH # preferred + HERMES_DASHBOARD_BASIC_AUTH_PASSWORD # plaintext fallback + HERMES_DASHBOARD_BASIC_AUTH_SECRET + HERMES_DASHBOARD_BASIC_AUTH_TTL_SECONDS + +If ``secret`` is not configured, a random per-process secret is generated +at startup. That's fine for a single-process dashboard, but means all +sessions are invalidated on restart and sessions don't survive across +multiple worker processes — set an explicit ``secret`` for stable +multi-worker / restart-surviving sessions. + +Password hashing uses stdlib :func:`hashlib.scrypt` (memory-hard, no +third-party dependency). ``complete_password_login`` runs a constant-time +comparison and always performs a hash even for an unknown username, so +the endpoint is not a username-enumeration timing oracle. + +Skip reasons: + Like the Nous provider, this exposes a module-level ``LAST_SKIP_REASON`` + the gate's fail-closed branch can surface when the plugin loads but + declines to register (no username/password configured). +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import logging +import os +import secrets +import time +from typing import Any, Optional + +from hermes_cli.dashboard_auth import ( + DashboardAuthProvider, + InvalidCredentialsError, + LoginStart, + RefreshExpiredError, + Session, +) + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Defaults +# --------------------------------------------------------------------------- + +# Access-token lifetime. The middleware transparently refreshes via the +# refresh token (30-day) when the access token lapses, so this controls +# how often a refresh round trip happens, not how long the user stays +# logged in. +_DEFAULT_TTL_SECONDS = 12 * 60 * 60 # 12h +_REFRESH_TTL_SECONDS = 30 * 24 * 60 * 60 # 30d + +# scrypt parameters (RFC 7914 / stdlib hashlib.scrypt). n must be a power +# of two; these are the widely-recommended interactive-login parameters +# (~16 MiB, a few ms on commodity hardware). +_SCRYPT_N = 2**14 +_SCRYPT_R = 8 +_SCRYPT_P = 1 +_SCRYPT_DKLEN = 32 +_SCRYPT_SALT_BYTES = 16 + +# Length of the HMAC-SHA256 digest appended as a fixed-length suffix to +# signed tokens (no separator — binary HMAC bytes can't be confused with +# a delimiter). +_SIG_LEN = hashlib.sha256().digest_size + + +LAST_SKIP_REASON: str = "" + + +# --------------------------------------------------------------------------- +# Password hashing (stdlib scrypt) +# --------------------------------------------------------------------------- + + +def hash_password(password: str) -> str: + """Return a ``scrypt$n$r$p$$`` hash string. + + Use this to precompute ``password_hash`` for config.yaml so plaintext + never sits at rest. Exposed as a module function so operators can run + ``python -c "from plugins.dashboard_auth.basic import hash_password; + print(hash_password('pw'))"``. + """ + salt = secrets.token_bytes(_SCRYPT_SALT_BYTES) + dk = hashlib.scrypt( + password.encode("utf-8"), + salt=salt, + n=_SCRYPT_N, + r=_SCRYPT_R, + p=_SCRYPT_P, + dklen=_SCRYPT_DKLEN, + maxmem=0, + ) + return ( + f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}$" + f"{base64.b64encode(salt).decode()}${base64.b64encode(dk).decode()}" + ) + + +def _verify_password(password: str, encoded: str) -> bool: + """Constant-time scrypt verify. False on any malformed hash string.""" + try: + scheme, n_s, r_s, p_s, salt_b64, dk_b64 = encoded.split("$") + if scheme != "scrypt": + return False + n, r, p = int(n_s), int(r_s), int(p_s) + salt = base64.b64decode(salt_b64) + expected = base64.b64decode(dk_b64) + except (ValueError, TypeError): + return False + try: + actual = hashlib.scrypt( + password.encode("utf-8"), + salt=salt, + n=n, + r=r, + p=p, + dklen=len(expected), + maxmem=0, + ) + except (ValueError, MemoryError): + return False + return hmac.compare_digest(actual, expected) + + +# A fixed dummy hash used to spend ~equal time when the username is +# unknown, so an attacker can't distinguish "no such user" (fast) from +# "wrong password" (slow scrypt) by timing. Computed once at import. +_DUMMY_HASH = hash_password("dummy-password-for-constant-time-verify") + + +# --------------------------------------------------------------------------- +# Token signing (stateless HMAC-signed blobs) +# --------------------------------------------------------------------------- + + +def _sign(payload: dict, secret: bytes) -> str: + raw = json.dumps(payload, separators=(",", ":")).encode() + sig = hmac.new(secret, raw, hashlib.sha256).digest() + return base64.urlsafe_b64encode(raw + sig).decode() + + +def _unsign(token: str, secret: bytes) -> Optional[dict]: + try: + blob = base64.urlsafe_b64decode(token.encode()) + if len(blob) <= _SIG_LEN: + return None + raw, sig = blob[:-_SIG_LEN], blob[-_SIG_LEN:] + expected = hmac.new(secret, raw, hashlib.sha256).digest() + if not hmac.compare_digest(sig, expected): + return None + return json.loads(raw) + except Exception: + return None + + +# --------------------------------------------------------------------------- +# Provider +# --------------------------------------------------------------------------- + + +class BasicAuthProvider(DashboardAuthProvider): + """Username/password provider with stateless HMAC-signed sessions.""" + + name = "basic" + display_name = "Username & Password" + supports_password = True + + def __init__( + self, + *, + username: str, + password_hash: str, + secret: bytes, + ttl_seconds: int = _DEFAULT_TTL_SECONDS, + ) -> None: + if not username: + raise ValueError("username must be non-empty") + if not password_hash: + raise ValueError("password_hash must be non-empty") + if len(secret) < 16: + raise ValueError("secret must be at least 16 bytes") + self._username = username + self._password_hash = password_hash + self._secret = secret + self._ttl = max(60, int(ttl_seconds)) + + # ---- OAuth methods: not used (pure-password provider) ------------------ + + def start_login(self, *, redirect_uri: str) -> LoginStart: + raise NotImplementedError( + "BasicAuthProvider is password-only; there is no OAuth redirect " + "flow. The login page POSTs to /auth/password-login instead." + ) + + def complete_login( + self, *, code: str, state: str, code_verifier: str, redirect_uri: str + ) -> Session: + raise NotImplementedError( + "BasicAuthProvider is password-only; use complete_password_login." + ) + + # ---- password login ---------------------------------------------------- + + def complete_password_login( + self, *, username: str, password: str + ) -> Session: + # Constant-time-ish: always run a scrypt verify (against the real + # hash if the username matches, else a dummy hash) so an unknown + # username and a wrong password take comparable time. Compare the + # username with compare_digest too, to avoid a length/byte timing + # leak on the username itself. + username_ok = hmac.compare_digest( + username.encode("utf-8"), self._username.encode("utf-8") + ) + target_hash = self._password_hash if username_ok else _DUMMY_HASH + password_ok = _verify_password(password, target_hash) + if not (username_ok and password_ok): + raise InvalidCredentialsError("invalid username or password") + return self._mint_session(self._username) + + # ---- session lifecycle ------------------------------------------------- + + def verify_session(self, *, access_token: str) -> Optional[Session]: + payload = _unsign(access_token, self._secret) + if ( + payload is None + or payload.get("kind") != "access" + or payload.get("exp", 0) <= int(time.time()) + ): + return None + return self._session_from_payload(access_token, "", payload) + + def refresh_session(self, *, refresh_token: str) -> Session: + if not refresh_token: + raise RefreshExpiredError("no refresh token present in session") + payload = _unsign(refresh_token, self._secret) + if ( + payload is None + or payload.get("kind") != "refresh" + or payload.get("exp", 0) <= int(time.time()) + ): + raise RefreshExpiredError("refresh token expired or invalid") + return self._mint_session(str(payload.get("sub", self._username))) + + def revoke_session(self, *, refresh_token: str) -> None: + # Stateless tokens — nothing to revoke server-side. The session + # expires within its TTL. Best-effort no-op, must not raise. + _ = refresh_token + return None + + # ---- internals --------------------------------------------------------- + + def _mint_session(self, user_id: str) -> Session: + now = int(time.time()) + exp = now + self._ttl + access_token = _sign( + {"sub": user_id, "kind": "access", "exp": exp}, self._secret + ) + refresh_token = _sign( + {"sub": user_id, "kind": "refresh", "exp": now + _REFRESH_TTL_SECONDS}, + self._secret, + ) + return Session( + user_id=user_id, + email="", + display_name=user_id, + org_id="", + provider=self.name, + expires_at=exp, + access_token=access_token, + refresh_token=refresh_token, + ) + + def _session_from_payload( + self, access_token: str, refresh_token: str, payload: dict + ) -> Session: + user_id = str(payload.get("sub", "")) + return Session( + user_id=user_id, + email="", + display_name=user_id, + org_id="", + provider=self.name, + expires_at=int(payload["exp"]), + access_token=access_token, + refresh_token=refresh_token, + ) + + +# --------------------------------------------------------------------------- +# Plugin entry point +# --------------------------------------------------------------------------- + + +def _load_config_basic_auth_section() -> dict: + """Return ``dashboard.basic_auth`` from config.yaml, or ``{}``. + + Robust to load_config() raising, the keys being absent, or the value + not being a dict — every shape falls through to ``{}``. + """ + try: + from hermes_cli.config import cfg_get, load_config + + cfg = load_config() + except Exception as exc: # noqa: BLE001 — broad catch is intentional + logger.debug( + "dashboard-auth-basic: load_config() raised %s; " + "falling back to env-only configuration", + exc, + ) + return {} + section = cfg_get(cfg, "dashboard", "basic_auth", default=None) + return section if isinstance(section, dict) else {} + + +def _resolve(env_name: str, cfg_section: dict, cfg_key: str) -> str: + """Env-wins-over-config resolution; empty env treated as unset.""" + env = os.environ.get(env_name, "").strip() + if env: + return env + return str(cfg_section.get(cfg_key, "") or "").strip() + + +def _resolve_secret(cfg_section: dict) -> bytes: + """Resolve the token-signing secret. + + Accepts base64 or hex or raw text from config/env. When unset, + generates a random per-process secret (sessions then don't survive a + restart or span multiple workers — logged at INFO). + """ + raw = _resolve( + "HERMES_DASHBOARD_BASIC_AUTH_SECRET", cfg_section, "secret" + ) + if not raw: + logger.info( + "dashboard-auth-basic: no 'secret' configured; generating a " + "random per-process signing key. Sessions will not survive a " + "restart or span multiple workers. Set dashboard.basic_auth." + "secret (or HERMES_DASHBOARD_BASIC_AUTH_SECRET) for stable " + "sessions." + ) + return secrets.token_bytes(32) + # Try base64, then hex, then fall back to the raw UTF-8 bytes. + for decoder in (base64.b64decode, bytes.fromhex): + try: + decoded = decoder(raw) + if len(decoded) >= 16: + return decoded + except (ValueError, TypeError): + pass + return raw.encode("utf-8") + + +def register(ctx) -> None: + """Plugin entry — registers BasicAuthProvider when credentials exist. + + Loopback / ``--insecure`` operators and anyone using the OAuth + provider leave ``dashboard.basic_auth`` unset, so this plugin is a + no-op for them. When username + (password or password_hash) are + configured, it registers a password provider that the login page + renders as a credential form. + """ + global LAST_SKIP_REASON + LAST_SKIP_REASON = "" + + section = _load_config_basic_auth_section() + username = _resolve( + "HERMES_DASHBOARD_BASIC_AUTH_USERNAME", section, "username" + ) + password_hash = _resolve( + "HERMES_DASHBOARD_BASIC_AUTH_PASSWORD_HASH", section, "password_hash" + ) + plaintext = _resolve( + "HERMES_DASHBOARD_BASIC_AUTH_PASSWORD", section, "password" + ) + ttl_raw = _resolve( + "HERMES_DASHBOARD_BASIC_AUTH_TTL_SECONDS", section, "session_ttl_seconds" + ) + + if not username: + LAST_SKIP_REASON = ( + "dashboard.basic_auth.username is not set (and " + "HERMES_DASHBOARD_BASIC_AUTH_USERNAME is empty). Set a username " + "and a password (or password_hash) under dashboard.basic_auth in " + "config.yaml to enable username/password dashboard login, or use " + "the OAuth provider, or pass --insecure to skip the auth gate." + ) + logger.debug("dashboard-auth-basic: %s", LAST_SKIP_REASON) + return + + if not password_hash and not plaintext: + LAST_SKIP_REASON = ( + "dashboard.basic_auth.username is set but neither password_hash " + "nor password is configured. Provide one of them (password_hash " + "is preferred — compute it with " + "plugins.dashboard_auth.basic.hash_password)." + ) + logger.warning("dashboard-auth-basic: %s", LAST_SKIP_REASON) + return + + # Precedence (env-wins convention): a password supplied via the + # HERMES_DASHBOARD_BASIC_AUTH_PASSWORD env var overrides a config.yaml + # password_hash, so an operator can rotate the password by setting an + # env var without editing config. A password_hash (precomputed) wins + # over a config-only plaintext password at the same tier — it's the + # preferred at-rest form. Concretely: + # * env password set → hash it (overrides any config hash) + # * else config password_hash set → use it + # * else config plaintext password → hash it in-memory + plaintext_from_env = os.environ.get( + "HERMES_DASHBOARD_BASIC_AUTH_PASSWORD", "" + ).strip() + if plaintext_from_env: + password_hash = hash_password(plaintext_from_env) + logger.info( + "dashboard-auth-basic: hashed env-supplied password in-memory " + "(overrides any config password_hash)." + ) + elif not password_hash: + # config-only plaintext password. + password_hash = hash_password(plaintext) + logger.info( + "dashboard-auth-basic: hashed plaintext password in-memory. " + "For production, precompute dashboard.basic_auth.password_hash " + "and remove the plaintext password from config." + ) + + secret = _resolve_secret(section) + + try: + ttl = int(ttl_raw) if ttl_raw else _DEFAULT_TTL_SECONDS + except ValueError: + ttl = _DEFAULT_TTL_SECONDS + + try: + provider = BasicAuthProvider( + username=username, + password_hash=password_hash, + secret=secret, + ttl_seconds=ttl, + ) + except ValueError as exc: + LAST_SKIP_REASON = f"BasicAuthProvider construction failed: {exc}" + logger.warning("dashboard-auth-basic: %s", LAST_SKIP_REASON) + return + + ctx.register_dashboard_auth_provider(provider) + logger.info( + "dashboard-auth-basic: registered password provider (username=%s)", + username, + ) diff --git a/plugins/dashboard_auth/basic/plugin.yaml b/plugins/dashboard_auth/basic/plugin.yaml new file mode 100644 index 0000000000000..586688f71b259 --- /dev/null +++ b/plugins/dashboard_auth/basic/plugin.yaml @@ -0,0 +1,7 @@ +name: basic +version: 1.0.0 +description: "Dashboard auth provider — username/password (no OAuth IDP). A self-hosted 'just put a password on my dashboard' provider. Activates when dashboard.basic_auth.username plus a password (or password_hash) are configured via config.yaml (canonical surface) or the HERMES_DASHBOARD_BASIC_AUTH_* env vars. Sessions are stateless HMAC-signed tokens minted by the provider; password hashing uses stdlib scrypt (no third-party dependency). Set dashboard.basic_auth.secret for restart-surviving / multi-worker sessions." +author: NousResearch +kind: backend +requires_env: + - HERMES_DASHBOARD_BASIC_AUTH_USERNAME From 71740f62396f5b6d9887cb471eb5d911feaf181f Mon Sep 17 00:00:00 2001 From: Ben Date: Thu, 4 Jun 2026 17:26:46 +1000 Subject: [PATCH 3/4] test(dashboard-auth): cover password login route, provider, and plugin - test_dashboard_auth_password_login.py: drives /auth/password-login end-to-end through the REAL gated_auth_middleware (login -> session cookie -> authenticated /api/auth/me -> transparent refresh via the RT cookie), plus protocol-extension checks, the generic-401/404 oracle properties, the rate limiter, and login-page rendering (form+script when supports_password, script-free otherwise, both for mixed providers). Reuses the existing StubAuthProvider harness convention. - test_basic_provider.py: scrypt hash/verify, login mint, kind-claim enforcement (access != refresh), cross-secret rejection, and the register() config/env precedence + skip reasons. Mutation-tested: dropping the kind-claim check in verify_session makes test_access_token_not_accepted_as_refresh fail, confirming the test isn't theater. --- .../test_dashboard_auth_password_login.py | 448 ++++++++++++++++++ .../dashboard_auth/test_basic_provider.py | 246 ++++++++++ 2 files changed, 694 insertions(+) create mode 100644 tests/hermes_cli/test_dashboard_auth_password_login.py create mode 100644 tests/plugins/dashboard_auth/test_basic_provider.py diff --git a/tests/hermes_cli/test_dashboard_auth_password_login.py b/tests/hermes_cli/test_dashboard_auth_password_login.py new file mode 100644 index 0000000000000..a863ede5b1738 --- /dev/null +++ b/tests/hermes_cli/test_dashboard_auth_password_login.py @@ -0,0 +1,448 @@ +"""Tests for the password (non-redirect) dashboard-auth login flow. + +Covers the protocol extension (``supports_password`` + +``complete_password_login``), the ``/auth/password-login`` route end-to-end +through the REAL ``gated_auth_middleware`` (session-cookie mint → +authenticated request → transparent refresh), the login-page credential +form rendering, and the route's rate limiter. + +The E2E harness mirrors ``test_dashboard_auth_401_reauth.py``: register a +provider, flip ``app.state.auth_required = True``, drive a ``TestClient``. +""" + +from __future__ import annotations + +import time + +import pytest + +# These tests mutate ``web_server.app.state.auth_required`` at module level, +# so they share the dashboard-auth app-state xdist group to avoid racing +# other gate tests. +pytestmark = pytest.mark.xdist_group("dashboard_auth_app_state") + +from fastapi.testclient import TestClient + +from hermes_cli import web_server +from hermes_cli.dashboard_auth import ( + DashboardAuthProvider, + InvalidCredentialsError, + ProviderError, + Session, + assert_protocol_compliance, + clear_providers, + register_provider, +) +from hermes_cli.dashboard_auth.cookies import SESSION_AT_COOKIE, SESSION_RT_COOKIE +from hermes_cli.dashboard_auth.login_page import render_login_html +from hermes_cli.dashboard_auth.routes import _reset_password_rate_limit +from tests.hermes_cli.conftest_dashboard_auth import StubAuthProvider + + +# --------------------------------------------------------------------------- +# Test password provider — minimal, in-memory, signed tokens. +# --------------------------------------------------------------------------- + + +def _sign(secret: bytes, sub: str, kind: str, ttl: int) -> str: + import base64 + import hashlib + import hmac + import json + + raw = json.dumps( + {"sub": sub, "kind": kind, "exp": int(time.time()) + ttl}, + separators=(",", ":"), + ).encode() + sig = hmac.new(secret, raw, hashlib.sha256).digest() + return base64.urlsafe_b64encode(raw + sig).decode() + + +def _unsign(secret: bytes, token: str): + import base64 + import hashlib + import hmac + import json + + try: + blob = base64.urlsafe_b64decode(token.encode()) + raw, sig = blob[:-32], blob[-32:] + if not hmac.compare_digest( + sig, hmac.new(secret, raw, hashlib.sha256).digest() + ): + return None + return json.loads(raw) + except Exception: + return None + + +class PasswordProvider(DashboardAuthProvider): + """In-test username/password provider (admin / hunter2).""" + + name = "testpw" + display_name = "Test Password" + supports_password = True + + def __init__(self, *, ttl: int = 3600, secret: bytes = b"test-secret-1234567890"): + self._ttl = ttl + self._secret = secret + self.unreachable = False # flip to simulate a ProviderError + + def start_login(self, *, redirect_uri: str): + raise NotImplementedError + + def complete_login(self, **kwargs): + raise NotImplementedError + + def complete_password_login(self, *, username: str, password: str) -> Session: + if self.unreachable: + raise ProviderError("backing store down") + if username != "admin" or password != "hunter2": + raise InvalidCredentialsError("bad creds") + exp = int(time.time()) + self._ttl + return Session( + user_id="admin", + email="", + display_name="admin", + org_id="", + provider=self.name, + expires_at=exp, + access_token=_sign(self._secret, "admin", "access", self._ttl), + refresh_token=_sign(self._secret, "admin", "refresh", 30 * 86400), + ) + + def verify_session(self, *, access_token: str): + p = _unsign(self._secret, access_token) + if not p or p.get("kind") != "access" or p["exp"] <= int(time.time()): + return None + return Session( + user_id=p["sub"], email="", display_name=p["sub"], org_id="", + provider=self.name, expires_at=p["exp"], + access_token=access_token, refresh_token="", + ) + + def refresh_session(self, *, refresh_token: str) -> Session: + from hermes_cli.dashboard_auth import RefreshExpiredError + + p = _unsign(self._secret, refresh_token) + if not p or p.get("kind") != "refresh" or p["exp"] <= int(time.time()): + raise RefreshExpiredError("dead rt") + exp = int(time.time()) + self._ttl + return Session( + user_id=p["sub"], email="", display_name=p["sub"], org_id="", + provider=self.name, expires_at=exp, + access_token=_sign(self._secret, p["sub"], "access", self._ttl), + refresh_token=_sign(self._secret, p["sub"], "refresh", 30 * 86400), + ) + + def revoke_session(self, *, refresh_token: str) -> None: + return None + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def pw_provider(): + return PasswordProvider() + + +@pytest.fixture +def gated_app(pw_provider): + clear_providers() + register_provider(pw_provider) + _reset_password_rate_limit() + prev_host = getattr(web_server.app.state, "bound_host", None) + prev_port = getattr(web_server.app.state, "bound_port", None) + prev_required = getattr(web_server.app.state, "auth_required", None) + web_server.app.state.bound_host = "fly-app.fly.dev" + web_server.app.state.bound_port = 443 + web_server.app.state.auth_required = True + client = TestClient(web_server.app, base_url="https://fly-app.fly.dev") + yield client + clear_providers() + _reset_password_rate_limit() + web_server.app.state.bound_host = prev_host + web_server.app.state.bound_port = prev_port + web_server.app.state.auth_required = prev_required + + +# --------------------------------------------------------------------------- +# Protocol extension +# --------------------------------------------------------------------------- + + +class TestProtocolExtension: + def test_password_provider_is_protocol_compliant(self): + assert assert_protocol_compliance(PasswordProvider) is None + + def test_default_supports_password_is_false(self): + # OAuth providers (the Stub) inherit the False default. + assert StubAuthProvider.supports_password is False + + def test_default_complete_password_login_raises_not_implemented(self): + # A provider that doesn't override the method (the Stub) raises, + # rather than silently accepting any credentials. + with pytest.raises(NotImplementedError): + StubAuthProvider().complete_password_login( + username="x", password="y" + ) + + +# --------------------------------------------------------------------------- +# /api/auth/providers exposes the supports_password flag +# --------------------------------------------------------------------------- + + +class TestProviderListFlag: + def test_providers_endpoint_reports_supports_password(self, gated_app): + resp = gated_app.get("/api/auth/providers") + assert resp.status_code == 200 + prov = {p["name"]: p for p in resp.json()["providers"]} + assert prov["testpw"]["supports_password"] is True + + def test_oauth_provider_reports_false(self): + clear_providers() + register_provider(StubAuthProvider()) + prev = getattr(web_server.app.state, "auth_required", None) + web_server.app.state.auth_required = True + try: + client = TestClient( + web_server.app, base_url="https://fly-app.fly.dev" + ) + resp = client.get("/api/auth/providers") + prov = {p["name"]: p for p in resp.json()["providers"]} + assert prov["stub"]["supports_password"] is False + finally: + clear_providers() + web_server.app.state.auth_required = prev + + +# --------------------------------------------------------------------------- +# /auth/password-login — end-to-end through the real middleware +# --------------------------------------------------------------------------- + + +class TestPasswordLoginRoute: + def test_valid_credentials_set_session_cookies_and_return_next( + self, gated_app + ): + resp = gated_app.post( + "/auth/password-login", + json={ + "provider": "testpw", + "username": "admin", + "password": "hunter2", + "next": "/sessions", + }, + ) + assert resp.status_code == 200 + assert resp.json() == {"ok": True, "next": "/sessions"} + set_cookie = resp.headers.get("set-cookie", "") + # HTTPS request → __Host- prefixed access-token cookie is set. + assert SESSION_AT_COOKIE in set_cookie + assert SESSION_RT_COOKIE in set_cookie + + def test_session_cookie_then_grants_authenticated_access(self, gated_app): + # Log in, then hit an auth-required endpoint with the cookie jar + # the TestClient retains — proving the minted session is accepted + # by the real gated_auth_middleware. + login = gated_app.post( + "/auth/password-login", + json={"provider": "testpw", "username": "admin", "password": "hunter2"}, + ) + assert login.status_code == 200 + me = gated_app.get("/api/auth/me") + assert me.status_code == 200 + assert me.json()["user_id"] == "admin" + assert me.json()["provider"] == "testpw" + + def test_wrong_password_returns_generic_401(self, gated_app): + resp = gated_app.post( + "/auth/password-login", + json={"provider": "testpw", "username": "admin", "password": "WRONG"}, + ) + assert resp.status_code == 401 + # Generic detail — no user-vs-password distinction. + assert resp.json()["detail"] == "Invalid credentials" + assert "set-cookie" not in {k.lower() for k in resp.headers} + + def test_unknown_user_returns_same_generic_401(self, gated_app): + resp = gated_app.post( + "/auth/password-login", + json={"provider": "testpw", "username": "ghost", "password": "hunter2"}, + ) + assert resp.status_code == 401 + assert resp.json()["detail"] == "Invalid credentials" + + def test_unknown_provider_returns_404(self, gated_app): + resp = gated_app.post( + "/auth/password-login", + json={"provider": "nope", "username": "admin", "password": "hunter2"}, + ) + assert resp.status_code == 404 + + def test_oauth_provider_rejects_password_login_with_404(self): + # An OAuth-only provider (supports_password False) must not be + # reachable via the password route — same 404 as unknown, so the + # endpoint isn't a provider-capability oracle. + clear_providers() + register_provider(StubAuthProvider()) + _reset_password_rate_limit() + prev = getattr(web_server.app.state, "auth_required", None) + web_server.app.state.auth_required = True + try: + client = TestClient( + web_server.app, base_url="https://fly-app.fly.dev" + ) + resp = client.post( + "/auth/password-login", + json={"provider": "stub", "username": "x", "password": "y"}, + ) + assert resp.status_code == 404 + finally: + clear_providers() + _reset_password_rate_limit() + web_server.app.state.auth_required = prev + + def test_provider_unreachable_returns_503(self, gated_app, pw_provider): + pw_provider.unreachable = True + resp = gated_app.post( + "/auth/password-login", + json={"provider": "testpw", "username": "admin", "password": "hunter2"}, + ) + assert resp.status_code == 503 + + def test_open_redirect_next_is_dropped(self, gated_app): + resp = gated_app.post( + "/auth/password-login", + json={ + "provider": "testpw", + "username": "admin", + "password": "hunter2", + "next": "https://evil.example/phish", + }, + ) + assert resp.status_code == 200 + # Malicious absolute URL dropped → lands at root. + assert resp.json()["next"] == "/" + + def test_route_is_public_unauthenticated(self, gated_app): + # The login route itself must be reachable without a session — + # otherwise you could never log in. + resp = gated_app.post( + "/auth/password-login", + json={"provider": "testpw", "username": "admin", "password": "hunter2"}, + ) + assert resp.status_code == 200 + + +# --------------------------------------------------------------------------- +# Transparent refresh — expired access token, live refresh token +# --------------------------------------------------------------------------- + + +class TestPasswordSessionRefresh: + def test_expired_access_token_refreshes_via_rt_cookie(self): + # TTL=0 → access token born expired; the RT cookie should drive a + # transparent refresh on the next request (the same machinery the + # OAuth provider uses). + clear_providers() + provider = PasswordProvider(ttl=0) + register_provider(provider) + _reset_password_rate_limit() + prev = getattr(web_server.app.state, "auth_required", None) + web_server.app.state.auth_required = True + try: + client = TestClient( + web_server.app, base_url="https://fly-app.fly.dev" + ) + login = client.post( + "/auth/password-login", + json={"provider": "testpw", "username": "admin", "password": "hunter2"}, + ) + assert login.status_code == 200 + # Give the provider a live TTL so the refreshed token verifies. + provider._ttl = 3600 + me = client.get("/api/auth/me") + assert me.status_code == 200 + assert me.json()["user_id"] == "admin" + finally: + clear_providers() + _reset_password_rate_limit() + web_server.app.state.auth_required = prev + + +# --------------------------------------------------------------------------- +# Rate limiter +# --------------------------------------------------------------------------- + + +class TestRateLimit: + def test_repeated_failures_eventually_429(self, gated_app): + # The limiter caps attempts per IP per window (default 10). After + # the budget is exhausted, even a VALID credential gets 429. + last = None + for _ in range(15): + last = gated_app.post( + "/auth/password-login", + json={"provider": "testpw", "username": "admin", "password": "WRONG"}, + ) + assert last.status_code == 429 + # Even correct creds are throttled once the window is saturated. + good = gated_app.post( + "/auth/password-login", + json={"provider": "testpw", "username": "admin", "password": "hunter2"}, + ) + assert good.status_code == 429 + + +# --------------------------------------------------------------------------- +# Login page rendering +# --------------------------------------------------------------------------- + + +class TestLoginPageRender: + def test_password_provider_renders_credential_form_and_script(self): + clear_providers() + register_provider(PasswordProvider()) + try: + html = render_login_html(next_path="/sessions") + assert '
" in html + assert "/auth/password-login" in html + finally: + clear_providers() + + def test_oauth_only_page_stays_script_free(self): + clear_providers() + register_provider(StubAuthProvider()) + try: + html = render_login_html() + assert "provider-btn" in html + assert "