002 — CPython http.server 3.14.5

{
  "name": "http-server-recut",
  "what_it_is": "A minimal HTTP request handler, re-cut so that a single verb handler can be exercised with no connection.",
  "examined": {
    "project": "CPython http.server",
    "version": "3.14.5",
    "license": "PSF-2.0",
    "measured": {
      "module_lines": 1441,
      "BaseHTTPRequestHandler_methods": 28,
      "SimpleHTTPRequestHandler_methods": 11,
      "handle_one_request_lines": 36,
      "parse_request_lines": 116,
      "log_message_lines": 25,
      "dispatch_mechanism": "getattr(self, 'do_' + command)"
    }
  },

  "capabilities": {
    "SMS": {
      "message": "parse(bytes) -> Request | Response, and serialise(Response) -> bytes. Owns no connection.",
      "dispatch": "Chooses a handler from a registry it is given, checks SCL, calls it."
    },
    "SCL": {
      "policy": "Which handler may answer which method, and which may read the filesystem. Confinement to the served root lives here, not in the handler."
    },
    "TMS": {
      "handlers/health": "Answers a liveness probe. Needs nothing.",
      "handlers/static-file": "Serves a file from a confined root."
    },
    "DMS": {
      "access_log": "Events, written to a sink the caller supplies."
    }
  },

  "the_finding": "Upstream's extension point is inheritance. A new verb is a method on a subclass of the whole handler, so the smallest thing that can hold that verb is the whole handler. There is no subset to load, and therefore no island test to run.",

  "non_goals": [
    "Being an HTTP server. There is no socket here at all — requests arrive as bytes, which is the point.",
    "Being faster, safer or more complete than the standard library. It is smaller because it does less."
  ]
}

{
  "policy_version": "1.0",

  "_comment": [
    "Upstream, a handler's permissions are whatever its code happens to do,",
    "because a handler is a method on the same object that owns the socket and",
    "the filesystem helpers. translate_path exists to keep a request from",
    "escaping the served directory, and it is a method the subclass may",
    "override — so the boundary is enforced by the subclass agreeing to it.",
    "",
    "Here the permission is data next to the handler name, and dispatch checks",
    "it before calling anything."
  ],

  "handlers": {
    "handlers/static-file": {
      "methods": ["GET"],
      "read_filesystem": true,
      "root": "public"
    },
    "handlers/health": {
      "methods": ["GET"],
      "read_filesystem": false
    }
  }
}

"""What each handler may do.

`may_handle` is called by SMS/dispatch before a handler runs, so a handler
cannot answer a method its policy entry does not list — regardless of what it
would have done.

`confine` is the one worth comparing with upstream. `SimpleHTTPRequestHandler`
keeps requests inside the served directory in `translate_path`, a method the
subclass may override; the confinement holds because subclasses do not override
it. Here it is a function the handler must call and cannot replace.
"""
from __future__ import annotations

import json
from pathlib import Path

_POLICY = json.loads(Path(__file__).with_name("policy.json").read_text(encoding="utf-8"))
HANDLERS: dict[str, dict] = _POLICY["handlers"]


class Refused(Exception):
    """Raised when a handler asks for something its entry does not grant."""


def may_handle(handler: str, method: str) -> bool:
    return method in HANDLERS.get(handler, {}).get("methods", [])


def confine(handler: str, target: str, root: Path) -> Path:
    """Resolve a request target inside `root`, or refuse."""
    if not HANDLERS.get(handler, {}).get("read_filesystem"):
        raise Refused(f"{handler} has read_filesystem=false")

    candidate = (root / target.lstrip("/")).resolve()
    if root.resolve() not in candidate.parents and candidate != root.resolve():
        raise Refused(f"{handler} tried to leave {root}: {target}")
    return candidate

"""Choosing a handler for a request.

Upstream this is one line inside `handle_one_request`:

    method = getattr(self, mname)

which is why adding a verb means subclassing. The set of handlers is the set of
attributes on the instance, so there is no way to hand the dispatcher a
different set — and no way to have one without having all of them.

Here handlers are values in a registry. The dispatcher is told what exists,
which is what lets a caller pass exactly one.
"""
from __future__ import annotations

import sys
from pathlib import Path
from typing import Callable, Protocol

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from SCL import policy                          # noqa: E402
from SMS.message import Request, Response       # noqa: E402


class Handler(Protocol):
    name: str
    method: str

    def matches(self, request: Request) -> bool: ...

    def handle(self, request: Request) -> Response: ...


