How to Secure a WebSocket Endpoint in FastAPI and Stop Cross-Site Hijacking, Leaked Tokens, Floods, and Slow Readers

Your app needs to push live updates to a browser, a chat message here, a build log there, and you reach for a WebSocket. It works on your laptop in five minutes. This tutorial is about the five weeks after that: another website quietly opens a socket to your server as your signed-in user, one client that stops reading freezes everyone else’s updates, a proxy hangs up on quiet connections, and a deploy drops every client at once. You will build a small notice board in FastAPI, break it in each of those ways, and fix each break with a few lines you can explain. Everything was run on Windows 11 with Python 3.13.14, FastAPI 0.162.0, Starlette 1.10.0, uvicorn 0.54.0 and Microsoft Edge 156, and the final app is pinned by 25 tests that run in about half a second.

A WebSocket deserves its own security tutorial because it does not follow the habits you have built for HTTP. The browser does not apply CORS to it, cookies ride along on the handshake by themselves, and a connection lives for hours instead of milliseconds, so every client costs you memory, a task and a place in a send loop for as long as it stays. Step 3 shows the first consequence in a real browser: a page on another origin opens a socket as the signed-in user and reads their API key.

What you will be able to do

• Explain the WebSocket handshake and read one by hand.
• Reproduce a cross-site WebSocket hijack in Edge, and say which cookie settings stop it and which do not.
• Check the Origin header correctly, and replace cookie authentication on the socket with one-time tickets.
• Limit message size, message shape, message rate and connections per user.
• Keep one stalled client from blocking the others, notice clients that have gone silent, and run the app behind nginx.
• Shut down and reconnect cleanly, and pin all of it with tests that fail when someone weakens a defense.

Five ideas to know first

What a WebSocket is

A WebSocket starts life as an ordinary HTTP request. The client asks to switch protocols, the server answers 101 Switching Protocols, and from then on the same TCP connection carries small messages, called frames, in both directions at any time. There is no pairing of request and response: the server can send ten messages while the client sends none. That is what makes live updates cheap, and it is also why a socket is a different thing to secure. RFC 6455, the protocol’s specification, leaves authentication to you. It says that the protocol “doesn’t prescribe any particular way that servers can authenticate clients during the WebSocket handshake” and that “the WebSocket server can use any client authentication mechanism available to a generic HTTP server, such as cookies, HTTP authentication, or TLS authentication”.

The handshake and the Origin header

When a browser opens a socket, it sends the handshake with an Origin header that names the page that ran new WebSocket(...), and it adds the cookies it holds for the server’s host. MDN’s guide for people who write WebSocket servers says that “All browsers send an Origin header”. RFC 6455 section 10.2 says that a server meant only for certain sites “SHOULD verify the |Origin| field is an origin they expect” and “SHOULD respond to the WebSocket handshake with a reply containing HTTP 403 Forbidden status code” when the origin is not acceptable.

Notice who enforces what. With fetch(), CORS makes the browser refuse to show a response to a page from another origin. For a WebSocket there is no such gate: the browser opens the connection and sends the cookies, and the server has to look at Origin and decide. Your CORS settings from the CORS tutorial do not protect a socket: in Starlette 1.10.0, CORSMiddleware hands any connection that is not plain HTTP straight to the app.

Origin versus site

Two words matter for the experiments below. web.dev says that websites with “the same combination of scheme, hostname, and port” are same-origin, and that everything else is cross-origin. It then defines a looser idea: websites with “the same scheme and the same eTLD+1” are same-site. That second idea is what a cookie’s SameSite attribute uses, and it ignores ports and subdomains. The lab uses three pages, and the table says how each relates to the notice board at http://localhost:8101.

Close codes

Every WebSocket ends with a close code, and you choose most of them. These are the ones this tutorial uses. The meanings come from RFC 6455 section 7.4.1, and the names of 1012 and 1013 from the IANA WebSocket close code registry.

Two more numbers appear in the browser, and neither is ever sent on the wire. RFC 6455 reserves 1005 for “no status code was actually present” and 1006 for a connection “closed abnormally, e.g., without sending or receiving a Close control frame”. In this lab, 1005 is what Edge reports when our own page closes a socket politely, and 1006 is what it reports when the handshake never completed.

Why long-lived connections change the maths

An HTTP request is over in milliseconds, so slow clients are rare and short. A socket stays open, and a client can stop reading, stop answering, or vanish without a goodbye, while your server still holds a buffer, a task and a slot in some list for it. Steps 5 and 6 are about not letting those leftovers add up. Step 7 covers the proxy in front of you and Step 8 the deploy that restarts you.

Prerequisites

• Python 3.13 (I used 3.13.14), pip and a terminal. The commands work in Git Bash or any POSIX shell; where PowerShell differs I say so.
• A Chromium-based browser for Step 3. Every browser result below was measured in Microsoft Edge 156.0.4314.58 driven by Playwright. Cookie defaults differ between browsers (MDN says that “Some browsers use Lax as the default value if SameSite is not specified”), so repeat the cookie experiments in each browser you support.
• Enough Python to read a small FastAPI app and async functions. You do not need to know anything about WebSockets yet.
• Optional, for Step 7: nginx. The lab downloads the Windows zip itself; on Linux or macOS install nginx 1.30 or later with your package manager.

Step 1: Set up the lab and watch a handshake

1a. Install the packages

Make a folder called ws_lab with a site folder inside it, and install pinned versions so your results match mine.

python -m venv labvenv
labvenv\Scripts\activate          # macOS and Linux: source labvenv/bin/activate
pip install fastapi==0.162.0 "uvicorn[standard]==0.54.0" websockets==17.4 httpx2==2.16.0 pytest==9.2.0 requests==2.34.2 playwright==1.64.1 psutil==7.2.2

1b. The shared helper

Every experiment in this tutorial starts a server, talks to it, and stops it. labkit.py does the starting and stopping, reads a server’s memory use, and holds two small helpers that speak raw WebSocket handshakes. It binds servers to localhost on purpose, to avoid a trap I hit on Windows: with a server listening only on 127.0.0.1, requests to the name localhost took a median of 2,077 milliseconds, and 5 milliseconds once the server listened on localhost itself.

"""labkit.py: start and stop lab servers, read memory use, and a few raw-socket helpers for the step scripts."""
import base64
import functools
import hashlib
import http.server
import os
import socket
import subprocess
import sys
import threading
import time

import psutil

HERE = os.path.dirname(os.path.abspath(__file__))
HOST = "localhost"
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"  # the fixed string from RFC 6455 section 1.3
os.makedirs(os.path.join(HERE, "out"), exist_ok=True)


def wait_port(port: int, seconds: float = 15.0) -> None:
    deadline = time.time() + seconds
    while time.time() < deadline:
        try:
            socket.create_connection((HOST, port), timeout=0.3).close()
            return
        except OSError:
            time.sleep(0.1)
    raise RuntimeError(f"nothing is listening on port {port}")


def start(app: str, port: int, *extra: str, env: dict | None = None, log: str | None = None) -> subprocess.Popen:
    """Run `uvicorn app --host localhost --port port` in the lab folder and wait until it accepts connections."""
    cmd = [sys.executable, "-m", "uvicorn", app, "--host", HOST, "--port", str(port), *extra]
    sink = open(os.path.join(HERE, "out", log), "w", encoding="utf-8") if log else subprocess.DEVNULL
    environment = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1", **(env or {})}
    proc = subprocess.Popen(cmd, cwd=HERE, env=environment, stdout=sink, stderr=subprocess.STDOUT)
    wait_port(port)
    return proc


def stop(proc: subprocess.Popen) -> None:
    subprocess.run(["taskkill", "/PID", str(proc.pid), "/T", "/F"], capture_output=True)
    try:
        proc.wait(timeout=10)
    except subprocess.TimeoutExpired:
        pass


def rss_mb(proc: subprocess.Popen) -> float:
    """Resident memory of the server (the venv launcher has a child; take the largest process in the tree)."""
    root = psutil.Process(proc.pid)
    return max(p.memory_info().rss for p in [root, *root.children(recursive=True)]) / 2**20


def serve_static(port: int, directory: str, bind: str = "127.0.0.1") -> http.server.ThreadingHTTPServer:
    class Quiet(http.server.SimpleHTTPRequestHandler):
        def log_message(self, *args):
            pass

    handler = functools.partial(Quiet, directory=directory)
    server = http.server.ThreadingHTTPServer((bind, port), handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return server


def raw_handshake(port: int, path: str, headers: dict | None = None, rcvbuf: int | None = None):
    """Open a TCP socket and do the WebSocket upgrade by hand. Returns (socket, request_text, response_text)."""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    if rcvbuf:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, rcvbuf)
    sock.connect(("127.0.0.1", port))
    key = base64.b64encode(os.urandom(16)).decode()
    lines = [f"GET {path} HTTP/1.1", f"Host: localhost:{port}", "Upgrade: websocket", "Connection: Upgrade",
             f"Sec-WebSocket-Key: {key}", "Sec-WebSocket-Version: 13"]
    lines += [f"{name}: {value}" for name, value in (headers or {}).items()]
    request = "\r\n".join(lines) + "\r\n\r\n"
    sock.sendall(request.encode())
    response = b""
    while b"\r\n\r\n" not in response:
        chunk = sock.recv(4096)
        if not chunk:
            break
        response += chunk
    head = response.partition(b"\r\n\r\n")[0]
    return sock, request, head.decode("latin-1"), key


def accept_for(key: str) -> str:
    """What the server must answer in Sec-WebSocket-Accept: base64(sha1(key + GUID))."""
    return base64.b64encode(hashlib.sha1((key + GUID).encode()).digest()).decode()

1c. The smallest WebSocket server

This is a complete server. @app.websocket declares a route for WebSocket connections. accept() finishes the handshake. After that the loop reads a text message and sends one back, and WebSocketDisconnect is the exception Starlette raises when the client leaves.

"""app_echo.py: the smallest useful WebSocket server (FastAPI 0.162.0, uvicorn 0.54.0)."""
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()


