From 1742ba3f088c827af645ed0f38b45fb4814e0317 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Fri, 21 Aug 2026 23:17:33 -0500 Subject: [PATCH 1/5] feat: add timeout-bounded regex routing --- README.md | 58 +++++++- benchmarks/route_benchmark.py | 50 +++++++ pyproject.toml | 1 + smallserver/__init__.py | 4 + smallserver/app.py | 60 +++++--- smallserver/http.py | 44 +++++- smallserver/routing.py | 265 ++++++++++++++++++++++++++++++++++ smallserver/server.py | 33 ++++- tests/test_http.py | 26 ++++ tests/test_regex_routing.py | 231 +++++++++++++++++++++++++++++ tests/test_routing.py | 16 ++ tests/test_server.py | 18 +++ tests/test_server_runtime.py | 37 +++++ 13 files changed, 810 insertions(+), 33 deletions(-) create mode 100644 benchmarks/route_benchmark.py create mode 100644 smallserver/routing.py create mode 100644 tests/test_regex_routing.py diff --git a/README.md b/README.md index 37cbe90..0992a4e 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,17 @@ # SmallServer SmallServer is a SmallOS-native web framework in early development. It provides -a bounded HTTP/1.1 server, static async routing for GET, POST, PUT, PATCH, and -DELETE, and explicit escape hatches for blocking and asyncio-native libraries. +a bounded HTTP/1.1 server and async routing for GET, POST, PUT, PATCH, and +DELETE. Static routes are built in, timeout-bounded regular-expression routes +are available through an optional dependency, and explicit escape hatches +support blocking and asyncio-native libraries. ## Current scope The current package provides an HTTP/1.1 baseline over a SmallOS runtime. It can: -- register static async routes for GET, POST, PUT, PATCH, and DELETE; +- register static or optional regular-expression async routes for GET, POST, + PUT, PATCH, and DELETE; - dispatch an already-created `Request` to a handler; - return deterministic `Response` values, including HTTP/1.1 bytes; - return 404 for an unknown path and 405 with `Allow` for a known path using @@ -18,7 +21,8 @@ The current package provides an HTTP/1.1 baseline over a SmallOS runtime. It can - parse one `Content-Length` HTTP/1.1 request per connection and close after its response. -Keep-alive/pipelining, TLS, path parameters, and HTTP/2 are not implemented yet. +Keep-alive/pipelining, TLS, automatic path templates, and HTTP/2 are not +implemented yet. ## Install for development @@ -28,6 +32,14 @@ python3 -m pip install -e . python3 -m unittest discover -s tests -v ``` +Install the optional matching engine when an application uses raw regex routes: + +```bash +python3 -m pip install -e '.[regex-routes]' +``` + +Static routing neither imports nor requires that dependency. + SmallOS is installed from the canonical `master` branch in `requirements.txt`. It owns scheduling, socket readiness, and foreign execution adapters. @@ -101,8 +113,42 @@ async def delete_widgets(request: Request) -> Response: return Response(status=204) ``` -Route paths are static in this release. Path parameters and richer lifecycle -hooks are deferred; the current `ServerHandle` provides explicit shutdown. +Static route lookup is dictionary-based and always takes precedence over a +regex route for the same method and path. Richer lifecycle hooks are deferred; +the current `ServerHandle` provides explicit shutdown. + +## Define regular-expression routes + +Regex routes use full-path matching and run in registration order after static +lookup. Only named captures become immutable `request.path_params`; an optional +group that did not participate is omitted. + +```python +@app.get_regex(r"/users/(?P[0-9]+)") +async def get_user(request: Request) -> Response: + return Response.json({"user_id": request.path_params["user_id"]}) + +@app.route_regex( + r"/articles/(?P[a-z0-9]+(?:-[a-z0-9]+)*)", + methods=("GET", "PATCH"), +) +async def article(request: Request) -> Response: + return Response.json({"slug": request.path_params["slug"]}) +``` + +Patterns must begin with a literal `/` and do not need `^` or `$`. They are +trusted application configuration, but paths are hostile input: SmallServer +bounds pattern length, route count, named captures, path bytes, each match, and +the total matching time. Prefer unambiguous repetition and narrow character +classes even with these deadlines. A timeout raises `RouteMatchTimeout` with an +opaque route ID and becomes a sanitized 500 response on the network path. + +Requests retain the exact ASCII origin-form target in `request.raw_target`. +Routing uses `request.path`, which excludes the query string; +`request.query_string` contains the raw text after `?`. Neither paths nor named +captures are percent-decoded, so `/files/a%2Fb` remains distinct from +`/files/a/b`. `request.route_pattern` identifies the selected static path or +regex pattern. ## Dispatch a request diff --git a/benchmarks/route_benchmark.py b/benchmarks/route_benchmark.py new file mode 100644 index 0000000..29bb1b2 --- /dev/null +++ b/benchmarks/route_benchmark.py @@ -0,0 +1,50 @@ +"""Small routing microbenchmark; run with ``python benchmarks/route_benchmark.py``.""" + +from __future__ import annotations + +import asyncio +import importlib.util +from pathlib import Path +import sys +import time + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from smallserver import Headers, RegexRouteConfig, Request, Response, RouteMatchTimeout, SmallServer + + +async def benchmark() -> None: + app = SmallServer() + + @app.get("/health") + async def health(request): + return Response() + + request = Request("GET", "/health", Headers()) + iterations = 25_000 + started = time.perf_counter() + for _ in range(iterations): + await app.dispatch(request) + static_elapsed = time.perf_counter() - started + print("static: {:.0f} dispatches/second".format(iterations / static_elapsed)) + + if importlib.util.find_spec("regex") is None: + print("regex: skipped (install smallserver[regex-routes])") + return + + bounded = SmallServer(RegexRouteConfig(match_timeout=0.002, total_match_timeout=0.005)) + + @bounded.get_regex(r"/(a+)+$") + async def hostile(request): + return Response() + + started = time.perf_counter() + try: + await bounded.dispatch(Request("GET", "/" + "a" * 5000 + "!", Headers())) + except RouteMatchTimeout: + pass + print("worst-case regex timeout: {:.4f}s".format(time.perf_counter() - started)) + + +if __name__ == "__main__": + asyncio.run(benchmark()) diff --git a/pyproject.toml b/pyproject.toml index 337debe..d78894a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,6 +14,7 @@ dependencies = [] [project.optional-dependencies] dev = ["build>=1.2"] +regex-routes = ["regex>=2023.10.3"] [tool.setuptools.packages.find] include = ["smallserver*"] diff --git a/smallserver/__init__.py b/smallserver/__init__.py index a421318..e470f75 100644 --- a/smallserver/__init__.py +++ b/smallserver/__init__.py @@ -4,6 +4,7 @@ from .adapters import AdapterRegistry, AdapterShutdownError, http_error_from_adapter from .errors import HTTPError from .http import Headers, Request, Response +from .routing import RegexRouteConfig, RegexRoutesUnavailable, RouteMatchTimeout from .server import ServerConfig, ServerHandle __all__ = [ @@ -11,8 +12,11 @@ "AdapterShutdownError", "Headers", "HTTPError", + "RegexRouteConfig", + "RegexRoutesUnavailable", "Request", "Response", + "RouteMatchTimeout", "ServerConfig", "ServerHandle", "SmallServer", diff --git a/smallserver/app.py b/smallserver/app.py index 416542f..aabb735 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -4,40 +4,35 @@ import inspect from collections.abc import Awaitable, Callable, Iterable +from dataclasses import replace import socket from typing import Any from .errors import HTTPError from .http import Request, Response +from .routing import RegexRouteConfig, Router from .server import HTTPParseError, HTTPRequestParser, ServerConfig, ServerHandle Handler = Callable[[Request], Awaitable[Response]] -_METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"}) class SmallServer: """Register static HTTP routes and dispatch requests to async handlers.""" - def __init__(self) -> None: - self._routes: dict[tuple[str, str], Handler] = {} + def __init__(self, regex_config: RegexRouteConfig | None = None) -> None: + self._router = Router(regex_config) def route(self, path: str, methods: Iterable[str]) -> Callable[[Handler], Handler]: if not isinstance(path, str) or not path.startswith("/"): raise ValueError("route path must start with '/'") - normalized = tuple(dict.fromkeys(method.upper() for method in methods)) - if not normalized or any(method not in _METHODS for method in normalized): - raise ValueError("routes must use one or more supported HTTP methods") + if "?" in path or "#" in path: + raise ValueError("route path must not contain a query string or fragment") + normalized = self._router.normalize_methods(methods) def register(handler: Handler) -> Handler: if not callable(handler): raise TypeError("route handler must be callable") - keys = [(method, path) for method in normalized] - for method, key_path in keys: - key = (method, key_path) - if key in self._routes: - raise ValueError("route already registered: {} {}".format(method, path)) - for key in keys: - self._routes[key] = handler + self._router.add_static(path, normalized, handler) return handler return register @@ -57,6 +52,33 @@ def patch(self, path: str) -> Callable[[Handler], Handler]: def delete(self, path: str) -> Callable[[Handler], Handler]: return self.route(path, ("DELETE",)) + def route_regex(self, pattern: str, methods: Iterable[str]) -> Callable[[Handler], Handler]: + """Register a timeout-bounded full-path regular-expression route.""" + normalized = self._router.normalize_methods(methods) + + def register(handler: Handler) -> Handler: + if not callable(handler): + raise TypeError("route handler must be callable") + self._router.add_regex(pattern, normalized, handler) + return handler + + return register + + def get_regex(self, pattern: str) -> Callable[[Handler], Handler]: + return self.route_regex(pattern, ("GET",)) + + def post_regex(self, pattern: str) -> Callable[[Handler], Handler]: + return self.route_regex(pattern, ("POST",)) + + def put_regex(self, pattern: str) -> Callable[[Handler], Handler]: + return self.route_regex(pattern, ("PUT",)) + + def patch_regex(self, pattern: str) -> Callable[[Handler], Handler]: + return self.route_regex(pattern, ("PATCH",)) + + def delete_regex(self, pattern: str) -> Callable[[Handler], Handler]: + return self.route_regex(pattern, ("DELETE",)) + def serve( self, runtime: Any, @@ -110,12 +132,13 @@ def serve( async def dispatch(self, request: Request) -> Response: """Run a registered handler or return a deterministic HTTP response.""" - handler = self._routes.get((request.method.upper(), request.path)) - if handler is None: - allowed = sorted(method for method, path in self._routes if path == request.path) - if allowed: - return Response.text("method not allowed", status=405, headers={"Allow": ", ".join(allowed)}) + match = self._router.resolve(request.method, request.path) + if match.handler is None: + if match.allowed_methods: + return Response.text("method not allowed", status=405, headers={"Allow": ", ".join(match.allowed_methods)}) return Response.text("not found", status=404) + handler = match.handler + request = replace(request, path_params=match.path_params, route_pattern=match.route_pattern) try: result = handler(request) if not inspect.isawaitable(result): @@ -169,6 +192,7 @@ async def _connection_loop(self, task: Any, handle: ServerHandle, client: socket handle._config.max_header_bytes, handle._config.max_header_count, handle._config.max_body_bytes, + handle._config.max_request_target_bytes, ) try: while not handle.closed: diff --git a/smallserver/http.py b/smallserver/http.py index 42b43c8..f90afbd 100644 --- a/smallserver/http.py +++ b/smallserver/http.py @@ -3,14 +3,25 @@ from __future__ import annotations from collections.abc import Iterable, Iterator, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field import json import re from types import MappingProxyType from typing import Any _TOKEN = re.compile(r"^[!#$%&'*+.^_`|~0-9A-Za-z-]+$") -_REASONS = {200: "OK", 201: "Created", 204: "No Content", 400: "Bad Request", 404: "Not Found", 405: "Method Not Allowed", 413: "Payload Too Large", 500: "Internal Server Error", 503: "Service Unavailable"} +_REASONS = { + 200: "OK", + 201: "Created", + 204: "No Content", + 400: "Bad Request", + 404: "Not Found", + 405: "Method Not Allowed", + 413: "Payload Too Large", + 414: "URI Too Long", + 500: "Internal Server Error", + 503: "Service Unavailable", +} class Headers(Mapping[str, str]): @@ -60,16 +71,45 @@ class Request: headers: Headers body: bytes = b"" version: str = "HTTP/1.1" + raw_target: str | None = None + query_string: str = "" + path_params: Mapping[str, str] = field(default_factory=dict) + route_pattern: str | None = None def __post_init__(self) -> None: if not _TOKEN.fullmatch(self.method): raise ValueError("invalid HTTP method") + if not isinstance(self.query_string, str): + raise TypeError("query_string must be a string") + if self.raw_target is None and "?" not in self.path and self.query_string: + raw_target = self.path + "?" + self.query_string + else: + raw_target = self.path if self.raw_target is None else self.raw_target + if not isinstance(raw_target, str) or not raw_target.startswith("/"): + raise ValueError("request path/target must start with '/'") + target_path, separator, target_query = raw_target.partition("?") + if self.raw_target is None: + if self.query_string and self.query_string != (target_query if separator else ""): + raise ValueError("request target fields are inconsistent") + object.__setattr__(self, "path", target_path) + object.__setattr__(self, "query_string", target_query if separator else "") + object.__setattr__(self, "raw_target", raw_target) + elif self.path != target_path or self.query_string != (target_query if separator else ""): + raise ValueError("request target fields are inconsistent") if not self.path.startswith("/"): raise ValueError("request path must start with '/'") if not isinstance(self.headers, Headers): object.__setattr__(self, "headers", Headers(self.headers)) if not isinstance(self.body, bytes): raise TypeError("request body must be bytes") + if not isinstance(self.path_params, Mapping): + raise TypeError("path_params must be a mapping") + params = dict(self.path_params) + if any(not isinstance(name, str) or not isinstance(value, str) for name, value in params.items()): + raise TypeError("path_params must map strings to strings") + object.__setattr__(self, "path_params", MappingProxyType(params)) + if self.route_pattern is not None and not isinstance(self.route_pattern, str): + raise TypeError("route_pattern must be a string or None") @dataclass(frozen=True) diff --git a/smallserver/routing.py b/smallserver/routing.py new file mode 100644 index 0000000..a6c9eff --- /dev/null +++ b/smallserver/routing.py @@ -0,0 +1,265 @@ +"""Deterministic static and timeout-bounded regular-expression routing.""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable, Iterable, Mapping +from dataclasses import dataclass +import importlib +import math +import time +from types import MappingProxyType +from typing import Any + +from .http import Request, Response + +Handler = Callable[[Request], Awaitable[Response]] +SUPPORTED_METHODS = frozenset({"GET", "POST", "PUT", "PATCH", "DELETE"}) + + +class RegexRoutesUnavailable(RuntimeError): + """Raised when regex routes are used without their optional dependency.""" + + +class RouteMatchTimeout(RuntimeError): + """Raised when a bounded regex route match exceeds its deadline.""" + + def __init__(self, route_id: str) -> None: + self.route_id = route_id + super().__init__("regular-expression route matching timed out ({})".format(route_id)) + + +@dataclass(frozen=True) +class RegexRouteConfig: + """Finite limits applied to regex registration and hostile request paths.""" + + max_path_bytes: int = 8 * 1024 + max_pattern_length: int = 1024 + max_routes: int = 100 + match_timeout: float = 0.01 + total_match_timeout: float = 0.05 + max_named_captures: int = 20 + + def __post_init__(self) -> None: + for name in ("max_path_bytes", "max_pattern_length", "max_routes", "max_named_captures"): + value = getattr(self, name) + if type(value) is not int or value <= 0: + raise ValueError("{} must be a positive integer".format(name)) + for name in ("match_timeout", "total_match_timeout"): + value = getattr(self, name) + if type(value) not in (int, float) or not math.isfinite(value) or value <= 0: + raise ValueError("{} must be a finite positive number".format(name)) + + +@dataclass(frozen=True) +class RouteMatch: + handler: Handler | None + path_params: Mapping[str, str] + route_pattern: str | None + allowed_methods: tuple[str, ...] = () + + +@dataclass(frozen=True) +class _RegexRoute: + route_id: str + pattern: str + compiled: Any + handlers: Mapping[str, Handler] + literal_prefix: str + + +class Router: + """Resolve static and ordered regex routes without slowing static lookup.""" + + def __init__(self, regex_config: RegexRouteConfig | None = None) -> None: + self._static: dict[tuple[str, str], Handler] = {} + self._regex: list[_RegexRoute] = [] + self._regex_by_pattern: dict[str, int] = {} + self.regex_config = regex_config or RegexRouteConfig() + + @staticmethod + def normalize_methods(methods: Iterable[str]) -> tuple[str, ...]: + try: + normalized = tuple(dict.fromkeys(method.upper() for method in methods)) + except AttributeError as exc: + raise ValueError("routes must use one or more supported HTTP methods") from exc + if not normalized or any(method not in SUPPORTED_METHODS for method in normalized): + raise ValueError("routes must use one or more supported HTTP methods") + return normalized + + def add_static(self, path: str, methods: tuple[str, ...], handler: Handler) -> None: + keys = [(method, path) for method in methods] + for key in keys: + if key in self._static: + raise ValueError("route already registered: {} {}".format(key[0], path)) + for key in keys: + self._static[key] = handler + + def add_regex(self, pattern: str, methods: tuple[str, ...], handler: Handler) -> None: + if not isinstance(pattern, str): + raise TypeError("regex route pattern must be a string") + existing_index = self._regex_by_pattern.get(pattern) + if existing_index is not None: + existing = self._regex[existing_index] + duplicate = next((method for method in methods if method in existing.handlers), None) + if duplicate is not None: + raise ValueError("regex route already registered: {}".format(duplicate)) + handlers = dict(existing.handlers) + handlers.update((method, handler) for method in methods) + self._regex[existing_index] = _RegexRoute( + existing.route_id, + existing.pattern, + existing.compiled, + MappingProxyType(handlers), + existing.literal_prefix, + ) + return + + if len(self._regex) >= self.regex_config.max_routes: + raise ValueError("maximum registered regex routes exceeded") + compiled = self._compile(pattern) + route = _RegexRoute( + "regex-route-{}".format(len(self._regex) + 1), + pattern, + compiled, + MappingProxyType({method: handler for method in methods}), + _literal_prefix(pattern), + ) + self._regex_by_pattern[pattern] = len(self._regex) + self._regex.append(route) + + def resolve(self, method: str, path: str) -> RouteMatch: + method = method.upper() + static = self._static.get((method, path)) + if static is not None: + return RouteMatch(static, MappingProxyType({}), path) + + if not self._regex: + allowed = tuple(sorted(method for method, registered_path in self._static if registered_path == path)) + return RouteMatch(None, MappingProxyType({}), None, allowed) + self._validate_path(path) + deadline = time.monotonic() + self.regex_config.total_match_timeout + cached: dict[int, Any] = {} + for index, route in enumerate(self._regex): + if method not in route.handlers: + continue + match = self._match(route, path, deadline) + cached[index] = match + if match is not None: + return RouteMatch( + route.handlers[method], + MappingProxyType(_captures(match)), + route.pattern, + ) + + allowed = {registered_method for registered_method, registered_path in self._static if registered_path == path} + for index, route in enumerate(self._regex): + match = cached.get(index) + if index not in cached: + match = self._match(route, path, deadline) + if match is not None: + allowed.update(route.handlers) + return RouteMatch(None, MappingProxyType({}), None, tuple(sorted(allowed))) + + def _compile(self, pattern: str) -> Any: + if not isinstance(pattern, str): + raise TypeError("regex route pattern must be a string") + if not pattern.startswith("/"): + raise ValueError("regex route pattern must start with a literal '/'") + if len(pattern) > self.regex_config.max_pattern_length: + raise ValueError("regex route pattern is too long") + names = _named_group_names(pattern) + if len(names) != len(set(names)): + raise ValueError("regex route pattern contains duplicate named groups") + if len(names) > self.regex_config.max_named_captures: + raise ValueError("regex route pattern has too many named captures") + try: + engine = importlib.import_module("regex") + except ImportError as exc: + raise RegexRoutesUnavailable( + "regular-expression routes require 'smallserver[regex-routes]'" + ) from exc + try: + compiled = engine.compile(pattern) + except Exception: + raise ValueError("invalid regex route pattern") from None + try: + empty_match = compiled.fullmatch("", timeout=self.regex_config.match_timeout) + except TimeoutError as exc: + raise ValueError("regex route pattern validation timed out") from exc + if empty_match is not None: + raise ValueError("regex route pattern must not match an empty path") + return compiled + + def _validate_path(self, path: str) -> None: + try: + size = len(path.encode("ascii")) + except UnicodeEncodeError as exc: + raise ValueError("request path must contain ASCII characters only") from exc + if size > self.regex_config.max_path_bytes: + raise ValueError("request path is too large for routing") + + def _match(self, route: _RegexRoute, path: str, deadline: float) -> Any: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RouteMatchTimeout(route.route_id) + if len(route.literal_prefix) > 1 and not path.startswith(route.literal_prefix): + return None + timeout = min(float(self.regex_config.match_timeout), remaining) + try: + return route.compiled.fullmatch(path, timeout=timeout) + except TimeoutError as exc: + raise RouteMatchTimeout(route.route_id) from exc + + +def _captures(match: Any) -> dict[str, str]: + return {name: value for name, value in match.groupdict().items() if value is not None} + + +def _literal_prefix(pattern: str) -> str: + """Return only the leading literals that are safe to use as a rejection index.""" + special = frozenset(".[](){}*+?|^$\\") + end = 0 + while end < len(pattern) and pattern[end] not in special: + end += 1 + return pattern[:end] + + +def _named_group_names(pattern: str) -> list[str]: + """Find named-group declarations while ignoring escapes and character classes.""" + names: list[str] = [] + escaped = False + in_class = False + index = 0 + while index < len(pattern): + character = pattern[index] + if escaped: + escaped = False + index += 1 + continue + if character == "\\": + escaped = True + index += 1 + continue + if character == "[": + in_class = True + index += 1 + continue + if character == "]" and in_class: + in_class = False + index += 1 + continue + marker_length = 0 + if not in_class and pattern.startswith("(?P<", index): + marker_length = 4 + elif not in_class and pattern.startswith("(?<", index): + next_character = pattern[index + 3 : index + 4] + if next_character not in ("=", "!"): + marker_length = 3 + if marker_length: + end = pattern.find(">", index + marker_length) + if end >= 0: + names.append(pattern[index + marker_length : end]) + index = end + 1 + continue + index += 1 + return names diff --git a/smallserver/server.py b/smallserver/server.py index 09701fe..a3510d6 100644 --- a/smallserver/server.py +++ b/smallserver/server.py @@ -21,10 +21,17 @@ def __init__(self, status: int, detail: str) -> None: class HTTPRequestParser: """Incrementally parse one bounded HTTP/1.1 request with Content-Length.""" - def __init__(self, max_header_bytes: int, max_header_count: int, max_body_bytes: int) -> None: + def __init__( + self, + max_header_bytes: int, + max_header_count: int, + max_body_bytes: int, + max_request_target_bytes: int = 8 * 1024, + ) -> None: self._max_header_bytes = max_header_bytes self._max_header_count = max_header_count self._max_body_bytes = max_body_bytes + self._max_request_target_bytes = max_request_target_bytes self._buffer = bytearray() self._request_head: tuple[str, str, Headers, int] | None = None @@ -42,13 +49,22 @@ def feed(self, data: bytes) -> Request | None: self._request_head = self._parse_head(bytes(self._buffer[:marker])) del self._buffer[:header_length] - method, path, headers, content_length = self._request_head + method, raw_target, headers, content_length = self._request_head if len(self._buffer) > content_length: raise HTTPParseError(400, "pipelined requests are not supported") if len(self._buffer) < content_length: return None try: - return Request(method, path, headers, bytes(self._buffer), "HTTP/1.1") + path, separator, query_string = raw_target.partition("?") + return Request( + method, + path, + headers, + bytes(self._buffer), + "HTTP/1.1", + raw_target=raw_target, + query_string=query_string if separator else "", + ) except ValueError as exc: raise HTTPParseError(400, str(exc)) from exc @@ -59,10 +75,12 @@ def _parse_head(self, raw: bytes) -> tuple[str, str, Headers, int]: raise HTTPParseError(400, "request headers are not valid bytes") from exc if not lines or len(lines[0].split(" ")) != 3: raise HTTPParseError(400, "malformed request line") - method, path, version = lines[0].split(" ") - if version != "HTTP/1.1" or not path.startswith("/"): + method, raw_target, version = lines[0].split(" ") + if version != "HTTP/1.1" or not raw_target.startswith("/"): raise HTTPParseError(400, "only origin-form HTTP/1.1 requests are supported") - if "#" in path or any(not 0x21 <= ord(character) <= 0x7E for character in path): + if len(raw_target.encode("iso-8859-1")) > self._max_request_target_bytes: + raise HTTPParseError(414, "request target is too large") + if "#" in raw_target or any(not 0x21 <= ord(character) <= 0x7E for character in raw_target): raise HTTPParseError(400, "request target is not valid origin-form") if len(lines) - 1 > self._max_header_count: raise HTTPParseError(413, "too many request headers") @@ -94,7 +112,7 @@ def _parse_head(self, raw: bytes) -> tuple[str, str, Headers, int]: raise HTTPParseError(413, "request body is too large") if not headers.get("host"): raise HTTPParseError(400, "HTTP/1.1 requests require a Host header") - return method, path, headers, length + return method, raw_target, headers, length @dataclass(frozen=True) @@ -105,6 +123,7 @@ class ServerConfig: max_header_bytes: int = 16 * 1024 max_header_count: int = 100 max_body_bytes: int = 1024 * 1024 + max_request_target_bytes: int = 8 * 1024 receive_chunk_bytes: int = 8 * 1024 listener_priority: int = 1 connection_priority: int = 2 diff --git a/tests/test_http.py b/tests/test_http.py index 841d1ee..95679a6 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -1,4 +1,5 @@ import unittest +from types import MappingProxyType from smallserver import Headers, Request, Response @@ -31,3 +32,28 @@ def test_header_values_match_the_wire_encoding(self) -> None: with self.subTest(invalid=invalid): with self.assertRaisesRegex(ValueError, "header value"): Headers({"X-Test": invalid}) + + def test_request_splits_raw_target_and_keeps_captures_immutable(self) -> None: + request = Request( + "GET", + "/items?tag=a%2Fb&empty=", + Headers(), + path_params={"item_id": "a%2Fb"}, + route_pattern=r"/items/(?P[^/]+)", + ) + self.assertEqual(request.raw_target, "/items?tag=a%2Fb&empty=") + self.assertEqual(request.path, "/items") + self.assertEqual(request.query_string, "tag=a%2Fb&empty=") + self.assertIsInstance(request.path_params, MappingProxyType) + with self.assertRaises(TypeError): + request.path_params["item_id"] = "changed" # type: ignore[index] + + def test_request_rejects_inconsistent_explicit_target_fields(self) -> None: + with self.assertRaisesRegex(ValueError, "inconsistent"): + Request("GET", "/one", Headers(), raw_target="/two") + + def test_request_can_be_built_from_separate_path_and_query(self) -> None: + request = Request("GET", "/items", Headers(), query_string="page=2") + self.assertEqual(request.raw_target, "/items?page=2") + self.assertEqual(request.path, "/items") + self.assertEqual(request.query_string, "page=2") diff --git a/tests/test_regex_routing.py b/tests/test_regex_routing.py new file mode 100644 index 0000000..fba24b4 --- /dev/null +++ b/tests/test_regex_routing.py @@ -0,0 +1,231 @@ +import importlib.util +import time +import unittest +from unittest.mock import patch + +from smallserver import ( + Headers, + RegexRouteConfig, + RegexRoutesUnavailable, + Request, + Response, + RouteMatchTimeout, + SmallServer, +) + + +HAS_REGEX = importlib.util.find_spec("regex") is not None + + +class OptionalRegexDependencyTests(unittest.TestCase): + def test_static_routes_do_not_import_optional_engine(self) -> None: + app = SmallServer() + + async def handler(request): + return Response() + + with patch("smallserver.routing.importlib.import_module", side_effect=AssertionError("imported")): + app.get("/health")(handler) + + def test_registration_explains_missing_extra(self) -> None: + app = SmallServer() + + async def handler(request): + return Response() + + with patch("smallserver.routing.importlib.import_module", side_effect=ImportError): + with self.assertRaisesRegex(RegexRoutesUnavailable, r"smallserver\[regex-routes\]"): + app.get_regex(r"/users/[0-9]+")(handler) + + +@unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") +class RegexRoutingTests(unittest.IsolatedAsyncioTestCase): + async def test_named_captures_are_immutable_and_optional_groups_are_omitted(self) -> None: + app = SmallServer() + seen = [] + + @app.get_regex(r"/users/(?P[0-9]+)(?:/(?P
[a-z]+))?") + async def user(request): + seen.append(request) + return Response.json(dict(request.path_params)) + + response = await app.dispatch(Request("GET", "/users/42?debug=1", Headers())) + self.assertEqual(response.body, b'{"user_id":"42"}') + self.assertEqual(seen[0].raw_target, "/users/42?debug=1") + self.assertEqual(seen[0].query_string, "debug=1") + self.assertEqual(seen[0].route_pattern, r"/users/(?P[0-9]+)(?:/(?P
[a-z]+))?") + with self.assertRaises(TypeError): + seen[0].path_params["user_id"] = "1" # type: ignore[index] + await app.dispatch(Request("GET", "/users/43/profile", Headers())) + self.assertEqual(dict(seen[0].path_params), {"user_id": "42"}) + self.assertEqual(dict(seen[1].path_params), {"user_id": "43", "section": "profile"}) + + async def test_captures_preserve_percent_encoded_octets(self) -> None: + app = SmallServer() + + @app.get_regex(r"/files/(?P[^/]+)") + async def file(request): + return Response.text(request.path_params["name"]) + + response = await app.dispatch(Request("GET", "/files/a%2Fb", Headers())) + self.assertEqual(response.body, b"a%2Fb") + + async def test_static_precedence_and_regex_registration_order(self) -> None: + app = SmallServer() + + @app.get_regex(r"/items/(?P.+)") + async def broad(request): + return Response.text("broad") + + @app.get_regex(r"/items/(?P[0-9]+)") + async def narrow(request): + return Response.text("narrow") + + @app.get("/items/7") + async def exact(request): + return Response.text("static") + + self.assertEqual((await app.dispatch(Request("GET", "/items/7", Headers()))).body, b"static") + self.assertEqual((await app.dispatch(Request("GET", "/items/8", Headers()))).body, b"broad") + + async def test_method_first_matching_and_sorted_allow(self) -> None: + app = SmallServer() + + @app.post("/records/1") + async def static_post(request): + return Response.text("post") + + @app.delete_regex(r"/records/(?P[0-9]+)") + async def regex_delete(request): + return Response.text("delete") + + @app.get_regex(r"/records/(?P.+)") + async def regex_get(request): + return Response.text("get") + + self.assertEqual((await app.dispatch(Request("GET", "/records/1", Headers()))).body, b"get") + response = await app.dispatch(Request("PATCH", "/records/1", Headers())) + self.assertEqual(response.status, 405) + self.assertEqual(response.headers["allow"], "DELETE, GET, POST") + + async def test_all_regex_method_decorators_dispatch(self) -> None: + app = SmallServer() + for method in ("get", "post", "put", "patch", "delete"): + decorator = getattr(app, method + "_regex") + + @decorator("/" + method + r"/(?P[0-9]+)") + async def handler(request, expected=method): + return Response.text(expected + request.path_params["value"]) + + for method in ("get", "post", "put", "patch", "delete"): + response = await app.dispatch(Request(method.upper(), "/" + method + "/2", Headers())) + self.assertEqual(response.body, (method + "2").encode()) + + async def test_same_pattern_disjoint_methods_merge_and_duplicates_are_atomic(self) -> None: + app = SmallServer() + + async def first(request): + return Response.text("first") + + async def second(request): + return Response.text("second") + + app.get_regex(r"/merged/(?P[0-9]+)")(first) + app.post_regex(r"/merged/(?P[0-9]+)")(second) + with self.assertRaisesRegex(ValueError, "already registered"): + app.route_regex(r"/merged/(?P[0-9]+)", ("PATCH", "GET"))(second) + self.assertEqual((await app.dispatch(Request("PATCH", "/merged/1", Headers()))).status, 405) + self.assertEqual((await app.dispatch(Request("POST", "/merged/1", Headers()))).body, b"second") + + def test_registration_limits_and_invalid_patterns(self) -> None: + async def handler(request): + return Response() + + cases = ( + (r"items/[0-9]+", "literal '/'"), + (r"/[", "invalid"), + (r"/(?Px)(?Py)", "duplicate"), + ) + for pattern, message in cases: + with self.subTest(pattern=pattern): + with self.assertRaisesRegex((ValueError, TypeError), message): + SmallServer().get_regex(pattern)(handler) + + with self.assertRaisesRegex(ValueError, "too long"): + SmallServer(RegexRouteConfig(max_pattern_length=4)).get_regex("/long")(handler) + with self.assertRaisesRegex(ValueError, "too many"): + SmallServer(RegexRouteConfig(max_named_captures=1)).get_regex( + r"/(?Px)(?Py)" + )(handler) + limited = SmallServer(RegexRouteConfig(max_routes=1)) + limited.get_regex(r"/one")(handler) + with self.assertRaisesRegex(ValueError, "maximum"): + limited.get_regex(r"/two")(handler) + + async def test_catastrophic_backtracking_is_bounded_and_path_is_not_disclosed(self) -> None: + app = SmallServer(RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005)) + + @app.get_regex(r"/(a+)+$") + async def handler(request): + return Response() + + hostile_path = "/" + "a" * 5000 + "!" + started = time.monotonic() + with self.assertRaises(RouteMatchTimeout) as raised: + await app.dispatch(Request("GET", hostile_path, Headers())) + elapsed = time.monotonic() - started + self.assertLess(elapsed, 0.25) + self.assertNotIn(hostile_path, str(raised.exception)) + self.assertEqual(raised.exception.route_id, "regex-route-1") + + async def test_total_budget_bounds_many_individually_fast_misses(self) -> None: + class SlowPattern: + def fullmatch(self, value, timeout): + if not value: + return None + time.sleep(min(timeout / 2, 0.001)) + return None + + class Engine: + @staticmethod + def compile(pattern): + return SlowPattern() + + app = SmallServer( + RegexRouteConfig(max_routes=20, match_timeout=0.01, total_match_timeout=0.003) + ) + + async def handler(request): + return Response() + + with patch("smallserver.routing.importlib.import_module", return_value=Engine()): + for index in range(10): + app.get_regex("/(?:route-{}).*".format(index))(handler) + started = time.monotonic() + with self.assertRaises(RouteMatchTimeout): + await app.dispatch(Request("GET", "/route", Headers())) + self.assertLess(time.monotonic() - started, 0.05) + + async def test_manual_dispatch_path_limit_is_enforced_before_matching(self) -> None: + app = SmallServer(RegexRouteConfig(max_path_bytes=8)) + + @app.get_regex(r"/.*") + async def handler(request): + return Response() + + with self.assertRaisesRegex(ValueError, "too large"): + await app.dispatch(Request("GET", "/12345678", Headers())) + + +class RegexRouteConfigTests(unittest.TestCase): + def test_rejects_nonfinite_or_nonpositive_limits(self) -> None: + for kwargs in ( + {"max_routes": 0}, + {"max_routes": True}, + {"match_timeout": 0}, + {"match_timeout": float("inf")}, + {"total_match_timeout": float("nan")}, + ): + with self.subTest(kwargs=kwargs): + with self.assertRaises(ValueError): + RegexRouteConfig(**kwargs) diff --git a/tests/test_routing.py b/tests/test_routing.py index 19da695..d152500 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -30,6 +30,20 @@ async def items(request): self.assertEqual(response.status, 405) self.assertEqual(response.headers["allow"], "GET") + async def test_query_string_does_not_participate_in_static_matching(self) -> None: + app = SmallServer() + seen = [] + + @app.get("/items") + async def items(request): + seen.append(request) + return Response() + + response = await app.dispatch(Request("GET", "/items?tag=a%2Fb", Headers())) + self.assertEqual(response.status, 200) + self.assertEqual(seen[0].path, "/items") + self.assertEqual(seen[0].query_string, "tag=a%2Fb") + async def test_http_error_becomes_response(self) -> None: app = SmallServer() @@ -61,6 +75,8 @@ async def one(request): app.get("/one")(one) with self.assertRaisesRegex(ValueError, "supported HTTP methods"): app.route("/trace", ("TRACE",)) + with self.assertRaisesRegex(ValueError, "query string"): + app.get("/one?debug=1") async def test_failed_multi_method_registration_is_atomic(self) -> None: app = SmallServer() diff --git a/tests/test_server.py b/tests/test_server.py index bf6e193..27b1766 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -35,11 +35,29 @@ def test_requires_host_and_rejects_invalid_origin_form(self) -> None: with self.assertRaisesRegex(HTTPParseError, "origin-form"): self.parser().feed(b"GET /items#fragment HTTP/1.1\r\nHost: localhost\r\n\r\n") + def test_splits_query_without_decoding_or_normalizing_path(self) -> None: + request = self.parser().feed( + b"GET /items/a%2Fb?tag=x%20y HTTP/1.1\r\nHost: localhost\r\n\r\n" + ) + self.assertIsNotNone(request) + assert request is not None + self.assertEqual(request.raw_target, "/items/a%2Fb?tag=x%20y") + self.assertEqual(request.path, "/items/a%2Fb") + self.assertEqual(request.query_string, "tag=x%20y") + + def test_enforces_request_target_limit_independently(self) -> None: + parser = HTTPRequestParser(256, 2, 32, max_request_target_bytes=8) + with self.assertRaisesRegex(HTTPParseError, "request target") as raised: + parser.feed(b"GET /12345678 HTTP/1.1\r\nHost: x\r\n\r\n") + self.assertEqual(raised.exception.status, 414) + def test_config_rejects_unbounded_limits(self) -> None: with self.assertRaisesRegex(ValueError, "max_connections"): ServerConfig(max_connections=0) with self.assertRaisesRegex(ValueError, "max_connections"): ServerConfig(max_connections=True) + with self.assertRaisesRegex(ValueError, "max_request_target_bytes"): + ServerConfig(max_request_target_bytes=0) def test_serve_closes_bound_socket_when_runtime_fork_fails(self) -> None: class Listener: diff --git a/tests/test_server_runtime.py b/tests/test_server_runtime.py index 2bfa26b..9d80992 100644 --- a/tests/test_server_runtime.py +++ b/tests/test_server_runtime.py @@ -1,3 +1,4 @@ +import importlib.util import socket import threading import unittest @@ -8,6 +9,9 @@ from smallserver import AdapterRegistry, Response, SmallServer +HAS_REGEX = importlib.util.find_spec("regex") is not None + + class SmallOSServerIntegrationTests(unittest.TestCase): def _request(self, port: int, path: str) -> bytes: with socket.create_connection(("127.0.0.1", port), timeout=3) as connection: @@ -127,3 +131,36 @@ def fast_client() -> None: self.assertEqual(errors, []) self.assertIn(b"\r\n\r\nfast done", responses["fast"]) self.assertIn(b"\r\n\r\nslow done", responses["slow"]) + + @unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") + def test_loopback_regex_route_uses_path_without_query(self) -> None: + runtime = SmallOS().setKernel(Unix()) + app = SmallServer() + + @app.get_regex(r"/files/(?P[^/]+)") + async def file(request): + return Response.text(request.path_params["name"] + "?" + request.query_string) + + try: + server = app.serve(runtime, host="127.0.0.1", port=0) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + received = [] + errors = [] + + def client() -> None: + try: + received.append(self._request(server.port, "/files/a%2Fb?download=1")) + except BaseException as exc: + errors.append(exc) + finally: + server.close() + + worker = threading.Thread(target=client, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=2) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertIn(b"\r\n\r\na%2Fb?download=1", b"".join(received)) From 8625d569f734890a79a371e053f10867d0743e41 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Fri, 21 Aug 2026 23:37:26 -0500 Subject: [PATCH 2/5] fix: harden regex routing after review --- README.md | 31 +++++++++- benchmarks/route_benchmark.py | 58 ++++++++++++++--- pyproject.toml | 2 +- smallserver/__init__.py | 3 +- smallserver/app.py | 30 ++++++++- smallserver/routing.py | 110 +++++++++++++++++++++------------ smallserver/server.py | 2 +- tests/installed_regex_smoke.py | 21 +++++++ tests/test_regex_routing.py | 89 ++++++++++++++++++++++++-- tests/test_server.py | 11 ++++ tests/test_server_runtime.py | 85 ++++++++++++++++++++++++- tests/typing/regex_routes.py | 20 ++++++ 12 files changed, 399 insertions(+), 63 deletions(-) create mode 100644 tests/installed_regex_smoke.py create mode 100644 tests/typing/regex_routes.py diff --git a/README.md b/README.md index 0992a4e..991183c 100644 --- a/README.md +++ b/README.md @@ -136,13 +136,30 @@ async def article(request: Request) -> Response: return Response.json({"slug": request.path_params["slug"]}) ``` -Patterns must begin with a literal `/` and do not need `^` or `$`. They are -trusted application configuration, but paths are hostile input: SmallServer +Patterns must begin with a literal `/` and do not need `^` or `$`. SmallServer +wraps the complete expression in a slash guard, so every top-level alternative +is constrained to an origin-form path. They are trusted application +configuration, but paths are hostile input: SmallServer bounds pattern length, route count, named captures, path bytes, each match, and the total matching time. Prefer unambiguous repetition and narrow character classes even with these deadlines. A timeout raises `RouteMatchTimeout` with an opaque route ID and becomes a sanitized 500 response on the network path. +Applications can observe that failure without receiving the hostile target: + +```python +from smallserver import RouteMatchTimeout + +def observe_route_error(error: RouteMatchTimeout) -> None: + logger.error("route matching failed: %s", error.route_id) + +app = SmallServer(route_error_observer=observe_route_error) +``` + +The synchronous observer runs once on the connection task and should return +quickly; observer failures are isolated from the response path. A path above +the configured regex-routing byte limit returns 414 before matching begins. + Requests retain the exact ASCII origin-form target in `request.raw_target`. Routing uses `request.path`, which excludes the query string; `request.query_string` contains the raw text after `?`. Neither paths nor named @@ -150,6 +167,16 @@ captures are percent-decoded, so `/files/a%2Fb` remains distinct from `/files/a/b`. `request.route_pattern` identifies the selected static path or regex pattern. +Run `python benchmarks/route_benchmark.py` for a same-process comparison of +the pre-router dictionary dispatch model and current router dispatch. Its JSON +also records configured versus observed hostile-pattern timeout when the extra +is installed; rates are machine-specific and should be compared on the same +host. + +Release validation can run `python tests/installed_regex_smoke.py` from an +environment where the built `smallserver[regex-routes]` wheel is installed. +The project requires Python 3.10 or newer. + ## Dispatch a request The listener creates requests and calls `dispatch()`. The same boundary is diff --git a/benchmarks/route_benchmark.py b/benchmarks/route_benchmark.py index 29bb1b2..a3b247e 100644 --- a/benchmarks/route_benchmark.py +++ b/benchmarks/route_benchmark.py @@ -1,9 +1,12 @@ -"""Small routing microbenchmark; run with ``python benchmarks/route_benchmark.py``.""" +"""Repeatable routing comparison; run with ``python benchmarks/route_benchmark.py``.""" from __future__ import annotations +import argparse import asyncio import importlib.util +import inspect +import json from pathlib import Path import sys import time @@ -13,7 +16,19 @@ from smallserver import Headers, RegexRouteConfig, Request, Response, RouteMatchTimeout, SmallServer -async def benchmark() -> None: +async def _legacy_dispatch(routes, request): + """Model the pre-router static dictionary dispatch for same-run comparison.""" + handler = routes[(request.method.upper(), request.path)] + result = handler(request) + if not inspect.isawaitable(result): + raise TypeError("benchmark handler must be awaitable") + response = await result + if not isinstance(response, Response): + raise TypeError("benchmark handler must return Response") + return response + + +async def benchmark(iterations: int) -> dict[str, float | int | str]: app = SmallServer() @app.get("/health") @@ -21,16 +36,30 @@ async def health(request): return Response() request = Request("GET", "/health", Headers()) - iterations = 25_000 + legacy_routes = {("GET", "/health"): health} + + started = time.perf_counter() + for _ in range(iterations): + await _legacy_dispatch(legacy_routes, request) + legacy_elapsed = time.perf_counter() - started + started = time.perf_counter() for _ in range(iterations): await app.dispatch(request) - static_elapsed = time.perf_counter() - started - print("static: {:.0f} dispatches/second".format(iterations / static_elapsed)) + router_elapsed = time.perf_counter() - started + + legacy_rate = iterations / legacy_elapsed + router_rate = iterations / router_elapsed + result: dict[str, float | int | str] = { + "iterations": iterations, + "legacy_static_dispatches_per_second": round(legacy_rate, 2), + "router_static_dispatches_per_second": round(router_rate, 2), + "router_to_legacy_ratio": round(router_rate / legacy_rate, 4), + } if importlib.util.find_spec("regex") is None: - print("regex: skipped (install smallserver[regex-routes])") - return + result["regex"] = "skipped; install smallserver[regex-routes]" + return result bounded = SmallServer(RegexRouteConfig(match_timeout=0.002, total_match_timeout=0.005)) @@ -43,8 +72,19 @@ async def hostile(request): await bounded.dispatch(Request("GET", "/" + "a" * 5000 + "!", Headers())) except RouteMatchTimeout: pass - print("worst-case regex timeout: {:.4f}s".format(time.perf_counter() - started)) + result["configured_regex_match_timeout_seconds"] = 0.002 + result["observed_worst_case_regex_seconds"] = round(time.perf_counter() - started, 6) + return result + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--iterations", type=int, default=25_000) + arguments = parser.parse_args() + if arguments.iterations <= 0: + parser.error("--iterations must be positive") + print(json.dumps(asyncio.run(benchmark(arguments.iterations)), indent=2, sort_keys=True)) if __name__ == "__main__": - asyncio.run(benchmark()) + main() diff --git a/pyproject.toml b/pyproject.toml index d78894a..ced2d56 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,7 @@ dependencies = [] [project.optional-dependencies] dev = ["build>=1.2"] -regex-routes = ["regex>=2023.10.3"] +regex-routes = ["regex>=2023.10.3,<2027"] [tool.setuptools.packages.find] include = ["smallserver*"] diff --git a/smallserver/__init__.py b/smallserver/__init__.py index e470f75..476d64e 100644 --- a/smallserver/__init__.py +++ b/smallserver/__init__.py @@ -4,7 +4,7 @@ from .adapters import AdapterRegistry, AdapterShutdownError, http_error_from_adapter from .errors import HTTPError from .http import Headers, Request, Response -from .routing import RegexRouteConfig, RegexRoutesUnavailable, RouteMatchTimeout +from .routing import RegexRouteConfig, RegexRoutesUnavailable, RouteMatchTimeout, RoutePathTooLarge from .server import ServerConfig, ServerHandle __all__ = [ @@ -17,6 +17,7 @@ "Request", "Response", "RouteMatchTimeout", + "RoutePathTooLarge", "ServerConfig", "ServerHandle", "SmallServer", diff --git a/smallserver/app.py b/smallserver/app.py index aabb735..a009025 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -10,17 +10,26 @@ from .errors import HTTPError from .http import Request, Response -from .routing import RegexRouteConfig, Router +from .routing import RegexRouteConfig, RouteMatchTimeout, RoutePathTooLarge, Router from .server import HTTPParseError, HTTPRequestParser, ServerConfig, ServerHandle Handler = Callable[[Request], Awaitable[Response]] +RouteErrorObserver = Callable[[RouteMatchTimeout], None] class SmallServer: """Register static HTTP routes and dispatch requests to async handlers.""" - def __init__(self, regex_config: RegexRouteConfig | None = None) -> None: + def __init__( + self, + regex_config: RegexRouteConfig | None = None, + *, + route_error_observer: RouteErrorObserver | None = None, + ) -> None: + if route_error_observer is not None and not callable(route_error_observer): + raise TypeError("route_error_observer must be callable or None") self._router = Router(regex_config) + self._route_error_observer = route_error_observer def route(self, path: str, methods: Iterable[str]) -> Callable[[Handler], Handler]: if not isinstance(path, str) or not path.startswith("/"): @@ -132,7 +141,10 @@ def serve( async def dispatch(self, request: Request) -> Response: """Run a registered handler or return a deterministic HTTP response.""" - match = self._router.resolve(request.method, request.path) + try: + match = self._router.resolve(request.method, request.path) + except RoutePathTooLarge: + return Response.text("request target is too large", status=414) if match.handler is None: if match.allowed_methods: return Response.text("method not allowed", status=405, headers={"Allow": ", ".join(match.allowed_methods)}) @@ -214,6 +226,9 @@ async def _connection_loop(self, task: Any, handle: ServerHandle, client: socket continue try: response = await self.dispatch(request) + except RouteMatchTimeout as exc: + self._observe_route_error(exc) + response = Response.text("internal server error", status=500) except Exception: response = Response.text("internal server error", status=500) await self._send_response(task, client, response) @@ -239,3 +254,12 @@ async def _send_response(self, task: Any, client: socket.socket, response: Respo if sent <= 0: return offset += sent + + def _observe_route_error(self, error: RouteMatchTimeout) -> None: + observer = self._route_error_observer + if observer is None: + return + try: + observer(error) + except Exception: + pass diff --git a/smallserver/routing.py b/smallserver/routing.py index a6c9eff..b53588d 100644 --- a/smallserver/routing.py +++ b/smallserver/routing.py @@ -6,6 +6,7 @@ from dataclasses import dataclass import importlib import math +import re import time from types import MappingProxyType from typing import Any @@ -28,6 +29,10 @@ def __init__(self, route_id: str) -> None: super().__init__("regular-expression route matching timed out ({})".format(route_id)) +class RoutePathTooLarge(RuntimeError): + """Raised before matching when a request path exceeds its routing bound.""" + + @dataclass(frozen=True) class RegexRouteConfig: """Finite limits applied to regex registration and hostile request paths.""" @@ -64,7 +69,6 @@ class _RegexRoute: pattern: str compiled: Any handlers: Mapping[str, Handler] - literal_prefix: str class Router: @@ -110,7 +114,6 @@ def add_regex(self, pattern: str, methods: tuple[str, ...], handler: Handler) -> existing.pattern, existing.compiled, MappingProxyType(handlers), - existing.literal_prefix, ) return @@ -122,7 +125,6 @@ def add_regex(self, pattern: str, methods: tuple[str, ...], handler: Handler) -> pattern, compiled, MappingProxyType({method: handler for method in methods}), - _literal_prefix(pattern), ) self._regex_by_pattern[pattern] = len(self._regex) self._regex.append(route) @@ -167,11 +169,6 @@ def _compile(self, pattern: str) -> Any: raise ValueError("regex route pattern must start with a literal '/'") if len(pattern) > self.regex_config.max_pattern_length: raise ValueError("regex route pattern is too long") - names = _named_group_names(pattern) - if len(names) != len(set(names)): - raise ValueError("regex route pattern contains duplicate named groups") - if len(names) > self.regex_config.max_named_captures: - raise ValueError("regex route pattern has too many named captures") try: engine = importlib.import_module("regex") except ImportError as exc: @@ -179,15 +176,17 @@ def _compile(self, pattern: str) -> Any: "regular-expression routes require 'smallserver[regex-routes]'" ) from exc try: - compiled = engine.compile(pattern) + compiled = engine.compile("(?=/)(?:{})".format(pattern)) except Exception: raise ValueError("invalid regex route pattern") from None - try: - empty_match = compiled.fullmatch("", timeout=self.regex_config.match_timeout) - except TimeoutError as exc: - raise ValueError("regex route pattern validation timed out") from exc - if empty_match is not None: - raise ValueError("regex route pattern must not match an empty path") + names = _named_group_names(pattern) + compiled_names = set(compiled.groupindex) + if set(names) != compiled_names: + raise ValueError("regex route named groups could not be validated") + if len(names) != len(compiled_names): + raise ValueError("regex route pattern contains duplicate named groups") + if len(compiled_names) > self.regex_config.max_named_captures: + raise ValueError("regex route pattern has too many named captures") return compiled def _validate_path(self, path: str) -> None: @@ -196,14 +195,12 @@ def _validate_path(self, path: str) -> None: except UnicodeEncodeError as exc: raise ValueError("request path must contain ASCII characters only") from exc if size > self.regex_config.max_path_bytes: - raise ValueError("request path is too large for routing") + raise RoutePathTooLarge("request path is too large for routing") def _match(self, route: _RegexRoute, path: str, deadline: float) -> Any: remaining = deadline - time.monotonic() if remaining <= 0: raise RouteMatchTimeout(route.route_id) - if len(route.literal_prefix) > 1 and not path.startswith(route.literal_prefix): - return None timeout = min(float(self.regex_config.match_timeout), remaining) try: return route.compiled.fullmatch(path, timeout=timeout) @@ -215,18 +212,10 @@ def _captures(match: Any) -> dict[str, str]: return {name: value for name, value in match.groupdict().items() if value is not None} -def _literal_prefix(pattern: str) -> str: - """Return only the leading literals that are safe to use as a rejection index.""" - special = frozenset(".[](){}*+?|^$\\") - end = 0 - while end < len(pattern) and pattern[end] not in special: - end += 1 - return pattern[:end] - - def _named_group_names(pattern: str) -> list[str]: - """Find named-group declarations while ignoring escapes and character classes.""" + """Find declarations while honoring regex comments and scoped verbose mode.""" names: list[str] = [] + verbose_stack = [False] escaped = False in_class = False index = 0 @@ -248,18 +237,59 @@ def _named_group_names(pattern: str) -> list[str]: in_class = False index += 1 continue - marker_length = 0 - if not in_class and pattern.startswith("(?P<", index): - marker_length = 4 - elif not in_class and pattern.startswith("(?<", index): - next_character = pattern[index + 3 : index + 4] - if next_character not in ("=", "!"): - marker_length = 3 - if marker_length: - end = pattern.find(">", index + marker_length) - if end >= 0: - names.append(pattern[index + marker_length : end]) - index = end + 1 + if not in_class and verbose_stack[-1] and character == "#": + newline = pattern.find("\n", index + 1) + index = len(pattern) if newline < 0 else newline + 1 + continue + if not in_class and pattern.startswith("(?#", index): + index = _comment_end(pattern, index + 3) + continue + if not in_class and character == "(": + flags = re.match(r"\(\?([A-Za-z]*)(?:-([A-Za-z]*))?([:)])", pattern[index:]) + if flags is not None: + enabled, disabled, delimiter = flags.groups() + verbose = (verbose_stack[-1] or "x" in enabled) and "x" not in (disabled or "") + index += flags.end() + if delimiter == ":": + verbose_stack.append(verbose) + else: + verbose_stack[-1] = verbose continue + marker_length = 0 + if pattern.startswith("(?P<", index): + marker_length = 4 + elif pattern.startswith("(?<", index): + next_character = pattern[index + 3 : index + 4] + if next_character not in ("=", "!"): + marker_length = 3 + if marker_length: + end = pattern.find(">", index + marker_length) + if end >= 0: + names.append(pattern[index + marker_length : end]) + verbose_stack.append(verbose_stack[-1]) + index = end + 1 + continue + verbose_stack.append(verbose_stack[-1]) + index += 1 + continue + if not in_class and character == ")": + if len(verbose_stack) > 1: + verbose_stack.pop() + index += 1 + continue index += 1 return names + + +def _comment_end(pattern: str, index: int) -> int: + escaped = False + while index < len(pattern): + character = pattern[index] + if escaped: + escaped = False + elif character == "\\": + escaped = True + elif character == ")": + return index + 1 + index += 1 + return index diff --git a/smallserver/server.py b/smallserver/server.py index a3510d6..9ef851d 100644 --- a/smallserver/server.py +++ b/smallserver/server.py @@ -123,10 +123,10 @@ class ServerConfig: max_header_bytes: int = 16 * 1024 max_header_count: int = 100 max_body_bytes: int = 1024 * 1024 - max_request_target_bytes: int = 8 * 1024 receive_chunk_bytes: int = 8 * 1024 listener_priority: int = 1 connection_priority: int = 2 + max_request_target_bytes: int = 8 * 1024 def __post_init__(self) -> None: for name, value in self.__dict__.items(): diff --git a/tests/installed_regex_smoke.py b/tests/installed_regex_smoke.py new file mode 100644 index 0000000..7689ff0 --- /dev/null +++ b/tests/installed_regex_smoke.py @@ -0,0 +1,21 @@ +"""Smoke an installed ``smallserver[regex-routes]`` package outside the source path.""" + +import asyncio + +from smallserver import Headers, Request, Response, SmallServer + + +async def main() -> None: + app = SmallServer() + + @app.get_regex(r"/users/(?P[0-9]+)") + async def user(request: Request) -> Response: + return Response.text(request.path_params["user_id"]) + + response = await app.dispatch(Request("GET", "/users/42?source=smoke", Headers())) + assert response.status == 200 + assert response.body == b"42" + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_regex_routing.py b/tests/test_regex_routing.py index fba24b4..71c0b73 100644 --- a/tests/test_regex_routing.py +++ b/tests/test_regex_routing.py @@ -12,12 +12,17 @@ RouteMatchTimeout, SmallServer, ) +from smallserver.routing import Router HAS_REGEX = importlib.util.find_spec("regex") is not None class OptionalRegexDependencyTests(unittest.TestCase): + def test_error_observer_must_be_callable(self) -> None: + with self.assertRaisesRegex(TypeError, "route_error_observer"): + SmallServer(route_error_observer=object()) # type: ignore[arg-type] + def test_static_routes_do_not_import_optional_engine(self) -> None: app = SmallServer() @@ -88,6 +93,51 @@ async def exact(request): self.assertEqual((await app.dispatch(Request("GET", "/items/7", Headers()))).body, b"static") self.assertEqual((await app.dispatch(Request("GET", "/items/8", Headers()))).body, b"broad") + async def test_alternation_and_optional_literals_preserve_order_and_405(self) -> None: + app = SmallServer() + + @app.get_regex(r"/foo|/bar") + async def top_level(request): + return Response.text("top") + + @app.get_regex(r"/fo?bar") + async def optional(request): + return Response.text("optional") + + @app.get_regex(r"/(?:red|blue)") + async def nested(request): + return Response.text("nested") + + for path, expected in ( + ("/foo", b"top"), + ("/bar", b"top"), + ("/fbar", b"optional"), + ("/fobar", b"optional"), + ("/red", b"nested"), + ("/blue", b"nested"), + ): + with self.subTest(path=path): + self.assertEqual((await app.dispatch(Request("GET", path, Headers()))).body, expected) + response = await app.dispatch(Request("POST", "/bar", Headers())) + self.assertEqual(response.status, 405) + self.assertEqual(response.headers["allow"], "GET") + + def test_compiled_routes_are_structurally_guarded_to_slash_paths(self) -> None: + async def handler(request): + return Response() + + for pattern, slash_path, non_slash_path in ( + (r"/foo|bar", "/foo", "bar"), + (r"/?foo", "/foo", "foo"), + ): + with self.subTest(pattern=pattern): + router = Router() + router.add_regex(pattern, ("GET",), handler) + self.assertIs(router.resolve("GET", slash_path).handler, handler) + miss = router.resolve("GET", non_slash_path) + self.assertIsNone(miss.handler) + self.assertEqual(miss.allowed_methods, ()) + async def test_method_first_matching_and_sorted_allow(self) -> None: app = SmallServer() @@ -162,6 +212,19 @@ async def handler(request): with self.assertRaisesRegex(ValueError, "maximum"): limited.get_regex(r"/two")(handler) + async def test_verbose_comment_group_text_does_not_create_false_duplicate(self) -> None: + app = SmallServer() + + @app.get_regex( + "/(?x:items/ # (?Pthis is comment text)\n" + " (?P[0-9]+))" + ) + async def item(request): + return Response.text(request.path_params["item_id"]) + + response = await app.dispatch(Request("GET", "/items/42", Headers())) + self.assertEqual(response.body, b"42") + async def test_catastrophic_backtracking_is_bounded_and_path_is_not_disclosed(self) -> None: app = SmallServer(RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005)) @@ -180,9 +243,9 @@ async def handler(request): async def test_total_budget_bounds_many_individually_fast_misses(self) -> None: class SlowPattern: + groupindex = {} + def fullmatch(self, value, timeout): - if not value: - return None time.sleep(min(timeout / 2, 0.001)) return None @@ -207,14 +270,30 @@ async def handler(request): self.assertLess(time.monotonic() - started, 0.05) async def test_manual_dispatch_path_limit_is_enforced_before_matching(self) -> None: + calls = [] + + class Pattern: + groupindex = {} + + def fullmatch(self, value, timeout): + calls.append(value) + return None + + class Engine: + @staticmethod + def compile(pattern): + return Pattern() + app = SmallServer(RegexRouteConfig(max_path_bytes=8)) - @app.get_regex(r"/.*") async def handler(request): return Response() - with self.assertRaisesRegex(ValueError, "too large"): - await app.dispatch(Request("GET", "/12345678", Headers())) + with patch("smallserver.routing.importlib.import_module", return_value=Engine()): + app.get_regex(r"/.*")(handler) + response = await app.dispatch(Request("GET", "/12345678", Headers())) + self.assertEqual(response.status, 414) + self.assertEqual(calls, []) class RegexRouteConfigTests(unittest.TestCase): diff --git a/tests/test_server.py b/tests/test_server.py index 27b1766..86eef5c 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -59,6 +59,17 @@ def test_config_rejects_unbounded_limits(self) -> None: with self.assertRaisesRegex(ValueError, "max_request_target_bytes"): ServerConfig(max_request_target_bytes=0) + def test_config_preserves_legacy_positional_field_mapping(self) -> None: + config = ServerConfig(1, 2, 3, 4, 5, 6, 7) + self.assertEqual(config.max_connections, 1) + self.assertEqual(config.max_header_bytes, 2) + self.assertEqual(config.max_header_count, 3) + self.assertEqual(config.max_body_bytes, 4) + self.assertEqual(config.receive_chunk_bytes, 5) + self.assertEqual(config.listener_priority, 6) + self.assertEqual(config.connection_priority, 7) + self.assertEqual(config.max_request_target_bytes, 8 * 1024) + def test_serve_closes_bound_socket_when_runtime_fork_fails(self) -> None: class Listener: closed = False diff --git a/tests/test_server_runtime.py b/tests/test_server_runtime.py index 9d80992..1ada18c 100644 --- a/tests/test_server_runtime.py +++ b/tests/test_server_runtime.py @@ -6,7 +6,7 @@ from SmallPackage import SmallOS, Unix from SmallPackage.adapters.threads import ThreadAdapter -from smallserver import AdapterRegistry, Response, SmallServer +from smallserver import AdapterRegistry, RegexRouteConfig, Response, RouteMatchTimeout, SmallServer HAS_REGEX = importlib.util.find_spec("regex") is not None @@ -164,3 +164,86 @@ def client() -> None: self.assertFalse(worker.is_alive()) self.assertEqual(errors, []) self.assertIn(b"\r\n\r\na%2Fb?download=1", b"".join(received)) + + @unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") + def test_regex_timeout_is_observed_once_and_does_not_stop_server(self) -> None: + runtime = SmallOS().setKernel(Unix()) + observed: list[RouteMatchTimeout] = [] + app = SmallServer( + RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005), + route_error_observer=observed.append, + ) + + @app.get_regex(r"/(a+)+$") + async def expensive(request): + return Response() + + @app.get("/health") + async def health(request): + return Response.text("healthy") + + try: + server = app.serve(runtime, host="127.0.0.1", port=0) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + hostile_path = "/" + "a" * 5000 + "!" + received = [] + errors = [] + + def client() -> None: + try: + received.append(self._request(server.port, hostile_path)) + received.append(self._request(server.port, "/health")) + except BaseException as exc: + errors.append(exc) + finally: + server.close() + + worker = threading.Thread(target=client, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=3) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(len(observed), 1) + self.assertEqual(observed[0].route_id, "regex-route-1") + self.assertNotIn(hostile_path, str(observed[0])) + self.assertTrue(received[0].startswith(b"HTTP/1.1 500 Internal Server Error\r\n")) + self.assertNotIn(hostile_path.encode("ascii"), received[0]) + self.assertTrue(received[1].startswith(b"HTTP/1.1 200 OK\r\n")) + self.assertTrue(received[1].endswith(b"healthy")) + + @unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") + def test_regex_path_limit_returns_414_before_matching(self) -> None: + runtime = SmallOS().setKernel(Unix()) + app = SmallServer(RegexRouteConfig(max_path_bytes=8)) + + @app.get_regex(r"/.*") + async def route(request): + return Response.text("must not run") + + try: + server = app.serve(runtime, host="127.0.0.1", port=0) + except PermissionError: + self.skipTest("the current sandbox does not permit loopback TCP binds") + + received = [] + errors = [] + + def client() -> None: + try: + received.append(self._request(server.port, "/12345678")) + except BaseException as exc: + errors.append(exc) + finally: + server.close() + + worker = threading.Thread(target=client, daemon=True) + worker.start() + runtime.start() + worker.join(timeout=2) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertTrue(received[0].startswith(b"HTTP/1.1 414 URI Too Long\r\n")) + self.assertNotIn(b"must not run", received[0]) diff --git a/tests/typing/regex_routes.py b/tests/typing/regex_routes.py new file mode 100644 index 0000000..2239829 --- /dev/null +++ b/tests/typing/regex_routes.py @@ -0,0 +1,20 @@ +"""Public typing fixture for mypy/pyright and compile-only release checks.""" + +from smallserver import Request, Response, RouteMatchTimeout, SmallServer + + +def observe(error: RouteMatchTimeout) -> None: + route_id: str = error.route_id + assert route_id + + +def application() -> SmallServer: + app = SmallServer(route_error_observer=observe) + + @app.get_regex(r"/users/(?P[0-9]+)") + async def user(request: Request) -> Response: + user_id: str = request.path_params["user_id"] + pattern: str | None = request.route_pattern + return Response.json({"user_id": user_id, "pattern": pattern}) + + return app From 3aa94ea523e5d315d954d7c390649acc6f27fec6 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Fri, 21 Aug 2026 23:47:09 -0500 Subject: [PATCH 3/5] fix: isolate route events and optimize static dispatch --- README.md | 21 +++++---- benchmarks/route_benchmark.py | 67 ++++++++++++++++++++-------- smallserver/__init__.py | 9 +++- smallserver/app.py | 35 +++++++++------ smallserver/routing.py | 12 ++++++ tests/installed_regex_smoke.py | 4 +- tests/test_routing.py | 16 +++++++ tests/test_server_runtime.py | 79 +++++++++++++++++++++++++++++----- tests/typing/regex_routes.py | 6 +-- 9 files changed, 195 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index 991183c..93ffe35 100644 --- a/README.md +++ b/README.md @@ -148,30 +148,35 @@ opaque route ID and becomes a sanitized 500 response on the network path. Applications can observe that failure without receiving the hostile target: ```python -from smallserver import RouteMatchTimeout +from smallserver import RouteErrorEvent -def observe_route_error(error: RouteMatchTimeout) -> None: - logger.error("route matching failed: %s", error.route_id) +def observe_route_error(event: RouteErrorEvent) -> None: + logger.error("route matching failed: %s (%s)", event.route_id, event.category) app = SmallServer(route_error_observer=observe_route_error) ``` -The synchronous observer runs once on the connection task and should return -quickly; observer failures are isolated from the response path. A path above -the configured regex-routing byte limit returns 414 before matching begins. +The synchronous observer receives a fresh, immutable, traceback-free event +containing only an opaque route ID and category. It runs once on the connection +task and should return quickly; observer failures are isolated from the +response path. A path above the configured regex-routing byte limit returns +414 before matching begins. Requests retain the exact ASCII origin-form target in `request.raw_target`. Routing uses `request.path`, which excludes the query string; `request.query_string` contains the raw text after `?`. Neither paths nor named captures are percent-decoded, so `/files/a%2Fb` remains distinct from -`/files/a/b`. `request.route_pattern` identifies the selected static path or -regex pattern. +`/files/a/b`. `request.route_pattern` identifies a selected regex pattern; +static dispatch passes the original request through without adding route +context. Run `python benchmarks/route_benchmark.py` for a same-process comparison of the pre-router dictionary dispatch model and current router dispatch. Its JSON also records configured versus observed hostile-pattern timeout when the extra is installed; rates are machine-specific and should be compared on the same host. +Use `python benchmarks/route_benchmark.py --release` to enforce the documented +static-dispatch floor of 80% of the legacy model across repeated rounds. Release validation can run `python tests/installed_regex_smoke.py` from an environment where the built `smallserver[regex-routes]` wheel is installed. diff --git a/benchmarks/route_benchmark.py b/benchmarks/route_benchmark.py index a3b247e..81ff591 100644 --- a/benchmarks/route_benchmark.py +++ b/benchmarks/route_benchmark.py @@ -8,6 +8,7 @@ import inspect import json from pathlib import Path +import statistics import sys import time @@ -15,6 +16,8 @@ from smallserver import Headers, RegexRouteConfig, Request, Response, RouteMatchTimeout, SmallServer +STATIC_DISPATCH_RATIO_FLOOR = 0.80 + async def _legacy_dispatch(routes, request): """Model the pre-router static dictionary dispatch for same-run comparison.""" @@ -28,7 +31,14 @@ async def _legacy_dispatch(routes, request): return response -async def benchmark(iterations: int) -> dict[str, float | int | str]: +async def _measure(operation, iterations: int) -> float: + started = time.perf_counter() + for _ in range(iterations): + await operation() + return iterations / (time.perf_counter() - started) + + +async def benchmark(iterations: int, rounds: int) -> dict[str, float | int | str | bool]: app = SmallServer() @app.get("/health") @@ -38,23 +48,37 @@ async def health(request): request = Request("GET", "/health", Headers()) legacy_routes = {("GET", "/health"): health} - started = time.perf_counter() - for _ in range(iterations): - await _legacy_dispatch(legacy_routes, request) - legacy_elapsed = time.perf_counter() - started - - started = time.perf_counter() - for _ in range(iterations): - await app.dispatch(request) - router_elapsed = time.perf_counter() - started - - legacy_rate = iterations / legacy_elapsed - router_rate = iterations / router_elapsed - result: dict[str, float | int | str] = { + async def legacy_operation(): + return await _legacy_dispatch(legacy_routes, request) + + async def router_operation(): + return await app.dispatch(request) + + await _measure(legacy_operation, min(iterations, 1_000)) + await _measure(router_operation, min(iterations, 1_000)) + legacy_rates = [] + router_rates = [] + ratios = [] + for round_number in range(rounds): + if round_number % 2: + router_rate = await _measure(router_operation, iterations) + legacy_rate = await _measure(legacy_operation, iterations) + else: + legacy_rate = await _measure(legacy_operation, iterations) + router_rate = await _measure(router_operation, iterations) + legacy_rates.append(legacy_rate) + router_rates.append(router_rate) + ratios.append(router_rate / legacy_rate) + + median_ratio = statistics.median(ratios) + result: dict[str, float | int | str | bool] = { "iterations": iterations, - "legacy_static_dispatches_per_second": round(legacy_rate, 2), - "router_static_dispatches_per_second": round(router_rate, 2), - "router_to_legacy_ratio": round(router_rate / legacy_rate, 4), + "rounds": rounds, + "legacy_static_dispatches_per_second": round(statistics.median(legacy_rates), 2), + "router_static_dispatches_per_second": round(statistics.median(router_rates), 2), + "router_to_legacy_ratio": round(median_ratio, 4), + "static_dispatch_ratio_floor": STATIC_DISPATCH_RATIO_FLOOR, + "static_dispatch_floor_passed": median_ratio >= STATIC_DISPATCH_RATIO_FLOOR, } if importlib.util.find_spec("regex") is None: @@ -80,10 +104,17 @@ async def hostile(request): def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--iterations", type=int, default=25_000) + parser.add_argument("--rounds", type=int, default=5) + parser.add_argument("--release", action="store_true") arguments = parser.parse_args() if arguments.iterations <= 0: parser.error("--iterations must be positive") - print(json.dumps(asyncio.run(benchmark(arguments.iterations)), indent=2, sort_keys=True)) + if arguments.rounds <= 0: + parser.error("--rounds must be positive") + result = asyncio.run(benchmark(arguments.iterations, arguments.rounds)) + print(json.dumps(result, indent=2, sort_keys=True)) + if arguments.release and not result["static_dispatch_floor_passed"]: + raise SystemExit(1) if __name__ == "__main__": diff --git a/smallserver/__init__.py b/smallserver/__init__.py index 476d64e..2fe0c87 100644 --- a/smallserver/__init__.py +++ b/smallserver/__init__.py @@ -4,7 +4,13 @@ from .adapters import AdapterRegistry, AdapterShutdownError, http_error_from_adapter from .errors import HTTPError from .http import Headers, Request, Response -from .routing import RegexRouteConfig, RegexRoutesUnavailable, RouteMatchTimeout, RoutePathTooLarge +from .routing import ( + RegexRouteConfig, + RegexRoutesUnavailable, + RouteErrorEvent, + RouteMatchTimeout, + RoutePathTooLarge, +) from .server import ServerConfig, ServerHandle __all__ = [ @@ -16,6 +22,7 @@ "RegexRoutesUnavailable", "Request", "Response", + "RouteErrorEvent", "RouteMatchTimeout", "RoutePathTooLarge", "ServerConfig", diff --git a/smallserver/app.py b/smallserver/app.py index a009025..eee2386 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -10,11 +10,17 @@ from .errors import HTTPError from .http import Request, Response -from .routing import RegexRouteConfig, RouteMatchTimeout, RoutePathTooLarge, Router +from .routing import ( + RegexRouteConfig, + RouteErrorEvent, + RouteMatchTimeout, + RoutePathTooLarge, + Router, +) from .server import HTTPParseError, HTTPRequestParser, ServerConfig, ServerHandle Handler = Callable[[Request], Awaitable[Response]] -RouteErrorObserver = Callable[[RouteMatchTimeout], None] +RouteErrorObserver = Callable[[RouteErrorEvent], None] class SmallServer: @@ -141,16 +147,18 @@ def serve( async def dispatch(self, request: Request) -> Response: """Run a registered handler or return a deterministic HTTP response.""" - try: - match = self._router.resolve(request.method, request.path) - except RoutePathTooLarge: - return Response.text("request target is too large", status=414) - if match.handler is None: - if match.allowed_methods: - return Response.text("method not allowed", status=405, headers={"Allow": ", ".join(match.allowed_methods)}) - return Response.text("not found", status=404) - handler = match.handler - request = replace(request, path_params=match.path_params, route_pattern=match.route_pattern) + handler = self._router.static_handler(request.method, request.path) + if handler is None: + try: + match = self._router.resolve(request.method, request.path) + except RoutePathTooLarge: + return Response.text("request target is too large", status=414) + if match.handler is None: + if match.allowed_methods: + return Response.text("method not allowed", status=405, headers={"Allow": ", ".join(match.allowed_methods)}) + return Response.text("not found", status=404) + handler = match.handler + request = replace(request, path_params=match.path_params, route_pattern=match.route_pattern) try: result = handler(request) if not inspect.isawaitable(result): @@ -259,7 +267,8 @@ def _observe_route_error(self, error: RouteMatchTimeout) -> None: observer = self._route_error_observer if observer is None: return + event = RouteErrorEvent(route_id=error.route_id, category="route_match_timeout") try: - observer(error) + observer(event) except Exception: pass diff --git a/smallserver/routing.py b/smallserver/routing.py index b53588d..f8db1ed 100644 --- a/smallserver/routing.py +++ b/smallserver/routing.py @@ -33,6 +33,14 @@ class RoutePathTooLarge(RuntimeError): """Raised before matching when a request path exceeds its routing bound.""" +@dataclass(frozen=True) +class RouteErrorEvent: + """Traceback-free, immutable routing failure data safe for observation.""" + + route_id: str + category: str + + @dataclass(frozen=True) class RegexRouteConfig: """Finite limits applied to regex registration and hostile request paths.""" @@ -98,6 +106,10 @@ def add_static(self, path: str, methods: tuple[str, ...], handler: Handler) -> N for key in keys: self._static[key] = handler + def static_handler(self, method: str, path: str) -> Handler | None: + """Return an exact static handler without allocating match context.""" + return self._static.get((method.upper(), path)) + def add_regex(self, pattern: str, methods: tuple[str, ...], handler: Handler) -> None: if not isinstance(pattern, str): raise TypeError("regex route pattern must be a string") diff --git a/tests/installed_regex_smoke.py b/tests/installed_regex_smoke.py index 7689ff0..e439c6d 100644 --- a/tests/installed_regex_smoke.py +++ b/tests/installed_regex_smoke.py @@ -2,10 +2,12 @@ import asyncio -from smallserver import Headers, Request, Response, SmallServer +from smallserver import Headers, Request, Response, RouteErrorEvent, SmallServer async def main() -> None: + event = RouteErrorEvent("regex-route-smoke", "route_match_timeout") + assert (event.route_id, event.category) == ("regex-route-smoke", "route_match_timeout") app = SmallServer() @app.get_regex(r"/users/(?P[0-9]+)") diff --git a/tests/test_routing.py b/tests/test_routing.py index d152500..72349f6 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -44,6 +44,22 @@ async def items(request): self.assertEqual(seen[0].path, "/items") self.assertEqual(seen[0].query_string, "tag=a%2Fb") + async def test_static_dispatch_passes_original_request_without_route_context_copy(self) -> None: + app = SmallServer() + seen = [] + + @app.get("/health") + async def health(request): + seen.append(request) + return Response() + + request = Request("GET", "/health", Headers()) + response = await app.dispatch(request) + self.assertEqual(response.status, 200) + self.assertIs(seen[0], request) + self.assertIsNone(seen[0].route_pattern) + self.assertEqual(dict(seen[0].path_params), {}) + async def test_http_error_becomes_response(self) -> None: app = SmallServer() diff --git a/tests/test_server_runtime.py b/tests/test_server_runtime.py index 1ada18c..c7bbb61 100644 --- a/tests/test_server_runtime.py +++ b/tests/test_server_runtime.py @@ -1,4 +1,5 @@ import importlib.util +from dataclasses import FrozenInstanceError import socket import threading import unittest @@ -6,18 +7,23 @@ from SmallPackage import SmallOS, Unix from SmallPackage.adapters.threads import ThreadAdapter -from smallserver import AdapterRegistry, RegexRouteConfig, Response, RouteMatchTimeout, SmallServer +from smallserver import ( + AdapterRegistry, + RegexRouteConfig, + Request, + Response, + RouteErrorEvent, + SmallServer, +) HAS_REGEX = importlib.util.find_spec("regex") is not None class SmallOSServerIntegrationTests(unittest.TestCase): - def _request(self, port: int, path: str) -> bytes: + def _exchange(self, port: int, payload: bytes) -> bytes: with socket.create_connection(("127.0.0.1", port), timeout=3) as connection: - connection.sendall( - "GET {} HTTP/1.1\r\nHost: localhost\r\n\r\n".format(path).encode("ascii") - ) + connection.sendall(payload) chunks = [] while True: chunk = connection.recv(4096) @@ -25,6 +31,12 @@ def _request(self, port: int, path: str) -> bytes: return b"".join(chunks) chunks.append(chunk) + def _request(self, port: int, path: str) -> bytes: + return self._exchange( + port, + "GET {} HTTP/1.1\r\nHost: localhost\r\n\r\n".format(path).encode("ascii"), + ) + def test_loopback_server_accepts_fragmented_request_and_shuts_down(self) -> None: runtime = SmallOS().setKernel(Unix()) app = SmallServer() @@ -168,13 +180,15 @@ def client() -> None: @unittest.skipUnless(HAS_REGEX, "regex-routes extra is not installed") def test_regex_timeout_is_observed_once_and_does_not_stop_server(self) -> None: runtime = SmallOS().setKernel(Unix()) - observed: list[RouteMatchTimeout] = [] + observed: list[RouteErrorEvent] = [] app = SmallServer( RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005), route_error_observer=observed.append, ) - @app.get_regex(r"/(a+)+$") + pattern_secret = "sensitive-pattern-marker" + + @app.post_regex(r"/(a+)+$(?#sensitive-pattern-marker)") async def expensive(request): return Response() @@ -188,12 +202,20 @@ async def health(request): self.skipTest("the current sandbox does not permit loopback TCP binds") hostile_path = "/" + "a" * 5000 + "!" + authorization_secret = "Bearer sensitive-authorization-marker" + body_secret = b"sensitive-body-marker" received = [] errors = [] def client() -> None: try: - received.append(self._request(server.port, hostile_path)) + request = ( + "POST {} HTTP/1.1\r\n" + "Host: localhost\r\n" + "Authorization: {}\r\n" + "Content-Length: {}\r\n\r\n" + ).format(hostile_path, authorization_secret, len(body_secret)).encode("ascii") + received.append(self._exchange(server.port, request + body_secret)) received.append(self._request(server.port, "/health")) except BaseException as exc: errors.append(exc) @@ -207,8 +229,24 @@ def client() -> None: self.assertFalse(worker.is_alive()) self.assertEqual(errors, []) self.assertEqual(len(observed), 1) - self.assertEqual(observed[0].route_id, "regex-route-1") - self.assertNotIn(hostile_path, str(observed[0])) + event = observed[0] + self.assertEqual(event.route_id, "regex-route-1") + self.assertEqual(event.category, "route_match_timeout") + with self.assertRaises(FrozenInstanceError): + event.route_id = "changed" # type: ignore[misc] + self.assertFalse(hasattr(event, "__traceback__")) + self.assertFalse(hasattr(event, "__cause__")) + self.assertFalse(hasattr(event, "__context__")) + + reachable = _reachable_objects(event) + reachable_strings = {value for value in reachable if isinstance(value, str)} + self.assertEqual( + reachable_strings, + {"route_id", "category", "regex-route-1", "route_match_timeout"}, + ) + self.assertFalse(any(isinstance(value, Request) for value in reachable)) + for secret in (hostile_path, authorization_secret, body_secret.decode("ascii"), pattern_secret): + self.assertNotIn(secret, reachable_strings) self.assertTrue(received[0].startswith(b"HTTP/1.1 500 Internal Server Error\r\n")) self.assertNotIn(hostile_path.encode("ascii"), received[0]) self.assertTrue(received[1].startswith(b"HTTP/1.1 200 OK\r\n")) @@ -247,3 +285,24 @@ def client() -> None: self.assertEqual(errors, []) self.assertTrue(received[0].startswith(b"HTTP/1.1 414 URI Too Long\r\n")) self.assertNotIn(b"must not run", received[0]) + + +def _reachable_objects(root): + pending = [root] + seen = set() + result = [] + while pending: + value = pending.pop() + identity = id(value) + if identity in seen: + continue + seen.add(identity) + result.append(value) + if isinstance(value, dict): + pending.extend(value.keys()) + pending.extend(value.values()) + elif isinstance(value, (tuple, list, set, frozenset)): + pending.extend(value) + elif hasattr(value, "__dict__"): + pending.append(vars(value)) + return result diff --git a/tests/typing/regex_routes.py b/tests/typing/regex_routes.py index 2239829..7f66494 100644 --- a/tests/typing/regex_routes.py +++ b/tests/typing/regex_routes.py @@ -1,10 +1,10 @@ """Public typing fixture for mypy/pyright and compile-only release checks.""" -from smallserver import Request, Response, RouteMatchTimeout, SmallServer +from smallserver import Request, Response, RouteErrorEvent, SmallServer -def observe(error: RouteMatchTimeout) -> None: - route_id: str = error.route_id +def observe(event: RouteErrorEvent) -> None: + route_id: str = event.route_id assert route_id From 4fed85273a2fcfadfee08d86a097763a480beac8 Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Fri, 21 Aug 2026 23:57:49 -0500 Subject: [PATCH 4/5] fix: isolate regex observer execution --- README.md | 13 +++-- benchmarks/route_benchmark.py | 4 ++ smallserver/app.py | 106 +++++++++++++++++++++++++++++----- smallserver/server.py | 19 +++++- tests/test_route_benchmark.py | 35 +++++++++++ tests/test_routing.py | 22 +++++++ tests/test_server_runtime.py | 47 +++++++++++++-- 7 files changed, 222 insertions(+), 24 deletions(-) create mode 100644 tests/test_route_benchmark.py diff --git a/README.md b/README.md index 93ffe35..87b894b 100644 --- a/README.md +++ b/README.md @@ -156,11 +156,14 @@ def observe_route_error(event: RouteErrorEvent) -> None: app = SmallServer(route_error_observer=observe_route_error) ``` -The synchronous observer receives a fresh, immutable, traceback-free event -containing only an opaque route ID and category. It runs once on the connection -task and should return quickly; observer failures are isolated from the -response path. A path above the configured regex-routing byte limit returns -414 before matching begins. +The observer receives a fresh, immutable, traceback-free event containing only +an opaque route ID and category. A bounded single-worker dispatcher invokes it +outside the SmallOS/request call stack and exits when its queue drains. The +observer should return quickly; failures are isolated from responses, reported +through `threading.excepthook`, and counted by +`server.route_observer_failures`. Capacity drops are counted by +`server.dropped_route_error_events`. A path above the configured regex-routing +byte limit returns 414 before matching begins. Requests retain the exact ASCII origin-form target in `request.raw_target`. Routing uses `request.path`, which excludes the query string; diff --git a/benchmarks/route_benchmark.py b/benchmarks/route_benchmark.py index 81ff591..1efab47 100644 --- a/benchmarks/route_benchmark.py +++ b/benchmarks/route_benchmark.py @@ -111,6 +111,10 @@ def main() -> None: parser.error("--iterations must be positive") if arguments.rounds <= 0: parser.error("--rounds must be positive") + if arguments.release and arguments.iterations < 10_000: + parser.error("--release requires at least 10000 iterations") + if arguments.release and arguments.rounds < 5: + parser.error("--release requires at least 5 rounds") result = asyncio.run(benchmark(arguments.iterations, arguments.rounds)) print(json.dumps(result, indent=2, sort_keys=True)) if arguments.release and not result["static_dispatch_floor_passed"]: diff --git a/smallserver/app.py b/smallserver/app.py index eee2386..53c5d51 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -3,9 +3,11 @@ from __future__ import annotations import inspect +from collections import deque from collections.abc import Awaitable, Callable, Iterable from dataclasses import replace import socket +import threading from typing import Any from .errors import HTTPError @@ -23,6 +25,77 @@ RouteErrorObserver = Callable[[RouteErrorEvent], None] +class _RouteObserverDispatcher: + """Run sanitized events on one bounded, short-lived observer thread.""" + + def __init__(self, observer: RouteErrorObserver) -> None: + self._observer = observer + self._events: deque[RouteErrorEvent] = deque() + self._lock = threading.Lock() + self._worker_active = False + self._pending = 0 + self._dropped = 0 + self._failures = 0 + + @property + def dropped(self) -> int: + with self._lock: + return self._dropped + + @property + def failures(self) -> int: + with self._lock: + return self._failures + + def schedule(self, event: RouteErrorEvent, max_pending: int) -> bool: + with self._lock: + if self._pending >= max_pending: + self._dropped += 1 + return False + self._events.append(event) + self._pending += 1 + if self._worker_active: + return True + self._worker_active = True + try: + threading.Thread( + target=self._run, + name="smallserver-route-observer", + daemon=True, + ).start() + except BaseException: + self._worker_active = False + self._events.pop() + self._pending -= 1 + self._dropped += 1 + return False + return True + + def _run(self) -> None: + while True: + with self._lock: + if not self._events: + self._worker_active = False + return + event = self._events.popleft() + try: + self._observer(event) + except BaseException as exc: + with self._lock: + self._failures += 1 + try: + threading.excepthook( + threading.ExceptHookArgs( + (type(exc), exc, exc.__traceback__, threading.current_thread()) + ) + ) + except BaseException: + pass + finally: + with self._lock: + self._pending -= 1 + + class SmallServer: """Register static HTTP routes and dispatch requests to async handlers.""" @@ -35,7 +108,11 @@ def __init__( if route_error_observer is not None and not callable(route_error_observer): raise TypeError("route_error_observer must be callable or None") self._router = Router(regex_config) - self._route_error_observer = route_error_observer + self._route_observer_dispatcher = ( + _RouteObserverDispatcher(route_error_observer) + if route_error_observer is not None + else None + ) def route(self, path: str, methods: Iterable[str]) -> Callable[[Handler], Handler]: if not isinstance(path, str) or not path.startswith("/"): @@ -123,7 +200,7 @@ def serve( except BaseException: listener.close() raise - handle = ServerHandle(runtime, listener, config) + handle = ServerHandle(runtime, listener, config, self._route_observer_dispatcher) listener_task = SmallTask( config.listener_priority, self._accept_loop, @@ -148,7 +225,10 @@ def serve( async def dispatch(self, request: Request) -> Response: """Run a registered handler or return a deterministic HTTP response.""" handler = self._router.static_handler(request.method, request.path) - if handler is None: + if handler is not None: + if request.path_params or request.route_pattern is not None: + request = replace(request, path_params={}, route_pattern=None) + else: try: match = self._router.resolve(request.method, request.path) except RoutePathTooLarge: @@ -235,7 +315,15 @@ async def _connection_loop(self, task: Any, handle: ServerHandle, client: socket try: response = await self.dispatch(request) except RouteMatchTimeout as exc: - self._observe_route_error(exc) + dispatcher = self._route_observer_dispatcher + if dispatcher is not None: + dispatcher.schedule( + RouteErrorEvent( + route_id=exc.route_id, + category="route_match_timeout", + ), + handle._config.max_connections, + ) response = Response.text("internal server error", status=500) except Exception: response = Response.text("internal server error", status=500) @@ -262,13 +350,3 @@ async def _send_response(self, task: Any, client: socket.socket, response: Respo if sent <= 0: return offset += sent - - def _observe_route_error(self, error: RouteMatchTimeout) -> None: - observer = self._route_error_observer - if observer is None: - return - event = RouteErrorEvent(route_id=error.route_id, category="route_match_timeout") - try: - observer(event) - except Exception: - pass diff --git a/smallserver/server.py b/smallserver/server.py index 9ef851d..39f1bfe 100644 --- a/smallserver/server.py +++ b/smallserver/server.py @@ -137,10 +137,17 @@ def __post_init__(self) -> None: class ServerHandle: """A bound listener and its cooperative shutdown signal.""" - def __init__(self, runtime: Any, listener: socket.socket, config: ServerConfig) -> None: + def __init__( + self, + runtime: Any, + listener: socket.socket, + config: ServerConfig, + route_observer_dispatcher: Any = None, + ) -> None: self._runtime = runtime self._listener = listener self._config = config + self._route_observer_dispatcher = route_observer_dispatcher self._wake_read, self._wake_write = socket.socketpair() self._wake_read.setblocking(False) self._wake_write.setblocking(False) @@ -161,6 +168,16 @@ def port(self) -> int: def closed(self) -> bool: return self._closed + @property + def dropped_route_error_events(self) -> int: + dispatcher = self._route_observer_dispatcher + return 0 if dispatcher is None else int(dispatcher.dropped) + + @property + def route_observer_failures(self) -> int: + dispatcher = self._route_observer_dispatcher + return 0 if dispatcher is None else int(dispatcher.failures) + def close(self) -> None: """Request shutdown safely from any thread without closing live FDs there.""" if self._closed: diff --git a/tests/test_route_benchmark.py b/tests/test_route_benchmark.py new file mode 100644 index 0000000..8b0abdd --- /dev/null +++ b/tests/test_route_benchmark.py @@ -0,0 +1,35 @@ +import contextlib +import io +import sys +import unittest +from unittest.mock import patch + +from benchmarks import route_benchmark + + +class RouteBenchmarkCLITests(unittest.TestCase): + def test_release_mode_rejects_undersized_samples(self) -> None: + cases = ( + ("9999", "5", "10000 iterations"), + ("10000", "4", "5 rounds"), + ) + for iterations, rounds, message in cases: + with self.subTest(iterations=iterations, rounds=rounds): + stderr = io.StringIO() + with patch.object( + sys, + "argv", + [ + "route_benchmark.py", + "--release", + "--iterations", + iterations, + "--rounds", + rounds, + ], + ): + with contextlib.redirect_stderr(stderr): + with self.assertRaises(SystemExit) as raised: + route_benchmark.main() + self.assertEqual(raised.exception.code, 2) + self.assertIn(message, stderr.getvalue()) diff --git a/tests/test_routing.py b/tests/test_routing.py index 72349f6..e6aaed4 100644 --- a/tests/test_routing.py +++ b/tests/test_routing.py @@ -60,6 +60,28 @@ async def health(request): self.assertIsNone(seen[0].route_pattern) self.assertEqual(dict(seen[0].path_params), {}) + async def test_static_dispatch_clears_spoofed_route_context(self) -> None: + app = SmallServer() + seen = [] + + @app.get("/health") + async def health(request): + seen.append(request) + return Response() + + dirty = Request( + "GET", + "/health", + Headers(), + path_params={"spoofed": "value"}, + route_pattern="sensitive-spoofed-pattern", + ) + response = await app.dispatch(dirty) + self.assertEqual(response.status, 200) + self.assertIsNot(seen[0], dirty) + self.assertIsNone(seen[0].route_pattern) + self.assertEqual(dict(seen[0].path_params), {}) + async def test_http_error_becomes_response(self) -> None: app = SmallServer() diff --git a/tests/test_server_runtime.py b/tests/test_server_runtime.py index c7bbb61..916e9c4 100644 --- a/tests/test_server_runtime.py +++ b/tests/test_server_runtime.py @@ -1,8 +1,10 @@ import importlib.util from dataclasses import FrozenInstanceError +import inspect import socket import threading import unittest +from unittest.mock import patch from SmallPackage import SmallOS, Unix from SmallPackage.adapters.threads import ThreadAdapter @@ -13,6 +15,7 @@ Request, Response, RouteErrorEvent, + RouteMatchTimeout, SmallServer, ) @@ -181,9 +184,29 @@ def client() -> None: def test_regex_timeout_is_observed_once_and_does_not_stop_server(self) -> None: runtime = SmallOS().setKernel(Unix()) observed: list[RouteErrorEvent] = [] + observer_graph = [] + observer_finished = threading.Event() + hook_finished = threading.Event() + hook_events = [] + + def observe(event: RouteErrorEvent) -> None: + observed.append(event) + caller_locals = [] + frame = inspect.currentframe() + while frame is not None: + caller_locals.append(dict(frame.f_locals)) + frame = frame.f_back + observer_graph.extend(_reachable_objects(caller_locals)) + observer_finished.set() + raise RuntimeError("intentional observer failure") + + def observe_thread_failure(arguments) -> None: + hook_events.append(arguments) + hook_finished.set() + app = SmallServer( RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005), - route_error_observer=observed.append, + route_error_observer=observe, ) pattern_secret = "sensitive-pattern-marker" @@ -217,15 +240,20 @@ def client() -> None: ).format(hostile_path, authorization_secret, len(body_secret)).encode("ascii") received.append(self._exchange(server.port, request + body_secret)) received.append(self._request(server.port, "/health")) + if not observer_finished.wait(2): + raise TimeoutError("route observer did not run") + if not hook_finished.wait(2): + raise TimeoutError("observer failure was not reported") except BaseException as exc: errors.append(exc) finally: server.close() worker = threading.Thread(target=client, daemon=True) - worker.start() - runtime.start() - worker.join(timeout=3) + with patch("threading.excepthook", side_effect=observe_thread_failure): + worker.start() + runtime.start() + worker.join(timeout=3) self.assertFalse(worker.is_alive()) self.assertEqual(errors, []) self.assertEqual(len(observed), 1) @@ -247,6 +275,17 @@ def client() -> None: self.assertFalse(any(isinstance(value, Request) for value in reachable)) for secret in (hostile_path, authorization_secret, body_secret.decode("ascii"), pattern_secret): self.assertNotIn(secret, reachable_strings) + + caller_strings = {value for value in observer_graph if isinstance(value, str)} + self.assertFalse(any(isinstance(value, Request) for value in observer_graph)) + self.assertFalse(any(isinstance(value, RouteMatchTimeout) for value in observer_graph)) + for secret in (hostile_path, authorization_secret, body_secret.decode("ascii"), pattern_secret): + self.assertNotIn(secret, caller_strings) + self.assertNotIn(body_secret, observer_graph) + self.assertEqual(server.route_observer_failures, 1) + self.assertEqual(server.dropped_route_error_events, 0) + self.assertEqual(len(hook_events), 1) + self.assertIsInstance(hook_events[0].exc_value, RuntimeError) self.assertTrue(received[0].startswith(b"HTTP/1.1 500 Internal Server Error\r\n")) self.assertNotIn(hostile_path.encode("ascii"), received[0]) self.assertTrue(received[1].startswith(b"HTTP/1.1 200 OK\r\n")) From 9639a4bdd5c8b04d907486d2aeacf3538e3ee76f Mon Sep 17 00:00:00 2001 From: Michael Emperador Date: Sat, 22 Aug 2026 00:12:10 -0500 Subject: [PATCH 5/5] fix: keep route observers on SmallOS --- README.md | 15 +++-- smallserver/app.py | 123 ++++++++++------------------------- smallserver/server.py | 80 +++++++++++++++++++++-- tests/test_server.py | 94 +++++++++++++++++++++++++- tests/test_server_runtime.py | 94 +++++++++++++++----------- 5 files changed, 262 insertions(+), 144 deletions(-) diff --git a/README.md b/README.md index 87b894b..8fe7a07 100644 --- a/README.md +++ b/README.md @@ -157,13 +157,14 @@ app = SmallServer(route_error_observer=observe_route_error) ``` The observer receives a fresh, immutable, traceback-free event containing only -an opaque route ID and category. A bounded single-worker dispatcher invokes it -outside the SmallOS/request call stack and exits when its queue drains. The -observer should return quickly; failures are isolated from responses, reported -through `threading.excepthook`, and counted by -`server.route_observer_failures`. Capacity drops are counted by -`server.dropped_route_error_events`. A path above the configured regex-routing -byte limit returns 414 before matching begins. +an opaque route ID and category. A bounded scheduler-local queue delivers it on +one dedicated SmallOS task, separate from the request task. The synchronous +observer must return quickly and must not block; blocking and async observers +remain an execution-adapter follow-up. Failures are isolated from responses +and counted by `server.route_observer_failures`. Capacity drops are counted by +`server.dropped_route_error_events`; tune the positive queue bound with +`ServerConfig(max_route_error_events=...)`. A path above the configured +regex-routing byte limit returns 414 before matching begins. Requests retain the exact ASCII origin-form target in `request.raw_target`. Routing uses `request.path`, which excludes the query string; diff --git a/smallserver/app.py b/smallserver/app.py index 53c5d51..952791f 100644 --- a/smallserver/app.py +++ b/smallserver/app.py @@ -3,11 +3,9 @@ from __future__ import annotations import inspect -from collections import deque from collections.abc import Awaitable, Callable, Iterable from dataclasses import replace import socket -import threading from typing import Any from .errors import HTTPError @@ -19,83 +17,19 @@ RoutePathTooLarge, Router, ) -from .server import HTTPParseError, HTTPRequestParser, ServerConfig, ServerHandle +from .server import ( + HTTPParseError, + HTTPRequestParser, + RouteObserverChannel, + ServerConfig, + ServerHandle, + run_route_observer, +) Handler = Callable[[Request], Awaitable[Response]] RouteErrorObserver = Callable[[RouteErrorEvent], None] -class _RouteObserverDispatcher: - """Run sanitized events on one bounded, short-lived observer thread.""" - - def __init__(self, observer: RouteErrorObserver) -> None: - self._observer = observer - self._events: deque[RouteErrorEvent] = deque() - self._lock = threading.Lock() - self._worker_active = False - self._pending = 0 - self._dropped = 0 - self._failures = 0 - - @property - def dropped(self) -> int: - with self._lock: - return self._dropped - - @property - def failures(self) -> int: - with self._lock: - return self._failures - - def schedule(self, event: RouteErrorEvent, max_pending: int) -> bool: - with self._lock: - if self._pending >= max_pending: - self._dropped += 1 - return False - self._events.append(event) - self._pending += 1 - if self._worker_active: - return True - self._worker_active = True - try: - threading.Thread( - target=self._run, - name="smallserver-route-observer", - daemon=True, - ).start() - except BaseException: - self._worker_active = False - self._events.pop() - self._pending -= 1 - self._dropped += 1 - return False - return True - - def _run(self) -> None: - while True: - with self._lock: - if not self._events: - self._worker_active = False - return - event = self._events.popleft() - try: - self._observer(event) - except BaseException as exc: - with self._lock: - self._failures += 1 - try: - threading.excepthook( - threading.ExceptHookArgs( - (type(exc), exc, exc.__traceback__, threading.current_thread()) - ) - ) - except BaseException: - pass - finally: - with self._lock: - self._pending -= 1 - - class SmallServer: """Register static HTTP routes and dispatch requests to async handlers.""" @@ -108,11 +42,7 @@ def __init__( if route_error_observer is not None and not callable(route_error_observer): raise TypeError("route_error_observer must be callable or None") self._router = Router(regex_config) - self._route_observer_dispatcher = ( - _RouteObserverDispatcher(route_error_observer) - if route_error_observer is not None - else None - ) + self._route_error_observer = route_error_observer def route(self, path: str, methods: Iterable[str]) -> Callable[[Handler], Handler]: if not isinstance(path, str) or not path.startswith("/"): @@ -200,7 +130,12 @@ def serve( except BaseException: listener.close() raise - handle = ServerHandle(runtime, listener, config, self._route_observer_dispatcher) + observer_channel = ( + RouteObserverChannel(self._route_error_observer, config.max_route_error_events) + if self._route_error_observer is not None + else None + ) + handle = ServerHandle(runtime, listener, config, observer_channel) listener_task = SmallTask( config.listener_priority, self._accept_loop, @@ -214,7 +149,16 @@ def serve( name="smallserver-close-watcher", ) handle._listener_task = listener_task - tasks = (listener_task, close_task) + tasks: tuple[Any, ...] = (listener_task, close_task) + if observer_channel is not None: + observer_task = SmallTask( + config.listener_priority, + run_route_observer, + args=(observer_channel,), + name="smallserver-route-observer", + ) + observer_channel.bind(observer_task) + tasks += (observer_task,) try: runtime.fork(list(tasks)) except BaseException: @@ -294,6 +238,7 @@ async def _connection_loop(self, task: Any, handle: ServerHandle, client: socket handle._config.max_body_bytes, handle._config.max_request_target_bytes, ) + route_error_event: RouteErrorEvent | None = None try: while not handle.closed: try: @@ -315,15 +260,10 @@ async def _connection_loop(self, task: Any, handle: ServerHandle, client: socket try: response = await self.dispatch(request) except RouteMatchTimeout as exc: - dispatcher = self._route_observer_dispatcher - if dispatcher is not None: - dispatcher.schedule( - RouteErrorEvent( - route_id=exc.route_id, - category="route_match_timeout", - ), - handle._config.max_connections, - ) + route_error_event = RouteErrorEvent( + route_id=exc.route_id, + category="route_match_timeout", + ) response = Response.text("internal server error", status=500) except Exception: response = Response.text("internal server error", status=500) @@ -335,6 +275,9 @@ async def _connection_loop(self, task: Any, handle: ServerHandle, client: socket client.close() except OSError: pass + observer_channel = handle._route_observer_channel + if route_error_event is not None and observer_channel is not None: + observer_channel.enqueue(route_error_event, task) async def _send_response(self, task: Any, client: socket.socket, response: Response) -> None: headers = {name: value for name, value in response.headers.items() if name.lower() != "connection"} diff --git a/smallserver/server.py b/smallserver/server.py index 39f1bfe..c9570af 100644 --- a/smallserver/server.py +++ b/smallserver/server.py @@ -2,11 +2,15 @@ from __future__ import annotations +from collections import deque from dataclasses import dataclass import socket from typing import Any from .http import Headers, Request, Response +from .routing import RouteErrorEvent + +_ROUTE_OBSERVER_SIGNAL = 31 class HTTPParseError(Exception): @@ -127,6 +131,7 @@ class ServerConfig: listener_priority: int = 1 connection_priority: int = 2 max_request_target_bytes: int = 8 * 1024 + max_route_error_events: int = 16 def __post_init__(self) -> None: for name, value in self.__dict__.items(): @@ -134,6 +139,58 @@ def __post_init__(self) -> None: raise ValueError("{} must be a positive integer".format(name)) +class RouteObserverChannel: + """Bounded scheduler-local delivery state for one server invocation.""" + + def __init__(self, observer: Any, max_events: int) -> None: + self.observer = observer + self.max_events = max_events + self.events: deque[RouteErrorEvent] = deque() + self.task: Any = None + self.accepting = True + self.dropped = 0 + self.failures = 0 + + def bind(self, task: Any) -> None: + self.task = task + + def enqueue(self, event: RouteErrorEvent, source_task: Any) -> bool: + if not self.accepting or len(self.events) >= self.max_events: + self.dropped += 1 + return False + self.events.append(event) + try: + signalled = ( + self.task is not None + and source_task.sendSignal(self.task.getID(), _ROUTE_OBSERVER_SIGNAL) == 0 + ) + except BaseException: + signalled = False + if not signalled: + self.events.pop() + self.dropped += 1 + return False + return True + + def stop(self) -> None: + self.accepting = False + self.dropped += len(self.events) + self.events.clear() + + +async def run_route_observer(task: Any, channel: RouteObserverChannel) -> None: + """Drain sanitized events on a dedicated SmallOS task.""" + while channel.accepting: + while channel.events: + event = channel.events.popleft() + try: + channel.observer(event) + except BaseException: + channel.failures += 1 + if channel.accepting: + await task.wait_signal(_ROUTE_OBSERVER_SIGNAL) + + class ServerHandle: """A bound listener and its cooperative shutdown signal.""" @@ -142,12 +199,12 @@ def __init__( runtime: Any, listener: socket.socket, config: ServerConfig, - route_observer_dispatcher: Any = None, + route_observer_channel: RouteObserverChannel | None = None, ) -> None: self._runtime = runtime self._listener = listener self._config = config - self._route_observer_dispatcher = route_observer_dispatcher + self._route_observer_channel = route_observer_channel self._wake_read, self._wake_write = socket.socketpair() self._wake_read.setblocking(False) self._wake_write.setblocking(False) @@ -170,13 +227,13 @@ def closed(self) -> bool: @property def dropped_route_error_events(self) -> int: - dispatcher = self._route_observer_dispatcher - return 0 if dispatcher is None else int(dispatcher.dropped) + channel = self._route_observer_channel + return 0 if channel is None else int(channel.dropped) @property def route_observer_failures(self) -> int: - dispatcher = self._route_observer_dispatcher - return 0 if dispatcher is None else int(dispatcher.failures) + channel = self._route_observer_channel + return 0 if channel is None else int(channel.failures) def close(self) -> None: """Request shutdown safely from any thread without closing live FDs there.""" @@ -194,6 +251,12 @@ def _finish_close(self) -> None: self._connections.clear() if self._listener_task is not None: self._runtime.resume_task(self._listener_task) + channel = self._route_observer_channel + if channel is not None: + channel.stop() + if channel.task is not None: + self._runtime.cancel_task(channel.task) + channel.task = None for sock in (self._listener, self._wake_read, self._wake_write): try: sock.close() @@ -203,6 +266,9 @@ def _finish_close(self) -> None: def _abort_startup(self, tasks: tuple[Any, ...]) -> None: """Release bound resources after task registration fails.""" self._closed = True + channel = self._route_observer_channel + if channel is not None: + channel.stop() cancel_task = getattr(self._runtime, "cancel_task", None) if callable(cancel_task): for task in tasks: @@ -210,6 +276,8 @@ def _abort_startup(self, tasks: tuple[Any, ...]) -> None: cancel_task(task) except BaseException: pass + if channel is not None: + channel.task = None for sock in (self._listener, self._wake_read, self._wake_write): try: sock.close() diff --git a/tests/test_server.py b/tests/test_server.py index 86eef5c..d386393 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,8 +1,8 @@ import unittest from unittest.mock import patch -from smallserver import SmallServer -from smallserver.server import HTTPParseError, HTTPRequestParser, ServerConfig +from smallserver import RouteErrorEvent, SmallServer +from smallserver.server import HTTPParseError, HTTPRequestParser, RouteObserverChannel, ServerConfig class HTTPRequestParserTests(unittest.TestCase): @@ -58,6 +58,8 @@ def test_config_rejects_unbounded_limits(self) -> None: ServerConfig(max_connections=True) with self.assertRaisesRegex(ValueError, "max_request_target_bytes"): ServerConfig(max_request_target_bytes=0) + with self.assertRaisesRegex(ValueError, "max_route_error_events"): + ServerConfig(max_route_error_events=0) def test_config_preserves_legacy_positional_field_mapping(self) -> None: config = ServerConfig(1, 2, 3, 4, 5, 6, 7) @@ -69,6 +71,46 @@ def test_config_preserves_legacy_positional_field_mapping(self) -> None: self.assertEqual(config.listener_priority, 6) self.assertEqual(config.connection_priority, 7) self.assertEqual(config.max_request_target_bytes, 8 * 1024) + self.assertEqual(config.max_route_error_events, 16) + + def test_route_observer_channel_has_deterministic_capacity_and_stop(self) -> None: + class ObserverTask: + @staticmethod + def getID() -> int: + return 9 + + class SourceTask: + signals = [] + + def sendSignal(self, pid, signal) -> int: + self.signals.append((pid, signal)) + return 0 + + channel = RouteObserverChannel(lambda event: None, max_events=1) + channel.bind(ObserverTask()) + source = SourceTask() + first = RouteErrorEvent("regex-route-1", "route_match_timeout") + second = RouteErrorEvent("regex-route-2", "route_match_timeout") + self.assertTrue(channel.enqueue(first, source)) + self.assertFalse(channel.enqueue(second, source)) + self.assertEqual(list(channel.events), [first]) + self.assertEqual(channel.dropped, 1) + self.assertEqual(source.signals, [(9, 31)]) + channel.stop() + self.assertFalse(channel.accepting) + self.assertEqual(list(channel.events), []) + self.assertEqual(channel.dropped, 2) + + failing_channel = RouteObserverChannel(lambda event: None, max_events=1) + failing_channel.bind(ObserverTask()) + + class FailingSourceTask: + def sendSignal(self, pid, signal) -> int: + raise RuntimeError("signal failed") + + self.assertFalse(failing_channel.enqueue(first, FailingSourceTask())) + self.assertEqual(list(failing_channel.events), []) + self.assertEqual(failing_channel.dropped, 1) def test_serve_closes_bound_socket_when_runtime_fork_fails(self) -> None: class Listener: @@ -105,3 +147,51 @@ def cancel_task(self, task) -> None: SmallServer().serve(runtime) self.assertTrue(listener.closed) self.assertEqual(runtime.cancelled, 2) + + def test_observer_task_is_included_in_startup_rollback(self) -> None: + class Listener: + closed = False + + def setsockopt(self, *args) -> None: + pass + + def bind(self, address) -> None: + pass + + def listen(self, backlog) -> None: + pass + + def setblocking(self, blocking) -> None: + pass + + def close(self) -> None: + self.closed = True + + class Runtime: + tasks = [] + cancelled = [] + + def fork(self, tasks) -> None: + self.tasks = list(tasks) + raise RuntimeError("no task capacity") + + def cancel_task(self, task) -> None: + self.cancelled.append(task) + + listener = Listener() + runtime = Runtime() + app = SmallServer(route_error_observer=lambda event: None) + with patch("smallserver.app.socket.socket", return_value=listener): + with self.assertRaisesRegex(RuntimeError, "capacity"): + app.serve(runtime) + self.assertTrue(listener.closed) + self.assertEqual(len(runtime.tasks), 3) + self.assertEqual(runtime.cancelled, runtime.tasks) + self.assertEqual( + [task.name for task in runtime.tasks], + [ + "smallserver-listener", + "smallserver-close-watcher", + "smallserver-route-observer", + ], + ) diff --git a/tests/test_server_runtime.py b/tests/test_server_runtime.py index 916e9c4..3937cf9 100644 --- a/tests/test_server_runtime.py +++ b/tests/test_server_runtime.py @@ -4,7 +4,6 @@ import socket import threading import unittest -from unittest.mock import patch from SmallPackage import SmallOS, Unix from SmallPackage.adapters.threads import ThreadAdapter @@ -18,6 +17,7 @@ RouteMatchTimeout, SmallServer, ) +from smallserver.server import run_route_observer HAS_REGEX = importlib.util.find_spec("regex") is not None @@ -186,24 +186,22 @@ def test_regex_timeout_is_observed_once_and_does_not_stop_server(self) -> None: observed: list[RouteErrorEvent] = [] observer_graph = [] observer_finished = threading.Event() - hook_finished = threading.Event() - hook_events = [] + observer_threads = [] def observe(event: RouteErrorEvent) -> None: observed.append(event) + observer_threads.append(threading.current_thread()) caller_locals = [] frame = inspect.currentframe() while frame is not None: caller_locals.append(dict(frame.f_locals)) + if frame.f_code is run_route_observer.__code__: + break frame = frame.f_back - observer_graph.extend(_reachable_objects(caller_locals)) + observer_graph.extend(_reachable_container_values(caller_locals)) observer_finished.set() raise RuntimeError("intentional observer failure") - def observe_thread_failure(arguments) -> None: - hook_events.append(arguments) - hook_finished.set() - app = SmallServer( RegexRouteConfig(match_timeout=0.001, total_match_timeout=0.005), route_error_observer=observe, @@ -227,35 +225,24 @@ async def health(request): hostile_path = "/" + "a" * 5000 + "!" authorization_secret = "Bearer sensitive-authorization-marker" body_secret = b"sensitive-body-marker" - received = [] - errors = [] - - def client() -> None: - try: - request = ( - "POST {} HTTP/1.1\r\n" - "Host: localhost\r\n" - "Authorization: {}\r\n" - "Content-Length: {}\r\n\r\n" - ).format(hostile_path, authorization_secret, len(body_secret)).encode("ascii") - received.append(self._exchange(server.port, request + body_secret)) - received.append(self._request(server.port, "/health")) - if not observer_finished.wait(2): - raise TimeoutError("route observer did not run") - if not hook_finished.wait(2): - raise TimeoutError("observer failure was not reported") - except BaseException as exc: - errors.append(exc) - finally: - server.close() - - worker = threading.Thread(target=client, daemon=True) - with patch("threading.excepthook", side_effect=observe_thread_failure): - worker.start() - runtime.start() - worker.join(timeout=3) - self.assertFalse(worker.is_alive()) - self.assertEqual(errors, []) + runtime_thread = threading.Thread( + target=runtime.start, + name="smallos-runtime-test", + daemon=True, + ) + runtime_thread.start() + request = ( + "POST {} HTTP/1.1\r\n" + "Host: localhost\r\n" + "Authorization: {}\r\n" + "Content-Length: {}\r\n\r\n" + ).format(hostile_path, authorization_secret, len(body_secret)).encode("ascii") + received = [self._exchange(server.port, request + body_secret)] + received.append(self._request(server.port, "/health")) + self.assertTrue(observer_finished.wait(2), "route observer did not run") + server.close() + runtime_thread.join(timeout=3) + self.assertFalse(runtime_thread.is_alive()) self.assertEqual(len(observed), 1) event = observed[0] self.assertEqual(event.route_id, "regex-route-1") @@ -284,8 +271,17 @@ def client() -> None: self.assertNotIn(body_secret, observer_graph) self.assertEqual(server.route_observer_failures, 1) self.assertEqual(server.dropped_route_error_events, 0) - self.assertEqual(len(hook_events), 1) - self.assertIsInstance(hook_events[0].exc_value, RuntimeError) + self.assertEqual(observer_threads, [runtime_thread]) + self.assertNotIn( + "smallserver-route-observer", + {thread.name for thread in threading.enumerate()}, + ) + channel = server._route_observer_channel + self.assertIsNotNone(channel) + assert channel is not None + self.assertFalse(channel.accepting) + self.assertEqual(list(channel.events), []) + self.assertIsNone(channel.task) self.assertTrue(received[0].startswith(b"HTTP/1.1 500 Internal Server Error\r\n")) self.assertNotIn(hostile_path.encode("ascii"), received[0]) self.assertTrue(received[1].startswith(b"HTTP/1.1 200 OK\r\n")) @@ -345,3 +341,23 @@ def _reachable_objects(root): elif hasattr(value, "__dict__"): pending.append(vars(value)) return result + + +def _reachable_container_values(root): + """Walk frame-local containers without traversing scheduler object graphs.""" + pending = [root] + seen = set() + result = [] + while pending: + value = pending.pop() + identity = id(value) + if identity in seen: + continue + seen.add(identity) + result.append(value) + if isinstance(value, dict): + pending.extend(value.keys()) + pending.extend(value.values()) + elif isinstance(value, (tuple, list, set, frozenset)): + pending.extend(value) + return result