def dispatch(
    request: Request,
    handlers: list[Handler],
    *,
    on_event: Callable[..., None] = lambda **_: None,
) -> Response:
    # Method *and* target, which is the difference that matters. Upstream
    # dispatches on the verb alone, because the verb is the attribute name:
    # `getattr(self, 'do_GET')` can only ever find one thing. Two capabilities
    # that both answer GET — a health probe and a file server — cannot both
    # exist as handlers there; one of them has to become a branch inside the
    # other's method, or arrive by inheritance order.
    #
    # The first version of this file matched on method alone and reproduced the
    # same limitation by accident: the health handler answered every GET and the
    # file handler never ran once, including for the request written to test
    # that it refuses to leave its root.
    handler = next((h for h in handlers if h.method == request.method and h.matches(request)), None)

    if handler is None:
        on_event(kind="unhandled", method=request.method, target=request.target)
        allowed = ", ".join(sorted({h.method for h in handlers})) or "none"
        return Response(405, "Method Not Allowed", headers={"Allow": allowed})

    if not policy.may_handle(handler.name, request.method):
        on_event(kind="refused", handler=handler.name, method=request.method)
        return Response(403, "Forbidden", b"handler is not permitted this method")

    response = handler.handle(request)
    on_event(kind="handled", handler=handler.name, status=response.status,
             method=request.method, target=request.target)
    return response

"""Requests and responses as values. Nothing here owns a socket.

This is the first thing the re-cut moves. Upstream, `parse_request` is 116 lines
that read from `self.rfile` and call `self.send_error(...)` when the input is
malformed — so parsing a request and answering one are the same operation, and
neither can happen without a connection.

Here `parse` takes bytes and returns either a Request or a Response. A rejection
is a *value* the caller may choose to send. Nothing is written to anyone.
"""
from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True)
class Request:
    method: str
    target: str
    version: str
    headers: dict[str, str] = field(default_factory=dict)


@dataclass(frozen=True)
class Response:
    status: int
    reason: str
    body: bytes = b""
    headers: dict[str, str] = field(default_factory=dict)


def parse(raw: bytes) -> Request | Response:
    """bytes -> Request, or the Response explaining why not.

    Returning the rejection instead of sending it is the whole difference. A
    caller with no connection — a test, say — gets the same answer as one with.
    """
    try:
        head, _, _ = raw.partition(b"\r\n\r\n")
        lines = head.decode("iso-8859-1").split("\r\n")
        parts = lines[0].split()
    except Exception:
        return Response(400, "Bad Request", b"unreadable request line")

    if len(parts) != 3:
        return Response(400, "Bad Request", b"malformed request line")

    method, target, version = parts
    if not version.startswith("HTTP/"):
        return Response(400, "Bad Request", b"unrecognised protocol version")

    headers: dict[str, str] = {}
    for line in lines[1:]:
        name, sep, value = line.partition(":")
        if sep:
            headers[name.strip().lower()] = value.strip()

    return Request(method=method, target=target, version=version, headers=headers)


def serialise(response: Response) -> bytes:
    head = f"HTTP/1.1 {response.status} {response.reason}\r\n"
    headers = {"Content-Length": str(len(response.body)), **response.headers}
    head += "".join(f"{k}: {v}\r\n" for k, v in headers.items())
    return head.encode("latin-1") + b"\r\n" + response.body

"""Answers a liveness probe. Touches nothing.

The whole point of this file is how little it needs. It is 20 lines, it holds no
reference to a connection, and `island-test.py` runs it with nothing else
loaded. Upstream, the same capability is a `do_GET` on a subclass of
`BaseHTTPRequestHandler`, and instantiating one requires a socket, a client
address and a server object — so the smallest possible version of this is the
entire 1,441-line module plus a method.
"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from SMS.message import Request, Response  # noqa: E402


class HealthHandler:
    name = "handlers/health"
    method = "GET"

    def matches(self, request: Request) -> bool:
        return request.target == "/health"

    def handle(self, request: Request) -> Response:
        return Response(200, "OK", b"ok", {"Content-Type": "text/plain"})

"""Serves a file from a directory.

The confinement it needs — "do not leave the served root" — is asked of SCL
rather than implemented here. Upstream the equivalent lives in
`SimpleHTTPRequestHandler.translate_path`, on the class a subclass extends,
which means a subclass that overrides it takes the confinement with it. The
protection is real but it is held in the same place as the thing it protects
against.
"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from SCL.policy import Refused, confine       # noqa: E402
from SMS.message import Request, Response     # noqa: E402

