42. HTTP/1.1 from a bare socket

Before you use an HTTP library, speak HTTP yourself once (Net Ch 28, Net Ch 39).

42.1 A client

import socket

def http_get(host: str, path: str = "/", port: int = 80) -> tuple[str, dict[str, str], bytes]:
    req = (
        f"GET {path} HTTP/1.1\r\n"
        f"Host: {host}\r\n"
        "User-Agent: py-labs/0.1\r\n"
        "Accept: */*\r\n"
        "Connection: close\r\n"          # ask the server to close after the response
        "\r\n"
    )
    with socket.create_connection((host, port), timeout=5) as s:
        s.sendall(req.encode("ascii"))
        raw = b""
        while chunk := s.recv(65536):
            raw += chunk
    head, _, body = raw.partition(b"\r\n\r\n")
    status, *header_lines = head.decode("iso-8859-1").split("\r\n")
    headers = {}
    for line in header_lines:
        k, _, v = line.partition(":")
        headers[k.strip().lower()] = v.strip()
    return status, headers, body

status, headers, body = http_get("example.com")
print(status)
print(headers.get("content-type"), len(body), "bytes")

Reading until the server closes only works because we sent Connection: close. With keep-alive, the client must use Content-Length or decode Transfer-Encoding: chunked to know where the body ends. Getting that wrong, or two parties disagreeing about it, is request smuggling (Security in Depth Ch 30).

42.2 A server

# labs/tinyhttp.py: a tiny threaded HTTP/1.1 server with strict limits
import json
import socket
import threading
from datetime import datetime, timezone

MAX_HEADER = 16 * 1024

def read_request(conn: socket.socket) -> tuple[str, str, dict[str, str]] | None:
    buf = b""
    while b"\r\n\r\n" not in buf:
        chunk = conn.recv(4096)
        if not chunk:
            return None
        buf += chunk
        if len(buf) > MAX_HEADER:
            raise ValueError("headers too large")
    head = buf.split(b"\r\n\r\n", 1)[0].decode("iso-8859-1")
    request_line, *lines = head.split("\r\n")
    method, target, version = request_line.split(" ", 2)
    headers = {}
    for line in lines:
        name, sep, value = line.partition(":")
        if not sep or name != name.strip():          # reject "Name : value" (a smuggling vector)
            raise ValueError("malformed header")
        headers[name.lower()] = value.strip()
    return method, target, headers

def respond(conn: socket.socket, status: str, body: bytes, ctype: str = "text/plain") -> None:
    conn.sendall(
        f"HTTP/1.1 {status}\r\nContent-Type: {ctype}\r\nContent-Length: {len(body)}\r\n"
        f"Date: {datetime.now(timezone.utc):%a, %d %b %Y %H:%M:%S GMT}\r\n"
        "Connection: close\r\n\r\n".encode() + body
    )

def handle(conn: socket.socket) -> None:
    with conn:
        conn.settimeout(10)                           # slowloris defense: no idle sockets forever
        try:
            req = read_request(conn)
            if req is None:
                return
            method, target, headers = req
            if method != "GET":
                respond(conn, "405 Method Not Allowed", b"only GET\n")
            elif target == "/health":
                respond(conn, "200 OK", json.dumps({"ok": True}).encode(), "application/json")
            else:
                respond(conn, "404 Not Found", b"not found\n")
        except (ValueError, TimeoutError):
            respond(conn, "400 Bad Request", b"bad request\n")

if __name__ == "__main__":
    with socket.create_server(("127.0.0.1", 8080)) as srv:
        print("http://127.0.0.1:8080/health")
        while True:
            c, _ = srv.accept()
            threading.Thread(target=handle, args=(c,), daemon=True).start()

curl -v http://127.0.0.1:8080/health
printf 'GET /health HTTP/1.1\r\nHost: x\r\nBad Header : 1\r\n\r\n' | nc 127.0.0.1 8080   # 400

The standard library has http.server for local file serving and http.client for a low-level client. Neither is meant for production. For real services, use the tools in the next two chapters.