WebSocket Internals: A Deep Dive into Full-Duplex Communication

GET /ws/chat HTTP/1.1
Host: example.com
Origin: https://example.com
Connection: Upgrade
Upgrade: websocket
Sec-WebSocket-Version: 13
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat
Sec-WebSocket-Extensions: permessage-deflate

import base64
import hashlib
import secrets

# Client sends this key
client_key = "dGhlIHNhbXBsZSBub25jZQ=="

# Magic string required by protocol
magic_string = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

# Concatenate and hash
combined = client_key + magic_string
accept = base64.b64encode(hashlib.sha1(combined.encode()).digest()).decode()

# Result: "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
print(accept)

// Frame breakdown
struct websocket_frame {
    // First byte
    uint8_t FIN : 1;        // Final fragment
    uint8_t RSV1 : 1;       // Reserved for extension
    uint8_t RSV2 : 1;       // Reserved for extension
    uint8_t RSV3 : 1;       // Reserved for extension
    uint8_t OPCODE : 4;     // Frame type
    
    // Second byte
    uint8_t MASK : 1;       // Masked payload
    uint8_t PAYLOAD_LEN : 7; // Payload length
    
    // Extended payload length (if needed)
    uint16_t extended_len16;  // For 126
    uint64_t extended_len64;  // For 127
    
    // Masking key (if MASK=1)
    uint8_t masking_key[4];
    
    // Payload
    uint8_t payload[];
};

// Text frame: "Hello"
/*
FIN=1, RSV=0, OPCODE=1 (text)
MASK=0, PAYLOAD_LEN=5
Payload: "Hello"
*/
const textFrame = Buffer.from([0x81, 0x05, 0x48, 0x65, 0x6c, 0x6c, 0x6f]);

// Binary frame: 0xDEADBEEF
/*
FIN=1, RSV=0, OPCODE=2 (binary)
MASK=0, PAYLOAD_LEN=4
Payload: 0xDE 0xAD 0xBE 0xEF
*/
const binaryFrame = Buffer.from([0x82, 0x04, 0xDE, 0xAD, 0xBE, 0xEF]);

// Close frame
/*
FIN=1, RSV=0, OPCODE=8 (close)
MASK=0, PAYLOAD_LEN=0
*/
const closeFrame = Buffer.from([0x88, 0x00]);

// Ping frame
/*
FIN=1, RSV=0, OPCODE=9 (ping)
MASK=0, PAYLOAD_LEN=0
*/
const pingFrame = Buffer.from([0x89, 0x00]);

const WebSocket = require('ws');

const wss = new WebSocket.Server({ port: 8080 });

wss.on('connection', (ws, req) => {
    const clientIP = req.socket.remoteAddress;
    console.log(`Client connected: ${clientIP}`);
    
    // Send welcome message
    ws.send(JSON.stringify({
        type: 'welcome',
        message: 'Connected to WebSocket server'
    }));
    
    // Handle incoming messages
    ws.on('message', (message) => {
        console.log('Received:', message.toString());
        
        // Parse and respond
        try {
            const data = JSON.parse(message);
            handleMessage(ws, data);
        } catch (e) {
            ws.send(JSON.stringify({
                type: 'error',
                message: 'Invalid JSON'
            }));
        }
    });
    
    // Handle pong (response to ping)
    ws.on('pong', () => {
        console.log('Pong received');
    });
    
    // Handle close
    ws.on('close', (code, reason) => {
        console.log(`Connection closed: ${code} - ${reason}`);
    });
    
    // Handle errors
    ws.on('error', (error) => {
        console.error('WebSocket error:', error);
    });
    
    // Implement heartbeat
    const interval = setInterval(() => {
        if (ws.isAlive) {
            ws.isAlive = false;
            ws.ping();
        } else {
            console.log('Client did not respond to ping');
            ws.terminate();
            clearInterval(interval);
        }
    }, 30000);
    
    ws.isAlive = true;
});

