WebSocket authentication in FastAPI

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

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.

Practice

1. What is the main purpose of WebSocket authentication in FastAPI?

easy

A. To speed up the WebSocket connection process

B. To confirm the user's identity before allowing the WebSocket connection

C. To encrypt the WebSocket messages automatically

D. To log all WebSocket messages for debugging

Solution

1. Step 1: Understand WebSocket authentication role

   Authentication ensures only authorized users can connect to the WebSocket.

2. Step 2: Identify the main goal in FastAPI context

   FastAPI uses authentication to confirm user identity before opening the connection.

3. Final Answer:

   To confirm the user's identity before allowing the WebSocket connection -> Option B

4. Quick Check:

   Authentication = Confirm identity [OK]

Hint: Authentication means confirming user identity before connection [OK]

Common Mistakes:

• Confusing authentication with encryption
• Thinking authentication speeds up connection
• Assuming authentication logs messages

2. Which of the following is the correct way to accept a WebSocket connection with token authentication in FastAPI?

easy

A. await websocket.accept() without checking token

B. Call await websocket.close() before checking token

C. Use await websocket.send_text() before accepting

D. Check token, then call await websocket.accept() if valid

Solution

1. Step 1: Understand WebSocket connection flow

   You must check the token first to decide if connection is allowed.

2. Step 2: Accept connection only if token is valid

   Call await websocket.accept() only after confirming token validity.

3. Final Answer:

   Check token, then call await websocket.accept() if valid -> Option D

4. Quick Check:

   Accept after token check = C [OK]

Hint: Accept connection only after validating token [OK]

Common Mistakes:

• Accepting connection before token check
• Closing connection before checking token
• Sending messages before accepting connection

3. Given this FastAPI WebSocket code snippet, what will happen if the token is invalid?

async def websocket_endpoint(websocket: WebSocket):
    token = websocket.query_params.get('token')
    if token != 'secret123':
        await websocket.close(code=1008)
    else:
        await websocket.accept()
        await websocket.send_text('Welcome!')

medium

A. The connection closes immediately with code 1008

B. The connection accepts and sends 'Welcome!' anyway

C. The server raises a runtime error

D. The connection stays open but no message is sent

Solution

1. Step 1: Check token validation logic

   If token is not 'secret123', the code calls await websocket.close(code=1008).

2. Step 2: Understand WebSocket close behavior

   Calling close immediately ends the connection with the given code.

3. Final Answer:

   The connection closes immediately with code 1008 -> Option A

4. Quick Check:

   Invalid token triggers close(1008) [OK]

Hint: Invalid token triggers immediate close with code 1008 [OK]

Common Mistakes:

• Assuming connection accepts anyway
• Expecting a runtime error
• Thinking connection stays open silently

4. Identify the error in this FastAPI WebSocket authentication code:

async def websocket_endpoint(websocket: WebSocket):
    token = websocket.headers.get('Authorization')
    if token != 'Bearer secret123':
        await websocket.accept()
        await websocket.send_text('Unauthorized')
        await websocket.close()
    else:
        await websocket.accept()
        await websocket.send_text('Welcome')

medium

A. Using headers.get instead of query_params.get

B. Not closing connection on invalid token

C. Accepting connection before sending 'Unauthorized' message

D. Missing await before websocket.close()

Solution

1. Step 1: Analyze connection flow on invalid token

   The code calls await websocket.accept() before sending 'Unauthorized' and closing.

2. Step 2: Understand best practice for unauthorized access

   It is better to close connection immediately without accepting it to avoid opening unauthorized connection.

3. Final Answer:

   Accepting connection before sending 'Unauthorized' message -> Option C

4. Quick Check:

   Unauthorized should close without accept [OK]

Hint: Do not accept connection before rejecting unauthorized users [OK]

Common Mistakes:

• Accepting connection before rejecting
• Ignoring connection close on invalid token
• Confusing headers and query parameters

5. You want to authenticate WebSocket connections in FastAPI using a token passed as a query parameter. Which approach correctly handles authentication and rejects unauthorized users cleanly?

hard

A. Check token from query params; if invalid, call await websocket.close(1008) before accept; else accept connection

B. Accept connection first, then check token; if invalid, send 'Unauthorized' message and close

C. Ignore token and accept all connections; handle authorization in message handlers

D. Check token in headers only; ignore query parameters

Solution

1. Step 1: Understand token location and validation

   Token is passed as query parameter, so check websocket.query_params for token.

2. Step 2: Handle unauthorized users before accepting connection

   If token is invalid, close connection immediately with code 1008 to reject cleanly.

3. Step 3: Accept connection only if token is valid

   Only call await websocket.accept() after confirming token validity.

4. Final Answer:

   Check token from query params; if invalid, call await websocket.close(1008) before accept; else accept connection -> Option A

5. Quick Check:

   Reject before accept with close(1008) = A [OK]

Hint: Reject unauthorized before accept using close(1008) [OK]

Common Mistakes:

• Accepting connection before token check
• Ignoring query parameters for token
• Handling authorization after connection open