| 1 | """HTTP/1.x request counting on top of the rootless byte proxy.
|
| 2 |
|
| 3 | ``CountingProxy`` observes TCP connections and bytes. Loopback benchmark
|
| 4 | transports are plaintext HTTP/1.1, so the client-to-server byte stream can also
|
| 5 | be framed passively into requests: request line + headers, then a
|
| 6 | ``Content-Length`` or ``chunked`` body. Bytes are forwarded unchanged; the
|
| 7 | parser only watches them.
|
| 8 |
|
| 9 | Honesty rules (ADR-0002):
|
| 10 |
|
| 11 | - A stream the parser cannot frame (non-HTTP/1.x bytes, oversized headers,
|
| 12 | malformed chunk sizes, unsupported framing) marks the proxy's request counts
|
| 13 | as unobserved. ``RequestCountingProxy.request_stats()`` then reports
|
| 14 | ``request_count=None`` rather than a partial number that looks measured.
|
| 15 | - ``injected_request_delay_ms`` is a fixed sleep the proxy adds before
|
| 16 | forwarding the first byte of each request. It is an explicit, labeled
|
| 17 | harness delay, not a shaped network (no netem/tc, no bandwidth model, no
|
| 18 | response-direction latency).
|
| 19 | """
|
| 20 |
|
| 21 | from __future__ import annotations
|
| 22 |
|
| 23 | import socket
|
| 24 | import time
|
| 25 | from dataclasses import dataclass
|
| 26 | from typing import Optional
|
| 27 |
|
| 28 | from oakbench.byteproxy import CountingProxy
|
| 29 |
|
| 30 | MAX_HEADER_BYTES = 256 * 1024
|
| 31 | _METHODS = {b"GET", b"POST", b"PUT", b"DELETE", b"HEAD", b"PATCH", b"OPTIONS"}
|
| 32 |
|
| 33 |
|
| 34 | class HttpFramingError(ValueError):
|
| 35 | """The client stream is not HTTP/1.x framing this parser can follow."""
|
| 36 |
|
| 37 |
|
| 38 | @dataclass(frozen=True)
|
| 39 | class HttpRequest:
|
| 40 | method: str
|
| 41 | target: str
|
| 42 |
|
| 43 | @property
|
| 44 | def path(self) -> str:
|
| 45 | return self.target.split("?", 1)[0]
|
| 46 |
|
| 47 |
|
| 48 | class HttpRequestStreamParser:
|
| 49 | """Incremental framer for one client-to-server HTTP/1.x byte stream."""
|
| 50 |
|
| 51 | def __init__(self) -> None:
|
| 52 | self._buffer = bytearray()
|
| 53 | self._state = "idle" # idle | headers | body | chunk_size | chunk_data | chunk_crlf | trailers
|
| 54 | self._remaining = 0
|
| 55 | self.requests: list[HttpRequest] = []
|
| 56 |
|
| 57 | @property
|
| 58 | def at_request_boundary(self) -> bool:
|
| 59 | return self._state == "idle" and not self._buffer
|
| 60 |
|
| 61 | def feed(self, data: bytes) -> int:
|
| 62 | """Consume bytes; return how many new requests *started* in ``data``."""
|
| 63 | started = 0
|
| 64 | view = memoryview(data)
|
| 65 | index = 0
|
| 66 | while index < len(view):
|
| 67 | if self._state == "idle":
|
| 68 | self._state = "headers"
|
| 69 | started += 1
|
| 70 | if self._state == "headers":
|
| 71 | self._buffer.extend(view[index:])
|
| 72 | index = len(view)
|
| 73 | end = self._buffer.find(b"\r\n\r\n")
|
| 74 | if end < 0:
|
| 75 | if len(self._buffer) > MAX_HEADER_BYTES:
|
| 76 | raise HttpFramingError("request headers exceed limit")
|
| 77 | break
|
| 78 | header = bytes(self._buffer[:end])
|
| 79 | rest = bytes(self._buffer[end + 4 :])
|
| 80 | self._buffer.clear()
|
| 81 | self._start_body(header)
|
| 82 | if rest:
|
| 83 | view = memoryview(rest)
|
| 84 | index = 0
|
| 85 | continue
|
| 86 | if self._state == "body":
|
| 87 | take = min(self._remaining, len(view) - index)
|
| 88 | self._remaining -= take
|
| 89 | index += take
|
| 90 | if self._remaining == 0:
|
| 91 | self._state = "idle"
|
| 92 | continue
|
| 93 | if self._state in {"chunk_size", "trailers", "chunk_crlf"}:
|
| 94 | self._buffer.extend(view[index : index + 1])
|
| 95 | index += 1
|
| 96 | if not self._buffer.endswith(b"\r\n"):
|
| 97 | if len(self._buffer) > MAX_HEADER_BYTES:
|
| 98 | raise HttpFramingError("chunk framing line exceeds limit")
|
| 99 | continue
|
| 100 | line = bytes(self._buffer[:-2])
|
| 101 | self._buffer.clear()
|
| 102 | if self._state == "chunk_crlf":
|
| 103 | if line:
|
| 104 | raise HttpFramingError("missing CRLF after chunk data")
|
| 105 | self._state = "chunk_size"
|
| 106 | elif self._state == "chunk_size":
|
| 107 | try:
|
| 108 | size = int(line.split(b";", 1)[0].strip(), 16)
|
| 109 | except ValueError as error:
|
| 110 | raise HttpFramingError("malformed chunk size") from error
|
| 111 | if size == 0:
|
| 112 | self._state = "trailers"
|
| 113 | else:
|
| 114 | self._remaining = size
|
| 115 | self._state = "chunk_data"
|
| 116 | else: # trailers: an empty line ends the message
|
| 117 | if not line:
|
| 118 | self._state = "idle"
|
| 119 | continue
|
| 120 | if self._state == "chunk_data":
|
| 121 | take = min(self._remaining, len(view) - index)
|
| 122 | self._remaining -= take
|
| 123 | index += take
|
| 124 | if self._remaining == 0:
|
| 125 | self._state = "chunk_crlf"
|
| 126 | continue
|
| 127 | return started
|
| 128 |
|
| 129 | def _start_body(self, header: bytes) -> None:
|
| 130 | lines = header.split(b"\r\n")
|
| 131 | parts = lines[0].split(b" ")
|
| 132 | if len(parts) != 3 or parts[0] not in _METHODS or not parts[2].startswith(b"HTTP/1."):
|
| 133 | raise HttpFramingError("not an HTTP/1.x request line")
|
| 134 | length: int | None = None
|
| 135 | chunked = False
|
| 136 | for line in lines[1:]:
|
| 137 | name, sep, value = line.partition(b":")
|
| 138 | if not sep:
|
| 139 | raise HttpFramingError("malformed header line")
|
| 140 | key = name.strip().lower()
|
| 141 | if key == b"content-length":
|
| 142 | try:
|
| 143 | length = int(value.strip())
|
| 144 | except ValueError as error:
|
| 145 | raise HttpFramingError("malformed content-length") from error
|
| 146 | elif key == b"transfer-encoding":
|
| 147 | codings = [item.strip().lower() for item in value.split(b",")]
|
| 148 | if codings[-1:] != [b"chunked"]:
|
| 149 | raise HttpFramingError("unsupported transfer-encoding")
|
| 150 | chunked = True
|
| 151 | elif key == b"upgrade":
|
| 152 | raise HttpFramingError("protocol upgrade is not framed")
|
| 153 | self.requests.append(
|
| 154 | HttpRequest(parts[0].decode("ascii"), parts[1].decode("latin-1"))
|
| 155 | )
|
| 156 | if chunked:
|
| 157 | self._state = "chunk_size"
|
| 158 | elif length:
|
| 159 | self._remaining = length
|
| 160 | self._state = "body"
|
| 161 | else:
|
| 162 | self._state = "idle"
|
| 163 |
|
| 164 |
|
| 165 | @dataclass(frozen=True)
|
| 166 | class RequestStats:
|
| 167 | """Request-level observation; ``None`` means the stream was not framable."""
|
| 168 |
|
| 169 | request_count: int | None
|
| 170 | requests_by_path_suffix: dict[str, int | None]
|
| 171 | methods: dict[str, int] | None
|
| 172 | framing_errors: int
|
| 173 | injected_request_delay_ms: float
|
| 174 |
|
| 175 | def count_suffix(self, suffix: str) -> int | None:
|
| 176 | return self.requests_by_path_suffix.get(suffix)
|
| 177 |
|
| 178 |
|
| 179 | class RequestCountingProxy(CountingProxy):
|
| 180 | """Byte proxy that also frames and counts client HTTP/1.x requests."""
|
| 181 |
|
| 182 | def __init__(
|
| 183 | self,
|
| 184 | upstream_host: str,
|
| 185 | upstream_port: int,
|
| 186 | *,
|
| 187 | path_suffixes: tuple[str, ...] = (),
|
| 188 | injected_request_delay_ms: float = 0.0,
|
| 189 | **kwargs, # type: ignore[no-untyped-def]
|
| 190 | ) -> None:
|
| 191 | if injected_request_delay_ms < 0:
|
| 192 | raise ValueError("injected_request_delay_ms must be >= 0")
|
| 193 | super().__init__(upstream_host, upstream_port, **kwargs)
|
| 194 | self.path_suffixes = tuple(path_suffixes)
|
| 195 | self.injected_request_delay_ms = float(injected_request_delay_ms)
|
| 196 | self._requests: list[HttpRequest] = []
|
| 197 | self._framing_errors = 0
|
| 198 |
|
| 199 | def reset_counts(self) -> None:
|
| 200 | super().reset_counts()
|
| 201 | with self._lock:
|
| 202 | self._requests = []
|
| 203 | self._framing_errors = 0
|
| 204 |
|
| 205 | def request_stats(self) -> RequestStats:
|
| 206 | with self._lock:
|
| 207 | requests = list(self._requests)
|
| 208 | errors = self._framing_errors
|
| 209 | if errors:
|
| 210 | return RequestStats(
|
| 211 | request_count=None,
|
| 212 | requests_by_path_suffix={suffix: None for suffix in self.path_suffixes},
|
| 213 | methods=None,
|
| 214 | framing_errors=errors,
|
| 215 | injected_request_delay_ms=self.injected_request_delay_ms,
|
| 216 | )
|
| 217 | methods: dict[str, int] = {}
|
| 218 | for request in requests:
|
| 219 | methods[request.method] = methods.get(request.method, 0) + 1
|
| 220 | return RequestStats(
|
| 221 | request_count=len(requests),
|
| 222 | requests_by_path_suffix={
|
| 223 | suffix: sum(1 for request in requests if request.path.endswith(suffix))
|
| 224 | for suffix in self.path_suffixes
|
| 225 | },
|
| 226 | methods=methods,
|
| 227 | framing_errors=0,
|
| 228 | injected_request_delay_ms=self.injected_request_delay_ms,
|
| 229 | )
|
| 230 |
|
| 231 | def request_log(self) -> list[HttpRequest]:
|
| 232 | with self._lock:
|
| 233 | return list(self._requests)
|
| 234 |
|
| 235 | def _pump(self, source: socket.socket, dest: socket.socket, direction: str) -> None:
|
| 236 | if direction != "client_to_upstream":
|
| 237 | super()._pump(source, dest, direction)
|
| 238 | return
|
| 239 | parser: Optional[HttpRequestStreamParser] = HttpRequestStreamParser()
|
| 240 | delay_s = self.injected_request_delay_ms / 1000.0
|
| 241 | try:
|
| 242 | while not self._stop.is_set():
|
| 243 | chunk = source.recv(64 * 1024)
|
| 244 | if not chunk:
|
| 245 | break
|
| 246 | started = 0
|
| 247 | if parser is not None:
|
| 248 | seen = len(parser.requests)
|
| 249 | try:
|
| 250 | started = parser.feed(chunk)
|
| 251 | except HttpFramingError:
|
| 252 | parser = None
|
| 253 | with self._lock:
|
| 254 | self._framing_errors += 1
|
| 255 | else:
|
| 256 | with self._lock:
|
| 257 | self._requests.extend(parser.requests[seen:])
|
| 258 | if started and delay_s > 0:
|
| 259 | time.sleep(delay_s * started)
|
| 260 | with self._lock:
|
| 261 | self._client_to_upstream_bytes += len(chunk)
|
| 262 | dest.sendall(chunk)
|
| 263 | except OSError:
|
| 264 | pass
|
| 265 | finally:
|
| 266 | if parser is not None and not parser.at_request_boundary:
|
| 267 | # A connection that ended mid-request (e.g. client abort) is
|
| 268 | # still counted by its header; a mid-header cut is unframed.
|
| 269 | if parser._state == "headers":
|
| 270 | with self._lock:
|
| 271 | self._framing_errors += 1
|
| 272 | try:
|
| 273 | dest.shutdown(socket.SHUT_WR)
|
| 274 | except OSError:
|
| 275 | pass
|
| 276 |
|
| 277 |
|
| 278 | __all__ = [
|
| 279 | "HttpFramingError",
|
| 280 | "HttpRequest",
|
| 281 | "HttpRequestStreamParser",
|
| 282 | "RequestCountingProxy",
|
| 283 | "RequestStats",
|
| 284 | ]
|