Compare commits

...

6 Commits

Author SHA1 Message Date
didericis-codex 288b205a44 test(macos): isolate registration tests from container discovery
test / stage-firecracker-inputs (pull_request) Successful in 2s
test / integration-docker (pull_request) Successful in 6s
tracker-policy-pr / check-pr (pull_request) Successful in 5s
test / unit (pull_request) Successful in 30s
lint / lint (push) Successful in 42s
test / build-infra (pull_request) Successful in 3m18s
test / integration-firecracker (pull_request) Successful in 1m27s
test / coverage (pull_request) Successful in 1m51s
test / publish-infra (pull_request) Has been skipped
2026-07-21 03:27:16 +00:00
didericis 0c1d27b605 fix(egress): the unattributed message must name the token-mismatch cause too
`/resolve` fail-closes on a missing/ambiguous registry row *and* on a
request whose identity token doesn't match. The message named only the
first, so a bottle that was registered correctly but sent no token read as
"not registered" and sent the reader looking for a deregistered bottle.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:27:16 +00:00
didericis 69361114d1 fix(egress): name the real fault when a deny-all is not an allowlist miss
An unattributed bottle, an unreachable orchestrator, and an unparseable
policy all become a deny-all Config, and a deny-all is indistinguishable
from "policy loaded, host not allowed" at the decision point — both are just
"no matching route". So every one of them reported `host X is not in the
bottle's egress.routes allowlist`, which reads as a config problem and sends
the operator hunting for a route that was never missing. Diagnosing a
bricked registration cost hours for exactly this reason.

