feat(supervise+orchestrator): slice 12 — multi-tenant git-gate hook + supervise server
Finishes the supervise data-plane writers for the shared gateway. Both still keyed off a single SUPERVISE_BOTTLE_SLUG env; now each proposal is attributed to the calling bottle, keeping slices 10/11's per-bottle model. - git_http_backend: in consolidated mode stamp SUPERVISE_BOTTLE_SLUG= <bottle_id> into the CGI env. The gitleaks-allow pre-receive hook runs as a child of the `git http-backend` we spawn, so it inherits the stamp and queues under the right bottle — no change to the (fragile) embedded hook. The bottle id is the namespaced root's final component (slice 10), the same key egress uses. Single-tenant leaves the container-stamped slug. - supervise_server: resolve the proposal slug per request by source IP when BOT_BOTTLE_ORCHESTRATOR_URL is set (`_attributed_config`), fail-closed — an unattributed / unreachable source raises rather than queue under the wrong or empty slug. `serve`/`main` build + attach the resolver and make the env slug optional in consolidated mode. Single-tenant unchanged. Tests: a real git push proving the slug reaches the hook's env; the supervise attribution matrix (single-tenant slug kept, source-IP bottle bound, unattributed + resolver-error fail closed). Remaining supervise gap (noted): websocket DLP still self.config-only. pyright 0 errors; pylint 9.83/10; unit suite green (1705 tests; the 13 test_sidecar_init /bin/sleep errors are pre-existing NixOS-local noise). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
This commit is contained in:
@@ -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
|
||||
# `<base>/<bottle_id>`, 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"),
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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=<bottle_id> 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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user