function handleMessage(ws, data) {
    switch (data.type) {
        case 'chat':
            // Broadcast to all clients
            wss.clients.forEach(client => {
                if (client.readyState === WebSocket.OPEN) {
                    client.send(JSON.stringify({
                        type: 'chat',
                        message: data.message,
                        timestamp: Date.now()
                    }));
                }
            });
            break;
            
        case 'echo':
            ws.send(JSON.stringify({
                type: 'echo',
                message: data.message
            }));
            break;
            
        case 'private':
            // Send to specific client
            const targetClient = [...wss.clients].find(
                c => c.id === data.targetId
            );
            if (targetClient && targetClient.readyState === WebSocket.OPEN) {
                targetClient.send(JSON.stringify({
                    type: 'private',
                    from: data.from,
                    message: data.message
                }));
            }
            break;
    }
}

console.log('WebSocket server started on port 8080');

class WebSocketClient {
    constructor(url) {
        this.url = url;
        this.ws = null;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 5;
        this.reconnectDelay = 1000;
        this.listeners = new Map();
    }
    
    connect() {
        return new Promise((resolve, reject) => {
            this.ws = new WebSocket(this.url);
            
            this.ws.onopen = () => {
                console.log('Connected');
                this.reconnectAttempts = 0;
                this.emit('open');
                resolve();
            };
            
            this.ws.onmessage = (event) => {
                try {
                    const data = JSON.parse(event.data);
                    this.emit('message', data);
                } catch (e) {
                    this.emit('message', event.data);
                }
            };
            
            this.ws.onclose = (event) => {
                console.log('Disconnected:', event.code, event.reason);
                this.emit('close', event);
                this.attemptReconnect();
            };
            
            this.ws.onerror = (error) => {
                console.error('Error:', error);
                this.emit('error', error);
                reject(error);
            };
        });
    }
    
    send(data) {
        if (this.ws && this.ws.readyState === WebSocket.OPEN) {
            const message = typeof data === 'string' ? data : JSON.stringify(data);
            this.ws.send(message);
        } else {
            console.warn('WebSocket not connected');
        }
    }
    
    close(code = 1000, reason = '') {
        this.reconnectAttempts = this.maxReconnectAttempts; // Prevent reconnect
        this.ws.close(code, reason);
    }
    
    attemptReconnect() {
        if (this.reconnectAttempts < this.maxReconnectAttempts) {
            this.reconnectAttempts++;
            const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts - 1);
            console.log(`Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts})`);
            
            setTimeout(() => {
                this.connect().catch(console.error);
            }, delay);
        }
    }
    
    // Event emitter methods
    on(event, callback) {
        if (!this.listeners.has(event)) {
            this.listeners.set(event, []);
        }
        this.listeners.get(event).push(callback);
    }
    
    off(event, callback) {
        if (this.listeners.has(event)) {
            const callbacks = this.listeners.get(event);
            const index = callbacks.indexOf(callback);
            if (index > -1) callbacks.splice(index, 1);
        }
    }
    
    emit(event, data) {
        if (this.listeners.has(event)) {
            this.listeners.get(event).forEach(cb => cb(data));
        }
    }
}

// Usage
const client = new WebSocketClient('ws://localhost:8080');

client.on('open', () => {
    client.send({ type: 'chat', message: 'Hello!' });
});

client.on('message', (data) => {
    console.log('Received:', data);
});

client.connect();

import asyncio
import websockets
import json
from datetime import datetime

connected_clients = set()

async def handle_client(websocket, path):
    client_id = f"{websocket.remote_address}"
    connected_clients.add(websocket)
    print(f"Client connected: {client_id}")
    
    try:
        # Send welcome message
        await websocket.send(json.dumps({
            "type": "welcome",
            "message": "Connected to server",
            "timestamp": datetime.now().isoformat()
        }))
        
        async for message in websocket:
            try:
                data = json.loads(message)
                await process_message(websocket, data)
            except json.JSONDecodeError:
                await websocket.send(json.dumps({
                    "type": "error",
                    "message": "Invalid JSON"
                }))
                
    except websockets.exceptions.ConnectionClosed:
        print(f"Client disconnected: {client_id}")
    finally:
        connected_clients.remove(websocket)

