Python 3 Standard Library Socket Programming Cookbook: TCP, UDP, select.select(), struct Packing, and Checksums

A zero-dependency cookbook for beginners. Everything below uses only the Python 3 standard library (socket, select, struct, hashlib, zlib, json, sys), flat procedural code, no asyncio, threads, or classes. Every script prints connections, messages and state changes so you can watch the protocol work. Run servers and clients in separate terminal tabs with python3 script.py <args>.

Part 1: Core Concepts & Standard Library Rules

• What a Socket Is: An Operating System endpoint that sends and receives raw bytes over a network.
• IP vs. Port: The IP address ('localhost' or '127.0.0.1') identifies the machine; the Port (an integer like 10000) identifies the specific Python process.
• Networks Only Speak Bytes (b"..."): Convert str to bytes before sending with .encode(), and convert received bytes back to str with .decode().
• TCP (SOCK_STREAM) vs. UDP (SOCK_DGRAM):
  • TCP is connection-oriented (bind -> listen -> accept on server; connect -> sendall -> recv on client).
  • UDP is connectionless (sendto and recvfrom). A UDP server uses a single socket for all clients and never uses select.select().
• Socket Port Reuse (SO_REUSEADDR): Always add server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) before bind() on TCP servers and proxies to avoid Address already in use errors when restarting.
• Binary Struct Packing (struct.Struct):
  • Format codes: i (4-byte int), I (4-byte unsigned int), f (4-byte float), ? (1-byte bool), c or 1s (1-byte char as bytes), Xs like 7s (X-byte string as bytes).
  • C Alignment Padding Rule: Never count bytes manually ('c i' is 8 bytes due to 3 padding bytes; 'i c' is 5 bytes; 'i i c' is 9 bytes; 'I I 1s' is 12 bytes; '7s6s' is 13 bytes). Always pass packer.size to recv(packer.size).
  • Null-Byte Padding Rule: Packing a shorter string like b"FAIL" into a 7s field pads the trailing bytes with null bytes (b"FAIL\x00\x00\x00"). Always strip null bytes after unpacking: status = status_bytes.decode().strip('\x00').

---

Part 2: Essential Building Blocks

import sys, struct, json, hashlib, zlib

# 1. Command-Line Arguments (python3 script.py localhost 10000 input.txt)
ip = sys.argv[1]
port = int(sys.argv[2])
filename = sys.argv[3]

# 2. Struct Packing & Unpacking
packer = struct.Struct('i i 1s')
packed_bytes = packer.pack(10, 25, b'+')
a, b, raw_op = packer.unpack(packed_bytes)
op = raw_op.decode().strip('\x00')

# 3. Delimiter-Separated Payloads (':', ';', '|')
numbers = [3, 7, 12, 18, 20]
bet = 500
msg_bytes = (":".join(str(x) for x in numbers) + f":{bet}").encode()
parts = [int(x) for x in msg_bytes.decode().split(":")]
recv_nums, recv_bet = parts[:-1], parts[-1]

# 4. Reading Text, JSON, and Binary Files
with open("input.json", "r") as f:
    data = json.load(f)

# 5. MD5 and CRC32 Checksums
md5_hex = hashlib.md5(b"1:5:12:19:20").hexdigest()
crc_int = zlib.crc32(b"1:5:12:19:20")

---

Part 3: Master Network Templates (TCP, Select, UDP, TCP-UDP Proxy)

Template 1: Single-Client TCP Server & Client

# --- tcp_server.py ---
import socket, sys

ip, port = sys.argv[1], int(sys.argv[2])
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((ip, port))
    server.listen(5)
    print(f"TCP Server listening on {ip}:{port}")
    while True:
        conn, addr = server.accept()
        with conn:
            print(f"Connected by {addr}")
            data = conn.recv(1024)
            if data:
                print("Received:", data.decode())
                conn.sendall(b"Hello client")

# --- tcp_client.py ---
import socket, sys

ip, port = sys.argv[1], int(sys.argv[2])
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((ip, port))
    client.sendall(b"Hello server")
    print("Server replied:", client.recv(1024).decode())

Template 2: Multi-Client TCP Server with select.select()

# --- select_tcp_server.py ---
import socket, select, sys

ip, port = sys.argv[1], int(sys.argv[2])
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((ip, port))
server.listen(5)

inputs = [server]
print(f"Multi-client TCP Server running on {ip}:{port}...")

