4.2 HTTP/1.1 Framing 与最小服务器

一次 recv 可能只返回半行 POST /echo HT，下一次也可能同时带来剩余 header 和另一条 request 的开头。若实现把“一次 recv”当成“一条 HTTP 消息”，解析结果就会依赖网络分段这种并不受应用控制的偶然条件。

HTTP/1.1 在 TCP byte stream 上定义 message boundary。Parser 必须增量读取 start-line、field section 和 content，同时限制时间与空间；代理和 origin 若对边界理解不同，还会给 request smuggling 留出缝隙。

这一课先把接收合同写清，再实现一个只支持 HTTP/1.1、Content-Length 和每连接一条 request 的教学服务器。功能刻意收窄，边界会逐项说明。

1. 一次 recv 没有协议意义

下面的写法不成立：

request = connection.recv(65536)
method, target, version = request.split(b"\r\n", 1)[0].split()

recv 返回“当前最多可取多少 byte”，不是“下一条 HTTP message”：

• start-line 或 \r\n\r\n 可能被拆开；
• body 可能尚未全部抵达；
• header 与 body 可能一次到达；
• persistent connection 上，下一条 request 也可能已经进入 buffer；
• peer 可以只发一点后停住；
• EOF 可能发生在声明长度之前。

因此 parser 需要一个跨 recv 保存的 buffer，以及清晰的读取状态。

2. HTTP/1.1 request 的边界

对普通 request，先寻找：

request-line CRLF
field-line CRLF
...
CRLF

最后一个空行结束 header section。是否存在 message body，不由“POST 通常有 body”这种经验决定，而由 framing field 指示。

本课接受：

Content-Length: 11

暂不实现：

Transfer-Encoding: chunked

不支持某项功能时要明确拒绝，不能退回“读到 socket 暂时没数据为止”。在 persistent connection 上，那会同时失去 message boundary 与连接活性。

3. 解析器先写资源上限

代码开始前先定合同：

项目	教学服务器策略
HTTP version	只接受 HTTP/1.1
request-target	只接受 origin-form 的 ASCII URI 子集，检查转义、控制字符与 fragment
header section	最多 16 KiB，包括结束空行
body	最多 1 MiB
Host	恰好一次；非空 reg-name 或带方括号 IPv6，可带端口
Content-Length	最多一个，只接受十进制非负整数
Transfer-Encoding	不支持；与 Content-Length 同时出现直接 400
timeout	accept 后 header 与 body 共用 5 秒 deadline；另设 5 秒写超时
connection reuse	response 后发送 Connection: close

这些数字是本实验的配置，不是 HTTP 的全球常量。生产 server 应按业务、proxy chain 和资源预算选择限制，并区分 header 过大、content 过大与读取超时。

这只是教学子集，不是符合全部 HTTP/1.1 要求的通用 server。RFC 9112 要求接收方能解码 chunked，而本例明确未实现；另外拒绝 IPvFuture/IPv6 zone 标识等 Host 扩展，不做虚拟主机路由，也不解码 URL path。

4. 模糊 framing 为什么危险

假设前端 proxy 按 Content-Length 切 request，后端却优先按 Transfer-Encoding 切。同一串 byte 会被两端划成不同 message，攻击者便可能把隐藏 request “走私”到后端。

RFC 9112 对 framing 冲突给出严格规则。本实验采取更窄的策略：

• 同时出现 Transfer-Encoding 与 Content-Length：400 并关闭；
• 出现任何 Transfer-Encoding：501 并关闭；
• 重复 Content-Length：400 并关闭；
• header name、空白或控制字符非法：400 并关闭。

教学代码不处在 proxy chain 中，但仍不使用宽松、猜测式解析。不同 hop 对同一 message 必须得到同一边界。

5. 可运行的最小服务器

下面保存为 mini_http11.py，使用 Python 3.10+。它支持：

• GET /health；
• HEAD /health；
• POST /echo；
• 404、405 与若干 parser error；
• 任意 TCP 分段下的 header/body 重组。

import re
import socket
import time
import logging
from ipaddress import IPv6Address
from email.utils import formatdate
from dataclasses import dataclass

