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
This commit is contained in:
2026-07-29 02:48:53 +00:00
parent 6d30f1387e
commit 48f8d0be18
11 changed files with 335 additions and 65 deletions
+87 -15
View File
@@ -14,6 +14,7 @@ from daemon.iface import (
DELETE_AUTH_WEBAUTHN_CREDENTIAL, DELETE_AUTH_WEBAUTHN_CREDENTIAL,
GET_AUTH_SESSION, GET_AUTH_SESSION,
GET_AUTH_USERS, GET_AUTH_USERS,
GET_AUTH_WEBAUTHN_CAPABLE,
GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS, GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS,
GET_AUTH_WEBAUTHN_CREDENTIALS, GET_AUTH_WEBAUTHN_CREDENTIALS,
POST_AUTH_LOGIN, POST_AUTH_LOGIN,
@@ -36,6 +37,8 @@ from lib.auth import (
clear_active_refresh_token, clear_active_refresh_token,
generate_tokens, generate_tokens,
get_access_ttl, get_access_ttl,
record_login_failure,
record_login_success,
validate_token, validate_token,
) )
from lib.auth_users import ( from lib.auth_users import (
@@ -51,12 +54,16 @@ from lib.auth_users import (
from lib.webauthn import ( from lib.webauthn import (
create_authentication_options, create_authentication_options,
create_registration_options, create_registration_options,
get_all_credential_counts, get_management_domains,
get_rp_name,
list_credentials, list_credentials,
remove_credential, remove_credential,
verify_authentication, verify_authentication,
verify_registration, verify_registration,
) )
from lib.webauthn import (
is_enabled as webauthn_is_enabled,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -96,8 +103,11 @@ def auth_login(_request: Any, body: Any) -> dict[str, Any]:
user = verify_user_password(username, password) user = verify_user_password(username, password)
if user is None: if user is None:
record_login_failure(username, client_ip)
raise ValueError("Invalid credentials") raise ValueError("Invalid credentials")
record_login_success(username, client_ip)
permissions = user["permissions"] permissions = user["permissions"]
tokens = generate_tokens(username, permissions) tokens = generate_tokens(username, permissions)
@@ -125,7 +135,7 @@ def auth_logout(request: Any, body: Any) -> dict[str, Any]:
Success response. Success response.
""" """
if not body: if not body:
return {"ok": True} return {}
jti = body.get("jti") jti = body.get("jti")
if jti: if jti:
@@ -135,7 +145,7 @@ def auth_logout(request: Any, body: Any) -> dict[str, Any]:
if username: if username:
blacklist_active_refresh_token(username) blacklist_active_refresh_token(username)
return {"ok": True} return {}
@registry.register(POST_AUTH_REFRESH) @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) @registry.register(POST_AUTH_WEBAUTHN_REGISTER_BEGIN)
def webauthn_register_begin(_request: Any, body: Any) -> dict[str, Any]: def webauthn_register_begin(_request: Any, body: Any) -> dict[str, Any]:
"""Begin WebAuthn registration — return options for ``credentials.create()``.""" """Begin WebAuthn registration — return options for ``credentials.create()``."""
if not body or not isinstance(body, dict): origin, rp_id = _check_webauthn_domain(body)
raise ValueError("Request body is required")
username = body.get("username") username = body.get("username")
if not username: if not username:
raise ValueError("username is required") raise ValueError("username is required")
options = create_registration_options(username) options = create_registration_options(
username,
origin=origin,
rp_id=rp_id,
)
return options return options
@registry.register(POST_AUTH_WEBAUTHN_REGISTER_FINISH) @registry.register(POST_AUTH_WEBAUTHN_REGISTER_FINISH)
def webauthn_register_finish(_request: Any, body: Any) -> dict[str, Any]: def webauthn_register_finish(_request: Any, body: Any) -> dict[str, Any]:
"""Finish WebAuthn registration — verify credential and persist.""" """Finish WebAuthn registration — verify credential and persist."""
if not body or not isinstance(body, dict): origin, rp_id = _check_webauthn_domain(body)
raise ValueError("Request body is required")
username = body.get("username") username = body.get("username")
credential_response = body.get("credential_response") credential_response = body.get("credential_response")
@@ -410,7 +468,12 @@ def webauthn_register_finish(_request: Any, body: Any) -> dict[str, Any]:
) )
cred = verify_registration( 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} 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 Public endpoint — no JWT required. Returns ``{"noWebAuthn": true}`` if the
user has no registered credentials (so the frontend can fall back to password). user has no registered credentials (so the frontend can fall back to password).
""" """
if not body or not isinstance(body, dict): _, rp_id = _check_webauthn_domain(body)
raise ValueError("Request body is required")
username = body.get("username") username = body.get("username")
if not username: if not username:
raise ValueError("username is required") raise ValueError("username is required")
options = create_authentication_options(username) options = create_authentication_options(
username,
rp_id=rp_id,
)
if options is None: if options is None:
return {"no_webauthn": True} return {"no_webauthn": True}
return options return options
@@ -441,8 +506,7 @@ def webauthn_authenticate_finish(_request: Any, body: Any) -> dict[str, Any]:
Public endpoint — no JWT required. Public endpoint — no JWT required.
""" """
if not body or not isinstance(body, dict): origin, rp_id = _check_webauthn_domain(body)
raise ValueError("Request body is required")
username = body.get("username") username = body.get("username")
assertion_response = body.get("assertion_response") 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): if not check_webauthn_rate(username, client_ip):
raise ValueError("Too many WebAuthn attempts. Please try again later.") 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) user = get_user(username)
if user is None: 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) @registry.register(GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS)
def webauthn_credential_counts(_request: Any, body: Any) -> dict[str, int]: def webauthn_credential_counts(_request: Any, body: Any) -> dict[str, int]:
"""Return credential counts for all users (admin endpoint).""" """Return credential counts for all users (admin endpoint)."""
from lib.webauthn import get_all_credential_counts
return get_all_credential_counts() return get_all_credential_counts()
+1
View File
@@ -199,6 +199,7 @@ GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS: Endpoint = _ep(
DELETE_AUTH_WEBAUTHN_CREDENTIAL: Endpoint = _ep( DELETE_AUTH_WEBAUTHN_CREDENTIAL: Endpoint = _ep(
"DELETE", "/auth/webauthn/creds/<credential_id>" "DELETE", "/auth/webauthn/creds/<credential_id>"
) )
GET_AUTH_WEBAUTHN_CAPABLE: Endpoint = _ep("GET", "/auth/webauthn/capable")
# ---- Server infra (not going through client) ---- # ---- Server infra (not going through client) ----
GET_HEALTH: Endpoint = _ep("GET", "/health") GET_HEALTH: Endpoint = _ep("GET", "/health")
+11 -1
View File
@@ -397,7 +397,9 @@ async def _handle_ws(request: web.Request) -> web.Response:
) )
session_header = request.headers.get("X-Session-Id") 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: if payload is None:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401) return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
@@ -556,6 +558,14 @@ def main() -> None:
setup_logging() 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() _register_routes()
app = create_app() app = create_app()
+36 -17
View File
@@ -346,19 +346,19 @@ def blacklist_expired() -> None:
class RateLimiter: 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 Failures are counted against the limit. A successful operation resets
the allowed count within the window. the failure counter for that key.
""" """
def __init__(self, max_attempts: int = 5, window_seconds: int = 300) -> None: def __init__(self, max_attempts: int = 5, window_seconds: int = 300) -> None:
self.max_attempts = max_attempts self.max_attempts = max_attempts
self.window = window_seconds self.window = window_seconds
self.attempts: dict[str, list[float]] = {} self.failures: dict[str, list[float]] = {}
def is_allowed(self, key: str) -> bool: 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: Args:
key: Identifier for the rate limit bucket (e.g., username or IP). key: Identifier for the rate limit bucket (e.g., username or IP).
@@ -368,25 +368,29 @@ class RateLimiter:
""" """
now = time.time() now = time.time()
cutoff = now - self.window cutoff = now - self.window
timestamps = self.attempts.get(key, []) timestamps = self.failures.get(key, [])
# Clean old entries # 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 len(self.failures[key]) < self.max_attempts
return False
self.attempts[key].append(now) def record_failure(self, key: str) -> None:
return True """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: def cleanup(self) -> None:
"""Remove expired entries from all buckets.""" """Remove expired entries from all buckets."""
now = time.time() now = time.time()
cutoff = now - self.window cutoff = now - self.window
for key in list(self.attempts): for key in list(self.failures):
self.attempts[key] = [t for t in self.attempts[key] if t > cutoff] self.failures[key] = [t for t in self.failures[key] if t > cutoff]
if not self.attempts[key]: if not self.failures[key]:
del self.attempts[key] del self.failures[key]
# Global rate limiters # 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: def check_login_rate(username: str, client_ip: str | None = None) -> bool:
"""Check if login is rate-limited for the given username. """Check if login is rate-limited for the given username.
Uses dual-key tracking: always records by IP (catches enumeration attacks), Checks failure counts for both IP and username buckets without
additionally records by username (catches legitimate users who forget password). recording anything. Callers must invoke record_login_failure() or
record_login_success() after the password verification step.
Args: Args:
username: The login attempt username. 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) 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: def check_webauthn_rate(username: str, client_ip: str | None = None) -> bool:
"""Check if WebAuthn authentication is rate-limited for the given username. """Check if WebAuthn authentication is rate-limited for the given username.
+8 -3
View File
@@ -31,6 +31,10 @@ from lib.password import hash_password, needs_rehash, verify_password
logger = logging.getLogger(__name__) 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 — hardcoded, full access, cannot be modified/deleted
BUILTIN_ADMIN_USERNAME = "admin" BUILTIN_ADMIN_USERNAME = "admin"
@@ -113,9 +117,10 @@ def verify_user_password(username: str, password: str) -> dict[str, Any] | None:
""" """
user = find_user(username) user = find_user(username)
if user is None: if user is None:
# Run a dummy Argon2id verification to prevent timing-based user enumeration. # Run a dummy Argon2id verification against a pre-computed hash to
# The timing for both paths is now equivalent. # prevent timing-based user enumeration. Uses module-level hash
verify_password(password, hash_password(secrets.token_hex(32))) # so both paths take ~1 verify call (~200ms) instead of ~400ms.
verify_password(password, _DUMMY_HASH)
return None return None
if not verify_password(password, user["password_hash"]): if not verify_password(password, user["password_hash"]):
return None return None
+9 -3
View File
@@ -292,8 +292,10 @@ def _seed_builtin_admin(db: Database) -> None:
if rows: if rows:
return return
# Create with a placeholder password that should be changed # Generate a random password — this fallback should only fire if
placeholder_hash = hash_password("CHANGEME") # 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) jwt_secret = secrets.token_urlsafe(32)
with db.in_transaction() as tx: with db.in_transaction() as tx:
@@ -303,7 +305,11 @@ def _seed_builtin_admin(db: Database) -> None:
for subsystem in ALL_SUBSYSTEMS: for subsystem in ALL_SUBSYSTEMS:
tx.run(Q_UPSERT_PERMISSION, (BUILTIN_ADMIN_USERNAME, subsystem, "rw")) 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: def reset_db_for_test() -> None:
+28
View File
@@ -225,6 +225,33 @@ def get_domains() -> list[dict[str, Any]]:
return result 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 # Domain CRUD
# ------------------------------------------------------------------ # ------------------------------------------------------------------
@@ -577,6 +604,7 @@ __all__ = [
"generate_server_conf", "generate_server_conf",
"get_config", "get_config",
"get_domains", "get_domains",
"get_management_domains",
"remove_domain", "remove_domain",
"save_config", "save_config",
"test_config", "test_config",
+69 -18
View File
@@ -73,9 +73,10 @@ def _get_webauthn_config() -> dict[str, Any]:
return raw.get("webauthn", {}) return raw.get("webauthn", {})
def get_rp_id() -> str: def is_enabled() -> bool:
"""Return the Relying Party ID from config.""" """Return whether WebAuthn is enabled in config."""
return _get_webauthn_config().get("rp_id", "localhost") cfg = _get_webauthn_config()
return cfg.get("enabled", True)
def get_rp_name() -> str: def get_rp_name() -> str:
@@ -83,9 +84,39 @@ def get_rp_name() -> str:
return _get_webauthn_config().get("rp_name", "Vacuum Wall") return _get_webauthn_config().get("rp_name", "Vacuum Wall")
def get_origin() -> str: def get_management_domains() -> list[str]:
"""Return the WebAuthn origin from config.""" """Return domain names eligible for WebAuthn.
return _get_webauthn_config().get("origin", "http://localhost")
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. """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 Returns a dict serializable to JSON, matching the format expected by
``navigator.credentials.create()``. ``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)) user_id = user_id + b"\x00" * (8 - len(user_id))
options = generate_registration_options( options = generate_registration_options(
rp_id=get_rp_id(), rp_id=rp_id,
rp_name=get_rp_name(), rp_name=get_rp_name(),
user_name=username, user_name=username,
user_display_name=username, user_display_name=username,
@@ -136,6 +176,8 @@ def verify_registration(
credential_response: dict[str, Any], credential_response: dict[str, Any],
registration_options: dict[str, Any], registration_options: dict[str, Any],
credential_name: str = "", credential_name: str = "",
origin: str = "",
rp_id: str = "",
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Verify a registration response and persist the credential. """Verify a registration response and persist the credential.
@@ -144,12 +186,12 @@ def verify_registration(
credential_response: Browser response from ``credentials.create()``. credential_response: Browser response from ``credentials.create()``.
registration_options: The options dict from ``create_registration_options``. registration_options: The options dict from ``create_registration_options``.
credential_name: Optional human-readable label. credential_name: Optional human-readable label.
origin: WebAuthn origin for verification.
rp_id: Relying Party ID for verification.
Returns: Returns:
Dict with ``id``, ``name``, ``transports``, ``sign_count``. Dict with ``id``, ``name``, ``transports``, ``sign_count``.
""" """
expected_origin = get_origin()
expected_rp_id = get_rp_id()
challenge = b64u_decode(registration_options["challenge"]) challenge = b64u_decode(registration_options["challenge"])
# The library accepts the credential response as a JSON-serializable dict # The library accepts the credential response as a JSON-serializable dict
@@ -157,8 +199,8 @@ def verify_registration(
col = verify_registration_response( col = verify_registration_response(
credential=credential_response, credential=credential_response,
expected_challenge=challenge, expected_challenge=challenge,
expected_origin=expected_origin, expected_origin=origin,
expected_rp_id=expected_rp_id, expected_rp_id=rp_id,
require_user_verification=False, 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. """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()``), Returns a dict serializable to JSON (for ``navigator.credentials.get()``),
or ``None`` if the user has no registered credentials. 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( options = generate_authentication_options(
rp_id=get_rp_id(), rp_id=rp_id,
allow_credentials=allow_credentials, allow_credentials=allow_credentials,
user_verification=UserVerificationRequirement.PREFERRED, user_verification=UserVerificationRequirement.PREFERRED,
) )
@@ -247,6 +296,8 @@ def verify_authentication(
username: str, username: str,
assertion_response: dict[str, Any], assertion_response: dict[str, Any],
auth_options: dict[str, Any], auth_options: dict[str, Any],
origin: str = "",
rp_id: str = "",
) -> bool: ) -> bool:
"""Verify an authentication assertion. """Verify an authentication assertion.
@@ -254,6 +305,8 @@ def verify_authentication(
username: The user authenticating. username: The user authenticating.
assertion_response: Browser response from ``credentials.get()``. assertion_response: Browser response from ``credentials.get()``.
auth_options: The options dict from ``create_authentication_options``. auth_options: The options dict from ``create_authentication_options``.
origin: WebAuthn origin for verification.
rp_id: Relying Party ID for verification.
Returns: Returns:
True on successful verification. True on successful verification.
@@ -261,8 +314,6 @@ def verify_authentication(
Raises: Raises:
ValueError: On verification failure. ValueError: On verification failure.
""" """
expected_origin = get_origin()
expected_rp_id = get_rp_id()
challenge = b64u_decode(auth_options["challenge"]) challenge = b64u_decode(auth_options["challenge"])
cred_id_str = assertion_response["id"] cred_id_str = assertion_response["id"]
@@ -279,8 +330,8 @@ def verify_authentication(
col = verify_authentication_response( col = verify_authentication_response(
credential=assertion_response, credential=assertion_response,
expected_challenge=challenge, expected_challenge=challenge,
expected_origin=expected_origin, expected_origin=origin,
expected_rp_id=expected_rp_id, expected_rp_id=rp_id,
credential_public_key=public_key, credential_public_key=public_key,
credential_current_sign_count=old_sign_count, credential_current_sign_count=old_sign_count,
require_user_verification=False, require_user_verification=False,
+27 -8
View File
@@ -429,11 +429,13 @@ class TestWebAuthnConfig:
"""Test WebAuthn configuration helpers.""" """Test WebAuthn configuration helpers."""
def test_get_rp_defaults(self) -> None: 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_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( @patch.dict(
@@ -482,14 +484,18 @@ class TestWebAuthnRegistration:
def test_create_registration_options_basic(self) -> None: def test_create_registration_options_basic(self) -> None:
from lib.webauthn import create_registration_options 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 isinstance(options, dict)
assert "challenge" in options assert "challenge" in options
assert "rp" in options assert "rp" in options
assert "user" in options assert "user" in options
assert "pubKeyCredParams" in options assert "pubKeyCredParams" in options
assert options["rp"]["id"] == "localhost" assert options["rp"]["id"] == "wall.example.com"
assert options["user"]["name"] == "testuser" assert options["user"]["name"] == "testuser"
assert len(options["pubKeyCredParams"]) >= 1 assert len(options["pubKeyCredParams"]) >= 1
@@ -524,6 +530,8 @@ class TestWebAuthnRegistration:
}, },
{"challenge": b64u_encode(b"testchallenge123")}, {"challenge": b64u_encode(b"testchallenge123")},
"My Key", "My Key",
origin="https://wall.example.com",
rp_id="wall.example.com",
) )
assert "id" in result assert "id" in result
@@ -542,7 +550,11 @@ class TestWebAuthnRegistration:
from lib.webauthn import create_registration_options 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 # Should contain the existing credential in exclude list
exclude_ids = [c["id"] for c in options.get("excludeCredentials", [])] 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: def test_create_auth_options_no_credentials(self) -> None:
from lib.webauthn import create_authentication_options 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 assert result is None
def test_create_auth_options_with_credentials(self) -> None: def test_create_auth_options_with_credentials(self) -> None:
@@ -598,7 +612,8 @@ class TestWebAuthnAuthentication:
from lib.webauthn import create_authentication_options 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 result is not None
assert len(result["allowCredentials"]) == 1 assert len(result["allowCredentials"]) == 1
@@ -618,6 +633,8 @@ class TestWebAuthnAuthentication:
}, },
}, },
{"challenge": _FAKE_CHALLENGE}, {"challenge": _FAKE_CHALLENGE},
origin="https://wall.example.com",
rp_id="wall.example.com",
) )
def test_verify_authentication_success(self) -> None: def test_verify_authentication_success(self) -> None:
@@ -643,6 +660,8 @@ class TestWebAuthnAuthentication:
}, },
}, },
{"challenge": _FAKE_CHALLENGE, "sign_count": 0}, {"challenge": _FAKE_CHALLENGE, "sign_count": 0},
origin="https://wall.example.com",
rp_id="wall.example.com",
) )
assert result is True assert result is True
+45
View File
@@ -15,6 +15,7 @@ from daemon.iface import (
DELETE_AUTH_WEBAUTHN_CREDENTIAL, DELETE_AUTH_WEBAUTHN_CREDENTIAL,
GET_AUTH_SESSION, GET_AUTH_SESSION,
GET_AUTH_USERS, GET_AUTH_USERS,
GET_AUTH_WEBAUTHN_CAPABLE,
GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS, GET_AUTH_WEBAUTHN_CREDENTIAL_COUNTS,
GET_AUTH_WEBAUTHN_CREDENTIALS, GET_AUTH_WEBAUTHN_CREDENTIALS,
POST_AUTH_LOGIN, 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"]) @bp.route("/webauthn/register-begin", methods=["POST"])
def webauthn_register_begin(): def webauthn_register_begin():
"""Begin WebAuthn registration. """Begin WebAuthn registration.
@@ -232,6 +266,9 @@ def webauthn_register_begin():
""" """
try: try:
body = request.get_json(silent=True) or {} 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)) return _ok(post(POST_AUTH_WEBAUTHN_REGISTER_BEGIN, body))
except Exception as exc: except Exception as exc:
logger.error("WebAuthn register begin failed: %s", exc) logger.error("WebAuthn register begin failed: %s", exc)
@@ -251,6 +288,9 @@ def webauthn_register_finish():
""" """
try: try:
body = request.get_json(silent=True) or {} 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)) return _ok(post(POST_AUTH_WEBAUTHN_REGISTER_FINISH, body))
except Exception as exc: except Exception as exc:
logger.error("WebAuthn register finish failed: %s", exc) logger.error("WebAuthn register finish failed: %s", exc)
@@ -271,6 +311,8 @@ def webauthn_authenticate_begin():
""" """
try: try:
body = request.get_json(silent=True) or {} 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)) return _ok(post(POST_AUTH_WEBAUTHN_AUTHENTICATE_BEGIN, body))
except Exception as exc: except Exception as exc:
logger.error("WebAuthn authenticate begin failed: %s", exc) logger.error("WebAuthn authenticate begin failed: %s", exc)
@@ -291,6 +333,9 @@ def webauthn_authenticate_finish():
try: try:
body = request.get_json(silent=True) or {} body = request.get_json(silent=True) or {}
body["client_ip"] = request.headers.get("X-Real-IP") or request.remote_addr 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)) return _ok(post(POST_AUTH_WEBAUTHN_AUTHENTICATE_FINISH, body))
except Exception as exc: except Exception as exc:
logger.error("WebAuthn authenticate finish failed: %s", exc) logger.error("WebAuthn authenticate finish failed: %s", exc)
+14
View File
@@ -143,6 +143,20 @@ export function webauthnSupported() {
return typeof window !== 'undefined' && !!window.PublicKeyCredential; 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<object>} { 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 ──────────────────────────────────────────────── */ /* ─── Base64url helpers ──────────────────────────────────────────────── */
/** /**