while True:
    readable, _, _ = select.select(inputs, [], [], 1.0)
    for s in readable:
        if s is server:
            conn, addr = server.accept()
            print(f"[+] New client connected: {addr}")
            inputs.append(conn)
        else:
            data = s.recv(1024)
            if not data:
                print("[-] Client disconnected")
                inputs.remove(s)
                s.close()
            else:
                print(f"[<] Received: {data.decode()}")
                s.sendall(b"OK")

Template 3: UDP Server & Client with Timeout

# --- udp_server.py ---
import socket, sys

sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((sys.argv[1], int(sys.argv[2])))
while True:
    data, client_addr = sock.recvfrom(1024)
    print(f"Received {data.decode()} from {client_addr}")
    sock.sendto(b"Hello Client", client_addr)

# --- udp_client.py ---
import socket, sys

server_addr = (sys.argv[1], int(sys.argv[2]))
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(2.0)
try:
    sock.sendto(b"Hello Server", server_addr)
    data, _ = sock.recvfrom(1024)
    print("Received:", data.decode())
except (socket.timeout, ConnectionResetError):
    print("UDP Server did not respond!")
finally:
    sock.close()

Template 4: TCP-to-UDP Bridge Proxy

# --- tcp_udp_proxy.py ---
# Usage: python3 tcp_udp_proxy.py <proxy_ip> <proxy_port> <server_ip> <server_port>
import socket, select, struct, sys

proxy_addr = (sys.argv[1], int(sys.argv[2]))
server_addr = (sys.argv[3], int(sys.argv[4]))

tcp_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tcp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
tcp_sock.bind(proxy_addr)
tcp_sock.listen(5)

udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_sock.settimeout(2.0)
packer = struct.Struct("i i c")
inputs = [tcp_sock]

while True:
    readable, _, _ = select.select(inputs, [], [])
    for s in readable:
        if s is tcp_sock:
            conn, addr = tcp_sock.accept()
            print("Client connected to proxy:", addr)
            inputs.append(conn)
        else:
            msg = s.recv(packer.size)
            if not msg:
                inputs.remove(s)
                s.close()
                continue
            udp_sock.sendto(msg, server_addr)
            result, _ = udp_sock.recvfrom(packer.size)
            val, zero, r_char = packer.unpack(result)
            s.sendall(packer.pack(val * 2, zero, r_char))

---

Part 4: 5 Complete Protocol Implementations

Implementation 1: Schedule & Seat Reservation Protocol (7s6s) + 50% Intercept Proxy

Sample screenings.txt:

Dune;Hall 1;17:30;5
Oppenheimer;Hall 2;19:00;3
Barbie;Hall 1;20:45;0

# --- reservation_server.py ---
# Usage: python3 reservation_server.py <ip> <port> <screenings.txt>
import socket, select, struct, sys, random, string

ip, port, filename = sys.argv[1], int(sys.argv[2]), sys.argv[3]
screenings = {}
with open(filename, "r") as f:
    for line in f:
        line = line.strip()
        if line:
            title, hall, start_time, seats = line.split(";")
            screenings[f"{title};{start_time}"] = {
                "title": title, "hall": hall, "time": start_time, "seats": int(seats)
            }

packer = struct.Struct("7s6s")
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((ip, port))
server.listen(5)
inputs = [server]
print(f"Server running on {ip}:{port}...")

while True:
    readable, _, _ = select.select(inputs, [], [])
    for s in readable:
        if s is server:
            conn, addr = server.accept()
            print(f"[+] Client connected: {addr}")
            inputs.append(conn)
        else:
            data = s.recv(1024)
            if not data:
                print("[-] Client disconnected")
                inputs.remove(s)
                s.close()
                continue

            msg = data.decode().strip()
            print(f"[<] Received: {msg}")

            if msg == "GET_SCREENINGS":
                reply_str = "|".join(
                    f"{v['title']};{v['hall']};{v['time']}" for v in screenings.values()
                )
                s.sendall(reply_str.encode())
                print(f"[>] Sent screenings: {reply_str}")
            else:
                if msg in screenings and screenings[msg]["seats"] > 0:
                    screenings[msg]["seats"] -= 1
                    ticket_id = "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
                    s.sendall(packer.pack(b"SUCCESS", ticket_id.encode()))
                    print(f"[>] SUCCESS for {msg}. ID: {ticket_id} (Seats left: {screenings[msg]['seats']})")
                else:
                    s.sendall(packer.pack(b"FAIL", b"000000"))
                    print(f"[>] FAIL for {msg}.")

