Compare commits

...

11 Commits

Author SHA1 Message Date
didericis c5ba33bf0a feat(orchestrator): slice 13c(iv) — gateway joins a shared network
test / unit (pull_request) Successful in 1m5s
test / integration (pull_request) Successful in 22s
test / coverage (pull_request) Successful in 1m15s
lint / lint (push) Successful in 2m18s
The consolidated gateway ran on the default bridge; the model needs it on a
dedicated user-defined network that every agent bottle also joins, reaching
the gateway's egress / git-http / supervise ports by its address (no host
port publishing) — and the source IP the gateway attributes by is the
bottle's address on this network.

- GATEWAY_NETWORK = "bot-bottle-gateway"; DockerGateway gains a `network`
  param.
- ensure_running now ensures the network exists (idempotent create,
  tolerating a concurrent 'already exists') then runs with `--network`.
- Docker picks the subnet; the launcher reads it back (13c source-IP
  allocator) to hand each bottle a pinned address.

pyright 0 errors; pylint 9.83/10; unit suite green (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
2026-07-13 21:55:10 -04:00
didericis 72286f188d feat(orchestrator): slice 13c(iii) — host-side control-plane client
test / unit (pull_request) Successful in 1m9s
test / integration (pull_request) Successful in 29s
test / coverage (pull_request) Successful in 1m18s
lint / lint (push) Successful in 2m10s
The launch path's counterpart to the gateway-side PolicyResolver: a trusted
host-side HTTP client for the orchestrator control plane, so the CLI can
register/teardown/re-policy bottles. Stdlib-only.

- OrchestratorClient.register_bottle(source_ip, *, image_ref, metadata,
  policy) -> RegisteredBottle(bottle_id, identity_token)  (POST /bottles)
- teardown_bottle(id) -> bool (DELETE; 404 -> False, idempotent for cleanup)
- set_policy(id, policy) -> bool (PUT; live reload)
- list_bottles() / health()
- Unlike the fail-closed data-plane resolver, this is the trusted caller: a
  non-2xx (other than the meaningful 404s) raises OrchestratorClientError so
  the launch path surfaces failures rather than swallowing them.

pyright 0 errors; pylint 9.82/10; unit suite green (1742 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
2026-07-13 21:51:27 -04:00
didericis 77aaabae63 feat(orchestrator): slice 13c(ii) — shared-gateway source-IP allocator
lint / lint (push) Successful in 2m2s
test / unit (pull_request) Successful in 1m4s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m12s
Deterministic per-bottle address assignment on the consolidated docker
gateway's shared network — the address the gateway uses as its attribution
key. Pure ipaddress logic; the docker-specific inputs (subnet CIDR, the
gateway container's own address) are gathered by the caller and passed as
`taken`, keeping it testable without docker.

- next_free_ip(cidr, taken): lowest host address not reserved (network +
  broadcast excluded by hosts(); docker's router .1 excluded here) and not
  already assigned (gateway container + live bottles from the registry).
  NoFreeAddressError when the subnet is exhausted.

pyright 0 errors; pylint 9.83/10; unit suite green (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
2026-07-13 21:20:13 -04:00
didericis d0b35b5506 feat(git-gate+orchestrator): slice 13c(i) — per-bottle git-gate provisioning render
lint / lint (push) Successful in 2m9s
test / unit (pull_request) Successful in 1m8s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Successful in 1m14s
The consolidated gateway serves every bottle's repos under /git/<bottle_id>/
(slice 10), so each bottle's bare repos + per-repo credential config must be
provisioned into that namespace. This adds the renderer for that; placing
the creds + running it inside the gateway is the launch-wiring step.

- Extract the credential-wiring `init_repo` shell into a shared
  `_git_gate_init_repo_fn(repo_root, creds_dir)` — one source for the
  security-sensitive logic. The single-tenant daemon entrypoint is
  byte-unchanged (still /git + /git-gate/creds; existing tests green).
- git_gate_render_provision(bottle_id, upstreams): posix-sh that inits ONE
  bottle's bare repos under /git/<bottle_id>/ reading creds from
  /git-gate/creds/<bottle_id>/. Init-only (no `git daemon` — the shared
  gateway already serves). Isolating each bottle's repo root + creds dir by
  id is what keeps one bottle's push credentials out of another's repos.
- bottle_id is embedded unquoted in the script, so it's restricted to a
  shell/path-safe alphabet (registry ids are token_hex; defense in depth).

pyright 0 errors; pylint 9.83/10; unit suite green (1724 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
2026-07-13 21:15:21 -04:00
didericis ea0c070cfe feat(orchestrator): slice 13b — consolidated registration inputs (egress policy)
lint / lint (push) Successful in 2m10s
test / unit (pull_request) Successful in 1m12s
test / integration (pull_request) Successful in 29s
test / coverage (pull_request) Successful in 1m20s
Bridges the per-bottle `prepare` output to the consolidated registry:
`registration_inputs(plan)` turns a prepared egress plan into the
backend-neutral inputs `Orchestrator.launch_bottle` takes.

- egress_policy(plan): the policy blob = the exact routes YAML the
  per-bottle sidecar read from a file; the multi-tenant gateway's
  PolicyResolver now fetches it from the registry per request instead.
  Same egress_render_routes, so a bottle's allow-list is byte-identical
  moving onto the shared gateway.
- RegistrationInputs bundles policy + metadata; metadata carries the human
  slug so the shared registry can map a minted bottle id back to a name
  (console / supervise display).

Host-side glue (imports bot_bottle.egress) used by the launch path — not
the lean orchestrator control-plane process. Not yet wired into launch
(that's 13c/13d), consistent with the slice-6/10 pattern of landing the
primitive before the cut-over.

Key test: the policy round-trips through load_config back to the same
routes + log level — the resolver-served policy reconstructs the per-bottle
egress config exactly.

pyright 0 errors; pylint 9.83/10; unit suite green (1718 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
2026-07-13 20:40:38 -04:00
didericis 24df322c31 feat(orchestrator): slice 13a — orchestrator process lifecycle (idempotent singleton)
lint / lint (push) Successful in 2m9s
test / unit (pull_request) Successful in 1m4s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m14s
First sub-slice of docker launch integration: the foundation the CLI needs
to ensure exactly one orchestrator control plane + shared gateway is up
before registering/launching bottles. Does NOT touch the real launch path
yet — it's the lifecycle primitive the cut-over slices build on.

- OrchestratorProcess.ensure_running(): idempotent singleton — returns the
  control-plane URL if a healthy one already answers /health, else spawns
  `python -m bot_bottle.orchestrator --broker docker --gateway` detached
  (start_new_session, output tee'd to <root>/orchestrator.log) and polls
  /health until healthy or timeout (OrchestratorStartError).
- The control-plane port is the singleton key: a second orchestrator can't
  bind it, so a stray double-start fails fast rather than forking a rival.
- Host-process (not container) per the PRD's dev-harness sequencing — it
  already has the host user's docker access to broker launches; the
  data-plane gateway it manages is the container.

pyright 0 errors; pylint 9.83/10; unit suite green (1714 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
2026-07-13 20:33:14 -04:00
didericis 80d75de08a feat(supervise+orchestrator): slice 12 — multi-tenant git-gate hook + supervise server
lint / lint (push) Successful in 2m2s
test / unit (pull_request) Successful in 1m0s
test / integration (pull_request) Successful in 22s
test / coverage (pull_request) Successful in 1m11s
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
2026-07-13 20:04:26 -04:00
didericis 8827c64a83 feat(supervise+orchestrator): slice 11 — per-bottle supervise queue + DLP safelist
lint / lint (push) Successful in 2m4s
test / unit (pull_request) Successful in 1m6s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Successful in 1m15s
Consolidated egress ran every bottle through one process but keyed the
supervise proposal queue off a single SUPERVISE_BOTTLE_SLUG env and kept
one *global* DLP safelist — so in the shared gateway an operator's token
approval for bottle A would (a) be attributed to the wrong bottle and
(b) leak into bottle B's DLP scan (A's approved secret passes B's egress).
This slice keys both per bottle, resolved by source IP.

- policy_resolver: add `resolve_policy_and_bottle_id` — policy + bottle id
  in one `/resolve`, so egress keys routing *and* the supervise
  queue/safelist from a single round-trip. Fail-closed (403 -> (None,None)).
- egress_addon_core: add `resolve_client_context` (+ `ContextResolverLike`)
  returning `(Config, bottle_id)`, sharing the fail-closed parse with
  `resolve_client_config` via `_config_from_policy`.
- egress_addon: `_active_config` -> `_resolve_flow` returns `(Config, slug)`;
  `safe_tokens` set -> per-bottle `_safe_tokens_for(slug)`; the token-allow
  write/await/archive + the approved-token add all use the resolved slug.
  Single-tenant (no resolver) unchanged — slug = the env SUPERVISE_BOTTLE_SLUG.

New tests cover the resolver, the fail-closed context matrix, and the
cross-tenant isolation (an approval lands only in the calling bottle's
safelist; the proposal is keyed by the source-IP-attributed bottle;
unattributed IPs can't supervise).

Out of scope (noted): the git-gate gitleaks-allow hook + supervise_server
agent-proposal paths, and websocket DLP (still self.config-only, inert in
consolidated mode) — follow-up slices.

pyright 0 errors; pylint 9.83/10; unit suite green (1700 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
2026-07-13 19:48:29 -04:00
didericis 115a64c161 fix(git-gate): review — sandbox naming, ResolverLike Protocol, token-required note
test / unit (pull_request) Successful in 1m8s
test / integration (pull_request) Successful in 22s
test / coverage (pull_request) Successful in 1m20s
lint / lint (push) Successful in 2m7s
Addresses PR #365 review:
- Type `resolve_sandbox_root`'s resolver param as a `ResolverLike` Protocol
  (structural, `resolve_bottle_id` only) — fixes the pyright errors from
  passing a duck-typed fake resolver in tests; mirrors
  egress_addon_core.PolicyResolverLike.
- Rename `_project_root`/`project_root`/`resolve_repo_root`/
  `DEFAULT_PROJECT_ROOT` -> `_sandbox_root`/`sandbox_root`/
  `resolve_sandbox_root`/`DEFAULT_REPO_ROOT`: "sandbox" names the per-tenant
  scope; "project" was too amorphous. `GIT_PROJECT_ROOT` keeps git's own
  env-var name.
- Note the single-tenant path is transitional (stripped once every backend
  runs the consolidated gateway).
- policy_resolver: document that the optional identity token is
  transitional — the consolidated end state requires it (flips at the
  /resolve boundary once token *delivery* lands, a PRD 0070 open question).
- Split the long line in main() (was 105 cols).

pyright 0 errors; pylint 9.95/10; unit suite green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-13 18:47:53 -04:00
didericis 08aef30d87 feat(git-gate): source-IP-keyed multi-tenant repo namespace (PRD 0070)
lint / lint (push) Failing after 2m0s
test / unit (pull_request) Successful in 1m1s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m19s
First slice of git-gate consolidation: make the smart-HTTP git backend
serve every bottle from one process, selecting each request's repo root by
the calling bottle's source IP — the same attribution invariant + resolver
the multi-tenant egress addon uses. `git daemon` can't source-IP-route per
connection, so the consolidated gateway serves git-gate over this HTTP
backend (the transport firecracker/macOS already use); wiring the docker
path onto it lands with the launch-integration slice.

- policy_resolver: factor out `_post_resolve`; add `resolve_bottle_id`
  (source IP -> bottle id, same fail-closed 403->None contract as
  `resolve`) — the git-gate has no policy blob to parse, the bottle *is*
  the namespace.
- git_http_backend: `resolve_repo_root(resolver, base, source_ip, token)`
  — single-tenant passthrough when no resolver; else `<base>/<bottle_id>`,
  fail-closed on unattributed / resolver error / namespace escape.
  `BOT_BOTTLE_ORCHESTRATOR_URL` (same env as egress) flips the shared
  gateway multi-tenant; the identity header is read for attribution and
  never forwarded to the CGI. Per-repo creds + hooks scope by repo dir, so
  isolating the root per bottle isolates its creds too.

Single-tenant path unchanged (existing real-git-push tests green). New unit
coverage for the resolver + repo-root selection matrix. Full unit suite
green (1690 tests; the 13 test_sidecar_init /bin/sleep errors are the
pre-existing NixOS-local noise).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-13 18:31:03 -04:00
didericis e0b0429cd1 refactor(orchestrator): rename Sidecar -> Gateway for the consolidated data plane
test / unit (pull_request) Successful in 1m10s
test / integration (pull_request) Successful in 25s
test / coverage (pull_request) Successful in 1m17s
lint / lint (push) Successful in 2m11s
Retire "sidecar" for the consolidated per-host path (PRD 0070 naming
decision): the orchestrator is the umbrella/control plane, and the
egress/git/supervise data-plane unit it runs is the "gateway".

- git mv sidecar.py -> gateway.py and the two integration + one unit test
  files; DockerSidecar->DockerGateway, Sidecar->Gateway,
  SidecarError->GatewayError, SIDECAR_*->GATEWAY_*, ensure_sidecar->
  ensure_gateway, sidecar_status->gateway_status, container name
  bot-bottle-orch-sidecar->bot-bottle-orch-gateway.
- Prose rename across broker/registry/egress/policy_resolver + PRD 0070.
- Preserved: the image name bot-bottle-sidecars, the
  BOT_BOTTLE_SIDECAR_IMAGE env var, Dockerfile.sidecars, and PRD 0069's
  own stage-name cross-references (that doc still uses "sidecar").

No behavior change. Full unit suite green (1679 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
2026-07-13 18:12:17 -04:00
36 changed files with 1662 additions and 228 deletions
+47
View File
@@ -0,0 +1,47 @@
"""Shared-gateway source-IP allocation for the consolidated docker backend
(PRD 0070).
In the consolidated model one gateway container serves every bottle over a
single shared docker network, and each agent bottle attaches with a pinned,
deterministic address that the gateway uses as its **attribution key**. This
allocates those addresses from the network's subnet, skipping the reserved
ones — the network address and broadcast (excluded by `hosts()`), docker's
router `.1`, and everything already in use (`taken`: the gateway container
plus every live bottle, which the caller reads from the registry).
Pure `ipaddress` logic — the docker-specific bits (the subnet CIDR, the
gateway container's own address) are gathered by the caller and passed in, so
this stays testable without docker.
"""
from __future__ import annotations
import ipaddress
from collections.abc import Iterable
class NoFreeAddressError(RuntimeError):
"""The shared gateway network's subnet is exhausted — every host address
is reserved or already assigned to a bottle."""
def next_free_ip(cidr: str, taken: Iterable[str]) -> str:
"""The lowest host address in `cidr` not in `taken` and not docker's
router (`.1`). `taken` must include the gateway container's own address
and every live bottle's. Raises `NoFreeAddressError` if the subnet is
full."""
net = ipaddress.ip_network(cidr, strict=False)
reserved = {str(a) for a in taken}
# Docker assigns the network's first host (.1) to the bridge router; a
# bottle must never be handed that address.
reserved.add(str(net.network_address + 1))
for host in net.hosts(): # hosts() already excludes network + broadcast
candidate = str(host)
if candidate not in reserved:
return candidate
raise NoFreeAddressError(
f"no free address in {cidr} ({len(reserved)} reserved/assigned)"
)
__all__ = ["next_free_ip", "NoFreeAddressError"]
+54 -31
View File
@@ -1,4 +1,4 @@
"""mitmproxy addon entrypoint for the egress sidecar (PRD 0017, PRD 0053). """mitmproxy addon entrypoint for the egress gateway (PRD 0017, PRD 0053).
Loaded by `mitmdump -s /app/egress_addon.py` inside the Loaded by `mitmdump -s /app/egress_addon.py` inside the
egress container.""" egress container."""
@@ -32,7 +32,7 @@ from egress_addon_core import ( # type: ignore[import-not-found] # pylint: dis
is_git_push_request, is_git_push_request,
load_config, load_config,
match_route, match_route,
resolve_client_config, resolve_client_context,
outbound_scan_headers, outbound_scan_headers,
route_to_yaml_dict, route_to_yaml_dict,
scan_inbound, scan_inbound,
@@ -99,17 +99,29 @@ class EgressAddon:
# Absent → single-tenant (static routes file); behaviour unchanged. # Absent → single-tenant (static routes file); behaviour unchanged.
orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip() orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
self._resolver = PolicyResolver(orch_url) if orch_url else None self._resolver = PolicyResolver(orch_url) if orch_url else None
# Tokens the operator has approved this session (PRD 0062). In-memory # Tokens the operator has approved this session (PRD 0062), keyed by
# only — a restart re-prompts. Mutated only from the asyncio loop that # bottle so the shared gateway keeps each bottle's safelist separate —
# runs the addon hooks, so no lock is needed. # a global set would let bottle A's approved secret pass bottle B's DLP
self.safe_tokens: set[str] = set() # scan. In-memory only (a restart re-prompts); mutated only from the
# asyncio loop that runs the addon hooks, so no lock is needed.
self._safe_tokens: dict[str, set[str]] = {}
self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip() self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
self._token_allow_timeout = _token_allow_timeout_from_env(os.environ) self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
self._reload(initial=True) self._reload(initial=True)
self._install_sighup() self._install_sighup()
def _supervise_available(self) -> bool: @staticmethod
return bool(self._supervise_slug) def _supervise_available(slug: str) -> bool:
"""Supervise is reachable for this request iff we resolved a bottle to
attribute its proposals to (single-tenant env slug, or a source-IP
-attributed bottle id). Empty → fail closed (no queue to write to)."""
return bool(slug)
def _safe_tokens_for(self, slug: str) -> set[str]:
"""This bottle's operator-approved DLP safelist (PRD 0062), created on
first use. Keyed by bottle so the shared gateway never leaks one
bottle's approved token into another's scan."""
return self._safe_tokens.setdefault(slug, set())
def _reload(self, *, initial: bool = False) -> None: def _reload(self, *, initial: bool = False) -> None:
try: try:
@@ -218,19 +230,21 @@ class EgressAddon:
+ "\n" + "\n"
) )
def _active_config(self, flow: http.HTTPFlow) -> Config: def _resolve_flow(self, flow: http.HTTPFlow) -> "tuple[Config, str]":
"""The Config to apply to this request. Single-tenant → the static """The `(Config, supervise slug)` to apply to this request. Single-tenant
`self.config`. Consolidated → the calling bottle's Config, resolved → the static `self.config` and the env slug. Consolidated → the calling
by source IP (fail-closed to deny-all if unattributed). The identity bottle's Config and its bottle id, resolved by source IP in one
token, if the agent injected one, is read then stripped so it never round-trip (fail-closed to deny-all + empty slug if unattributed). The
leaks upstream.""" identity token, if the agent injected one, is read then stripped so it
never leaks upstream. The slug keys both the proposal queue and the
per-bottle safelist, so an approval only ever affects its own bottle."""
if self._resolver is None: if self._resolver is None:
return self.config return self.config, self._supervise_slug
conn = flow.client_conn conn = flow.client_conn
client_ip = conn.peername[0] if conn and conn.peername else "" client_ip = conn.peername[0] if conn and conn.peername else ""
token = flow.request.headers.get(IDENTITY_HEADER, "") token = flow.request.headers.get(IDENTITY_HEADER, "")
flow.request.headers.pop(IDENTITY_HEADER, None) flow.request.headers.pop(IDENTITY_HEADER, None)
return resolve_client_config(self._resolver, client_ip, token) return resolve_client_context(self._resolver, client_ip, token)
async def request(self, flow: http.HTTPFlow) -> None: async def request(self, flow: http.HTTPFlow) -> None:
request_path, _, query = flow.request.path.partition("?") request_path, _, query = flow.request.path.partition("?")
@@ -239,14 +253,14 @@ class EgressAddon:
self._serve_introspection(flow, request_path) self._serve_introspection(flow, request_path)
return return
config = self._active_config(flow) config, slug = self._resolve_flow(flow)
# DLP outbound scan BEFORE stripping auth — catches tokens the # DLP outbound scan BEFORE stripping auth — catches tokens the
# agent tried to smuggle in any header, path, query param, or body. # agent tried to smuggle in any header, path, query param, or body.
# Hostname is included to catch DNS-tunnelling exfiltration attempts. # Hostname is included to catch DNS-tunnelling exfiltration attempts.
route = match_route(config.routes, flow.request.pretty_host) route = match_route(config.routes, flow.request.pretty_host)
if route is not None: if route is not None:
if not await self._handle_outbound_dlp(flow, route): if not await self._handle_outbound_dlp(flow, route, slug):
return return
# The redact policy may have rewritten the request line; recompute # The redact policy may have rewritten the request line; recompute
# the path/query the git checks below rely on. # the path/query the git checks below rely on.
@@ -275,7 +289,7 @@ class EgressAddon:
return return
# Strip agent-set Authorization after DLP scan so smuggled tokens # Strip agent-set Authorization after DLP scan so smuggled tokens
# are caught above; the route may inject sidecar-owned auth below. # are caught above; the route may inject gateway-owned auth below.
flow.request.headers.pop("authorization", None) flow.request.headers.pop("authorization", None)
# Build headers mapping for match evaluation # Build headers mapping for match evaluation
@@ -310,6 +324,7 @@ class EgressAddon:
self, self,
flow: http.HTTPFlow, flow: http.HTTPFlow,
route: Route, route: Route,
slug: str,
) -> bool: ) -> bool:
"""Scan the outbound request and apply the route's on-match policy """Scan the outbound request and apply the route's on-match policy
(PRD 0062). Returns True if the request may be forwarded, False if a (PRD 0062). Returns True if the request may be forwarded, False if a
@@ -331,7 +346,7 @@ class EgressAddon:
) )
result = scan_outbound( result = scan_outbound(
route, scan_text, os.environ, route, scan_text, os.environ,
safe_tokens=self.safe_tokens, crlf_text=crlf_text, safe_tokens=self._safe_tokens_for(slug), crlf_text=crlf_text,
) )
if result is None or result.severity != "block": if result is None or result.severity != "block":
return True return True
@@ -366,10 +381,10 @@ class EgressAddon:
# supervise (default): hold the request for operator approval. # supervise (default): hold the request for operator approval.
# Fall back to a hard 403 when supervise isn't wired for the bottle. # Fall back to a hard 403 when supervise isn't wired for the bottle.
if not self._supervise_available(): if not self._supervise_available(slug):
self._block_dlp(flow, result) self._block_dlp(flow, result)
return False return False
approved = await self._supervise_token_block(flow, request_path, result) approved = await self._supervise_token_block(flow, request_path, result, slug)
if not approved: if not approved:
return False # _supervise_token_block wrote the 403 response return False # _supervise_token_block wrote the 403 response
# loop: the approved value is now in safe_tokens; re-scan. # loop: the approved value is now in safe_tokens; re-scan.
@@ -412,12 +427,16 @@ class EgressAddon:
flow: http.HTTPFlow, flow: http.HTTPFlow,
request_path: str, request_path: str,
result: ScanResult, result: ScanResult,
slug: str,
) -> bool: ) -> bool:
"""Route a token DLP block to the operator's supervisor queue and wait. """Route a token DLP block to the operator's supervisor queue and wait.
Returns True if the operator approved (the matched value is added to `slug` attributes the proposal to the calling bottle (its own queue +
`self.safe_tokens` and the caller re-scans); False if the request must safelist) — in the shared gateway this is what keeps one bottle's
be blocked (a 403 response has been written to `flow`).""" approval from unblocking another's request. Returns True if the operator
approved (the matched value is added to that bottle's safelist and the
caller re-scans); False if the request must be blocked (a 403 response
has been written to `flow`)."""
host = flow.request.pretty_host host = flow.request.pretty_host
payload = build_token_allow_payload( payload = build_token_allow_payload(
redact_tokens(host, env=os.environ), redact_tokens(host, env=os.environ),
@@ -426,7 +445,7 @@ class EgressAddon:
result, result,
) )
proposal = _sv.Proposal.new( proposal = _sv.Proposal.new(
bottle_slug=self._supervise_slug, bottle_slug=slug,
tool=_sv.TOOL_EGRESS_TOKEN_ALLOW, tool=_sv.TOOL_EGRESS_TOKEN_ALLOW,
proposed_file=payload, proposed_file=payload,
justification=_TOKEN_ALLOW_JUSTIFICATION, justification=_TOKEN_ALLOW_JUSTIFICATION,
@@ -449,13 +468,13 @@ class EgressAddon:
**self._req_ctx(flow), **self._req_ctx(flow),
}) + "\n") }) + "\n")
response = await self._await_token_response(proposal.id) response = await self._await_token_response(proposal.id, slug)
_sv.archive_proposal(self._supervise_slug, proposal.id) _sv.archive_proposal(slug, proposal.id)
if response is not None and response.status in ( if response is not None and response.status in (
_sv.STATUS_APPROVED, _sv.STATUS_MODIFIED, _sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
): ):
self.safe_tokens.add(result.matched) self._safe_tokens_for(slug).add(result.matched)
if self.config.log >= LOG_BLOCKS: if self.config.log >= LOG_BLOCKS:
sys.stderr.write(json.dumps({ sys.stderr.write(json.dumps({
"event": "egress_token_allowed", "event": "egress_token_allowed",
@@ -478,6 +497,7 @@ class EgressAddon:
async def _await_token_response( async def _await_token_response(
self, self,
proposal_id: str, proposal_id: str,
slug: str,
) -> "_sv.Response | None": ) -> "_sv.Response | None":
"""Poll the DB for the operator's response without blocking the """Poll the DB for the operator's response without blocking the
proxy event loop. Returns the Response, or None on timeout.""" proxy event loop. Returns the Response, or None on timeout."""
@@ -485,7 +505,7 @@ class EgressAddon:
deadline = loop.time() + self._token_allow_timeout deadline = loop.time() + self._token_allow_timeout
while True: while True:
try: try:
return _sv.read_response(self._supervise_slug, proposal_id) return _sv.read_response(slug, proposal_id)
except (OSError, ValueError, KeyError): except (OSError, ValueError, KeyError):
# Not written yet, or a partial/malformed write — retry until # Not written yet, or a partial/malformed write — retry until
# the deadline, then fail closed. # the deadline, then fail closed.
@@ -539,6 +559,9 @@ class EgressAddon:
""" """
if flow.websocket is None: # type: ignore[union-attr] if flow.websocket is None: # type: ignore[union-attr]
return return
# WebSocket DLP runs against the static config only (single-tenant); in
# the consolidated gateway self.config has no routes, so this is inert
# until websocket routing is made source-IP-aware (a separate slice).
route = match_route(self.config.routes, flow.request.pretty_host) route = match_route(self.config.routes, flow.request.pretty_host)
if route is None: if route is None:
return return
@@ -549,7 +572,7 @@ class EgressAddon:
# not an injection vector here — scan only for credential leakage. # not an injection vector here — scan only for credential leakage.
result = scan_outbound( result = scan_outbound(
route, content, os.environ, route, content, os.environ,
safe_tokens=self.safe_tokens, crlf_text="", safe_tokens=self._safe_tokens_for(self._supervise_slug), crlf_text="",
) )
if result is not None and result.severity == "block": if result is not None and result.severity == "block":
sys.stderr.write(f"egress DLP: {result.reason}\n") sys.stderr.write(f"egress DLP: {result.reason}\n")
+45 -10
View File
@@ -3,7 +3,7 @@
Split out of `egress_addon.py` so the host's unit tests can Split out of `egress_addon.py` so the host's unit tests can
exercise the parse + decision functions without depending on the exercise the parse + decision functions without depending on the
`mitmproxy` package. The companion module wraps these with the `mitmproxy` package. The companion module wraps these with the
`mitmproxy.http.HTTPFlow` API and is loaded inside the sidecar `mitmproxy.http.HTTPFlow` API and is loaded inside the gateway
container. container.
Imports: stdlib + `yaml_subset` (which is itself stdlib-only and Imports: stdlib + `yaml_subset` (which is itself stdlib-only and
@@ -22,7 +22,7 @@ except ImportError: # pragma: no cover - host-side path
from .yaml_subset import YamlSubsetError, parse_yaml_subset from .yaml_subset import YamlSubsetError, parse_yaml_subset
# DLP detector-config parsing lives in a sibling module (also flat-bundled # DLP detector-config parsing lives in a sibling module (also flat-bundled
# into the sidecar — see Dockerfile.sidecars). Re-exported below so existing # into the gateway — see Dockerfile.sidecars). Re-exported below so existing
# `from egress_addon_core import ON_MATCH_*` callers keep working. # `from egress_addon_core import ON_MATCH_*` callers keep working.
try: try:
from egress_dlp_config import ( # type: ignore[import-not-found] from egress_dlp_config import ( # type: ignore[import-not-found]
@@ -120,7 +120,7 @@ class ScanResult:
reason: str reason: str
location: str = "" # where the match was found, e.g. "body", "authorization header" location: str = "" # where the match was found, e.g. "body", "authorization header"
context: str = "" # surrounding text with the match replaced by REDACT context: str = "" # surrounding text with the match replaced by REDACT
# Raw substring the detector matched. Used inside the sidecar to key the # Raw substring the detector matched. Used inside the gateway to key the
# supervisor-approved "safe tokens" set (PRD 0062); never logged or written # supervisor-approved "safe tokens" set (PRD 0062); never logged or written
# to a proposal file. Empty for structural detectors (CRLF) that carry no # to a proposal file. Empty for structural detectors (CRLF) that carry no
# safelist-able value. # safelist-able value.
@@ -421,6 +421,18 @@ class PolicyResolverLike(typing.Protocol):
... ...
def _config_from_policy(policy: "str | None") -> "Config":
"""Parse a resolved policy blob into a Config, fail-closed: None / empty /
unparseable all become a deny-all Config (no routes → every request
blocked)."""
if not policy:
return Config(routes=()) # unattributed or empty → deny-all
try:
return load_config(policy)
except ValueError:
return Config(routes=()) # unparseable policy → deny
def resolve_client_config( def resolve_client_config(
resolver: PolicyResolverLike, client_ip: str, identity_token: str = "" resolver: PolicyResolverLike, client_ip: str, identity_token: str = ""
) -> "Config": ) -> "Config":
@@ -433,12 +445,33 @@ def resolve_client_config(
policy = resolver.resolve(client_ip, identity_token) policy = resolver.resolve(client_ip, identity_token)
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return Config(routes=()) # orchestrator unreachable/errored → deny return Config(routes=()) # orchestrator unreachable/errored → deny
if not policy: return _config_from_policy(policy)
return Config(routes=()) # unattributed or empty → deny-all
class ContextResolverLike(typing.Protocol):
"""The bit of `policy_resolver.PolicyResolver` `resolve_client_context`
needs — one round-trip returning both policy and bottle id."""
def resolve_policy_and_bottle_id(
self, source_ip: str, identity_token: str = ...,
) -> "tuple[str | None, str | None]":
...
def resolve_client_context(
resolver: ContextResolverLike, client_ip: str, identity_token: str = "",
) -> "tuple[Config, str]":
"""The calling client's `(Config, bottle_id)` in one round-trip —
**fail-closed**. The Config follows `resolve_client_config`'s deny-all
rules; the bottle id is `""` whenever unattributed or the orchestrator
errored, which the caller treats as "supervise unavailable for this
bottle" (never another bottle's queue). One `/resolve` keys both the
per-request egress policy and the per-bottle supervise queue + safelist."""
try: try:
return load_config(policy) policy, bottle_id = resolver.resolve_policy_and_bottle_id(client_ip, identity_token)
except ValueError: except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return Config(routes=()) # unparseable policy → deny return Config(routes=()), "" # orchestrator unreachable/errored → deny
return _config_from_policy(policy), (bottle_id or "")
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -641,7 +674,7 @@ def outbound_scan_headers(
) -> dict[str, str]: ) -> dict[str, str]:
"""Return request headers that should be included in outbound DLP. """Return request headers that should be included in outbound DLP.
Routes that inject sidecar-owned auth always strip the agent's Routes that inject gateway-owned auth always strip the agent's
Authorization header before forwarding. Scanning that header first Authorization header before forwarding. Scanning that header first
creates false positives for provider clients that insist on sending creates false positives for provider clients that insist on sending
their own bearer-shaped placeholder, while still not changing what their own bearer-shaped placeholder, while still not changing what
@@ -692,7 +725,7 @@ def scan_outbound(
crlf_text: str | None = None, crlf_text: str | None = None,
) -> ScanResult | None: ) -> ScanResult | None:
# Lazy import to avoid circular deps and keep dlp_detectors optional # Lazy import to avoid circular deps and keep dlp_detectors optional
# at import time (the sidecar copies it flat alongside this file). # at import time (the gateway copies it flat alongside this file).
try: try:
from dlp_detectors import ( # type: ignore[import-not-found] from dlp_detectors import ( # type: ignore[import-not-found]
scan_crlf_injection, scan_crlf_injection,
@@ -833,7 +866,9 @@ __all__ = [
"is_git_fetch_request", "is_git_fetch_request",
"load_config", "load_config",
"resolve_client_config", "resolve_client_config",
"resolve_client_context",
"PolicyResolverLike", "PolicyResolverLike",
"ContextResolverLike",
"match_route", "match_route",
"outbound_scan_headers", "outbound_scan_headers",
"parse_config", "parse_config",
+2
View File
@@ -46,6 +46,7 @@ from .git_gate_render import (
git_gate_known_hosts_line, git_gate_known_hosts_line,
git_gate_render_access_hook, git_gate_render_access_hook,
git_gate_render_entrypoint, git_gate_render_entrypoint,
git_gate_render_provision,
git_gate_render_gitconfig, git_gate_render_gitconfig,
git_gate_render_hook, git_gate_render_hook,
git_gate_upstreams_for_bottle, git_gate_upstreams_for_bottle,
@@ -155,6 +156,7 @@ __all__ = [
"git_gate_render_gitconfig", "git_gate_render_gitconfig",
"git_gate_known_hosts_line", "git_gate_known_hosts_line",
"git_gate_render_entrypoint", "git_gate_render_entrypoint",
"git_gate_render_provision",
"git_gate_render_hook", "git_gate_render_hook",
"git_gate_render_access_hook", "git_gate_render_access_hook",
"provision_git_gate_dynamic_keys", "provision_git_gate_dynamic_keys",
+57 -21
View File
@@ -9,6 +9,7 @@ own; `git_gate` re-exports these names for API stability."""
from __future__ import annotations from __future__ import annotations
import re
import shlex import shlex
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -125,22 +126,18 @@ def git_gate_known_hosts_line(host: str, port: str, key: str) -> str:
return f"{target} {key}\n" return f"{target} {key}\n"
def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str: def _git_gate_init_repo_fn(repo_root: str, creds_dir: str) -> list[str]:
"""Posix-sh entrypoint. One `init_repo` call per upstream, then """The `init_repo` shell function, parameterized by the bare-repo root
`exec git daemon`. The function reads and the per-bottle creds dir. Single source of the credential-wiring
`/git-gate/creds/<name>-{key,known_hosts}` (bind-mounted into logic, shared by the single-tenant daemon entrypoint (`/git`,
the bundle by the renderer) and wires them into each bare repo's `/git-gate/creds`) and the consolidated per-bottle provisioning
config; the access-hook + pre-receive hook pick those paths up (`/git/<bottle_id>`, `/git-gate/creds/<bottle_id>`)."""
at fetch / push time.""" return [
lines = [
"#!/bin/sh",
"set -eu",
"",
"init_repo() {", "init_repo() {",
" name=$1", " name=$1",
" upstream_url=$2", " upstream_url=$2",
" keyfile=/git-gate/creds/${name}-key", f" keyfile={creds_dir}/${{name}}-key",
" hostsfile=/git-gate/creds/${name}-known_hosts", f" hostsfile={creds_dir}/${{name}}-known_hosts",
"", "",
# `|| true`: PRD 0018 chunk 3+ bind-mounts these RO from the # `|| true`: PRD 0018 chunk 3+ bind-mounts these RO from the
# host, so chmod-syscalls fail with EROFS. The files already # host, so chmod-syscalls fail with EROFS. The files already
@@ -153,14 +150,14 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
" chmod 600 \"$hostsfile\" 2>/dev/null || true", " chmod 600 \"$hostsfile\" 2>/dev/null || true",
" fi", " fi",
"", "",
" repo=/git/${name}.git", f" repo={repo_root}/${{name}}.git",
" if [ ! -d \"$repo\" ]; then", " if [ ! -d \"$repo\" ]; then",
" git init --bare \"$repo\" >/dev/null", " git init --bare \"$repo\" >/dev/null",
# --mirror=fetch sets remote.origin.fetch = +refs/*:refs/* so", # --mirror=fetch sets remote.origin.fetch = +refs/*:refs/* so a later
# a later `git fetch origin` mirrors the upstream's full ref", # `git fetch origin` mirrors the upstream's full ref graph (heads,
# graph (heads, tags, notes) into the bare repo at canonical", # tags, notes) into the bare repo at canonical paths. It does NOT set
# paths. It does NOT set remote.origin.mirror=true, so an", # remote.origin.mirror=true, so an explicit `git push origin
# explicit `git push origin <ref>:<ref>` still pushes one ref.", # <ref>:<ref>` still pushes one ref.
" git -C \"$repo\" remote add --mirror=fetch origin \"$upstream_url\"", " git -C \"$repo\" remote add --mirror=fetch origin \"$upstream_url\"",
" fi", " fi",
" git -C \"$repo\" config git-gate.identityFile \"$keyfile\"", " git -C \"$repo\" config git-gate.identityFile \"$keyfile\"",
@@ -170,9 +167,19 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
" git -C \"$repo\" config http.receivepack true", " git -C \"$repo\" config http.receivepack true",
" install -m 755 /etc/git-gate/pre-receive \"$repo/hooks/pre-receive\"", " install -m 755 /etc/git-gate/pre-receive \"$repo/hooks/pre-receive\"",
"}", "}",
"",
"mkdir -p /git",
] ]
def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
"""Posix-sh entrypoint. One `init_repo` call per upstream, then
`exec git daemon`. The function reads
`/git-gate/creds/<name>-{key,known_hosts}` (bind-mounted into
the bundle by the renderer) and wires them into each bare repo's
config; the access-hook + pre-receive hook pick those paths up
at fetch / push time."""
lines = ["#!/bin/sh", "set -eu", ""]
lines += _git_gate_init_repo_fn("/git", "/git-gate/creds")
lines += ["", "mkdir -p /git"]
for u in upstreams: for u in upstreams:
lines.append(f"init_repo {shlex.quote(u.name)} {shlex.quote(u.upstream_url)}") lines.append(f"init_repo {shlex.quote(u.name)} {shlex.quote(u.upstream_url)}")
lines.extend([ lines.extend([
@@ -190,6 +197,35 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
return "\n".join(lines) + "\n" return "\n".join(lines) + "\n"
# A bottle id namespaces the consolidated gateway's repo + creds dirs; it is
# embedded unquoted in the provisioning script, so restrict it to a shell- and
# path-safe alphabet (registry ids are token_hex — this is defense in depth).
_SAFE_BOTTLE_ID = re.compile(r"[A-Za-z0-9_-]+")
def git_gate_render_provision(
bottle_id: str, upstreams: tuple[GitGateUpstream, ...],
) -> str:
"""Posix-sh script that provisions ONE bottle's bare repos into the
consolidated gateway (PRD 0070), under `/git/<bottle_id>/` with creds
read from `/git-gate/creds/<bottle_id>/`. Init-only no `git daemon`,
since the shared gateway already serves every bottle; run inside the
running gateway when the bottle is registered.
Isolating each bottle's repo root and creds dir by id is what keeps one
bottle's push credentials out of another's repos on the shared gateway."""
if not _SAFE_BOTTLE_ID.fullmatch(bottle_id):
raise ValueError(f"git-gate: unsafe bottle id {bottle_id!r}")
repo_root = f"/git/{bottle_id}"
creds_dir = f"/git-gate/creds/{bottle_id}"
lines = ["#!/bin/sh", "set -eu", ""]
lines += _git_gate_init_repo_fn(repo_root, creds_dir)
lines += ["", f"mkdir -p {shlex.quote(repo_root)}"]
for u in upstreams:
lines.append(f"init_repo {shlex.quote(u.name)} {shlex.quote(u.upstream_url)}")
return "\n".join(lines) + "\n"
def git_gate_render_hook() -> str: def git_gate_render_hook() -> str:
"""The shared pre-receive hook: gitleaks-scan all incoming refs, """The shared pre-receive hook: gitleaks-scan all incoming refs,
then forward each accepted ref to the real upstream (`origin`) then forward each accepted ref to the real upstream (`origin`)
+116 -5
View File
@@ -6,6 +6,15 @@ where the guest reaches the sidecar over the point-to-point TAP). The
wrapper serves the same `/git/*.git` bare repos through wrapper serves the same `/git/*.git` bare repos through
`git http-backend`, so pre-receive and upstream forwarding remain the `git http-backend`, so pre-receive and upstream forwarding remain the
git-gate enforcement point. git-gate enforcement point.
Consolidated (PRD 0070): when `BOT_BOTTLE_ORCHESTRATOR_URL` is set, one
shared gateway serves every bottle, and each request is served from the
calling bottle's repo namespace (`<root>/<bottle_id>`), attributed from
the unspoofable source IP via the orchestrator. Per-repo credentials +
hooks scope by repo directory, so isolating the *root* per bottle isolates
its creds too. Unattributed clients fail closed (404). Unset the legacy
per-bottle single-tenant flat root, unchanged a transitional path that
gets stripped out once every backend runs the consolidated gateway.
""" """
from __future__ import annotations from __future__ import annotations
@@ -13,13 +22,86 @@ from __future__ import annotations
import os import os
import subprocess import subprocess
import sys import sys
import typing
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path from pathlib import Path
from urllib.parse import urlsplit from urllib.parse import urlsplit
# policy_resolver ships flat alongside this file in the sidecar bundle
# image (see Dockerfile.sidecars); the bot_bottle.* fallback is the
# host-side / test path. Mirrors egress_addon's import shape.
try:
from policy_resolver import ( # type: ignore[import-not-found]
PolicyResolveError,
PolicyResolver,
)
except ImportError: # pragma: no cover - host-side path
from bot_bottle.policy_resolver import PolicyResolveError, PolicyResolver
DEFAULT_PORT = 9420 DEFAULT_PORT = 9420
# Consolidated (multi-tenant) mode: when this points at the per-host
# orchestrator's control plane, the backend serves each request from the
# *calling* bottle's repo namespace, selected by source IP, instead of a
# single flat repo root. Unset → legacy per-bottle single-tenant mode
# (unchanged). Same env the egress addon reads, so one orchestrator setting
# flips the whole shared gateway multi-tenant.
ORCHESTRATOR_URL_ENV = "BOT_BOTTLE_ORCHESTRATOR_URL"
# App-layer identity token (defense-in-depth over the source-IP invariant);
# the agent injects it, the backend reads it for attribution and never
# forwards it to `git http-backend`. Mirrors egress_addon.IDENTITY_HEADER
# (duplicated, not imported: egress_addon pulls in mitmproxy).
IDENTITY_HEADER = "x-bot-bottle-identity"
# Default flat repo root (single-tenant, and the base under which
# consolidated mode nests each sandbox's namespace).
DEFAULT_REPO_ROOT = "/git"
class ResolverLike(typing.Protocol):
"""Structural type for the resolver `resolve_sandbox_root` needs — just
`resolve_bottle_id`. A Protocol so tests can pass a fake without
importing PolicyResolver (mirrors egress_addon_core.PolicyResolverLike)."""
def resolve_bottle_id(
self, source_ip: str, identity_token: str = ...,
) -> "str | None": ...
def resolve_sandbox_root(
resolver: "ResolverLike | None",
base_root: Path,
source_ip: str,
identity_token: str = "",
) -> Path | None:
"""The per-sandbox repo root to serve this request from, or None to
deny (404).
Single-tenant (`resolver is None`): the flat `base_root`, unchanged.
NOTE: this legacy per-bottle single-tenant path is transitional it
will be stripped out once every backend runs the consolidated gateway
(PRD 0070), leaving only the source-IP-attributed path below.
Consolidated: `base_root/<bottle_id>`, where the sandbox is attributed
from the source IP via the orchestrator. Fail-closed an unattributed
client, a resolver error, or a namespace that would escape `base_root`
all deny, so one sandbox can never reach another's repos."""
if resolver is None:
return base_root
try:
bottle_id = resolver.resolve_bottle_id(source_ip, identity_token)
except PolicyResolveError:
return None # orchestrator unreachable/errored → deny
if not bottle_id:
return None # unattributed → deny
base = base_root.resolve()
namespace = (base / bottle_id).resolve()
if base not in namespace.parents:
return None # bottle_id tried to escape the root → deny
return namespace
# Mirrors git_gate_render.GIT_GATE_TIMEOUT_SECS. Duplicated rather than # Mirrors git_gate_render.GIT_GATE_TIMEOUT_SECS. Duplicated rather than
# imported: this module ships as a flat top-level sibling in the sidecar # imported: this module ships as a flat top-level sibling in the sidecar
# bundle image (see Dockerfile.sidecars), not as part of the bot_bottle # bundle image (see Dockerfile.sidecars), not as part of the bot_bottle
@@ -40,10 +122,25 @@ class GitHttpHandler(BaseHTTPRequestHandler):
def do_POST(self) -> None: def do_POST(self) -> None:
self._run_backend() self._run_backend()
def _sandbox_root(self) -> Path | None:
"""This request's per-sandbox repo root, or None to deny. Single-tenant
unless the server was started with a resolver (consolidated mode), in
which case the root is the calling sandbox's source-IP-selected
namespace. `GIT_PROJECT_ROOT` keeps git's own env-var name."""
base = Path(os.environ.get("GIT_PROJECT_ROOT", DEFAULT_REPO_ROOT))
resolver = getattr(self.server, "policy_resolver", None)
token = self.headers.get(IDENTITY_HEADER, "")
return resolve_sandbox_root(resolver, base, self.client_address[0], token)
def _run_backend(self) -> None: def _run_backend(self) -> None:
sandbox_root = self._sandbox_root()
if sandbox_root is None:
# Unattributed / resolver error: deny before touching any repo.
self.send_error(404)
return
parsed = urlsplit(self.path) parsed = urlsplit(self.path)
if self._is_upload_pack(parsed.path, parsed.query): if self._is_upload_pack(parsed.path, parsed.query):
repo_dir = self._repo_dir(parsed.path) repo_dir = self._repo_dir(sandbox_root, parsed.path)
if repo_dir is None: if repo_dir is None:
self.send_error(404) self.send_error(404)
return return
@@ -77,7 +174,7 @@ class GitHttpHandler(BaseHTTPRequestHandler):
return return
env = os.environ.copy() env = os.environ.copy()
env.update({ env.update({
"GIT_PROJECT_ROOT": os.environ.get("GIT_PROJECT_ROOT", "/git"), "GIT_PROJECT_ROOT": str(sandbox_root),
"GIT_HTTP_EXPORT_ALL": "1", "GIT_HTTP_EXPORT_ALL": "1",
"REQUEST_METHOD": self.command, "REQUEST_METHOD": self.command,
"PATH_INFO": parsed.path, "PATH_INFO": parsed.path,
@@ -91,6 +188,14 @@ class GitHttpHandler(BaseHTTPRequestHandler):
"SERVER_PORT": str(self.server.server_port), # type: ignore "SERVER_PORT": str(self.server.server_port), # type: ignore
"SERVER_PROTOCOL": self.request_version, "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 ( for header, variable in (
("accept", "HTTP_ACCEPT"), ("accept", "HTTP_ACCEPT"),
("content-encoding", "HTTP_CONTENT_ENCODING"), ("content-encoding", "HTTP_CONTENT_ENCODING"),
@@ -123,8 +228,8 @@ class GitHttpHandler(BaseHTTPRequestHandler):
) )
self._write_cgi_response(proc.stdout) self._write_cgi_response(proc.stdout)
def _repo_dir(self, path: str) -> Path | None: def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None:
root = Path(os.environ.get("GIT_PROJECT_ROOT", "/git")).resolve() root = sandbox_root.resolve()
relative = path.lstrip("/").split(".git", 1)[0] + ".git" relative = path.lstrip("/").split(".git", 1)[0] + ".git"
candidate = (root / relative).resolve() candidate = (root / relative).resolve()
if root not in (candidate, *candidate.parents): if root not in (candidate, *candidate.parents):
@@ -181,7 +286,13 @@ class GitHttpHandler(BaseHTTPRequestHandler):
def main() -> int: def main() -> int:
port = int(os.environ.get("GIT_HTTP_PORT", str(DEFAULT_PORT))) port = int(os.environ.get("GIT_HTTP_PORT", str(DEFAULT_PORT)))
server = ThreadingHTTPServer(("0.0.0.0", port), GitHttpHandler) server = ThreadingHTTPServer(("0.0.0.0", port), GitHttpHandler)
sys.stdout.write(f"git-http listening on 0.0.0.0:{port}\n") orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
# Consolidated mode: resolve each request's sandbox namespace by source
# IP. Absent → single-tenant (flat repo root); behaviour unchanged.
resolver = PolicyResolver(orch_url) if orch_url else None
server.policy_resolver = resolver # type: ignore[attr-defined]
mode = "multi-tenant" if orch_url else "single-tenant"
sys.stdout.write(f"git-http listening on 0.0.0.0:{port} ({mode})\n")
sys.stdout.flush() sys.stdout.flush()
server.serve_forever() server.serve_forever()
return 0 return 0
+9 -9
View File
@@ -1,6 +1,6 @@
"""Per-host orchestrator service (PRD 0070). """Per-host orchestrator service (PRD 0070).
A single persistent per-host service that will run the sidecar functions A single persistent per-host service that will run the gateway functions
(egress / git-gate / supervise), coordinate with the console, and broker (egress / git-gate / supervise), coordinate with the console, and broker
agent launches. This package is being built bottom-up, starting with the agent launches. This package is being built bottom-up, starting with the
backend-neutral "consolidation core" that needs no VM packaging: backend-neutral "consolidation core" that needs no VM packaging:
@@ -10,15 +10,15 @@ backend-neutral "consolidation core" that needs no VM packaging:
* `broker` the signed, structured launch-request contract + a * `broker` the signed, structured launch-request contract + a
`LaunchBroker` (stub for the harness) that verifies `LaunchBroker` (stub for the harness) that verifies
provenance before acting. provenance before acting.
* `sidecar` the consolidated per-host sidecar: a `Sidecar` * `gateway` the consolidated per-host gateway: a `Gateway`
lifecycle contract (idempotent singleton) + a lifecycle contract (idempotent singleton) + a
`DockerSidecar` impl. One sidecar shared by all `DockerGateway` impl. One gateway shared by all
bottles instead of one per bottle. bottles instead of one per bottle.
* `service` the `Orchestrator`: owns the registry, brokers the * `service` the `Orchestrator`: owns the registry, brokers the
launch lifecycle (launch/teardown), manages the launch lifecycle (launch/teardown), manages the
shared sidecar, attributes. shared gateway, attributes.
* `control_plane` the HTTP control-plane RPC (launch / teardown / * `control_plane` the HTTP control-plane RPC (launch / teardown /
list / attribute / sidecar / health). list / attribute / gateway / health).
The actual backend-native launch (a real docker/firecracker broker) and The actual backend-native launch (a real docker/firecracker broker) and
the egress/git/supervise data plane land in later slices once this core is the egress/git/supervise data plane land in later slices once this core is
@@ -38,7 +38,7 @@ from .broker import (
verify_request, verify_request,
) )
from .docker_broker import DockerBroker, DockerBrokerError from .docker_broker import DockerBroker, DockerBrokerError
from .sidecar import DockerSidecar, Sidecar, SidecarError from .gateway import DockerGateway, Gateway, GatewayError
from .service import Orchestrator from .service import Orchestrator
from .control_plane import ControlPlaneServer, dispatch, make_server from .control_plane import ControlPlaneServer, dispatch, make_server
@@ -52,9 +52,9 @@ __all__ = [
"StubBroker", "StubBroker",
"DockerBroker", "DockerBroker",
"DockerBrokerError", "DockerBrokerError",
"Sidecar", "Gateway",
"DockerSidecar", "DockerGateway",
"SidecarError", "GatewayError",
"sign_request", "sign_request",
"verify_request", "verify_request",
"Orchestrator", "Orchestrator",
+8 -8
View File
@@ -21,7 +21,7 @@ from .control_plane import make_server
from .docker_broker import DockerBroker from .docker_broker import DockerBroker
from .registry import RegistryStore, default_db_path from .registry import RegistryStore, default_db_path
from .service import Orchestrator from .service import Orchestrator
from .sidecar import DockerSidecar, Sidecar from .gateway import DockerGateway, Gateway
def main(argv: list[str] | None = None) -> int: def main(argv: list[str] | None = None) -> int:
@@ -38,7 +38,7 @@ def main(argv: list[str] | None = None) -> int:
help="launch broker: 'stub' records requests; 'docker' runs containers", help="launch broker: 'stub' records requests; 'docker' runs containers",
) )
parser.add_argument( parser.add_argument(
"--sidecar", action="store_true", "--gateway", action="store_true",
help="run one consolidated per-host sidecar bundle (build-if-missing)", help="run one consolidated per-host sidecar bundle (build-if-missing)",
) )
args = parser.parse_args(argv) args = parser.parse_args(argv)
@@ -51,14 +51,14 @@ def main(argv: list[str] | None = None) -> int:
# anything; 'docker' runs real containers (firecracker drops in later). # anything; 'docker' runs real containers (firecracker drops in later).
secret = secrets.token_bytes(32) secret = secrets.token_bytes(32)
broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret) broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret)
sidecar: Sidecar | None = DockerSidecar() if args.sidecar else None gateway: Gateway | None = DockerGateway() if args.gateway else None
orchestrator = Orchestrator(registry, broker, secret, sidecar) orchestrator = Orchestrator(registry, broker, secret, gateway)
# One persistent per-host sidecar, shared by every bottle: build the # One persistent per-host gateway, shared by every bottle: build the
# bundle image if missing, then bring the singleton up (idempotent). # bundle image if missing, then bring the singleton up (idempotent).
if sidecar is not None: if gateway is not None:
orchestrator.ensure_sidecar() orchestrator.ensure_gateway()
log.info("consolidated sidecar ensured", context={"name": sidecar.name}) log.info("consolidated gateway ensured", context={"name": gateway.name})
server = make_server(orchestrator, host=args.host, port=args.port) server = make_server(orchestrator, host=args.host, port=args.port)
bound_host, bound_port = server.server_address[0], server.server_address[1] bound_host, bound_port = server.server_address[0], server.server_address[1]
+1 -1
View File
@@ -9,7 +9,7 @@ The request it sends is:
request can't be coerced into launching an arbitrary payload; and request can't be coerced into launching an arbitrary payload; and
* **signed** wrapped as a compact JWS/JWT the broker verifies before * **signed** wrapped as a compact JWS/JWT the broker verifies before
acting, so a *compromised co-located component* (an agent-facing acting, so a *compromised co-located component* (an agent-facing
sidecar, say) can't forge a launch. This is the concrete form of the gateway, say) can't forge a launch. This is the concrete form of the
"structured requests only" rule in the PRD security review. "structured requests only" rule in the PRD security review.
Signing is **HS256** over a secret shared by the orchestrator (signer) and Signing is **HS256** over a secret shared by the orchestrator (signer) and
+140
View File
@@ -0,0 +1,140 @@
"""Host-side control-plane client (PRD 0070).
The launch path talks to the orchestrator over its HTTP control plane to
register, re-policy, and tear down bottles the counterpart to the
gateway-side `PolicyResolver` (which only reads `/resolve`). Where
`PolicyResolver` is fail-closed and lives in the untrusted data plane, this
is the trusted control-plane caller: a non-success response is an error the
launch path must surface, not silently swallow.
Stdlib-only, so the CLI can drive the orchestrator without any dependency.
"""
from __future__ import annotations
import json
import urllib.error
import urllib.request
from dataclasses import dataclass
DEFAULT_TIMEOUT_SECONDS = 5.0
class OrchestratorClientError(RuntimeError):
"""A control-plane call failed (unreachable, or an unexpected status)."""
@dataclass(frozen=True)
class RegisteredBottle:
"""What `POST /bottles` returns: the minted bottle id and the per-bottle
identity token the agent presents for app-layer attribution."""
bottle_id: str
identity_token: str
class OrchestratorClient:
"""Trusted host-side client for the orchestrator control plane."""
def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT_SECONDS) -> None:
self._base = base_url.rstrip("/")
self._timeout = timeout
def _request(
self, method: str, path: str, body: dict[str, object] | None = None,
) -> tuple[int, dict[str, object]]:
"""Send one request; return `(status, payload)`. Raises
`OrchestratorClientError` only when the orchestrator can't be reached
or returns malformed data HTTP *status* codes are returned so
callers can treat 404 as a meaningful "no such bottle"."""
data = json.dumps(body).encode() if body is not None else None
headers = {"Content-Type": "application/json"} if data is not None else {}
req = urllib.request.Request(
f"{self._base}{path}", data=data, method=method, headers=headers,
)
try:
with urllib.request.urlopen(req, timeout=self._timeout) as resp:
raw = resp.read()
payload = json.loads(raw) if raw else {}
return resp.status, payload if isinstance(payload, dict) else {}
except urllib.error.HTTPError as e:
# A structured error response still carries a usable status.
try:
payload = json.loads(e.read() or b"{}")
except (ValueError, OSError):
payload = {}
return e.code, payload if isinstance(payload, dict) else {}
except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
raise OrchestratorClientError(f"{method} {path}: {e}") from e
def _ok(self, method: str, path: str, body: dict[str, object] | None = None) -> dict[str, object]:
"""`_request` that requires a 2xx, raising otherwise."""
status, payload = self._request(method, path, body)
if not 200 <= status < 300:
detail = payload.get("error", "")
raise OrchestratorClientError(f"{method} {path}: HTTP {status} {detail}".rstrip())
return payload
def health(self) -> bool:
"""True iff the control plane answers `GET /health` with 200."""
try:
status, _ = self._request("GET", "/health")
except OrchestratorClientError:
return False
return status == 200
def register_bottle(
self,
source_ip: str,
*,
image_ref: str = "",
metadata: str = "",
policy: str = "",
) -> RegisteredBottle:
"""Register a bottle and broker its launch (`POST /bottles`). Returns
its minted id + identity token."""
payload = self._ok("POST", "/bottles", {
"source_ip": source_ip,
"image_ref": image_ref,
"metadata": metadata,
"policy": policy,
})
bottle_id = payload.get("bottle_id")
token = payload.get("identity_token")
if not isinstance(bottle_id, str) or not isinstance(token, str):
raise OrchestratorClientError("register: response missing bottle_id/identity_token")
return RegisteredBottle(bottle_id=bottle_id, identity_token=token)
def teardown_bottle(self, bottle_id: str) -> bool:
"""Tear a bottle down (`DELETE /bottles/<id>`). False if the
orchestrator didn't know it (404) — idempotent for cleanup paths."""
status, _ = self._request("DELETE", f"/bottles/{bottle_id}")
if status == 404:
return False
if not 200 <= status < 300:
raise OrchestratorClientError(f"teardown {bottle_id}: HTTP {status}")
return True
def set_policy(self, bottle_id: str, policy: str) -> bool:
"""Live-reload a bottle's policy (`PUT /bottles/<id>/policy`). False on
404 (unknown bottle)."""
status, _ = self._request("PUT", f"/bottles/{bottle_id}/policy", {"policy": policy})
if status == 404:
return False
if not 200 <= status < 300:
raise OrchestratorClientError(f"set_policy {bottle_id}: HTTP {status}")
return True
def list_bottles(self) -> list[dict[str, object]]:
"""Every registered bottle's redacted record (`GET /bottles`)."""
payload = self._ok("GET", "/bottles")
bottles = payload.get("bottles")
return bottles if isinstance(bottles, list) else []
__all__ = [
"OrchestratorClient",
"OrchestratorClientError",
"RegisteredBottle",
"DEFAULT_TIMEOUT_SECONDS",
]
+4 -4
View File
@@ -5,7 +5,7 @@ The backend-agnostic control-plane RPC (CLI / console -> orchestrator) over
vsock / unix-socket portability caveats): vsock / unix-socket portability caveats):
GET /health -> 200 {"status": "ok"} GET /health -> 200 {"status": "ok"}
GET /sidecar -> 200 {"configured", ["name","running"]} GET /gateway -> 200 {"configured", ["name","running"]}
GET /bottles -> 200 {"bottles": [ <redacted record>, ...]} GET /bottles -> 200 {"bottles": [ <redacted record>, ...]}
POST /bottles -> 201 {"bottle_id","identity_token"} (launch) POST /bottles -> 201 {"bottle_id","identity_token"} (launch)
body: {"source_ip", ["image_ref"], body: {"source_ip", ["image_ref"],
@@ -63,8 +63,8 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
if method == "GET" and route == "/health": if method == "GET" and route == "/health":
return 200, {"status": "ok"} return 200, {"status": "ok"}
if method == "GET" and route == "/sidecar": if method == "GET" and route == "/gateway":
return 200, orch.sidecar_status() return 200, orch.gateway_status()
if method == "GET" and route == "/bottles": if method == "GET" and route == "/bottles":
return 200, {"bottles": [r.redacted() for r in orch.registry.all()]} return 200, {"bottles": [r.redacted() for r in orch.registry.all()]}
@@ -122,7 +122,7 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
return 200, {"bottle_id": rec.bottle_id} return 200, {"bottle_id": rec.bottle_id}
if method == "POST" and route == "/resolve": if method == "POST" and route == "/resolve":
# The per-request lookup the multi-tenant sidecar makes: returns the # The per-request lookup the multi-tenant gateway makes: returns the
# bottle's policy. identity_token is OPTIONAL — absent means resolve # bottle's policy. identity_token is OPTIONAL — absent means resolve
# by source IP alone (network-layer attribution). # by source IP alone (network-layer attribution).
try: try:
@@ -1,18 +1,18 @@
"""The consolidated per-host sidecar (PRD 0070). """The consolidated per-host gateway (PRD 0070).
The core consolidation win: **one** persistent sidecar per host, shared by The core consolidation win: **one** persistent gateway per host, shared by
every bottle, instead of a sidecar bundle per bottle. It's safe to share every bottle, instead of a sidecar bundle per bottle. It's safe to share
because the attribution invariant (source IP + identity token, see because the attribution invariant (source IP + identity token, see
`registry`) lets the sidecar attribute each request to the right bottle `registry`) lets the gateway attribute each request to the right bottle
so per-bottle policy lives in one long-lived process keyed on who's calling. so per-bottle policy lives in one long-lived process keyed on who's calling.
`Sidecar` is the backend-neutral lifecycle contract (mirrors `LaunchBroker`): `Gateway` is the backend-neutral lifecycle contract (mirrors `LaunchBroker`):
ensure the single instance is up, report it, tear it down. `DockerSidecar` ensure the single instance is up, report it, tear it down. `DockerGateway`
is the docker implementation; a firecracker sidecar VM slots in later. is the docker implementation; a firecracker gateway VM slots in later.
The defining behaviour is **idempotent singleton**: `ensure_running` starts The defining behaviour is **idempotent singleton**: `ensure_running` starts
the instance if absent and is a no-op if it's already up, so N bottle the instance if absent and is a no-op if it's already up, so N bottle
launches never spawn N sidecars. launches never spawn N gateways.
""" """
from __future__ import annotations from __future__ import annotations
@@ -23,49 +23,54 @@ from pathlib import Path
from ..docker_cmd import run_docker from ..docker_cmd import run_docker
SIDECAR_NAME = "bot-bottle-orch-sidecar" GATEWAY_NAME = "bot-bottle-orch-gateway"
SIDECAR_LABEL = "bot-bottle-orch-sidecar=1" GATEWAY_LABEL = "bot-bottle-orch-gateway=1"
# The single user-defined network the gateway and every agent bottle share.
# Agents attach here with a pinned IP and reach the gateway's egress /
# git-http / supervise ports by its address — no host port publishing, and
# the source IP the gateway attributes by is the address on this network.
GATEWAY_NETWORK = "bot-bottle-gateway"
# The real sidecar-bundle image + its Dockerfile. Kept as a local constant # The real sidecar-bundle image + its Dockerfile. Kept as a local constant
# rather than imported from backend.docker.sidecar_bundle, which would drag # rather than imported from backend.docker.sidecar_bundle, which would drag
# the whole backend layer into the lean orchestrator (see #359); unify when # the whole backend layer into the lean orchestrator (see #359); unify when
# that lands. Env override matches the backend's BOT_BOTTLE_SIDECAR_IMAGE. # that lands. Env override matches the backend's BOT_BOTTLE_SIDECAR_IMAGE.
SIDECAR_BUNDLE_IMAGE = os.environ.get("BOT_BOTTLE_SIDECAR_IMAGE", "bot-bottle-sidecars:latest") GATEWAY_IMAGE = os.environ.get("BOT_BOTTLE_SIDECAR_IMAGE", "bot-bottle-sidecars:latest")
SIDECAR_DOCKERFILE = "Dockerfile.sidecars" GATEWAY_DOCKERFILE = "Dockerfile.sidecars"
_REPO_ROOT = Path(__file__).resolve().parents[2] _REPO_ROOT = Path(__file__).resolve().parents[2]
class SidecarError(Exception): class GatewayError(Exception):
"""The shared sidecar failed to build/start/stop (non-zero `docker` exit).""" """The shared gateway failed to build/start/stop (non-zero `docker` exit)."""
class Sidecar(abc.ABC): class Gateway(abc.ABC):
"""Lifecycle of the single per-host sidecar. Backend-neutral.""" """Lifecycle of the single per-host gateway. Backend-neutral."""
name: str name: str
def ensure_built(self) -> None: def ensure_built(self) -> None:
"""Ensure the sidecar's image / rootfs exists, building it if needed. """Ensure the gateway's image / rootfs exists, building it if needed.
Default: nothing to build (e.g. a stub or a pre-pulled image).""" Default: nothing to build (e.g. a stub or a pre-pulled image)."""
return return
@abc.abstractmethod @abc.abstractmethod
def ensure_running(self) -> None: def ensure_running(self) -> None:
"""Start the sidecar if it isn't already up. Idempotent: a no-op """Start the gateway if it isn't already up. Idempotent: a no-op
when it's already running (that's the whole point one per host). when it's already running (that's the whole point one per host).
Assumes the image exists call `ensure_built()` first.""" Assumes the image exists call `ensure_built()` first."""
@abc.abstractmethod @abc.abstractmethod
def is_running(self) -> bool: def is_running(self) -> bool:
"""True iff the sidecar instance is currently up.""" """True iff the gateway instance is currently up."""
@abc.abstractmethod @abc.abstractmethod
def stop(self) -> None: def stop(self) -> None:
"""Remove the sidecar. Idempotent — absent is success.""" """Remove the gateway. Idempotent — absent is success."""
class DockerSidecar(Sidecar): class DockerGateway(Gateway):
"""The consolidated sidecar as a single, fixed-name Docker container. """The consolidated gateway as a single, fixed-name Docker container.
`image_ref` defaults to the real sidecar-bundle image; `ensure_built` `image_ref` defaults to the real sidecar-bundle image; `ensure_built`
builds it from `Dockerfile.sidecars` when it's missing. (Note: slice 5 builds it from `Dockerfile.sidecars` when it's missing. (Note: slice 5
@@ -74,14 +79,16 @@ class DockerSidecar(Sidecar):
def __init__( def __init__(
self, self,
image_ref: str = SIDECAR_BUNDLE_IMAGE, image_ref: str = GATEWAY_IMAGE,
*, *,
name: str = SIDECAR_NAME, name: str = GATEWAY_NAME,
network: str = GATEWAY_NETWORK,
build_context: Path | None = None, build_context: Path | None = None,
dockerfile: str | None = SIDECAR_DOCKERFILE, dockerfile: str | None = GATEWAY_DOCKERFILE,
) -> None: ) -> None:
self.image_ref = image_ref self.image_ref = image_ref
self.name = name self.name = name
self.network = network
self._build_context = build_context or _REPO_ROOT self._build_context = build_context or _REPO_ROOT
self._dockerfile = dockerfile self._dockerfile = dockerfile
@@ -101,7 +108,7 @@ class DockerSidecar(Sidecar):
str(self._build_context), str(self._build_context),
]) ])
if proc.returncode != 0: if proc.returncode != 0:
raise SidecarError(f"sidecar image build failed: {proc.stderr.strip()}") raise GatewayError(f"gateway image build failed: {proc.stderr.strip()}")
def is_running(self) -> bool: def is_running(self) -> bool:
proc = run_docker([ proc = run_docker([
@@ -112,28 +119,42 @@ class DockerSidecar(Sidecar):
]) ])
return self.name in proc.stdout.split() return self.name in proc.stdout.split()
def _ensure_network(self) -> None:
"""Create the shared gateway network if it doesn't exist. Idempotent —
a concurrent create loses harmlessly (the loser sees 'already exists').
Docker picks the subnet; the launcher reads it back to allocate IPs."""
if run_docker(["docker", "network", "inspect", self.network]).returncode == 0:
return
proc = run_docker(["docker", "network", "create", self.network])
if proc.returncode != 0 and "already exists" not in proc.stderr:
raise GatewayError(
f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
)
def ensure_running(self) -> None: def ensure_running(self) -> None:
if self.is_running(): if self.is_running():
return return
self._ensure_network()
# Clear any stale (stopped) container holding the fixed name, then # Clear any stale (stopped) container holding the fixed name, then
# start fresh. `rm --force` on an absent name is a tolerated no-op. # start fresh. `rm --force` on an absent name is a tolerated no-op.
run_docker(["docker", "rm", "--force", self.name]) run_docker(["docker", "rm", "--force", self.name])
proc = run_docker([ proc = run_docker([
"docker", "run", "--detach", "docker", "run", "--detach",
"--name", self.name, "--name", self.name,
"--label", SIDECAR_LABEL, "--label", GATEWAY_LABEL,
"--network", self.network,
self.image_ref, self.image_ref,
]) ])
if proc.returncode != 0: if proc.returncode != 0:
raise SidecarError(f"sidecar failed to start: {proc.stderr.strip()}") raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
def stop(self) -> None: def stop(self) -> None:
proc = run_docker(["docker", "rm", "--force", self.name]) proc = run_docker(["docker", "rm", "--force", self.name])
if proc.returncode != 0 and "No such container" not in proc.stderr: if proc.returncode != 0 and "No such container" not in proc.stderr:
raise SidecarError(f"sidecar failed to stop: {proc.stderr.strip()}") raise GatewayError(f"gateway failed to stop: {proc.stderr.strip()}")
__all__ = [ __all__ = [
"Sidecar", "DockerSidecar", "SidecarError", "Gateway", "DockerGateway", "GatewayError",
"SIDECAR_NAME", "SIDECAR_LABEL", "SIDECAR_BUNDLE_IMAGE", "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
] ]
+141
View File
@@ -0,0 +1,141 @@
"""Orchestrator process lifecycle (PRD 0070, docker slice).
Before the CLI can register or launch bottles against the consolidated
model, exactly one orchestrator control plane and the single per-host
gateway it manages must be running. This starts the orchestrator
dev-harness (`python -m bot_bottle.orchestrator`) as a background host
process and health-checks it.
It is an **idempotent singleton**: `ensure_running` returns immediately if a
healthy control plane already answers on the port, and otherwise spawns one
and waits for it to come up. The control-plane port is the singleton key
a second orchestrator can't bind it, so a stray double-start fails fast
rather than forking a rival.
Host-process (not container) on purpose: the PRD sequences the orchestrator
as a plain-process dev-harness first (fast iteration, and it already has the
host user's docker access to broker launches), while the data-plane
*gateway* it manages runs as a container. Wrapping the orchestrator itself
in a backend-native unit is a later step.
"""
from __future__ import annotations
import subprocess
import sys
import time
import urllib.error
import urllib.request
from .. import log
from ..paths import bot_bottle_root
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8080
# Poll cadence + default ceiling while waiting for a freshly-spawned control
# plane to answer /health (the first start also builds/boots the gateway).
_HEALTH_POLL_SECONDS = 0.25
DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
_HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
class OrchestratorStartError(RuntimeError):
"""The orchestrator process did not become healthy within the timeout."""
class OrchestratorProcess:
"""Manages the local orchestrator control-plane process for the docker
backend. Backend-neutral callers only need `ensure_running()` + `url`."""
def __init__(
self,
host: str = DEFAULT_HOST,
port: int = DEFAULT_PORT,
*,
broker: str = "docker",
gateway: bool = True,
) -> None:
self.host = host
self.port = port
self._broker = broker
self._gateway = gateway
@property
def url(self) -> str:
"""The control-plane base URL — also what the data plane's
BOT_BOTTLE_ORCHESTRATOR_URL points at."""
return f"http://{self.host}:{self.port}"
def is_healthy(self, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS) -> bool:
"""True iff a control plane answers `GET /health` with 200 — the
singleton liveness check."""
try:
with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp:
return resp.status == 200
except (urllib.error.URLError, TimeoutError, OSError):
return False
def ensure_running(
self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
) -> str:
"""Return the control-plane URL, starting the orchestrator first if it
isn't already healthy. Idempotent — a healthy control plane is left
untouched. Raises `OrchestratorStartError` if a freshly-spawned one
doesn't answer within `startup_timeout`."""
if self.is_healthy():
return self.url
log.info("starting orchestrator", context={"url": self.url, "broker": self._broker})
self._spawn()
deadline = time.monotonic() + startup_timeout
while time.monotonic() < deadline:
if self.is_healthy():
log.info("orchestrator healthy", context={"url": self.url})
return self.url
time.sleep(_HEALTH_POLL_SECONDS)
raise OrchestratorStartError(
f"orchestrator at {self.url} did not become healthy within "
f"{startup_timeout:g}s"
)
def _argv(self) -> list[str]:
"""`python -m bot_bottle.orchestrator ...` — static flags only."""
argv = [
sys.executable, "-m", "bot_bottle.orchestrator",
"--host", self.host, "--port", str(self.port),
"--broker", self._broker,
]
if self._gateway:
argv.append("--gateway")
return argv
def _log_path(self) -> str:
"""Where the detached orchestrator's stdout/stderr goes so a failed
start is diagnosable after the CLI has moved on."""
return str(bot_bottle_root() / "orchestrator.log")
def _spawn(self) -> None:
"""Launch the orchestrator detached so it outlives this CLI process,
with its output tee'd to a log file under the bot-bottle root."""
root = bot_bottle_root()
root.mkdir(parents=True, exist_ok=True)
logfile = open(self._log_path(), "a", encoding="utf-8") # noqa: SIM115 # pylint: disable=consider-using-with
try:
subprocess.Popen( # noqa: S603 # pylint: disable=consider-using-with
self._argv(),
stdout=logfile,
stderr=subprocess.STDOUT,
stdin=subprocess.DEVNULL,
start_new_session=True,
)
finally:
# The child inherits its own dup'd fd; this handle is ours to drop.
logfile.close()
__all__ = [
"OrchestratorProcess",
"OrchestratorStartError",
"DEFAULT_HOST",
"DEFAULT_PORT",
"DEFAULT_STARTUP_TIMEOUT_SECONDS",
]
+57
View File
@@ -0,0 +1,57 @@
"""Consolidated registration inputs (PRD 0070, docker slice).
Bridges the existing per-bottle `prepare` output to the consolidated
registry: turns a prepared bottle's egress plan into the backend-neutral
inputs `Orchestrator.launch_bottle` takes the egress **policy** blob and
launch **metadata**.
The policy blob is the exact routes YAML the per-bottle egress sidecar used
to read from a file; in the consolidated model the multi-tenant gateway's
`PolicyResolver` fetches it from the registry per request (keyed by source
IP) instead. Same render, so consolidated and single-tenant egress apply
byte-identical policy a bottle's allow-list doesn't change when it moves
onto the shared gateway.
Host-side glue (imports `bot_bottle.egress`), used by the launch path not
by the lean orchestrator control-plane process itself.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from ..egress import EgressPlan, egress_render_routes
@dataclass(frozen=True)
class RegistrationInputs:
"""What `Orchestrator.launch_bottle` needs to register a bottle, derived
from its prepared plan. `policy` is served verbatim by the gateway's
`/resolve`; `metadata` is opaque forward-compat state it carries the
human slug so the console / supervise can show a name, not just the
minted bottle id."""
policy: str
metadata: str
def egress_policy(plan: EgressPlan) -> str:
"""The bottle's egress policy blob: the routes YAML the gateway serves
and the addon parses with `load_config`. Identical to the per-bottle
`routes.yaml` render, so the consolidated path applies the same
allow-list."""
return egress_render_routes(plan.routes, log=plan.log)
def registration_inputs(plan: EgressPlan) -> RegistrationInputs:
"""Assemble the orchestrator registration inputs from a prepared egress
plan. `metadata` records the slug so the shared registry can map a minted
bottle id back to its human name."""
return RegistrationInputs(
policy=egress_policy(plan),
metadata=json.dumps({"slug": plan.slug}),
)
__all__ = ["RegistrationInputs", "egress_policy", "registration_inputs"]
+5 -5
View File
@@ -67,9 +67,9 @@ class BottleRecord:
# Opaque JSON blob for forward-compat (pool slot, ...) — the registry # Opaque JSON blob for forward-compat (pool slot, ...) — the registry
# doesn't interpret it, so new fields need no migration. # doesn't interpret it, so new fields need no migration.
metadata: str = "" metadata: str = ""
# The bottle's sidecar policy (opaque JSON — egress allowlist / routes / # The bottle's gateway policy (opaque JSON — egress allowlist / routes /
# git config). The registry stores and serves it verbatim, keyed by # git config). The registry stores and serves it verbatim, keyed by
# source IP; the multi-tenant sidecar interprets it. # source IP; the multi-tenant gateway interprets it.
policy: str = "" policy: str = ""
def redacted(self) -> dict[str, object]: def redacted(self) -> dict[str, object]:
@@ -102,9 +102,9 @@ _MIGRATIONS = TableMigrations(
# the "one active bottle per source IP" check cheap. # the "one active bottle per source IP" check cheap.
"CREATE INDEX IF NOT EXISTS idx_orchestrator_bottles_source_ip " "CREATE INDEX IF NOT EXISTS idx_orchestrator_bottles_source_ip "
"ON orchestrator_bottles (source_ip)", "ON orchestrator_bottles (source_ip)",
# v3 — per-bottle policy (opaque JSON the sidecar interprets): the # v3 — per-bottle policy (opaque JSON the gateway interprets): the
# egress allowlist / routes / git config selected by source IP. The # egress allowlist / routes / git config selected by source IP. The
# multi-tenant sidecar resolves it per request via `attribute`. # multi-tenant gateway resolves it per request via `attribute`.
"ALTER TABLE orchestrator_bottles ADD COLUMN policy TEXT NOT NULL DEFAULT ''", "ALTER TABLE orchestrator_bottles ADD COLUMN policy TEXT NOT NULL DEFAULT ''",
], ],
) )
@@ -217,7 +217,7 @@ class RegistryStore(DbStore):
IP, or None if unknown or ambiguous (more than one a IP, or None if unknown or ambiguous (more than one a
misconfiguration). Safe as the *sole* attributor only where the misconfiguration). Safe as the *sole* attributor only where the
source IP is unspoofable (Firecracker `/31` + nft) and the control source IP is unspoofable (Firecracker `/31` + nft) and the control
plane is reachable only by the trusted sidecar; pair with the plane is reachable only by the trusted gateway; pair with the
identity token (`attribute`) elsewhere.""" identity token (`attribute`) elsewhere."""
with self._connect() as conn: with self._connect() as conn:
rows = conn.execute( rows = conn.execute(
+20 -20
View File
@@ -19,12 +19,12 @@ from __future__ import annotations
from .broker import LaunchBroker, LaunchRequest, sign_request from .broker import LaunchBroker, LaunchRequest, sign_request
from .registry import BottleRecord, RegistryStore from .registry import BottleRecord, RegistryStore
from .sidecar import Sidecar from .gateway import Gateway
class Orchestrator: class Orchestrator:
"""Owns the registry + brokers launches, and manages the single """Owns the registry + brokers launches, and manages the single
consolidated per-host sidecar. Backend-neutral (broker and sidecar consolidated per-host gateway. Backend-neutral (broker and gateway
abstract the backend-native pieces).""" abstract the backend-native pieces)."""
def __init__( def __init__(
@@ -32,12 +32,12 @@ class Orchestrator:
registry: RegistryStore, registry: RegistryStore,
broker: LaunchBroker, broker: LaunchBroker,
sign_secret: bytes, sign_secret: bytes,
sidecar: Sidecar | None = None, gateway: Gateway | None = None,
) -> None: ) -> None:
self.registry = registry self.registry = registry
self._broker = broker self._broker = broker
self._secret = sign_secret self._secret = sign_secret
self._sidecar = sidecar self._gateway = gateway
def launch_bottle( def launch_bottle(
self, self,
@@ -48,7 +48,7 @@ class Orchestrator:
metadata: str = "", metadata: str = "",
policy: str = "", policy: str = "",
) -> BottleRecord: ) -> BottleRecord:
"""Register a bottle (with its sidecar policy) and broker its launch. """Register a bottle (with its gateway policy) and broker its launch.
Rolls the registry entry back if the launch doesn't take, so a Rolls the registry entry back if the launch doesn't take, so a
failure leaves no orphan.""" failure leaves no orphan."""
rec = self.registry.register(source_ip, metadata=metadata, policy=policy) rec = self.registry.register(source_ip, metadata=metadata, policy=policy)
@@ -84,37 +84,37 @@ class Orchestrator:
def resolve(self, source_ip: str, identity_token: str = "") -> BottleRecord | None: def resolve(self, source_ip: str, identity_token: str = "") -> BottleRecord | None:
"""Resolve the bottle behind a request — the source-IP-keyed lookup """Resolve the bottle behind a request — the source-IP-keyed lookup
the multi-tenant sidecar makes per request; the returned record the multi-tenant gateway makes per request; the returned record
carries its `policy`. With a token, full attribution (source IP + carries its `policy`. With a token, full attribution (source IP +
token); without, network-layer attribution by source IP alone token); without, network-layer attribution by source IP alone
(valid where the IP is unspoofable and the control plane is (valid where the IP is unspoofable and the control plane is
sidecar-only).""" gateway-only)."""
if identity_token: if identity_token:
return self.registry.attribute(source_ip, identity_token) return self.registry.attribute(source_ip, identity_token)
return self.registry.by_source_ip(source_ip) return self.registry.by_source_ip(source_ip)
def set_policy(self, bottle_id: str, policy: str) -> bool: def set_policy(self, bottle_id: str, policy: str) -> bool:
"""Update a bottle's sidecar policy in place (live reload). False if """Update a bottle's gateway policy in place (live reload). False if
the bottle is unknown.""" the bottle is unknown."""
return self.registry.set_policy(bottle_id, policy) return self.registry.set_policy(bottle_id, policy)
# --- consolidated sidecar ---------------------------------------------- # --- consolidated gateway ----------------------------------------------
def ensure_sidecar(self) -> None: def ensure_gateway(self) -> None:
"""Ensure the single per-host sidecar is built and up (idempotent). """Ensure the single per-host gateway is built and up (idempotent).
No-op when no sidecar is configured.""" No-op when no gateway is configured."""
if self._sidecar is not None: if self._gateway is not None:
self._sidecar.ensure_built() self._gateway.ensure_built()
self._sidecar.ensure_running() self._gateway.ensure_running()
def sidecar_status(self) -> dict[str, object]: def gateway_status(self) -> dict[str, object]:
"""Report the shared sidecar for the control plane / console.""" """Report the shared gateway for the control plane / console."""
if self._sidecar is None: if self._gateway is None:
return {"configured": False} return {"configured": False}
return { return {
"configured": True, "configured": True,
"name": self._sidecar.name, "name": self._gateway.name,
"running": self._sidecar.is_running(), "running": self._gateway.is_running(),
} }
+56 -9
View File
@@ -1,6 +1,6 @@
"""Sidecar-side per-client policy resolver (PRD 0070). """Gateway-side per-client policy resolver (PRD 0070).
The consolidated sidecar serves every bottle from one process, so for each The consolidated gateway serves every bottle from one process, so for each
request it must apply the *calling* bottle's policy, selected by the source request it must apply the *calling* bottle's policy, selected by the source
IP the attribution invariant makes unspoofable. This resolves that policy IP the attribution invariant makes unspoofable. This resolves that policy
from the orchestrator's control plane (`POST /resolve`), keyed on the from the orchestrator's control plane (`POST /resolve`), keyed on the
@@ -47,12 +47,11 @@ class PolicyResolver:
self._base = base_url.rstrip("/") self._base = base_url.rstrip("/")
self._timeout = timeout self._timeout = timeout
def resolve(self, source_ip: str, identity_token: str = "") -> str | None: def _post_resolve(self, source_ip: str, identity_token: str) -> dict[str, object] | None:
"""The calling bottle's policy blob, or None if unattributed. Always """The orchestrator's `/resolve` payload for this client, or None if
fetches from the orchestrator so revocations / changes / teardowns unattributed (a clean `403`). Raises `PolicyResolveError` on an
are honored immediately. `identity_token` is optional omit it to unreachable / unexpected-status / malformed response so every caller
resolve by source IP alone (the network-layer attribution). can fail closed. Shared by `resolve` and `resolve_bottle_id`."""
Raises `PolicyResolveError` if the orchestrator can't be reached."""
body = json.dumps( body = json.dumps(
{"source_ip": source_ip, "identity_token": identity_token} {"source_ip": source_ip, "identity_token": identity_token}
).encode() ).encode()
@@ -69,8 +68,56 @@ class PolicyResolver:
raise PolicyResolveError(f"/resolve returned HTTP {e.code}") from e raise PolicyResolveError(f"/resolve returned HTTP {e.code}") from e
except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e: except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
raise PolicyResolveError(f"/resolve unreachable or malformed: {e}") from e raise PolicyResolveError(f"/resolve unreachable or malformed: {e}") from e
policy = payload.get("policy") if isinstance(payload, dict) else None return payload if isinstance(payload, dict) else {}
def resolve(self, source_ip: str, identity_token: str = "") -> str | None:
"""The calling bottle's policy blob, or None if unattributed. Always
fetches from the orchestrator so revocations / changes / teardowns
are honored immediately.
`identity_token` is optional *transitionally* omitting it resolves
by source IP alone (the network-layer attribution, sound by
construction on Firecracker's `/31`+nft, weaker elsewhere). The
consolidated end state makes the token mandatory so the app-layer
defense is always enforced, not silently degraded; that flips at the
`/resolve` boundary once identity-token *delivery* lands (a PRD 0070
open question we can't require what the agent can't yet present).
Raises `PolicyResolveError` if the orchestrator can't be reached."""
payload = self._post_resolve(source_ip, identity_token)
if payload is None:
return None
policy = payload.get("policy")
return policy if isinstance(policy, str) else "" return policy if isinstance(policy, str) else ""
def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str | None:
"""The calling bottle's id, or None if unattributed. This is the
source-IP-keyed identity the git-gate uses to select the bottle's
repo namespace (there is no per-bottle policy blob to parse the
bottle *is* the namespace). Same fail-closed contract as `resolve`."""
payload = self._post_resolve(source_ip, identity_token)
if payload is None:
return None
bottle_id = payload.get("bottle_id")
return bottle_id if isinstance(bottle_id, str) and bottle_id else None
def resolve_policy_and_bottle_id(
self, source_ip: str, identity_token: str = "",
) -> tuple[str | None, str | None]:
"""Both the policy blob and the bottle id in a single `/resolve` — so
a caller that needs each (the egress addon: policy for routing, bottle
id to key the calling bottle's supervise queue + safelist) makes one
round-trip, not two. Returns `(None, None)` when unattributed (a clean
`403`). Raises `PolicyResolveError` if the orchestrator can't be
reached, so the caller still fails closed."""
payload = self._post_resolve(source_ip, identity_token)
if payload is None:
return None, None
policy = payload.get("policy")
bottle_id = payload.get("bottle_id")
return (
policy if isinstance(policy, str) else "",
bottle_id if isinstance(bottle_id, str) and bottle_id else None,
)
__all__ = ["PolicyResolver", "PolicyResolveError"] __all__ = ["PolicyResolver", "PolicyResolveError"]
+50 -5
View File
@@ -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 container creation by the backend's start step). SUPERVISE_DB_PATH
points at the bind-mounted host database. 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: Speaks MCP over HTTP+JSON-RPC. Methods handled:
* `initialize` handshake; returns server info + caps. * `initialize` handshake; returns server info + caps.
@@ -40,16 +46,18 @@ import time
import typing import typing
import urllib.error import urllib.error
import urllib.request import urllib.request
from dataclasses import dataclass from dataclasses import dataclass, replace
try: try:
# Same-directory imports inside the bundle container; these files are # Same-directory imports inside the bundle container; these files are
# COPYed flat under /app by Dockerfile.sidecars. # COPYed flat under /app by Dockerfile.sidecars.
from egress_addon_core import LOG_OFF, load_config from egress_addon_core import LOG_OFF, load_config
from policy_resolver import PolicyResolveError, PolicyResolver
import supervise as _sv import supervise as _sv
except ModuleNotFoundError: except ModuleNotFoundError:
# Package imports for host-side tests and tooling. # Package imports for host-side tests and tooling.
from .egress_addon_core import LOG_OFF, load_config from .egress_addon_core import LOG_OFF, load_config
from .policy_resolver import PolicyResolveError, PolicyResolver
from . import supervise as _sv from . import supervise as _sv
@@ -73,6 +81,12 @@ DEFAULT_RESPONSE_TIMEOUT_SECONDS = 30.0
MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.05 MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.05
EGRESS_LIST_TIMEOUT_SECONDS = 5.0 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) @dataclass(frozen=True)
class JsonRpcRequest: class JsonRpcRequest:
@@ -511,9 +525,30 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
if method == "tools/list": if method == "tools/list":
return handle_tools_list(req.params) return handle_tools_list(req.params)
if method == "tools/call": 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}") 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: def _write_jsonrpc(self, body: bytes) -> None:
self.send_response(200) self.send_response(200)
self.send_header("Content-Type", "application/json") self.send_header("Content-Type", "application/json")
@@ -537,6 +572,9 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
allow_reuse_address = True allow_reuse_address = True
daemon_threads = True daemon_threads = True
config: ServerConfig = ServerConfig(bottle_slug="") 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 ----------------------------------------------------------- # --- Entry point -----------------------------------------------------------
@@ -548,15 +586,17 @@ def serve(
port: int = _sv.SUPERVISE_PORT, port: int = _sv.SUPERVISE_PORT,
bind: str = "0.0.0.0", bind: str = "0.0.0.0",
response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS, response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS,
resolver: "PolicyResolver | None" = None,
) -> typing.NoReturn: ) -> typing.NoReturn:
server = MCPServer((bind, port), MCPHandler) server = MCPServer((bind, port), MCPHandler)
server.config = ServerConfig( server.config = ServerConfig(
bottle_slug=bottle_slug, bottle_slug=bottle_slug,
response_timeout_seconds=response_timeout_seconds, response_timeout_seconds=response_timeout_seconds,
) )
server.policy_resolver = resolver
mode = "multi-tenant" if resolver else f"slug={bottle_slug!r}"
sys.stderr.write( sys.stderr.write(
f"supervise listening on {bind}:{port}; " f"supervise listening on {bind}:{port}; {mode}; "
f"slug={bottle_slug!r}; "
f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n" # type: ignore[arg-type] f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n" # type: ignore[arg-type]
) )
sys.stderr.flush() sys.stderr.flush()
@@ -571,8 +611,12 @@ def serve(
def main(argv: list[str]) -> int: def main(argv: list[str]) -> int:
del argv # config is env-only, no CLI flags 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", "") 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") sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n")
return 2 return 2
port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT))) port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT)))
@@ -587,6 +631,7 @@ def main(argv: list[str]) -> int:
port=port, port=port,
bind=bind, bind=bind,
response_timeout_seconds=response_timeout_seconds, response_timeout_seconds=response_timeout_seconds,
resolver=resolver,
) )
return 0 # serve() does not return return 0 # serve() does not return
+20 -20
View File
@@ -10,8 +10,8 @@
## Summary ## Summary
Replace the **per-bottle sidecar bundle** with a single **persistent, Replace the **per-bottle gateway bundle** with a single **persistent,
per-host orchestrator**: one long-lived service that runs the sidecar per-host orchestrator**: one long-lived service that runs the gateway
functions (egress / git-gate / supervise), coordinates with the console, functions (egress / git-gate / supervise), coordinates with the console,
and brokers agent launches and teardown. It is **virtualized from the and brokers agent launches and teardown. It is **virtualized from the
start** using each backend's native isolation primitive — a Firecracker start** using each backend's native isolation primitive — a Firecracker
@@ -22,13 +22,13 @@ container on the legacy backend — and is fronted by a single
## Motivation ## Motivation
Today each bottle spins up its own sidecar bundle (egress mitmproxy + Today each bottle spins up its own gateway bundle (egress mitmproxy +
git-gate + supervise). That costs: git-gate + supervise). That costs:
- **Resources.** N bottles → N heavy bundles booting and idling. - **Resources.** N bottles → N heavy bundles booting and idling.
- **Operational churn.** Per-launch container/VM lifecycle for the - **Operational churn.** Per-launch container/VM lifecycle for the
sidecars, a control path baked at launch and torn down at exit. gateways, a control path baked at launch and torn down at exit.
- **A blurry contract.** "How a bottle talks to its sidecar" is - **A blurry contract.** "How a bottle talks to its gateway" is
re-implemented per backend instead of being one agreed interface. re-implemented per backend instead of being one agreed interface.
A per-host orchestrator collapses the first two and forces the third to A per-host orchestrator collapses the first two and forces the third to
@@ -52,11 +52,11 @@ weaken, and the weakened ones must be mitigated by design, not hand-waved.
structured, auditable privileged core instead of a fat socket. structured, auditable privileged core instead of a fat socket.
- **Attribution is enforced, not assumed.** Making source-IP identity a - **Attribution is enforced, not assumed.** Making source-IP identity a
first-class contract invariant (below) means each backend must *prove* first-class contract invariant (below) means each backend must *prove*
it, rather than the sidecar implicitly trusting network position. it, rather than the gateway implicitly trusting network position.
### What gets weaker, and the mitigation ### What gets weaker, and the mitigation
1. **Secret concentration.** Per-bottle sidecars isolate secrets at the 1. **Secret concentration.** Per-bottle gateways isolate secrets at the
process boundary — each holds only its bottle's tokens/keys. A host process boundary — each holds only its bottle's tokens/keys. A host
orchestrator concentrates **every bottle's** egress tokens, git deploy orchestrator concentrates **every bottle's** egress tokens, git deploy
keys, and the console credential in one long-lived process. A single keys, and the console credential in one long-lived process. A single
@@ -75,7 +75,7 @@ weaken, and the weakened ones must be mitigated by design, not hand-waved.
fleet, plus launch authority, plus the console token. fleet, plus launch authority, plus the console token.
- *Mitigation:* the orchestrator is itself confined (its own VM/container - *Mitigation:* the orchestrator is itself confined (its own VM/container
with its own fail-closed egress); make it **restartable without killing with its own fail-closed egress); make it **restartable without killing
running agent VMs** (agents keep running; they briefly lose sidecar running agent VMs** (agents keep running; they briefly lose gateway
connectivity until it's back); persist state to a host volume so a connectivity until it's back); persist state to a host volume so a
restart re-adopts live bottles rather than losing them. restart re-adopts live bottles rather than losing them.
@@ -114,13 +114,13 @@ responsibility** and part of the contract:
- **Apple** — the host-only network. - **Apple** — the host-only network.
If a backend can't honor the invariant, source-IP consolidation is not If a backend can't honor the invariant, source-IP consolidation is not
safe there and that backend keeps per-bottle sidecars. The invariant is a safe there and that backend keeps per-bottle gateways. The invariant is a
hard precondition, not an aspiration. hard precondition, not an aspiration.
**Defense-in-depth — a per-bottle identity token.** On top of the **Defense-in-depth — a per-bottle identity token.** On top of the
network-layer invariant, inject a per-bottle secret token into every network-layer invariant, inject a per-bottle secret token into every
request the agent makes to the orchestrator (the agent already egresses request the agent makes to the orchestrator (the agent already egresses
through the sidecar proxy, so this is cheap to add). It gives an through the gateway proxy, so this is cheap to add). It gives an
**application-layer** proof of identity independent of the network layer: **application-layer** proof of identity independent of the network layer:
- On **Firecracker** the `/31` + nft already make source IP unspoofable - On **Firecracker** the `/31` + nft already make source IP unspoofable
@@ -133,7 +133,7 @@ Requirements: the token is **per-bottle, unguessable, and
non-cross-leakable** — a bottle can only ever prove it is *itself* (it non-cross-leakable** — a bottle can only ever prove it is *itself* (it
can't learn another bottle's token, given bottle isolation), so a hostile can't learn another bottle's token, given bottle isolation), so a hostile
agent gains nothing by presenting it. The orchestrator provisions the agent gains nothing by presenting it. The orchestrator provisions the
token at launch and the sidecar requires it to attribute + authorize. token at launch and the gateway requires it to attribute + authorize.
Note this hardens *attribution*, not *secret exposure*: it's app-layer, so Note this hardens *attribution*, not *secret exposure*: it's app-layer, so
a compromised orchestrator still sees every token (that's the a compromised orchestrator still sees every token (that's the
concentration problem, addressed separately under Secret handling). concentration problem, addressed separately under Secret handling).
@@ -178,7 +178,7 @@ manifest anywhere a raw token value is accepted, discovered from
`AgentProvider`s are — because maintaining every forge/cloud/OAuth `AgentProvider`s are — because maintaining every forge/cloud/OAuth
provider in-tree is untenable. The orchestrator's vault mints via these provider in-tree is untenable. The orchestrator's vault mints via these
providers; but the abstraction is shippable independently and today's providers; but the abstraction is shippable independently and today's
per-bottle sidecars can use it too. per-bottle gateways can use it too.
## Design ## Design
@@ -195,7 +195,7 @@ Three surfaces; only one is per-backend.
every host (no vsock/unix-socket portability caveats); a local unix every host (no vsock/unix-socket portability caveats); a local unix
socket is a fine optimization, but HTTP is the wire contract. socket is a fine optimization, but HTTP is the wire contract.
2. **Data plane (agent → orchestrator)** — the egress / git / supervise 2. **Data plane (agent → orchestrator)** — the egress / git / supervise
endpoints. Already agnostic today (agents dial `http://sidecar:9099`); endpoints. Already agnostic today (agents dial `http://gateway:9099`);
only the *address* and *how packets get there* are per-backend. only the *address* and *how packets get there* are per-backend.
3. **Launch / wire (orchestrator → backend)** — the irreducibly 3. **Launch / wire (orchestrator → backend)** — the irreducibly
backend-specific part; lives on `BottleBackend`. backend-specific part; lives on `BottleBackend`.
@@ -249,7 +249,7 @@ forged one from a compromised co-located component. The orchestrator signs;
the broker verifies with the orchestrator's public key (provisioned at the broker verifies with the orchestrator's public key (provisioned at
broker install). This is the concrete form of security #3's "structured broker install). This is the concrete form of security #3's "structured
requests only." Same JSON-with-ids + JWT shape for all requests only." Same JSON-with-ids + JWT shape for all
orchestrator↔broker/sidecar communication; cases that don't fit get orchestrator↔broker/gateway communication; cases that don't fit get
handled as they arise, with this as the default. handled as they arise, with this as the default.
If `BottleBackend` bloats, the pressure valve is composition one level If `BottleBackend` bloats, the pressure valve is composition one level
@@ -299,10 +299,10 @@ and integrate a console against.
the data plane and the console reach state through the control-plane RPC, the data plane and the console reach state through the control-plane RPC,
never a direct file handle.** No agent-facing component gets the file, so never a direct file handle.** No agent-facing component gets the file, so
none can forge attribution. (This supersedes the earlier `ro`-mount idea.) none can forge attribution. (This supersedes the earlier `ro`-mount idea.)
- *Transitional caveat:* today the per-bottle **supervise sidecar - *Transitional caveat:* today the per-bottle **supervise gateway
rw-bind-mounts `bot-bottle.db`** to write proposals — exactly the rw-bind-mounts `bot-bottle.db`** to write proposals — exactly the
pattern the orchestrator removes (supervise consolidates into the pattern the orchestrator removes (supervise consolidates into the
orchestrator; sidecar writes become RPC calls). Until that lands, don't orchestrator; gateway writes become RPC calls). Until that lands, don't
put the attribution registry behind a data-plane-writable mount. put the attribution registry behind a data-plane-writable mount.
Implementation note for the VM slices: SQLite **WAL** over a guest share Implementation note for the VM slices: SQLite **WAL** over a guest share
@@ -327,7 +327,7 @@ orchestrator becomes a VM. Decouple the two risks instead:
Backend order (cheapest proof → hardest → last): Backend order (cheapest proof → hardest → last):
1. **Docker orchestrator** — nearly free (the sidecar bundle is already 1. **Docker orchestrator** — nearly free (the gateway bundle is already
containers; collapse N bundles into one persistent container). Proves containers; collapse N bundles into one persistent container). Proves
consolidation + the `BottleBackend` seam with the least moving parts. consolidation + the `BottleBackend` seam with the least moving parts.
2. **Firecracker orchestrator** — the real work: the shim + VM-to-VM 2. **Firecracker orchestrator** — the real work: the shim + VM-to-VM
@@ -336,13 +336,13 @@ Backend order (cheapest proof → hardest → last):
the dev-harness so the app logic is already proven. the dev-harness so the app logic is already proven.
3. **macOS (Apple container)** — last (container-to-container networking). 3. **macOS (Apple container)** — last (container-to-container networking).
Keep the sidecar **service one shared thing** throughout. Keep the gateway **service one shared thing** throughout.
## Non-goals ## Non-goals
- Removing OCI/Dockerfile support for agent images (0069's concern). - Removing OCI/Dockerfile support for agent images (0069's concern).
- A single database for *all* config (see the three-tier table). - A single database for *all* config (see the three-tier table).
- Changing the per-bottle isolation of agent workloads — only the sidecar - Changing the per-bottle isolation of agent workloads — only the gateway
is consolidated; agents stay one-VM/container-each. is consolidated; agents stay one-VM/container-each.
## Relationship to other work ## Relationship to other work
@@ -393,7 +393,7 @@ Keep the sidecar **service one shared thing** throughout.
`BottleBackend.wire()` (DNAT+forward for fc, shared-net for `BottleBackend.wire()` (DNAT+forward for fc, shared-net for
docker/apple), not orchestrator logic. **Not a blocker** — resolvable at docker/apple), not orchestrator logic. **Not a blocker** — resolvable at
implementation time (may require a bit of host modification, as the pool implementation time (may require a bit of host modification, as the pool
setup already does on NixOS); an earlier "sidecars in VMs" spike showed setup already does on NixOS); an earlier "gateways in VMs" spike showed
it's feasible. it's feasible.
- **Live-reload protocol** for per-bottle policy over the HTTP control - **Live-reload protocol** for per-bottle policy over the HTTP control
plane (add/remove routes/keys/proposals without a restart). plane (add/remove routes/keys/proposals without a restart).
@@ -1,7 +1,7 @@
"""Integration: the consolidated Docker sidecar is an idempotent singleton. """Integration: the consolidated Docker gateway is an idempotent singleton.
Gated on a reachable Docker daemon. Uses a unique name so it never Gated on a reachable Docker daemon. Uses a unique name so it never
collides with a real per-host sidecar or a leftover from another run. collides with a real per-host gateway or a leftover from another run.
""" """
from __future__ import annotations from __future__ import annotations
@@ -10,17 +10,17 @@ import secrets
import subprocess import subprocess
import unittest import unittest
from bot_bottle.orchestrator.sidecar import DockerSidecar from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker from tests._docker import skip_unless_docker
IMAGE = "busybox" IMAGE = "busybox"
@skip_unless_docker() @skip_unless_docker()
class TestDockerSidecarIntegration(unittest.TestCase): class TestDockerGatewayIntegration(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self.name = "bot-bottle-orch-sidecar-itest-" + secrets.token_hex(4) self.name = "bot-bottle-orch-gateway-itest-" + secrets.token_hex(4)
self.sc = DockerSidecar(IMAGE, name=self.name) self.sc = DockerGateway(IMAGE, name=self.name)
self.addCleanup(self.sc.stop) self.addCleanup(self.sc.stop)
def _count(self) -> int: def _count(self) -> int:
@@ -1,4 +1,4 @@
"""Integration: DockerSidecar.image_exists reflects real docker state. """Integration: DockerGateway.image_exists reflects real docker state.
Gated on a reachable Docker daemon. Deliberately does NOT build the full Gated on a reachable Docker daemon. Deliberately does NOT build the full
sidecar bundle (a heavy, slow image build) it exercises the `image_exists` sidecar bundle (a heavy, slow image build) it exercises the `image_exists`
@@ -11,14 +11,14 @@ from __future__ import annotations
import subprocess import subprocess
import unittest import unittest
from bot_bottle.orchestrator.sidecar import DockerSidecar from bot_bottle.orchestrator.gateway import DockerGateway
from tests._docker import skip_unless_docker from tests._docker import skip_unless_docker
IMAGE = "busybox" IMAGE = "busybox"
@skip_unless_docker() @skip_unless_docker()
class TestDockerSidecarImageExists(unittest.TestCase): class TestDockerGatewayImageExists(unittest.TestCase):
def test_image_exists_true_for_present_false_for_absent(self) -> None: def test_image_exists_true_for_present_false_for_absent(self) -> None:
# Ensure the tiny image is present (build_if_missing is disabled here # Ensure the tiny image is present (build_if_missing is disabled here
# so this never triggers a Dockerfile.sidecars build). # so this never triggers a Dockerfile.sidecars build).
@@ -26,9 +26,9 @@ class TestDockerSidecarImageExists(unittest.TestCase):
["docker", "pull", IMAGE], ["docker", "pull", IMAGE],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
) )
self.assertTrue(DockerSidecar(IMAGE, dockerfile=None).image_exists()) self.assertTrue(DockerGateway(IMAGE, dockerfile=None).image_exists())
self.assertFalse( self.assertFalse(
DockerSidecar("bot-bottle-nonexistent:doesnotexist", dockerfile=None).image_exists() DockerGateway("bot-bottle-nonexistent:doesnotexist", dockerfile=None).image_exists()
) )
@@ -46,7 +46,7 @@ def _addon() -> EgressAddon:
"""Return a bare EgressAddon with LOG_FULL config and no routes file.""" """Return a bare EgressAddon with LOG_FULL config and no routes file."""
a: EgressAddon = EgressAddon.__new__(EgressAddon) a: EgressAddon = EgressAddon.__new__(EgressAddon)
a.config = Config(routes=(), log=LOG_FULL) a.config = Config(routes=(), log=LOG_FULL)
a.safe_tokens = set() a._safe_tokens = {}
a._supervise_slug = "" a._supervise_slug = ""
a._token_allow_timeout = 300.0 a._token_allow_timeout = 300.0
return a return a
+83 -2
View File
@@ -211,7 +211,7 @@ def _addon(config: Config) -> EgressAddon:
"""Bare EgressAddon with a supplied config and no supervise wiring.""" """Bare EgressAddon with a supplied config and no supervise wiring."""
a: EgressAddon = EgressAddon.__new__(EgressAddon) a: EgressAddon = EgressAddon.__new__(EgressAddon)
a.config = config a.config = config
a.safe_tokens = set() a._safe_tokens = {}
a._supervise_slug = "" a._supervise_slug = ""
a._token_allow_timeout = 300.0 a._token_allow_timeout = 300.0
a.routes_path = "/nonexistent/routes.yaml" a.routes_path = "/nonexistent/routes.yaml"
@@ -222,6 +222,29 @@ def _run_request(addon: EgressAddon, flow: _Flow) -> None:
asyncio.run(addon.request(flow)) # type: ignore[arg-type] asyncio.run(addon.request(flow)) # type: ignore[arg-type]
def _with_client_ip(flow: _Flow, ip: str) -> _Flow:
"""Attach a mitmproxy-style client_conn so consolidated resolution can read
the source IP (`flow.client_conn.peername[0]`)."""
flow.client_conn = types.SimpleNamespace(peername=(ip, 54321)) # type: ignore[attr-defined]
return flow
class _CtxResolver:
"""Fake orchestrator resolver: maps source IP -> bottle id, and grants the
same allow-list to any attributed bottle (unattributed -> deny)."""
def __init__(self, ip_to_bottle: dict[str, str]) -> None:
self._map = ip_to_bottle
def resolve_policy_and_bottle_id(
self, source_ip: str, identity_token: str = "",
) -> tuple[str | None, str | None]:
del identity_token
bottle_id = self._map.get(source_ip)
policy = "routes:\n - host: api.example.com\n" if bottle_id else None
return policy, bottle_id
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Introspection endpoint # Introspection endpoint
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -418,7 +441,8 @@ class TestSuperviseBranch(unittest.TestCase):
with patch.object(_ea_mod, "_sv", _fake_sv("approved")): with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
_run_request(addon, flow) _run_request(addon, flow)
self.assertIsNone(flow.response) # forwarded after approval self.assertIsNone(flow.response) # forwarded after approval
self.assertIn(_OPENAI_KEY, addon.safe_tokens) # Approval lands in the calling bottle's safelist (keyed by slug).
self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("test-bottle"))
def test_operator_rejection_blocks(self) -> None: def test_operator_rejection_blocks(self) -> None:
addon = self._supervised_addon() addon = self._supervised_addon()
@@ -735,5 +759,62 @@ class TestLogFullRequest(unittest.TestCase):
self.assertTrue(any(e.get("event") == "egress_request" for e in logged)) self.assertTrue(any(e.get("event") == "egress_request" for e in logged))
class TestSuperviseMultiTenant(unittest.TestCase):
"""Consolidated gateway: supervise proposals + the DLP safelist are keyed
per bottle, resolved by source IP (PRD 0070)."""
def _consolidated_addon(self) -> EgressAddon:
# Static config is empty; the resolver supplies each bottle's config.
addon = _addon(Config(routes=()))
addon._resolver = cast(Any, _CtxResolver({"10.0.0.1": "bottle-a", "10.0.0.2": "bottle-b"}))
addon._token_allow_timeout = 0.05
return addon
def test_approval_is_scoped_to_the_calling_bottle(self) -> None:
addon = self._consolidated_addon()
# bottle-a (10.0.0.1) sends the token; the operator approves.
flow = _with_client_ip(
_Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
"10.0.0.1",
)
with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
_run_request(addon, flow)
self.assertIsNone(flow.response) # forwarded after approval
# The approval lands ONLY in bottle-a's safelist — never bottle-b's.
# A global set here would be the cross-tenant leak this slice closes.
self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-a"))
self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-b"))
def test_proposal_is_attributed_to_the_source_ip_bottle(self) -> None:
addon = self._consolidated_addon()
seen: list[str] = []
fake = _fake_sv("approved")
def _capture(**kw: Any) -> Any:
seen.append(kw["bottle_slug"])
return types.SimpleNamespace(id="p")
fake.Proposal = types.SimpleNamespace(new=_capture)
flow = _with_client_ip(
_Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
"10.0.0.2",
)
with patch.object(_ea_mod, "_sv", fake):
_run_request(addon, flow)
self.assertEqual(["bottle-b"], seen) # proposal keyed by the resolved bottle
def test_unattributed_source_ip_cannot_supervise(self) -> None:
addon = self._consolidated_addon()
# 10.9.9.9 is not in the resolver map -> deny-all config, empty slug.
flow = _with_client_ip(
_Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
"10.9.9.9",
)
with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
_run_request(addon, flow)
self.assertIsNotNone(flow.response) # blocked (no route, no supervise)
self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for(""))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+49 -2
View File
@@ -1,10 +1,10 @@
"""Unit: resolve_client_config — fail-closed per-client egress config (PRD 0070).""" """Unit: fail-closed per-client egress resolution — config + context (PRD 0070)."""
from __future__ import annotations from __future__ import annotations
import unittest import unittest
from bot_bottle.egress_addon_core import resolve_client_config from bot_bottle.egress_addon_core import resolve_client_config, resolve_client_context
from bot_bottle.policy_resolver import PolicyResolveError from bot_bottle.policy_resolver import PolicyResolveError
@@ -49,5 +49,52 @@ class TestResolveClientConfig(unittest.TestCase):
self.assertEqual(("10.243.0.1", "tok"), r.calls[0]) self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
class _FakeContextResolver:
def __init__(
self, policy: str | None = None, bottle_id: str | None = None, raises: bool = False,
) -> None:
self._policy = policy
self._bottle_id = bottle_id
self._raises = raises
def resolve_policy_and_bottle_id(
self, source_ip: str, identity_token: str = "",
) -> tuple[str | None, str | None]:
if self._raises:
raise PolicyResolveError("orchestrator down")
return self._policy, self._bottle_id
class TestResolveClientContext(unittest.TestCase):
def test_returns_config_and_bottle_id(self) -> None:
cfg, slug = resolve_client_context(
_FakeContextResolver(policy="routes:\n - host: example.com\n", bottle_id="b1"),
"10.243.0.1",
)
self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
self.assertEqual("b1", slug)
def test_unattributed_denies_and_empty_slug(self) -> None:
cfg, slug = resolve_client_context(
_FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
)
self.assertEqual((), cfg.routes)
self.assertEqual("", slug) # no bottle → supervise unavailable
def test_resolver_error_denies_and_empty_slug(self) -> None:
cfg, slug = resolve_client_context(_FakeContextResolver(raises=True), "10.243.0.1")
self.assertEqual((), cfg.routes)
self.assertEqual("", slug)
def test_unparseable_policy_denies_but_keeps_slug(self) -> None:
# A bad policy denies egress, but the bottle is still attributed (its
# supervise queue is keyed by the id, independent of route parsing).
cfg, slug = resolve_client_context(
_FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
)
self.assertEqual((), cfg.routes)
self.assertEqual("b2", slug)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+42
View File
@@ -0,0 +1,42 @@
"""Unit: shared-gateway source-IP allocation (PRD 0070)."""
from __future__ import annotations
import unittest
from bot_bottle.backend.docker.gateway_net import NoFreeAddressError, next_free_ip
class TestNextFreeIp(unittest.TestCase):
def test_skips_router_and_returns_first_host(self) -> None:
# .1 is docker's router; the first assignable address is .2.
self.assertEqual("172.18.0.2", next_free_ip("172.18.0.0/16", []))
def test_skips_taken_addresses(self) -> None:
# Gateway container holds .2, a live bottle holds .3 -> next is .4.
self.assertEqual(
"172.18.0.4", next_free_ip("172.18.0.0/16", ["172.18.0.2", "172.18.0.3"]),
)
def test_taken_order_does_not_matter(self) -> None:
self.assertEqual(
"172.18.0.2", next_free_ip("172.18.0.0/16", ["172.18.0.3", "172.18.0.5"]),
)
def test_allocation_is_deterministic(self) -> None:
taken = ["172.18.0.2"]
self.assertEqual(next_free_ip("172.18.0.0/16", taken),
next_free_ip("172.18.0.0/16", taken))
def test_raises_when_subnet_exhausted(self) -> None:
# /30: hosts are .1 (router, reserved) and .2; taking .2 leaves none.
with self.assertRaises(NoFreeAddressError):
next_free_ip("10.9.9.0/30", ["10.9.9.2"])
def test_accepts_host_bits_set_cidr(self) -> None:
# A container's inspected address arrives as e.g. 172.18.0.2/16.
self.assertEqual("172.18.0.2", next_free_ip("172.18.0.5/16", []))
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,67 @@
"""Unit: consolidated per-bottle git-gate provisioning render (PRD 0070)."""
from __future__ import annotations
import unittest
from bot_bottle.git_gate_render import (
GitGateUpstream,
git_gate_render_entrypoint,
git_gate_render_provision,
)
def _ups(*names: str) -> tuple[GitGateUpstream, ...]:
return tuple(
GitGateUpstream(
name=n,
upstream_url=f"ssh://git@github.com/x/{n}.git",
upstream_host="github.com",
upstream_port="22",
identity_file="",
known_host_key="",
)
for n in names
)
class TestProvisionRender(unittest.TestCase):
def test_namespaces_repos_and_creds_by_bottle_id(self) -> None:
script = git_gate_render_provision("bottleab12", _ups("foo"))
self.assertIn("repo=/git/bottleab12/${name}.git", script)
self.assertIn("keyfile=/git-gate/creds/bottleab12/${name}-key", script)
self.assertIn("mkdir -p /git/bottleab12", script)
def test_one_init_repo_call_per_upstream(self) -> None:
script = git_gate_render_provision("b1", _ups("foo", "bar"))
calls = [l for l in script.splitlines() if l.startswith("init_repo ")]
self.assertEqual(2, len(calls))
def test_provision_does_not_start_the_daemon(self) -> None:
# The shared gateway already serves; provisioning is init-only.
self.assertNotIn("git daemon", git_gate_render_provision("b1", _ups("foo")))
def test_installs_pre_receive_hook(self) -> None:
script = git_gate_render_provision("b1", _ups("foo"))
self.assertIn("install -m 755 /etc/git-gate/pre-receive", script)
def test_rejects_unsafe_bottle_id(self) -> None:
for bad in ("../etc", "a/b", "a b", "a;rm", ""):
with self.assertRaises(ValueError):
git_gate_render_provision(bad, _ups("foo"))
class TestEntrypointUnchanged(unittest.TestCase):
"""The shared `_git_gate_init_repo_fn` refactor must not alter the
single-tenant daemon entrypoint's output."""
def test_entrypoint_still_single_tenant_flat(self) -> None:
script = git_gate_render_entrypoint(_ups("foo"))
self.assertIn("repo=/git/${name}.git", script) # flat, not namespaced
self.assertIn("keyfile=/git-gate/creds/${name}-key", script)
self.assertIn("--base-path=/git", script)
self.assertIn("exec git daemon", script)
if __name__ == "__main__":
unittest.main()
+67
View File
@@ -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 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): class TestGitHttpBackend(unittest.TestCase):
def test_real_git_push_reaches_bare_repo(self): def test_real_git_push_reaches_bare_repo(self):
from http.server import ThreadingHTTPServer from http.server import ThreadingHTTPServer
@@ -94,6 +105,62 @@ class TestGitHttpBackend(unittest.TestCase):
).strip() ).strip()
self.assertEqual(head, cloned) 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): def test_post_forwards_git_cgi_headers(self):
from http.server import ThreadingHTTPServer from http.server import ThreadingHTTPServer
+65
View File
@@ -0,0 +1,65 @@
"""Unit: resolve_sandbox_root — source-IP-keyed git-gate repo namespace (PRD 0070).
The consolidated git-http backend serves every bottle from one process and
selects each request's repo root by the calling bottle's source IP. This is
the fail-closed selection logic, tested without a live server.
"""
from __future__ import annotations
import unittest
from pathlib import Path
from bot_bottle.git_http_backend import resolve_sandbox_root
from bot_bottle.policy_resolver import PolicyResolveError
_BASE = Path("/git")
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[tuple[str, str]] = []
def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str | None:
self.calls.append((source_ip, identity_token))
if self._raises:
raise PolicyResolveError("orchestrator down")
return self._bottle_id
class TestResolveRepoRoot(unittest.TestCase):
def test_single_tenant_passthrough(self) -> None:
# No resolver → the flat base root, unchanged (legacy per-bottle mode).
self.assertEqual(_BASE, resolve_sandbox_root(None, _BASE, "10.243.0.1"))
def test_attributed_bottle_gets_namespaced_root(self) -> None:
root = resolve_sandbox_root(_FakeResolver(bottle_id="ab12cd34"), _BASE, "10.243.0.1")
self.assertEqual(Path("/git/ab12cd34"), root)
def test_unattributed_denies(self) -> None:
self.assertIsNone(resolve_sandbox_root(_FakeResolver(bottle_id=None), _BASE, "10.243.0.9"))
def test_empty_bottle_id_denies(self) -> None:
self.assertIsNone(resolve_sandbox_root(_FakeResolver(bottle_id=""), _BASE, "10.243.0.1"))
def test_resolver_error_denies(self) -> None:
# Orchestrator unreachable/errored must never serve repos.
self.assertIsNone(resolve_sandbox_root(_FakeResolver(raises=True), _BASE, "10.243.0.1"))
def test_namespace_escape_denies(self) -> None:
# A bottle_id that would climb out of the root is rejected (defense in
# depth — registry ids are token_hex, but never trust the namespace).
self.assertIsNone(
resolve_sandbox_root(_FakeResolver(bottle_id="../etc"), _BASE, "10.243.0.1")
)
def test_forwards_source_ip_and_token(self) -> None:
r = _FakeResolver(bottle_id="b1")
resolve_sandbox_root(r, _BASE, "10.243.0.1", "tok")
self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
if __name__ == "__main__":
unittest.main()
+106
View File
@@ -0,0 +1,106 @@
"""Unit: host-side orchestrator control-plane client (PRD 0070). HTTP mocked."""
from __future__ import annotations
import json
import unittest
import urllib.error
from unittest.mock import MagicMock, patch
from bot_bottle.orchestrator.client import (
OrchestratorClient,
OrchestratorClientError,
RegisteredBottle,
)
_URLOPEN = "bot_bottle.orchestrator.client.urllib.request.urlopen"
def _resp(status: int, payload: object) -> MagicMock:
m = MagicMock()
inner = m.__enter__.return_value
inner.status = status
inner.read.return_value = json.dumps(payload).encode()
return m
def _http_error(code: int, payload: object = None) -> urllib.error.HTTPError:
del payload # the client tolerates an empty error body; keep the signature
return urllib.error.HTTPError("http://x", code, "err", {}, None) # type: ignore[arg-type]
class TestRegister(unittest.TestCase):
def setUp(self) -> None:
self.c = OrchestratorClient("http://orch:8080")
def test_register_returns_id_and_token(self) -> None:
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1", "identity_token": "tok"})):
got = self.c.register_bottle("10.0.0.2", policy="routes: []\n", metadata="{}")
self.assertEqual(RegisteredBottle("b1", "tok"), got)
def test_register_posts_source_ip_and_policy(self) -> None:
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b", "identity_token": "t"})) as m:
self.c.register_bottle("10.0.0.9", policy="P", metadata="M", image_ref="img")
sent = json.loads(m.call_args.args[0].data)
self.assertEqual("10.0.0.9", sent["source_ip"])
self.assertEqual("P", sent["policy"])
self.assertEqual("img", sent["image_ref"])
def test_register_missing_fields_raises(self) -> None:
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1"})):
with self.assertRaises(OrchestratorClientError):
self.c.register_bottle("10.0.0.2")
def test_register_non_2xx_raises(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(400, {"error": "bad"})):
with self.assertRaises(OrchestratorClientError):
self.c.register_bottle("")
class TestTeardown(unittest.TestCase):
def setUp(self) -> None:
self.c = OrchestratorClient("http://orch:8080")
def test_teardown_true_on_success(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})):
self.assertTrue(self.c.teardown_bottle("b1"))
def test_teardown_false_on_404(self) -> None:
# Idempotent: an already-gone bottle is a clean no-op.
with patch(_URLOPEN, side_effect=_http_error(404)):
self.assertFalse(self.c.teardown_bottle("gone"))
def test_teardown_uses_delete(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})) as m:
self.c.teardown_bottle("b1")
self.assertEqual("DELETE", m.call_args.args[0].get_method())
class TestHealthAndPolicy(unittest.TestCase):
def setUp(self) -> None:
self.c = OrchestratorClient("http://orch:8080")
def test_health_true_on_200(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"status": "ok"})):
self.assertTrue(self.c.health())
def test_health_false_when_unreachable(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
self.assertFalse(self.c.health())
def test_set_policy_false_on_404(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(404)):
self.assertFalse(self.c.set_policy("gone", "P"))
def test_set_policy_true_on_success(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"updated": True})):
self.assertTrue(self.c.set_policy("b1", "P"))
def test_unreachable_raises(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
with self.assertRaises(OrchestratorClientError):
self.c.register_bottle("10.0.0.2")
if __name__ == "__main__":
unittest.main()
@@ -118,8 +118,8 @@ class TestDispatch(unittest.TestCase):
status, _ = dispatch(self.orch, "GET", "/health/", b"") status, _ = dispatch(self.orch, "GET", "/health/", b"")
self.assertEqual(200, status) self.assertEqual(200, status)
def test_sidecar_status_unconfigured(self) -> None: def test_gateway_status_unconfigured(self) -> None:
status, payload = dispatch(self.orch, "GET", "/sidecar", b"") status, payload = dispatch(self.orch, "GET", "/gateway", b"")
self.assertEqual(200, status) self.assertEqual(200, status)
self.assertEqual(False, payload["configured"]) self.assertEqual(False, payload["configured"])
@@ -1,29 +1,29 @@
"""Unit tests for the consolidated Docker sidecar (PRD 0070). Docker mocked.""" """Unit tests for the consolidated Docker gateway (PRD 0070). Docker mocked."""
from __future__ import annotations from __future__ import annotations
import unittest import unittest
from unittest.mock import Mock, patch from unittest.mock import Mock, patch
from bot_bottle.orchestrator.sidecar import ( from bot_bottle.orchestrator.gateway import (
SIDECAR_NAME, GATEWAY_NAME,
DockerSidecar, DockerGateway,
SidecarError, GatewayError,
) )
_RUN_DOCKER = "bot_bottle.orchestrator.sidecar.run_docker" _RUN_DOCKER = "bot_bottle.orchestrator.gateway.run_docker"
def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock: def _proc(returncode: int = 0, stdout: str = "", stderr: str = "") -> Mock:
return Mock(returncode=returncode, stdout=stdout, stderr=stderr) return Mock(returncode=returncode, stdout=stdout, stderr=stderr)
class TestDockerSidecar(unittest.TestCase): class TestDockerGateway(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self.sc = DockerSidecar("bot-bottle-sidecars:latest") self.sc = DockerGateway("bot-bottle-sidecars:latest")
def test_default_name(self) -> None: def test_default_name(self) -> None:
self.assertEqual(SIDECAR_NAME, self.sc.name) self.assertEqual(GATEWAY_NAME, self.sc.name)
def test_is_running_reads_docker_ps(self) -> None: def test_is_running_reads_docker_ps(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(stdout=self.sc.name + "\n")): with patch(_RUN_DOCKER, return_value=_proc(stdout=self.sc.name + "\n")):
@@ -51,6 +51,33 @@ class TestDockerSidecar(unittest.TestCase):
self.assertEqual(1, len(runs)) self.assertEqual(1, len(runs))
self.assertIn(self.sc.name, runs[0]) self.assertIn(self.sc.name, runs[0])
self.assertIn("bot-bottle-sidecars:latest", runs[0]) self.assertIn("bot-bottle-sidecars:latest", runs[0])
# Runs on the shared gateway network so agents can reach it by IP.
self.assertEqual(self.sc.network, runs[0][runs[0].index("--network") + 1])
def test_ensure_running_creates_network_when_missing(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
calls.append(argv)
if argv[:3] == ["docker", "network", "inspect"]:
return _proc(returncode=1, stderr="No such network")
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
with patch(_RUN_DOCKER, side_effect=fake):
self.sc.ensure_running()
creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
self.assertEqual([["docker", "network", "create", self.sc.network]], creates)
def test_ensure_running_reuses_existing_network(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
calls.append(argv)
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
with patch(_RUN_DOCKER, side_effect=fake):
self.sc.ensure_running() # network inspect returns 0 → exists
self.assertEqual([], [c for c in calls if c[:3] == ["docker", "network", "create"]])
def test_ensure_running_raises_on_docker_failure(self) -> None: def test_ensure_running_raises_on_docker_failure(self) -> None:
def fake(argv: list[str]) -> Mock: def fake(argv: list[str]) -> Mock:
@@ -61,7 +88,7 @@ class TestDockerSidecar(unittest.TestCase):
return _proc() return _proc()
with patch(_RUN_DOCKER, side_effect=fake): with patch(_RUN_DOCKER, side_effect=fake):
with self.assertRaises(SidecarError): with self.assertRaises(GatewayError):
self.sc.ensure_running() self.sc.ensure_running()
def test_stop_is_idempotent_on_missing(self) -> None: def test_stop_is_idempotent_on_missing(self) -> None:
@@ -71,13 +98,13 @@ class TestDockerSidecar(unittest.TestCase):
def test_stop_raises_on_other_failure(self) -> None: def test_stop_raises_on_other_failure(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="daemon down")): with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="daemon down")):
with self.assertRaises(SidecarError): with self.assertRaises(GatewayError):
self.sc.stop() self.sc.stop()
class TestDockerSidecarBuild(unittest.TestCase): class TestDockerGatewayBuild(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self.sc = DockerSidecar() # defaults to the real bundle image + dockerfile self.sc = DockerGateway() # defaults to the real bundle image + dockerfile
def test_image_exists_reads_docker_inspect(self) -> None: def test_image_exists_reads_docker_inspect(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(returncode=0)): with patch(_RUN_DOCKER, return_value=_proc(returncode=0)):
@@ -109,7 +136,7 @@ class TestDockerSidecarBuild(unittest.TestCase):
self.assertTrue(any(a.endswith("Dockerfile.sidecars") for a in builds[0])) self.assertTrue(any(a.endswith("Dockerfile.sidecars") for a in builds[0]))
def test_ensure_built_noop_when_no_dockerfile(self) -> None: def test_ensure_built_noop_when_no_dockerfile(self) -> None:
sc = DockerSidecar("busybox", dockerfile=None) sc = DockerGateway("busybox", dockerfile=None)
with patch(_RUN_DOCKER) as m: with patch(_RUN_DOCKER) as m:
sc.ensure_built() sc.ensure_built()
m.assert_not_called() m.assert_not_called()
@@ -121,7 +148,7 @@ class TestDockerSidecarBuild(unittest.TestCase):
return _proc(returncode=1, stderr="build boom") return _proc(returncode=1, stderr="build boom")
with patch(_RUN_DOCKER, side_effect=fake): with patch(_RUN_DOCKER, side_effect=fake):
with self.assertRaises(SidecarError): with self.assertRaises(GatewayError):
self.sc.ensure_built() self.sc.ensure_built()
+87
View File
@@ -0,0 +1,87 @@
"""Unit: orchestrator process lifecycle — idempotent singleton (PRD 0070)."""
from __future__ import annotations
import tempfile
import unittest
import urllib.error
from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle.orchestrator.lifecycle import (
OrchestratorProcess,
OrchestratorStartError,
)
from tests.unit import use_bottle_root
_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen"
_POPEN = "bot_bottle.orchestrator.lifecycle.subprocess.Popen"
_SLEEP = "bot_bottle.orchestrator.lifecycle.time.sleep"
_MONOTONIC = "bot_bottle.orchestrator.lifecycle.time.monotonic"
def _health(status: int) -> MagicMock:
"""A urlopen() context-manager whose `.status` is `status`."""
m = MagicMock()
m.__enter__.return_value.status = status
return m
class TestOrchestratorProcess(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.addCleanup(use_bottle_root(Path(self._tmp.name)))
self.p = OrchestratorProcess(port=8099)
def test_url(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.p.url)
def test_is_healthy_true_on_200(self) -> None:
with patch(_URLOPEN, return_value=_health(200)):
self.assertTrue(self.p.is_healthy())
def test_is_healthy_false_on_error(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
self.assertFalse(self.p.is_healthy())
def test_ensure_running_noop_when_already_healthy(self) -> None:
with patch(_URLOPEN, return_value=_health(200)), patch(_POPEN) as popen:
self.assertEqual(self.p.url, self.p.ensure_running())
popen.assert_not_called() # a live control plane is left untouched
def test_ensure_running_spawns_then_waits_for_health(self) -> None:
# First check (before spawn) fails; after spawn the poll succeeds.
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_POPEN) as popen, patch(_SLEEP):
url = self.p.ensure_running()
self.assertEqual(self.p.url, url)
popen.assert_called_once()
def test_ensure_running_raises_on_startup_timeout(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_POPEN), patch(_SLEEP), \
patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
with self.assertRaises(OrchestratorStartError):
self.p.ensure_running(startup_timeout=1.0)
def test_argv_includes_gateway_and_broker(self) -> None:
argv = OrchestratorProcess(port=8099, broker="docker", gateway=True)._argv()
self.assertIn("--gateway", argv)
self.assertIn("bot_bottle.orchestrator", argv)
self.assertEqual("docker", argv[argv.index("--broker") + 1])
def test_argv_omits_gateway_when_disabled(self) -> None:
self.assertNotIn("--gateway", OrchestratorProcess(gateway=False)._argv())
def test_spawn_launches_detached_and_logs(self) -> None:
with patch(_POPEN) as popen:
self.p._spawn()
popen.assert_called_once()
kwargs = popen.call_args.kwargs
self.assertTrue(kwargs["start_new_session"]) # outlives the CLI
self.assertTrue((Path(self._tmp.name) / "orchestrator.log").exists())
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,56 @@
"""Unit: consolidated registration inputs — egress policy round-trip (PRD 0070)."""
from __future__ import annotations
import json
import unittest
from pathlib import Path
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.egress_addon_core import LOG_BLOCKS, load_config
from bot_bottle.orchestrator.registration import (
RegistrationInputs,
egress_policy,
registration_inputs,
)
def _plan(routes: tuple[EgressRoute, ...], *, slug: str = "demo", log: int = 0) -> EgressPlan:
return EgressPlan(
slug=slug,
routes_path=Path("/unused/routes.yaml"),
routes=routes,
token_env_map={},
log=log,
)
class TestEgressPolicy(unittest.TestCase):
def test_policy_round_trips_through_load_config(self) -> None:
# The policy the gateway serves must parse back to the same allow-list
# the per-bottle sidecar applied — moving onto the shared gateway must
# not change a bottle's egress.
routes = (EgressRoute(host="api.example.com"), EgressRoute(host="pypi.org"))
cfg = load_config(egress_policy(_plan(routes)))
self.assertEqual(("api.example.com", "pypi.org"), tuple(r.host for r in cfg.routes))
def test_policy_preserves_log_level(self) -> None:
plan = _plan((EgressRoute(host="x.example.com"),), log=LOG_BLOCKS)
self.assertEqual(LOG_BLOCKS, load_config(egress_policy(plan)).log)
def test_empty_routes_yield_deny_all(self) -> None:
cfg = load_config(egress_policy(_plan(())))
self.assertEqual((), cfg.routes) # no routes → default-deny
class TestRegistrationInputs(unittest.TestCase):
def test_bundles_policy_and_slug_metadata(self) -> None:
plan = _plan((EgressRoute(host="api.example.com"),), slug="my-bot")
inputs = registration_inputs(plan)
self.assertIsInstance(inputs, RegistrationInputs)
self.assertEqual("my-bot", json.loads(inputs.metadata)["slug"])
self.assertEqual(egress_policy(plan), inputs.policy) # same blob egress_policy renders
if __name__ == "__main__":
unittest.main()
+16 -16
View File
@@ -10,7 +10,7 @@ from pathlib import Path
from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker
from bot_bottle.orchestrator.registry import RegistryStore from bot_bottle.orchestrator.registry import RegistryStore
from bot_bottle.orchestrator.service import Orchestrator from bot_bottle.orchestrator.service import Orchestrator
from bot_bottle.orchestrator.sidecar import Sidecar from bot_bottle.orchestrator.gateway import Gateway
class _FailingBroker(LaunchBroker): class _FailingBroker(LaunchBroker):
@@ -24,11 +24,11 @@ class _FailingBroker(LaunchBroker):
pass pass
class _FakeSidecar(Sidecar): class _FakeGateway(Gateway):
"""In-memory sidecar for wiring tests.""" """In-memory gateway for wiring tests."""
def __init__(self) -> None: def __init__(self) -> None:
self.name = "fake-sidecar" self.name = "fake-gateway"
self.ensured = 0 self.ensured = 0
self.built = 0 self.built = 0
self._running = False self._running = False
@@ -120,23 +120,23 @@ class TestOrchestrator(unittest.TestCase):
orch.launch_bottle("10.243.0.9") orch.launch_bottle("10.243.0.9")
self.assertEqual([], self.store.all()) # no orphan self.assertEqual([], self.store.all()) # no orphan
def test_sidecar_unconfigured_by_default(self) -> None: def test_gateway_unconfigured_by_default(self) -> None:
self.assertEqual({"configured": False}, self.orch.sidecar_status()) self.assertEqual({"configured": False}, self.orch.gateway_status())
self.orch.ensure_sidecar() # no-op, must not raise self.orch.ensure_gateway() # no-op, must not raise
def test_sidecar_wired_and_ensured(self) -> None: def test_gateway_wired_and_ensured(self) -> None:
sc = _FakeSidecar() sc = _FakeGateway()
orch = Orchestrator(self.store, self.broker, self.secret, sidecar=sc) orch = Orchestrator(self.store, self.broker, self.secret, gateway=sc)
self.assertEqual( self.assertEqual(
{"configured": True, "name": "fake-sidecar", "running": False}, {"configured": True, "name": "fake-gateway", "running": False},
orch.sidecar_status(), orch.gateway_status(),
) )
orch.ensure_sidecar() orch.ensure_gateway()
self.assertEqual(1, sc.built) # ensure_sidecar builds first, self.assertEqual(1, sc.built) # ensure_gateway builds first,
self.assertEqual(1, sc.ensured) # then runs self.assertEqual(1, sc.ensured) # then runs
self.assertEqual( self.assertEqual(
{"configured": True, "name": "fake-sidecar", "running": True}, {"configured": True, "name": "fake-gateway", "running": True},
orch.sidecar_status(), orch.gateway_status(),
) )
+32 -1
View File
@@ -1,4 +1,4 @@
"""Unit tests for the sidecar-side PolicyResolver (PRD 0070). HTTP mocked.""" """Unit tests for the gateway-side PolicyResolver (PRD 0070). HTTP mocked."""
from __future__ import annotations from __future__ import annotations
@@ -71,6 +71,37 @@ class TestPolicyResolver(unittest.TestCase):
self.r.resolve("10.243.0.7") # token optional self.r.resolve("10.243.0.7") # token optional
self.assertEqual("", json.loads(m.call_args.args[0].data)["identity_token"]) self.assertEqual("", json.loads(m.call_args.args[0].data)["identity_token"])
def test_resolve_bottle_id_returns_id(self) -> None:
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})):
self.assertEqual("b1", self.r.resolve_bottle_id("10.243.0.1", "tok"))
def test_resolve_bottle_id_403_is_none_fail_closed(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(403)):
self.assertIsNone(self.r.resolve_bottle_id("10.243.0.9", "tok"))
def test_resolve_bottle_id_missing_field_is_none(self) -> None:
with patch(_URLOPEN, return_value=_resp({"policy": "P"})):
self.assertIsNone(self.r.resolve_bottle_id("10.243.0.1", "tok"))
def test_resolve_bottle_id_error_raises(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(500)):
with self.assertRaises(PolicyResolveError):
self.r.resolve_bottle_id("10.243.0.1", "tok")
def test_resolve_policy_and_bottle_id_one_call(self) -> None:
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})) as m:
self.assertEqual(("P", "b1"), self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"))
self.assertEqual(1, m.call_count) # both from a single /resolve
def test_resolve_policy_and_bottle_id_403_is_none_none(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(403)):
self.assertEqual((None, None), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
with self.assertRaises(PolicyResolveError):
self.r.resolve_policy_and_bottle_id("10.243.0.1")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+53
View File
@@ -6,6 +6,7 @@ import sys
import tempfile import tempfile
import threading import threading
import time import time
import types
import unittest import unittest
from pathlib import Path from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
@@ -629,5 +630,57 @@ class TestHttpEndToEnd(unittest.TestCase):
conn.close() 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__": if __name__ == "__main__":
unittest.main() unittest.main()