(websocket: WebSocket)
| 977 | |
| 978 | @router.websocket("/captcha_ws") |
| 979 | async def captcha_websocket_endpoint(websocket: WebSocket): |
| 980 | from ..core.logger import debug_logger |
| 981 | api_key = ( |
| 982 | websocket.query_params.get("key") |
| 983 | or websocket.query_params.get("api_key") |
| 984 | or websocket.headers.get("x-goog-api-key") |
| 985 | or "" |
| 986 | ).strip() |
| 987 | authorization = (websocket.headers.get("authorization") or "").strip() |
| 988 | if authorization.lower().startswith("bearer "): |
| 989 | api_key = authorization[7:].strip() |
| 990 | |
| 991 | if not api_key or not AuthManager.verify_api_key(api_key): |
| 992 | await websocket.close(code=1008) |
| 993 | return |
| 994 | |
| 995 | service = await ExtensionCaptchaService.get_instance() |
| 996 | await service.connect(websocket) |
| 997 | try: |
| 998 | while True: |
| 999 | data = await websocket.receive_text() |
| 1000 | await service.handle_message(websocket, data) |
| 1001 | except WebSocketDisconnect: |
| 1002 | service.disconnect(websocket) |
| 1003 | except Exception as e: |
| 1004 | debug_logger.log_error(f"WebSocket error: {e}") |
| 1005 | service.disconnect(websocket) |
nothing calls this directly
no test coverage detected