# --- reservation_client.py ---
# Usage: python3 reservation_client.py <ip> <port> <film_title> <start_time>
import socket, struct, sys

ip, port = sys.argv[1], int(sys.argv[2])
film = sys.argv[3] if len(sys.argv) > 3 else "Dune"
time_str = sys.argv[4] if len(sys.argv) > 4 else "17:30"
packer = struct.Struct("7s6s")

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((ip, port))
    client.sendall(b"GET_SCREENINGS")
    print("[>] Sent: GET_SCREENINGS")

    screenings_raw = client.recv(1024).decode()
    print("[<] Available screenings:")
    for item in screenings_raw.split("|"):
        print("   -", item)

    buy_req = f"{film};{time_str}"
    client.sendall(buy_req.encode())
    print(f"[>] Attempting to reserve: {buy_req}")

    resp_data = client.recv(packer.size)
    status_bytes, ticket_bytes = packer.unpack(resp_data)
    status = status_bytes.decode().strip('\x00')
    ticket_id = ticket_bytes.decode().strip('\x00')
    print(f"[<] Server Result: {status}, Ticket ID: {ticket_id}")

# --- intercept_proxy.py ---
# Usage: python3 intercept_proxy.py <proxy_ip> <proxy_port> <server_ip> <server_port>
import socket, struct, sys, random

proxy_addr = (sys.argv[1], int(sys.argv[2]))
server_addr = (sys.argv[3], int(sys.argv[4]))
packer = struct.Struct("7s6s")
intercepted_tickets = []

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as proxy_listener:
    proxy_listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    proxy_listener.bind(proxy_addr)
    proxy_listener.listen(1)
    print(f"Proxy listening on {proxy_addr}...")

    while True:
        client_conn, client_info = proxy_listener.accept()
        with client_conn:
            print(f"\n[+] Client connected to proxy: {client_info}")
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server_conn:
                server_conn.connect(server_addr)

                req1 = client_conn.recv(1024)
                server_conn.sendall(req1)
                resp1 = server_conn.recv(1024)
                client_conn.sendall(resp1)

                req2 = client_conn.recv(1024)
                server_conn.sendall(req2)

                resp2 = server_conn.recv(packer.size)
                status_b, ticket_b = packer.unpack(resp2)
                status = status_b.decode().strip('\x00')
                ticket_id = ticket_b.decode().strip('\x00')

                if status == "FAIL":
                    print(f"[Proxy] Failed ({status}, {ticket_id}). Forwarding to client.")
                    client_conn.sendall(resp2)
                else:
                    if random.random() < 0.5:
                        print(f"[Proxy] Forwarding valid ticket: {ticket_id}")
                        client_conn.sendall(resp2)
                    else:
                        intercepted_tickets.append(ticket_id)
                        print(f"[Proxy] INTERCEPTED {ticket_id}! All: {intercepted_tickets}")
                        client_conn.sendall(packer.pack(b"FAIL", b"000000"))

Implementation 2: Multi-Format Number Draw Service (TCP/UDP, Delimiter, struct, Checksum, JSON/CLI/Stdin)

# --- draw_server.py ---
# Usage: python3 draw_server.py <ip> <port>
import socket, select, struct, sys, random, hashlib

ip, port = sys.argv[1], int(sys.argv[2])
drawn = random.sample(range(1, 21), 5)
print(f"[*] Server drawn numbers: {drawn}")
packer = struct.Struct("i i i i i i")

def process_payload(data, mode="struct"):
    if mode == "colon":
        parts = [int(x) for x in data.decode().split(":")]
        picks, bet = parts[:5], parts[5]
        hits = len(set(picks) & set(drawn))
        win = bet * hits
        return (":".join(str(x) for x in drawn) + f":{win}").encode()
    elif mode == "struct":
        unpacked = packer.unpack(data)
        picks, bet = unpacked[:5], unpacked[5]
        hits = len(set(picks) & set(drawn))
        win = bet * hits
        return packer.pack(*drawn, win)
    elif mode == "checksum":
        msg_part, recv_md5 = data.decode().split("|")
        calc_md5 = hashlib.md5(msg_part.encode()).hexdigest()
        if recv_md5 != calc_md5:
            print("[!] Checksum mismatch!")
        parts = [int(x) for x in msg_part.split(":")]
        picks, bet = parts[:5], parts[5]
        hits = len(set(picks) & set(drawn))
        win = bet * hits
        reply_body = ":".join(str(x) for x in drawn) + f":{win}"
        reply_md5 = hashlib.md5(reply_body.encode()).hexdigest()
        return f"{reply_body}|{reply_md5}".encode()

