Compare commits

...

9 Commits

Author SHA1 Message Date
didericis 389a2c10ce feat(orchestrator): slice 13c(vii) — stable shared gateway CA
lint / lint (push) Successful in 2m4s
test / unit (pull_request) Successful in 1m2s
test / integration (pull_request) Successful in 20s
test / coverage (pull_request) Successful in 1m13s
The consolidated gateway TLS-intercepts every bottle, so all agents must
trust ONE CA. Rather than host-generate + mount a CA, let the gateway's own
mitmproxy generate its CA into a persistent named volume and extract the
cert for agents — no host openssl, keeps gateway.py lean.

- DockerGateway mounts GATEWAY_CA_VOLUME at /home/mitmproxy/.mitmproxy so the
  self-generated CA survives container recreation (it must not rotate, or
  already-launched agents would stop trusting the gateway).
- ca_cert_pem(): read the CA cert (PEM) out of the running gateway
  (docker exec cat mitmproxy-ca-cert.pem) — the agent installs this into its
  trust store to accept the gateway's bumped certs.

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 22:12:18 -04:00
didericis 9896986810 feat(orchestrator): slice 13c(vi) — consolidated launch sequence (composition)
test / unit (pull_request) Successful in 1m3s
test / integration (pull_request) Successful in 21s
test / coverage (pull_request) Successful in 1m12s
lint / lint (push) Successful in 2m5s
Composes the orchestrator primitives into the register/teardown sequence
that replaces the per-bottle bundle:

  launch_consolidated(egress_plan, git_gate_plan): ensure orchestrator +
    gateway up -> read the gateway network's subnet + the gateway's own
    address -> allocate the bottle a pinned source IP (skip the gateway +
    live bottles) -> register (egress policy blob + slug metadata) ->
    provision its git-gate repos/creds into the gateway. Returns a
    LaunchContext (bottle_id, identity_token, source_ip, network, gateway_ip,
    orchestrator_url) — everything the agent container needs to attach.
    Rolls the registration back if provisioning fails, so no orphan.

  teardown_consolidated(bottle_id): deregister + deprovision (idempotent).

The agent `docker run` itself stays the backend's job (provider
provisioning); this owns the orchestrator-facing wiring so the sequence is
unit-testable — collaborators mocked, next_free_ip + registration_inputs run
for real (IP allocation + policy blob asserted end to end).

