auth: fix WS session_id extraction and track WebAuthn success/failure

Browsers cannot send custom X-Session-Id header on WebSocket connections,
so decode the token payload to extract session_id. Add WebAuthn
success/failure recording to support rate limiter counter resets.
This commit is contained in:
2026-08-12 04:27:14 +00:00
parent 6f728cf853
commit 3654209b78
3 changed files with 38 additions and 11 deletions
+15 -7
View File
@@ -39,6 +39,8 @@ from lib.auth import (
get_access_ttl,
record_login_failure,
record_login_success,
record_webauthn_failure,
record_webauthn_success,
validate_token,
)
from lib.auth_users import (
@@ -519,13 +521,19 @@ 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,
origin=origin,
rp_id=rp_id,
)
try:
verify_authentication(
username,
assertion_response,
auth_options,
origin=origin,
rp_id=rp_id,
)
except ValueError:
record_webauthn_failure(username, client_ip)
raise
record_webauthn_success(username, client_ip)
user = get_user(username)
if user is None:
+9 -4
View File
@@ -373,7 +373,7 @@ async def _handle_ws(request: web.Request) -> web.Response:
1. WebSocket subprotocol header (Sec-WebSocket-Protocol: "Bearer <token>")
2. X-Auth-Token header (nginx-injected)
"""
from lib.auth import validate_token
from lib.auth import decode_token, validate_token
token_param = None
@@ -392,11 +392,16 @@ async def _handle_ws(request: web.Request) -> web.Response:
{"ok": False, "error": "authentication required"}, status=401
)
session_header = request.headers.get("X-Session-Id")
if not session_header:
# Decode token to extract session_id from payload (browsers can't send
# X-Session-Id header on WebSocket connections, only subprotocols)
raw_payload = decode_token(token_param)
if raw_payload is None:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
payload = validate_token(
token_param, token_type="access", session_id=session_header
token_param,
token_type="access",
session_id=raw_payload.get("session_id"),
)
if payload is None:
return web.json_response({"ok": False, "error": "unauthorized"}, status=401)
+14
View File
@@ -453,3 +453,17 @@ def check_webauthn_rate(username: str, client_ip: str | None = None) -> bool:
if client_ip and not _webauthn_limiter.is_allowed(client_ip):
return False
return _webauthn_limiter.is_allowed(username)
def record_webauthn_failure(username: str, client_ip: str | None = None) -> None:
"""Record a failed WebAuthn attempt."""
if client_ip:
_webauthn_limiter.record_failure(client_ip)
_webauthn_limiter.record_failure(username)
def record_webauthn_success(username: str, client_ip: str | None = None) -> None:
"""Record a successful WebAuthn attempt (resets failure counter)."""
if client_ip:
_webauthn_limiter.record_success(client_ip)
_webauthn_limiter.record_success(username)