MODE = "struct"  # "colon", "struct", or "checksum"
USE_TCP = True   # True for TCP+select, False for UDP

if USE_TCP:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((ip, port))
    server.listen(5)
    inputs = [server]
    while True:
        readable, _, _ = select.select(inputs, [], [])
        for s in readable:
            if s is server:
                conn, addr = server.accept()
                print(f"[+] Client connected: {addr}")
                inputs.append(conn)
            else:
                data = s.recv(1024)
                if not data:
                    inputs.remove(s)
                    s.close()
                else:
                    s.sendall(process_payload(data, MODE))
else:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((ip, port))
    while True:
        data, client_addr = sock.recvfrom(1024)
        sock.sendto(process_payload(data, MODE), client_addr)

# --- draw_client.py ---
# Usage: python3 draw_client.py <ip> <port> [extra_args]
import socket, struct, sys, random, json, hashlib

ip, port = sys.argv[1], int(sys.argv[2])
packer = struct.Struct("i i i i i i")

INPUT_SOURCE = "random"  # "random", "stdin", "json", or "argv"
if INPUT_SOURCE == "random":
    picks = random.sample(range(1, 21), 5)
    bet = 100
elif INPUT_SOURCE == "stdin":
    raw = input("Enter 5 numbers and bet separated by space: ").split()
    picks, bet = [int(x) for x in raw[:5]], int(raw[5])
elif INPUT_SOURCE == "json":
    with open(sys.argv[3], "r") as f:
        d = json.load(f)
        picks, bet = d["numbers"], d["bet"]
elif INPUT_SOURCE == "argv":
    picks, bet = [int(x) for x in sys.argv[3:8]], int(sys.argv[8])

print(f"[>] Client picks: {picks}, Bet: {bet}")

MODE = "struct"  # "colon", "struct", or "checksum"
if MODE == "colon":
    payload = (":".join(str(x) for x in picks) + f":{bet}").encode()
elif MODE == "struct":
    payload = packer.pack(*picks, bet)
elif MODE == "checksum":
    body = ":".join(str(x) for x in picks) + f":{bet}"
    payload = f"{body}|{hashlib.md5(body.encode()).hexdigest()}".encode()

USE_TCP = True
if USE_TCP:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.connect((ip, port))
        sock.sendall(payload)
        resp = sock.recv(1024)
else:
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.sendto(payload, (ip, port))
        resp, _ = sock.recvfrom(1024)

if MODE == "struct":
    unpacked = packer.unpack(resp)
    print(f"[<] Server drawn: {unpacked[:5]}, Winnings: {unpacked[5]}")
else:
    print(f"[<] Server response: {resp.decode()}")

Implementation 3: Multi-Client select TCP Server & Automated Binary Search Client (struct 'c i')

# --- search_server.py ---
# Usage: python3 search_server.py <hostname> <port>
import socket, select, struct, sys, random

ip, port = sys.argv[1], int(sys.argv[2])
secret = random.randint(1, 100)
game_over = False
print(f"[*] Target number is: {secret}")
packer = struct.Struct('c i')

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((ip, port))
server.listen(5)
inputs = [server]

while True:
    readable, _, _ = select.select(inputs, [], [])
    for s in readable:
        if s is server:
            conn, addr = server.accept()
            print(f"[+] Client connected: {addr}")
            inputs.append(conn)
        else:
            data = s.recv(packer.size)
            if not data:
                inputs.remove(s)
                s.close()
                continue
            op_byte, guess = packer.unpack(data)
            op = op_byte.decode()
            print(f"[<] Client asked: {op} {guess}")

            if game_over:
                s.sendall(packer.pack(b'V', 0))
            elif op == '<':
                s.sendall(packer.pack(b'I' if secret < guess else b'N', 0))
            elif op == '>':
                s.sendall(packer.pack(b'I' if secret > guess else b'N', 0))
            elif op == '=':
                if secret == guess:
                    game_over = True
                    s.sendall(packer.pack(b'Y', 0))
                    print("[!] Target found! Session over.")
                else:
                    s.sendall(packer.pack(b'K', 0))

# --- search_client.py ---
# Usage: python3 search_client.py <hostname> <port>
import socket, struct, sys, time

