LeoVasanko
3ff41ca354
- validate_config: reject multiple auth-host marks per domain; sanitize_config clears extras (first wins) and coerces junk entry values to presence-only - origin_key: lowercase keys, strip trailing dots (bare hosts/wildcards) - Passkey._allowlisted: tolerate trailing-dot wildcard bases - wschat: stamp remote-flow sessions with the session host's domain, not the approver's - auth_host redirects: keep the port (redirect to the configured auth host instead of the normalized, port-less current host) - update_domain: required fields (wholesale replace) — no silent wipes - admin: fix pre-existing lockout-guard order in org permission removal; permission PATCH keeps domain restriction when omitted; 400 instead of 500 on unknown permission UUIDs - Drop dead code: db.update_config/set_session_host/delete_reset_token, Session.metadata, oidjwt.clear_key, background aliases, avatar.current_avatar_url/media_root, wsutil.require_pow - Prune stale/duplicated comments and docstrings
51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
"""
|
|
Shared WebSocket utilities for FastAPI endpoints.
|
|
"""
|
|
|
|
import logging
|
|
from functools import wraps
|
|
|
|
from fastapi import WebSocket, WebSocketDisconnect
|
|
from webauthn.helpers.exceptions import InvalidAuthenticationResponse
|
|
|
|
from paskia.domains import current_domain
|
|
from paskia.fastapi import authz
|
|
|
|
|
|
def websocket_error_handler(func):
|
|
"""Decorator for WebSocket endpoints that handles common errors."""
|
|
|
|
@wraps(func)
|
|
async def wrapper(ws: WebSocket, *args, **kwargs):
|
|
try:
|
|
await ws.accept()
|
|
return await func(ws, *args, **kwargs)
|
|
except WebSocketDisconnect:
|
|
pass
|
|
except authz.AuthException as e:
|
|
await ws.send_json(
|
|
{
|
|
"status": e.status_code,
|
|
**(await authz.auth_error_content(e)),
|
|
}
|
|
)
|
|
except (ValueError, InvalidAuthenticationResponse) as e:
|
|
await ws.send_json({"status": 401, "detail": str(e)})
|
|
except Exception:
|
|
logging.exception("Internal Server Error")
|
|
await ws.send_json({"status": 500, "detail": "Internal Server Error"})
|
|
|
|
return wrapper
|
|
|
|
|
|
def validate_origin(ws: WebSocket) -> str:
|
|
"""Extract and validate origin from WebSocket request headers.
|
|
|
|
Raises:
|
|
ValueError: If origin header is missing or not allowed in the current domain
|
|
"""
|
|
origin = ws.headers.get("origin")
|
|
if not origin:
|
|
raise ValueError("Origin header is required for WebSocket connections")
|
|
return current_domain().passkey.validate_origin(origin)
|