_TYPES = {".html": "text/html", ".css": "text/css", ".json": "application/json", ".txt": "text/plain"}


class StaticFileHandler:
    name = "handlers/static-file"
    method = "GET"

    def __init__(self, root: Path) -> None:
        self._root = root

    def matches(self, request: Request) -> bool:
        # Everything the probe did not claim. Order in the registry decides, and
        # the registry is a list the caller controls.
        return True

    def handle(self, request: Request) -> Response:
        try:
            path = confine(self.name, request.target, self._root)
        except Refused as exc:
            # A refusal is reported as a value, like every other outcome here.
            return Response(403, "Forbidden", str(exc).encode())

        if not path.is_file():
            return Response(404, "Not Found", b"no such file")

        body = path.read_bytes()
        ctype = _TYPES.get(path.suffix, "application/octet-stream")
        return Response(200, "OK", body, {"Content-Type": ctype})

"""The access log as a sink you pass in.

Upstream, `log_message` is 25 lines that write to `sys.stderr`. Not to a stream
the server was given — to `sys.stderr`, named in the function. Redirecting it
means subclassing the request handler, which is the same extension mechanism as
adding a verb, so "I want the log somewhere else" and "I want to serve a new
method" are the same kind of change.

Here the sink is an argument. Passing `list.append` is how the island test reads
what happened without capturing a stream.
"""
from __future__ import annotations

from typing import Callable


class AccessLog:
    def __init__(self, sink: Callable[[str], None]) -> None:
        self._sink = sink
        self.events: list[dict] = []

    def record(self, **event) -> None:
        self.events.append(event)
        kind = event.pop("kind", "event")
        self._sink(f"{kind:<10} " + " ".join(f"{k}={v}" for k, v in event.items()))

"""The island test, and the measurement that motivated the re-cut.

    python src/island-test.py

Section 1 runs one handler with nothing else loaded and no connection.
Section 2 measures what the same thing costs upstream, by importing the real
module and asking what it takes to obtain one verb.
"""
from __future__ import annotations

import inspect
import re
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from DMS.access_log import AccessLog                     # noqa: E402
from SCL import policy                                   # noqa: E402
from SMS.dispatch import dispatch                        # noqa: E402
from SMS.message import Request, Response, parse         # noqa: E402

failures: list[str] = []


def report(label: str, ok: bool, detail: str = "") -> None:
    print(f"  {'PASS' if ok else 'FAIL'}  {label}" + (f" - {detail}" if detail else ""))
    if not ok:
        failures.append(label)


print("\n== 1. one handler, alone, with no connection")

from TMS.handlers.health import HealthHandler            # noqa: E402

lines: list[str] = []
log = AccessLog(sink=lines.append)
parsed = parse(b"GET /health HTTP/1.1\r\n\r\n")
response = dispatch(parsed, [HealthHandler()], on_event=log.record)

report("handlers/health answered with only itself loaded", response.status == 200,
       f"HTTP {response.status} {response.reason}")
report("no socket, no server, no client address were needed",
       isinstance(parsed, Request) and response.body == b"ok")
report("the log went to the caller's sink", len(lines) == 1, f"{len(lines)} line(s)")

# the static handler is absent, so its method must not be answerable
absent = dispatch(parse(b"GET /index.html HTTP/1.1\r\n\r\n"), [HealthHandler()], on_event=log.record)
report("an absent handler is absent", absent.status == 405,
       f"HTTP {absent.status} - nothing loaded claims /index.html, and that is reported rather than guessed")

print("\n== 2. a rejection is a value, not a transmission")
bad = parse(b"nonsense\r\n\r\n")
report("the parser returns the rejection", isinstance(bad, Response) and bad.status == 400,
       "upstream parse_request calls send_error, so parsing needs a connection")

print("\n== 3. SCL refuses before the handler runs")
denied = dispatch(Request("DELETE", "/health", "HTTP/1.1"), [HealthHandler()], on_event=log.record)
report("a method no handler declares is refused", denied.status == 405,
       f"HTTP {denied.status}, Allow: {denied.headers.get('Allow')}")
report("policy is data, not prose", policy.may_handle("handlers/health", "GET")
       and not policy.may_handle("handlers/health", "DELETE"))