ip, port = sys.argv[1], int(sys.argv[2])
packer = struct.Struct('c i')
low, high = 1, 100

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
    client.connect((ip, port))
    while low <= high:
        mid = (low + high) // 2
        op = b'=' if low == high else b'>'
        print(f"[>] Sending: ({op.decode()}, {mid}) [Range: {low}..{high}]")
        client.sendall(packer.pack(op, mid))

        resp_data = client.recv(packer.size)
        ans_byte, _ = packer.unpack(resp_data)
        ans = ans_byte.decode()
        print(f"[<] Server replied: {ans}")

        if ans in ('Y', 'K', 'V'):
            print(f"[*] Terminating with status: {ans}")
            break
        if ans == 'I':
            low = mid + 1
        elif ans == 'N':
            high = mid
        time.sleep(0.3)

Implementation 4: Chunked TCP File Transfer & Expiring MD5 Checksum Server (BE| / KI|)

# --- checksum_srv.py ---
# Usage: python3 checksum_srv.py <ip> <port>
import socket, select, sys, time

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((sys.argv[1], int(sys.argv[2])))
server.listen(5)
inputs = [server]
store = {}

while True:
    readable, _, _ = select.select(inputs, [], [])
    for s in readable:
        if s is server:
            conn, _ = server.accept()
            inputs.append(conn)
        else:
            data = s.recv(1024)
            if not data:
                inputs.remove(s)
                s.close()
                continue
            parts = data.decode().strip().split("|")
            if parts[0] == "BE":
                fid, valid_sec, _, chsum = parts[1], int(parts[2]), parts[3], parts[4]
                store[fid] = {"hash": chsum, "expires": time.time() + valid_sec}
                s.sendall(b"OK")
            elif parts[0] == "KI":
                fid = parts[1]
                if fid in store and time.time() <= store[fid]["expires"]:
                    h = store[fid]["hash"]
                    s.sendall(f"{len(h)}|{h}".encode())
                else:
                    s.sendall(b"0|")

# --- file_transfer_cli.py ---
# Usage: python3 file_transfer_cli.py <srv_ip> <srv_port> <chsum_ip> <chsum_port> <file_id> <filename>
import socket, sys, hashlib

srv_addr = (sys.argv[1], int(sys.argv[2]))
chsum_addr = (sys.argv[3], int(sys.argv[4]))
file_id, filename = sys.argv[5], sys.argv[6]
md5 = hashlib.md5()

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s_net:
    s_net.connect(srv_addr)
    with open(filename, "rb") as f:
        while chunk := f.read(1024):
            s_net.sendall(chunk)
            md5.update(chunk)

hex_digest = md5.hexdigest()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s_ch:
    s_ch.connect(chsum_addr)
    s_ch.sendall(f"BE|{file_id}|60|{len(hex_digest)}|{hex_digest}".encode())
    s_ch.recv(1024)

# --- file_transfer_srv.py ---
# Usage: python3 file_transfer_srv.py <srv_ip> <srv_port> <chsum_ip> <chsum_port> <file_id> <out_filename>
import socket, sys, hashlib

srv_addr = (sys.argv[1], int(sys.argv[2]))
chsum_addr = (sys.argv[3], int(sys.argv[4]))
file_id, out_filename = sys.argv[5], sys.argv[6]
md5 = hashlib.md5()

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(srv_addr)
    server.listen(1)
    conn, _ = server.accept()
    with conn, open(out_filename, "wb") as f:
        while chunk := conn.recv(1024):
            f.write(chunk)
            md5.update(chunk)

local_hash = md5.hexdigest()
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s_ch:
    s_ch.connect(chsum_addr)
    s_ch.sendall(f"KI|{file_id}".encode())
    ch_len, remote_hash = s_ch.recv(1024).decode().split("|")

print("CSUM OK" if int(ch_len) > 0 and local_hash == remote_hash else "CSUM CORRUPTED")

Implementation 5: Unpacking & Packing Binary Records (struct.Struct)

# --- binary_records.py ---
# Usage: python3 binary_records.py db1.bin db2.bin db3.bin db4.bin
import sys, struct

formats = ['c 9s i', 'i f ?', '? 9s c', 'i c f']
for i in range(4):
    packer = struct.Struct(formats[i])
    with open(sys.argv[i + 1], 'rb') as f:
        print(packer.unpack(f.read(packer.size)))

print(struct.pack('16s i ?', b'elso', 83, True))
print(struct.pack('f ? c', 86.5, False, b'X'))
print(struct.pack('i 14s f', 74, b'masodik', 94.4))
print(struct.pack('c i 16s', b'Z', 105, b'harmadik'))