WebSocket — HTTP থেকে full-duplex upgrade

~110 লাইনে পুরো protocol-এর core — কোনো dependency নেই:

#!/usr/bin/env python3
"""ws_server.py — minimal WebSocket echo server (stdlib only, RFC 6455 subset)"""
import base64, hashlib, socket, threading

GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'


def recv_exact(conn, n):
    buf = b''
    while len(buf) < n:
        part = conn.recv(n - len(buf))
        if not part:
            raise ConnectionError('closed')
        buf += part
    return buf


def read_http_request(conn):
    buf = b''
    while b'\r\n\r\n' not in buf:
        buf += conn.recv(4096)
    head = buf.split(b'\r\n\r\n')[0].decode('latin1')
    lines = head.split('\r\n')
    hdrs = {}
    for line in lines[1:]:
        k, _, v = line.partition(':')
        hdrs[k.strip().lower()] = v.strip()
    return hdrs


def handshake_response(hdrs):
    if 'websocket' not in hdrs.get('upgrade', '').lower():
        raise ValueError('not a websocket upgrade')
    key = hdrs['sec-websocket-key']
    accept = base64.b64encode(
        hashlib.sha1((key + GUID).encode()).digest()).decode()
    return ('HTTP/1.1 101 Switching Protocols\r\n'
            'Upgrade: websocket\r\n'
            'Connection: Upgrade\r\n'
            f'Sec-WebSocket-Accept: {accept}\r\n\r\n').encode()


def read_frame(conn):
    b1, b2 = recv_exact(conn, 2)
    fin, opcode = b1 >> 7, b1 & 0x0F
    masked, length = b2 >> 7, b2 & 0x7F
    if length == 126:
        length = int.from_bytes(recv_exact(conn, 2), 'big')
    elif length == 127:
        length = int.from_bytes(recv_exact(conn, 8), 'big')
    mask = recv_exact(conn, 4) if masked else None
    payload = bytearray(recv_exact(conn, length)) if length else bytearray()
    if mask:
        for i in range(len(payload)):
            payload[i] ^= mask[i % 4]
    return fin, opcode, bytes(payload)


def encode_frame(opcode, payload):
    head = bytes([0x80 | opcode])
    n = len(payload)
    if n < 126:
        head += bytes([n])
    elif n < 65536:
        head += bytes([126]) + n.to_bytes(2, 'big')
    else:
        head += bytes([127]) + n.to_bytes(8, 'big')
    return head + payload


def serve_conn(conn, addr):
    print(f'[+] {addr} connected')
    try:
        conn.sendall(handshake_response(read_http_request(conn)))
        print(f'[+] {addr}: handshake ok — frames now')
        while True:
            fin, op, payload = read_frame(conn)
            if op == 8:                                   # CLOSE
                conn.sendall(encode_frame(8, payload))
                break
            if op == 9:                                   # PING → PONG
                conn.sendall(encode_frame(10, payload))
                continue
            if op == 10:                                  # PONG (unsolicited)
                continue
            if op in (1, 2):                              # TEXT/BINARY echo
                text = payload.decode(errors='replace')
                print(f'[{addr}] ← {text!r}')
                conn.sendall(encode_frame(1, f'echo: {text}'.encode()))
    except (ConnectionError, ValueError) as exc:
        print(f'[-] {addr}: {exc}')
    finally:
        conn.close()
        print(f'[-] {addr} closed')


srv = socket.socket()
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(('127.0.0.1', 8765))
srv.listen(8)
print('[ws] listening on ws://127.0.0.1:8765')
while True:
    conn, addr = srv.accept()
    threading.Thread(target=serve_conn, args=(conn, addr), daemon=True).start()

Test — experiment-এর client-টা localhost-এ ঘুরিয়ে:

import asyncio, websockets

async def main():
    async with websockets.connect('ws://127.0.0.1:8765') as ws:
        await ws.send('local test')
        print(await ws.recv())      # echo: local test
asyncio.run(main())

নিজে বাড়ান: (১) fragmentation support (FIN=0 continuation frames জমান); (২) broadcast mode — সব connected client-কে relay (chat!); (৩) close-frame-এর status-code parse করে log করুন; (৪) একটা malicious frame পাঠান যেখানে declared-length আসল payload-এর চেয়ে বড় — আপনার reader কীভাবে hang/crash করে দেখুন, তারপর timeout-guard বসান (protocol hardening-এর প্রথম হাতে-কলমে পাঠ)।