From 48f8d0be184ee895ec9b2f3820168ae5daf3763d Mon Sep 17 00:00:00 2001 From: Mike Teehan Date: Wed, 29 Jul 2026 02:48:53 +0000 Subject: [PATCH] Auth: rate limiter, WebAuthn domain awareness, misc fixes - Rate limiter tracks failures only; success resets counter - Record failures/successes after password verification, not before - WebAuthn rp_id/origin resolved dynamically from request domain - Management domains auto-discovered from nginx backend config - All WebAuthn operations validate domain against management list - Add GET /api/auth/webauthn/capable endpoint for frontend checks - Frontend checkWebAuthnCapable() function for domain-gated UI - Timing side-channel fix: pre-compute dummy hash at module load - Builtin admin seeded with random password (logged at WARNING) - Logout handler returns consistent response shape --- daemon/handlers/auth.py | 102 +++++++++++++++++++++---- daemon/iface.py | 1 + daemon/server.py | 12 ++- lib/auth.py | 53 ++++++++----- lib/auth_users.py | 11 ++- lib/db.py | 12 ++- lib/nginx.py | 28 +++++++ lib/webauthn.py | 87 ++++++++++++++++----- tests/test_auth.py | 35 +++++++-- webui/api/auth.py | 45 +++++++++++ webui/static/hoover/components/auth.js | 14 ++++ 11 files changed, 335 insertions(+), 65 deletions(-) diff --git a/daemon/handlers/auth.py b/daemon/handlers/auth.py index 25e73f4..6bdf796 100644 --- a/daemon/handlers/auth.py +++ b/daemon/handlers/auth.py @@ -14,6 +14,7 @@ from daemon.iface import ( DELETE_AUTH_WEBAUTHN_CREDENTIAL, GET_AUTH_SESSION, GET_AUTH_USERS, + GET_AUTH_WEBAUTHN_CAPABLE, GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS, GET_AUTH_WEBAUTHN_CREDENTIALS, POST_AUTH_LOGIN, @@ -36,6 +37,8 @@ from lib.auth import ( clear_active_refresh_token, generate_tokens, get_access_ttl, + record_login_failure, + record_login_success, validate_token, ) from lib.auth_users import ( @@ -51,12 +54,16 @@ from lib.auth_users import ( from lib.webauthn import ( create_authentication_options, create_registration_options, - get_all_credential_counts, + get_management_domains, + get_rp_name, list_credentials, remove_credential, verify_authentication, verify_registration, ) +from lib.webauthn import ( + is_enabled as webauthn_is_enabled, +) logger = logging.getLogger(__name__) @@ -96,8 +103,11 @@ def auth_login(_request: Any, body: Any) -> dict[str, Any]: user = verify_user_password(username, password) if user is None: + record_login_failure(username, client_ip) raise ValueError("Invalid credentials") + record_login_success(username, client_ip) + permissions = user["permissions"] tokens = generate_tokens(username, permissions) @@ -125,7 +135,7 @@ def auth_logout(request: Any, body: Any) -> dict[str, Any]: Success response. """ if not body: - return {"ok": True} + return {} jti = body.get("jti") if jti: @@ -135,7 +145,7 @@ def auth_logout(request: Any, body: Any) -> dict[str, Any]: if username: blacklist_active_refresh_token(username) - return {"ok": True} + return {} @registry.register(POST_AUTH_REFRESH) @@ -379,25 +389,73 @@ def auth_delete_user(_request: Any, body: Any) -> dict[str, Any]: # --------------------------------------------------------------------------- +def _check_webauthn_domain(body: Any) -> tuple[str, str]: + """Validate that the request domain is eligible for WebAuthn. + + Returns (origin, rp_id) if valid. Raises ValueError otherwise. + """ + if not body or not isinstance(body, dict): + raise ValueError("Request body is required") + + if not webauthn_is_enabled(): + raise ValueError("WebAuthn is disabled") + + rp_id = body.get("webauthn_rp_id") + origin = body.get("webauthn_origin") + if not rp_id or not origin: + raise ValueError("Missing WebAuthn domain configuration") + + if rp_id not in get_management_domains(): + raise ValueError("WebAuthn is not available on this domain") + + return origin, rp_id + + +@registry.register(GET_AUTH_WEBAUTHN_CAPABLE) +def webauthn_capable(_request: Any, body: Any) -> dict[str, Any]: + """Check if WebAuthn is available for the current request domain.""" + rp_id = body.get("webauthn_rp_id") if body else None + origin = body.get("webauthn_origin") if body else None + + if not webauthn_is_enabled(): + return {"enabled": False, "reason": "WebAuthn is disabled in config"} + + if not rp_id or not origin: + return {"enabled": False, "reason": "Domain information unavailable"} + + mgmt_domains = get_management_domains() + if rp_id not in mgmt_domains: + return {"enabled": False, "reason": "Not a management domain"} + + return { + "enabled": True, + "rp_id": rp_id, + "rp_name": get_rp_name(), + "origin": origin, + } + + @registry.register(POST_AUTH_WEBAUTHN_REGISTER_BEGIN) def webauthn_register_begin(_request: Any, body: Any) -> dict[str, Any]: """Begin WebAuthn registration — return options for ``credentials.create()``.""" - if not body or not isinstance(body, dict): - raise ValueError("Request body is required") + origin, rp_id = _check_webauthn_domain(body) username = body.get("username") if not username: raise ValueError("username is required") - options = create_registration_options(username) + options = create_registration_options( + username, + origin=origin, + rp_id=rp_id, + ) return options @registry.register(POST_AUTH_WEBAUTHN_REGISTER_FINISH) def webauthn_register_finish(_request: Any, body: Any) -> dict[str, Any]: """Finish WebAuthn registration — verify credential and persist.""" - if not body or not isinstance(body, dict): - raise ValueError("Request body is required") + origin, rp_id = _check_webauthn_domain(body) username = body.get("username") credential_response = body.get("credential_response") @@ -410,7 +468,12 @@ def webauthn_register_finish(_request: Any, body: Any) -> dict[str, Any]: ) cred = verify_registration( - username, credential_response, registration_options, credential_name + username, + credential_response, + registration_options, + credential_name, + origin=origin, + rp_id=rp_id, ) return {"ok": True, "credential": cred} @@ -422,14 +485,16 @@ def webauthn_authenticate_begin(_request: Any, body: Any) -> dict[str, Any]: Public endpoint — no JWT required. Returns ``{"noWebAuthn": true}`` if the user has no registered credentials (so the frontend can fall back to password). """ - if not body or not isinstance(body, dict): - raise ValueError("Request body is required") + _, rp_id = _check_webauthn_domain(body) username = body.get("username") if not username: raise ValueError("username is required") - options = create_authentication_options(username) + options = create_authentication_options( + username, + rp_id=rp_id, + ) if options is None: return {"no_webauthn": True} return options @@ -441,8 +506,7 @@ def webauthn_authenticate_finish(_request: Any, body: Any) -> dict[str, Any]: Public endpoint — no JWT required. """ - if not body or not isinstance(body, dict): - raise ValueError("Request body is required") + origin, rp_id = _check_webauthn_domain(body) username = body.get("username") assertion_response = body.get("assertion_response") @@ -455,7 +519,13 @@ def webauthn_authenticate_finish(_request: Any, body: Any) -> dict[str, Any]: if not check_webauthn_rate(username, client_ip): raise ValueError("Too many WebAuthn attempts. Please try again later.") - verify_authentication(username, assertion_response, auth_options) + verify_authentication( + username, + assertion_response, + auth_options, + origin=origin, + rp_id=rp_id, + ) user = get_user(username) if user is None: @@ -512,4 +582,6 @@ def webauthn_remove_credential(request: Any, body: Any) -> dict[str, Any]: @registry.register(GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS) def webauthn_credential_counts(_request: Any, body: Any) -> dict[str, int]: """Return credential counts for all users (admin endpoint).""" + from lib.webauthn import get_all_credential_counts + return get_all_credential_counts() diff --git a/daemon/iface.py b/daemon/iface.py index c614c8d..940916c 100644 --- a/daemon/iface.py +++ b/daemon/iface.py @@ -199,6 +199,7 @@ GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS: Endpoint = _ep( DELETE_AUTH_WEBAUTHN_CREDENTIAL: Endpoint = _ep( "DELETE", "/auth/webauthn/creds/" ) +GET_AUTH_WEBAUTHN_CAPABLE: Endpoint = _ep("GET", "/auth/webauthn/capable") # ---- Server infra (not going through client) ---- GET_HEALTH: Endpoint = _ep("GET", "/health") diff --git a/daemon/server.py b/daemon/server.py index f64239f..b62a833 100644 --- a/daemon/server.py +++ b/daemon/server.py @@ -397,7 +397,9 @@ async def _handle_ws(request: web.Request) -> web.Response: ) session_header = request.headers.get("X-Session-Id") - payload = validate_token(token_param, token_type="access", session_id=session_header) + payload = validate_token( + token_param, token_type="access", session_id=session_header + ) if payload is None: return web.json_response({"ok": False, "error": "unauthorized"}, status=401) @@ -556,6 +558,14 @@ def main() -> None: setup_logging() + # Startup checks + try: + from lib import webauthn as lib_webauthn + + lib_webauthn.check_webauthn_config() + except Exception: + pass # ignore if webauthn module import failed + _register_routes() app = create_app() diff --git a/lib/auth.py b/lib/auth.py index 166ba9c..16999f5 100644 --- a/lib/auth.py +++ b/lib/auth.py @@ -346,19 +346,19 @@ def blacklist_expired() -> None: class RateLimiter: - """Simple sliding-window rate limiter for login attempts. + """Sliding-window rate limiter that tracks successes and failures separately. - Maintains per-key attempt timestamps and rejects requests that exceed - the allowed count within the window. + Failures are counted against the limit. A successful operation resets + the failure counter for that key. """ def __init__(self, max_attempts: int = 5, window_seconds: int = 300) -> None: self.max_attempts = max_attempts self.window = window_seconds - self.attempts: dict[str, list[float]] = {} + self.failures: dict[str, list[float]] = {} def is_allowed(self, key: str) -> bool: - """Check if a request from *key* is allowed. + """Check if a request from *key* is allowed (does NOT record the attempt). Args: key: Identifier for the rate limit bucket (e.g., username or IP). @@ -368,25 +368,29 @@ class RateLimiter: """ now = time.time() cutoff = now - self.window - timestamps = self.attempts.get(key, []) + timestamps = self.failures.get(key, []) # Clean old entries - self.attempts[key] = [t for t in timestamps if t > cutoff] + self.failures[key] = [t for t in timestamps if t > cutoff] - if len(self.attempts[key]) >= self.max_attempts: - return False + return len(self.failures[key]) < self.max_attempts - self.attempts[key].append(now) - return True + def record_failure(self, key: str) -> None: + """Record a failed attempt for *key*.""" + self.failures.setdefault(key, []).append(time.time()) + + def record_success(self, key: str) -> None: + """Reset the failure counter for *key* on a successful operation.""" + self.failures.pop(key, None) def cleanup(self) -> None: """Remove expired entries from all buckets.""" now = time.time() cutoff = now - self.window - for key in list(self.attempts): - self.attempts[key] = [t for t in self.attempts[key] if t > cutoff] - if not self.attempts[key]: - del self.attempts[key] + for key in list(self.failures): + self.failures[key] = [t for t in self.failures[key] if t > cutoff] + if not self.failures[key]: + del self.failures[key] # Global rate limiters @@ -397,8 +401,9 @@ _webauthn_limiter = RateLimiter(max_attempts=5, window_seconds=600) def check_login_rate(username: str, client_ip: str | None = None) -> bool: """Check if login is rate-limited for the given username. - Uses dual-key tracking: always records by IP (catches enumeration attacks), - additionally records by username (catches legitimate users who forget password). + Checks failure counts for both IP and username buckets without + recording anything. Callers must invoke record_login_failure() or + record_login_success() after the password verification step. Args: username: The login attempt username. @@ -412,6 +417,20 @@ def check_login_rate(username: str, client_ip: str | None = None) -> bool: return _login_limiter.is_allowed(username) +def record_login_failure(username: str, client_ip: str | None = None) -> None: + """Record a failed login attempt.""" + if client_ip: + _login_limiter.record_failure(client_ip) + _login_limiter.record_failure(username) + + +def record_login_success(username: str, client_ip: str | None = None) -> None: + """Record a successful login (resets failure counter).""" + if client_ip: + _login_limiter.record_success(client_ip) + _login_limiter.record_success(username) + + def check_webauthn_rate(username: str, client_ip: str | None = None) -> bool: """Check if WebAuthn authentication is rate-limited for the given username. diff --git a/lib/auth_users.py b/lib/auth_users.py index 6702bde..d69e312 100644 --- a/lib/auth_users.py +++ b/lib/auth_users.py @@ -31,6 +31,10 @@ from lib.password import hash_password, needs_rehash, verify_password logger = logging.getLogger(__name__) +# Pre-computed dummy hash for constant-time verification on nonexistent users. +# Generated once at module load to avoid timing leaks from per-call hash generation. +_DUMMY_HASH = hash_password(secrets.token_hex(32)) + # Builtin admin — hardcoded, full access, cannot be modified/deleted BUILTIN_ADMIN_USERNAME = "admin" @@ -113,9 +117,10 @@ def verify_user_password(username: str, password: str) -> dict[str, Any] | None: """ user = find_user(username) if user is None: - # Run a dummy Argon2id verification to prevent timing-based user enumeration. - # The timing for both paths is now equivalent. - verify_password(password, hash_password(secrets.token_hex(32))) + # Run a dummy Argon2id verification against a pre-computed hash to + # prevent timing-based user enumeration. Uses module-level hash + # so both paths take ~1 verify call (~200ms) instead of ~400ms. + verify_password(password, _DUMMY_HASH) return None if not verify_password(password, user["password_hash"]): return None diff --git a/lib/db.py b/lib/db.py index 62dd667..791f6db 100644 --- a/lib/db.py +++ b/lib/db.py @@ -292,8 +292,10 @@ def _seed_builtin_admin(db: Database) -> None: if rows: return - # Create with a placeholder password that should be changed - placeholder_hash = hash_password("CHANGEME") + # Generate a random password — this fallback should only fire if + # bootstrap_auth.py was skipped. Log the password prominently. + random_password = secrets.token_urlsafe(24) + placeholder_hash = hash_password(random_password) jwt_secret = secrets.token_urlsafe(32) with db.in_transaction() as tx: @@ -303,7 +305,11 @@ def _seed_builtin_admin(db: Database) -> None: for subsystem in ALL_SUBSYSTEMS: tx.run(Q_UPSERT_PERMISSION, (BUILTIN_ADMIN_USERNAME, subsystem, "rw")) - logger.info("Builtin admin user created with full access") + logger.warning( + "Builtin admin user created. THIS IS A FALLBACK — bootstrap_auth.py " + "should have run during install. Admin password: %s", + random_password, + ) def reset_db_for_test() -> None: diff --git a/lib/nginx.py b/lib/nginx.py index e93b1f0..48c0e03 100644 --- a/lib/nginx.py +++ b/lib/nginx.py @@ -225,6 +225,33 @@ def get_domains() -> list[dict[str, Any]]: return result +def get_management_domains() -> list[str]: + """Return domain names that serve the management UI. + + Checks both backend-referenced paths (for migrated configs) and + inline paths (for legacy configs pending migration). + + Returns: + List of domain name strings. + """ + cfg = get_config() + backends = cfg.get("backends", {}) + domains: list[str] = [] + for name, dom in cfg.get("domains", {}).items(): + # Check inline paths (pre-migration format) + inline_paths = dom.get("paths", {}) + if any(p.get("is_management") for p in inline_paths.values()): + domains.append(name) + continue + # Check backend-referenced paths + backend_name = dom.get("backend", "") + if backend_name and backend_name in backends: + paths = backends[backend_name].get("paths", {}) + if any(p.get("is_management") for p in paths.values()): + domains.append(name) + return domains + + # ------------------------------------------------------------------ # Domain CRUD # ------------------------------------------------------------------ @@ -577,6 +604,7 @@ __all__ = [ "generate_server_conf", "get_config", "get_domains", + "get_management_domains", "remove_domain", "save_config", "test_config", diff --git a/lib/webauthn.py b/lib/webauthn.py index 228060b..aec6141 100644 --- a/lib/webauthn.py +++ b/lib/webauthn.py @@ -73,9 +73,10 @@ def _get_webauthn_config() -> dict[str, Any]: return raw.get("webauthn", {}) -def get_rp_id() -> str: - """Return the Relying Party ID from config.""" - return _get_webauthn_config().get("rp_id", "localhost") +def is_enabled() -> bool: + """Return whether WebAuthn is enabled in config.""" + cfg = _get_webauthn_config() + return cfg.get("enabled", True) def get_rp_name() -> str: @@ -83,9 +84,39 @@ def get_rp_name() -> str: return _get_webauthn_config().get("rp_name", "Vacuum Wall") -def get_origin() -> str: - """Return the WebAuthn origin from config.""" - return _get_webauthn_config().get("origin", "http://localhost") +def get_management_domains() -> list[str]: + """Return domain names eligible for WebAuthn. + + Delegates to lib.nginx.get_management_domains() to read the + live proxy config and discover which domains serve the management UI. + """ + from lib.nginx import get_management_domains as _resolve + + return _resolve() + + +def is_domain_valid(domain: str) -> bool: + """Check if *domain* is eligible for WebAuthn.""" + return domain in get_management_domains() + + +def check_webauthn_config() -> None: + """Validate WebAuthn config at startup.""" + enabled = is_enabled() + if not enabled: + logger.info("WebAuthn is disabled in config") + return + try: + domains = get_management_domains() + if not domains: + logger.warning( + "WebAuthn is enabled but no management domains are configured. " + "Add a domain with backend 'webui' to nginx config, or disable WebAuthn." + ) + else: + logger.info("WebAuthn enabled for domains: %s", domains) + except Exception as exc: + logger.error("Failed to resolve management domains for WebAuthn: %s", exc) # --------------------------------------------------------------------------- @@ -93,9 +124,18 @@ def get_origin() -> str: # --------------------------------------------------------------------------- -def create_registration_options(username: str) -> dict[str, Any]: +def create_registration_options( + username: str, + origin: str, + rp_id: str, +) -> dict[str, Any]: """Create WebAuthn registration options for a new credential. + Args: + username: The user registering the credential. + origin: WebAuthn origin (must match the request origin). + rp_id: Relying Party ID (must match the request domain). + Returns a dict serializable to JSON, matching the format expected by ``navigator.credentials.create()``. """ @@ -113,7 +153,7 @@ def create_registration_options(username: str) -> dict[str, Any]: user_id = user_id + b"\x00" * (8 - len(user_id)) options = generate_registration_options( - rp_id=get_rp_id(), + rp_id=rp_id, rp_name=get_rp_name(), user_name=username, user_display_name=username, @@ -136,6 +176,8 @@ def verify_registration( credential_response: dict[str, Any], registration_options: dict[str, Any], credential_name: str = "", + origin: str = "", + rp_id: str = "", ) -> dict[str, Any]: """Verify a registration response and persist the credential. @@ -144,12 +186,12 @@ def verify_registration( credential_response: Browser response from ``credentials.create()``. registration_options: The options dict from ``create_registration_options``. credential_name: Optional human-readable label. + origin: WebAuthn origin for verification. + rp_id: Relying Party ID for verification. Returns: Dict with ``id``, ``name``, ``transports``, ``sign_count``. """ - expected_origin = get_origin() - expected_rp_id = get_rp_id() challenge = b64u_decode(registration_options["challenge"]) # The library accepts the credential response as a JSON-serializable dict @@ -157,8 +199,8 @@ def verify_registration( col = verify_registration_response( credential=credential_response, expected_challenge=challenge, - expected_origin=expected_origin, - expected_rp_id=expected_rp_id, + expected_origin=origin, + expected_rp_id=rp_id, require_user_verification=False, ) @@ -218,9 +260,16 @@ def verify_registration( # --------------------------------------------------------------------------- -def create_authentication_options(username: str) -> dict[str, Any] | None: +def create_authentication_options( + username: str, + rp_id: str, +) -> dict[str, Any] | None: """Create authentication options for a user. + Args: + username: The user authenticating. + rp_id: Relying Party ID (must match the request domain). + Returns a dict serializable to JSON (for ``navigator.credentials.get()``), or ``None`` if the user has no registered credentials. """ @@ -235,7 +284,7 @@ def create_authentication_options(username: str) -> dict[str, Any] | None: ] options = generate_authentication_options( - rp_id=get_rp_id(), + rp_id=rp_id, allow_credentials=allow_credentials, user_verification=UserVerificationRequirement.PREFERRED, ) @@ -247,6 +296,8 @@ def verify_authentication( username: str, assertion_response: dict[str, Any], auth_options: dict[str, Any], + origin: str = "", + rp_id: str = "", ) -> bool: """Verify an authentication assertion. @@ -254,6 +305,8 @@ def verify_authentication( username: The user authenticating. assertion_response: Browser response from ``credentials.get()``. auth_options: The options dict from ``create_authentication_options``. + origin: WebAuthn origin for verification. + rp_id: Relying Party ID for verification. Returns: True on successful verification. @@ -261,8 +314,6 @@ def verify_authentication( Raises: ValueError: On verification failure. """ - expected_origin = get_origin() - expected_rp_id = get_rp_id() challenge = b64u_decode(auth_options["challenge"]) cred_id_str = assertion_response["id"] @@ -279,8 +330,8 @@ def verify_authentication( col = verify_authentication_response( credential=assertion_response, expected_challenge=challenge, - expected_origin=expected_origin, - expected_rp_id=expected_rp_id, + expected_origin=origin, + expected_rp_id=rp_id, credential_public_key=public_key, credential_current_sign_count=old_sign_count, require_user_verification=False, diff --git a/tests/test_auth.py b/tests/test_auth.py index d51ed12..8713e1a 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -429,11 +429,13 @@ class TestWebAuthnConfig: """Test WebAuthn configuration helpers.""" def test_get_rp_defaults(self) -> None: - from lib.webauthn import get_origin, get_rp_id, get_rp_name + from lib.webauthn import get_management_domains, get_rp_name, is_enabled - assert get_rp_id() == "localhost" + assert is_enabled() is True assert get_rp_name() == "Vacuum Wall" - assert get_origin() == "http://localhost" + # get_management_domains reads from nginx config + domains = get_management_domains() + assert isinstance(domains, list) @patch.dict( @@ -482,14 +484,18 @@ class TestWebAuthnRegistration: def test_create_registration_options_basic(self) -> None: from lib.webauthn import create_registration_options - options = create_registration_options("testuser") + options = create_registration_options( + "testuser", + origin="https://wall.example.com", + rp_id="wall.example.com", + ) assert isinstance(options, dict) assert "challenge" in options assert "rp" in options assert "user" in options assert "pubKeyCredParams" in options - assert options["rp"]["id"] == "localhost" + assert options["rp"]["id"] == "wall.example.com" assert options["user"]["name"] == "testuser" assert len(options["pubKeyCredParams"]) >= 1 @@ -524,6 +530,8 @@ class TestWebAuthnRegistration: }, {"challenge": b64u_encode(b"testchallenge123")}, "My Key", + origin="https://wall.example.com", + rp_id="wall.example.com", ) assert "id" in result @@ -542,7 +550,11 @@ class TestWebAuthnRegistration: from lib.webauthn import create_registration_options - options = create_registration_options("testuser") + options = create_registration_options( + "testuser", + origin="https://wall.example.com", + rp_id="wall.example.com", + ) # Should contain the existing credential in exclude list exclude_ids = [c["id"] for c in options.get("excludeCredentials", [])] @@ -590,7 +602,9 @@ class TestWebAuthnAuthentication: def test_create_auth_options_no_credentials(self) -> None: from lib.webauthn import create_authentication_options - result = create_authentication_options("nonexistentuser") + result = create_authentication_options( + "nonexistentuser", rp_id="wall.example.com" + ) assert result is None def test_create_auth_options_with_credentials(self) -> None: @@ -598,7 +612,8 @@ class TestWebAuthnAuthentication: from lib.webauthn import create_authentication_options - result = create_authentication_options("testuser") + result = create_authentication_options("testuser", rp_id="wall.example.com") + assert result is not None assert result is not None assert len(result["allowCredentials"]) == 1 @@ -618,6 +633,8 @@ class TestWebAuthnAuthentication: }, }, {"challenge": _FAKE_CHALLENGE}, + origin="https://wall.example.com", + rp_id="wall.example.com", ) def test_verify_authentication_success(self) -> None: @@ -643,6 +660,8 @@ class TestWebAuthnAuthentication: }, }, {"challenge": _FAKE_CHALLENGE, "sign_count": 0}, + origin="https://wall.example.com", + rp_id="wall.example.com", ) assert result is True diff --git a/webui/api/auth.py b/webui/api/auth.py index 167a949..a50f48a 100644 --- a/webui/api/auth.py +++ b/webui/api/auth.py @@ -15,6 +15,7 @@ from daemon.iface import ( DELETE_AUTH_WEBAUTHN_CREDENTIAL, GET_AUTH_SESSION, GET_AUTH_USERS, + GET_AUTH_WEBAUTHN_CAPABLE, GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS, GET_AUTH_WEBAUTHN_CREDENTIALS, POST_AUTH_LOGIN, @@ -219,6 +220,39 @@ def delete_user(username: str): # --------------------------------------------------------------------------- +def _resolve_webauthn_origin() -> tuple[str, str]: + """Extract WebAuthn origin and rp_id from the current request. + + Returns (origin, rp_id) derived from the actual request, falling back + to config values when the request metadata is unavailable. + """ + scheme = request.headers.get("X-Forwarded-Proto", request.scheme) + host = request.headers.get("X-Forwarded-Host", request.host.split(":")[0]) + origin = f"{scheme}://{host}" + # rp_id is the registered domain (strip port numbers) + rp_id = host.split(":")[0] + return origin, rp_id + + +@bp.route("/webauthn/capable", methods=["GET"]) +def webauthn_capable(): + """Check if WebAuthn is available on the current request domain. + + Endpoint: + GET /api/auth/webauthn/capable + Returns: + { "enabled": true/false, "rp_id": "...", "rp_name": "...", "origin": "..." } + or { "enabled": false, "reason": "..." } + """ + try: + origin, rp_id = _resolve_webauthn_origin() + body = {"webauthn_origin": origin, "webauthn_rp_id": rp_id} + return _ok(get(GET_AUTH_WEBAUTHN_CAPABLE, body)) + except Exception as exc: + logger.error("WebAuthn capable check failed: %s", exc) + return _error(str(exc), 500) + + @bp.route("/webauthn/register-begin", methods=["POST"]) def webauthn_register_begin(): """Begin WebAuthn registration. @@ -232,6 +266,9 @@ def webauthn_register_begin(): """ try: body = request.get_json(silent=True) or {} + origin, rp_id = _resolve_webauthn_origin() + body["webauthn_origin"] = origin + body["webauthn_rp_id"] = rp_id return _ok(post(POST_AUTH_WEBAUTHN_REGISTER_BEGIN, body)) except Exception as exc: logger.error("WebAuthn register begin failed: %s", exc) @@ -251,6 +288,9 @@ def webauthn_register_finish(): """ try: body = request.get_json(silent=True) or {} + origin, rp_id = _resolve_webauthn_origin() + body["webauthn_origin"] = origin + body["webauthn_rp_id"] = rp_id return _ok(post(POST_AUTH_WEBAUTHN_REGISTER_FINISH, body)) except Exception as exc: logger.error("WebAuthn register finish failed: %s", exc) @@ -271,6 +311,8 @@ def webauthn_authenticate_begin(): """ try: body = request.get_json(silent=True) or {} + _, rp_id = _resolve_webauthn_origin() + body["webauthn_rp_id"] = rp_id return _ok(post(POST_AUTH_WEBAUTHN_AUTHENTICATE_BEGIN, body)) except Exception as exc: logger.error("WebAuthn authenticate begin failed: %s", exc) @@ -291,6 +333,9 @@ def webauthn_authenticate_finish(): try: body = request.get_json(silent=True) or {} body["client_ip"] = request.headers.get("X-Real-IP") or request.remote_addr + origin, rp_id = _resolve_webauthn_origin() + body["webauthn_origin"] = origin + body["webauthn_rp_id"] = rp_id return _ok(post(POST_AUTH_WEBAUTHN_AUTHENTICATE_FINISH, body)) except Exception as exc: logger.error("WebAuthn authenticate finish failed: %s", exc) diff --git a/webui/static/hoover/components/auth.js b/webui/static/hoover/components/auth.js index dae6654..44eef3f 100644 --- a/webui/static/hoover/components/auth.js +++ b/webui/static/hoover/components/auth.js @@ -143,6 +143,20 @@ export function webauthnSupported() { return typeof window !== 'undefined' && !!window.PublicKeyCredential; } +/** + * Check if WebAuthn is enabled and available on the current domain. + * Calls GET /api/auth/webauthn/capable to query the server. + * + * @returns {Promise} { enabled, rp_id, rp_name, origin, reason? } + */ +export async function checkWebAuthnCapable() { + const result = await apiFetch('/api/auth/webauthn/capable'); + if (!result.ok) { + return { enabled: false, reason: 'Unable to check WebAuthn capability' }; + } + return result.data || { enabled: false, reason: 'Server returned no data' }; +} + /* ─── Base64url helpers ──────────────────────────────────────────────── */ /**