MAX_HEADER_SECTION = 16 * 1024
MAX_BODY = 1024 * 1024
TOKEN = re.compile(rb"^[!#$%&'*+\-.^_\x60|~0-9A-Za-z]+$")

TARGET = re.compile(rb"/(?:[A-Za-z0-9._~!$&'()*+,;=:@/?-]|%[0-9A-Fa-f]{2})*")

def valid_host(value):
    # This subset accepts reg-name or bracketed IPv6, with an optional port.
    if value.startswith("["):
        match = re.fullmatch(r"\[([0-9A-Fa-f:.]+)\](?::([0-9]*))?", value)
        if not match:
            return False
        try:
            IPv6Address(match[1])
        except ValueError:
            return False
        port = match[2]
    else:
        host, separator, port = value.partition(":")
        if not re.fullmatch(r"(?:[A-Za-z0-9._~!$&'()*+,;=-]|%[0-9A-Fa-f]{2})+", host):
            return False
        if separator and (not port.isascii() or (port and not port.isdecimal())):
            return False
    digits = (port or "").lstrip("0") or "0"
    return len(digits) <= 5 and int(digits) <= 65535

class DeadlineReader:
    def __init__(self, connection, deadline):
        self.connection = connection
        self.deadline = deadline
        self.method = None

    def recv(self, length):
        remaining = self.deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError("request deadline exceeded")
        self.connection.settimeout(remaining)
        return self.connection.recv(length)

class HTTPError(Exception):
    def __init__(self, status, message):
        super().__init__(message)
        self.status = status
        self.message = message

@dataclass(frozen=True)
class Request:
    method: str
    target: str
    version: str
    headers: dict[str, list[str]]
    body: bytes

def receive_header_section(connection):
    buffer = bytearray()
    marker = b"\r\n\r\n"

    while marker not in buffer:
        if len(buffer) >= MAX_HEADER_SECTION:
            raise HTTPError(431, "header section too large")

        chunk = connection.recv(
            min(4096, MAX_HEADER_SECTION - len(buffer))
        )
        if chunk == b"":
            raise HTTPError(400, "EOF before header section")
        buffer.extend(chunk)

    marker_at = buffer.find(marker)
    return bytes(buffer[:marker_at]), bytes(buffer[marker_at + 4:])

def receive_exact(connection, initial, length):
    body = bytearray(initial[:length])
    while len(body) < length:
        chunk = connection.recv(min(4096, length - len(body)))
        if chunk == b"":
            raise HTTPError(400, "EOF before complete request content")
        body.extend(chunk)
    return bytes(body)

def parse_request(connection):
    head, buffered_after_head = receive_header_section(connection)
    lines = head.split(b"\r\n")
    if not lines or not lines[0]:
        raise HTTPError(400, "missing request line")

    request_parts = lines[0].split(b" ")
    if len(request_parts) != 3 or any(not part for part in request_parts):
        raise HTTPError(400, "malformed request line")

    method_bytes, target_bytes, version_bytes = request_parts
    if not TOKEN.fullmatch(method_bytes):
        raise HTTPError(400, "invalid method")

    try:
        method = method_bytes.decode("ascii")
        target = target_bytes.decode("ascii")
        version = version_bytes.decode("ascii")
    except UnicodeDecodeError as error:
        raise HTTPError(400, "non-ASCII request line") from error

    connection.method = method
    if version != "HTTP/1.1":
        raise HTTPError(400, "only HTTP/1.1 is supported")
    if not TARGET.fullmatch(target_bytes):
        raise HTTPError(400, "only origin-form targets are supported")

    headers: dict[str, list[str]] = {}
    for raw_line in lines[1:]:
        if not raw_line:
            raise HTTPError(400, "unexpected empty field line")
        if raw_line[:1] in (b" ", b"\t"):
            raise HTTPError(400, "obsolete folded field")
        if b":" not in raw_line:
            raise HTTPError(400, "field without colon")

        name_bytes, value_bytes = raw_line.split(b":", 1)
        if not TOKEN.fullmatch(name_bytes):
            raise HTTPError(400, "invalid field name")

        value_bytes = value_bytes.strip(b" \t")
        if any(
            byte < 32 and byte != 9 or byte == 127
            for byte in value_bytes
        ):
            raise HTTPError(400, "control character in field value")

        name = name_bytes.decode("ascii").lower()
        value = value_bytes.decode("latin-1")
        headers.setdefault(name, []).append(value)

    hosts = headers.get("host", [])
    if len(hosts) != 1 or not valid_host(hosts[0]):
        raise HTTPError(400, "need exactly one valid Host")

    content_lengths = headers.get("content-length", [])
    transfer_encodings = headers.get("transfer-encoding", [])
    if transfer_encodings and content_lengths:
        raise HTTPError(400, "ambiguous message framing")
    if transfer_encodings:
        raise HTTPError(501, "Transfer-Encoding is not supported")
    if len(content_lengths) > 1:
        raise HTTPError(400, "duplicate Content-Length")

    body_length = 0
    if content_lengths:
        raw_length = content_lengths[0]
        if not raw_length.isascii() or not raw_length.isdecimal():
            raise HTTPError(400, "invalid Content-Length")
        digits = raw_length.lstrip("0") or "0"
        limit = str(MAX_BODY)
        if len(digits) > len(limit) or (len(digits) == len(limit) and digits > limit):
            raise HTTPError(413, "request content too large")
        body_length = int(digits)

    if "expect" in headers:
        raise HTTPError(417, "Expect is not supported")

    body = receive_exact(
        connection,
        buffered_after_head,
        body_length,
    )
    return Request(method, target, version, headers, body)

