From c00097d99870c3aaeecb22224ee4827aa7d53132 Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 27 Jul 2026 03:14:30 +0000 Subject: [PATCH] fix(gateway): bound stdlib HTTP request work --- bot_bottle/gateway/bounded_http.py | 109 ++++++++++++++++++ bot_bottle/gateway/git_gate/http_backend.py | 34 +++--- bot_bottle/gateway/supervisor/mcp_dispatch.py | 27 +++-- bot_bottle/gateway/supervisor/server.py | 62 ++++++---- tests/unit/test_gateway_bounded_http.py | 79 +++++++++++++ tests/unit/test_supervise_server.py | 18 ++- 6 files changed, 280 insertions(+), 49 deletions(-) create mode 100644 bot_bottle/gateway/bounded_http.py create mode 100644 tests/unit/test_gateway_bounded_http.py diff --git a/bot_bottle/gateway/bounded_http.py b/bot_bottle/gateway/bounded_http.py new file mode 100644 index 00000000..730b70b4 --- /dev/null +++ b/bot_bottle/gateway/bounded_http.py @@ -0,0 +1,109 @@ +"""Shared resource boundaries for gateway stdlib HTTP services.""" + +from __future__ import annotations + +import http.server +import socket +import threading +import time +from dataclasses import dataclass +from typing import Any, Protocol + + +class Readable(Protocol): + def read(self, size: int = -1, /) -> bytes: ... + + +@dataclass(frozen=True) +class BodyReadError(Exception): + status: int + message: str + + +def read_declared_body( + stream: Readable, + connection: socket.socket, + raw_length: str | None, + *, + maximum: int, + timeout_seconds: float, + require_length: bool, +) -> bytes: + """Validate and read exactly one declared body under a read deadline.""" + if raw_length is None: + if require_length: + raise BodyReadError(411, "Content-Length required") + raw_length = "0" + try: + length = int(raw_length) + except ValueError as exc: + raise BodyReadError(400, "invalid Content-Length") from exc + if length < 0: + raise BodyReadError(400, "invalid Content-Length") + if length > maximum: + raise BodyReadError(413, "request body too large") + previous_timeout = connection.gettimeout() + deadline = time.monotonic() + timeout_seconds + chunks: list[bytes] = [] + remaining = length + try: + while remaining: + timeout = deadline - time.monotonic() + if timeout <= 0: + raise BodyReadError(408, "request body read timed out") + connection.settimeout(timeout) + chunk = stream.read(min(remaining, 64 * 1024)) + if not chunk: + raise BodyReadError(400, "incomplete request body") + chunks.append(chunk) + remaining -= len(chunk) + except TimeoutError as exc: + raise BodyReadError(408, "request body read timed out") from exc + finally: + connection.settimeout(previous_timeout) + return b"".join(chunks) + + +class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer): + """ThreadingHTTPServer with a hard cap on in-flight request threads.""" + + daemon_threads = True + + def __init__( # pylint: disable=consider-using-with + self, *args, max_workers: int = 32, **kwargs, # type: ignore[no-untyped-def] + ): + if max_workers < 1: + raise ValueError("max_workers must be positive") + self._request_slots = threading.BoundedSemaphore(max_workers) + super().__init__(*args, **kwargs) + + def process_request( + self, request: Any, client_address: Any, + ) -> None: + if not self._request_slots.acquire( # pylint: disable=consider-using-with + blocking=False, + ): + try: + request.sendall( + b"HTTP/1.1 503 Service Unavailable\r\n" + b"Content-Length: 0\r\nConnection: close\r\n\r\n" + ) + finally: + self.shutdown_request(request) + return + try: + super().process_request(request, client_address) + except BaseException: + self._request_slots.release() + raise + + def process_request_thread( + self, request: Any, client_address: Any, + ) -> None: + try: + super().process_request_thread(request, client_address) + finally: + self._request_slots.release() + + +__all__ = ["BodyReadError", "BoundedThreadingHTTPServer", "read_declared_body"] diff --git a/bot_bottle/gateway/git_gate/http_backend.py b/bot_bottle/gateway/git_gate/http_backend.py index c4860e7e..cb33d239 100644 --- a/bot_bottle/gateway/git_gate/http_backend.py +++ b/bot_bottle/gateway/git_gate/http_backend.py @@ -22,11 +22,16 @@ import os import subprocess import sys import typing -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from http.server import BaseHTTPRequestHandler from pathlib import Path from urllib.parse import urlsplit from bot_bottle.constants import GIT_GATE_TIMEOUT_SECS, IDENTITY_HEADER +from bot_bottle.gateway.bounded_http import ( + BodyReadError, + BoundedThreadingHTTPServer, + read_declared_body, +) from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver @@ -77,6 +82,8 @@ def resolve_sandbox_root( # Bound memory use while still allowing ordinary git push packfiles. MAX_BODY_BYTES = 100 * 1024 * 1024 +REQUEST_BODY_TIMEOUT_SECONDS = 30.0 +MAX_REQUEST_WORKERS = 16 class GitHttpHandler(BaseHTTPRequestHandler): @@ -184,19 +191,18 @@ class GitHttpHandler(BaseHTTPRequestHandler): value = self.headers.get(header) if value: env[variable] = value - raw_length = self.headers.get("content-length", "0") or "0" try: - length = int(raw_length) - except ValueError: - self.send_error(400, "Bad Content-Length") + body = read_declared_body( + self.rfile, + self.connection, + self.headers.get("content-length"), + maximum=MAX_BODY_BYTES, + timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS, + require_length=False, + ) + except BodyReadError as exc: + self.send_error(exc.status, exc.message) return - if length < 0: - self.send_error(400, "Negative Content-Length") - return - if length > MAX_BODY_BYTES: - self.send_error(413, "Request body too large") - return - body = self.rfile.read(length) if length else b"" proc = subprocess.run( ["git", "http-backend"], input=body, @@ -273,7 +279,9 @@ def main() -> int: "(no single-tenant flat-root fallback)\n" ) return 1 - server = ThreadingHTTPServer(("0.0.0.0", port), GitHttpHandler) + server = BoundedThreadingHTTPServer( + ("0.0.0.0", port), GitHttpHandler, max_workers=MAX_REQUEST_WORKERS, + ) # Resolve each request's sandbox namespace by source IP against the # orchestrator control plane. server.policy_resolver = PolicyResolver(orch_url) # type: ignore[attr-defined] diff --git a/bot_bottle/gateway/supervisor/mcp_dispatch.py b/bot_bottle/gateway/supervisor/mcp_dispatch.py index 79f7a18a..151eb2a2 100644 --- a/bot_bottle/gateway/supervisor/mcp_dispatch.py +++ b/bot_bottle/gateway/supervisor/mcp_dispatch.py @@ -6,9 +6,8 @@ import json from dataclasses import dataclass from typing import Callable, Protocol -from bot_bottle.gateway.egress.context import resolve_client_context -from bot_bottle.gateway.egress.schema import route_to_yaml_dict -from bot_bottle.gateway.policy_resolver import PolicyResolver +from bot_bottle.gateway.egress.schema import load_config, route_to_yaml_dict +from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver from bot_bottle.supervisor import types as _sv @@ -24,6 +23,10 @@ class MethodNotFoundError(Exception): """Raised when a JSON-RPC method has no MCP handler.""" +class RouteResolutionError(Exception): + """The caller's live route table could not be resolved authoritatively.""" + + Handler = Callable[[dict[str, object]], object] @@ -60,10 +63,19 @@ def resolved_routes_payload( source_ip: str, identity_token: str, ) -> dict[str, object]: - """Render the calling bottle's routes, failing closed to an empty list.""" - config, _slug, _tokens = resolve_client_context( - resolver, source_ip, identity_token, - ) + """Render an authoritatively resolved route table for the calling bottle.""" + try: + policy, bottle_id, _tokens = resolver.resolve_policy_and_bottle_id( + source_ip, identity_token, + ) + except PolicyResolveError as exc: + raise RouteResolutionError("orchestrator unavailable") from exc + if not bottle_id: + raise RouteResolutionError("request source is not attributed to a bottle") + try: + config = load_config(policy or "") + except ValueError as exc: + raise RouteResolutionError("resolved policy is invalid") from exc body = json.dumps( {"routes": [route_to_yaml_dict(route) for route in config.routes]}, indent=2, @@ -74,6 +86,7 @@ def resolved_routes_payload( __all__ = [ "Handlers", "MethodNotFoundError", + "RouteResolutionError", "dispatch", "resolved_routes_payload", ] diff --git a/bot_bottle/gateway/supervisor/server.py b/bot_bottle/gateway/supervisor/server.py index 8bfd2fd8..bbd5dc34 100644 --- a/bot_bottle/gateway/supervisor/server.py +++ b/bot_bottle/gateway/supervisor/server.py @@ -51,19 +51,24 @@ from __future__ import annotations import http.server import json import os -import socketserver import sys import time import typing from dataclasses import dataclass from bot_bottle.constants import IDENTITY_HEADER +from bot_bottle.gateway.bounded_http import ( + BodyReadError, + BoundedThreadingHTTPServer, + read_declared_body, +) from bot_bottle.gateway.egress.schema import load_config from bot_bottle.gateway.egress.types import LOG_OFF from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver from bot_bottle.gateway.supervisor.mcp_dispatch import ( Handlers as DispatchHandlers, MethodNotFoundError, + RouteResolutionError, dispatch, resolved_routes_payload, ) @@ -570,6 +575,8 @@ def format_unknown_proposal_text(proposal_id: str) -> str: # Max request body the server accepts. 1 MB is well above any realistic # routes.yaml proposal. MAX_BODY_BYTES = 1 * 1024 * 1024 +REQUEST_BODY_TIMEOUT_SECONDS = 10.0 +MAX_REQUEST_WORKERS = 32 class MCPHandler(http.server.BaseHTTPRequestHandler): @@ -592,19 +599,18 @@ class MCPHandler(http.server.BaseHTTPRequestHandler): self._write_text(405, "use POST for MCP requests\n") def do_POST(self) -> None: - length_header = self.headers.get("Content-Length") - if length_header is None: - self._write_text(411, "Content-Length required\n") - return try: - length = int(length_header) - except ValueError: - self._write_text(400, "invalid Content-Length\n") + body = read_declared_body( + self.rfile, + self.connection, + self.headers.get("Content-Length"), + maximum=MAX_BODY_BYTES, + timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS, + require_length=True, + ) + except BodyReadError as exc: + self._write_text(exc.status, exc.message + "\n") return - if length < 0 or length > MAX_BODY_BYTES: - self._write_text(413, "request body too large\n") - return - body = self.rfile.read(length) try: req = parse_jsonrpc(body) @@ -660,20 +666,25 @@ class MCPHandler(http.server.BaseHTTPRequestHandler): identity_token=self._identity_token(), ) - return dispatch( - req, - DispatchHandlers( - initialize=handle_initialize, - tools_list=handle_tools_list, - list_routes=lambda _params: resolved_routes_payload( + def list_routes(_params: dict[str, object]) -> object: + try: + return resolved_routes_payload( self._resolver_or_fail(), self.client_address[0], self._identity_token(), - ), - check_proposal=check, - propose=propose, - ), - ) + ) + except RouteResolutionError as exc: + raise _RpcInternalError( + f"could not resolve live egress routes: {exc}" + ) from exc + + return dispatch(req, DispatchHandlers( + initialize=handle_initialize, + tools_list=handle_tools_list, + list_routes=list_routes, + check_proposal=check, + propose=propose, + )) def _identity_token(self) -> str: """The agent's per-bottle identity token from the request header (the @@ -711,7 +722,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler): self.wfile.write(encoded) -class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): +class MCPServer(BoundedThreadingHTTPServer): allow_reuse_address = True daemon_threads = True config: ServerConfig = ServerConfig() @@ -720,6 +731,9 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): # closed per request (see `_resolver_or_fail`). policy_resolver: "PolicyResolver | None" = None + def __init__(self, *args, **kwargs): # type: ignore[no-untyped-def] + super().__init__(*args, max_workers=MAX_REQUEST_WORKERS, **kwargs) + # --- Entry point ----------------------------------------------------------- diff --git a/tests/unit/test_gateway_bounded_http.py b/tests/unit/test_gateway_bounded_http.py new file mode 100644 index 00000000..f90e6e0c --- /dev/null +++ b/tests/unit/test_gateway_bounded_http.py @@ -0,0 +1,79 @@ +"""Unit tests for shared gateway stdlib HTTP resource boundaries.""" +# pylint: disable=protected-access + +from __future__ import annotations + +import io +import socket +import unittest + +from bot_bottle.gateway.bounded_http import ( + BodyReadError, + BoundedThreadingHTTPServer, + read_declared_body, +) + + +class _Handler: + pass + + +class _TimeoutStream: + def read(self, _size: int = -1, /) -> bytes: + raise TimeoutError + + +class TestDeclaredBody(unittest.TestCase): + def setUp(self) -> None: + self.left, self.right = socket.socketpair() + + def tearDown(self) -> None: + self.left.close() + self.right.close() + + def test_rejects_incomplete_body(self) -> None: + with self.assertRaisesRegex(BodyReadError, "incomplete"): + read_declared_body( + io.BytesIO(b"short"), + self.left, + "10", + maximum=100, + timeout_seconds=1, + require_length=True, + ) + + def test_maps_read_timeout(self) -> None: + with self.assertRaises(BodyReadError) as caught: + read_declared_body( + _TimeoutStream(), + self.left, + "1", + maximum=100, + timeout_seconds=1, + require_length=True, + ) + self.assertEqual(408, caught.exception.status) + + +class TestBoundedServer(unittest.TestCase): + def test_saturated_server_rejects_without_spawning_thread(self) -> None: + client, peer = socket.socketpair() + with BoundedThreadingHTTPServer( + ("127.0.0.1", 0), _Handler, max_workers=1, # type: ignore[arg-type] + ) as server: + with client, peer: + # Directly reserve the only slot to model an in-flight handler. + self.assertTrue( + server._request_slots.acquire( # pylint: disable=consider-using-with + blocking=False, + ), + ) + try: + server.process_request(client, ("127.0.0.1", 1)) + self.assertIn(b"503 Service Unavailable", peer.recv(1024)) + finally: + server._request_slots.release() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_supervise_server.py b/tests/unit/test_supervise_server.py index 61f30e39..c133f5ec 100644 --- a/tests/unit/test_supervise_server.py +++ b/tests/unit/test_supervise_server.py @@ -717,18 +717,26 @@ class TestResolvedRoutesPayload(unittest.TestCase): hosts = {r["host"] for r in data["routes"]} self.assertEqual({"api.anthropic.com", "www.google.com"}, hosts) - def test_orchestrator_error_fails_closed_to_empty(self) -> None: - # resolve_client_context swallows resolver errors → deny-all (empty), - # never another bottle's routes. + def test_orchestrator_error_is_not_reported_as_empty_policy(self) -> None: + with self.assertRaises(supervise_server.RouteResolutionError): + resolved_routes_payload( + typing.cast( + supervise_server.PolicyResolver, + _FakeSuperviseResolver(raises=True), + ), + _SRC, + _TOK, + ) + + def test_authoritative_empty_policy_remains_successful(self) -> None: payload = resolved_routes_payload( typing.cast( supervise_server.PolicyResolver, - _FakeSuperviseResolver(raises=True), + _FakeSuperviseResolver(bottle_id="b1", policy="routes: []\n"), ), _SRC, _TOK, ) - assert payload is not None data = json.loads(payload["content"][0]["text"]) # type: ignore[index] self.assertEqual([], data["routes"])