async def process_message(websocket, data):
    msg_type = data.get("type")
    
    if msg_type == "broadcast":
        message = {
            "type": "broadcast",
            "content": data.get("content"),
            "sender": str(websocket.remote_address),
            "timestamp": datetime.now().isoformat()
        }
        # Send to all connected clients
        await asyncio.gather(
            *[client.send(json.dumps(message)) for client in connected_clients],
            return_exceptions=True
        )
    
    elif msg_type == "echo":
        await websocket.send(json.dumps({
            "type": "echo",
            "content": data.get("content"),
            "timestamp": datetime.now().isoformat()
        }))
    
    elif msg_type == "ping":
        await websocket.send(json.dumps({
            "type": "pong",
            "timestamp": datetime.now().isoformat()
        }))

async def main():
    async with websockets.serve(handle_client, "localhost", 8080):
        print("WebSocket server started on ws://localhost:8080")
        await asyncio.Future()  # Run forever

if __name__ == "__main__":
    asyncio.run(main())

// Server-side origin validation
const ALLOWED_ORIGINS = [
    'https://example.com',
    'https://www.example.com',
    'https://app.example.com'
];

wss.on('connection', (ws, req) => {
    const origin = req.headers.origin;
    
    if (!ALLOWED_ORIGINS.includes(origin)) {
        console.log(`Blocked connection from origin: ${origin}`);
        ws.close(4000, 'Invalid origin');
        return;
    }
    
    // Proceed with connection
    handleConnection(ws, req);
});

// Token-based authentication
const authClients = new Map();

wss.on('connection', (ws, req) => {
    // Extract token from query string
    const url = new URL(req.url, `http://${req.headers.host}`);
    const token = url.searchParams.get('token');
    
    if (!token) {
        ws.close(4001, 'Authentication required');
        return;
    }
    
    // Validate token
    const user = validateToken(token);
    if (!user) {
        ws.close(4002, 'Invalid token');
        return;
    }
    
    // Attach user to connection
    ws.user = user;
    authClients.set(user.id, ws);
});

function validateToken(token) {
    // Implement JWT validation or session lookup
    try {
        const decoded = jwt.verify(token, SECRET_KEY);
        return decoded;
    } catch (e) {
        return null;
    }
}

// Message validation
function validateMessage(data) {
    const errors = [];
    
    if (!data.type || typeof data.type !== 'string') {
        errors.push('Missing or invalid type');
    }
    
    if (data.type === 'chat') {
        if (!data.message || typeof data.message !== 'string') {
            errors.push('Missing or invalid message');
        }
        if (data.message.length > 10000) {
            errors.push('Message too long');
        }
    }
    
    return {
        valid: errors.length === 0,
        errors
    };
}

ws.on('message', (message) => {
    const data = JSON.parse(message);
    const validation = validateMessage(data);
    
    if (!validation.valid) {
        ws.send(JSON.stringify({
            type: 'error',
            errors: validation.errors
        }));
        return;
    }
    
    // Process valid message
});

const rateLimitMap = new Map();

function checkRateLimit(ws, maxMessages = 10, windowMs = 1000) {
    const clientId = ws.remoteAddress;
    const now = Date.now();
    
    if (!rateLimitMap.has(clientId)) {
        rateLimitMap.set(clientId, { count: 1, resetTime: now + windowMs });
        return true;
    }
    
    const clientData = rateLimitMap.get(clientId);
    
    if (now > clientData.resetTime) {
        clientData.count = 1;
        clientData.resetTime = now + windowMs;
        return true;
    }
    
    if (clientData.count >= maxMessages) {
        return false;
    }
    
    clientData.count++;
    return true;
}

ws.on('message', (message) => {
    if (!checkRateLimit(ws)) {
        ws.close(4003, 'Rate limit exceeded');
        return;
    }
    
    // Process message
});