@app.websocket("/echo")
async def echo(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            text = await websocket.receive_text()
            await websocket.send_text(f"echo: {text}")
    except WebSocketDisconnect:
        pass

Now the script that opens a connection by hand. It writes the HTTP request on a raw TCP socket, prints what the server answers, computes the one header value the answer must contain, and then talks to the same server through the websockets library.

"""step01_handshake.py: do the WebSocket upgrade by hand on a raw TCP socket, then talk to the same server with a library."""
import labkit
from websockets.sync.client import connect

server = labkit.start("app_echo:app", 8100, "--log-level", "warning")
try:
    sock, request, response, key = labkit.raw_handshake(8100, "/echo", headers={"Origin": "http://localhost:3000"})
    print("--- what the client sent ---")
    print(request.replace(chr(13) + chr(10), chr(10)).rstrip())
    print("--- what the server answered ---")
    print(response.replace(chr(13) + chr(10), chr(10)))
    print("--- the check ---")
    expected = labkit.accept_for(key)
    print("base64(sha1(key + GUID)) =", expected)
    print("server's Sec-WebSocket-Accept matches:", f"Sec-WebSocket-Accept: {expected}" in response)
    sock.close()

    print("--- the same server, through the websockets library ---")
    with connect("ws://localhost:8100/echo") as ws:
        ws.send("hello")
        print("sent 'hello', got back:", ws.recv())
finally:
    labkit.stop(server)

python step01_handshake.py

--- what the client sent ---
GET /echo HTTP/1.1
Host: localhost:8100
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: ZKUz6TNwPTtWlxWtLXrENA==
Sec-WebSocket-Version: 13
Origin: http://localhost:3000
--- what the server answered ---
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: noFQMsd9ETwY/s8icZDZAVXvevU=
date: Fri, 06 Nov 2026 11:31:59 GMT
server: uvicorn
--- the check ---
base64(sha1(key + GUID)) = noFQMsd9ETwY/s8icZDZAVXvevU=
server's Sec-WebSocket-Accept matches: True
--- the same server, through the websockets library ---
sent 'hello', got back: echo: hello

What the output teaches

The request is plain HTTP with three special headers. Upgrade: websocket and Connection: Upgrade ask for the protocol switch, and Sec-WebSocket-Version: 13 names the protocol version. Sec-WebSocket-Key is a random value the client invents for each attempt. The server proves that it read the request by returning Sec-WebSocket-Accept, which is the SHA-1 hash of the key plus a fixed GUID from the RFC, in base64. The script computes that value itself and it matches. This is a handshake check, not security: anyone can compute it, and it says nothing about who the client is.

The Origin line is the one to remember. I added it by hand to play the part of a browser. A script like this one can write any value there, which is why MDN warns that “non-browser agents can send a faked Origin”. The header protects you from browsers that run someone else’s JavaScript, and from nothing else. RFC 6455 says so directly: “The intent is not to prevent non-browsers from establishing connections but rather to ensure that trusted browsers under the control of potentially malicious JavaScript cannot fake a WebSocket handshake.” Your real authentication has to live somewhere else, and Step 4 puts it there. (The key, the accept value and the date change on every run; everything else is fixed.)

Step 2: Build the notice board the way most tutorials do

The app is a notice board. Users sign in, open a socket, and receive live notices. They can also post a notice to a public board. Each user has a private API key that the server sends over the socket when it connects, which stands in for any private data your app pushes to a user.

This first version is the shape you will find in most tutorials. The connection manager follows the pattern in the FastAPI WebSockets docs: it remembers every socket and loops over them, awaiting send_text for each one. Login sets a session cookie, and the socket authenticates with that cookie. Read it once, then look for what is missing.

"""app_v1.py: the notice board the way most tutorials build it. Cookie login, no Origin check, no limits."""
import json
import secrets

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, JSONResponse

app = FastAPI()

USERS = {"alice": "sk-alice-7f3a91", "bob": "sk-bob-22c4d0"}  # user -> private API key
SESSIONS: dict[str, str] = {}  # session id -> user
NOTICES: list[dict] = []  # the public notice board


class ConnectionManager:
    """The pattern from the FastAPI docs: remember every socket, loop over them to send."""

    def __init__(self):
        self.active: list[tuple[str, WebSocket]] = []

    def connect(self, user: str, websocket: WebSocket):
        self.active.append((user, websocket))

    def disconnect(self, user: str, websocket: WebSocket):
        self.active.remove((user, websocket))

    async def send_to_user(self, user: str, text: str):
        for who, websocket in list(self.active):
            if who == user:
                await websocket.send_text(text)


manager = ConnectionManager()


@app.get("/login")
def login(user: str, samesite: str = "lax"):
    """Lab shortcut: a GET that signs you in, so a browser tab can get the cookie in one navigation."""
    session_id = secrets.token_urlsafe(16)
    SESSIONS[session_id] = user
    response = JSONResponse({"ok": True, "user": user})
    response.set_cookie("sid", session_id, httponly=True, samesite=samesite, secure=(samesite == "none"))
    return response


@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
    user = SESSIONS.get(websocket.cookies.get("sid", ""))
    if user is None:
        await websocket.close(code=1008)
        return
    # NOTE: nothing here looks at the Origin header yet
    await websocket.accept()
    manager.connect(user, websocket)
    await websocket.send_json({"type": "hello", "user": user, "api_key": USERS[user]})
    try:
        while True:
            message = json.loads(await websocket.receive_text())
            if message["type"] == "post":
                NOTICES.append({"by": user, "text": message["text"]})
                await websocket.send_json({"type": "posted", "count": len(NOTICES)})
    except WebSocketDisconnect:
        manager.disconnect(user, websocket)


@app.post("/notify")
async def notify(user: str, count: int = 1, size: int = 0):
    """Lab stand-in for 'something happened for this user': push count messages to all their sockets."""
    padding = "x" * size
    for n in range(count):
        await manager.send_to_user(user, json.dumps({"type": "notice", "n": n, "pad": padding}))
    return {"sent": count}


@app.get("/notices")
def notices():
    return NOTICES


@app.get("/stats")
def stats():
    return {"connections": len(manager.active)}


@app.get("/probe.html")
def probe():
    return FileResponse("site/probe.html")

What each part does

• USERS and SESSIONS are in-memory stand-ins for a database and a session store.
• login is a lab shortcut: a GET that signs you in, so a browser tab can receive the cookie in one navigation. The samesite query parameter lets the experiments try lax, strict and none. A cookie marked none must also be marked secure, and Edge accepted that combination over plain HTTP on localhost.
• ws_endpoint reads the sid cookie from the handshake, closes with code 1008 when there is no session, accepts otherwise, sends a hello message with the user’s API key, and then reads messages. A post message adds a notice to the board as that user.
• The comment # NOTE: nothing here looks at the Origin header yet marks the bug this tutorial fixes first.
• notify pushes count messages to every socket a user has open. In a real app this would be your event source, and it is the endpoint Step 6 uses to flood a reader.

Common mistake: closing the socket after accept() to reject a client. Closing before accept() is better, because the server then refuses the handshake itself. The ASGI specification says that if a close is sent before the socket is accepted, “the server must close the connection with a HTTP 403 error code (Forbidden), and not complete the WebSocket handshake”. That is exactly what RFC 6455 asks for on a bad origin, and the experiments below show the 403.

Step 3: Hijack it from another website

This is the attack that the CWE catalog describes as a product that “uses a WebSocket, but it does not properly verify that the source of data or communication is valid”. PortSwigger names it cross-site WebSocket hijacking and defines it as a cross-site request forgery vulnerability “on a WebSocket handshake”. It arises, PortSwigger says, “when the WebSocket handshake request relies solely on HTTP cookies for session handling and does not contain any CSRF tokens or other unpredictable values”, and then “the application will handle the connection in the context of the victim user’s session”.

3a. The probe page

The probe is a static page. Its run() function opens a socket to whatever server you name, reads the first message, posts a notice, and reports what happened. The same file is served from three different origins, so the only thing that changes between experiments is who is asking.

<!doctype html>
<!-- site/probe.html -->
<meta charset="utf-8">
<title>probe</title>
<p>probe page: open the console or call run() from a test.</p>
<script>
// run({api, mode}) opens a WebSocket to `api` from whatever origin served this page and reports what happened.
// mode "cookie": just connect (the browser adds cookies by itself).
// mode "ticket": ask POST /ws-ticket for a one-time ticket first, then connect with ?ticket=...
async function run(opts) {
  const result = {page: location.origin, ticket: null, opened: false, closeCode: null, hello: null, posted: null};
  let url = opts.api.replace("http", "ws") + "/ws";
  if (opts.mode === "ticket") {
    try {
      const reply = await fetch(opts.api + "/ws-ticket", {method: "POST", credentials: "include"});
      if (reply.ok) {
        const body = await reply.json();
        result.ticket = "ok";
        url += "?ticket=" + body.ticket;
      } else {
        result.ticket = "HTTP " + reply.status;
      }
    } catch (error) {
      result.ticket = "blocked by the browser";
    }
  }
  return await new Promise((resolve) => {
    const socket = new WebSocket(url);
    const timer = setTimeout(() => { try { socket.close(); } catch (e) {} resolve(result); }, 2500);
    socket.onopen = () => {
      result.opened = true;
      socket.send(JSON.stringify({type: "post", text: "posted from " + location.origin}));
    };
    socket.onmessage = (event) => {
      const message = JSON.parse(event.data);
      if (message.type === "hello") result.hello = message;
      if (message.type === "posted") { result.posted = message; socket.close(); }
    };
    socket.onclose = (event) => { result.closeCode = event.code; clearTimeout(timer); resolve(result); };
  });
}
</script>

3b. Run the experiment in a real browser

The script starts the servers you name on the command line, each on its own port: v1 is app_v1.py on 8101, v2 is a copy with an Origin check on 8102 (you will create it in a moment) and v3 is the hardened app on 8103 (Steps 4 to 6). It then drives Edge through Playwright. For each case a fresh browser profile signs in as alice in a normal tab and then visits the probe page on one of the three origins. Playwright launches the Edge that is already installed, so there is no browser download.

"""browser_matrix.py: open a WebSocket to the notice board from three kinds of page in a real Edge.

Needs: pip install playwright (it drives the Edge that is already installed, no browser download).
The three pages: the board's own page, a sibling origin (same site, other port) and an unrelated site."""
import os
import sys

import requests
from playwright.sync_api import sync_playwright

import labkit

SITE = os.path.join(labkit.HERE, "site")
SIBLING = "http://localhost:3000"  # same site as the board (the port is ignored), but a different origin
OTHER = "http://127.0.0.1:4000"  # a different host, so a different site
PAGES = [("own page", None), ("sibling origin", SIBLING), ("other site", OTHER)]
PORTS = {"v1": 8101, "v2": 8102, "v3": 8103}


def main(versions: list[str]) -> None:
    labkit.serve_static(3000, SITE)
    labkit.serve_static(4000, SITE)
    quiet = ("--log-level", "warning")
    launch = {
        "v1": lambda: labkit.start("app_v1:app", PORTS["v1"], *quiet),
        "v2": lambda: labkit.start("app_v2:app", PORTS["v2"], *quiet, env={"APP_ORIGIN": f"http://localhost:{PORTS['v2']}"}),
        "v3": lambda: labkit.start("app:app", PORTS["v3"], *quiet, "--ws-max-size", "65536", env={"APP_ORIGIN": f"http://localhost:{PORTS['v3']}"}),
    }
    servers = [launch[version]() for version in versions]
    cases = []
    for version in ("v1", "v2"):
        if version in versions:
            for samesite in ("lax", "strict", "none"):
                cases += [(version, samesite, page, "cookie") for page in PAGES]
    if "v3" in versions:
        for samesite in ("lax", "none"):
            cases += [("v3", samesite, page, "ticket") for page in PAGES]
    print(f"{'server':<7}{'cookie':<8}{'page':<16}{'opened':<8}{'api key seen':<14}{'posted as alice':<17}{'ticket fetch':<26}{'close'}")
    try:
        with sync_playwright() as p:
            browser = p.chromium.launch(channel="msedge", headless=True)
            print("browser:", browser.version, file=sys.stderr)
            for version, samesite, (label, origin), mode in cases:
                api = f"http://localhost:{PORTS[version]}"
                context = browser.new_context()
                page = context.new_page()
                page.goto(f"{api}/login?user=alice&samesite={samesite}")  # the user signs in on the board in a normal tab
                before = len(requests.get(f"{api}/notices", timeout=5).json())
                page.goto(f"{origin or api}/probe.html")
                result = page.evaluate("options => run(options)", {"api": api, "mode": mode})
                notices = requests.get(f"{api}/notices", timeout=5).json()
                posted = len(notices) > before and notices[-1]["by"] == "alice"
                key = bool(result["hello"] and result["hello"].get("api_key"))
                print(f"{version:<7}{samesite:<8}{label:<16}{str(result['opened']):<8}{str(key):<14}{str(posted):<17}"
                      f"{str(result['ticket'] or '-'):<26}{result['closeCode']}")
                context.close()
            browser.close()
    finally:
        for server in servers:
            labkit.stop(server)


if __name__ == "__main__":
    main(sys.argv[1:] or ["v1", "v2", "v3"])  # for example: python browser_matrix.py v1

Run it against the first server only:

python browser_matrix.py v1

server cookie  page            opened  api key seen  posted as alice  ticket fetch              close
v1     lax     own page        True    True          True             -                         1005
v1     lax     sibling origin  True    True          True             -                         1005
v1     lax     other site      False   False         False            -                         1006
v1     strict  own page        True    True          True             -                         1005
v1     strict  sibling origin  True    True          True             -                         1005
v1     strict  other site      False   False         False            -                         1006
v1     none    own page        True    True          True             -                         1005
v1     none    sibling origin  True    True          True             -                         1005
v1     none    other site      True    True          True             -                         1005

3c. Read the table

Each row is one browser session. opened says whether the socket opened, api key seen says whether the page received the private key, and posted as alice says whether the notice the page sent appeared on the board under alice’s name. The close column is the code Edge reported.

All nine rows come from the v1 server, which has no Origin check:

• The board’s own page works with every cookie setting. That is the legitimate case.
• The sibling origin at localhost:3000 opened the socket, read the API key and posted as alice with lax, strict and none. The page is cross-origin but same-site, so the browser treated the handshake as same-site and sent even the strict cookie. This is the row that surprises people: SameSite does nothing against a sibling subdomain or another app on the same registrable domain.
• The unrelated site at 127.0.0.1:4000 was stopped by lax and strict (the handshake went out without the cookie, the server closed it, and Edge reported 1006) and was not stopped by none. So the cookie attribute is a real defense against other sites, but only against other sites, and only when it is set. If you leave it out you depend on the browser’s default, and MDN says only that “some browsers” default to Lax.

No CORS header was involved at any point: app_v1.py has no CORS configuration at all, and the page on 127.0.0.1:4000 still opened a socket to another origin. The only thing between that page and the user’s data was a cookie flag.

3d. Fix it: check the Origin before accept()

The fix is a few lines in the handshake. Compare the Origin header with an allowlist and refuse anything else before accepting. Here is what make_versions.py added to the handler, with the allowlist read from an environment variable so each lab server can name its own origin:

# excerpt of app_v2.py: the new lines in the /ws handler, ahead of accept()
    if websocket.headers.get("origin") not in ALLOWED_ORIGINS:
        await websocket.close(code=1008)  # before accept(): the browser sees an HTTP 403
        return

Create app_v2.py and run the same browser check against it. make_versions.py derives the copy from app_v1.py with three exact text swaps, so you can see precisely which lines differ.

"""make_versions.py: derive app_v2.py (v1 plus an Origin allowlist) from app_v1.py with two exact swaps."""
import os

HERE = os.path.dirname(os.path.abspath(__file__))
src = open(os.path.join(HERE, "app_v1.py"), encoding="utf-8").read()

swaps = [
    ("app = FastAPI()\n",
     "app = FastAPI()\n\nALLOWED_ORIGINS = {os.environ.get(\"APP_ORIGIN\", \"http://localhost:8000\")}\n"),
    ("    # NOTE: nothing here looks at the Origin header yet\n",
     "    if websocket.headers.get(\"origin\") not in ALLOWED_ORIGINS:\n"
     "        await websocket.close(code=1008)  # before accept(): the browser sees an HTTP 403\n"
     "        return\n"),
    ("import json\nimport secrets\n", "import json\nimport os\nimport secrets\n"),
]
for old, new in swaps:
    assert src.count(old) == 1, old
    src = src.replace(old, new)
src = src.replace('"""app_v1.py: the notice board the way most tutorials build it. Cookie login, no Origin check, no limits."""',
                  '"""app_v2.py: app_v1.py plus an exact Origin allowlist on the handshake. Everything else is unchanged."""')
with open(os.path.join(HERE, "app_v2.py"), "w", encoding="utf-8", newline="\n") as f:
    f.write(src)
print("wrote app_v2.py", len(src), "chars")

python make_versions.py

python browser_matrix.py v2

server cookie  page            opened  api key seen  posted as alice  ticket fetch              close
v2     lax     own page        True    True          True             -                         1005
v2     lax     sibling origin  False   False         False            -                         1006
v2     lax     other site      False   False         False            -                         1006
v2     strict  own page        True    True          True             -                         1005
v2     strict  sibling origin  False   False         False            -                         1006
v2     strict  other site      False   False         False            -                         1006
v2     none    own page        True    True          True             -                         1005
v2     none    sibling origin  False   False         False            -                         1006
v2     none    other site      False   False         False            -                         1006

The board’s own page still works in all three cookie modes, and both foreign pages are refused in all three, including the sibling origin that SameSite could not stop. The refusal is an HTTP 403 during the handshake, which Edge reports as 1006.

3e. What a correct Origin check looks like

• Compare the whole string, exactly. The header holds a scheme, a host and a port, and nothing else. Do not use startswith, endswith or in on the text: http://localhost:8000.evil.example starts with the real origin. The tests in Step 9 try that value, a one-character longer port, the wrong scheme, the literal string null (which sandboxed pages and local files send) and an empty value.
• Keep the list short and written down. One origin per front end you ship, read from configuration, not built from the request.
• Decide what a missing header means. Browsers always send it. Scripts, phones and server-to-server clients often do not. MDN notes that “Most applications reject requests without this header”. Whether you allow it depends on what authenticates the connection. With a cookie it is safest to reject it. With a one-time ticket (Step 4) a missing header is harmless, because there is no ambient credential for a browser to abuse.
• Remember what the check is. It is a defense against browsers, not an identity check. A script can send any value, as Step 1 showed.

Step 4: Stop trusting the cookie, and use one-time tickets

The Origin check closed the hole for browsers, but the socket still authenticates with an ambient cookie, which is the exact condition PortSwigger names: a handshake that “relies solely on HTTP cookies for session handling”. If the check has a bug, a proxy rewrites the header, or a future change loosens the allowlist, the cookie is again all an attacker needs. So give the socket its own credential, one that a foreign page cannot obtain and a log line cannot leak.

The credential is a ticket. The page asks for one with an ordinary authenticated POST, receives a random value that is good for 30 seconds and for one use, and puts it in the handshake URL. Because the ticket travels in the URL, it is not a secret you can afford to keep, and that is fine: by the time it appears in a log it has already been spent.

4a. The hardened app, part one: configuration, origin check, tickets and login

Create app.py from this block and the ones in Steps 5 and 6. This first block holds the settings, the Origin check from Step 3, the ticket store and the two HTTP endpoints that sign you in and hand out tickets. A few details are worth reading closely.

• origin_ok accepts a missing header and requires an exact match when the header is present. The reason a missing header is acceptable here is that the ticket, not the browser, is now the credential.
• Only the SHA-256 of a ticket is stored, so a memory dump or a debug print of the table is not a list of usable tickets. issue_ticket also sweeps out expired entries, which keeps the table from filling with tickets nobody used.
• redeem_ticket uses pop, not get. Taking the entry out is what makes a ticket single-use, and in a single-threaded event loop the pop cannot race with another handshake.
• Every handler is async def on purpose. The FastAPI docs say that a path operation function declared with plain def “is run in an external threadpool that is then awaited”, and the dictionaries, sets and queues in this app are only safe when one thread, the event loop, touches them.
• ws_ticket applies the same Origin check as the socket. A foreign page cannot read the reply anyway, but there is no reason to hand it a ticket.
• Two lines exist only for the lab: the LAB_SESSION seed (a fixed session that survives a restart, the way a shared session store would, used in Step 8) and the OUTBOX_MAX environment variable (Step 6 sets it to 0 to show what an unbounded queue costs).

"""app.py: the hardened notice board (FastAPI 0.162.0, Starlette 1.10.0, uvicorn 0.54.0, Python 3.13).

Ticket login, an exact Origin allowlist, validated messages, size and rate limits, a connection cap,
and a bounded outbox per client so that one slow reader cannot stall everyone else."""
import asyncio
import hashlib
import json
import logging
import os
import secrets
import time
from dataclasses import dataclass
from typing import Annotated, Literal

from fastapi import FastAPI, HTTPException, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import FileResponse, JSONResponse
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError

log = logging.getLogger("noticeboard")

APP_ORIGIN = os.environ.get("APP_ORIGIN", "http://localhost:8000")
ALLOWED_ORIGINS = {APP_ORIGIN}
TICKET_TTL = float(os.environ.get("TICKET_TTL", "30"))  # seconds a ticket stays valid
MAX_MESSAGE_CHARS = 2000
OUTBOX_MAX = int(os.environ.get("OUTBOX_MAX", "100"))  # 0 would mean unbounded; the lab uses it to show why not
MAX_CONNECTIONS_PER_USER = 3
RATE_BURST = 20
RATE_PER_SECOND = 5.0

app = FastAPI()

USERS = {"alice": "sk-alice-7f3a91", "bob": "sk-bob-22c4d0"}
SESSIONS: dict[str, str] = {}  # session id -> user
NOTICES: list[dict] = []
if os.environ.get("LAB_SESSION"):  # lab only: a fixed session that survives a restart, like a shared session store would
    SESSIONS[os.environ["LAB_SESSION"]] = "alice"


# --- 1. Who may open a socket: Origin allowlist and one-time tickets -------------------------------------------

def origin_ok(origin: str | None) -> bool:
    """Browsers always send Origin on a WebSocket handshake. Scripts and phones often send none, and with no
    browser involved there is no ambient cookie to abuse. A header that IS present must match exactly."""
    return origin is None or origin in ALLOWED_ORIGINS


@dataclass
class Ticket:
    user: str
    expires: float


TICKETS: dict[str, Ticket] = {}  # sha256(ticket) -> Ticket


def _key(raw: str) -> str:
    return hashlib.sha256(raw.encode()).hexdigest()


def issue_ticket(user: str) -> str:
    now = time.monotonic()
    for stale in [key for key, ticket in TICKETS.items() if ticket.expires < now]:
        del TICKETS[stale]  # keep the table from growing with tickets nobody used
    raw = secrets.token_urlsafe(32)
    TICKETS[_key(raw)] = Ticket(user, now + TICKET_TTL)
    return raw


def redeem_ticket(raw: str | None) -> str | None:
    """Return the user a ticket belongs to, and burn the ticket either way."""
    if not raw:
        return None
    ticket = TICKETS.pop(_key(raw), None)  # pop, not get: a ticket works exactly once
    if ticket is None or ticket.expires < time.monotonic():
        return None
    return ticket.user


# Every handler below is async def on purpose. FastAPI runs a plain def handler in a thread pool, and
# the dicts, sets and asyncio queues here are only safe when ONE thread (the event loop) touches them.
@app.get("/login")
async def login(user: str, samesite: str = "lax"):
    """Lab shortcut: a GET that signs you in, so a browser tab can get the cookie in one navigation."""
    session_id = secrets.token_urlsafe(16)
    SESSIONS[session_id] = user
    response = JSONResponse({"ok": True, "user": user})
    response.set_cookie("sid", session_id, httponly=True, samesite=samesite, secure=(samesite == "none"))
    return response


@app.post("/ws-ticket")
async def ws_ticket(request: Request):
    if not origin_ok(request.headers.get("origin")):
        raise HTTPException(status_code=403)
    user = SESSIONS.get(request.cookies.get("sid", ""))
    if user is None:
        raise HTTPException(status_code=401)
    return {"ticket": issue_ticket(user), "expires_in": TICKET_TTL}

4b. What the handshake does with them

The WebSocket handler appears in full in Step 6. These are the lines that run before it accepts anything. The order is deliberate: check the origin first, then redeem the ticket, then enforce the per-user connection cap. Every refusal happens before accept(), so the browser sees an HTTP 403 and nothing else, and the client is never told which check failed. The reason goes to your log, not to the caller.

# excerpt of app.py: the checks in ws_endpoint that run before accept()
    origin = websocket.headers.get("origin")
    if not origin_ok(origin):
        log.warning("websocket refused: origin %r", origin)
        await websocket.close(code=1008)  # before accept(): the client sees an HTTP 403
        return
    user = redeem_ticket(websocket.query_params.get("ticket"))
    if user is None or hub.count(user) >= MAX_CONNECTIONS_PER_USER:
        await websocket.close(code=1008)
        return

4c. Try it in the browser

Run the same experiment against the hardened app. The probe page now asks POST /ws-ticket for a ticket first and connects with it.

python browser_matrix.py v3

server cookie  page            opened  api key seen  posted as alice  ticket fetch              close
v3     lax     own page        True    False         True             ok                        1005
v3     lax     sibling origin  False   False         False            blocked by the browser    1006
v3     lax     other site      False   False         False            blocked by the browser    1006
v3     none    own page        True    False         True             ok                        1005
v3     none    sibling origin  False   False         False            blocked by the browser    1006
v3     none    other site      False   False         False            blocked by the browser    1006

The board’s own page works: the ticket request succeeds, the socket opens, and the notice is posted. The api key seen column is now False because this version no longer sends the key. For both foreign pages, with the cookie set to lax or to none, the ticket request was blocked by the browser, since the page is not allowed to read a reply from an origin that sent no CORS headers, and the socket was then refused. With none the browser may attach the cookie to the ticket request, which is the cookie setting that gave a foreign page the whole account in Step 3, but it still will not show the reply to that page. The server would also have answered 403, because the Origin header does not match.

4d. Abuse the tickets from a script

The browser only shows you the happy path. This script plays the careless and the hostile client in turn against the hardened app, which it starts with a two-second ticket lifetime so the expiry case does not make you wait.

"""step_ticket.py: one-time tickets in action: reuse, expiry, forged tickets, and what the access log records."""
import os
import re
import time

import requests
from websockets.exceptions import InvalidStatus
from websockets.sync.client import connect

import labkit

PORT = 8104
ORIGIN = f"http://localhost:{PORT}"
server = labkit.start("app:app", PORT, env={"APP_ORIGIN": ORIGIN, "TICKET_TTL": "2"}, log="uvicorn_ticket.log")  # default log level
session = requests.Session()
session.get(f"{ORIGIN}/login?user=alice", timeout=5)


def new_ticket() -> str:
    reply = session.post(f"{ORIGIN}/ws-ticket", headers={"Origin": ORIGIN}, timeout=5)
    return reply.json()["ticket"]


def try_connect(ticket: str) -> str:
    try:
        with connect(f"ws://localhost:{PORT}/ws?ticket={ticket}", additional_headers={"Origin": ORIGIN}, open_timeout=5) as ws:
            return "opened, first message " + ws.recv(timeout=3)
    except InvalidStatus as refused:
        return f"refused with HTTP {refused.response.status_code}"


try:
    first = new_ticket()
    print("fresh ticket:                ", try_connect(first))
    print("same ticket again (replay):  ", try_connect(first))
    old = new_ticket()
    time.sleep(2.5)
    print("ticket used after 2.5 s:     ", try_connect(old), "(lab TTL is 2 s)")
    print("ticket nobody issued:        ", try_connect("A" * 43))
    no_session = requests.post(f"{ORIGIN}/ws-ticket", headers={"Origin": ORIGIN}, timeout=5)
    print("ticket request, no cookie:   ", no_session.status_code)
    foreign = session.post(f"{ORIGIN}/ws-ticket", headers={"Origin": "http://127.0.0.1:4000"}, timeout=5)
    print("ticket request, foreign Origin:", foreign.status_code)
    print("ticket body keys:            ", sorted(session.post(f"{ORIGIN}/ws-ticket", headers={"Origin": ORIGIN}, timeout=5).json()))
finally:
    labkit.stop(server)

print("--- uvicorn access log, WebSocket lines (default log level) ---")
with open(os.path.join(labkit.HERE, "out", "uvicorn_ticket.log"), encoding="utf-8", errors="replace") as log:
    lines = [line.rstrip() for line in log if "WebSocket" in line]
for line in lines[:4]:
    print(re.sub(r"^INFO:\s+", "", line))

python step_ticket.py

fresh ticket:                 opened, first message {"type": "hello", "user": "alice"}
same ticket again (replay):   refused with HTTP 403
ticket used after 2.5 s:      refused with HTTP 403 (lab TTL is 2 s)
ticket nobody issued:         refused with HTTP 403
ticket request, no cookie:    401
ticket request, foreign Origin: 403
ticket body keys:             ['expires_in', 'ticket']
--- uvicorn access log, WebSocket lines (default log level) ---
::1:54166 - "WebSocket /ws?ticket=YYF0Nu6vCFWXGqoBAJQoyj4TRPLax6GH1Aaj6KVj9AY" [accepted]
::1:54172 - "WebSocket /ws?ticket=YYF0Nu6vCFWXGqoBAJQoyj4TRPLax6GH1Aaj6KVj9AY" 403
::1:54178 - "WebSocket /ws?ticket=jOK7-G97jr6OCh5t2PFVs_H5JaTwqBMGWCdzC-2gRfg" 403
::1:54184 - "WebSocket /ws?ticket=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" 403

What the output teaches

• A fresh ticket opens a socket. The same ticket a second time is refused with 403, as is a ticket used after it expired and a ticket the server never issued. All three look identical to the caller, which is what you want.
• Asking for a ticket without a session cookie gets 401, and asking with a foreign Origin gets 403.
• The access-log lines are the answer to “is a credential in a URL not a leak?”. uvicorn records the full path including the query string, so the first ticket is sitting in the log. The line after it shows the same ticket being refused. A logged ticket is a spent ticket. That is why the credential has to be a short-lived, single-use ticket and not a session ID or a JWT: those would stay valid for as long as they live, in your logs and in every proxy log between the browser and you.

Common mistakes: making the lifetime long “for slow networks” (the page asks for a ticket right before it connects, so seconds are enough); using get instead of pop so the ticket survives; and running several worker processes with an in-memory ticket table, where a ticket issued by one worker cannot be redeemed by another. With more than one worker you need a shared store that can take a ticket out in one atomic step. Browsers also cannot set an Authorization header on new WebSocket(), which is why a ticket in the query string, or a first message after connecting, is the usual pattern.

Step 5: Limit what a client can send

A signed-in client can still be a bad one: buggy, hostile, or a script stuck in a loop. Four limits each answer with a specific close code, so you can see in a log what happened.

5a. Message size, at two levels

There are two places to limit size, and they do different jobs. The transport limit protects memory: uvicorn’s --ws-max-size setting has a default of 16777216 bytes according to its settings page, and a message under that is read into memory in full before your code sees it. The application limit, MAX_MESSAGE_CHARS = 2000, expresses what a notice-board message is allowed to be. The script below sends one 15 MB message to two servers: one with the default transport limit and one started with --ws-max-size 65536.

5b. Message shape, and rate

The second block of app.py holds the pieces that decide whether a message is acceptable. Pydantic validates the text against a discriminated union of the two message types, and extra="forbid" means an unknown field is an error, not something silently ignored. Bucket is a token bucket: five messages a second on average, 20 at once. CloseWith is a small exception that carries the close code, and Step 6 shows how the connection turns it into a close frame.

# app.py (continued)
# --- 2. What a client may send: a schema, a size cap, a rate cap -----------------------------------------------

class Post(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["post"]
    text: str = Field(min_length=1, max_length=500)


class Ping(BaseModel):
    model_config = ConfigDict(extra="forbid")
    type: Literal["ping"]


ClientMessage = TypeAdapter(Annotated[Post | Ping, Field(discriminator="type")])


class Bucket:
    """Token bucket: RATE_PER_SECOND messages a second on average, RATE_BURST at once."""

    def __init__(self, burst: int, per_second: float):
        self.tokens, self.burst, self.rate, self.last = float(burst), burst, per_second, time.monotonic()

    def take(self) -> bool:
        now = time.monotonic()
        self.tokens = min(self.burst, self.tokens + (now - self.last) * self.rate)
        self.last = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False


class CloseWith(Exception):
    """Raised anywhere inside a connection to end it with this WebSocket close code."""

    def __init__(self, code: int):
        super().__init__(code)
        self.code = code

The checks run in a fixed order inside read_loop, which Step 6 shows in full: disconnect, then binary frames (1003), then size (1009), then rate (1008), then the schema (1008). Size comes before parsing so that an oversized message is never handed to the JSON parser, and rate comes before validation so that a flood of invalid messages cannot cost you CPU for every one of them.

# excerpt of app.py: the order of the checks in read_loop
async def read_loop(conn: Connection) -> None:
    while True:
        message = await conn.websocket.receive()
        if message["type"] == "websocket.disconnect":
            raise WebSocketDisconnect(message.get("code", 1005))
        text = message.get("text")
        if text is None:
            raise CloseWith(1003)  # binary frames are not part of this protocol
        if len(text) > MAX_MESSAGE_CHARS:
            raise CloseWith(1009)
        if not conn.bucket.take():
            raise CloseWith(1008)
        try:
            parsed = ClientMessage.validate_json(text)
        except ValidationError:
            raise CloseWith(1008) from None

5c. Run the limits

This script sends the oversized message, then 60 pings back to back, then opens four connections for the same user.

"""step_limits.py: what one signed-in client can do to the server, and what each limit costs it."""
import contextlib
import threading
import time

import requests
from websockets.exceptions import ConnectionClosed, InvalidStatus
from websockets.sync.client import connect

import labkit


def login_ticket(origin: str) -> str:
    session = requests.Session()
    session.get(f"{origin}/login?user=alice", timeout=5)
    return session.post(f"{origin}/ws-ticket", headers={"Origin": origin}, timeout=5).json()["ticket"]


def big_message(port: int, extra: list[str], megabytes: int) -> None:
    origin = f"http://localhost:{port}"
    server = labkit.start("app:app", port, "--log-level", "warning", *extra, env={"APP_ORIGIN": origin})
    try:
        ticket = login_ticket(origin)
        baseline = labkit.rss_mb(server)
        peak, stop = [baseline], threading.Event()

        def sample():
            while not stop.is_set():
                peak[0] = max(peak[0], labkit.rss_mb(server))
                time.sleep(0.005)

        threading.Thread(target=sample, daemon=True).start()
        code = None
        with connect(f"ws://localhost:{port}/ws?ticket={ticket}", additional_headers={"Origin": origin},
                     max_size=None, compression=None) as ws:
            ws.recv()
            started = time.perf_counter()
            ws.send("x" * (megabytes * 1_000_000))
            try:
                while True:
                    ws.recv(timeout=10)
            except ConnectionClosed as closed:
                code = closed.rcvd.code if closed.rcvd else None
            took = time.perf_counter() - started
        stop.set()
        flag = " ".join(extra) or "(default --ws-max-size 16777216)"
        print(f"{flag:<40} sent {megabytes} MB -> closed {code} after {took:4.2f} s, server memory {baseline:5.1f} -> peak {peak[0]:5.1f} MiB")
    finally:
        labkit.stop(server)


big_message(8105, [], 15)
big_message(8106, ["--ws-max-size", "65536"], 15)

# message rate and connection cap, on the hardened app
port, origin = 8107, "http://localhost:8107"
server = labkit.start("app:app", port, "--log-level", "warning", "--ws-max-size", "65536", env={"APP_ORIGIN": origin})
try:
    with connect(f"ws://localhost:{port}/ws?ticket={login_ticket(origin)}", additional_headers={"Origin": origin}) as ws:
        ws.recv()
        for _ in range(60):
            ws.send('{"type": "ping"}')
        pongs, code = 0, None
        try:
            while True:
                ws.recv(timeout=5)
                pongs += 1
        except ConnectionClosed as closed:
            code = closed.rcvd.code if closed.rcvd else None
    print(f"60 pings sent back to back -> {pongs} answered, then closed {code}")

    with contextlib.ExitStack() as stack:
        for attempt in range(1, 5):
            try:
                stack.enter_context(connect(f"ws://localhost:{port}/ws?ticket={login_ticket(origin)}", additional_headers={"Origin": origin}, compression=None))
                print(f"connection {attempt} for alice: open")
            except InvalidStatus as refused:
                print(f"connection {attempt} for alice: refused with HTTP {refused.response.status_code}")
finally:
    labkit.stop(server)

python step_limits.py

(default --ws-max-size 16777216)         sent 15 MB -> closed 1009 after 0.09 s, server memory  53.0 -> peak  78.9 MiB
--ws-max-size 65536                      sent 15 MB -> closed None after 0.03 s, server memory  52.9 -> peak  53.1 MiB
60 pings sent back to back -> 20 answered, then closed 1008
connection 1 for alice: open
connection 2 for alice: open
connection 3 for alice: open
connection 4 for alice: refused with HTTP 403

What the output teaches

• With the default transport limit, the 15 MB message was read in full. Our application check answered it with 1009 after 0.09 seconds, and the server’s memory went from 53.0 to a peak of 78.9 MiB while it held the message. With --ws-max-size 65536 the library refused the frame and memory did not move: 52.9 to 53.1 MiB. The client saw no close code at all (None) because the server hung up while the client was still uploading, and a browser would report that as 1006. I sampled memory every five milliseconds from outside the process, so read the peaks as indicative and the difference as real.
• 60 pings in a burst got 20 answers, which is the bucket’s burst size, and then the connection was closed with 1008.
• The fourth connection for alice was refused with a 403 at the handshake, so a single user cannot hold unlimited tasks and buffers.

Common mistake: checking the size in the application and assuming that protects memory. By the time your code runs, the library has already buffered the whole message. Set the transport limit to the smallest value your protocol needs.

Step 6: Deal with readers that are slow or gone

Everything so far was about who connects and what they send. This step is about what you send to them, and it is where most WebSocket servers fail in production.

6a. One client that stops reading

The ConnectionManager in the FastAPI WebSockets docs sends a message to every client with a loop that awaits send_text for each connection one after another, and app_v1.py uses the same kind of loop. Now give it two connections for the same user, one healthy and one that completes the handshake and then never reads another byte, and push 3,000 notices of about 2 KB each. The stalled client is a raw socket with a deliberately tiny receive buffer so it fills up quickly. The script runs this against the naive app, once with the stalled client connected first and once with it second, and against the hardened app, and then it measures memory for a reader that never reads.

"""step_slow.py: one client that stops reading, and what it does to everyone else."""
import threading
import time

import requests
from websockets.exceptions import ConnectionClosed
from websockets.sync.client import connect

import labkit

COUNT, SIZE = 3000, 2000  # 3000 notices of about 2 KB each, about 6 MB in total


def healthy_reader(url: str, headers: dict, arrivals: list, ready: threading.Event, stop: threading.Event) -> None:
    with connect(url, additional_headers=headers, compression=None, max_size=None) as ws:
        ws.recv()  # hello (or the first message)
        ready.set()
        while not stop.is_set():
            try:
                ws.recv(timeout=0.5)
            except TimeoutError:
                continue
            except ConnectionClosed:
                break
            arrivals.append(time.perf_counter())


def run_case(label: str, order: str, version: str) -> None:
    port = 8101 if version == "v1" else 8103
    origin = f"http://localhost:{port}"
    app = "app_v1:app" if version == "v1" else "app:app"
    server = labkit.start(app, port, "--log-level", "warning", env={"APP_ORIGIN": origin})

    def credentials():
        """v1 wants a cookie, the hardened app wants a one-time ticket and an Origin header."""
        session = requests.Session()
        session.get(f"{origin}/login?user=alice", timeout=5)
        if version == "v1":
            return "/ws", {"Cookie": f"sid={session.cookies.get('sid')}"}
        ticket = session.post(f"{origin}/ws-ticket", headers={"Origin": origin}, timeout=5).json()["ticket"]
        return f"/ws?ticket={ticket}", {"Origin": origin}

    arrivals, stop, ready = [], threading.Event(), threading.Event()
    stalled_socket = None

    def open_stalled():
        nonlocal stalled_socket
        path, headers = credentials()
        stalled_socket = labkit.raw_handshake(port, path, headers=headers, rcvbuf=2048)[0]  # tiny receive buffer, never reads

    def open_healthy():
        path, headers = credentials()
        thread = threading.Thread(target=healthy_reader, args=(f"ws://localhost:{port}{path}", headers, arrivals, ready, stop), daemon=True)
        thread.start()
        ready.wait(5)

    try:
        for who in (order.split("+")):
            (open_stalled if who == "stalled" else open_healthy)()
        outcome = {}

        def push():
            started = time.perf_counter()
            try:
                reply = requests.post(f"{origin}/notify", params={"user": "alice", "count": COUNT, "size": SIZE}, timeout=8)
                outcome["returned"] = f"{time.perf_counter() - started:.2f} s"
                outcome["body"] = reply.json()
            except requests.exceptions.ReadTimeout:
                outcome["returned"] = "still waiting after 8 s"

        t0 = time.perf_counter()
        pusher = threading.Thread(target=push)
        pusher.start()
        pusher.join(10)
        time.sleep(1.0)
        stop.set()
        last = f"{max(arrivals) - t0:.2f} s" if arrivals else "-"
        stats = requests.get(f"{origin}/stats", timeout=5).json()
        print(f"{label:<28}{order:<20}notify returned: {outcome['returned']:<24}healthy client got {len(arrivals):>4} of {COUNT}"
              f" (last at {last}), server stats {stats}")
    finally:
        labkit.stop(server)
        if stalled_socket:
            stalled_socket.close()


def memory_case(outbox_max: int) -> None:
    port = 8108 + (1 if outbox_max else 0)
    origin = f"http://localhost:{port}"
    server = labkit.start("app:app", port, "--log-level", "warning", env={"APP_ORIGIN": origin, "OUTBOX_MAX": str(outbox_max)})
    try:
        session = requests.Session()
        session.get(f"{origin}/login?user=alice", timeout=5)
        ticket = session.post(f"{origin}/ws-ticket", headers={"Origin": origin}, timeout=5).json()["ticket"]
        stalled = labkit.raw_handshake(port, f"/ws?ticket={ticket}", headers={"Origin": origin}, rcvbuf=2048)[0]
        before = labkit.rss_mb(server)
        reply = requests.post(f"{origin}/notify", params={"user": "alice", "count": 50000, "size": 2000}, timeout=60).json()
        time.sleep(1.5)
        stats = requests.get(f"{origin}/stats", timeout=5).json()
        label = "unbounded (OUTBOX_MAX=0)" if outbox_max == 0 else f"bounded (OUTBOX_MAX={outbox_max})"
        print(f"{label:<28}50,000 notices of 2 KB to one reader that never reads: server memory {before:6.1f} -> {labkit.rss_mb(server):6.1f} MiB, stats {stats}")
        stalled.close()
    finally:
        labkit.stop(server)


print("--- one stalled reader and one healthy reader, same user ---")
run_case("naive loop (app_v1)", "stalled+healthy", "v1")
run_case("naive loop (app_v1)", "healthy+stalled", "v1")
run_case("bounded outbox (app)", "stalled+healthy", "v3")
print("--- memory held for a reader that never reads ---")
memory_case(0)
memory_case(100)

python step_slow.py

--- one stalled reader and one healthy reader, same user ---
naive loop (app_v1)         stalled+healthy     notify returned: still waiting after 8 s healthy client got   68 of 3000 (last at 0.00 s), server stats {'connections': 2}
naive loop (app_v1)         healthy+stalled     notify returned: still waiting after 8 s healthy client got   69 of 3000 (last at 0.01 s), server stats {'connections': 2}
bounded outbox (app)        stalled+healthy     notify returned: 0.07 s                  healthy client got 3000 of 3000 (last at 0.07 s), server stats {'connections': 1, 'evicted': 1}
--- memory held for a reader that never reads ---
unbounded (OUTBOX_MAX=0)    50,000 notices of 2 KB to one reader that never reads: server memory   52.9 ->  158.3 MiB, stats {'connections': 1, 'evicted': 0}
bounded (OUTBOX_MAX=100)    50,000 notices of 2 KB to one reader that never reads: server memory   53.0 ->   53.4 MiB, stats {'connections': 0, 'evicted': 1}

What the output teaches

• With the naive loop the healthy client received 68 of 3,000 notices and then nothing, and the HTTP request that triggered the broadcast was still waiting after eight seconds. The order of the two clients did not matter: 68 when the stalled one connected first and 69 when it connected second.
• The reason is in uvicorn 0.54.0’s source. In websockets_sansio_impl.py, pause_writing clears an event when the transport’s write buffer passes its high-water mark, and send begins with await self.writable.wait(). A reader that never reads fills the kernel’s buffers and then that buffer, so one await never finishes and the loop never reaches the next connection. One bad client froze every other client of the same user, and also froze whatever HTTP handler was doing the broadcasting.
• The other way to fail is to stop waiting and queue everything. With an unbounded queue per client, 50,000 notices of 2 KB to a reader that never reads took the server from 52.9 to 158.3 MiB. With a bounded outbox it stayed at 53.0 to 53.4 MiB.

6b. The fix: one bounded outbox per connection

The design has three rules. Producers never wait: they call offer(), which puts a message in the connection’s queue or reports that the queue is full. Exactly one task per connection, pump(), sends from that queue and is the only place that waits for the network. And a full queue is a decision, not an accident: offer() sets an event, a watcher task raises CloseWith(1013), and the connection is closed with “Try Again Later” so the client can reconnect and catch up.

This block holds Connection, Hub, the goodbye helper and read_loop. Hub.send_to_user is a plain function with no await in it, which is the property that makes it impossible for one reader to hold up a broadcast.

# app.py (continued)
# --- 3. Who waits for whom: one bounded outbox per connection ----------------------------------------------------

class Connection:
    def __init__(self, websocket: WebSocket, user: str):
        self.websocket, self.user = websocket, user
        self.outbox: asyncio.Queue[str] = asyncio.Queue(maxsize=OUTBOX_MAX)
        self.slow = asyncio.Event()
        self.bucket = Bucket(RATE_BURST, RATE_PER_SECOND)

    def offer(self, text: str) -> bool:
        """Queue a message without ever waiting. False means this client cannot keep up."""
        try:
            self.outbox.put_nowait(text)
            return True
        except asyncio.QueueFull:
            self.slow.set()
            return False

    async def pump(self) -> None:
        """The only place that waits for the network: send queued messages in order."""
        while True:
            await self.websocket.send_text(await self.outbox.get())

    async def watch(self) -> None:
        await self.slow.wait()
        raise CloseWith(1013)  # Try Again Later: reconnect and catch up


class Hub:
    def __init__(self):
        self.by_user: dict[str, set[Connection]] = {}
        self.evicted = 0

    def count(self, user: str) -> int:
        return len(self.by_user.get(user, ()))

    def add(self, conn: Connection) -> None:
        self.by_user.setdefault(conn.user, set()).add(conn)

    def remove(self, conn: Connection) -> None:
        self.by_user.get(conn.user, set()).discard(conn)

    def send_to_user(self, user: str, payload: dict) -> int:
        """Hand a message to every socket the user has open. Never waits, so it cannot be held up by one reader."""
        text = json.dumps(payload)
        return sum(1 for conn in list(self.by_user.get(user, ())) if conn.offer(text))


hub = Hub()


async def close_quietly(websocket: WebSocket, code: int) -> None:
    """Try to say goodbye, but never wait on a client that is not reading."""
    try:
        await asyncio.wait_for(websocket.close(code=code), timeout=1.0)
    except (TimeoutError, RuntimeError):
        pass


async def read_loop(conn: Connection) -> None:
    while True:
        message = await conn.websocket.receive()
        if message["type"] == "websocket.disconnect":
            raise WebSocketDisconnect(message.get("code", 1005))
        text = message.get("text")
        if text is None:
            raise CloseWith(1003)  # binary frames are not part of this protocol
        if len(text) > MAX_MESSAGE_CHARS:
            raise CloseWith(1009)
        if not conn.bucket.take():
            raise CloseWith(1008)
        try:
            parsed = ClientMessage.validate_json(text)
        except ValidationError:
            raise CloseWith(1008) from None
        if isinstance(parsed, Ping):
            conn.offer(json.dumps({"type": "pong"}))
        else:
            NOTICES.append({"by": conn.user, "text": parsed.text})
            hub.send_to_user(conn.user, {"type": "posted", "count": len(NOTICES)})

The handler runs three tasks inside an asyncio.TaskGroup: the pump, the slow-consumer watcher and the read loop. When any of them ends or raises, the group cancels the others, and except* turns a CloseWith into the right close frame. If TaskGroup and except* are new to you, the asyncio TaskGroup tutorial covers them step by step. The helper close_quietly gives the goodbye one second and then gives up. That is not a nicety: close() goes through the same send that waits on the write buffer, so saying goodbye to a client that is not reading would hang for as long as the client stays silent.

# app.py (continued)
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket):
    origin = websocket.headers.get("origin")
    if not origin_ok(origin):
        log.warning("websocket refused: origin %r", origin)
        await websocket.close(code=1008)  # before accept(): the client sees an HTTP 403
        return
    user = redeem_ticket(websocket.query_params.get("ticket"))
    if user is None or hub.count(user) >= MAX_CONNECTIONS_PER_USER:
        await websocket.close(code=1008)
        return
    await websocket.accept()
    conn = Connection(websocket, user)
    hub.add(conn)
    conn.offer(json.dumps({"type": "hello", "user": user}))
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(conn.pump())
            group.create_task(conn.watch())
            group.create_task(read_loop(conn))
    except* CloseWith as problem:
        code = problem.exceptions[0].code
        if code == 1013:
            hub.evicted += 1
        await close_quietly(websocket, code)
    except* WebSocketDisconnect:
        pass
    finally:
        hub.remove(conn)

The last block holds the lab helpers: notify, which stands in for your event source, plus the stats, slow-request and page routes used by the experiments. notify yields to the event loop after each message with await asyncio.sleep(0), the way real events arrive one per turn of the loop. That line matters. If a producer offers more messages than the outbox holds before the loop gets a turn to run the pumps, even a healthy client overflows, and a test below pins that property.

# app.py (continued)
# --- 4. Lab helpers ----------------------------------------------------------------------------------------------

@app.post("/notify")
async def notify(user: str, count: int = 1, size: int = 0):
    """Lab stand-in for 'something happened for this user'. Returns how many sockets accepted the last message."""
    padding = "x" * size
    queued = 0
    for n in range(count):
        queued = hub.send_to_user(user, {"type": "notice", "n": n, "pad": padding})
        await asyncio.sleep(0)  # real events arrive one per loop turn; yield so the pumps can run in between
    return {"sent": count, "queued_last": queued}


@app.get("/notices")
async def notices():
    return NOTICES


@app.get("/stats")
async def stats():
    return {"connections": sum(len(group) for group in hub.by_user.values()), "evicted": hub.evicted}


@app.get("/slow")
async def slow(seconds: float = 3.0):
    await asyncio.sleep(seconds)
    return {"waited": seconds}


@app.get("/client.html")
async def client_page():
    return FileResponse("site/client.html")


@app.get("/probe.html")
async def probe():
    return FileResponse("site/probe.html")

What the earlier output shows for the hardened app

Look back at the third row of the output: the hardened app’s notify returned in 0.07 seconds, the healthy client received all 3,000 notices (the last one at 0.07 seconds), and the stats show one connection left and one eviction, so the stalled client was cut loose instead of holding everyone up. In the memory experiment the bounded server shows zero connections and one eviction afterwards, with memory flat at 53.4 MiB.

Your policy choices. Closing a slow client is the right default when the client can reconnect and re-sync, as in the browser page in Step 8. For a stream where only the latest value matters, such as a price or a cursor, you would instead drop the oldest queued message, or replace the queue with a single “latest value” slot. I implemented and tested only the closing policy here.

6c. Clients that go silent

A slow reader at least keeps its connection busy. A client that has simply gone away, a laptop that lost power or a phone that left coverage, sends nothing at all, and TCP can take a long time to notice. WebSocket has a built-in tool for this: RFC 6455 says that “a Ping frame may serve either as a keepalive or as a means to verify that the remote endpoint is still responsive”. uvicorn sends the pings for you. Its settings page lists --ws-ping-interval and --ws-ping-timeout, both with a default of 20.0 seconds.

This script connects a raw socket that never reads and never answers a ping, and times how long the server takes to notice, once with one-second settings and once with the defaults. It also checks that a normal client that does answer pings is not dropped.

"""step_dead.py: a client that goes silent without saying goodbye, and how long the server takes to notice."""
import threading
import time

import requests
from websockets.sync.client import connect

import labkit

results = {}


def watch(label: str, port: int, extra: list[str], limit: float) -> None:
    origin = f"http://localhost:{port}"
    server = labkit.start("app:app", port, "--log-level", "warning", *extra, env={"APP_ORIGIN": origin})
    try:
        session = requests.Session()
        session.get(f"{origin}/login?user=alice", timeout=5)
        ticket = session.post(f"{origin}/ws-ticket", headers={"Origin": origin}, timeout=5).json()["ticket"]
        silent = labkit.raw_handshake(port, f"/ws?ticket={ticket}", headers={"Origin": origin})[0]  # never reads, never answers
        started = time.perf_counter()
        while time.perf_counter() - started < limit:
            if requests.get(f"{origin}/stats", timeout=5).json()["connections"] == 0:
                results[label] = f"server noticed after {time.perf_counter() - started:5.1f} s"
                break
            time.sleep(0.25)
        else:
            results[label] = f"still counted after {limit:.0f} s"
        silent.close()
        if label.startswith("ping 1 s"):  # a client that does answer pings must not be dropped
            ticket = session.post(f"{origin}/ws-ticket", headers={"Origin": origin}, timeout=5).json()["ticket"]
            with connect(f"ws://localhost:{port}/ws?ticket={ticket}", additional_headers={"Origin": origin}) as ws:
                ws.recv()
                time.sleep(6)
                ws.send('{"type": "ping"}')
                results["answering client"] = f"still connected after 6 s of pings, reply {ws.recv()}"
    finally:
        labkit.stop(server)


threads = [
    threading.Thread(target=watch, args=("ping 1 s, timeout 1 s", 8112, ["--ws-ping-interval", "1", "--ws-ping-timeout", "1"], 30)),
    threading.Thread(target=watch, args=("uvicorn defaults (20 s, 20 s)", 8113, [], 70)),
]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
for label in ("ping 1 s, timeout 1 s", "answering client", "uvicorn defaults (20 s, 20 s)"):
    print(f"{label:<32}{results[label]}")

python step_dead.py

ping 1 s, timeout 1 s           server noticed after   2.1 s
answering client                still connected after 6 s of pings, reply {"type": "pong"}
uvicorn defaults (20 s, 20 s)   server noticed after  40.2 s

What the output teaches

• With a one-second interval and timeout, the silent client was noticed after 2.1 seconds. With the defaults it took 40.2 seconds: one interval until the first ping, then one timeout for the answer that never came.
• The client that answered pings stayed connected after six seconds of silence on its own side.
• Pick the interval from the slowest timer in the path. A proxy in front of you may close idle connections sooner than your server would notice, which is Step 7. Very short intervals cost traffic on every connection and can drop clients on a flaky network, so the one-second setting here is for measuring, not a recommendation.

Step 7: Put nginx in front

In production a reverse proxy usually sits between the browser and uvicorn, and it brings two surprises that never show up on a laptop. The lab runs nginx 1.30.6, downloaded as a zip from nginx.org, in front of the hardened app. The configuration has four locations, each demonstrating one situation:

1. /plain/ is a bare proxy_pass, which is what most people write first.
2. /upgrade/ adds the two headers that carry the WebSocket upgrade through.
3. /idle/ adds a three-second proxy_read_timeout so you do not have to wait a minute to see the idle behavior.
4. /idle-ping/ has the same short timeout but points at a backend that pings every second.

# nginx.conf
worker_processes 1;
error_log logs/error.log warn;
pid logs/nginx.pid;
events { worker_connections 64; }

http {
    access_log off;
    server {
        listen 127.0.0.1:8180;

        # 1. a bare proxy_pass: the upgrade request does not get through
        location /plain/ { proxy_pass http://127.0.0.1:8120/; }

        # 2. the two headers that carry the upgrade request to the app
        location /upgrade/ {
            proxy_pass http://127.0.0.1:8120/;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
        }

        # 3. the same, behind a proxy that hangs up on silent connections after 3 seconds (nginx's default is 60)
        location /idle/ {
            proxy_pass http://127.0.0.1:8120/;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
            proxy_read_timeout 3s;
        }

        # 4. the same short timeout, but the app sends a WebSocket ping every second
        location /idle-ping/ {
            proxy_pass http://127.0.0.1:8121/;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
            proxy_read_timeout 3s;
        }
    }
}

"""step_proxy.py: put nginx in front of the notice board: the missing Upgrade headers and the idle timeout."""
import io
import os
import subprocess
import time
import zipfile

import requests
from websockets.exceptions import ConnectionClosed, InvalidStatus
from websockets.sync.client import connect

import labkit

VERSION = "1.30.6"
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
dist = os.path.join(labkit.HERE, "nginx_dist")
exe = os.path.join(dist, f"nginx-{VERSION}", "nginx.exe")
if not os.path.exists(exe):
    data = requests.get(f"https://nginx.org/download/nginx-{VERSION}.zip", headers={"User-Agent": UA}, timeout=120).content
    zipfile.ZipFile(io.BytesIO(data)).extractall(dist)
prefix = os.path.join(labkit.HERE, "nginx_run")
for folder in ("logs", "temp"):
    os.makedirs(os.path.join(prefix, folder), exist_ok=True)
base = [exe, "-p", prefix, "-c", os.path.join(labkit.HERE, "nginx.conf")]
print("nginx:", subprocess.run(base + ["-v"], capture_output=True, text=True).stderr.strip(), "| config test:",
      subprocess.run(base + ["-t"], capture_output=True, text=True).stderr.strip().splitlines()[-1])


def ticket_for(port: int) -> str:
    origin = f"http://localhost:{port}"
    session = requests.Session()
    session.get(f"{origin}/login?user=alice", timeout=5)
    return session.post(f"{origin}/ws-ticket", headers={"Origin": origin}, timeout=5).json()["ticket"]


def through_nginx(location: str, backend_port: int, hold: float) -> str:
    url = f"ws://localhost:8180/{location}/ws?ticket={ticket_for(backend_port)}"
    try:
        with connect(url, open_timeout=5, compression=None) as ws:
            ws.recv(timeout=3)  # hello
            started = time.perf_counter()
            try:
                ws.recv(timeout=hold)
                return "got a message?"
            except TimeoutError:
                return f"connected, silent for {hold:.0f} s, still open"
            except ConnectionClosed as closed:
                code = closed.rcvd.code if closed.rcvd else "none, the connection just ended"
                return f"dropped after {time.perf_counter() - started:3.1f} s of silence (close code: {code})"
    except InvalidStatus as refused:
        return f"handshake refused with HTTP {refused.response.status_code}"


backends = [
    labkit.start("app:app", 8120, "--log-level", "warning", env={"APP_ORIGIN": "http://localhost:8120"}),
    labkit.start("app:app", 8121, "--log-level", "warning", "--ws-ping-interval", "1", "--ws-ping-timeout", "5",
                 env={"APP_ORIGIN": "http://localhost:8121"}),
]
web = subprocess.Popen(base + ["-g", "daemon off;"])
try:
    labkit.wait_port(8180)
    print(f"{'/plain/    bare proxy_pass':<44}{through_nginx('plain', 8120, 1)}")
    print(f"{'/upgrade/  Upgrade + Connection headers':<44}{through_nginx('upgrade', 8120, 1)}")
    print(f"{'/idle/     proxy_read_timeout 3s':<44}{through_nginx('idle', 8120, 8)}")
    print(f"{'/idle-ping/ same, app pings every 1 s':<44}{through_nginx('idle-ping', 8121, 7)}")
finally:
    subprocess.run(base + ["-s", "stop"], capture_output=True)
    try:
        web.wait(timeout=10)
    except subprocess.TimeoutExpired:
        web.kill()
    for backend in backends:
        labkit.stop(backend)

python step_proxy.py

nginx: nginx version: nginx/1.30.6 | config test: nginx: configuration file C:\Claude\Cowork\Sxz.io\sxz-automations\tmp\sxz-publisher\lab\nginx.conf test is successful
/plain/    bare proxy_pass                  handshake refused with HTTP 404
/upgrade/  Upgrade + Connection headers     connected, silent for 1 s, still open
/idle/     proxy_read_timeout 3s            dropped after 3.0 s of silence (close code: none, the connection just ended)
/idle-ping/ same, app pings every 1 s       connected, silent for 7 s, still open

What the output teaches

• A bare proxy_pass breaks the handshake. The client got an HTTP 404. nginx’s WebSocket page explains why: the Upgrade and Connection headers are hop-by-hop headers and are “not passed from a client to proxied server” unless you pass them explicitly. Without them the app saw an ordinary GET to a WebSocket-only path. With the two proxy_set_header lines the same backend worked. (The nginx page shows proxy_http_version 1.1 as needed before version 1.29.7. I kept it in the configuration so it also works on older versions.)
• nginx hangs up on quiet connections. The same page says that “By default, the connection will be closed if the proxied server does not transmit any data within 60 seconds”. I set the timeout to three seconds, and a silent connection was dropped after 3.0 seconds with no close frame at all, which a browser reports as 1006.
• Pings fix it. The page continues: “the proxied server can be configured to periodically send WebSocket ping frames to reset the timeout and check if the connection is still alive”. With the app pinging every second, the connection stayed open for the whole seven seconds, because nginx saw data from the upstream.

Common mistake: tuning the proxy timeout and the ping interval separately. Set the ping interval comfortably below the shortest idle timeout anywhere between the browser and the app: nginx, a cloud load balancer, a corporate proxy. Behind TLS the page’s origin is an https:// origin and the browser connects with wss://, so update ALLOWED_ORIGINS to match.

Step 8: Shut down and reconnect cleanly

Every deploy stops your server, and every WebSocket client is connected when it does. What the clients see, and what they do next, decides whether a deploy is invisible or an incident.

8a. What uvicorn does when told to stop

Reading server.py in uvicorn 0.54.0, the order is: stop accepting new connections, tell every open connection to shut down, wait for running tasks to finish (the --timeout-graceful-shutdown option sets a limit, and I did not exercise it), and only then run the application’s lifespan shutdown. For a WebSocket, “tell it to shut down” means a close frame with code 1012, Service Restart. Two consequences follow. Clients get a code that says the server is restarting, and your lifespan shutdown hook is too late to send anything of your own, because the sockets are already closed by then.

uvicorn treats SIGINT and SIGTERM as stop signals and, on Windows, SIGBREAK, which is what Ctrl+Break sends. This machine runs Windows, so the lab sends a real Ctrl+Break to a uvicorn process in its own process group, which takes the same handler as SIGTERM does on Linux. I did not send a real SIGTERM, so on Linux, check that your process manager’s stop signal reaches uvicorn and not a wrapper script. The first scenario connects a socket client and starts a three-second HTTP request, and then stops the server 0.7 seconds later.

8b. A browser client that reconnects without a stampede

When a server stops, every client gets the same close code at the same moment. If every client reconnects immediately, the new server receives all of them at once, which is how a restart turns into an overload. The standard answer is to wait a random time before each retry, and to let the ceiling for that random time grow with each failure. This page is the browser side of the notice board. It fetches a fresh ticket for every attempt, so a reconnect never reuses a spent one.

<!doctype html>
<!-- site/client.html -->
<meta charset="utf-8">
<title>Notice board</title>
<ul id="log"></ul>
<script>
// The browser side of the notice board: get a one-time ticket, open the socket, reconnect with full jitter.
const state = {tries: 0, opened: 0, messages: 0, closes: [], gaveUp: false};
window.state = state;
let failures = 0;

async function connect() {
  state.tries += 1;
  let ticket;
  try {
    const reply = await fetch("/ws-ticket", {method: "POST"});
    if (reply.status === 401 || reply.status === 403) { state.gaveUp = true; return; }  // asking again will not help
    if (!reply.ok) throw new Error("ticket request failed: " + reply.status);
    ticket = (await reply.json()).ticket;
  } catch (error) {
    return retry();  // the server is down or restarting
  }
  const scheme = location.protocol === "https:" ? "wss" : "ws";
  const socket = new WebSocket(`${scheme}://${location.host}/ws?ticket=${ticket}`);
  socket.onopen = () => { state.opened += 1; failures = 0; };
  socket.onmessage = (event) => {
    state.messages += 1;
    document.getElementById("log").append(Object.assign(document.createElement("li"), {textContent: event.data}));
  };
  socket.onclose = (event) => {
    state.closes.push(event.code);
    if (event.code !== 1008) retry();  // 1008 means the server said no on purpose
  };
}

function retry() {
  failures += 1;
  const ceiling = Math.min(30000, 500 * 2 ** failures);  // 1 s, 2 s, 4 s ... up to 30 s
  setTimeout(connect, Math.random() * ceiling);  // full jitter: anywhere between 0 and the ceiling
}

connect();
</script>

Read retry() closely. After the first failure the delay is a random number between 0 and one second, after the second between 0 and two, and so on up to 30 seconds. “Full jitter” means the whole range is random, not a fixed delay plus a little noise. Three details keep it from retrying uselessly: a close with 1008 stops retrying because the server said no on purpose, a 401 or 403 from the ticket endpoint stops it because asking again will not change the answer, and a network error from the ticket request retries because the server may simply be restarting. The retry and backoff tutorial covers the idea in general.

The second scenario loads this page in Edge, stops the server with Ctrl+Break, waits 1.5 seconds, starts a new one, and waits for the page to connect again.

"""step_shutdown.py: stop uvicorn with a real Ctrl+Break (uvicorn handles it like SIGTERM) and watch the clients.

Windows detail: uvicorn is started with the base interpreter, not the venv launcher, so the signal reaches uvicorn itself."""
import os
import signal
import subprocess
import sys
import threading
import time

import requests
from playwright.sync_api import sync_playwright
from websockets.exceptions import ConnectionClosed
from websockets.sync.client import connect

import labkit

PORT = 8131
ORIGIN = f"http://localhost:{PORT}"
SESSION = "lab-session-alice"


def launch(*extra: str) -> subprocess.Popen:
    env = {**os.environ, "APP_ORIGIN": ORIGIN, "LAB_SESSION": SESSION, "PYTHONDONTWRITEBYTECODE": "1",
           "PYTHONPATH": os.path.join(sys.prefix, "Lib", "site-packages")}
    python = os.path.join(sys.base_prefix, "python.exe")
    proc = subprocess.Popen([python, "-m", "uvicorn", "app:app", "--host", "localhost", "--port", str(PORT),
                             "--log-level", "warning", *extra], cwd=labkit.HERE, env=env,
                            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
    labkit.wait_port(PORT)
    return proc


def part_one() -> None:
    proc = launch()
    seen: dict = {}
    ticket = requests.post(f"{ORIGIN}/ws-ticket", headers={"Origin": ORIGIN}, cookies={"sid": SESSION}, timeout=5).json()["ticket"]

    def socket_client() -> None:
        try:
            with connect(f"ws://localhost:{PORT}/ws?ticket={ticket}", additional_headers={"Origin": ORIGIN}) as ws:
                ws.recv()
                seen["connected"] = True
                ws.recv(timeout=20)
        except ConnectionClosed as closed:
            seen["socket_closed"] = time.perf_counter()
            seen["code"] = closed.rcvd.code if closed.rcvd else None

    def slow_request() -> None:
        reply = requests.get(f"{ORIGIN}/slow", params={"seconds": 3}, timeout=20)
        seen["slow_done"] = time.perf_counter()
        seen["slow_status"] = reply.status_code

    threads = [threading.Thread(target=socket_client), threading.Thread(target=slow_request)]
    for thread in threads:
        thread.start()
    time.sleep(0.7)
    t0 = time.perf_counter()
    proc.send_signal(signal.CTRL_BREAK_EVENT)
    time.sleep(0.5)
    try:
        labkit.wait_port(PORT, seconds=0.5)
        late = "accepted"
    except RuntimeError:
        late = "refused (the listening socket is closed)"
    exit_code = proc.wait(timeout=20)
    t_exit = time.perf_counter()
    for thread in threads:
        thread.join()
    print("signal sent at 0.00 s")
    print(f"WebSocket client saw close code {seen['code']} at {seen['socket_closed'] - t0:4.2f} s")
    print(f"new connection 0.5 s after the signal: {late}")
    print(f"in-flight GET /slow (3 s, started 0.7 s before the signal) answered HTTP {seen['slow_status']} at {seen['slow_done'] - t0:4.2f} s")
    print(f"process exited with code {exit_code} at {t_exit - t0:4.2f} s")


def part_two() -> None:
    proc = launch()
    with sync_playwright() as p:
        browser = p.chromium.launch(channel="msedge", headless=True)
        context = browser.new_context()
        context.add_cookies([{"name": "sid", "value": SESSION, "url": ORIGIN}])
        page = context.new_page()
        page.goto(f"{ORIGIN}/client.html")
        page.wait_for_function("window.state.opened >= 1", timeout=10000)
        t0 = time.perf_counter()
        proc.send_signal(signal.CTRL_BREAK_EVENT)
        proc.wait(timeout=20)
        page.wait_for_function("window.state.closes.length >= 1", timeout=10000)
        closed_at = time.perf_counter() - t0
        time.sleep(1.5)  # the deploy takes a moment: the old process is gone, the new one is not up yet
        proc = launch()
        t1 = time.perf_counter()
        page.wait_for_function("window.state.opened >= 2", timeout=40000)
        state = page.evaluate("window.state")
        print(f"browser: close codes {state['closes']}, seen at {closed_at:4.2f} s; new server up, page connected again "
              f"{time.perf_counter() - t1:4.2f} s later, {state['tries']} connection tries in total")
        browser.close()
    labkit.stop(proc)


print("--- Ctrl+Break against a server with a socket client and a slow HTTP request ---")
part_one()
print("--- the same signal against a server with a browser page that reconnects ---")
part_two()

python step_shutdown.py

--- Ctrl+Break against a server with a socket client and a slow HTTP request ---
signal sent at 0.00 s
WebSocket client saw close code 1012 at 0.03 s
new connection 0.5 s after the signal: refused (the listening socket is closed)
in-flight GET /slow (3 s, started 0.7 s before the signal) answered HTTP 200 at 2.31 s
process exited with code 3 at 2.40 s
--- the same signal against a server with a browser page that reconnects ---
browser: close codes [1012], seen at 0.18 s; new server up, page connected again 0.18 s later, 5 connection tries in total

What the output teaches

• The socket client received close code 1012 within a few hundredths of a second of the signal, and a new connection 0.5 seconds later was refused, because the listening socket is closed first.
• The three-second HTTP request, which started 0.7 seconds before the signal, still got its HTTP 200, 2.31 seconds after the signal, and the process exited right after that. A graceful stop waits for work in flight, so give your orchestrator a stop timeout longer than your slowest request.
• The exit code was 3. uvicorn’s server.py re-raises the signal it captured once shutdown is complete, so the process ends the way the signal would have ended it. I did not dig into why Windows reports that as 3, so check what your supervisor makes of the exit status of a stopped uvicorn.
• In the browser, the page saw close code 1012 and reconnected 0.18 seconds after the new server was up. The tries count of 5 includes the attempts that failed while the server was down. Your numbers will differ because the delay is random.

Step 9: Test it, and prove the tests bite

Every defense so far is a few lines that someone will one day “simplify”. Tests keep them in place, and a quick experiment shows whether the tests can actually tell.

9a. The test suite

The tests use Starlette’s TestClient, which can open a WebSocket in-process with websocket_connect. Rejections before accept() surface as a WebSocketDisconnect raised when the connection is opened, which is how the tests tell “refused at the handshake” from “accepted and then closed”. The second half of the file tests the hub with fake sockets and no network, including the stalled reader and the burst case.

"""test_ws.py: tests for the hardened notice board (pytest 9.2.0, Starlette TestClient)."""
import asyncio
import contextlib
import time

import pytest
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect

import app as board

ORIGIN = board.APP_ORIGIN


@pytest.fixture(autouse=True)
def clean_state():
    for table in (board.SESSIONS, board.TICKETS, board.hub.by_user):
        table.clear()
    board.NOTICES.clear()
    board.hub.evicted = 0


@pytest.fixture
def client():
    return TestClient(board.app)


def get_ticket(client, user="alice"):
    client.get(f"/login?user={user}")
    reply = client.post("/ws-ticket", headers={"origin": ORIGIN})
    assert reply.status_code == 200
    return reply.json()["ticket"]


def total_connections():
    return sum(len(group) for group in board.hub.by_user.values())


def wait_until(condition, seconds=2.0):
    deadline = time.monotonic() + seconds
    while time.monotonic() < deadline:
        if condition():
            return True
        time.sleep(0.005)
    return False


@contextlib.contextmanager
def connect(client, ticket, origin=ORIGIN):
    """Open a socket and, on the way out, wait until the server has noticed we left.

    Starlette's TestClient cancels the app the moment the with block ends. Without this wait, that cancel can land
    while the endpoint is still inside its TaskGroup and surface as a CancelledError in the test."""
    headers = {"origin": origin} if origin is not None else {}
    with client.websocket_connect(f"/ws?ticket={ticket}", headers=headers) as ws:
        yield ws
        before = total_connections()
        if before:
            ws.close(1000)
            wait_until(lambda: total_connections() < before)


def refused(client, ticket, origin=ORIGIN):
    """True when the handshake itself fails (close sent before accept), which a browser sees as HTTP 403."""
    try:
        with connect(client, ticket, origin):
            return False
    except WebSocketDisconnect:
        return True


PING = '{"type": "ping"}'


def close_code(ws, replies=1):
    """Return the code the server closes the socket with, or None if it never does.

    `replies` is how many messages the server could still send before it closes. Reading no more than that
    keeps a broken server (one that never closes) from hanging the test."""
    for _ in range(replies):
        try:
            ws.receive_text()
        except WebSocketDisconnect as closed:
            return closed.code
    return None


def test_ticket_login_happy_path(client):
    with connect(client, get_ticket(client)) as ws:
        assert ws.receive_json() == {"type": "hello", "user": "alice"}


def test_foreign_origin_is_refused_before_accept(client):
    assert refused(client, get_ticket(client), "http://127.0.0.1:4000")


@pytest.mark.parametrize("origin", [
    "http://localhost:8000.evil.example",  # starts with the real origin
    "http://localhost:80001",  # one character longer
    "https://localhost:8000",  # other scheme
    "null",  # sandboxed iframes and local files send this
    "",  # present but empty
])
def test_lookalike_origins_are_refused(client, origin):
    assert refused(client, get_ticket(client), origin)


def test_no_origin_header_is_allowed_for_ticket_clients(client):
    with connect(client, get_ticket(client), origin=None) as ws:
        assert ws.receive_json()["type"] == "hello"


def test_cookie_alone_does_not_open_a_socket(client):
    client.get("/login?user=alice")
    with pytest.raises(WebSocketDisconnect):
        with client.websocket_connect("/ws", headers={"origin": ORIGIN}):
            pass


def test_ticket_works_once(client):
    ticket = get_ticket(client)
    with connect(client, ticket) as ws:
        ws.receive_json()
    assert refused(client, ticket)


def test_ticket_expires(client, monkeypatch):
    monkeypatch.setattr(board, "TICKET_TTL", -1.0)
    assert refused(client, get_ticket(client))


def test_unknown_ticket_is_refused(client):
    assert refused(client, "not-a-real-ticket")


def test_ticket_endpoint_checks_origin_and_session(client):
    assert client.post("/ws-ticket", headers={"origin": ORIGIN}).status_code == 401
    client.get("/login?user=alice")
    assert client.post("/ws-ticket", headers={"origin": "http://127.0.0.1:4000"}).status_code == 403


def test_oversize_message_closes_1009(client):
    with connect(client, get_ticket(client)) as ws:
        ws.receive_json()
        ws.send_text("x" * (board.MAX_MESSAGE_CHARS + 1))
        ws.send_text(PING)  # sentinel: a server that did not close answers this
        assert close_code(ws) == 1009


@pytest.mark.parametrize("payload", [
    "not json",
    '{"type": "launch"}',
    '{"type": "post", "text": ""}',
    '{"type": "post", "text": "hi", "admin": true}',
])
def test_invalid_messages_close_1008(client, payload):
    with connect(client, get_ticket(client)) as ws:
        ws.receive_json()
        ws.send_text(payload)
        ws.send_text(PING)
        assert close_code(ws) == 1008


def test_binary_frames_close_1003(client):
    with connect(client, get_ticket(client)) as ws:
        ws.receive_json()
        ws.send_bytes(bytes([0, 1]))
        ws.send_text(PING)
        assert close_code(ws) == 1003


def test_rate_limit_closes_1008(client):
    with connect(client, get_ticket(client)) as ws:
        ws.receive_json()
        for _ in range(board.RATE_BURST + 15):
            ws.send_text(PING)
        assert close_code(ws, replies=board.RATE_BURST + 15) == 1008


def test_connection_cap_per_user(client):
    with contextlib.ExitStack() as stack:
        for _ in range(board.MAX_CONNECTIONS_PER_USER):
            stack.enter_context(connect(client, get_ticket(client))).receive_json()
        assert refused(client, get_ticket(client))


def test_post_reaches_the_board_and_ping_gets_pong(client):
    with connect(client, get_ticket(client)) as ws:
        ws.receive_json()
        ws.send_text('{"type": "post", "text": "hello"}')
        assert ws.receive_json() == {"type": "posted", "count": 1}
        ws.send_text(PING)
        assert ws.receive_json() == {"type": "pong"}
    assert board.NOTICES == [{"by": "alice", "text": "hello"}]


# --- the hub on its own: no network, fake sockets ------------------------------------------------------------------

class FakeSocket:
    def __init__(self, stalled=False):
        self.stalled, self.sent = stalled, []

    async def send_text(self, text):
        if self.stalled:
            await asyncio.Event().wait()  # a client that never reads: the send never finishes
        self.sent.append(text)


def run(coro):
    return asyncio.run(coro)


def test_outbox_is_bounded():
    async def scenario():
        conn = board.Connection(FakeSocket(), "alice")
        results = [conn.offer(str(n)) for n in range(board.OUTBOX_MAX + 5)]
        return results, conn.slow.is_set()

    results, slow = run(scenario())
    assert results[:board.OUTBOX_MAX] == [True] * board.OUTBOX_MAX
    assert results[board.OUTBOX_MAX:] == [False] * 5 and slow


def test_a_stalled_reader_does_not_hold_up_a_healthy_one():
    async def scenario():
        hub = board.Hub()
        stalled, healthy = board.Connection(FakeSocket(stalled=True), "alice"), board.Connection(FakeSocket(), "alice")
        for conn in (stalled, healthy):
            hub.add(conn)
        pumps = [asyncio.create_task(conn.pump()) for conn in (stalled, healthy)]
        started = asyncio.get_running_loop().time()
        for n in range(500):
            hub.send_to_user("alice", {"n": n})
            await asyncio.sleep(0)  # let the pumps run, as the event loop would between requests
        elapsed = asyncio.get_running_loop().time() - started
        await asyncio.sleep(0.05)
        for task in pumps:
            task.cancel()
        return len(healthy.websocket.sent), stalled.slow.is_set(), elapsed

    delivered, stalled_flagged, elapsed = run(scenario())
    assert delivered == 500 and stalled_flagged and elapsed < 1.0


def test_a_burst_larger_than_the_outbox_evicts_even_a_healthy_client():
    async def scenario():
        hub = board.Hub()
        healthy = board.Connection(FakeSocket(), "alice")
        hub.add(healthy)
        task = asyncio.create_task(healthy.pump())
        for n in range(board.OUTBOX_MAX + 5):  # no await between offers: the pump never gets a turn
            hub.send_to_user("alice", {"n": n})
        flagged = healthy.slow.is_set()
        task.cancel()
        return flagged

    assert run(scenario()) is True

python -m pytest -q

......................... [100%]
25 passed in 0.54s

Two traps shaped the helpers at the top of the file, and both cost me time.

• The cancel on exit. TestClient cancels the application the moment the with block ends. My first version of the tests failed with CancelledError when that landed while the handler was still inside its TaskGroup. I did not chase the root cause. The connect() helper now closes the socket and waits until the hub reports the connection gone before it leaves the block, and the suite has been stable across repeated runs since.
• A test that waits forever. A test that reads until the server closes the socket hangs forever when the server never closes it, which is exactly the bug the test exists to catch. I found this when the first mutation run hung. Each such test now sends a sentinel ping after the bad message and reads at most as many replies as the server could legitimately send, so a broken server produces a failed assertion and not a frozen test run.

9b. Do the tests catch regressions?

A test that cannot fail proves nothing. This script plants one bug at a time in a copy of app.py, using exact text swaps, and counts how many tests fail. Each mutant is a mistake someone could plausibly make.

"""mutations.py: plant one bug at a time in app.py and count the tests that notice."""
import os
import re
import shutil
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
SOURCE = open(os.path.join(HERE, "app.py"), encoding="utf-8").read()

MUTANTS = [
    ("origin check removed", [("return origin is None or origin in ALLOWED_ORIGINS", "return True")]),
    ("origin compared with startswith", [("return origin is None or origin in ALLOWED_ORIGINS",
                                          "return origin is None or any(origin.startswith(o) for o in ALLOWED_ORIGINS)")]),
    ("missing Origin treated as hostile", [("return origin is None or origin in ALLOWED_ORIGINS", "return origin in ALLOWED_ORIGINS")]),
    ("ticket can be used twice", [("TICKETS.pop(_key(raw), None)  # pop, not get: a ticket works exactly once", "TICKETS.get(_key(raw))")]),
    ("ticket never expires", [("if ticket is None or ticket.expires < time.monotonic():", "if ticket is None:")]),
    ("accept before the checks", [
        ('    origin = websocket.headers.get("origin")\n', '    await websocket.accept()\n    origin = websocket.headers.get("origin")\n'),
        ("    await websocket.accept()\n    conn = Connection(websocket, user)", "    conn = Connection(websocket, user)")]),
    ("message size check removed", [("if len(text) > MAX_MESSAGE_CHARS:", "if False:")]),
    ("rate limit removed", [("if not conn.bucket.take():", "if False:")]),
    ("connection cap off by one", [("hub.count(user) >= MAX_CONNECTIONS_PER_USER", "hub.count(user) > MAX_CONNECTIONS_PER_USER")]),
    ("outbox unbounded", [("asyncio.Queue(maxsize=OUTBOX_MAX)", "asyncio.Queue()")]),
    ("binary frames ignored", [("            raise CloseWith(1003)  # binary frames are not part of this protocol", "            continue")]),
]


def run_tests(folder: str) -> tuple[int, list[str], str]:
    env = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1", "COLUMNS": "170"}
    try:
        done = subprocess.run([sys.executable, "-m", "pytest", "-q", "--tb=no", "-rf", "-p", "no:cacheprovider", "test_ws.py"],
                              cwd=folder, env=env, capture_output=True, text=True, timeout=90)
    except subprocess.TimeoutExpired:
        return 1, ["(a test hung: the server never closed)"], "timeout"
    failed = re.findall(r"^FAILED test_ws\.py::(\S+)", done.stdout, flags=re.M)
    summary = done.stdout.strip().splitlines()[-1] if done.stdout.strip() else done.stderr[-200:]
    return len(failed), failed, summary


with tempfile.TemporaryDirectory() as scratch:
    shutil.copy(os.path.join(HERE, "test_ws.py"), scratch)
    shutil.copytree(os.path.join(HERE, "site"), os.path.join(scratch, "site"))
    with open(os.path.join(scratch, "app.py"), "w", encoding="utf-8", newline="\n") as f:
        f.write(SOURCE)
    count, _, summary = run_tests(scratch)
    print(f"{'unmodified app.py':<36}{count} failing tests ({summary})")
    caught = 0
    for name, swaps in MUTANTS:
        source = SOURCE
        for old, new in swaps:
            assert source.count(old) == 1, (name, old)
            source = source.replace(old, new)
        with open(os.path.join(scratch, "app.py"), "w", encoding="utf-8", newline="\n") as f:
            f.write(source)
        count, failed, _ = run_tests(scratch)
        caught += count > 0
        names = ", ".join(sorted({re.sub(r"\[.*", "", n) for n in failed})[:3])
        print(f"{name:<36}{count:>2} failing  {names}")
    print(f"caught {caught} of {len(MUTANTS)} planted bugs")

python mutations.py

unmodified app.py                   0 failing tests (25 passed in 0.54s)
origin check removed                 7 failing  test_foreign_origin_is_refused_before_accept, test_lookalike_origins_are_refused, test_ticket_endpoint_checks_origin_and_session
origin compared with startswith      2 failing  test_lookalike_origins_are_refused
missing Origin treated as hostile    1 failing  test_no_origin_header_is_allowed_for_ticket_clients
ticket can be used twice             1 failing  test_ticket_works_once
ticket never expires                 1 failing  test_ticket_expires
accept before the checks            11 failing  test_connection_cap_per_user, test_cookie_alone_does_not_open_a_socket, test_foreign_origin_is_refused_before_accept
message size check removed           1 failing  test_oversize_message_closes_1009
rate limit removed                   1 failing  test_rate_limit_closes_1008
connection cap off by one            1 failing  test_connection_cap_per_user
outbox unbounded                     3 failing  test_a_burst_larger_than_the_outbox_evicts_even_a_healthy_client, test_a_stalled_reader_does_not_hold_up_a_healthy_one, test_outbox_is_bounded
binary frames ignored                1 failing  test_binary_frames_close_1003
caught 11 of 11 planted bugs

All 11 planted bugs were caught, and the unmodified file has none failing. The first line of each row names the mistake. Two deserve a comment. “Accept before the checks” fails 11 tests, because accepting first turns every HTTP 403 into a close after the handshake, which is a different thing for a browser and a weaker one for you. “Outbox unbounded” is caught by three tests, including the burst test: without a bound the outbox never fills, so the slow-reader defense silently disappears while every other test still passes.

Quick reference: symptom, cause and fix

Confirm the whole thing works

1. python -m pytest -q prints 25 passed.
2. python mutations.py ends with caught 11 of 11 planted bugs.
3. python browser_matrix.py v1 v2 v3 shows the sibling origin and the none cookie getting in on v1, only the board’s own page getting in on v2, and again only the own page on v3.
4. python step_slow.py shows the healthy client receiving 3000 of 3000 on the hardened app and a small, flat memory number for the bounded outbox.
5. In your own app, open the browser’s Network tab, filter by WS, and look at the handshake request: confirm the Origin header there, then load a page from a second origin you control and confirm that it is refused.

Where to go next

• Authorize every message, not only the handshake. A connection that was allowed in can still ask for things it should not see, which is the object-level problem in How to Prevent Broken Object Level Authorization (IDOR) in a FastAPI App.
• Replace the in-memory ticket table and session dictionary with a shared store so the app can run several workers, and test that a ticket issued by one worker cannot be redeemed twice by two.
• If the data only flows from server to browser, a plain event stream may be enough. How to Stream Model Output With Server-Sent Events in FastAPI covers it, including resuming after a dropped connection.
• Read the sibling defenses: CORS for the HTTP side of the same origin rules, CSRF protection for the cookie problem, backpressure and load shedding for the overload side, and a hardened rate limiter for limits that need a client identity.
• Repeat Step 3 in Firefox and Safari. Cookie defaults and cross-site rules differ between browsers, and the cookie settings decided the outcome for the first server.

Sources

• IETF, RFC 6455, The WebSocket Protocol: section 5.5.2 (Ping), 7.4.1 (status codes), 10.2 (Origin considerations) and 10.5 (client authentication).
• IANA, WebSocket Protocol Registries, for close codes 1012 and 1013.
• ASGI, HTTP and WebSocket specification, for the close event sent before accept.
• MDN, Writing WebSocket servers and Set-Cookie.
• web.dev, “Same-site” and “same-origin”.
• PortSwigger Web Security Academy, Cross-site WebSocket hijacking, and MITRE, CWE-1385: Missing Origin Validation in WebSockets.
• FastAPI, WebSockets and Concurrency and async / await; Starlette, WebSockets.
• uvicorn, Settings, and the installed uvicorn 0.54.0 source files server.py and protocols/websockets/websockets_sansio_impl.py.
• nginx, WebSocket proxying.