Connection Limits

~12 min · management, rate-limit, abuse

Level 0Poller

0 XP0/60 lessons0/10 achievements

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

Two axes of abuse

WebSocket abuse comes in two shapes: too many connections from one source, or too many messages on one connection. The mitigations are different. Connection limits cap server resource exhaustion (file descriptors, memory). Message rate limits cap noisy/malicious clients without dropping their connection.

Per-IP connection cap

Track active connections per source IP. Reject new connections beyond the cap with code 4029 (an application convention; 429 is the closest HTTP analog). Be careful behind a proxy: websocket.client.host may show the proxy IP, not the real client. Use X-Forwarded-For if your reverse proxy sets it.

Per-connection message rate limit

Track timestamps of recent messages per WebSocket. If the count in the last second exceeds the limit, drop the message and send back a {type: 'error', code: 'rate_limited'}. Do not close the connection — that escalates and looks like a network failure. Just drop the abusive message and keep going.

Code

Combined connection + message rate limits·python

from collections import defaultdict, deque
from fastapi import WebSocket
import time

class LimitedManager:
    def __init__(self, *, max_per_ip=20, max_msg_per_sec=20):
        self.max_per_ip = max_per_ip
        self.max_msg_per_sec = max_msg_per_sec
        self.ip_counts: Dict[str, int] = defaultdict(int)
        self.msg_window: Dict[WebSocket, deque] = {}

    async def connect(self, ws: WebSocket) -> bool:
        ip = (ws.headers.get('x-forwarded-for', '').split(',')[0].strip()
              or (ws.client.host if ws.client else 'unknown'))
        if self.ip_counts[ip] >= self.max_per_ip:
            await ws.close(code=4029, reason='too many connections')
            return False
        self.ip_counts[ip] += 1
        self.msg_window[ws] = deque()
        await ws.accept()
        return True

    def check_rate(self, ws: WebSocket) -> bool:
        now = time.time()
        win = self.msg_window.get(ws)
        if win is None:
            return False
        while win and now - win[0] > 1.0:
            win.popleft()
        if len(win) >= self.max_msg_per_sec:
            return False
        win.append(now)
        return True

External links

OWASP: Denial of Service Cheat Sheet

Exercise

Set max_per_ip=2 and open three tabs to your endpoint from the same browser. The third must be rejected with code 4029. Then set max_msg_per_sec=5 and write a tight loop that sends 100 messages — confirm only the first 5 succeed each second and the rest receive a rate_limited error.