Carry the structural reason on Config and prefer it in decide(). A genuine
allowlist miss — a policy that loaded and simply lacks the host — keeps the
original wording, so the message now tells the two cases apart.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:27:16 +00:00
didericis e4d53fd360 fix(orchestrator): reap registry rows whose bottle is no longer running
A registry row only ever left the registry two ways: an explicit
teardown_bottle (the launcher's cleanup callback) or the same-IP supersede
sweep in register(). Neither runs when the launching CLI dies hard, so the
row outlives its container.

That orphan is not inert. Source IPs are recycled by the backend's DHCP and
by_source_ip fail-closes on ambiguity, so a leftover row at a reused address
resolves no policy at all for the next bottle that lands there — and a
bottle with no policy denies every host, which surfaces to the agent as
"host X is not in the allowlist" for hosts that were never the problem.

Add reap_absent/reconcile and call it from the macOS launch path before
registering, so each launch self-heals the registry. Restores the invariant
the data plane needs: at most one active row per live address, and none for
a dead one. The second half matters as much as the first — when several rows
claim a *live* address the newest wins and the rest are swept, otherwise a
recycled address stays ambiguous, which is exactly the bricked state.

The host supplies the live set because the orchestrator runs inside the
infra container and cannot see the backend. A grace window exempts rows
younger than it, so reconciliation cannot race a bottle still coming up, and
a reconcile failure is logged rather than blocking an otherwise-fine launch.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:27:16 +00:00
didericis-codex 5e01c28016 test(macos): cover root container exec helper
test / integration-docker (pull_request) Successful in 11s
tracker-policy-pr / check-pr (pull_request) Successful in 20s
test / unit (pull_request) Successful in 32s
test / stage-firecracker-inputs (pull_request) Successful in 3s
test / build-infra (pull_request) Successful in 8m31s
test / integration-firecracker (pull_request) Successful in 1m32s
test / coverage (pull_request) Successful in 1m27s
test / publish-infra (pull_request) Has been skipped
test / stage-firecracker-inputs (push) Successful in 1s
test / integration-docker (push) Successful in 7s
test / unit (push) Successful in 29s
Update Quality Badges / update-badges (push) Failing after 33s
lint / lint (push) Successful in 42s
test / build-infra (push) Successful in 8m37s
test / integration-firecracker (push) Successful in 1m27s
test / coverage (push) Successful in 1m27s
test / publish-infra (push) Successful in 1m48s
2026-07-21 03:12:17 +00:00
didericis 2f8539c2c7 fix(macos): name the gateway instead of addressing it, so bottles survive it moving
test / integration-docker (pull_request) Successful in 14s
test / unit (pull_request) Successful in 38s
tracker-policy-pr / check-pr (pull_request) Successful in 25s
lint / lint (push) Successful in 49s
test / stage-firecracker-inputs (pull_request) Successful in 5s
test / build-infra (pull_request) Successful in 3m31s
test / integration-firecracker (pull_request) Successful in 1m51s
test / coverage (pull_request) Failing after 1m33s
test / publish-infra (pull_request) Has been skipped
The shared gateway's address is DHCP-assigned and changes whenever the infra
container is recreated — a source-hash bump, an image upgrade, a crash. Every
agent-facing URL embedded that address, and the proxy URL reaches the agent
as process environment at `container exec` time. A running process's environ
cannot be rewritten from outside, so a moved gateway stranded every running
bottle permanently: not degraded, unreachable, until relaunched and its agent
session thrown away.

Give the agent a stable name instead. `GATEWAY_HOSTNAME` replaces the address
in the egress proxy URL, NO_PROXY, git-http, and supervise URLs, and resolves
through the bottle's own /etc/hosts. Unlike environ that is a file, so it can
be rewritten inside a container that is already running — which is the whole
point: a gateway that returns at a new address is picked up by live bottles.

Launch writes the entry before anything execs (every agent URL names the
gateway, so it must resolve for the first connection), and re-points every
running bottle once the gateway is up, so one stranded by an earlier restart
re-attaches instead of needing a relaunch.

The write needs root and the agent runs as `node`: the host can repoint a
bottle's gateway name, the agent cannot repoint its own. Keep that asymmetry.

Apple Container 1.0 has no container-name DNS on a user network and
`container run` has no --add-host, so the entry is written by exec after
start rather than declared at run.

Does not address the other half of #443: per-bottle egress auth tokens are
held in memory by the orchestrator and are still lost across a restart, so a
re-attached bottle resolves its policy but not its injected credentials.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-20 23:08:59 -04:00
20 changed files with 898 additions and 27 deletions
@@ -36,9 +36,12 @@ from dataclasses import dataclass
from ...egress import EgressPlan
from ...git_gate import GitGatePlan
from ...orchestrator.client import OrchestratorClient
from ...log import info
from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
from ...orchestrator.registration import registration_inputs
from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
from . import util as container_mod
from .enumerate import CONTAINER_NAME_PREFIX, enumerate_active
from .gateway import GATEWAY_NETWORK
from .gateway_provision import AppleGatewayTransport
from .infra import MacosInfraService, OrchestratorStartError
@@ -89,6 +92,25 @@ def ensure_gateway(
)
def live_source_ips(network: str) -> list[str]:
"""Every running agent container's address on `network`.
The reconciliation input: the orchestrator lives inside the infra
container and cannot enumerate the host's containers, so the host has to
tell it which bottles are actually up. Containers that have not been
assigned an address yet contribute nothing (the read is non-fatal) — the
reap's grace window, not this list, is what protects an in-flight
launch."""
ips: list[str] = []
for agent in enumerate_active():
ip = container_mod.try_container_ipv4_on_network(
f"{CONTAINER_NAME_PREFIX}{agent.slug}", network,
)
if ip:
ips.append(ip)
return ips
def register_agent(
egress_plan: EgressPlan,
git_gate_plan: GitGatePlan,
@@ -103,6 +125,16 @@ def register_agent(
container — it is the attribution key the gateway resolves policy by.
Raises on failure; the caller tears down."""
client = OrchestratorClient(endpoint.orchestrator_url)
# Self-heal before registering: a launcher that died hard (SIGKILL, closed
# terminal, host sleep) never ran its teardown callback, leaving an active
# row with no container. vmnet recycles addresses, so such a row can
# collide with this bottle's — and `by_source_ip` fail-closes on ambiguity,
# which would resolve no policy at all and deny every host. Best-effort: a
# reconciliation failure must not block an otherwise-fine launch.
try:
client.reconcile(live_source_ips(endpoint.network))
except OrchestratorClientError as e:
info(f"registry reconciliation skipped: {e}")
inputs = registration_inputs(egress_plan)
reg = client.register_bottle(
source_ip, image_ref=image_ref, policy=inputs.policy,
@@ -136,6 +168,7 @@ __all__ = [
"GatewayEndpoint",
"LaunchContext",
"ensure_gateway",
"live_source_ips",
"register_agent",
"teardown_consolidated",
"ConsolidatedLaunchError",
@@ -8,7 +8,11 @@ from ...bottle_state import read_metadata
from .. import ActiveAgent
from .infra import INFRA_NAME
_PREFIX = "bot-bottle-"
# The name every agent container carries: `bot-bottle-<slug>`. Exported
# because callers that act on a running bottle (gateway-host rewrites,
# registry reconciliation) have to map an enumerated slug back to a
# container name.
CONTAINER_NAME_PREFIX = "bot-bottle-"
# The shared per-host infra container carries the same prefix as agent
# containers but is infrastructure, not a bottle — one control plane + gateway
# serves every agent, so listing it as an agent would invent one per host.
@@ -26,9 +30,9 @@ def enumerate_active() -> list[ActiveAgent]:
return []
out: list[ActiveAgent] = []
for name in sorted(line.strip() for line in result.stdout.splitlines()):
if not name.startswith(_PREFIX) or name in _INFRA_NAMES:
if not name.startswith(CONTAINER_NAME_PREFIX) or name in _INFRA_NAMES:
continue
slug = name[len(_PREFIX):]
slug = name[len(CONTAINER_NAME_PREFIX):]
metadata = read_metadata(slug)
out.append(ActiveAgent(
backend_name="macos-container",
@@ -0,0 +1,96 @@
"""Stable gateway name for macOS agents, via each bottle's `/etc/hosts`.
The shared gateway's address is assigned by vmnet's DHCP and changes whenever
the infra container is recreated — a source-hash bump, an image upgrade, a
crash. Every agent-facing URL (egress proxy, git-http, supervise) embeds that
address, and the proxy URL reaches the agent as **process environment** at
`container exec` time. A running process's `environ` cannot be rewritten from
outside, so a moved gateway used to strand every running bottle permanently:
not degraded, unreachable, until the bottle was relaunched and its session
thrown away.
So the agent never learns the address. It is given a stable *name*
(`GATEWAY_HOSTNAME`) in every URL, resolved through its own `/etc/hosts`.
Unlike `environ`, that is a file — it can be rewritten inside a container that
is already running, so a gateway that comes back at a new address is picked up
by live bottles instead of orphaning them.
Apple Container 1.0 offers no container-name DNS on a user network (the only
nameserver an agent sees is vmnet's, which does not know container names) and
`container run` has no `--add-host`, so the entry is written by exec after the
container starts.
Writing it needs root, and the agent runs as `node`: the agent therefore
cannot repoint its own gateway name, while the host (which drives `container
exec --user root`) can. That asymmetry is deliberate — keep it.
"""
from __future__ import annotations
from ...log import warn
from . import util as container_mod
from .enumerate import CONTAINER_NAME_PREFIX, enumerate_active
# The name every agent-facing gateway URL uses. Must not collide with a real
# DNS name the agent might resolve; it is bottle-local by construction.
GATEWAY_HOSTNAME = "bot-bottle-gateway"
# Marker so the rewrite is idempotent and only ever touches our own line —
# the rest of /etc/hosts (localhost, the container's own name) is preserved.
_MARKER = "# bot-bottle gateway"
def _rewrite_script(gateway_ip: str) -> str:
"""A shell one-liner that replaces our managed line in `/etc/hosts`.
Rewrites in place via a temp file + `cat` rather than `mv`, so the file
keeps its original inode, ownership, and mode — a bind-mounted or
pre-created `/etc/hosts` must not be replaced by a root-owned 0644 copy
that the runtime then refuses to update.
"""
return (
"set -e; "
f"grep -v '{_MARKER}' /etc/hosts > /tmp/.bb-hosts || true; "
f"printf '%s %s %s\\n' '{gateway_ip}' '{GATEWAY_HOSTNAME}' "
f"'{_MARKER}' >> /tmp/.bb-hosts; "
"cat /tmp/.bb-hosts > /etc/hosts; "
"rm -f /tmp/.bb-hosts"
)
def set_gateway_host(container_name: str, gateway_ip: str) -> None:
"""Point `GATEWAY_HOSTNAME` at `gateway_ip` inside one running container.
Must run before the agent is exec'd: the agent's proxy URL names the
gateway, so the entry has to exist for its first connection. Idempotent —
re-running with the same address is a no-op in effect.
"""
container_mod.exec_container_as_root(
container_name, ["sh", "-c", _rewrite_script(gateway_ip)],
)
def refresh_gateway_host(gateway_ip: str) -> list[str]:
"""Re-point every running bottle at the current gateway address.
Called once the shared gateway is known to be up, so a bottle stranded by
an earlier gateway restart re-attaches instead of needing a relaunch.
Returns the containers updated.
Best-effort per bottle: one container that refuses the write (already
exiting, say) must not stop the others from being repaired, and must not
fail the launch that triggered the sweep.
"""
updated: list[str] = []
for agent in enumerate_active():
name = f"{CONTAINER_NAME_PREFIX}{agent.slug}"
try:
set_gateway_host(name, gateway_ip)
updated.append(name)
# One bad bottle must not stop the sweep, so this is deliberately broad.
except Exception as e: # noqa: BLE001 # pylint: disable=broad-exception-caught
warn(f"could not re-point {name} at the gateway: {e}")
return updated
__all__ = ["GATEWAY_HOSTNAME", "set_gateway_host", "refresh_gateway_host"]
+38 -11
View File
@@ -59,6 +59,11 @@ from ..docker.egress import EGRESS_PORT
from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
from . import util as container_mod
from .bottle import MacosContainerBottle
from .gateway_hosts import (
GATEWAY_HOSTNAME,
refresh_gateway_host,
set_gateway_host,
)
from .bottle_plan import MacosContainerBottlePlan
from .consolidated_launch import (
GatewayEndpoint,
@@ -100,6 +105,11 @@ def launch(
# Step 1: the per-host singletons. Must precede the agent run — its
# proxy env needs the gateway's address at `container run` time.
endpoint = ensure_gateway()
# The gateway's address may have changed since these bottles launched
# (any infra recreate re-runs DHCP). They name the gateway rather than
# address it, so re-pointing /etc/hosts re-attaches them in place
# instead of leaving them stranded until relaunch.
refresh_gateway_host(endpoint.gateway_ip)
# Step 2: mint this bottle's deploy keys, then point it at the SHARED
# gateway's CA + git-http/supervise ports.
@@ -117,6 +127,9 @@ def launch(
# attribution key; `--cap-drop CAP_NET_RAW` at run is what makes it
# unforgeable. Poll: `container run --detach` can return before vmnet's
# DHCP has assigned the address.
# Resolve the gateway name before anything execs: every agent-facing
# URL uses it, so the entry must exist for the first connection.
set_gateway_host(plan.container_name, endpoint.gateway_ip)
source_ip = container_mod.wait_container_ipv4_on_network(
plan.container_name, endpoint.network,
)
@@ -231,13 +244,20 @@ def _stamp_agent_urls(
) -> MacosContainerBottlePlan:
"""Point the agent's git-gate insteadOf rewrites + supervise MCP at the
shared gateway's ports. Both bypass the egress proxy (NO_PROXY covers the
gateway address)."""
gateway name).
Addressed by `GATEWAY_HOSTNAME`, never by IP: these URLs are baked into
the agent's gitconfig and MCP config at provision time, so an address here
would strand the bottle the moment the gateway moved. The name is resolved
per connection through `/etc/hosts`, which stays rewritable while the
bottle runs."""
del endpoint # addressed by name; the address reaches the bottle via /etc/hosts
git_gate_url = (
f"http://{endpoint.gateway_ip}:{_GIT_HTTP_PORT}"
f"http://{GATEWAY_HOSTNAME}:{_GIT_HTTP_PORT}"
if plan.git_gate_plan.upstreams else ""
)
supervise_url = (
f"http://{endpoint.gateway_ip}:{SUPERVISE_PORT}/"
f"http://{GATEWAY_HOSTNAME}:{SUPERVISE_PORT}/"
if plan.supervise_plan is not None else ""
)
return dataclasses.replace(
@@ -247,19 +267,25 @@ def _stamp_agent_urls(
)
def _proxy_url(gateway_ip: str, identity_token: str = "") -> str:
def _proxy_url(identity_token: str = "") -> str:
"""The agent's egress proxy URL. The identity token rides as proxy
credentials — the gateway reads Proxy-Authorization, resolves the
(source_ip, token) pair against the control plane, and strips it before
upstream. Without a valid pair `/resolve` denies the request (#366)."""
upstream. Without a valid pair `/resolve` denies the request (#366).
Names the gateway rather than addressing it: this URL reaches the agent as
process environment, which cannot be rewritten once the agent is running,
so an address baked here is unfixable if the gateway moves."""
cred = f"bottle:{identity_token}@" if identity_token else ""
return f"http://{cred}{gateway_ip}:{EGRESS_PORT}"
return f"http://{cred}{GATEWAY_HOSTNAME}:{EGRESS_PORT}"
def _no_proxy(gateway_ip: str) -> str:
def _no_proxy() -> str:
# git-http + supervise live on the gateway and must NOT go through the
# egress proxy — the agent reaches them directly by its address.
return f"localhost,127.0.0.1,{gateway_ip}"
# egress proxy — the agent reaches them directly by name. Deliberately
# address-free: NO_PROXY is baked into the run-time env and is therefore
# just as unfixable as the proxy URL if the gateway moves.
return f"localhost,127.0.0.1,{GATEWAY_HOSTNAME}"
def _identity_proxy_env(
@@ -276,7 +302,8 @@ def _identity_proxy_env(
`_agent_env_entries`."""
if not identity_token:
return {}
url = _proxy_url(endpoint.gateway_ip, identity_token)
del endpoint # the gateway is named, not addressed
url = _proxy_url(identity_token)
return {
"HTTPS_PROXY": url, "HTTP_PROXY": url,
"https_proxy": url, "http_proxy": url,
@@ -338,7 +365,7 @@ def _agent_env_entries(
# silently dropping attribution for. Without it a process that egresses
# before the exec-time env still fails closed — the agent network is
# host-only, so there is no route off it except the gateway.
no_proxy = _no_proxy(endpoint.gateway_ip)
no_proxy = _no_proxy()
env = [
f"NO_PROXY={no_proxy}",
f"no_proxy={no_proxy}",
@@ -360,6 +360,21 @@ def exec_container(name: str, argv: list[str]) -> None:
)
def exec_container_as_root(name: str, argv: list[str]) -> None:
"""`exec_container`, but as uid 0 inside the container.
For host-driven maintenance the agent itself must not be able to perform —
rewriting `/etc/hosts` to point the gateway name at an address. The agent
runs as `node`, so it cannot repoint its own gateway; the host can.
"""
result = _run_container_op([_CONTAINER, "exec", "--user", "root", name, *argv])
if result.returncode != 0:
die(
f"container exec (root) in {name} failed: "
f"{(result.stderr or '').strip() or '<no stderr>'}"
)
def _run_container_op(cmd: list[str]) -> subprocess.CompletedProcess[str]:
result = subprocess.run(
cmd,
+1
View File
@@ -379,6 +379,7 @@ class EgressAddon:
env,
request_method=flow.request.method,
request_headers=req_headers,
deny_reason=config.deny_reason,
)
if decision.action == "block":
+45 -6
View File
@@ -89,6 +89,14 @@ LOG_FULL = 2 # log block/warn events + full request and response bodies
class Config:
routes: tuple[Route, ...]
log: int = LOG_OFF
# Why this Config is a deny-all, when it is one for a reason *other* than
# the bottle's own policy genuinely not listing the host. A deny-all is
# indistinguishable from "policy loaded, host not allowed" at the decision
# point — both are simply "no matching route" — so without this the
# operator sees `host X is not in the allowlist` and goes hunting for a
# missing route that was never the problem. Empty for a normally-parsed
# policy; `decide` prefers it over the allowlist wording when set.
deny_reason: str = ""
@dataclass(frozen=True)
@@ -405,16 +413,40 @@ class PolicyResolverLike(typing.Protocol):
...
# Deny-all explanations. Each names the *actual* failure so an operator isn't
# sent looking for a missing egress route when the bottle never had a policy
# to begin with — the failure mode that made a bricked registration read like
# a misconfigured allowlist.
DENY_UNATTRIBUTED = (
"egress: this request was not attributed to any bottle, so no egress "
"policy applies and every host is denied. Either the bottle's registry "
"row is missing/ambiguous (torn down, or another bottle claimed its "
"source IP), or the request carried no matching identity token — check "
"that the caller's proxy URL includes it. This is not an allowlist problem."
)
DENY_UNPARSEABLE = (
"egress: this bottle's egress policy could not be parsed, so it is being "
"treated as deny-all. Fix the bottle's egress.routes; every host is denied "
"until it loads."
)
DENY_RESOLVER_ERROR = (
"egress: the orchestrator could not be reached to resolve this bottle's "
"egress policy, so every host is denied (fail-closed). Check that the "
"control plane is up; this is not an allowlist problem."
)
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)."""
blocked). Each deny-all carries the reason it is one, so the block message
names the real fault instead of blaming the allowlist."""
if not policy:
return Config(routes=()) # unattributed or empty → deny-all
return Config(routes=(), deny_reason=DENY_UNATTRIBUTED)
try:
return load_config(policy)
except ValueError:
return Config(routes=()) # unparseable policy → deny
return Config(routes=(), deny_reason=DENY_UNPARSEABLE)
def resolve_client_config(
@@ -428,7 +460,7 @@ def resolve_client_config(
try:
policy = resolver.resolve(client_ip, identity_token)
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return Config(routes=()) # orchestrator unreachable/errored → deny
return Config(routes=(), deny_reason=DENY_RESOLVER_ERROR)
return _config_from_policy(policy)
@@ -457,7 +489,7 @@ def resolve_client_context(
client_ip, identity_token,
)
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return Config(routes=()), "", {} # orchestrator unreachable/errored → deny
return Config(routes=(), deny_reason=DENY_RESOLVER_ERROR), "", {}
return _config_from_policy(policy), (bottle_id or ""), tokens
@@ -572,12 +604,16 @@ def decide(
*,
request_method: str = "GET",
request_headers: typing.Mapping[str, str] | None = None,
deny_reason: str = "",
) -> Decision:
"""`deny_reason` is `Config.deny_reason`: when the deny-all came from a
missing/unparseable policy rather than the bottle's own allowlist, report
that instead of implying a route is merely absent."""
route = match_route(routes, request_host)
if route is None:
return Decision(
action="block",
reason=(
reason=deny_reason or (
f"egress: host {request_host!r} is not in the "
f"bottle's egress.routes allowlist. Declare a "
f"route for it or remove the request."
@@ -852,6 +888,9 @@ __all__ = [
"is_git_push_request",
"is_git_fetch_request",
"load_config",
"DENY_UNATTRIBUTED",
"DENY_UNPARSEABLE",
"DENY_RESOLVER_ERROR",
"resolve_client_config",
"resolve_client_context",
"PolicyResolverLike",
+15
View File
@@ -15,6 +15,7 @@ from __future__ import annotations
import json
import urllib.error
import urllib.request
from collections.abc import Iterable
from dataclasses import dataclass
from ..paths import host_control_plane_token
@@ -147,6 +148,20 @@ class OrchestratorClient:
raise OrchestratorClientError(f"teardown {bottle_id}: HTTP {status}")
return True
def reconcile(
self, live_source_ips: Iterable[str], *, grace_seconds: float | None = None,
) -> list[str]:
"""Drop registry rows for bottles that are no longer running
(`POST /reconcile`), returning the reaped bottle ids. `live_source_ips`
is the caller's enumeration of its live bottles — the orchestrator
can't see the backend from inside the infra container."""
body: dict[str, object] = {"live_source_ips": list(live_source_ips)}
if grace_seconds is not None:
body["grace_seconds"] = grace_seconds
payload = self._ok("POST", "/reconcile", body)
reaped = payload.get("reaped")
return [r for r in reaped if isinstance(r, str)] if isinstance(reaped, list) else []
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)."""
+24
View File
@@ -13,6 +13,9 @@ vsock / unix-socket portability caveats):
PUT /bottles/<bottle_id>/policy -> 200 {"updated": true} | 404 (live reload)
body: {"policy"}
DELETE /bottles/<bottle_id> -> 200 {"torn_down": true} | 404 (teardown)
POST /reconcile -> 200 {"reaped": [bottle_id, ...]}
body: {"live_source_ips": [...],
["grace_seconds"]}
POST /attribute -> 200 {"bottle_id"} | 403
POST /resolve -> 200 {"bottle_id","policy"} | 403
body: {"source_ip","identity_token"}
@@ -141,6 +144,27 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
return 200, {"torn_down": True}
return 404, {"error": "no such bottle"}
if method == "POST" and route == "/reconcile":
# Host-driven self-heal: the caller enumerates its live bottles (only
# the host can see the backend) and the orchestrator drops rows for
# every other active bottle. Trusted-caller only — an agent that could
# reach this would be able to unregister its neighbours.
try:
data = _parse_json_object(body)
except ValueError as e:
return 400, {"error": f"invalid JSON: {e}"}
raw_ips = data.get("live_source_ips")
if not isinstance(raw_ips, list):
return 400, {"error": "live_source_ips (list of strings) is required"}
live = [ip for ip in raw_ips if isinstance(ip, str) and ip]
grace = data.get("grace_seconds")
kwargs = (
{"grace_seconds": float(grace)}
if isinstance(grace, (int, float)) and not isinstance(grace, bool)
else {}
)
return 200, {"reaped": orch.reconcile(live, **kwargs)}
if method == "POST" and route == "/attribute":
try:
data = _parse_json_object(body)
+72
View File
@@ -32,6 +32,7 @@ import hmac
import secrets
import sqlite3
import time
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import Path
@@ -42,6 +43,12 @@ from ..paths import host_db_path
# 256 bits of urandom, URL-safe — unguessable per-bottle identity token.
IDENTITY_TOKEN_BYTES = 32
# How recently a row must have been registered to be exempt from
# `reap_absent`. Covers the window between `container run` and the address
# becoming visible to another launch's enumeration, so reconciliation never
# reaps a bottle that is still coming up.
DEFAULT_REAP_GRACE_SECONDS = 120.0
def new_identity_token() -> str:
"""A fresh per-bottle identity token (PRD 0070 attribution defence)."""
@@ -225,6 +232,70 @@ class RegistryStore(DbStore):
).fetchall()
return [_row_to_record(r) for r in rows]
def reap_absent(
self,
live_source_ips: Iterable[str],
*,
grace_seconds: float = DEFAULT_REAP_GRACE_SECONDS,
now: float | None = None,
) -> list[BottleRecord]:
"""Delete active rows whose source IP is not held by a live bottle.
A row only ever leaves the registry two ways: an explicit
`teardown_bottle` (the launcher's cleanup callback) or the supersede
sweep in `register`. Neither runs when the launching CLI dies hard
SIGKILL, a closed terminal, a host sleep/crash so the row outlives
its container. That orphan is not inert: source IPs are recycled by
the backend's DHCP, and `by_source_ip` fail-closes on ambiguity, so a
leftover row at a reused address can brick the *next* bottle that
lands on it (no policy resolved -> every host denied, reported to the
agent as "not in the allowlist"). Reconciling against the live set at
launch keeps the registry from accumulating those landmines.
Restores the invariant the data plane needs: **at most one active row
per live address, and none at all for a dead one.** Two cases, because
a dead bottle's address may already have been handed to a live one:
* no live bottle holds the address every row there is an orphan;
* a live bottle holds it but several rows claim it the newest
registration is authoritative and the rest are orphans, the same
rule `register`'s same-IP supersede sweep applies. Without this
second case a recycled address stays ambiguous, which is exactly
the state that resolves no policy.
`grace_seconds` protects an in-flight launch: registration happens
moments after `container run`, and a concurrent launch's address may
not be visible to the caller's enumeration yet. Rows younger than the
grace window are never reaped, so reconciliation can't race a bottle
that is still coming up. Returns the deleted records."""
live = {ip for ip in live_source_ips if ip}
cutoff = (time.time() if now is None else now) - grace_seconds
with self._connection() as conn:
rows = conn.execute(
"SELECT * FROM orchestrator_bottles WHERE state = 'active'",
).fetchall()
by_ip: dict[str, list[BottleRecord]] = {}
for row in rows:
rec = _row_to_record(row)
by_ip.setdefault(rec.source_ip, []).append(rec)
candidates: list[BottleRecord] = []
for ip, recs in by_ip.items():
if ip not in live:
candidates.extend(recs)
continue
# Keep the newest claim on a live address; supersede the rest.
recs.sort(key=lambda r: r.created_at)
candidates.extend(recs[:-1])
doomed = [r for r in candidates if r.created_at <= cutoff]
for rec in doomed:
conn.execute(
"DELETE FROM orchestrator_bottles WHERE bottle_id = ?",
(rec.bottle_id,),
)
if doomed:
self._chmod()
return doomed
def by_source_ip(self, source_ip: str) -> BottleRecord | None:
"""Network-layer attribution: the single active bottle at this source
IP, or None if unknown or ambiguous (more than one a
@@ -262,4 +333,5 @@ __all__ = [
"new_identity_token",
"default_db_path",
"IDENTITY_TOKEN_BYTES",
"DEFAULT_REAP_GRACE_SECONDS",
]
+30 -1
View File
@@ -13,15 +13,20 @@ Launch lifecycle:
and returns the record. If the broker rejects/fails, the registry entry
is rolled back so a failed launch leaves no orphan.
* `teardown_bottle` sends a signed teardown request, then deregisters.
* `reconcile` sweeps rows whose bottle is no longer running the
self-heal for the teardown paths that never got to run (a hard-killed
launcher), since an orphan row at a recycled source IP bricks the next
bottle that lands on it.
"""
from __future__ import annotations
import json
from collections.abc import Iterable
from datetime import datetime, timezone
from .broker import LaunchBroker, LaunchRequest, sign_request
from .registry import BottleRecord, RegistryStore
from .registry import DEFAULT_REAP_GRACE_SECONDS, BottleRecord, RegistryStore
from .gateway import Gateway
from ..supervise import (
AuditEntry,
@@ -117,6 +122,30 @@ class Orchestrator:
self._tokens.pop(bottle_id, None)
return True
def reconcile(
self,
live_source_ips: Iterable[str],
*,
grace_seconds: float = DEFAULT_REAP_GRACE_SECONDS,
) -> list[str]:
"""Drop registry rows for bottles that are no longer running, and
forget their in-memory egress tokens. Returns the reaped bottle ids.
The caller supplies the live set because only the host can enumerate
its own containers the orchestrator runs *inside* the infra
container and has no view of the backend. Deliberately does not
broker a teardown: the container is already gone, so there is nothing
to stop, and a broker error must not stop the sweep from clearing
the row that would otherwise brick the next bottle at that address.
See `RegistryStore.reap_absent` for why orphans accumulate and why
they are harmful rather than merely untidy."""
reaped = self.registry.reap_absent(
live_source_ips, grace_seconds=grace_seconds)
for rec in reaped:
self._tokens.pop(rec.bottle_id, None)
return [rec.bottle_id for rec in reaped]
def tokens_for(self, bottle_id: str) -> dict[str, str]:
"""The bottle's in-memory egress auth tokens (env_name -> value), or
empty. The gateway injects these per request; they are never
+60 -1
View File
@@ -4,7 +4,14 @@ from __future__ import annotations
import unittest
from bot_bottle.egress_addon_core import resolve_client_config, resolve_client_context
from bot_bottle.egress_addon_core import (
DENY_RESOLVER_ERROR,
DENY_UNATTRIBUTED,
DENY_UNPARSEABLE,
decide,
resolve_client_config,
resolve_client_context,
)
from bot_bottle.policy_resolver import PolicyResolveError
@@ -108,3 +115,55 @@ class TestResolveClientContext(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestDenyReasonNamesTheRealFault(unittest.TestCase):
"""A deny-all must not masquerade as a missing allowlist entry.
Regression: an unregistered bottle resolves no policy, so *every* host is
denied but the block message said `host X is not in the allowlist`,
which reads as a config problem and sends the operator hunting for a route
that was never missing. The structural reason wins over that wording.
"""
def _reason(self, resolver: object, host: str = "chatgpt.com") -> str:
cfg = resolve_client_config(resolver, "10.243.0.1") # type: ignore[arg-type]
return decide(cfg.routes, host, "/v1/x", {}, deny_reason=cfg.deny_reason).reason
def test_unattributed_says_unattributed_not_allowlist(self) -> None:
reason = self._reason(_FakeResolver(result=None))
self.assertEqual(DENY_UNATTRIBUTED, reason)
# The misleading claim is the one that must be gone: the host was
# never "not in the allowlist" — there was no allowlist at all.
self.assertNotIn("is not in the bottle's egress.routes allowlist", reason)
# Both causes must be named. `/resolve` fail-closes on a missing row
# *and* on a token mismatch, and the message pointing only at the row
# sent us hunting for a deregistered bottle that was registered fine.
self.assertIn("registry row", reason)
self.assertIn("identity token", reason)
def test_resolver_error_says_orchestrator_unreachable(self) -> None:
self.assertEqual(DENY_RESOLVER_ERROR, self._reason(_FakeResolver(raises=True)))
def test_unparseable_policy_says_so(self) -> None:
self.assertEqual(
DENY_UNPARSEABLE, self._reason(_FakeResolver(result="routes: notalist\n")))
def test_a_real_allowlist_miss_keeps_the_allowlist_wording(self) -> None:
"""The message only changes for structural deny-alls — a loaded policy
that genuinely lacks the host still points at the allowlist."""
reason = self._reason(_FakeResolver(result='routes:\n - host: "api.example.com"\n'))
self.assertIn("is not in the bottle's egress.routes allowlist", reason)
self.assertIn("chatgpt.com", reason)
def test_allowed_host_is_still_forwarded(self) -> None:
cfg = resolve_client_config(
_FakeResolver(result='routes:\n - host: "api.example.com"\n'), "10.243.0.1")
decision = decide(
cfg.routes, "api.example.com", "/v1/x", {}, deny_reason=cfg.deny_reason)
self.assertEqual("forward", decision.action)
def test_a_parsed_policy_carries_no_deny_reason(self) -> None:
cfg = resolve_client_config(
_FakeResolver(result='routes:\n - host: "api.example.com"\n'), "10.243.0.1")
self.assertEqual("", cfg.deny_reason)
+68 -1
View File
@@ -87,7 +87,8 @@ class TestRegisterAgent(unittest.TestCase):
*, source_ip: str = "192.168.128.9",
):
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.provision_git_gate", provision or Mock()):
patch(f"{_MOD}.provision_git_gate", provision or Mock()), \
patch(f"{_MOD}.live_source_ips", return_value=[]):
return register_agent(
_egress_plan(), _git_plan(),
source_ip=source_ip, endpoint=_endpoint(), image_ref="img:1",
@@ -134,3 +135,69 @@ class TestTeardown(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestLiveSourceIps(unittest.TestCase):
"""The reconciliation input: the host enumerates its own bottles because
the orchestrator, inside the infra container, cannot see the backend."""
def _agents(self, *slugs: str) -> list[Mock]:
return [Mock(slug=s) for s in slugs]
def test_maps_slugs_to_container_addresses(self) -> None:
from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
patch(f"{_MOD}.container_mod.try_container_ipv4_on_network",
side_effect=["10.0.0.1", "10.0.0.2"]) as ip:
got = live_source_ips("net0")
self.assertEqual(["10.0.0.1", "10.0.0.2"], got)
self.assertEqual("bot-bottle-a", ip.call_args_list[0].args[0])
def test_containers_without_an_address_are_skipped(self) -> None:
"""A container that hasn't been given a DHCP address yet contributes
nothing the reap's grace window, not this list, protects it."""
from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
patch(f"{_MOD}.container_mod.try_container_ipv4_on_network",
side_effect=["", "10.0.0.2"]):
self.assertEqual(["10.0.0.2"], live_source_ips("net0"))
class TestRegisterAgentReconciles(unittest.TestCase):
"""Registration self-heals the registry first: an orphan row at a recycled
address makes attribution ambiguous, which resolves no policy at all and
denies every host for the bottle being launched."""
def _register(self, client: Mock) -> None:
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.provision_git_gate"), \
patch(f"{_MOD}.live_source_ips", return_value=["10.0.0.7"]):
register_agent(
_egress_plan(), _git_plan(),
source_ip="10.0.0.7", endpoint=_endpoint(),
)
def test_reconciles_before_registering(self) -> None:
client = _client()
calls: list[str] = []
def _reconcile(*_args: object, **_kwargs: object) -> list[str]:
calls.append("reconcile")
return []
def _register_bottle(*_args: object, **_kwargs: object) -> RegisteredBottle:
calls.append("register")
return RegisteredBottle("b1", "tok")
client.reconcile.side_effect = _reconcile
client.register_bottle.side_effect = _register_bottle
self._register(client)
self.assertEqual(["reconcile", "register"], calls)
client.reconcile.assert_called_once_with(["10.0.0.7"])
def test_a_reconcile_failure_does_not_block_the_launch(self) -> None:
from bot_bottle.orchestrator.client import OrchestratorClientError
client = _client()
client.reconcile.side_effect = OrchestratorClientError("unreachable")
self._register(client)
client.register_bottle.assert_called_once()
@@ -17,6 +17,7 @@ from unittest.mock import patch
from bot_bottle.backend.macos_container.bottle import MacosContainerBottle
from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
from bot_bottle.backend.macos_container.consolidated_launch import GatewayEndpoint
from bot_bottle.backend.macos_container.gateway_hosts import GATEWAY_HOSTNAME
from bot_bottle.backend.macos_container.launch import (
_agent_run_argv,
_identity_proxy_env,
@@ -128,7 +129,13 @@ class TestAgentRunArgv(unittest.TestCase):
"""git-http + supervise live on the gateway and must be reached
directly, not through its own egress proxy."""
entry = next(a for a in self.argv if a.startswith("NO_PROXY="))
self.assertIn("192.168.128.3", entry)
self.assertIn(GATEWAY_HOSTNAME, entry)
def test_no_proxy_names_the_gateway_and_never_addresses_it(self) -> None:
"""NO_PROXY is baked into the run-time env, so an address here is as
unfixable as the proxy URL if the gateway moves."""
entry = next(a for a in self.argv if a.startswith("NO_PROXY="))
self.assertNotIn("192.168.128.3", entry)
def test_forwarded_secrets_stay_off_argv(self) -> None:
"""Bare name → inherited from the run process env, so the value never
@@ -146,7 +153,7 @@ class TestIdentityTokenDelivery(unittest.TestCase):
def test_exec_env_carries_the_token_as_proxy_credentials(self) -> None:
env = _identity_proxy_env(_endpoint(), "s3cret")
self.assertEqual(
"http://bottle:s3cret@192.168.128.3:9099", env["HTTP_PROXY"],
f"http://bottle:s3cret@{GATEWAY_HOSTNAME}:9099", env["HTTP_PROXY"],
)
self.assertEqual(env["HTTP_PROXY"], env["https_proxy"])
@@ -214,7 +221,7 @@ class TestPlanIdentityToken(unittest.TestCase):
self.assertIn("HTTP_PROXY", argv)
self.assertNotIn("s3cret", " ".join(argv))
self.assertEqual(
"http://bottle:s3cret@192.168.128.3:9099", kwargs["env"]["HTTP_PROXY"],
f"http://bottle:s3cret@{GATEWAY_HOSTNAME}:9099", kwargs["env"]["HTTP_PROXY"],
)
+24
View File
@@ -272,6 +272,30 @@ resolver #2
),
)
def test_exec_container_as_root_selects_root_user(self):
completed = util.subprocess.CompletedProcess(
args=[], returncode=0, stdout="", stderr="",
)
with patch.object(util, "_run_container_op", return_value=completed) as run:
util.exec_container_as_root("bot-bottle-demo", ["true"])
run.assert_called_once_with([
"container", "exec", "--user", "root", "bot-bottle-demo", "true",
])
def test_exec_container_as_root_reports_failure(self):
failed = util.subprocess.CompletedProcess(
args=[], returncode=1, stdout="", stderr="permission denied\n",
)
with patch.object(util, "_run_container_op", return_value=failed), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.exec_container_as_root("bot-bottle-demo", ["true"])
die.assert_called_once_with(
"container exec (root) in bot-bottle-demo failed: permission denied",
)
def _completed(stdout: str, returncode: int = 0):
return util.subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr="")
+140
View File
@@ -0,0 +1,140 @@
"""Unit: stable gateway name via each bottle's /etc/hosts (issue #443).
The gateway's address moves whenever the infra container is recreated. Agents
name it instead of addressing it, and the name resolves through `/etc/hosts`
a file, so it stays rewritable while the bottle runs, unlike the `environ` the
proxy URL is delivered in.
"""
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import patch
from bot_bottle.backend.macos_container.gateway_hosts import (
GATEWAY_HOSTNAME,
refresh_gateway_host,
set_gateway_host,
)
_MOD = "bot_bottle.backend.macos_container.gateway_hosts"
class TestSetGatewayHost(unittest.TestCase):
def _script(self, exec_root: object) -> str:
argv = exec_root.call_args.args[1] # type: ignore[attr-defined]
self.assertEqual(["sh", "-c"], argv[:2])
return argv[2]
def test_writes_the_address_against_the_stable_name(self) -> None:
with patch(f"{_MOD}.container_mod.exec_container_as_root") as ex:
set_gateway_host("bot-bottle-demo", "192.168.128.19")
self.assertEqual("bot-bottle-demo", ex.call_args.args[0])
script = self._script(ex)
self.assertIn("192.168.128.19", script)
self.assertIn(GATEWAY_HOSTNAME, script)
def test_runs_as_root_so_the_agent_cannot_repoint_itself(self) -> None:
"""The agent runs as `node`. If it could rewrite /etc/hosts it could
aim its own gateway name elsewhere, so the write must go through the
root-only helper."""
with patch(f"{_MOD}.container_mod.exec_container_as_root") as ex:
set_gateway_host("bot-bottle-demo", "10.0.0.1")
ex.assert_called_once()
def test_is_idempotent_by_removing_its_own_line_first(self) -> None:
"""Re-pointing must replace the managed entry, not append a second one
two entries for the same name would resolve by luck of ordering."""
with patch(f"{_MOD}.container_mod.exec_container_as_root") as ex:
set_gateway_host("bot-bottle-demo", "10.0.0.1")
self.assertIn("grep -v", self._script(ex))
def test_preserves_the_rest_of_the_hosts_file(self) -> None:
"""localhost and the container's own name must survive the rewrite."""
with patch(f"{_MOD}.container_mod.exec_container_as_root") as ex:
set_gateway_host("bot-bottle-demo", "10.0.0.1")
script = self._script(ex)
# Filter-and-append, never a truncating write of just our line.
self.assertIn("/etc/hosts >", script)
self.assertIn(">> /tmp/.bb-hosts", script)
def test_keeps_the_original_inode(self) -> None:
"""`cat >` rather than `mv`: a pre-created /etc/hosts must keep its
ownership and mode, not be replaced by a root-owned copy."""
script = None
with patch(f"{_MOD}.container_mod.exec_container_as_root") as ex:
set_gateway_host("bot-bottle-demo", "10.0.0.1")
script = self._script(ex)
self.assertIn("cat /tmp/.bb-hosts > /etc/hosts", script)
self.assertNotIn("mv ", script)
class TestRefreshGatewayHost(unittest.TestCase):
"""The re-attach sweep: bottles stranded by an earlier gateway restart get
re-pointed in place instead of needing a relaunch."""
def _agents(self, *slugs: str) -> list[SimpleNamespace]:
return [SimpleNamespace(slug=s) for s in slugs]
def test_repoints_every_running_bottle(self) -> None:
with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
patch(f"{_MOD}.set_gateway_host") as setter:
updated = refresh_gateway_host("192.168.128.19")
self.assertEqual(["bot-bottle-a", "bot-bottle-b"], updated)
self.assertEqual(
[("bot-bottle-a", "192.168.128.19"), ("bot-bottle-b", "192.168.128.19")],
[c.args for c in setter.call_args_list],
)
def test_one_failing_bottle_does_not_stop_the_sweep(self) -> None:
"""A container that is already exiting must not block the repair of
its neighbours, nor fail the launch that triggered the sweep."""
def _flaky(name: str, _ip: str) -> None:
if name == "bot-bottle-a":
raise RuntimeError("container is exiting")
with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
patch(f"{_MOD}.set_gateway_host", side_effect=_flaky), \
patch(f"{_MOD}.warn") as warn:
updated = refresh_gateway_host("10.0.0.1")
self.assertEqual(["bot-bottle-b"], updated)
warn.assert_called_once()
def test_no_running_bottles_is_a_clean_no_op(self) -> None:
with patch(f"{_MOD}.enumerate_active", return_value=[]), \
patch(f"{_MOD}.set_gateway_host") as setter:
self.assertEqual([], refresh_gateway_host("10.0.0.1"))
setter.assert_not_called()
class TestLaunchWiring(unittest.TestCase):
"""Ordering matters: the name must resolve before anything execs, and the
stranded-bottle sweep must run once the gateway is known to be up."""
def test_launch_sets_the_host_entry_before_reading_the_source_ip(self) -> None:
"""The agent's every URL names the gateway, so the entry has to exist
before the first connection which means before the agent execs."""
import inspect
from bot_bottle.backend.macos_container import launch
src = inspect.getsource(launch)
set_at = src.index("set_gateway_host(plan.container_name")
exec_at = src.index("wait_container_ipv4_on_network")
self.assertLess(set_at, exec_at)
def test_launch_refreshes_stranded_bottles_after_ensure_gateway(self) -> None:
import inspect
from bot_bottle.backend.macos_container import launch
src = inspect.getsource(launch)
ensure_at = src.index("endpoint = ensure_gateway()")
refresh_at = src.index("refresh_gateway_host(endpoint.gateway_ip)")
self.assertLess(ensure_at, refresh_at)
if __name__ == "__main__":
unittest.main()
+29
View File
@@ -104,3 +104,32 @@ class TestHealthAndPolicy(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestReconcile(unittest.TestCase):
def setUp(self) -> None:
self.c = OrchestratorClient("http://orch:8080")
def test_posts_live_ips_and_returns_reaped(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["b1", "b2"]})) as m:
got = self.c.reconcile(["10.0.0.2", "10.0.0.3"])
self.assertEqual(["b1", "b2"], got)
sent = json.loads(m.call_args.args[0].data)
self.assertEqual(["10.0.0.2", "10.0.0.3"], sent["live_source_ips"])
self.assertNotIn("grace_seconds", sent) # omitted -> server default
def test_grace_seconds_is_forwarded_when_given(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": []})) as m:
self.c.reconcile([], grace_seconds=30)
self.assertEqual(30, json.loads(m.call_args.args[0].data)["grace_seconds"])
def test_malformed_reaped_is_tolerated(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["ok", 5, None]})):
self.assertEqual(["ok"], self.c.reconcile([]))
with patch(_URLOPEN, return_value=_resp(200, {})):
self.assertEqual([], self.c.reconcile([]))
def test_error_status_raises(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(500)):
with self.assertRaises(OrchestratorClientError):
self.c.reconcile([])
@@ -8,11 +8,13 @@ from __future__ import annotations
import json
import secrets
import sqlite3
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from contextlib import closing
from pathlib import Path
from unittest.mock import patch
@@ -264,6 +266,7 @@ class TestControlPlaneAuth(unittest.TestCase):
("POST", "/bottles", _body({"source_ip": "10.0.0.1"})),
("PUT", "/bottles/x/policy", _body({"policy": "routes: []"})),
("DELETE", "/bottles/x", b""),
("POST", "/reconcile", _body({"live_source_ips": []})),
("POST", "/resolve", _body({"source_ip": "10.0.0.1", "identity_token": "t"})),
("POST", "/attribute", _body({"source_ip": "10.0.0.1", "identity_token": "t"})),
("GET", "/supervise/proposals", b""),
@@ -387,3 +390,56 @@ class TestDispatchSupervise(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestReconcileRoute(unittest.TestCase):
"""`POST /reconcile` — the host tells the orchestrator which bottles are
actually up, since the orchestrator can't see the backend from inside the
infra container."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.orch = _orchestrator(Path(self._tmp.name) / "r.db")
def tearDown(self) -> None:
self._tmp.cleanup()
def _old(self, source_ip: str) -> str:
rec = self.orch.registry.register(source_ip)
with closing(sqlite3.connect(self.orch.registry.db_path)) as conn:
conn.execute(
"UPDATE orchestrator_bottles SET created_at = 0.0 WHERE bottle_id = ?",
(rec.bottle_id,))
conn.commit()
return rec.bottle_id
def test_reaps_absent_and_reports_ids(self) -> None:
dead = self._old("10.0.0.1")
alive = self._old("10.0.0.2")
status, payload = dispatch(
self.orch, "POST", "/reconcile", _body({"live_source_ips": ["10.0.0.2"]}))
self.assertEqual(200, status)
self.assertEqual([dead], payload["reaped"])
self.assertIsNone(self.orch.registry.get(dead))
self.assertIsNotNone(self.orch.registry.get(alive))
def test_missing_live_source_ips_is_400(self) -> None:
status, _ = dispatch(self.orch, "POST", "/reconcile", _body({}))
self.assertEqual(400, status)
def test_grace_seconds_is_honoured(self) -> None:
"""A grace window wide enough to cover the row protects it."""
self.orch.registry.register("10.0.0.3")
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [], "grace_seconds": 3600}))
self.assertEqual(200, status)
self.assertEqual([], payload["reaped"])
def test_non_string_entries_are_ignored(self) -> None:
dead = self._old("10.0.0.4")
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [None, 7, "10.0.0.9"]}))
self.assertEqual(200, status)
self.assertEqual([dead], payload["reaped"])
+80
View File
@@ -4,7 +4,9 @@ from __future__ import annotations
import sqlite3
import tempfile
import time
import unittest
from contextlib import closing
from pathlib import Path
from bot_bottle.orchestrator.registry import (
@@ -169,3 +171,81 @@ class TestRegistryStore(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestReapAbsent(unittest.TestCase):
"""`reap_absent` — the self-heal for rows whose bottle is gone.
An orphan is not merely untidy: source IPs get recycled, and
`by_source_ip` fail-closes on ambiguity, so a leftover row at a reused
address resolves *no* policy for the next bottle that lands there and
every host it asks for is denied.
"""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db = Path(self._tmp.name) / "registry.db"
self.store = RegistryStore(self.db)
self.store.migrate()
def tearDown(self) -> None:
self._tmp.cleanup()
def _aged(self, source_ip: str, *, age: float) -> BottleRecord:
"""Register a bottle and backdate it past the grace window."""
rec = self.store.register(source_ip)
with closing(sqlite3.connect(self.db)) as conn:
conn.execute(
"UPDATE orchestrator_bottles SET created_at = ? WHERE bottle_id = ?",
(time.time() - age, rec.bottle_id),
)
conn.commit()
return rec
def test_reaps_row_with_no_live_container(self) -> None:
gone = self._aged("10.243.0.9", age=600)
reaped = self.store.reap_absent([])
self.assertEqual([gone.bottle_id], [r.bottle_id for r in reaped])
self.assertIsNone(self.store.get(gone.bottle_id))
def test_keeps_row_whose_ip_is_live(self) -> None:
alive = self._aged("10.243.0.9", age=600)
self.assertEqual([], self.store.reap_absent(["10.243.0.9"]))
self.assertIsNotNone(self.store.get(alive.bottle_id))
def test_grace_window_protects_an_in_flight_launch(self) -> None:
"""A bottle registered moments ago is never reaped, even though the
caller's enumeration didn't see its address yet."""
fresh = self.store.register("10.243.0.10")
self.assertEqual([], self.store.reap_absent([]))
self.assertIsNotNone(self.store.get(fresh.bottle_id))
def test_reaping_the_orphan_unbricks_the_reused_address(self) -> None:
"""The regression this exists for: an orphan at an address that vmnet
later hands to a new bottle makes `by_source_ip` ambiguous, so the new
bottle resolves no policy at all."""
orphan = self._aged("10.243.0.11", age=600)
# A new bottle lands on the recycled address. Force the row in directly
# so `register`'s own supersede sweep doesn't mask the ambiguity.
with closing(sqlite3.connect(self.db)) as conn:
conn.execute(
"INSERT INTO orchestrator_bottles "
"(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "
"VALUES ('newbottle', '10.243.0.11', 'tok-new', 'active', ?, '', 'routes: []')",
(time.time(),),
)
conn.commit()
self.assertIsNone(self.store.by_source_ip("10.243.0.11")) # bricked
reaped = self.store.reap_absent(["10.243.0.11"], grace_seconds=60)
self.assertEqual([orphan.bottle_id], [r.bottle_id for r in reaped])
rec = self.store.by_source_ip("10.243.0.11")
assert rec is not None
self.assertEqual("newbottle", rec.bottle_id)
def test_ignores_empty_ips_in_the_live_set(self) -> None:
gone = self._aged("10.243.0.12", age=600)
self.assertEqual(
[gone.bottle_id],
[r.bottle_id for r in self.store.reap_absent(["", "10.243.0.99"])],
)
+54
View File
@@ -4,8 +4,10 @@ from __future__ import annotations
import json
import secrets
import sqlite3
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from unittest.mock import patch
@@ -304,3 +306,55 @@ class TestOrchestratorSupervise(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestOrchestratorReconcile(unittest.TestCase):
"""`reconcile` — drop rows for bottles that are no longer running."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.secret = secrets.token_bytes(16)
self.db = Path(self._tmp.name) / "r.db"
self.store = RegistryStore(self.db)
self.store.migrate()
self.broker = StubBroker(self.secret)
self.orch = Orchestrator(self.store, self.broker, self.secret)
def tearDown(self) -> None:
self._tmp.cleanup()
def _age_all(self, seconds: float) -> None:
"""Backdate every row past the reap grace window."""
with closing(sqlite3.connect(self.db)) as conn:
conn.execute(
"UPDATE orchestrator_bottles SET created_at = created_at - ?", (seconds,))
conn.commit()
def test_reaps_dead_bottle_and_forgets_its_tokens(self) -> None:
dead = self.orch.launch_bottle("10.243.0.1", tokens={"EGRESS_TOKEN_0": "s3cret"})
live = self.orch.launch_bottle("10.243.0.2", tokens={"EGRESS_TOKEN_0": "keep"})
self._age_all(600)
self.assertEqual([dead.bottle_id], self.orch.reconcile(["10.243.0.2"]))
self.assertIsNone(self.store.get(dead.bottle_id))
self.assertIsNotNone(self.store.get(live.bottle_id))
# The in-memory egress credential goes with the row.
self.assertEqual({}, self.orch.tokens_for(dead.bottle_id))
self.assertEqual({"EGRESS_TOKEN_0": "keep"}, self.orch.tokens_for(live.bottle_id))
def test_reconcile_does_not_broker_a_teardown(self) -> None:
"""The container is already gone — there is nothing to stop, and a
broker error must not stop the sweep clearing the row."""
self.orch.launch_bottle("10.243.0.1")
self._age_all(600)
self.broker.launched.clear()
self.orch.reconcile([])
self.assertEqual([], self.broker.torn_down)
def test_reconcile_keeps_everything_when_all_are_live(self) -> None:
a = self.orch.launch_bottle("10.243.0.1")
b = self.orch.launch_bottle("10.243.0.2")
self._age_all(600)
self.assertEqual([], self.orch.reconcile(["10.243.0.1", "10.243.0.2"]))
self.assertIsNotNone(self.store.get(a.bottle_id))
self.assertIsNotNone(self.store.get(b.bottle_id))