72. Capstone: one request, every layer, every guide, in Python

The HTTPS guide's finale and the Rust capstone trace one request through every layer. This program does it in Python, and adds the layer this guide leads to: a model that reads the response.

 ┌──────────────── one run of lifecycle.py ────────────────┐
 │ OS       process, file descriptors, syscalls, rusage     │  Part IV · OS guide
 │ DNS      a raw UDP query, parsed by hand                 │  Ch 41 · Net Ch 18
 │ TCP      connect(), handshake time                       │  Ch 40 · Net Ch 21
 │ TLS      version, cipher, certificate chain and expiry   │  Ch 45 · Sec Ch 28-29
 │ HTTP     a hand-written request, headers, body           │  Ch 42 · Net Ch 28
 │ Security headers, SSRF check on the target               │  Part VI · Sec guides
 │ AI       classify the page text with a trained model     │  Part VII · AI guide
 └──────────────────────────────────────────────────────────┘

# labs/lifecycle.py: one HTTPS request, traced through every layer of the series
import ipaddress
import os
import random
import resource
import socket
import ssl
import struct
import sys
import time
from html.parser import HTMLParser
from urllib.parse import urlsplit

def section(title: str) -> None:
    print(f"\n── {title} " + "─" * (60 - len(title)))

# ---------- DNS: a raw query (Ch 41, compact) ----------
def dns_a(name: str, server: str = "1.1.1.1") -> tuple[list[str], float]:
    qid = random.randint(0, 0xFFFF)
    q = struct.pack("!HHHHHH", qid, 0x0100, 1, 0, 0, 0)
    q += b"".join(bytes([len(p)]) + p.encode("idna") for p in name.split(".")) + b"\0"
    q += struct.pack("!HH", 1, 1)
    t = time.perf_counter()
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.settimeout(3)
        s.sendto(q, (server, 53))
        msg = s.recv(4096)
    ms = (time.perf_counter() - t) * 1000
    if struct.unpack_from("!H", msg)[0] != qid:
        raise ValueError("DNS ID mismatch")
    an = struct.unpack_from("!H", msg, 6)[0]
    off = len(q)                                      # skip header + question (same bytes we sent)
    ips = []
    for _ in range(an):
        while msg[off] not in (0,) and msg[off] & 0xC0 != 0xC0:   # skip the (possibly compressed) name
            off += msg[off] + 1
        off += 2 if msg[off] & 0xC0 == 0xC0 else 1
        rtype, _, _, rdlen = struct.unpack_from("!HHIH", msg, off)
        off += 10
        if rtype == 1:
            ips.append(socket.inet_ntoa(msg[off:off + rdlen]))
        off += rdlen
    return ips, ms

# ---------- HTTP framing: Transfer-Encoding: chunked (Net Ch 39) ----------
def dechunk(body: bytes) -> bytes:
    out = b""
    while body:
        size_line, _, body = body.partition(b"\r\n")
        size = int(size_line.split(b";")[0], 16)      # hex size, optional ;extensions
        if size == 0:
            break                                     # last chunk (trailers ignored)
        out, body = out + body[:size], body[size + 2:]  # skip the CRLF after each chunk
    return out

# ---------- HTML to text, for the AI layer ----------
class TextOnly(HTMLParser):
    def __init__(self) -> None:
        super().__init__()
        self.parts: list[str] = []
        self._skip = 0
    def handle_starttag(self, tag, attrs):
        self._skip += tag in ("script", "style")
    def handle_endtag(self, tag):
        if tag in ("script", "style") and self._skip:
            self._skip -= 1
    def handle_data(self, data):
        if not self._skip and data.strip():
            self.parts.append(data.strip())

