WebSockets Security for Beginners (2026 Edition)

Complete Node.js WebSocket Server with Security:

// requirements: npm install ws jsonwebtoken express-rate-limit express-validator

const WebSocket = require('ws');
const http = require('http');
const jwt = require('jsonwebtoken');
const rateLimit = require('express-rate-limit');
const { body, validationResult } = require('express-validator');
const crypto = require('crypto');

const JWT_SECRET = process.env.JWT_SECRET || crypto.randomBytes(32).toString('hex');
const MAX_MESSAGE_SIZE = 8 * 1024; // 8 KB
const MAX_CONNECTIONS_PER_IP = 5;
const MESSAGE_RATE_LIMIT = 100; // messages per minute

// Rate limiting store (in production, use Redis)
const connectionCounts = new Map();
const messageCounts = new Map();

// Cleanup old entries periodically
setInterval(() => {
    const now = Date.now();
    for (const [key, value] of messageCounts.entries()) {
        if (now - value.resetTime > 60000) {
            messageCounts.delete(key);
        }
    }
}, 60000);

class SecureWebSocketServer {
    constructor(port = 8080) {
        this.port = port;
        this.server = http.createServer();
        this.wss = new WebSocket.Server({ 
            noServer: true,
            clientTracking: true
        });
        
        this.setupHandlers();
        this.start();
    }
    
    setupHandlers() {
        // Handle upgrade request
        this.server.on('upgrade', (request, socket, head) => {
            // Authenticate before upgrading
            const authenticated = this.authenticateHandshake(request);
            
            if (!authenticated) {
                socket.write('HTTP/1.1 401 Unauthorized\r\n\r\n');
                socket.destroy();
                return;
            }
            
            // Rate limit connections per IP
            const clientIP = this.getClientIP(request);
            if (!this.checkConnectionLimit(clientIP)) {
                socket.write('HTTP/1.1 429 Too Many Requests\r\n\r\n');
                socket.destroy();
                return;
            }
            
            // Upgrade to WebSocket
            this.wss.handleUpgrade(request, socket, head, (ws) => {
                this.wss.emit('connection', ws, request);
            });
        });
        
        // Handle WebSocket connections
        this.wss.on('connection', (ws, request) => {
            const clientIP = this.getClientIP(request);
            const token = this.extractToken(request);
            let user = null;
            
            try {
                const decoded = jwt.verify(token, JWT_SECRET);
                user = decoded.userId || decoded.sub;
            } catch (e) {
                ws.close(1008, 'Invalid token');
                return;
            }
            
            // Store connection metadata
            ws.userId = user;
            ws.clientIP = clientIP;
            ws.connectedAt = Date.now();
            ws.messageCount = 0;
            
            console.log(`[WebSocket] User ${user} connected from ${clientIP}`);
            
            // Message handler with validation
            ws.on('message', (data) => {
                this.handleMessage(ws, data);
            });
            
            // Error handler
            ws.on('error', (error) => {
                console.error(`[WebSocket] Error for user ${user}:`, error);
            });
            
            // Close handler
            ws.on('close', (code, reason) => {
                console.log(`[WebSocket] User ${user} disconnected: ${code} ${reason}`);
            });
            
            // Send welcome message
            ws.send(JSON.stringify({
                type: 'welcome',
                message: 'Connected successfully',
                timestamp: Date.now()
            }));
        });
    }
    
    authenticateHandshake(request) {
        // Extract token from query parameter or header
        const token = this.extractToken(request);
        
        if (!token) {
            return false;
        }
        
        try {
            jwt.verify(token, JWT_SECRET);
            return true;
        } catch (e) {
            return false;
        }
    }
    
    extractToken(request) {
        // Try query parameter first
        const url = new URL(request.url, `http://${request.headers.host}`);
        const tokenParam = url.searchParams.get('token');
        if (tokenParam) {
            return tokenParam;
        }
        
        // Try Authorization header
        const authHeader = request.headers.authorization;
        if (authHeader && authHeader.startsWith('Bearer ')) {
            return authHeader.substring(7);
        }
        
        // Try Sec-WebSocket-Protocol header
        const protocol = request.headers['sec-websocket-protocol'];
        if (protocol) {
            const protocols = protocol.split(',').map(p => p.trim());
            const tokenProtocol = protocols.find(p => p.startsWith('token.'));
            if (tokenProtocol) {
                return tokenProtocol.substring(6);
            }
        }
        
        return null;
    }
    
    getClientIP(request) {
        return request.headers['x-forwarded-for']?.split(',')[0] || 
               request.socket.remoteAddress || 
               'unknown';
    }
    
    checkConnectionLimit(clientIP) {
        const count = connectionCounts.get(clientIP) || 0;
        if (count >= MAX_CONNECTIONS_PER_IP) {
            return false;
        }
        connectionCounts.set(clientIP, count + 1);
        return true;
    }
    
