From 826106f76a96265a9d2fe2b6339893f83f853868 Mon Sep 17 00:00:00 2001 From: codex Date: Sun, 26 Jul 2026 22:56:56 +0000 Subject: [PATCH] fix(orchestrator): bound unauthenticated HTTP requests --- bot_bottle/orchestrator/server.py | 62 ++++++++++++++++++++++++-- tests/unit/test_orchestrator_server.py | 39 +++++++++++++++- 2 files changed, 97 insertions(+), 4 deletions(-) diff --git a/bot_bottle/orchestrator/server.py b/bot_bottle/orchestrator/server.py index 68347cc7..b69d1bb3 100644 --- a/bot_bottle/orchestrator/server.py +++ b/bot_bottle/orchestrator/server.py @@ -59,8 +59,10 @@ import http.server import json import math import os +import socket import socketserver import sys +import threading import typing from urllib.parse import urlsplit @@ -80,6 +82,9 @@ Json = dict[str, object] # token at all, and a compromised gateway holds only `gateway` — neither can # drive the operator routes (approve proposals, rewrite policy, read tokens). ORCHESTRATOR_AUTH_HEADER = "x-bot-bottle-orchestrator-auth" +MAX_BODY_BYTES = 1 * 1024 * 1024 +REQUEST_TIMEOUT_SECONDS = 10.0 +MAX_REQUEST_THREADS = 32 # The routes the data plane (role `gateway`) is allowed to reach — exactly the # per-request lookups PolicyResolver makes. Every other authenticated route is @@ -371,9 +376,30 @@ class Handler(http.server.BaseHTTPRequestHandler): plane down for the caller.""" server = self.server assert isinstance(server, OrchestratorServer) - length = int(self.headers.get("Content-Length") or 0) - body = self.rfile.read(length) if length > 0 else b"" role = server.role_for(self.headers.get(ORCHESTRATOR_AUTH_HEADER, "")) + route = urlsplit(self.path).path.rstrip("/") or "/" + if not (method == "GET" and route == "/health") and role is None: + self._write_json( + 401, {"error": "control-plane authentication required"}, + ) + return + length_header = self.headers.get("Content-Length") + try: + length = int(length_header) if length_header is not None else 0 + except ValueError: + self._write_json(400, {"error": "invalid Content-Length"}) + return + if length < 0: + self._write_json(400, {"error": "invalid Content-Length"}) + return + if length > MAX_BODY_BYTES: + self._write_json(413, {"error": "request body too large"}) + return + try: + body = self.rfile.read(length) if length else b"" + except (TimeoutError, socket.timeout): + self._write_json(408, {"error": "request body read timed out"}) + return try: status, payload = dispatch( server.orchestrator, method, self.path, body, role=role) @@ -387,6 +413,9 @@ class Handler(http.server.BaseHTTPRequestHandler): ) sys.stderr.flush() status, payload = 500, {"error": "internal error"} + self._write_json(status, payload) + + def _write_json(self, status: int, payload: Json) -> None: data = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") @@ -434,8 +463,35 @@ class OrchestratorServer(socketserver.ThreadingMixIn, http.server.HTTPServer): "orchestrator control-plane signing key is required; " "refusing to start without caller authentication" ) + self._request_slots = threading.BoundedSemaphore(MAX_REQUEST_THREADS) super().__init__(address, Handler) + def get_request(self) -> tuple[socket.socket, typing.Any]: + request, client_address = super().get_request() + request.settimeout(REQUEST_TIMEOUT_SECONDS) + return request, client_address + + def process_request( + self, request: socket.socket, client_address: typing.Any, + ) -> None: + # Bound concurrency before ThreadingMixIn creates a worker. Backpressure + # stays in the accept loop instead of allocating an unbounded thread per + # slow or malicious connection. + self._request_slots.acquire() + try: + super().process_request(request, client_address) + except BaseException: + self._request_slots.release() + raise + + def process_request_thread( + self, request: socket.socket, client_address: typing.Any, + ) -> None: + try: + super().process_request_thread(request, client_address) + finally: + self._request_slots.release() + def role_for(self, presented: str) -> str | None: """The verified caller role, or None for a missing/invalid token.""" return CONTROL_PLANE.verify(presented, self._signing_key) @@ -461,5 +517,5 @@ def make_server( __all__ = [ "dispatch", "Handler", "OrchestratorServer", "make_server", "Json", - "ORCHESTRATOR_AUTH_HEADER", + "ORCHESTRATOR_AUTH_HEADER", "MAX_BODY_BYTES", ] diff --git a/tests/unit/test_orchestrator_server.py b/tests/unit/test_orchestrator_server.py index 745d71ae..536d54bf 100644 --- a/tests/unit/test_orchestrator_server.py +++ b/tests/unit/test_orchestrator_server.py @@ -7,6 +7,7 @@ server tests), plus one real-socket round-trip to prove the handler wiring. from __future__ import annotations import base64 +import http.client import io import json import secrets @@ -22,7 +23,7 @@ from unittest.mock import MagicMock, patch from bot_bottle.orchestrator_auth import ROLE_CLI, ROLE_GATEWAY, mint from bot_bottle.orchestrator.broker import StubBroker -from bot_bottle.orchestrator.server import dispatch, make_server +from bot_bottle.orchestrator.server import MAX_BODY_BYTES, dispatch, make_server from bot_bottle.orchestrator.store.registry_store import BottleRecord, RegistryStore from bot_bottle.orchestrator.service import OrchestratorCore from bot_bottle.orchestrator.store.store_manager import StoreManager @@ -251,6 +252,42 @@ class TestDispatch(unittest.TestCase): class TestServerRoundTrip(unittest.TestCase): + def _raw_status(self, content_length: str, *, authenticated: bool = True) -> int: + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + key = "request-limits-key" + server = make_server( + _orchestrator(Path(tmp.name) / "r.db"), + "127.0.0.1", 0, signing_key=key, + ) + self.addCleanup(server.server_close) + threading.Thread(target=server.serve_forever, daemon=True).start() + self.addCleanup(server.shutdown) + conn = http.client.HTTPConnection( + server.server_address[0], server.server_address[1], timeout=5, + ) + self.addCleanup(conn.close) + conn.putrequest("POST", "/bottles") + conn.putheader("Content-Length", content_length) + if authenticated: + conn.putheader( + "x-bot-bottle-orchestrator-auth", mint(ROLE_CLI, key), + ) + conn.endheaders() + return conn.getresponse().status + + def test_rejects_malformed_content_length(self) -> None: + self.assertEqual(400, self._raw_status("not-a-number")) + + def test_rejects_oversized_body_without_reading_it(self) -> None: + self.assertEqual(413, self._raw_status(str(MAX_BODY_BYTES + 1))) + + def test_rejects_unauthenticated_request_before_reading_body(self) -> None: + self.assertEqual( + 401, + self._raw_status(str(MAX_BODY_BYTES), authenticated=False), + ) + def test_http_register_health_attribute(self) -> None: tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup)