SSL/TLS & Security

~13 min · production, security, wss

Level 0Poller

0 XP0/60 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

wss:// is non-negotiable

Production WebSocket runs over TLS. Many corporate firewalls and proxies block ws:// outright. Cleartext exposes auth tokens to anyone on the path. The setup is identical to HTTPS — TLS terminates at the proxy or the application, the WebSocket runs over the same TLS stream. Get a free certificate via Let's Encrypt; renew automatically.

Security checklist

Treat the handshake as the auth boundary. Validate origin (browsers do not enforce CORS for WebSocket). Authenticate the user (JWT, cookies, signed token). Rate-limit per-IP and per-user. Validate every incoming message (Pydantic). Cap message size (default Starlette limit is 16MB; lower it). Cap connection counts per user (multi-device fine, multi-thousand-device suspicious).

Code

Defense-in-depth WebSocket endpoint·python

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import time

ALLOWED_ORIGINS = {'https://app.example.com', 'https://admin.example.com'}
MAX_MSG_PER_SEC = 20

@app.websocket('/ws')
async def secure_ws(websocket: WebSocket):
    # 1. Origin validation (CORS for WebSocket is your job)
    origin = websocket.headers.get('origin', '')
    if origin not in ALLOWED_ORIGINS:
        await websocket.close(code=4003, reason='forbidden origin')
        return

    # 2. Auth — before accept()
    token = websocket.query_params.get('token')
    user = decode_jwt(token) if token else None
    if not user:
        await websocket.close(code=4001, reason='unauthorized')
        return

    await websocket.accept()

    # 3. Per-connection state for rate limiting
    msg_window = []  # timestamps in the last second

    try:
        async for msg in websocket.iter_json():
            now = time.time()
            msg_window = [t for t in msg_window if now - t < 1.0]
            if len(msg_window) >= MAX_MSG_PER_SEC:
                await websocket.send_json({
                    'type': 'error', 'code': 'rate_limited',
                })
                continue
            msg_window.append(now)

            # 4. Validate every message
            if not isinstance(msg, dict) or 'type' not in msg:
                await websocket.send_json({
                    'type': 'error', 'code': 'invalid_message',
                })
                continue

            await dispatch(websocket, user, msg)
    except WebSocketDisconnect:
        pass

External links

OWASP Cheat Sheet: WebSocket Security

Exercise

Audit your existing WebSocket endpoints against the checklist: origin, auth, rate limit, message validation, message size, connection cap. Score each on a 0/1/2 scale (0 missing, 1 partial, 2 done). Fix anything below 2. Then deploy wss:// via Caddy or Let's Encrypt and confirm ws:// is rejected.