    handleMessage(ws, data) {
        const clientIP = ws.clientIP;
        const userId = ws.userId;
        
        // Check message size
        if (data.length > MAX_MESSAGE_SIZE) {
            console.warn(`[WebSocket] Message too large from ${userId}: ${data.length} bytes`);
            ws.close(1009, 'Message too large');
            return;
        }
        
        // Rate limit messages
        if (!this.checkMessageRateLimit(clientIP, userId)) {
            console.warn(`[WebSocket] Rate limit exceeded for ${userId}`);
            ws.send(JSON.stringify({
                type: 'error',
                message: 'Rate limit exceeded'
            }));
            return;
        }
        
        // Parse and validate JSON
        let message;
        try {
            message = JSON.parse(data.toString());
        } catch (e) {
            console.warn(`[WebSocket] Invalid JSON from ${userId}:`, e.message);
            ws.send(JSON.stringify({
                type: 'error',
                message: 'Invalid JSON format'
            }));
            return;
        }
        
        // Validate message schema
        if (!this.validateMessageSchema(message)) {
            console.warn(`[WebSocket] Invalid message schema from ${userId}`);
            ws.send(JSON.stringify({
                type: 'error',
                message: 'Invalid message format'
            }));
            return;
        }
        
        // Increment message count
        ws.messageCount++;
        
        // Process message
        this.processMessage(ws, message);
    }
    
    checkMessageRateLimit(clientIP, userId) {
        const key = `${clientIP}:${userId}`;
        const now = Date.now();
        const record = messageCounts.get(key);
        
        if (!record) {
            messageCounts.set(key, {
                count: 1,
                resetTime: now + 60000
            });
            return true;
        }
        
        if (now > record.resetTime) {
            record.count = 1;
            record.resetTime = now + 60000;
            return true;
        }
        
        if (record.count >= MESSAGE_RATE_LIMIT) {
            return false;
        }
        
        record.count++;
        return true;
    }
    
    validateMessageSchema(message) {
        // Basic schema validation
        if (!message || typeof message !== 'object') {
            return false;
        }
        
        if (!message.type || typeof message.type !== 'string') {
            return false;
        }
        
        // Add more validation based on message types
        const allowedTypes = ['ping', 'pong', 'chat', 'data', 'command'];
        if (!allowedTypes.includes(message.type)) {
            return false;
        }
        
        return true;
    }
    
    processMessage(ws, message) {
        // Echo back for ping
        if (message.type === 'ping') {
            ws.send(JSON.stringify({
                type: 'pong',
                timestamp: Date.now()
            }));
            return;
        }
        
        // Handle other message types
        console.log(`[WebSocket] Message from ${ws.userId}:`, message.type);
        
        // Broadcast to other clients (example)
        this.wss.clients.forEach((client) => {
            if (client !== ws && client.readyState === WebSocket.OPEN) {
                client.send(JSON.stringify({
                    type: 'broadcast',
                    from: ws.userId,
                    message: message
                }));
            }
        });
    }
    
    start() {
        this.server.listen(this.port, () => {
            console.log(`[WebSocket] Secure server listening on port ${this.port}`);
        });
    }
}

// Start server
const server = new SecureWebSocketServer(8080);

Complete Python WebSocket Security Implementation:

#!/usr/bin/env python3
"""
Production-ready WebSocket Security Implementation
Complete with authentication, rate limiting, and validation
"""

import asyncio
import websockets
import json
import jwt
import time
from typing import Dict, Optional, Set
from collections import defaultdict
from dataclasses import dataclass
import logging
import secrets

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

JWT_SECRET = secrets.token_urlsafe(32)
MAX_MESSAGE_SIZE = 8 * 1024  # 8 KB
MAX_CONNECTIONS_PER_IP = 5
MESSAGE_RATE_LIMIT = 100  # messages per minute

@dataclass
class ConnectionInfo:
    """Connection metadata."""
    user_id: str
    client_ip: str
    connected_at: float
    message_count: int = 0