def classify(text: str) -> list[tuple[str, float]]:
    """The AI layer: a tiny TF-IDF + logistic regression topic model (Ch 56)."""
    from sklearn.feature_extraction.text import TfidfVectorizer
    from sklearn.linear_model import LogisticRegression
    from sklearn.pipeline import make_pipeline
    train = {
        "documentation": ["this domain is for use in documentation examples", "reference manual guide tutorial",
                          "example text for illustrative purposes", "read the docs api reference"],
        "programming":   ["python downloads source code release", "install the interpreter and packages",
                          "language features library functions", "developers community code"],
        "commerce":      ["add to cart checkout buy now price", "free shipping order today sale",
                          "customer reviews product discount", "payment card billing subscription"],
    }
    X = [t for ts in train.values() for t in ts]
    y = [label for label, ts in train.items() for _ in ts]
    model = make_pipeline(TfidfVectorizer(), LogisticRegression(max_iter=1000)).fit(X, y)
    probs = model.predict_proba([text])[0]
    return sorted(zip(model.classes_, probs), key=lambda p: -p[1])

def main(url: str) -> int:
    u = urlsplit(url)
    host, port = u.hostname or "", u.port or 443
    path = (u.path or "/") + (f"?{u.query}" if u.query else "")

    section("OS: the process making the request")
    print(f"pid={os.getpid()} python={sys.version.split()[0]} platform={sys.platform}")
    fds_before = len(os.listdir("/dev/fd"))
    print(f"open file descriptors: {fds_before}  (run under `strace -f -e trace=network` to see every syscall)")

    section("DNS: name -> address (raw UDP query to 1.1.1.1:53)")
    ips, dns_ms = dns_a(host)
    print(f"{host} -> {ips}  in {dns_ms:.1f} ms")
    ip = ips[0]

    section("Security: is this a public address? (SSRF guard, Ch 43.4)")
    addr = ipaddress.ip_address(ip)
    print(f"{ip}: global={addr.is_global} private={addr.is_private} loopback={addr.is_loopback}")
    if not addr.is_global:
        print("refusing to connect to a non-public address")
        return 1

    section("TCP: three-way handshake")
    t = time.perf_counter()
    raw = socket.create_connection((ip, port), timeout=5)
    tcp_ms = (time.perf_counter() - t) * 1000
    local, remote = raw.getsockname(), raw.getpeername()
    print(f"{local[0]}:{local[1]} -> {remote[0]}:{remote[1]}  connected in {tcp_ms:.1f} ms  (fd {raw.fileno()})")

    section("TLS: handshake, certificate, verification")
    ctx = ssl.create_default_context()
    t = time.perf_counter()
    tls = ctx.wrap_socket(raw, server_hostname=host)       # SNI + chain + hostname verification
    tls_ms = (time.perf_counter() - t) * 1000
    cert = tls.getpeercert()
    subject = dict(x[0] for x in cert["subject"]).get("commonName")
    issuer = dict(x[0] for x in cert["issuer"]).get("organizationName")
    days = int((ssl.cert_time_to_seconds(cert["notAfter"]) - time.time()) / 86400)
    print(f"{tls.version()} {tls.cipher()[0]}  in {tls_ms:.1f} ms")
    print(f"certificate CN={subject} issuer={issuer} expires in {days} days")

    section("HTTP: request and response")
    request = (f"GET {path} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: lifecycle.py\r\n"
               "Accept: text/html\r\nAccept-Encoding: identity\r\nConnection: close\r\n\r\n")
    t = time.perf_counter()
    tls.sendall(request.encode())
    data = b""
    while chunk := tls.recv(65536):
        data += chunk
        if len(data) > 5 * 2**20:
            break                                           # cap the body (Ch 47.4)
    http_ms = (time.perf_counter() - t) * 1000
    tls.close()
    head, _, body = data.partition(b"\r\n\r\n")
    lines = head.decode("iso-8859-1").split("\r\n")
    headers = {k.lower(): v.strip() for k, _, v in (l.partition(":") for l in lines[1:])}
    print(f"{lines[0]}  ({len(body)} body bytes, {http_ms:.1f} ms)")
    for h in ("content-type", "cache-control", "server"):
        print(f"  {h}: {headers.get(h, '-')}")
    if headers.get("transfer-encoding", "").lower() == "chunked":
        body = dechunk(body)
        print(f"  chunked body decoded: {len(body)} bytes")

    section("Security: response headers (Sec Ch 41-44)")
    for h in ("strict-transport-security", "content-security-policy", "x-content-type-options"):
        print(f"  {'✅' if h in headers else '⚠️ '} {h}")

    section("AI: what is this page about? (Ch 56, AI guide Ch 9)")
    parser = TextOnly()
    parser.feed(body.decode("utf-8", "replace"))
    text = " ".join(parser.parts)[:2000]
    print(f"  text: {text[:120]!r}...")
    for label, p in classify(text):
        print(f"  {label:<14} {p:.2f}")

    section("Summary")
    total = dns_ms + tcp_ms + tls_ms + http_ms
    for name, ms in (("DNS", dns_ms), ("TCP", tcp_ms), ("TLS", tls_ms), ("HTTP", http_ms)):
        print(f"  {name:<5} {ms:7.1f} ms  {'█' * max(1, int(40 * ms / total))}")
    ru = resource.getrusage(resource.RUSAGE_SELF)
    print(f"  max RSS {ru.ru_maxrss / (2**20 if sys.platform == 'darwin' else 2**10):.0f} MB, "
          f"fds now {len(os.listdir('/dev/fd'))} (were {fds_before}: the socket was closed)")
    return 0

if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1] if len(sys.argv) > 1 else "https://example.com/"))

── OS: the process making the request ──────────────────────────
pid=95038 python=3.13.9 platform=darwin
open file descriptors: 4  (run under `strace -f -e trace=network` to see every syscall)

── DNS: name -> address (raw UDP query to 1.1.1.1:53) ──────────
example.com -> ['172.66.147.243', '104.20.23.154']  in 13.4 ms

── Security: is this a public address? (SSRF guard, Ch 43.4) ───
172.66.147.243: global=True private=False loopback=False

── TCP: three-way handshake ────────────────────────────────────
192.168.1.4:55268 -> 172.66.147.243:443  connected in 16.9 ms  (fd 3)

── TLS: handshake, certificate, verification ───────────────────
TLSv1.3 TLS_AES_256_GCM_SHA384  in 132.2 ms
certificate CN=example.com issuer=SSL Corporation expires in 82 days

── HTTP: request and response ──────────────────────────────────
HTTP/1.1 200 OK  (589 body bytes, 72.2 ms)
  content-type: text/html; charset=utf-8
  cache-control: -
  server: cloudflare
  chunked body decoded: 577 bytes

── Security: response headers (Sec Ch 41-44) ───────────────────
  ⚠️  strict-transport-security
  ⚠️  content-security-policy
  ⚠️  x-content-type-options

── AI: what is this page about? (Ch 56, AI guide Ch 9) ─────────
  text: 'Example Domain This domain is for use in documentation examples without needing permission. This is not a service; avoid'...
  documentation  0.53
  programming    0.24
  commerce       0.23

── Summary ─────────────────────────────────────────────────────
  DNS      13.4 ms  ██
  TCP      16.9 ms  ██
  TLS     132.2 ms  ██████████████████████
  HTTP     72.2 ms  ████████████
  max RSS 189 MB, fds now 4 (were 4: the socket was closed)

Read the output top to bottom and you have walked the series:

Extend the capstone:

1. Run it under strace -f -e trace=network,read,write (Linux) and label every syscall with the layer it belongs to.
2. Capture it with tcpdump -w lifecycle.pcap host <ip> and open the file in Wireshark: find the DNS query, the TCP handshake, the TLS ClientHello with your SNI, and the encrypted application data.
3. Replace the hand-written HTTP with requests and confirm, with the §63.2 stack-trace trick, that it performs the same steps.
4. Replace the classifier with a call to the local LLM from Ch 61 that summarizes the page in one sentence, validated with pydantic.
5. Rewrite the whole program in Go, then in Rust, and compare line counts, speed, and which bugs each compiler caught for you.