REASONS = {
    200: "OK",
    400: "Bad Request",
    404: "Not Found",
    405: "Method Not Allowed",
    408: "Request Timeout",
    413: "Content Too Large",
    417: "Expectation Failed",
    431: "Request Header Fields Too Large",
    500: "Internal Server Error",
    501: "Not Implemented",
}

def encode_response(status, body=b"", headers=None, head_only=False):
    fields = {
        "Date": formatdate(usegmt=True),
        "Content-Type": "text/plain; charset=utf-8",
        "Connection": "close",
    }
    if headers:
        fields.update(headers)
    fields["Content-Length"] = str(len(body))

    start = f"HTTP/1.1 {status} {REASONS[status]}\r\n"
    field_lines = "".join(
        f"{name}: {value}\r\n"
        for name, value in fields.items()
    )
    head = (start + field_lines + "\r\n").encode("ascii")
    return head if head_only else head + body

def application(request):
    path = request.target.partition("?")[0]

    if path == "/health":
        if request.method in {"GET", "HEAD"}:
            return 200, b"healthy\n", {}
        return 405, b"method not allowed\n", {
            "Allow": "GET, HEAD",
        }

    if path == "/echo":
        if request.method == "POST":
            return 200, request.body, {
                "Content-Type": "application/octet-stream",
            }
        return 405, b"method not allowed\n", {
            "Allow": "POST",
        }

    return 404, b"not found\n", {}

def handle_connection(connection, read_budget=5):
    reader = DeadlineReader(connection, time.monotonic() + read_budget)
    try:
        request = parse_request(reader)
        status, body, headers = application(request)
    except TimeoutError:
        status, body, headers = 408, b"request timeout\n", {}
    except HTTPError as error:
        status = error.status
        body, headers = (error.message + "\n").encode("utf-8"), {}
    except OSError:
        return
    except Exception:
        logging.exception("request handling failed")
        status, body, headers = 500, b"internal error\n", {}

    response = encode_response(
        status, body, headers, head_only=reader.method == "HEAD",
    )
    try:
        connection.settimeout(5)
        connection.sendall(response)
    except OSError:
        # The connection owner closes it; this client cannot stop the listener.
        return

def serve(host="127.0.0.1", port=8080):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener:
        listener.setsockopt(
            socket.SOL_SOCKET,
            socket.SO_REUSEADDR,
            1,
        )
        listener.bind((host, port))
        listener.listen(128)
        print(f"listening on http://{host}:{port}")

        while True:
            connection, peer = listener.accept()
            with connection:
                connection.settimeout(5)
                handle_connection(connection)
                print("served", peer)

if __name__ == "__main__":
    try:
        serve()
    except KeyboardInterrupt:
        print("\nstopped")

启动：

python3 mini_http11.py

另一个终端验证：

