diff --git a/bot_bottle/git_http_backend.py b/bot_bottle/git_http_backend.py index 77a151e..25275ac 100644 --- a/bot_bottle/git_http_backend.py +++ b/bot_bottle/git_http_backend.py @@ -188,6 +188,14 @@ class GitHttpHandler(BaseHTTPRequestHandler): "SERVER_PORT": str(self.server.server_port), # type: ignore "SERVER_PROTOCOL": self.request_version, }) + # Consolidated mode: attribute the gitleaks-allow supervise proposal + # (written by receive-pack's pre-receive hook, a child of the CGI we + # spawn below) to the calling bottle. The namespaced root is + # `/`, so its final component is the bottle id — the + # same per-bottle key egress uses. Single-tenant leaves the hook's + # container-stamped SUPERVISE_BOTTLE_SLUG untouched. + if getattr(self.server, "policy_resolver", None) is not None: + env["SUPERVISE_BOTTLE_SLUG"] = sandbox_root.name for header, variable in ( ("accept", "HTTP_ACCEPT"), ("content-encoding", "HTTP_CONTENT_ENCODING"), diff --git a/bot_bottle/supervise_server.py b/bot_bottle/supervise_server.py index 089e106..e6c570d 100644 --- a/bot_bottle/supervise_server.py +++ b/bot_bottle/supervise_server.py @@ -15,6 +15,12 @@ The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at container creation by the backend's start step). SUPERVISE_DB_PATH points at the bind-mounted host database. +Consolidated (PRD 0070): when BOT_BOTTLE_ORCHESTRATOR_URL is set, one +shared server fronts every bottle and attributes each proposal to the +calling bottle by source IP (resolved from the orchestrator) instead of a +fixed slug — an unattributed source fails closed. Unset → the legacy +per-bottle single-tenant server, unchanged. + Speaks MCP over HTTP+JSON-RPC. Methods handled: * `initialize` — handshake; returns server info + caps. @@ -40,16 +46,18 @@ import time import typing import urllib.error import urllib.request -from dataclasses import dataclass +from dataclasses import dataclass, replace try: # Same-directory imports inside the bundle container; these files are # COPYed flat under /app by Dockerfile.sidecars. from egress_addon_core import LOG_OFF, load_config + from policy_resolver import PolicyResolveError, PolicyResolver import supervise as _sv except ModuleNotFoundError: # Package imports for host-side tests and tooling. from .egress_addon_core import LOG_OFF, load_config + from .policy_resolver import PolicyResolveError, PolicyResolver from . import supervise as _sv @@ -73,6 +81,12 @@ DEFAULT_RESPONSE_TIMEOUT_SECONDS = 30.0 MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.05 EGRESS_LIST_TIMEOUT_SECONDS = 5.0 +# Consolidated (multi-tenant) mode: when set, one shared supervise server +# fronts every bottle and attributes each proposal to the calling bottle by +# source IP (resolved from the orchestrator), instead of a single +# SUPERVISE_BOTTLE_SLUG env. Unset → legacy per-bottle single-tenant. +ORCHESTRATOR_URL_ENV = "BOT_BOTTLE_ORCHESTRATOR_URL" + @dataclass(frozen=True) class JsonRpcRequest: @@ -511,9 +525,30 @@ class MCPHandler(http.server.BaseHTTPRequestHandler): if method == "tools/list": return handle_tools_list(req.params) if method == "tools/call": - return handle_tools_call(req.params, config) + # Attribute the proposal to the calling bottle. Single-tenant → the + # env slug on `config`; consolidated → the source-IP-resolved + # bottle id, so one shared server queues each bottle's proposal + # under its own slug. + return handle_tools_call(req.params, self._attributed_config(config)) raise _RpcClientError(ERR_METHOD_NOT_FOUND, f"method not found: {method}") + def _attributed_config(self, config: ServerConfig) -> ServerConfig: + """The ServerConfig with `bottle_slug` bound to *this request's* bottle. + Single-tenant (no resolver): unchanged. Consolidated: the bottle id + attributed from the source IP — **fail-closed**, an unattributed or + unreachable source raises so no proposal is queued under the wrong (or + empty) slug.""" + resolver = getattr(self.server, "policy_resolver", None) + if resolver is None: + return config + try: + bottle_id = resolver.resolve_bottle_id(self.client_address[0]) + except PolicyResolveError as e: + raise _RpcInternalError(f"orchestrator unreachable, cannot attribute: {e}") from e + if not bottle_id: + raise _RpcInternalError("request source is not attributed to a bottle") + return replace(config, bottle_slug=bottle_id) + def _write_jsonrpc(self, body: bytes) -> None: self.send_response(200) self.send_header("Content-Type", "application/json") @@ -537,6 +572,9 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): allow_reuse_address = True daemon_threads = True config: ServerConfig = ServerConfig(bottle_slug="") + # None → single-tenant (proposals use config.bottle_slug); set → consolidated + # (each proposal attributed to the source-IP-resolved bottle). + policy_resolver: "PolicyResolver | None" = None # --- Entry point ----------------------------------------------------------- @@ -548,15 +586,17 @@ def serve( port: int = _sv.SUPERVISE_PORT, bind: str = "0.0.0.0", response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS, + resolver: "PolicyResolver | None" = None, ) -> typing.NoReturn: server = MCPServer((bind, port), MCPHandler) server.config = ServerConfig( bottle_slug=bottle_slug, response_timeout_seconds=response_timeout_seconds, ) + server.policy_resolver = resolver + mode = "multi-tenant" if resolver else f"slug={bottle_slug!r}" sys.stderr.write( - f"supervise listening on {bind}:{port}; " - f"slug={bottle_slug!r}; " + f"supervise listening on {bind}:{port}; {mode}; " f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n" # type: ignore[arg-type] ) sys.stderr.flush() @@ -571,8 +611,12 @@ def serve( def main(argv: list[str]) -> int: del argv # config is env-only, no CLI flags + orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip() + resolver = PolicyResolver(orch_url) if orch_url else None bottle_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "") - if not bottle_slug: + # Consolidated mode resolves the slug per request, so the env slug is + # optional there; single-tenant still requires it. + if not bottle_slug and resolver is None: sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n") return 2 port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT))) @@ -587,6 +631,7 @@ def main(argv: list[str]) -> int: port=port, bind=bind, response_timeout_seconds=response_timeout_seconds, + resolver=resolver, ) return 0 # serve() does not return diff --git a/tests/unit/test_git_http_backend.py b/tests/unit/test_git_http_backend.py index c605eb0..95672d8 100644 --- a/tests/unit/test_git_http_backend.py +++ b/tests/unit/test_git_http_backend.py @@ -13,6 +13,17 @@ from bot_bottle.git_gate import GIT_GATE_TIMEOUT_SECS from bot_bottle.git_http_backend import GitHttpHandler, MAX_BODY_BYTES +class _FixedResolver: + """Maps every source IP to one bottle id (consolidated-mode stub).""" + + def __init__(self, bottle_id: str) -> None: + self._bottle_id = bottle_id + + def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str: + del source_ip, identity_token + return self._bottle_id + + class TestGitHttpBackend(unittest.TestCase): def test_real_git_push_reaches_bare_repo(self): from http.server import ThreadingHTTPServer @@ -94,6 +105,62 @@ class TestGitHttpBackend(unittest.TestCase): ).strip() self.assertEqual(head, cloned) + def test_consolidated_push_stamps_bottle_slug_for_the_hook(self): + # In consolidated mode the backend attributes the push by source IP and + # stamps SUPERVISE_BOTTLE_SLUG= into the CGI env, so the + # gitleaks-allow pre-receive hook queues its proposal under the right + # bottle. The hook here just records what it received. + from http.server import ThreadingHTTPServer + + bottle_id = "bottleab12" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + bare = root / bottle_id / "repo.git" # namespaced per bottle (slice 10) + bare.parent.mkdir(parents=True) + subprocess.run(["git", "init", "--bare", str(bare)], + check=True, capture_output=True, text=True) + subprocess.run( + ["git", "-C", str(bare), "config", "http.receivepack", "true"], + check=True, + ) + capture = root / "slug-capture" + hook = bare / "hooks" / "pre-receive" + hook.write_text( + f"#!/bin/sh\nprintf '%s' \"${{SUPERVISE_BOTTLE_SLUG:-UNSET}}\" > " + f"{capture}\ncat >/dev/null\nexit 0\n" + ) + hook.chmod(0o755) + + old_root = os.environ.get("GIT_PROJECT_ROOT") + os.environ["GIT_PROJECT_ROOT"] = str(root) # base; backend nests per bottle + self.addCleanup(self._restore_env, old_root) + + server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler) + server.policy_resolver = _FixedResolver(bottle_id) # type: ignore[attr-defined] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + self.addCleanup(server.shutdown) + self.addCleanup(server.server_close) + + work = root / "work" + work.mkdir() + subprocess.run(["git", "init"], cwd=work, check=True, + capture_output=True, text=True) + subprocess.run(["git", "config", "user.name", "test"], cwd=work, check=True) + subprocess.run(["git", "config", "user.email", "t@example.invalid"], + cwd=work, check=True) + (work / "README.md").write_text("test\n") + subprocess.run(["git", "add", "README.md"], cwd=work, check=True) + subprocess.run(["git", "commit", "-m", "init"], cwd=work, + check=True, capture_output=True, text=True) + + url = f"http://127.0.0.1:{server.server_port}/repo.git" + subprocess.run( + ["git", "push", url, "HEAD:refs/heads/main"], + cwd=work, check=True, capture_output=True, text=True, timeout=5, + ) + self.assertEqual(bottle_id, capture.read_text()) + def test_post_forwards_git_cgi_headers(self): from http.server import ThreadingHTTPServer diff --git a/tests/unit/test_supervise_server.py b/tests/unit/test_supervise_server.py index a29f032..df81d3a 100644 --- a/tests/unit/test_supervise_server.py +++ b/tests/unit/test_supervise_server.py @@ -6,6 +6,7 @@ import sys import tempfile import threading import time +import types import unittest from pathlib import Path from unittest.mock import patch @@ -629,5 +630,57 @@ class TestHttpEndToEnd(unittest.TestCase): conn.close() +class _FakeResolver: + def __init__(self, bottle_id: str | None = None, raises: bool = False) -> None: + self._bottle_id = bottle_id + self._raises = raises + self.calls: list[str] = [] + + def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str | None: + del identity_token + self.calls.append(source_ip) + if self._raises: + # Raise the exact class supervise_server catches (it imports + # policy_resolver flat inside the bundle, package-side in tests). + raise supervise_server.PolicyResolveError("orchestrator down") + return self._bottle_id + + +def _handler(resolver: object) -> MCPHandler: + """A bare MCPHandler wired with a server (carrying the resolver) and a + client address, enough to exercise `_attributed_config` off-socket.""" + h: MCPHandler = MCPHandler.__new__(MCPHandler) + h.server = types.SimpleNamespace(policy_resolver=resolver) # type: ignore[assignment] + h.client_address = ("10.0.0.7", 4321) + return h + + +class TestAttributedConfig(unittest.TestCase): + """Consolidated supervise: each proposal is attributed to the calling + bottle by source IP; single-tenant keeps the env slug (PRD 0070).""" + + def test_single_tenant_keeps_env_slug(self) -> None: + cfg = _handler(None)._attributed_config(ServerConfig(bottle_slug="dev")) + self.assertEqual("dev", cfg.bottle_slug) + + def test_consolidated_binds_source_ip_bottle(self) -> None: + r = _FakeResolver(bottle_id="bottle-x") + cfg = _handler(r)._attributed_config(ServerConfig(bottle_slug="ignored")) + self.assertEqual("bottle-x", cfg.bottle_slug) # resolved slug wins + self.assertEqual(["10.0.0.7"], r.calls) + + def test_unattributed_source_fails_closed(self) -> None: + with self.assertRaises(_RpcInternalError): + _handler(_FakeResolver(bottle_id=None))._attributed_config( + ServerConfig(bottle_slug="x") + ) + + def test_resolver_error_fails_closed(self) -> None: + with self.assertRaises(_RpcInternalError): + _handler(_FakeResolver(raises=True))._attributed_config( + ServerConfig(bottle_slug="x") + ) + + if __name__ == "__main__": unittest.main()