Compare commits

..

7 Commits

Author SHA1 Message Date
didericis-codex 0255d29ff1 refactor(supervisor): separate MCP dispatch from transport
test / image-input-builds (pull_request) Successful in 42s
test / unit (pull_request) Successful in 59s
test / integration-docker (pull_request) Has been cancelled
tracker-policy-pr / check-pr (pull_request) Successful in 8s
test / coverage (pull_request) Has been cancelled
2026-07-27 02:58:30 +00:00
didericis-codex c916f4c1ed refactor(egress): extract outbound DLP request stage
test / integration-docker (pull_request) Blocked by required conditions
test / coverage (pull_request) Blocked by required conditions
test / unit (pull_request) Has started running
test / image-input-builds (pull_request) Has started running
tracker-policy-pr / check-pr (pull_request) Successful in 8s
2026-07-27 02:58:22 +00:00
didericis-codex 8abf52ccfa refactor(egress): extract request policy stages 2026-07-27 02:58:22 +00:00
didericis-codex 2234145ab4 fix(orchestrator): bound streamed request bodies
test / unit (pull_request) Waiting to run
test / integration-docker (pull_request) Blocked by required conditions
test / image-input-builds (pull_request) Waiting to run
test / coverage (pull_request) Blocked by required conditions
tracker-policy-pr / check-pr (pull_request) Failing after 10s
2026-07-27 02:58:18 +00:00
didericis-codex a044c37ad5 fix(orchestrator): keep host client dependency-free 2026-07-27 02:58:18 +00:00
didericis-codex 7ae8085606 refactor(orchestrator): replace manual HTTP dispatch with FastAPI 2026-07-27 02:58:18 +00:00
didericis-codex 44dbca82db build(orchestrator): pin FastAPI runtime dependencies 2026-07-27 02:58:18 +00:00
5 changed files with 228 additions and 125 deletions
@@ -0,0 +1,79 @@
"""Framework-neutral MCP method and tool dispatch."""
from __future__ import annotations
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.supervisor import types as _sv
class Request(Protocol):
@property
def method(self) -> str: ...
@property
def params(self) -> dict[str, object]: ...
class MethodNotFoundError(Exception):
"""Raised when a JSON-RPC method has no MCP handler."""
Handler = Callable[[dict[str, object]], object]
@dataclass(frozen=True)
class Handlers:
initialize: Handler
tools_list: Handler
list_routes: Handler
check_proposal: Handler
propose: Handler
def dispatch(request: Request, handlers: Handlers) -> object:
"""Route one parsed request without depending on the HTTP server."""
if request.method == "initialize":
return handlers.initialize(request.params)
if request.method == "notifications/initialized":
return None
if request.method == "tools/list":
return handlers.tools_list(request.params)
if request.method != "tools/call":
raise MethodNotFoundError(request.method)
tool = request.params.get("name")
if tool == _sv.TOOL_LIST_EGRESS_ROUTES:
return handlers.list_routes(request.params)
if tool == _sv.TOOL_CHECK_PROPOSAL:
return handlers.check_proposal(request.params)
return handlers.propose(request.params)
def resolved_routes_payload(
resolver: PolicyResolver,
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,
)
body = json.dumps(
{"routes": [route_to_yaml_dict(route) for route in config.routes]},
indent=2,
)
return {"content": [{"type": "text", "text": body}], "isError": False}
__all__ = [
"Handlers",
"MethodNotFoundError",
"dispatch",
"resolved_routes_payload",
]
+42 -50
View File
@@ -58,10 +58,15 @@ import typing
from dataclasses import dataclass
from bot_bottle.constants import IDENTITY_HEADER
from bot_bottle.gateway.egress.context import resolve_client_context
from bot_bottle.gateway.egress.schema import load_config, route_to_yaml_dict
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,
dispatch,
resolved_routes_payload,
)
from bot_bottle.supervisor import types as _sv
@@ -611,6 +616,11 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
try:
result = self._dispatch(req, config)
except MethodNotFoundError as e:
self._write_jsonrpc(
jsonrpc_error(req.id, ERR_METHOD_NOT_FOUND, f"method not found: {e}"),
)
return
except _RpcClientError as e:
self._write_jsonrpc(jsonrpc_error(req.id, e.code, e.message))
return
@@ -633,41 +643,37 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
self._write_jsonrpc(jsonrpc_result(req.id, result))
def _dispatch(self, req: JsonRpcRequest, config: ServerConfig) -> object:
method = req.method
if method == "initialize":
return handle_initialize(req.params)
if method == "notifications/initialized":
return None # ack-only
if method == "tools/list":
return handle_tools_list(req.params)
if method == "tools/call":
# `list-egress-routes` is read-only introspection. The shared gateway
# has no static route table (routes are resolved per request by
# source IP), so answer it from the calling bottle's resolved policy.
# Otherwise the agent sees an empty allowlist and composes an egress
# proposal that *replaces* the live routes instead of extending them
# — silently dropping base routes like api.anthropic.com on approval.
if req.params.get("name") == _sv.TOOL_LIST_EGRESS_ROUTES:
return self._resolved_routes_payload()
resolver = self._resolver_or_fail()
source_ip = self.client_address[0]
token = self._identity_token()
# `check-proposal` is a non-blocking read of the calling bottle's
# own queue — attributed by (source_ip, identity_token) like a
# proposal, but it never queues or blocks.
if req.params.get("name") == _sv.TOOL_CHECK_PROPOSAL:
return handle_check_proposal(
req.params, resolver=resolver,
source_ip=source_ip, identity_token=token,
)
# The control plane attributes the proposal to the source-IP + token
# resolved bottle, so the one shared queue holds each bottle's
# proposal under its own id — no slug is asserted by this daemon.
return handle_tools_call(
req.params, config, resolver=resolver,
source_ip=source_ip, identity_token=token,
def check(params: dict[str, object]) -> object:
return handle_check_proposal(
params,
resolver=self._resolver_or_fail(),
source_ip=self.client_address[0],
identity_token=self._identity_token(),
)
raise _RpcClientError(ERR_METHOD_NOT_FOUND, f"method not found: {method}")
def propose(params: dict[str, object]) -> object:
return handle_tools_call(
params,
config,
resolver=self._resolver_or_fail(),
source_ip=self.client_address[0],
identity_token=self._identity_token(),
)
return dispatch(
req,
DispatchHandlers(
initialize=handle_initialize,
tools_list=handle_tools_list,
list_routes=lambda _params: resolved_routes_payload(
self._resolver_or_fail(),
self.client_address[0],
self._identity_token(),
),
check_proposal=check,
propose=propose,
),
)
def _identity_token(self) -> str:
"""The agent's per-bottle identity token from the request header (the
@@ -686,20 +692,6 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
raise _RpcInternalError("supervise server has no policy resolver")
return resolver
def _resolved_routes_payload(self) -> dict[str, object]:
"""The calling bottle's live egress routes as the `list-egress-routes`
JSON payload, resolved by (source_ip, identity token). Fail-closed: an
unattributed source or an unreachable orchestrator yields an empty route
list (never another bottle's), courtesy of `resolve_client_context`."""
resolver = self._resolver_or_fail()
conf, _slug, _tokens = resolve_client_context(
resolver, self.client_address[0], self._identity_token(),
)
body = json.dumps(
{"routes": [route_to_yaml_dict(r) for r in conf.routes]}, indent=2,
)
return {"content": [{"type": "text", "text": body}], "isError": False}
def _write_jsonrpc(self, body: bytes) -> None:
self.send_response(200)
self.send_header("Content-Type", "application/json")
@@ -13,9 +13,7 @@ resource-consuming boundary revalidate the assumptions it acts on. This
finishes the focused quality work begun under #444 without broad rewrites:
cleanup cannot act on stale identities, policy introspection cannot publish a
fabricated empty policy, gateway servers bound untrusted work, and daemon
shutdown does not emit uncaught background-thread failures. Shared
control-plane storage and gateway credential provisioning also enforce their
filesystem security contract before sensitive data is written.
shutdown does not emit uncaught background-thread failures.
## Problem
@@ -38,27 +36,6 @@ misleading behavior:
destructive plan.
5. Gateway log-pump threads race stream closure during shutdown and emit
uncaught exceptions even when shutdown otherwise succeeds.
6. Firecracker discovers VMs through whitespace-split `pgrep -a` output.
A configured cache path containing spaces can hide a live VM from the
snapshot and make its run directory appear orphaned.
7. Docker cleanup asks compose for its project snapshot in best-effort mode.
A transient query failure can therefore become an empty stopped-project
set and authorize deletion of associated state directories.
8. Firecracker artifact downloads and registry publication have no network
deadline, so an unresponsive registry can hold setup or release work
indefinitely.
9. Authenticated secret blobs select the unauthenticated legacy decoder when
their in-band version prefix is changed, allowing storage tampering to
bypass tag verification.
10. Cleanup executes the entire post-confirmation snapshot rather than the
intersection with what the operator saw, and mutation failures are not
reflected in the command result.
11. Git smart-HTTP can retain sixteen 100 MiB request bodies concurrently,
cleanup mutations have no subprocess deadline, and Firecracker signalling
failures bypass shared mutation accounting.
12. SQLite creates the shared control-plane database before its mode is
restricted, then suppresses permission-repair failures. Gateway transports
also differ in whether copied deploy-key modes are preserved.
These are one design problem: state used to authorize deletion, replacement,
or resource allocation must be authoritative at the point of use.
@@ -69,8 +46,6 @@ or resource allocation must be authoritative at the point of use.
appeared in a pre-confirmation snapshot.
- Firecracker cleanup proves immediately before action that a PID is still the
same Firecracker process and that a run directory is still orphaned.
- Firecracker process discovery reads NUL-delimited argv from `/proc`; paths
are never reconstructed from whitespace-delimited process listings.
- All backend cleanup discovery primitives raise a typed enumeration error on
operational failure. No backend may independently continue from a partial
snapshot.
@@ -85,22 +60,6 @@ or resource allocation must be authoritative at the point of use.
callers because bottles themselves are untrusted.
- Gateway child-output pumping treats expected stream closure during shutdown
as completion while preserving diagnostics for unexpected failures.
- Artifact pull, existence-check, and publication requests use explicit
network deadlines.
- Persisted secrets accept only the authenticated format. The schema migration
intentionally clears legacy rows; local agents are reprovisioned rather
than retaining a ciphertext-controlled downgrade path.
- Cleanup executes only resources present in both the displayed and current
authoritative plans, attempts every approved mutation, and returns failure
when any mutation does not complete.
- Git request bodies spool to disk behind a separate heavy-work semaphore;
cleanup commands have configurable deadlines; Firecracker signalling
failures aggregate while identity-verification uncertainty still aborts.
- The shared database directory and file are private before SQLite writes any
control-plane state; an inability to enforce those modes aborts startup.
- Gateway credential directories and files receive explicit private modes
inside the gateway, independent of Docker, Apple Container, or SSH copy
semantics.
- Unit tests cover PID/path reuse, partial backend enumeration, transient
policy resolution failure, slow bodies, concurrency saturation, and stream
closure races.
@@ -135,13 +94,6 @@ Backend-specific primitives define how to identify a resource. Firecracker
uses process start identity plus canonical config/run paths; container
backends use authoritative CLI queries and stable resource names/labels.
Container engines expose destructive name-based commands without a portable
compare-and-delete operation. Cleanup therefore refreshes after confirmation
and requires every discovery query to succeed, minimizing but not claiming to
eliminate the final name-reuse race. A future engine-specific stable-ID
primitive may close that residual window without moving control flow back
into each backend.
### Enforcement state versus introspection state
Egress enforcement retains its deny-all fallback because uncertainty must not
@@ -163,17 +115,6 @@ The gateway output pump catches only stream-closure exceptions expected after
the supervisor closes child pipes. Other I/O failures remain visible and are
reported through the supervisor's normal diagnostic channel.
### Shared filesystem security
The common SQLite store owns database creation for every backend. It creates
the parent directory and an empty database with private modes before opening
SQLite, repairs existing modes, verifies the resulting state, and propagates
every enforcement failure. Backend launchers do not duplicate this policy.
The backend-neutral gateway provisioner likewise applies directory and file
modes after transport copies complete. This avoids relying on copy behavior
that differs among Docker, Apple Container, and Firecracker's SSH transport.
## Implementation chunks
1. Existing fail-closed security and backend enumeration fixes.
@@ -183,14 +124,6 @@ that differs among Docker, Apple Container, and Firecracker's SSH transport.
5. Shared cleanup refresh/revalidation plus authoritative macOS discovery.
6. Strict supervisor introspection and bounded supervisor/Git HTTP work.
7. Gateway shutdown log-pump closure handling.
8. Lossless Firecracker process identities, authoritative Docker cleanup
queries, and bounded Firecracker artifact transfers.
9. Mandatory authenticated secret storage, shared cleanup-plan intersection
and mutation accounting, and contained Git backend process failures.
10. Disk-spooled and separately bounded Git bodies, cleanup command deadlines,
and classified Firecracker signalling failures.
11. Fail-closed shared database creation and backend-neutral gateway credential
permissions.
## Open questions
+25 -7
View File
@@ -13,6 +13,7 @@ import tempfile
import threading
import time
import types
import typing
import unittest
from pathlib import Path
@@ -47,6 +48,7 @@ from bot_bottle.gateway.supervisor.server import (
jsonrpc_error,
jsonrpc_result,
parse_jsonrpc,
resolved_routes_payload,
validate_proposed_file,
)
@@ -701,9 +703,14 @@ class TestResolvedRoutesPayload(unittest.TestCase):
" - host: api.anthropic.com\n"
" - host: www.google.com\n"
)
payload = _handler(
_FakeSuperviseResolver(bottle_id="b1", policy=policy)
)._resolved_routes_payload()
payload = resolved_routes_payload(
typing.cast(
supervise_server.PolicyResolver,
_FakeSuperviseResolver(bottle_id="b1", policy=policy),
),
_SRC,
_TOK,
)
assert payload is not None
self.assertFalse(payload["isError"]) # type: ignore[index]
data = json.loads(payload["content"][0]["text"]) # type: ignore[index]
@@ -713,9 +720,14 @@ class TestResolvedRoutesPayload(unittest.TestCase):
def test_orchestrator_error_fails_closed_to_empty(self) -> None:
# resolve_client_context swallows resolver errors → deny-all (empty),
# never another bottle's routes.
payload = _handler(
_FakeSuperviseResolver(raises=True)
)._resolved_routes_payload()
payload = resolved_routes_payload(
typing.cast(
supervise_server.PolicyResolver,
_FakeSuperviseResolver(raises=True),
),
_SRC,
_TOK,
)
assert payload is not None
data = json.loads(payload["content"][0]["text"]) # type: ignore[index]
self.assertEqual([], data["routes"])
@@ -724,7 +736,13 @@ class TestResolvedRoutesPayload(unittest.TestCase):
# A server without a resolver is a misconfig, not a mode: raise rather
# than list anything.
with self.assertRaises(_RpcInternalError):
_handler(None)._resolved_routes_payload()
_handler(None)._dispatch(
parse_jsonrpc(
b'{"jsonrpc":"2.0","id":1,"method":"tools/call",'
b'"params":{"name":"list-egress-routes"}}',
),
ServerConfig(),
)
class TestNonBlockingSupervise(unittest.TestCase):
@@ -0,0 +1,81 @@
"""Unit tests for framework-neutral supervisor MCP dispatch."""
from __future__ import annotations
import unittest
from dataclasses import dataclass
from bot_bottle.gateway.supervisor.mcp_dispatch import (
Handlers,
MethodNotFoundError,
dispatch,
)
from bot_bottle.supervisor import types as _sv
@dataclass(frozen=True)
class _Request:
method: str
params: dict[str, object]
class TestDispatch(unittest.TestCase):
def setUp(self) -> None:
self.calls: list[str] = []
def handler(name: str):
def call(_params: dict[str, object]) -> str:
self.calls.append(name)
return name
return call
self.handlers = Handlers(
initialize=handler("initialize"),
tools_list=handler("tools_list"),
list_routes=handler("list_routes"),
check_proposal=handler("check_proposal"),
propose=handler("propose"),
)
def request(self, method: str, **params: object) -> _Request:
return _Request(method=method, params=params)
def test_routes_protocol_methods(self) -> None:
self.assertEqual(
"initialize", dispatch(self.request("initialize"), self.handlers),
)
self.assertEqual(
"tools_list", dispatch(self.request("tools/list"), self.handlers),
)
self.assertIsNone(
dispatch(self.request("notifications/initialized"), self.handlers),
)
def test_routes_each_tool_class(self) -> None:
self.assertEqual(
"list_routes",
dispatch(
self.request("tools/call", name=_sv.TOOL_LIST_EGRESS_ROUTES),
self.handlers,
),
)
self.assertEqual(
"check_proposal",
dispatch(
self.request("tools/call", name=_sv.TOOL_CHECK_PROPOSAL),
self.handlers,
),
)
self.assertEqual(
"propose",
dispatch(self.request("tools/call", name=_sv.TOOL_EGRESS_ALLOW), self.handlers),
)
def test_unknown_method_is_typed(self) -> None:
with self.assertRaisesRegex(MethodNotFoundError, "unknown"):
dispatch(self.request("unknown"), self.handlers)
if __name__ == "__main__":
unittest.main()