curl -i http://127.0.0.1:8080/health
curl -i -X POST --data-binary 'hello world' \
  http://127.0.0.1:8080/echo
curl -i -X DELETE http://127.0.0.1:8080/health

预期分别得到 200、200 和 405；405 response 带 Allow: GET, HEAD。

还可以执行 curl -I http://127.0.0.1:8080/health：HEAD 返回 Content-Length: 8，但线上不发送 healthy\n。已识别为 HEAD 的请求即使后来遇到解析错误，也不会发送错误 body。Expect 一律 417，避免客户端等 100 Continue、服务器却等 body 的互等。

6. Parser 的几个关键决定

为什么 header value 用 latin-1 解码

这里用 latin-1 做 byte 到 code point 的一一映射，避免宽松 UTF-8 替换悄悄改变原始 octet。它不表示所有 field value 的语义都是 ISO-8859-1。真正使用某个 field 前，仍要按该 field 的 ABNF 或 library API 解析。

为什么拒绝重复 Content-Length

RFC 9112 对相同值的重复 Content-Length 留有规范化选择，但 security-sensitive parser 可以直接拒绝。教学 server 不承担 compatibility proxy 的职责，确定性比宽容更重要。

长度比较先去掉前导零，再按有效数字位数和值检查上限，最后才转 int。这样数千位的长度字段会得到 413，而不会触发 Python 整数转换限制后变成 500。合法的长前导零仍可接受，但整个 header 受 16 KiB 限制。

为什么不处理 buffer 中多出来的 byte

receive_header_section 可能同时读到 body 之后的 byte。当前 server 每连接只处理一条 request，并在 response 后关闭，因此不会把这些 byte 当成另一条 message。若要支持 persistent connection，parser 必须返回 remainder，并在下一轮继续解析，不能丢弃。

为什么所有 response 都 Connection: close

读取由 DeadlineReader 共享绝对期限，成功短读不会重置预算；发送使用另一个 5 秒超时。这个单线程服务器一次仍只能处理一条连接，deadline 不覆盖在 accept queue 中等候的时间，也不限制全部已排队客户端占用的资源。

HTTP/1.1 默认支持 persistent connection，但“支持”意味着 server 要正确循环解析多条 request、按顺序发送 response、设置 idle timeout，并在 error 后决定是否还能安全复用。这个实验主动关闭，缩小状态机，不把未实现的能力伪装成 keep-alive。

7. 用碎片化输入测试边界

测试客户端仍分小块写入，但这本身不保证接收端短读。因此加一个 ReadCap，强制服务端每次最多取 3 字节，确保 CRLF 和长度字段经过分段路径。socketpair 在常见 Unix 上不占 TCP 端口，其他平台实现可能不同；测试不访问外网。把它保存为 test_mini_http11.py，与 server 文件放在同一目录。

import socket
import threading

from mini_http11 import handle_connection

class ReadCap:
    def __init__(self, connection):
        self.connection = connection

    def recv(self, length):
        return self.connection.recv(min(length, 3))

    def settimeout(self, seconds):
        self.connection.settimeout(seconds)

    def sendall(self, data):
        self.connection.sendall(data)

def exchange(request):
    client, server = socket.socketpair()
    client.settimeout(3)

    def serve_once():
        with server:
            server.settimeout(2)
            handle_connection(ReadCap(server), read_budget=2)

    worker = threading.Thread(target=serve_once)
    worker.start()

    with client:
        for offset in range(0, len(request), 3):
            client.sendall(request[offset:offset + 3])
        client.shutdown(socket.SHUT_WR)

        chunks = []
        while True:
            chunk = client.recv(4096)
            if chunk == b"":
                break
            chunks.append(chunk)

    worker.join(timeout=2)
    assert not worker.is_alive()
    return b"".join(chunks)

def body_of(response):
    return response.split(b"\r\n\r\n", 1)[1]

echo = exchange(
    b"POST /echo HTTP/1.1\r\n"
    b"Host: atlas.example\r\n"
    b"Content-Length: 11\r\n"
    b"\r\n"
    b"hello world"
)
assert echo.startswith(b"HTTP/1.1 200 OK\r\n")
assert body_of(echo) == b"hello world"