pyright 0 errors; pylint 9.83/10; unit suite green (1755 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 22:03:21 -04:00
didericis 20e83258bf feat(git-gate+orchestrator): slice 13c(v) — provision git-gate into the running gateway
lint / lint (push) Successful in 2m16s
test / unit (pull_request) Successful in 1m8s
test / integration (pull_request) Successful in 29s
test / coverage (pull_request) Successful in 1m21s
Places a bottle's git-gate credentials + bare repos into the live shared
gateway container when it registers — the execution side of the 13c(i)
provisioning render.

- provision_git_gate(gateway, bottle_id, plan): mkdir the per-bottle creds
  dir, docker cp each upstream's identity key (+ known_hosts when present)
  into /git-gate/creds/<bottle_id>/, then docker exec the namespaced,
  init-only script from git_gate_render_provision. No-op with no upstreams.
- deprovision_git_gate(gateway, bottle_id): rm the bottle's /git/<id> +
  creds on teardown (idempotent).
- bottle_id is validated (shell/path-safe alphabet) BEFORE any docker cp/rm,
  so a traversal id can never reach a path arg — the creds/repo dirs land in
  docker cp/rm arguments. Registry ids are token_hex; defense in depth.

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:59:12 -04:00
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
17 changed files with 1333 additions and 22 deletions
@@ -0,0 +1,145 @@
"""Consolidated bottle launch sequence for the docker backend (PRD 0070).
Composes the orchestrator primitives into the register/teardown sequence that
replaces the per-bottle sidecar bundle:
1. ensure the orchestrator control plane + shared gateway are up;
2. allocate the bottle a pinned source IP on the gateway network (the
attribution key), skipping the gateway's own address + live bottles;
3. register it (egress policy blob + slug metadata) → bottle id + identity
token;
4. provision its git-gate repos/creds into the running gateway.
It returns a `LaunchContext` with everything the agent container needs to
attach — network, pinned IP, the gateway's address (its proxy target), the
orchestrator URL, and the identity token. The agent `docker run` itself is
the backend's job (it owns provider provisioning); this owns the
orchestrator-facing wiring so that sequence stays testable in isolation.
"""
from __future__ import annotations
from dataclasses import dataclass
from ...docker_cmd import run_docker
from ...egress import EgressPlan
from ...git_gate import GitGatePlan
from ...orchestrator.client import OrchestratorClient
from ...orchestrator.gateway import GATEWAY_NAME, GATEWAY_NETWORK
from ...orchestrator.lifecycle import OrchestratorProcess
from ...orchestrator.registration import registration_inputs
from .gateway_net import next_free_ip
from .gateway_provision import deprovision_git_gate, provision_git_gate
class ConsolidatedLaunchError(RuntimeError):
"""The consolidated register/provision sequence could not complete."""
@dataclass(frozen=True)
class LaunchContext:
"""What the agent container needs to join the shared gateway."""
bottle_id: str
identity_token: str
source_ip: str # the agent's pinned address (attribution key)
network: str # the shared gateway network to attach to
gateway_ip: str # the gateway's address — the agent's proxy target
orchestrator_url: str
def _network_cidr(network: str) -> str:
"""The gateway network's IPv4 subnet, or raise."""
proc = run_docker([
"docker", "network", "inspect",
"--format", "{{range .IPAM.Config}}{{.Subnet}}{{end}}", network,
])
cidr = proc.stdout.strip()
if proc.returncode != 0 or not cidr:
raise ConsolidatedLaunchError(
f"gateway network {network} has no subnet: {proc.stderr.strip()}"
)
return cidr
def _container_ip(name: str, network: str) -> str:
"""A container's IPv4 address on `network`, or raise."""
proc = run_docker([
"docker", "inspect", "--format",
f'{{{{(index .NetworkSettings.Networks "{network}").IPAddress}}}}', name,
])
ip = proc.stdout.strip()
if proc.returncode != 0 or not ip:
raise ConsolidatedLaunchError(
f"gateway {name} has no address on {network}: {proc.stderr.strip()}"
)
return ip
def _taken_ips(client: OrchestratorClient, gateway_ip: str) -> list[str]:
"""Every address already in use on the gateway network: the gateway
container plus every live bottle the registry knows."""
taken = [gateway_ip]
for rec in client.list_bottles():
src = rec.get("source_ip")
if isinstance(src, str) and src:
taken.append(src)
return taken
def launch_consolidated(
egress_plan: EgressPlan,
git_gate_plan: GitGatePlan,
*,
image_ref: str = "",
process: OrchestratorProcess | None = None,
gateway_name: str = GATEWAY_NAME,
network: str = GATEWAY_NETWORK,
) -> LaunchContext:
"""Ensure the orchestrator + gateway are up, allocate + register the
bottle, and provision its git-gate state. Returns the agent's attach
context. Raises `ConsolidatedLaunchError` (or the primitives' own errors)
if any step fails — the caller tears down on failure."""
process = process or OrchestratorProcess()
url = process.ensure_running()
client = OrchestratorClient(url)
cidr = _network_cidr(network)
gateway_ip = _container_ip(gateway_name, network)
source_ip = next_free_ip(cidr, _taken_ips(client, gateway_ip))
inputs = registration_inputs(egress_plan)
reg = client.register_bottle(
source_ip, image_ref=image_ref, policy=inputs.policy, metadata=inputs.metadata,
)
try:
provision_git_gate(gateway_name, reg.bottle_id, git_gate_plan)
except Exception:
# Roll the registration back so a provisioning failure leaves no orphan.
client.teardown_bottle(reg.bottle_id)
raise
return LaunchContext(
bottle_id=reg.bottle_id,
identity_token=reg.identity_token,
source_ip=source_ip,
network=network,
gateway_ip=gateway_ip,
orchestrator_url=url,
)
def teardown_consolidated(
bottle_id: str, *, orchestrator_url: str, gateway_name: str = GATEWAY_NAME,
) -> None:
"""Deregister the bottle and remove its git-gate state from the gateway.
Both steps are idempotent so this is safe from a cleanup trap."""
OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
deprovision_git_gate(gateway_name, bottle_id)
__all__ = [
"LaunchContext",
"launch_consolidated",
"teardown_consolidated",
"ConsolidatedLaunchError",
]
+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"]
@@ -0,0 +1,94 @@
"""Provision one bottle's git-gate state into the running shared gateway
(PRD 0070, docker slice).
The consolidated gateway serves every bottle's repos under `/git/<bottle_id>/`
with per-repo credentials under `/git-gate/creds/<bottle_id>/`. When a bottle
is registered the launcher must place *its* deploy keys + known_hosts into
that per-bottle creds dir and init its bare repos there — so this copies the
credential files into the live gateway container and runs the (namespaced,
init-only) provisioning script produced by `git_gate_render_provision`.
Isolating each bottle's creds dir + repo root by id is what keeps one
bottle's push credentials out of another's repos on the shared gateway.
"""
from __future__ import annotations
import re
from ...docker_cmd import run_docker
from ...git_gate import GitGatePlan, git_gate_render_provision
# bottle ids index the gateway's per-bottle repo + creds dirs; they land in
# `docker cp`/`rm` path arguments, so validate before any path is built (a
# traversal id like "../etc" must never reach the container). Registry ids are
# token_hex — this is defense in depth at the docker boundary.
_SAFE_BOTTLE_ID = re.compile(r"[A-Za-z0-9_-]+")
class GatewayProvisionError(RuntimeError):
"""A git-gate provisioning step against the running gateway failed."""
def _require_safe(bottle_id: str) -> None:
if not _SAFE_BOTTLE_ID.fullmatch(bottle_id):
raise GatewayProvisionError(f"unsafe bottle id {bottle_id!r}")
def _creds_dir(bottle_id: str) -> str:
return f"/git-gate/creds/{bottle_id}"
def _exec(gateway: str, argv: list[str]) -> None:
"""`docker exec` a command in the gateway, raising on non-zero exit."""
proc = run_docker(["docker", "exec", gateway, *argv])
if proc.returncode != 0:
raise GatewayProvisionError(
f"gateway exec {argv!r} failed: {proc.stderr.strip()}"
)
def _cp_into(gateway: str, src: str, dest: str) -> None:
"""`docker cp` a host file into the gateway, raising on non-zero exit."""
proc = run_docker(["docker", "cp", src, f"{gateway}:{dest}"])
if proc.returncode != 0:
raise GatewayProvisionError(
f"gateway cp {src} -> {dest} failed: {proc.stderr.strip()}"
)
def provision_git_gate(gateway: str, bottle_id: str, plan: GitGatePlan) -> None:
"""Place `bottle_id`'s git-gate credentials into the running `gateway`
container and init its bare repos under `/git/<bottle_id>/`.
Copies each upstream's identity key (and known_hosts, when present) into
`/git-gate/creds/<bottle_id>/`, then runs the namespaced provisioning
script. No-op for a bottle with no git upstreams."""
_require_safe(bottle_id)
if not plan.upstreams:
return
creds = _creds_dir(bottle_id)
_exec(gateway, ["mkdir", "-p", creds])
for u in plan.upstreams:
if u.identity_file:
_cp_into(gateway, u.identity_file, f"{creds}/{u.name}-key")
known_hosts = str(u.known_hosts_file)
if known_hosts and known_hosts != ".":
_cp_into(gateway, known_hosts, f"{creds}/{u.name}-known_hosts")
# Init the bare repos + per-repo credential config for this namespace.
script = git_gate_render_provision(bottle_id, plan.upstreams)
_exec(gateway, ["sh", "-c", script])
def deprovision_git_gate(gateway: str, bottle_id: str) -> None:
"""Remove a bottle's repos + creds from the gateway on teardown. Idempotent
— an already-absent namespace is a clean no-op (best effort; a stray dir
can't leak, since attribution is by source IP and the bottle is gone)."""
_require_safe(bottle_id)
run_docker([
"docker", "exec", gateway, "rm", "-rf",
f"/git/{bottle_id}", _creds_dir(bottle_id),
])
__all__ = ["provision_git_gate", "deprovision_git_gate", "GatewayProvisionError"]
+2
View File
@@ -46,6 +46,7 @@ from .git_gate_render import (
git_gate_known_hosts_line,
git_gate_render_access_hook,
git_gate_render_entrypoint,
git_gate_render_provision,
git_gate_render_gitconfig,
git_gate_render_hook,
git_gate_upstreams_for_bottle,
@@ -155,6 +156,7 @@ __all__ = [
"git_gate_render_gitconfig",
"git_gate_known_hosts_line",
"git_gate_render_entrypoint",
"git_gate_render_provision",
"git_gate_render_hook",
"git_gate_render_access_hook",
"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
import re
import shlex
from dataclasses import dataclass
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"
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",
"",
def _git_gate_init_repo_fn(repo_root: str, creds_dir: str) -> list[str]:
"""The `init_repo` shell function, parameterized by the bare-repo root
and the per-bottle creds dir. Single source of the credential-wiring
logic, shared by the single-tenant daemon entrypoint (`/git`,
`/git-gate/creds`) and the consolidated per-bottle provisioning
(`/git/<bottle_id>`, `/git-gate/creds/<bottle_id>`)."""
return [
"init_repo() {",
" name=$1",
" upstream_url=$2",
" keyfile=/git-gate/creds/${name}-key",
" hostsfile=/git-gate/creds/${name}-known_hosts",
f" keyfile={creds_dir}/${{name}}-key",
f" hostsfile={creds_dir}/${{name}}-known_hosts",
"",
# `|| true`: PRD 0018 chunk 3+ bind-mounts these RO from the
# 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",
" fi",
"",
" repo=/git/${name}.git",
f" repo={repo_root}/${{name}}.git",
" if [ ! -d \"$repo\" ]; then",
" git init --bare \"$repo\" >/dev/null",
# --mirror=fetch sets remote.origin.fetch = +refs/*:refs/* so",
# a later `git fetch origin` mirrors the upstream's full ref",
# graph (heads, tags, notes) into the bare repo at canonical",
# paths. It does NOT set remote.origin.mirror=true, so an",
# explicit `git push origin <ref>:<ref>` still pushes one ref.",
# --mirror=fetch sets remote.origin.fetch = +refs/*:refs/* so a later
# `git fetch origin` mirrors the upstream's full ref graph (heads,
# tags, notes) into the bare repo at canonical paths. It does NOT set
# remote.origin.mirror=true, so an explicit `git push origin
# <ref>:<ref>` still pushes one ref.
" git -C \"$repo\" remote add --mirror=fetch origin \"$upstream_url\"",
" fi",
" 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",
" 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:
lines.append(f"init_repo {shlex.quote(u.name)} {shlex.quote(u.upstream_url)}")
lines.extend([
@@ -190,6 +197,35 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
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:
"""The shared pre-receive hook: gitleaks-scan all incoming refs,
then forward each accepted ref to the real upstream (`origin`)
+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",
]
+46 -1
View File
@@ -25,6 +25,19 @@ from ..docker_cmd import run_docker
GATEWAY_NAME = "bot-bottle-orch-gateway"
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"
# mitmproxy's CA dir in the bundle. A persistent named volume here keeps the
# gateway's self-generated CA STABLE across container recreation — every agent
# installs this one CA to trust the shared gateway's TLS interception, so it
# must not rotate when the gateway restarts.
MITMPROXY_HOME = "/home/mitmproxy/.mitmproxy"
GATEWAY_CA_VOLUME = "bot-bottle-gateway-mitmproxy"
GATEWAY_CA_CERT = f"{MITMPROXY_HOME}/mitmproxy-ca-cert.pem"
# The real sidecar-bundle image + its Dockerfile. Kept as a local constant
# rather than imported from backend.docker.sidecar_bundle, which would drag
@@ -77,11 +90,13 @@ class DockerGateway(Gateway):
image_ref: str = GATEWAY_IMAGE,
*,
name: str = GATEWAY_NAME,
network: str = GATEWAY_NETWORK,
build_context: Path | None = None,
dockerfile: str | None = GATEWAY_DOCKERFILE,
) -> None:
self.image_ref = image_ref
self.name = name
self.network = network
self._build_context = build_context or _REPO_ROOT
self._dockerfile = dockerfile
@@ -112,9 +127,22 @@ class DockerGateway(Gateway):
])
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:
if self.is_running():
return
self._ensure_network()
# Clear any stale (stopped) container holding the fixed name, then
# start fresh. `rm --force` on an absent name is a tolerated no-op.
run_docker(["docker", "rm", "--force", self.name])
@@ -122,11 +150,27 @@ class DockerGateway(Gateway):
"docker", "run", "--detach",
"--name", self.name,
"--label", GATEWAY_LABEL,
"--network", self.network,
# Persist the self-generated CA so it survives restarts (agents
# trust it) — see GATEWAY_CA_VOLUME.
"--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
self.image_ref,
])
if proc.returncode != 0:
raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
def ca_cert_pem(self) -> str:
"""The gateway's CA certificate (PEM) that agents install to trust its
TLS interception. Read from the running container; raises if the
gateway hasn't generated it yet (it's created on first mitmproxy
start)."""
proc = run_docker(["docker", "exec", self.name, "cat", GATEWAY_CA_CERT])
if proc.returncode != 0 or not proc.stdout.strip():
raise GatewayError(
f"gateway CA cert not available: {proc.stderr.strip() or 'empty'}"
)
return proc.stdout
def stop(self) -> None:
proc = run_docker(["docker", "rm", "--force", self.name])
if proc.returncode != 0 and "No such container" not in proc.stderr:
@@ -135,5 +179,6 @@ class DockerGateway(Gateway):
__all__ = [
"Gateway", "DockerGateway", "GatewayError",
"GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE",
"GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
"GATEWAY_CA_VOLUME", "GATEWAY_CA_CERT",
]
+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"]
+91
View File
@@ -0,0 +1,91 @@
"""Unit: consolidated launch sequence — compose the orchestrator primitives (PRD 0070)."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from bot_bottle.backend.docker.consolidated_launch import (
launch_consolidated,
teardown_consolidated,
)
from bot_bottle.egress import EgressPlan, EgressRoute
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.orchestrator.client import RegisteredBottle
_MOD = "bot_bottle.backend.docker.consolidated_launch"
def _egress_plan() -> EgressPlan:
return EgressPlan(
slug="demo", routes_path=Path("/x"), routes=(EgressRoute(host="api.example.com"),),
token_env_map={},
)
def _git_plan() -> GitGatePlan:
return GitGatePlan(
slug="demo", entrypoint_script=Path(), hook_script=Path(),
access_hook_script=Path(), upstreams=(),
)
def _client(*, bottles: list[dict[str, object]] | None = None) -> Mock:
c = Mock()
c.list_bottles.return_value = bottles or []
c.register_bottle.return_value = RegisteredBottle("b1", "tok")
return c
class TestLaunchConsolidated(unittest.TestCase):
def _run(self, client: Mock, provision: Mock | None = None):
process = MagicMock()
process.ensure_running.return_value = "http://orch:8080"
with patch(f"{_MOD}._network_cidr", return_value="172.18.0.0/16"), \
patch(f"{_MOD}._container_ip", return_value="172.18.0.2"), \
patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.provision_git_gate", provision or Mock()):
return launch_consolidated(_egress_plan(), _git_plan(), process=process)
def test_allocates_ip_registers_and_provisions(self) -> None:
client = _client()
provision = Mock()
ctx = self._run(client, provision)
# .1 is the router, .2 is the gateway → first bottle gets .3.
self.assertEqual("172.18.0.3", ctx.source_ip)
self.assertEqual("172.18.0.2", ctx.gateway_ip)
self.assertEqual("b1", ctx.bottle_id)
self.assertEqual("tok", ctx.identity_token)
self.assertEqual("http://orch:8080", ctx.orchestrator_url)
# Registered with the source IP + the egress policy blob.
kwargs = client.register_bottle.call_args
self.assertEqual("172.18.0.3", kwargs.args[0])
self.assertIn("api.example.com", kwargs.kwargs["policy"])
provision.assert_called_once()
def test_skips_gateway_and_live_bottle_addresses(self) -> None:
client = _client(bottles=[{"source_ip": "172.18.0.3"}])
ctx = self._run(client)
self.assertEqual("172.18.0.4", ctx.source_ip) # .2 gw, .3 taken → .4
def test_provision_failure_rolls_back_registration(self) -> None:
client = _client()
provision = Mock(side_effect=RuntimeError("provision boom"))
with self.assertRaises(RuntimeError):
self._run(client, provision)
client.teardown_bottle.assert_called_once_with("b1") # no orphan left
class TestTeardownConsolidated(unittest.TestCase):
def test_deregisters_and_deprovisions(self) -> None:
client = Mock()
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.deprovision_git_gate") as deprov:
teardown_consolidated("b1", orchestrator_url="http://orch:8080")
client.teardown_bottle.assert_called_once_with("b1")
deprov.assert_called_once()
if __name__ == "__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()
+111
View File
@@ -0,0 +1,111 @@
"""Unit: git-gate provisioning into the running shared gateway (PRD 0070)."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from bot_bottle.backend.docker.gateway_provision import (
GatewayProvisionError,
deprovision_git_gate,
provision_git_gate,
)
from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
_RUN = "bot_bottle.backend.docker.gateway_provision.run_docker"
def _proc(returncode: int = 0, stderr: str = "") -> Mock:
return Mock(returncode=returncode, stderr=stderr)
def _recorder(calls: list[list[str]]):
"""A run_docker side_effect that records argv and returns success."""
def fake(argv: list[str]) -> Mock:
calls.append(argv)
return _proc()
return fake
def _plan(*upstreams: GitGateUpstream) -> GitGatePlan:
return GitGatePlan(
slug="demo",
entrypoint_script=Path(),
hook_script=Path(),
access_hook_script=Path(),
upstreams=tuple(upstreams),
)
def _up(name: str, *, key: str = "/host/keys/id", known_hosts: str = "") -> GitGateUpstream:
return GitGateUpstream(
name=name,
upstream_url=f"ssh://git@github.com/x/{name}.git",
upstream_host="github.com",
upstream_port="22",
identity_file=key,
known_host_key="",
known_hosts_file=Path(known_hosts) if known_hosts else Path(),
)
class TestProvisionGitGate(unittest.TestCase):
def test_copies_creds_and_runs_namespaced_init(self) -> None:
calls: list[list[str]] = []
with patch(_RUN, side_effect=_recorder(calls)):
provision_git_gate("gw", "bottle1", _plan(_up("foo", known_hosts="/host/kh")))
cps = [c for c in calls if c[:2] == ["docker", "cp"]]
self.assertIn(["docker", "cp", "/host/keys/id", "gw:/git-gate/creds/bottle1/foo-key"], cps)
self.assertIn(
["docker", "cp", "/host/kh", "gw:/git-gate/creds/bottle1/foo-known_hosts"], cps,
)
# The init script runs in the gateway, namespaced under the bottle id.
exec_scripts = [c for c in calls if c[:3] == ["docker", "exec", "gw"] and c[3] == "sh"]
self.assertEqual(1, len(exec_scripts))
self.assertIn("repo=/git/bottle1/${name}.git", exec_scripts[0][-1])
def test_omits_known_hosts_copy_when_absent(self) -> None:
calls: list[list[str]] = []
with patch(_RUN, side_effect=_recorder(calls)):
provision_git_gate("gw", "b1", _plan(_up("foo"))) # no known_hosts
cps = [c for c in calls if c[:2] == ["docker", "cp"]]
self.assertEqual(1, len(cps)) # only the key, not known_hosts
self.assertTrue(cps[0][3].endswith("/foo-key"))
def test_no_upstreams_is_noop(self) -> None:
with patch(_RUN) as m:
provision_git_gate("gw", "b1", _plan())
m.assert_not_called()
def test_raises_on_docker_failure(self) -> None:
with patch(_RUN, return_value=_proc(returncode=1, stderr="boom")):
with self.assertRaises(GatewayProvisionError):
provision_git_gate("gw", "b1", _plan(_up("foo")))
def test_rejects_unsafe_bottle_id_before_any_docker(self) -> None:
with patch(_RUN) as m:
with self.assertRaises(GatewayProvisionError):
provision_git_gate("gw", "../etc", _plan(_up("foo")))
m.assert_not_called() # rejected before a single docker call
class TestDeprovision(unittest.TestCase):
def test_removes_repo_and_creds(self) -> None:
with patch(_RUN, return_value=_proc()) as m:
deprovision_git_gate("gw", "b1")
argv = m.call_args.args[0]
self.assertEqual(["docker", "exec", "gw", "rm", "-rf"], argv[:5])
self.assertIn("/git/b1", argv)
self.assertIn("/git-gate/creds/b1", argv)
def test_rejects_unsafe_bottle_id(self) -> None:
with patch(_RUN) as m:
with self.assertRaises(GatewayProvisionError):
deprovision_git_gate("gw", "a/b")
m.assert_not_called()
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()
+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()
+44
View File
@@ -6,11 +6,15 @@ import unittest
from unittest.mock import Mock, patch
from bot_bottle.orchestrator.gateway import (
GATEWAY_CA_CERT,
GATEWAY_NAME,
DockerGateway,
GatewayError,
)
_CA_PEM = "-----BEGIN CERTIFICATE-----\nMII...\n-----END CERTIFICATE-----\n"
_RUN_DOCKER = "bot_bottle.orchestrator.gateway.run_docker"
@@ -51,6 +55,46 @@ class TestDockerGateway(unittest.TestCase):
self.assertEqual(1, len(runs))
self.assertIn(self.sc.name, 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])
# Persists its CA on a named volume so agents keep trusting it.
self.assertTrue(any("mitmproxy" in a for a in runs[0]))
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_ca_cert_pem_reads_from_container(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(stdout=_CA_PEM)) as m:
self.assertEqual(_CA_PEM, self.sc.ca_cert_pem())
argv = m.call_args.args[0]
self.assertEqual(["docker", "exec", self.sc.name, "cat", GATEWAY_CA_CERT], argv)
def test_ca_cert_pem_raises_when_absent(self) -> None:
with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="No such file")):
with self.assertRaises(GatewayError):
self.sc.ca_cert_pem()
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 fake(argv: list[str]) -> Mock:
+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()