refactor(egress): extract outbound DLP request stage

bot_bottle/gateway/supervisor/mcp_dispatch.py

@@ -1,79 +0,0 @@
-"""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",
-]

bot_bottle/gateway/supervisor/server.py

@@ -58,15 +58,10 @@ import typing
 from dataclasses import dataclass
 
 from bot_bottle.constants import IDENTITY_HEADER
-from bot_bottle.gateway.egress.schema import load_config
+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.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
 
 
@@ -616,11 +611,6 @@ 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
@@ -643,37 +633,41 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
         self._write_jsonrpc(jsonrpc_result(req.id, result))
 
     def _dispatch(self, req: JsonRpcRequest, config: ServerConfig) -> object:
-        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(),
-            )
-
-        def propose(params: dict[str, object]) -> 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(
-                params,
-                config,
-                resolver=self._resolver_or_fail(),
-                source_ip=self.client_address[0],
-                identity_token=self._identity_token(),
+                req.params, config, resolver=resolver,
+                source_ip=source_ip, identity_token=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,
-            ),
-        )
+        raise _RpcClientError(ERR_METHOD_NOT_FOUND, f"method not found: {method}")
 
     def _identity_token(self) -> str:
         """The agent's per-bottle identity token from the request header (the
@@ -692,6 +686,20 @@ 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")

docs/prds/0082-authoritative-failure-boundaries.md

@@ -13,7 +13,9 @@ 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.
+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.
 
 ## Problem
 
@@ -45,6 +47,18 @@ misleading behavior:
 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.
@@ -73,6 +87,20 @@ or resource allocation must be authoritative at the point of use.
   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,6 +163,17 @@ 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.
@@ -146,6 +185,12 @@ reported through the supervisor's normal diagnostic channel.
 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
 

tests/unit/test_supervise_server.py

@@ -13,7 +13,6 @@ import tempfile
 import threading
 import time
 import types
-import typing
 import unittest
 from pathlib import Path
 
@@ -48,7 +47,6 @@ from bot_bottle.gateway.supervisor.server import (
     jsonrpc_error,
     jsonrpc_result,
     parse_jsonrpc,
-    resolved_routes_payload,
     validate_proposed_file,
 )
 
@@ -703,14 +701,9 @@ class TestResolvedRoutesPayload(unittest.TestCase):
             "  - host: api.anthropic.com\n"
             "  - host: www.google.com\n"
         )
-        payload = resolved_routes_payload(
-            typing.cast(
-                supervise_server.PolicyResolver,
-                _FakeSuperviseResolver(bottle_id="b1", policy=policy),
-            ),
-            _SRC,
-            _TOK,
-        )
+        payload = _handler(
+            _FakeSuperviseResolver(bottle_id="b1", policy=policy)
+        )._resolved_routes_payload()
         assert payload is not None
         self.assertFalse(payload["isError"])  # type: ignore[index]
         data = json.loads(payload["content"][0]["text"])  # type: ignore[index]
@@ -720,14 +713,9 @@ 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 = resolved_routes_payload(
-            typing.cast(
-                supervise_server.PolicyResolver,
-                _FakeSuperviseResolver(raises=True),
-            ),
-            _SRC,
-            _TOK,
-        )
+        payload = _handler(
+            _FakeSuperviseResolver(raises=True)
+        )._resolved_routes_payload()
         assert payload is not None
         data = json.loads(payload["content"][0]["text"])  # type: ignore[index]
         self.assertEqual([], data["routes"])
@@ -736,13 +724,7 @@ 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)._dispatch(
-                parse_jsonrpc(
-                    b'{"jsonrpc":"2.0","id":1,"method":"tools/call",'
-                    b'"params":{"name":"list-egress-routes"}}',
-                ),
-                ServerConfig(),
-            )
+            _handler(None)._resolved_routes_payload()
 
 
 class TestNonBlockingSupervise(unittest.TestCase):

tests/unit/test_supervisor_mcp_dispatch.py

@@ -1,81 +0,0 @@
-"""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()