0
0
FastAPIframework~7 mins

WebSocket authentication in FastAPI

Choose your learning style9 modes available
LearnWhyDeepVisualTryChallengeProjectRecallPerf
Introduction

WebSocket authentication ensures only allowed users can open and use a WebSocket connection. It protects your app from unwanted access.

When you want to keep a chat app private to logged-in users.
When sending real-time updates that only certain users should see.
When you need to verify user identity before allowing a WebSocket connection.
When you want to close connections if the user is not authorized.
When you want to reuse existing login tokens or cookies for WebSocket security.
Syntax
FastAPI
from fastapi import WebSocket, WebSocketDisconnect, Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # Verify token and return user info or raise error
    ...

async def websocket_endpoint(websocket: WebSocket, user=Depends(get_current_user)):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        pass

Use Depends to inject authentication logic into WebSocket endpoints.

Call await websocket.accept() only after verifying the user.

Examples
This example checks a token sent as a query parameter before accepting the WebSocket.
FastAPI
from fastapi import WebSocket, Depends

async def get_token(websocket: WebSocket):
    token = websocket.query_params.get('token')
    if token != 'secret':
        await websocket.close(code=1008)  # Policy Violation
    return token

async def websocket_endpoint(websocket: WebSocket, token=Depends(get_token)):
    await websocket.accept()
    await websocket.send_text(f"Connected with token: {token}")
This example uses OAuth2 token from headers to authenticate before accepting the WebSocket.
FastAPI
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != "validtoken":
        raise Exception("Invalid token")
    return {"username": "user1"}

async def websocket_endpoint(websocket: WebSocket, user=Depends(get_current_user)):
    await websocket.accept()
    await websocket.send_text(f"Hello {user['username']}")
Sample Program

This FastAPI app shows a simple WebSocket server that authenticates users by checking a token sent as a query parameter. If the token is correct, the connection is accepted and messages are echoed back. The HTML page lets you test the connection by sending a message.

FastAPI
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, status
from fastapi.security import OAuth2PasswordBearer
from fastapi.responses import HTMLResponse

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    if token != "secrettoken":
        raise Exception("Invalid token")
    return {"username": "alice"}

@app.get("/", response_class=HTMLResponse)
async def get():
    return """
    <html>
        <head>
            <title>WebSocket Auth Test</title>
        </head>
        <body>
            <h1>WebSocket Auth Test</h1>
            <button onclick="connect()">Connect WebSocket</button>
            <ul id="messages"></ul>
            <script>
                function connect() {
                    const token = 'secrettoken';
                    const ws = new WebSocket(`ws://localhost:8000/ws?token=${token}`);
                    ws.onmessage = function(event) {
                        const messages = document.getElementById('messages');
                        const li = document.createElement('li');
                        li.textContent = event.data;
                        messages.appendChild(li);
                    };
                    ws.onopen = () => ws.send('Hello Server');
                }
            </script>
        </body>
    </html>
    """

async def get_token(websocket: WebSocket):
    token = websocket.query_params.get('token')
    if token != 'secrettoken':
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        raise Exception("Invalid token")
    return token

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str = Depends(get_token)):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        pass
OutputSuccess
Important Notes

WebSocket connections do not send headers like normal HTTP requests, so tokens are often sent as query parameters or in the first message.

Always close the WebSocket with a proper code if authentication fails to inform the client.

Use HTTPS and WSS (secure WebSocket) in production to protect tokens in transit.

Summary

WebSocket authentication protects your real-time connections from unauthorized users.

Use dependency injection with Depends to check tokens before accepting connections.

Tokens can be sent as query parameters or in headers depending on your setup.