print("\n== 4. what one verb costs upstream")
import http.server as upstream                            # noqa: E402

module_lines = len(inspect.getsource(upstream).splitlines())
base_methods = [m for m in upstream.BaseHTTPRequestHandler.__dict__ if not m.startswith("__")]
mro = [c.__name__ for c in upstream.SimpleHTTPRequestHandler.__mro__]
dispatch_src = inspect.getsource(upstream.BaseHTTPRequestHandler.handle_one_request)
uses_getattr = bool(re.search(r"getattr\(self,\s*mname\)", dispatch_src))
log_src = inspect.getsource(upstream.BaseHTTPRequestHandler.log_message)

print(f"    module                       {module_lines} lines")
print(f"    BaseHTTPRequestHandler       {len(base_methods)} methods")
print(f"    SimpleHTTPRequestHandler MRO {' -> '.join(mro)}")
print(f"    dispatch by getattr          {uses_getattr}")
print(f"    log_message names sys.stderr {'sys.stderr' in log_src}")

report("upstream dispatches by attribute lookup, so handlers cannot be passed in", uses_getattr,
       "if this stops being true the re-cut's premise needs revisiting")
report("upstream's log destination is named in the function", "sys.stderr" in log_src)
report("a verb handler is reached only through the whole handler",
       "BaseHTTPRequestHandler" in mro and len(mro) >= 4,
       f"{len(mro)}-deep MRO, {len(base_methods)} inherited methods before the verb")

print()
if failures:
    print(f"  {len(failures)} check(s) failed: {', '.join(failures)}")
    raise SystemExit(1)
print("  island test passed")

"""Run the re-cut against a few raw requests.

    python src/main.py

No socket is opened. Requests are bytes, which is what makes the handlers
testable and what upstream cannot offer without a connection.
"""
from __future__ import annotations

import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

from DMS.access_log import AccessLog                     # noqa: E402
from SMS.dispatch import dispatch                        # noqa: E402
from SMS.message import Request, Response, parse, serialise  # noqa: E402
from TMS.handlers.health import HealthHandler            # noqa: E402
from TMS.handlers.static_file import StaticFileHandler   # noqa: E402

REQUESTS = [
    b"GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n",
    b"GET /index.html HTTP/1.1\r\nHost: localhost\r\n\r\n",
    b"GET /../secrets.txt HTTP/1.1\r\nHost: localhost\r\n\r\n",
    b"POST /health HTTP/1.1\r\nHost: localhost\r\n\r\n",
    b"nonsense\r\n\r\n",
]


def main() -> int:
    root = Path(tempfile.mkdtemp()) / "public"
    root.mkdir()
    (root / "index.html").write_text("<h1>served</h1>\n", encoding="utf-8")
    (root.parent / "secrets.txt").write_text("not for you\n", encoding="utf-8")

    log = AccessLog(sink=lambda line: print(f"    {line}"))
    handlers = [HealthHandler(), StaticFileHandler(root=root)]

    print("\n== http-server re-cut (no socket)")
    for raw in REQUESTS:
        first = raw.split(b"\r\n", 1)[0].decode("latin-1")
        print(f"\n  -> {first}")
        parsed = parse(raw)
        # A rejection from the parser is a Response, not something it sent.
        response = parsed if isinstance(parsed, Response) else dispatch(
            parsed, handlers, on_event=log.record
        )
        head = serialise(response).split(b"\r\n", 1)[0].decode("latin-1")
        print(f"    {head}   {len(response.body)} byte(s)")

    print(f"\n  {len(log.events)} event(s) recorded to the sink the caller supplied")

    # The confinement is the one outcome worth asserting rather than eyeballing,
    # and the assertion has to require the refusal rather than merely permit it.
    # Written as "no 200 for a path containing secrets", it passed while the file
    # handler was never reached at all.
    escape = next((e for e in log.events if "secrets" in str(e.get("target", ""))), None)
    problems = []
    if escape is None:
        problems.append("the escape attempt reached no handler, so nothing was tested")
    elif escape.get("handler") != "handlers/static-file":
        problems.append(f"the escape attempt went to {escape.get('handler')}, not the file handler")
    elif escape.get("status") != 403:
        problems.append(f"the escape attempt returned {escape.get('status')}, not 403")

    if problems:
        print("  CONFINEMENT NOT DEMONSTRATED: " + "; ".join(problems))
        return 1
    print("  confinement demonstrated: the file handler refused a path outside its root")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())