FastAPI WebSocket集成HTTPBasic认证报错：缺少request参数

使用FastAPI开发带HTTPBasic认证的简易WebSocket应用时，/{id}的HTTP请求可正常通过认证，但/ws/{client_id}的WebSocket连接失败，报错信息如下：

ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "D:\Installations\envs\chat-app\lib\site-packages\uvicorn\protocols\websockets\websockets_impl.py", line 240, in run_asgi
    result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
  File "D:\Installations\envs\chat-app\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 69, in __call__
    return await self.app(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\applications.py", line 1054, in __call__
    await super().__call__(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\applications.py", line 123, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\middleware\errors.py", line 151, in __call__
    await self.app(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\middleware\exceptions.py", line 65, in __call__
    await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 756, in __call__
    await self.middleware_stack(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 776, in app
    await route.handle(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 373, in handle
    await self.app(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 96, in app
    await wrap_app_handling_exceptions(app, session)(scope, receive, send)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 64, in wrapped_app
    raise exc
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
    await app(scope, receive, sender)
  File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 94, in app
    await func(session)
  File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\routing.py", line 338, in app
    solved_result = await solve_dependencies(
  File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\dependencies\utils.py", line 572, in solve_dependencies
    solved_result = await solve_dependencies(
  File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\dependencies\utils.py", line 600, in solve_dependencies
    solved = await call(**sub_values)
TypeError: HTTPBasic.__call__() missing 1 required positional argument: 'request'
INFO:     connection open
INFO:     connection closed

原代码 #

from fastapi import FastAPI, HTTPException, status, Depends, Request, WebSocket, WebSocketDisconnect
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.templating import Jinja2Templates
from typing import Annotated, List

app = FastAPI()
security = HTTPBasic()
templates = Jinja2Templates(directory="templates")


class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str, websocket: WebSocket) :
        for connection in self.active_connections:
            if (connection == websocket):
                continue
            await connection.send_text(message)

connectionmanager = ConnectionManager()

def validate_user(username: bytes, password: bytes):
    # 自定义用户验证逻辑
    return username == b"test", password == b"test"

def get_current_username(
    credentials: Annotated[HTTPBasicCredentials, Depends(security)],
):
    is_correct_username, is_correct_password= validate_user(credentials.username.encode("utf8"),credentials.password.encode("utf8"))
    
    if not (is_correct_username and is_correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )
    return credentials.username

@app.get("/{id}", response_class=HTMLResponse)
def read_user(id: int,  username: Annotated[str, Depends(get_current_username)], request: Request):
    return templates.TemplateResponse("index.html", {"request": request})


@app.websocket("/ws/{client_id}")
async def websocket_endpoint(
        websocket: WebSocket, client_id: int, username: Annotated[str, Depends(get_current_username)], request: Request):
    await connectionmanager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await connectionmanager.send_personal_message(f"You : {data}", websocket)
            await connectionmanager.broadcast(f"Client #{client_id}: {data}", websocket)

    except WebSocketDisconnect:
        connectionmanager.disconnect(websocket)
        await connectionmanager.broadcast(f"Client #{client_id} left the chat")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host='localhost', port=8000)

错误根源是：HTTPBasic依赖是为HTTP请求设计的，它默认需要注入Request对象，但WebSocket的ASGI Scope和HTTP请求结构不同，无法自动注入Request，导致调用HTTPBasic.__call__时缺少必填参数。

解决方法是手动从WebSocket的Scope中提取认证信息，替代原有的HTTPBasic依赖：

1. 编写WebSocket专属的认证依赖 #

from base64 import b64decode

def get_current_username_from_websocket(websocket: WebSocket):
    # 从WebSocket headers中提取Authorization头
    auth_header = None
    for key, value in websocket.scope["headers"]:
        if key == b"authorization":
            auth_header = value.decode()
            break
    
    if not auth_header:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing credentials",
            headers={"WWW-Authenticate": "Basic"},
        )
    
    # 解析Basic认证格式
    auth_parts = auth_header.split(" ", 1)
    if len(auth_parts) != 2 or auth_parts[0].lower() != "basic":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Unsupported authentication method",
            headers={"WWW-Authenticate": "Basic"},
        )
    
    try:
        decoded_creds = b64decode(auth_parts[1]).decode("utf-8")
        username, password = decoded_creds.split(":", 1)
    except (ValueError, UnicodeDecodeError):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials format",
            headers={"WWW-Authenticate": "Basic"},
        )
    
    # 复用原有用户验证逻辑
    is_correct_username, is_correct_password = validate_user(username.encode("utf8"), password.encode("utf8"))
    if not (is_correct_username and is_correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Basic"},
        )
    
    return username

2. 修改WebSocket路由 #

替换依赖为上面的函数，同时移除不需要的Request参数：

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(
        websocket: WebSocket, client_id: int, 
        username: Annotated[str, Depends(get_current_username_from_websocket)]):
    # 先完成认证，再接受连接
    await connectionmanager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await connectionmanager.send_personal_message(f"You : {data}", websocket)
            await connectionmanager.broadcast(f"Client #{client_id}: {data}", websocket)

    except WebSocketDisconnect:
        connectionmanager.disconnect(websocket)
        await connectionmanager.broadcast(f"Client #{client_id} left the chat")

3. 客户端连接说明 #

WebSocket客户端在发起连接时，需要在握手请求中携带Authorization: Basic <base64编码的用户名:密码>头，否则会被拒绝连接。

原有的HTTP路由/{id}可以继续使用get_current_username依赖，不受影响。

内容的提问来源于stack exchange，提问作者Aravindan vaithialingam