class SecureWebSocketServer:
    """Secure WebSocket server with comprehensive security."""
    
    def __init__(self, host='localhost', port=8765):
        self.host = host
        self.port = port
        self.connections: Dict[websockets.WebSocketServerProtocol, ConnectionInfo] = {}
        self.connection_counts: Dict[str, int] = defaultdict(int)
        self.message_counts: Dict[str, Dict] = defaultdict(dict)
        
    async def authenticate(self, path: str, headers: Dict) -> Optional[str]:
        """Authenticate WebSocket handshake.
        
        Args:
            path: Request path
            headers: Request headers
            
        Returns:
            User ID if authenticated, None otherwise
        """
        # Extract token from query parameter
        if '?' in path:
            query = path.split('?', 1)[1]
            params = dict(p.split('=') for p in query.split('&') if '=' in p)
            token = params.get('token')
        else:
            token = None
        
        # Try Authorization header
        if not token:
            auth_header = headers.get('authorization', '')
            if auth_header.startswith('Bearer '):
                token = auth_header[7:]
        
        if not token:
            return None
        
        try:
            decoded = jwt.decode(token, JWT_SECRET, algorithms=['HS256'])
            return decoded.get('user_id') or decoded.get('sub')
        except jwt.InvalidTokenError:
            return None
    
    def check_connection_limit(self, client_ip: str) -> bool:
        """Check connection limit per IP.
        
        Args:
            client_ip: Client IP address
            
        Returns:
            True if within limit
        """
        count = self.connection_counts[client_ip]
        if count >= MAX_CONNECTIONS_PER_IP:
            return False
        self.connection_counts[client_ip] += 1
        return True
    
    def check_message_rate_limit(self, client_ip: str, user_id: str) -> bool:
        """Check message rate limit.
        
        Args:
            client_ip: Client IP
            user_id: User ID
            
        Returns:
            True if within limit
        """
        key = f"{client_ip}:{user_id}"
        now = time.time()
        record = self.message_counts.get(key)
        
        if not record:
            self.message_counts[key] = {'count': 1, 'reset_time': now + 60}
            return True
        
        if now > record['reset_time']:
            record['count'] = 1
            record['reset_time'] = now + 60
            return True
        
        if record['count'] >= MESSAGE_RATE_LIMIT:
            return False
        
        record['count'] += 1
        return True
    
    def validate_message(self, message: str) -> Optional[Dict]:
        """Validate message format and size.
        
        Args:
            message: Message string
            
        Returns:
            Parsed message dict if valid, None otherwise
        """
        if len(message.encode()) > MAX_MESSAGE_SIZE:
            logger.warning(f"Message too large: {len(message)} bytes")
            return None
        
        try:
            data = json.loads(message)
        except json.JSONDecodeError:
            logger.warning("Invalid JSON format")
            return None
        
        if not isinstance(data, dict) or 'type' not in data:
            return None
        
        allowed_types = ['ping', 'pong', 'chat', 'data', 'command']
        if data['type'] not in allowed_types:
            return None
        
        return data
    
    async def handle_client(self, websocket, path):
        """Handle WebSocket client connection.
        
        Args:
            websocket: WebSocket connection
            path: Request path
        """
        client_ip = websocket.remote_address[0] if websocket.remote_address else 'unknown'
        
        # Authenticate
        user_id = await self.authenticate(path, websocket.request_headers)
        if not user_id:
            logger.warning(f"Authentication failed for {client_ip}")
            await websocket.close(code=1008, reason="Authentication failed")
            return
        
        # Check connection limit
        if not self.check_connection_limit(client_ip):
            logger.warning(f"Connection limit exceeded for {client_ip}")
            await websocket.close(code=1008, reason="Too many connections")
            return
        
        # Store connection info
        conn_info = ConnectionInfo(
            user_id=user_id,
            client_ip=client_ip,
            connected_at=time.time()
        )
        self.connections[websocket] = conn_info
        
        logger.info(f"User {user_id} connected from {client_ip}")
        
        try:
            # Send welcome message
            await websocket.send(json.dumps({
                'type': 'welcome',
                'message': 'Connected successfully',
                'timestamp': time.time()
            }))
            
            # Handle messages
            async for message in websocket:
                # Validate message
                validated = self.validate_message(message)
                if not validated:
                    await websocket.send(json.dumps({
                        'type': 'error',
                        'message': 'Invalid message format'
                    }))
                    continue
                
                # Check rate limit
                if not self.check_message_rate_limit(client_ip, user_id):
                    await websocket.send(json.dumps({
                        'type': 'error',
                        'message': 'Rate limit exceeded'
                    }))
                    continue
                
                # Process message
                conn_info.message_count += 1
                await self.process_message(websocket, validated)
        
        except websockets.exceptions.ConnectionClosed:
            logger.info(f"User {user_id} disconnected")
        finally:
            # Cleanup
            if websocket in self.connections:
                del self.connections[websocket]
            self.connection_counts[client_ip] = max(0, self.connection_counts[client_ip] - 1)
    
    async def process_message(self, websocket, message: Dict):
        """Process validated message.
        
        Args:
            websocket: WebSocket connection
            message: Validated message dict
        """
        if message['type'] == 'ping':
            await websocket.send(json.dumps({
                'type': 'pong',
                'timestamp': time.time()
            }))
        else:
            logger.info(f"Processing message type: {message['type']}")
            # Handle other message types
    
    async def start(self):
        """Start WebSocket server."""
        logger.info(f"Starting secure WebSocket server on {self.host}:{self.port}")
        async with websockets.serve(self.handle_client, self.host, self.port):
            await asyncio.Future()  # run forever

if __name__ == '__main__':
    server = SecureWebSocketServer()
    asyncio.run(server.start())