const Redis = require('ioredis');
const redis = new Redis();

const PUBSUB_CHANNEL = 'websocket:events';

wss.on('connection', (ws, req) => {
    const clientId = generateClientId();
    ws.clientId = clientId;
    
    // Store client info in Redis
    await redis.hset(`client:${clientId}`, {
        connected: 'true',
        server: process.env.SERVER_ID,
        connectedAt: Date.now()
    });
    
    // Subscribe to client-specific events
    const subscriber = redis.duplicate();
    await subscriber.subscribe(`client:${clientId}:events`);
    
    subscriber.on('message', (channel, message) => {
        if (ws.readyState === WebSocket.OPEN) {
            ws.send(message);
        }
    });
    
    ws.on('message', async (message) => {
        // Broadcast to other servers
        await redis.publish(PUBSUB_CHANNEL, JSON.stringify({
            type: 'broadcast',
            clientId,
            message: message.toString()
        }));
    });
    
    ws.on('close', async () => {
        await redis.hdel(`client:${clientId}`, 'connected');
        await subscriber.unsubscribe();
        await subscriber.quit();
    });
});

// Handle cross-server messages
redis.subscribe(PUBSUB_CHANNEL);
redis.on('message', (channel, message) => {
    const event = JSON.parse(message);
    
    // Forward to local WebSocket clients
    wss.clients.forEach(client => {
        if (client.clientId !== event.clientId && client.readyState === WebSocket.OPEN) {
            client.send(event.message);
        }
    });
});

# Nginx WebSocket configuration
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

upstream websocket_backend {
    server ws1.example.com:8080;
    server ws2.example.com:8080;
    server ws3.example.com:8080;
}

server {
    location /ws/ {
        proxy_pass http://websocket_backend;
        
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        proxy_read_timeout 86400;
        proxy_send_timeout 86400;
    }
}

const wss = new WebSocket.Server({
    port: 8080,
    perMessageDeflate: {
        threshold: 1024,
        clientNoContextTakeover: true,
        serverNoContextTakeover: true,
        memLevel: 9
    }
});

// Server-side heartbeat
const HEARTBEAT_INTERVAL = 30000;

wss.on('connection', (ws) => {
    ws.isAlive = true;
    
    ws.on('pong', () => {
        ws.isAlive = true;
    });
    
    const heartbeat = setInterval(() => {
        if (ws.isAlive === false) {
            clearInterval(heartbeat);
            return ws.terminate();
        }
        
        ws.isAlive = false;
        ws.ping();
    }, HEARTBEAT_INTERVAL);
    
    ws.on('close', () => {
        clearInterval(heartbeat);
    });
});

// Prefer binary for large data
const data = new Float32Array([1.5, 2.5, 3.5, 4.5]);

// Send as binary (more efficient)
ws.send(data.buffer);

// Or use Blob
const blob = new Blob([data], { type: 'application/octet-stream' });
ws.send(blob);

function logFrame(buffer, isSent) {
    const firstByte = buffer[0];
    const secondByte = buffer[1];
    
    const fin = (firstByte & 0x80) >> 7;
    const opcode = firstByte & 0x0f;
    const masked = (secondByte & 0x80) >> 7;
    let payloadLen = secondByte & 0x7f;
    
    let offset = 2;
    if (payloadLen === 126) {
        payloadLen = buffer.readUInt16BE(2);
        offset = 4;
    } else if (payloadLen === 127) {
        payloadLen = Number(buffer.readBigUInt64BE(2));
        offset = 10;
    }
    
    const opcodeNames = {
        0x0: 'Continuation',
        0x1: 'Text',
        0x2: 'Binary',
        0x8: 'Close',
        0x9: 'Ping',
        0xA: 'Pong'
    };
    
    console.log(`${isSent ? '→' : '←'} ${opcodeNames[opcode] || opcode} ` +
                `FIN:${fin} MASK:${masked} LEN:${payloadLen}`);
}