diff --git a/bot_bottle/orchestrator/broker.py b/bot_bottle/orchestrator/broker.py index 1cd91687..6d5c567f 100644 --- a/bot_bottle/orchestrator/broker.py +++ b/bot_bottle/orchestrator/broker.py @@ -38,7 +38,21 @@ _ALLOWED_OPS = ("launch", "teardown") class BrokerAuthError(Exception): """A broker request failed provenance or schema verification — bad/absent signature, malformed token, or a payload that doesn't match - the fixed launch-request shape. Fail-closed: the broker must not act.""" + the fixed launch-request shape. Fail-closed: the broker must not act. + + A **definite** negative: nothing was launched, so a caller may safely roll + back as if the op never happened.""" + + +class BrokerUnavailableError(Exception): + """A brokered request could not be carried to a verdict: the broker (or the + wire to it) was unreachable, timed out, or dropped the response. + + Crucially **ambiguous** — unlike `BrokerAuthError`, the op MAY already have + taken effect on the backend before the response was lost, so a caller must + NOT assume it did nothing (e.g. must not roll a registry row back as if no + launch happened, which would orphan a running container). Only the in-process + brokers never raise this; the out-of-process `BrokerClient` does.""" @dataclass(frozen=True) @@ -179,6 +193,7 @@ class StubBroker(LaunchBroker): __all__ = [ "BrokerAuthError", + "BrokerUnavailableError", "LaunchRequest", "SubmitBroker", "LaunchBroker", diff --git a/bot_bottle/orchestrator/broker_client.py b/bot_bottle/orchestrator/broker_client.py index 79879def..62783418 100644 --- a/bot_bottle/orchestrator/broker_client.py +++ b/bot_bottle/orchestrator/broker_client.py @@ -20,15 +20,17 @@ import json import urllib.error import urllib.request -from .broker import BrokerAuthError, LaunchRequest +from .broker import BrokerAuthError, BrokerUnavailableError, LaunchRequest DEFAULT_TIMEOUT_SECONDS = 5.0 class BrokerClientError(RuntimeError): - """A brokered launch/teardown could not be delivered to the host control - server: unreachable, or an unexpected status other than the fail-closed 401 - (which surfaces as `BrokerAuthError`, matching the in-process broker).""" + """The host control server *responded*, but with an unexpected status other + than the fail-closed 401 (which surfaces as `BrokerAuthError`) — e.g. a 502 + backend failure or a malformed body. A definite negative: the host processed + the request and it did not launch. (A *no-response* failure — unreachable / + timeout / dropped — is the ambiguous `BrokerUnavailableError` instead.)""" class BrokerClient: @@ -45,9 +47,11 @@ class BrokerClient: verified and acted on. Raises `BrokerAuthError` on a fail-closed 401 (bad provenance/schema — - the same exception the in-process broker raises), or `BrokerClientError` - if the controller is unreachable, times out, or returns any other - non-success status.""" + the same exception the in-process broker raises); `BrokerClientError` if + the host *responds* with any other non-success status or a malformed + body (a definite negative); or `BrokerUnavailableError` if no response is + obtained (unreachable / timeout / dropped) — the **ambiguous** case, where + the host may already have acted, so the caller must not roll back.""" data = json.dumps({"token": token}).encode() req = urllib.request.Request( f"{self._base}/broker", data=data, method="POST", @@ -65,8 +69,11 @@ class BrokerClient: raise BrokerClientError( f"POST /broker: HTTP {e.code} {detail}".rstrip() ) from e - except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e: - raise BrokerClientError(f"POST /broker: {e}") from e + except (urllib.error.URLError, TimeoutError, OSError) as e: + # No usable response — unreachable, timed out, or the connection + # dropped mid-exchange. Ambiguous: the request may already have + # launched the bottle, so this is NOT a definite failure. + raise BrokerUnavailableError(f"POST /broker: {e}") from e def _json_object(raw: bytes) -> dict[str, object]: diff --git a/bot_bottle/orchestrator/host_server.py b/bot_bottle/orchestrator/host_server.py index e974f887..fe8130a0 100644 --- a/bot_bottle/orchestrator/host_server.py +++ b/bot_bottle/orchestrator/host_server.py @@ -57,6 +57,16 @@ BROKER_SECRET_ENV = "BOT_BOTTLE_BROKER_SECRET" # (8099) — a separate privileged component listening on its own socket. DEFAULT_PORT = 8091 +# Cap on the request body. A signed broker request is tiny, so rejecting anything +# larger *before reading it* keeps a caller that can merely reach the socket (no +# signed token needed) from exhausting memory or a handler thread with a huge +# Content-Length — the signed token, not mere reachability, is the authority. +MAX_BODY_BYTES = 64 * 1024 + +# Per-request socket timeout, bounding how long a stalled / slow-loris caller can +# hold a handler thread on this privileged listener. +REQUEST_TIMEOUT_SECONDS = 15 + def _parse_json_object(body: bytes) -> Json: """Parse a JSON object body. Raises ValueError for non-objects / bad JSON.""" @@ -130,6 +140,10 @@ def dispatch( # pylint: disable=too-many-return-statements class Handler(http.server.BaseHTTPRequestHandler): """Thin stdlib adapter: read the body, call `dispatch`, write JSON.""" + # Socket timeout per request (applied by StreamRequestHandler.setup) so a + # stalled caller can't pin a handler thread on this privileged listener. + timeout = REQUEST_TIMEOUT_SECONDS + # Quiet by default; opt back into stdlib access logging with # BOT_BOTTLE_HOST_CONTROLLER_DEBUG (the controller has its own logging). def log_message(self, format: str, *args: typing.Any) -> None: # noqa: A002 @@ -137,12 +151,21 @@ class Handler(http.server.BaseHTTPRequestHandler): super().log_message(format, *args) def _serve(self, method: str) -> None: - """Read the request body, dispatch it, and write the JSON reply. A - dispatch that raises (it shouldn't — dispatch is total) still returns a - 500 rather than dropping the connection.""" + """Read the request body (bounded), dispatch it, and write the JSON + reply. A dispatch that raises (it shouldn't — dispatch is total) still + returns a 500 rather than dropping the connection.""" server = self.server assert isinstance(server, HostControlServer) - length = int(self.headers.get("Content-Length") or 0) + try: + length = int(self.headers.get("Content-Length") or 0) + except ValueError: + self._reply(400, {"error": "invalid Content-Length"}) + return + if length < 0 or length > MAX_BODY_BYTES: + # Reject before reading: nothing legitimate is this big, so an + # oversized declared length is a bug or a resource-exhaustion attempt. + self._reply(413, {"error": "request body too large"}) + return body = self.rfile.read(length) if length > 0 else b"" try: status, payload = dispatch(server.broker, method, self.path, body) @@ -150,6 +173,10 @@ class Handler(http.server.BaseHTTPRequestHandler): sys.stderr.write(f"host controller: {method} {self.path} failed: {e!r}\n") sys.stderr.flush() status, payload = 500, {"error": f"internal error: {e}"} + self._reply(status, payload) + + def _reply(self, status: int, payload: typing.Mapping[str, object]) -> None: + """Write one JSON response with an explicit Content-Length.""" data = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index 96ca9f19..70de69d7 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -25,7 +25,7 @@ import json from collections.abc import Iterable from datetime import datetime, timezone -from .broker import LaunchRequest, SubmitBroker, sign_request +from .broker import BrokerUnavailableError, LaunchRequest, SubmitBroker, sign_request from .store.registry_store import DEFAULT_REAP_GRACE_SECONDS, BottleRecord, RegistryStore from .supervisor import ( AuditEntry, @@ -111,14 +111,23 @@ class OrchestratorCore: image_ref=image_ref, slot=slot, ) - launched = False try: self._broker.submit(sign_request(req, self._secret)) - launched = True - finally: - if not launched: - self.registry.deregister(rec.bottle_id) - self._tokens.pop(rec.bottle_id, None) + except BrokerUnavailableError: + # Ambiguous delivery failure (timeout / dropped response): the broker + # may already have launched the bottle before the response was lost. + # Do NOT deregister — that would orphan a running container with no + # registry row (reconcile reaps rows, never containers). Keep the row + # so reconcile reaps it iff the bottle is not actually live; surface + # the error so the caller knows the launch is unconfirmed. + raise + except Exception: + # A definite failure — a fail-closed rejection, a backend launch + # error, or the host reporting it did not launch: nothing is running, + # so roll the registry entry back to leave no orphan. + self.registry.deregister(rec.bottle_id) + self._tokens.pop(rec.bottle_id, None) + raise return rec def teardown_bottle(self, bottle_id: str) -> bool: diff --git a/tests/unit/test_orchestrator_broker_client.py b/tests/unit/test_orchestrator_broker_client.py index e85d51d2..e0de4747 100644 --- a/tests/unit/test_orchestrator_broker_client.py +++ b/tests/unit/test_orchestrator_broker_client.py @@ -8,7 +8,11 @@ import unittest import urllib.error from unittest.mock import MagicMock, patch -from bot_bottle.orchestrator.broker import BrokerAuthError, LaunchRequest +from bot_bottle.orchestrator.broker import ( + BrokerAuthError, + BrokerUnavailableError, + LaunchRequest, +) from bot_bottle.orchestrator.broker_client import BrokerClient, BrokerClientError _URLOPEN = "bot_bottle.orchestrator.broker_client.urllib.request.urlopen" @@ -58,14 +62,25 @@ class TestSubmit(unittest.TestCase): with self.assertRaises(BrokerAuthError): self.c.submit("forged") - def test_502_raises_broker_client_error(self) -> None: + def test_502_is_a_definite_client_error(self) -> None: + # The host responded — it processed the request and did not launch, so a + # definite BrokerClientError (the caller may safely roll back). with patch(_URLOPEN, side_effect=_http_error(502, {"error": "docker down"})): with self.assertRaises(BrokerClientError): self.c.submit("tok") - def test_unreachable_raises_broker_client_error(self) -> None: + def test_unreachable_is_ambiguous_unavailable(self) -> None: + # No response at all — the request may already have launched, so the + # AMBIGUOUS BrokerUnavailableError (the caller must NOT roll back). with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")): - with self.assertRaises(BrokerClientError): + with self.assertRaises(BrokerUnavailableError): + self.c.submit("tok") + + def test_timeout_is_ambiguous_unavailable(self) -> None: + # A dropped/late response after the request was sent is the exact orphan + # risk: the host may have launched. Must be ambiguous, not a definite fail. + with patch(_URLOPEN, side_effect=TimeoutError("read timed out")): + with self.assertRaises(BrokerUnavailableError): self.c.submit("tok") def test_malformed_success_body_raises(self) -> None: @@ -79,6 +94,24 @@ class TestSubmit(unittest.TestCase): with self.assertRaises(BrokerAuthError): self.c.submit("forged") + def test_non_json_success_body_raises(self) -> None: + # A 200 whose body isn't JSON is tolerated into {} then fails the + # missing-field check — a definite client error, not a crash. + m = MagicMock() + m.__enter__.return_value.read.return_value = b"not json at all" + with patch(_URLOPEN, return_value=m): + with self.assertRaises(BrokerClientError): + self.c.submit("tok") + + def test_unreadable_error_body_is_tolerated(self) -> None: + # An HTTPError whose body can't be read (fp=None) still classifies by + # status — the error detail is best-effort. + err = urllib.error.HTTPError( + "http://host/broker", 502, "err", {}, None) # type: ignore[arg-type] + with patch(_URLOPEN, side_effect=err): + with self.assertRaises(BrokerClientError): + self.c.submit("tok") + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_host_server.py b/tests/unit/test_orchestrator_host_server.py index 3428280a..0217a2c8 100644 --- a/tests/unit/test_orchestrator_host_server.py +++ b/tests/unit/test_orchestrator_host_server.py @@ -7,10 +7,14 @@ the full sign -> POST -> verify -> act seam over HTTP. from __future__ import annotations +import http.client +import io import json import secrets import threading +import typing import unittest +from unittest.mock import MagicMock, patch from bot_bottle.orchestrator.broker import ( BrokerAuthError, @@ -21,8 +25,12 @@ from bot_bottle.orchestrator.broker import ( ) from bot_bottle.orchestrator.broker_client import BrokerClient from bot_bottle.orchestrator.host_server import ( + MAX_BODY_BYTES, + Handler, + HostControlServer, broker_secret_from_env, dispatch, + main, make_host_server, ) @@ -97,6 +105,16 @@ class TestDispatch(unittest.TestCase): status, _ = dispatch(self.broker, "POST", "/broker", b"{not json") self.assertEqual(400, status) + def test_empty_body_is_missing_token_400(self) -> None: + # Empty body parses to {} (no token) → 400, never reaching the broker. + status, _ = dispatch(self.broker, "POST", "/broker", b"") + self.assertEqual(400, status) + self.assertEqual([], self.broker.launched) + + def test_non_object_body_is_400(self) -> None: + status, _ = dispatch(self.broker, "POST", "/broker", b"[1, 2]") + self.assertEqual(400, status) + def test_unknown_route_404(self) -> None: status, _ = dispatch(self.broker, "GET", "/nope", b"") self.assertEqual(404, status) @@ -152,5 +170,116 @@ class TestSeamRoundTrip(unittest.TestCase): self.assertEqual([], broker.launched) # fail-closed across the wire +class TestRequestLimits(unittest.TestCase): + """The privileged listener must not let a caller that can merely reach the + socket (no signed token) exhaust it via an oversized declared body — and it + rejects on the Content-Length *header*, before reading the body.""" + + def _addr(self) -> tuple[str, int]: + self.broker = StubBroker(secrets.token_bytes(16)) + server = make_host_server(self.broker, "127.0.0.1", 0) + self.addCleanup(server.server_close) + threading.Thread(target=server.serve_forever, daemon=True).start() + self.addCleanup(server.shutdown) + host, port = server.server_address[:2] + return typing.cast(str, host), port + + def test_oversized_content_length_is_rejected_before_reading(self) -> None: + host, port = self._addr() + conn = http.client.HTTPConnection(host, port, timeout=5) + self.addCleanup(conn.close) + # Declare an oversized body but send only a sliver: the server must reject + # on the header before reading, so the caller gets a clean, deterministic + # 413 (no large unread body to race a connection reset). + conn.putrequest("POST", "/broker", skip_accept_encoding=True) + conn.putheader("Content-Type", "application/json") + conn.putheader("Content-Length", str(MAX_BODY_BYTES + 1)) + conn.endheaders() + conn.send(b"{}") # far short of the declared length; never read + resp = conn.getresponse() + self.assertEqual(413, resp.status) + self.assertEqual([], self.broker.launched) # never reached the broker + + +class TestServeUnit(unittest.TestCase): + """Drive `Handler._serve` directly (no socket). The real per-request handler + runs in a daemon thread whose coverage/trace data is lost, so the + bounded-body and error paths are exercised here in the main thread instead.""" + + def _handler(self, broker: LaunchBroker, headers: dict[str, str], + body: bytes = b"") -> tuple[Handler, MagicMock]: + server = HostControlServer.__new__(HostControlServer) + server.broker = broker + h = Handler.__new__(Handler) + h.server = server + h.headers = headers # type: ignore[assignment] — dict is a valid .get() stand-in + h.path = "/broker" + h.rfile = io.BytesIO(body) + h.wfile = io.BytesIO() + send_response = MagicMock() + h.send_response = send_response # type: ignore[method-assign] + h.send_header = MagicMock() # type: ignore[method-assign] + h.end_headers = MagicMock() # type: ignore[method-assign] + return h, send_response + + def test_oversized_content_length_is_413(self) -> None: + broker = StubBroker(secrets.token_bytes(16)) + h, send_response = self._handler(broker, {"Content-Length": str(MAX_BODY_BYTES + 1)}) + h.do_POST() # exercises do_POST -> _serve + send_response.assert_called_once_with(413) + self.assertEqual([], broker.launched) # rejected before the broker + + def test_invalid_content_length_is_400(self) -> None: + h, send_response = self._handler(StubBroker(secrets.token_bytes(16)), + {"Content-Length": "not-a-number"}) + h._serve("POST") + send_response.assert_called_once_with(400) + + def test_valid_request_dispatches_200(self) -> None: + secret = secrets.token_bytes(16) + broker = StubBroker(secret) + body = _body({"token": sign_request( + LaunchRequest(op="teardown", bottle_id="b1"), secret)}) + h, send_response = self._handler(broker, {"Content-Length": str(len(body))}, body) + h._serve("POST") + send_response.assert_called_once_with(200) + self.assertEqual(["b1"], [r.bottle_id for r in broker.torn_down]) + + def test_dispatch_exception_becomes_500(self) -> None: + # dispatch is total, but the handler still guards it: a raised dispatch + # returns 500 rather than dropping the connection. + h, send_response = self._handler( + StubBroker(secrets.token_bytes(16)), {"Content-Length": "0"}) + with patch("bot_bottle.orchestrator.host_server.dispatch", + side_effect=RuntimeError("boom")): + h._serve("POST") + send_response.assert_called_once_with(500) + + def test_health_over_do_get(self) -> None: + h, send_response = self._handler(StubBroker(secrets.token_bytes(16)), {}) + h.path = "/health" + h.do_GET() + send_response.assert_called_once_with(200) + + +class TestMain(unittest.TestCase): + def test_fail_closed_without_secret(self) -> None: + with patch("bot_bottle.orchestrator.host_server.broker_secret_from_env", + return_value=None): + self.assertEqual(2, main(["--port", "0"])) + + def test_serves_then_shuts_down_cleanly(self) -> None: + fake = MagicMock() + fake.server_address = ("127.0.0.1", 0) + fake.serve_forever.side_effect = KeyboardInterrupt + with patch("bot_bottle.orchestrator.host_server.broker_secret_from_env", + return_value=b"k"), \ + patch("bot_bottle.orchestrator.host_server.make_host_server", + return_value=fake): + self.assertEqual(0, main(["--port", "0"])) + fake.serve_forever.assert_called_once() + fake.server_close.assert_called_once() + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_main.py b/tests/unit/test_orchestrator_main.py new file mode 100644 index 00000000..366eaf12 --- /dev/null +++ b/tests/unit/test_orchestrator_main.py @@ -0,0 +1,64 @@ +"""Unit: the orchestrator dev-harness entrypoint (`python -m bot_bottle.orchestrator`). + +Exercises broker selection (stub / docker / http) and the fail-closed http path, +patching `make_server` so the serve loop returns instead of blocking. +""" + +from __future__ import annotations + +import os +import secrets +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +from bot_bottle.orchestrator.__main__ import main + + +def _fake_server() -> MagicMock: + fake = MagicMock() + fake.server_address = ("127.0.0.1", 0) + # Break out of serve_forever immediately, exercising the try/finally. + fake.serve_forever.side_effect = KeyboardInterrupt + return fake + + +class TestMain(unittest.TestCase): + def _run(self, broker: str, env: dict[str, str] | None = None) -> tuple[int, MagicMock]: + fake = _fake_server() + with tempfile.TemporaryDirectory() as d: + argv = ["--db", str(Path(d) / "r.db"), "--port", "0", "--broker", broker] + with patch("bot_bottle.orchestrator.__main__.make_server", return_value=fake), \ + patch.dict("os.environ", env or {}, clear=False): + if env is None: + os.environ.pop("BOT_BOTTLE_BROKER_SECRET", None) + rc = main(argv) + return rc, fake + + def test_stub_broker_serves_and_closes(self) -> None: + rc, fake = self._run("stub") + self.assertEqual(0, rc) + fake.serve_forever.assert_called_once() + fake.server_close.assert_called_once() + + def test_docker_broker_serves(self) -> None: + rc, _ = self._run("docker") + self.assertEqual(0, rc) + + def test_http_broker_with_secret_serves(self) -> None: + rc, _ = self._run( + "http", env={"BOT_BOTTLE_BROKER_SECRET": secrets.token_bytes(16).hex()}) + self.assertEqual(0, rc) + + def test_http_broker_without_secret_exits(self) -> None: + # Fail-closed: --broker http with no shared secret is a usage error. + with tempfile.TemporaryDirectory() as d: + with patch.dict("os.environ", {}, clear=False): + os.environ.pop("BOT_BOTTLE_BROKER_SECRET", None) + with self.assertRaises(SystemExit): + main(["--db", str(Path(d) / "r.db"), "--broker", "http"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index 7f9b12f8..c9031209 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -11,7 +11,12 @@ from contextlib import closing from pathlib import Path from unittest.mock import patch -from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker +from bot_bottle.orchestrator.broker import ( + BrokerUnavailableError, + LaunchBroker, + LaunchRequest, + StubBroker, +) from bot_bottle.orchestrator.store.registry_store import RegistryStore from bot_bottle.orchestrator.service import OrchestratorCore from bot_bottle.orchestrator.store.secret_store import new_env_var_secret @@ -25,8 +30,8 @@ from bot_bottle.orchestrator.supervisor import ( class _FailingBroker(LaunchBroker): - """Verifies the token like any broker, then fails the launch — to - exercise the orchestrator's registry rollback.""" + """Verifies the token like any broker, then fails the launch *definitely* — + to exercise the orchestrator's registry rollback.""" def _launch(self, req: LaunchRequest) -> None: raise RuntimeError("launch failed") @@ -35,6 +40,18 @@ class _FailingBroker(LaunchBroker): pass +class _UnavailableBroker(LaunchBroker): + """Verifies the token, then raises the *ambiguous* BrokerUnavailableError — + the host may already have launched — so the orchestrator must KEEP the + registry row rather than orphan a running container.""" + + def _launch(self, req: LaunchRequest) -> None: + raise BrokerUnavailableError("delivery dropped after send") + + def _teardown(self, req: LaunchRequest) -> None: + pass + + class TestOrchestrator(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() @@ -136,11 +153,20 @@ class TestOrchestrator(unittest.TestCase): self.assertIsNotNone(self.orch.resolve("10.243.0.3", rec.identity_token)) self.assertIsNone(self.orch.resolve("10.243.0.3", "wrong-token")) - def test_launch_rolls_back_registry_on_broker_failure(self) -> None: + def test_launch_rolls_back_registry_on_definite_broker_failure(self) -> None: orch = OrchestratorCore(self.store, _FailingBroker(self.secret), self.secret) with self.assertRaises(RuntimeError): orch.launch_bottle("10.243.0.9") - self.assertEqual([], self.store.all()) # no orphan + self.assertEqual([], self.store.all()) # no orphan row + + def test_launch_keeps_registry_on_ambiguous_broker_failure(self) -> None: + # The host may already have launched the bottle before the response was + # lost, so deregistering would orphan a running container with no row. + # The row is kept for reconcile to reap iff the bottle is not live. + orch = OrchestratorCore(self.store, _UnavailableBroker(self.secret), self.secret) + with self.assertRaises(BrokerUnavailableError): + orch.launch_bottle("10.243.0.9") + self.assertEqual(1, len(self.store.all())) # row survives — no orphan container def test_gateway_status_reports_unconfigured(self) -> None: # The orchestrator no longer owns a standalone gateway lifecycle; the