missing_host = exchange(
    b"GET /health HTTP/1.1\r\n\r\n"
)
assert missing_host.startswith(b"HTTP/1.1 400 Bad Request\r\n")

ambiguous = exchange(
    b"POST /echo HTTP/1.1\r\n"
    b"Host: atlas.example\r\n"
    b"Content-Length: 4\r\n"
    b"Transfer-Encoding: chunked\r\n"
    b"\r\n"
)
assert ambiguous.startswith(b"HTTP/1.1 400 Bad Request\r\n")

too_large = exchange(
    b"POST /echo HTTP/1.1\r\n"
    b"Host: atlas.example\r\n"
    b"Content-Length: 1048577\r\n"
    b"\r\n"
)
assert too_large.startswith(
    b"HTTP/1.1 413 Content Too Large\r\n"
)

head = exchange(
    b"HEAD /health HTTP/1.1\r\n"
    b"Host: atlas.example\r\n"
    b"\r\n"
)
assert b"Content-Length: 8\r\n" in head
assert body_of(head) == b""

print("HTTP/1.1 fragmented-input tests passed")

运行：

python3 test_mini_http11.py

预期输出：

HTTP/1.1 fragmented-input tests passed

这些用例验证了固定 3 字节读取上限下的重组、Host 缺失、冲突长度、body 上限和正常 HEAD。它们不穷尽所有分段或 RFC 语法，也没有直接证明 TCP 重传、超时、reset 和并发安全；这些需要额外测试。

8. 从教学 server 到生产 server

• persistent connection、pipelining 与完整 remainder buffer；
• chunked transfer coding 和 trailer；
• absolute-form、authority-form、asterisk-form 等 request-target；
• TLS、ALPN 与 HTTP/2；
• 并发、排队、backpressure、graceful shutdown；
• access log、request ID、metrics 与 trace；
• routing、media type、authentication、authorization；
• header timeout、body deadline、rate limit 与全局资源预算；
• proxy trust boundary，例如哪些 forwarded field 可以相信；
• fuzzing、差分测试和 request smuggling 回归集。

这些不是“再补几个 if”就能安全完成的功能。真实服务优先使用维护良好的 HTTP library/server；手写版本适合学习协议边界和构造隔离实验。

9. 常见错误

“没读到数据就说明 request 结束。” Blocking socket 会等待，timeout 也不等于 message boundary。

“POST 才检查 Content-Length。” Request framing 独立于 method semantics。

“出现两个相同 Content-Length 可以随便取第一个。” Parser disagreement 会形成安全边界，应按统一规则处理或拒绝。

“限制 body 大小就不会耗尽资源。” Header、连接数、读取时间、response queue 和 parser CPU 都要有预算。

“server 回 400 后还能继续复用这条连接。” Framing error 后 buffer 边界可能已经不可信；本实验始终关闭。

“自己写的 server 通过 curl 就能上线。” curl 正常路径不能覆盖畸形输入、slow client、代理差异和并发生命周期。

10. 练习与验收

1. 把 header 每次拆成 1 byte，测试结果应保持不变。
2. 增加重复 Host、带空白的 field name、负 Content-Length 测试，均应 400。
3. 为 POST /echo 增加 Content-Type allowlist，不支持时返回 415。
4. 设计 persistent connection 状态机：标出 remainder buffer、idle timeout、何时关闭。
5. 阅读 RFC 9112 的 chunked decoding algorithm，列出实现 trailer 前还需要哪些上限。

验收条件：你能解释 parser 每次为什么继续读取或停止，能指出每个内存/time limit，并且不会把一次 recv、EOF 或 timeout 当作通用 message delimiter。

这条 POST /echo 即使分段读取，仍按 CRLF 与 Content-Length 恢复同一份 body；冲突 framing 会得到错误并关闭连接。能解释这些边界，比只看到一行 200 更接近理解 HTTP parser。

下一章进入HTTPS 与 TLS。HTTP 已经有了语义和边界，接下来要确认对面的服务端是谁，并让沿途观察者无法读取或篡改 content。

参考

• RFC 9110：HTTP Semantics
• RFC 9112：HTTP/1.1
• RFC 9205：Building Protocols with HTTP