Compare commits
39 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| dcf81ee2ce | |||
| 65af83b441 | |||
| 716928315e | |||
| 0c2d0aca63 | |||
| 610c4173a5 | |||
| 6492cc258a | |||
| d9c8a4645d | |||
| 5537b29bbc | |||
| 90c69fb30b | |||
| 21f3436d03 | |||
| 2f71d48189 | |||
| 1f9b957604 | |||
| 9e2e953de6 | |||
| d9e3b61c37 | |||
| 8a79450b0a | |||
| b6eb728a4d | |||
| 71f40fe528 | |||
| fd7448c6a2 | |||
| 3178453f83 | |||
| 47268f6fd6 | |||
| fc86b0309f | |||
| 21a89c7c1f | |||
| af1d1ef304 | |||
| 4692a92c19 | |||
| 004c530194 | |||
| 713566ad85 | |||
| 47bc627ead | |||
| 13d9f3843d | |||
| 219fd7493f | |||
| 1175e17d4e | |||
| 590f3cebd7 | |||
| 36bb8a30ef | |||
| 124b1f473c | |||
| 7b2098cad0 | |||
| 0e89d6ae5a | |||
| 0fb4a026ac | |||
| 7c57785f48 | |||
| 58e5275e8f | |||
| 3b1b716822 |
@@ -5,7 +5,7 @@
|
||||
# bot-bottle
|
||||
|
||||
[](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
|
||||
[](https://coverage.readthedocs.io/)
|
||||
[](https://coverage.readthedocs.io/)
|
||||
[](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
|
||||
|
||||
**Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.
|
||||
|
||||
@@ -40,7 +40,7 @@ from abc import ABC, abstractmethod
|
||||
from contextlib import AbstractContextManager
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, Generic, Sequence, TypeVar
|
||||
from typing import Any, Generic, Sequence, TypeVar
|
||||
|
||||
from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
|
||||
from ..egress import EgressPlan
|
||||
@@ -54,9 +54,6 @@ from ..workspace import WorkspacePlan, workspace_plan
|
||||
from .print_util import print_multi, visible_agent_env_names
|
||||
from .util import host_skill_dir
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from .freeze import CommitCancelled, Freezer, get_freezer
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class BottleSpec:
|
||||
@@ -575,63 +572,28 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
|
||||
Not called by the launch path or the test suite."""
|
||||
|
||||
|
||||
# _backends is None until the first call to _get_backends(), at which
|
||||
# point all three concrete backend classes are imported and instantiated.
|
||||
# Keeping the imports out of module scope means that importing any
|
||||
# backend sub-module (e.g. `backend.docker.util`) no longer drags the
|
||||
# firecracker and macos-container implementations into memory.
|
||||
#
|
||||
# Tests may replace _backends with a {name: fake} dict via patch.object;
|
||||
# _get_backends() returns the current module-level value as-is when it
|
||||
# is not None, so test fakes take effect without triggering real imports.
|
||||
_backends: dict[str, BottleBackend[Any, Any]] | None = None
|
||||
# Import concrete backend classes AFTER the base types are defined, so
|
||||
# each backend module can pull BottleSpec / BottlePlan / BottleBackend
|
||||
# via `from . import ...` without hitting a partially-initialized module.
|
||||
from .docker import DockerBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
|
||||
from .firecracker import FirecrackerBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
|
||||
from .macos_container import MacosContainerBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
|
||||
|
||||
# Freezer is imported after the backend classes for the same reason:
|
||||
# Freezer.commit_slug constructs ActiveAgent, which must be fully
|
||||
# defined first.
|
||||
from .freeze import CommitCancelled, Freezer, get_freezer # noqa: E402 # pylint: disable=wrong-import-position
|
||||
|
||||
|
||||
def _get_backends() -> dict[str, BottleBackend[Any, Any]]:
|
||||
"""Return the registry of all backend instances, loading lazily on first call."""
|
||||
global _backends # pylint: disable=global-statement
|
||||
if _backends is None:
|
||||
from .docker import DockerBottleBackend
|
||||
from .firecracker import FirecrackerBottleBackend
|
||||
from .macos_container import MacosContainerBottleBackend
|
||||
_backends = {
|
||||
"docker": DockerBottleBackend(),
|
||||
"firecracker": FirecrackerBottleBackend(),
|
||||
"macos-container": MacosContainerBottleBackend(),
|
||||
}
|
||||
return _backends
|
||||
|
||||
|
||||
def __getattr__(name: str) -> Any:
|
||||
"""Lazily surface concrete backend classes and freeze symbols at the
|
||||
package level so existing `from bot_bottle.backend import X` and
|
||||
`patch.object(backend_mod, X, ...)` call-sites keep working without
|
||||
forcing an import of every backend at module-init time."""
|
||||
if name == "DockerBottleBackend":
|
||||
from .docker import DockerBottleBackend
|
||||
globals()[name] = DockerBottleBackend
|
||||
return DockerBottleBackend
|
||||
if name == "FirecrackerBottleBackend":
|
||||
from .firecracker import FirecrackerBottleBackend
|
||||
globals()[name] = FirecrackerBottleBackend
|
||||
return FirecrackerBottleBackend
|
||||
if name == "MacosContainerBottleBackend":
|
||||
from .macos_container import MacosContainerBottleBackend
|
||||
globals()[name] = MacosContainerBottleBackend
|
||||
return MacosContainerBottleBackend
|
||||
if name == "CommitCancelled":
|
||||
from .freeze import CommitCancelled
|
||||
globals()[name] = CommitCancelled
|
||||
return CommitCancelled
|
||||
if name == "Freezer":
|
||||
from .freeze import Freezer
|
||||
globals()[name] = Freezer
|
||||
return Freezer
|
||||
if name == "get_freezer":
|
||||
from .freeze import get_freezer
|
||||
globals()[name] = get_freezer
|
||||
return get_freezer
|
||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||||
# The dict is heterogeneous: each value is a BottleBackend specialized
|
||||
# over its own plan type. Concrete plan types are erased here because
|
||||
# the registry is selected at runtime and the CLI only needs the
|
||||
# unparameterized methods (prepare → plan → launch(plan), cleanup, etc.).
|
||||
_BACKENDS: dict[str, BottleBackend[Any, Any]] = {
|
||||
"docker": DockerBottleBackend(),
|
||||
"firecracker": FirecrackerBottleBackend(),
|
||||
"macos-container": MacosContainerBottleBackend(),
|
||||
}
|
||||
|
||||
|
||||
def get_bottle_backend(
|
||||
@@ -649,11 +611,10 @@ def get_bottle_backend(
|
||||
Dies with a pointer at the known backends if the chosen name
|
||||
isn't implemented."""
|
||||
resolved = name or os.environ.get("BOT_BOTTLE_BACKEND") or _default_backend_name()
|
||||
backends = _get_backends()
|
||||
if resolved not in backends:
|
||||
known = ", ".join(sorted(backends))
|
||||
if resolved not in _BACKENDS:
|
||||
known = ", ".join(sorted(_BACKENDS))
|
||||
die(f"unknown backend {resolved!r}; known backends: {known}")
|
||||
return backends[resolved]
|
||||
return _BACKENDS[resolved]
|
||||
|
||||
|
||||
def _default_backend_name() -> str:
|
||||
@@ -663,17 +624,16 @@ def _default_backend_name() -> str:
|
||||
# `firecracker` binary isn't installed yet: selecting it here routes
|
||||
# start through firecracker's preflight, which prints an install
|
||||
# pointer, instead of silently falling back to docker.
|
||||
from .firecracker import FirecrackerBottleBackend
|
||||
if FirecrackerBottleBackend.is_host_capable():
|
||||
return "firecracker"
|
||||
return "docker"
|
||||
|
||||
|
||||
def known_backend_names() -> tuple[str, ...]:
|
||||
"""Sorted tuple of all backend keys in `_get_backends()`. Used by
|
||||
"""Sorted tuple of all backend keys in `_BACKENDS`. Used by
|
||||
argparse (`--backend` choices) and the dashboard's backend
|
||||
picker."""
|
||||
return tuple(sorted(_get_backends()))
|
||||
return tuple(sorted(_BACKENDS))
|
||||
|
||||
|
||||
def has_backend(name: str) -> bool:
|
||||
@@ -685,10 +645,9 @@ def has_backend(name: str) -> bool:
|
||||
|
||||
Returns False for unknown names so callers can pass
|
||||
arbitrary input without separate validation."""
|
||||
backends = _get_backends()
|
||||
if name not in backends:
|
||||
if name not in _BACKENDS:
|
||||
return False
|
||||
return backends[name].is_available()
|
||||
return _BACKENDS[name].is_available()
|
||||
|
||||
|
||||
def enumerate_active_agents() -> list[ActiveAgent]:
|
||||
@@ -704,11 +663,10 @@ def enumerate_active_agents() -> list[ActiveAgent]:
|
||||
deterministic tiebreaker. Agents with missing metadata
|
||||
(`started_at == ""`) sort first."""
|
||||
out: list[ActiveAgent] = []
|
||||
backends = _get_backends()
|
||||
for name in sorted(backends):
|
||||
if not backends[name].is_available():
|
||||
for name in known_backend_names():
|
||||
if not has_backend(name):
|
||||
continue
|
||||
out.extend(backends[name].enumerate_active())
|
||||
out.extend(_BACKENDS[name].enumerate_active())
|
||||
out.sort(key=lambda a: (a.started_at, a.slug))
|
||||
return out
|
||||
|
||||
|
||||
@@ -98,7 +98,6 @@ def launch_consolidated(
|
||||
git_gate_plan: GitGatePlan,
|
||||
*,
|
||||
image_ref: str = "",
|
||||
tokens: dict[str, str] | None = None,
|
||||
service: OrchestratorService | None = None,
|
||||
gateway_name: str = GATEWAY_NAME,
|
||||
network: str = GATEWAY_NETWORK,
|
||||
@@ -117,8 +116,7 @@ def launch_consolidated(
|
||||
|
||||
inputs = registration_inputs(egress_plan)
|
||||
reg = client.register_bottle(
|
||||
source_ip, image_ref=image_ref, policy=inputs.policy,
|
||||
metadata=inputs.metadata, tokens=tokens,
|
||||
source_ip, image_ref=image_ref, policy=inputs.policy, metadata=inputs.metadata,
|
||||
)
|
||||
try:
|
||||
provision_git_gate(gateway_name, reg.bottle_id, git_gate_plan)
|
||||
|
||||
@@ -36,7 +36,6 @@ from contextlib import ExitStack, contextmanager
|
||||
from pathlib import Path
|
||||
from typing import Callable, Generator
|
||||
|
||||
from ...egress import egress_resolve_token_values
|
||||
from ...git_gate import (
|
||||
provision_git_gate_dynamic_keys,
|
||||
revoke_git_gate_provisioned_keys,
|
||||
@@ -121,17 +120,8 @@ def launch(
|
||||
|
||||
# Step 3: register on the orchestrator + provision this bottle's
|
||||
# git-gate state into the shared gateway; get the agent's attach
|
||||
# context (pinned source IP, gateway address, shared network). The
|
||||
# per-bottle egress auth tokens are resolved from the host env now and
|
||||
# handed to the orchestrator (in memory) for the gateway to inject —
|
||||
# the agent never sees them.
|
||||
effective_env = {**os.environ, **plan.agent_provision.provisioned_env}
|
||||
token_values = egress_resolve_token_values(
|
||||
plan.egress_plan.token_env_map, effective_env,
|
||||
)
|
||||
ctx = launch_consolidated(
|
||||
plan.egress_plan, git_gate_plan, image_ref=plan.image, tokens=token_values,
|
||||
)
|
||||
# context (pinned source IP, gateway address, shared network).
|
||||
ctx = launch_consolidated(plan.egress_plan, git_gate_plan, image_ref=plan.image)
|
||||
stack.callback(
|
||||
teardown_consolidated, ctx.bottle_id, orchestrator_url=ctx.orchestrator_url,
|
||||
)
|
||||
|
||||
@@ -7,9 +7,8 @@ from __future__ import annotations
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from typing import Iterator
|
||||
from typing import Iterable, Iterator
|
||||
|
||||
from ...docker_cmd import run_docker
|
||||
from ...log import die, info
|
||||
# from ...workspace import WorkspacePlan
|
||||
|
||||
@@ -31,7 +30,12 @@ def container_name_candidates(base: str) -> Iterator[str]:
|
||||
def runsc_available() -> bool:
|
||||
"""Return True if the Docker daemon has the gVisor (`runsc`) runtime
|
||||
registered. Called once per prepare; the result lives on the plan."""
|
||||
r = run_docker(["docker", "info", "--format", "{{json .Runtimes}}"])
|
||||
r = subprocess.run(
|
||||
["docker", "info", "--format", "{{json .Runtimes}}"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
return r.returncode == 0 and "runsc" in r.stdout
|
||||
|
||||
|
||||
@@ -45,15 +49,20 @@ def require_docker() -> None:
|
||||
|
||||
|
||||
def image_exists(ref: str) -> bool:
|
||||
return run_docker(["docker", "image", "inspect", ref]).returncode == 0
|
||||
return _silent_run(["docker", "image", "inspect", ref]) == 0
|
||||
|
||||
|
||||
def container_exists(name: str) -> bool:
|
||||
"""Returns True if a container (running or stopped) with the given
|
||||
name exists. Uses `docker ps -a -q -f name=^<name>$` so substring
|
||||
matches don't false-positive."""
|
||||
result = run_docker(["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"])
|
||||
return result.returncode == 0 and bool(result.stdout.strip())
|
||||
result = subprocess.run(
|
||||
["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
return bool(result.stdout.strip())
|
||||
|
||||
|
||||
def force_remove_container(name: str) -> None:
|
||||
@@ -61,7 +70,12 @@ def force_remove_container(name: str) -> None:
|
||||
doesn't — and the rm itself is best-effort (errors swallowed) so
|
||||
this is safe to register as a teardown callback."""
|
||||
if container_exists(name):
|
||||
run_docker(["docker", "rm", "-f", name])
|
||||
subprocess.run(
|
||||
["docker", "rm", "-f", name],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
def docker_exec_root(container: str, argv: list[str]) -> None:
|
||||
@@ -141,10 +155,22 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
|
||||
def commit_container(container_name: str, image_tag: str) -> None:
|
||||
"""Run `docker commit <container_name> <image_tag>` to snapshot the
|
||||
running container's filesystem state as a local Docker image."""
|
||||
result = run_docker(["docker", "commit", container_name, image_tag])
|
||||
result = subprocess.run(
|
||||
["docker", "commit", container_name, image_tag],
|
||||
capture_output=True, text=True, check=False,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
die(
|
||||
f"docker commit {container_name!r} → {image_tag!r} failed: "
|
||||
f"{(result.stderr or '').strip() or '<no stderr>'}"
|
||||
)
|
||||
info(f"committed {container_name!r} → {image_tag!r}")
|
||||
|
||||
|
||||
def _silent_run(cmd: Iterable[str]) -> int:
|
||||
return subprocess.run(
|
||||
list(cmd),
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
check=False,
|
||||
).returncode
|
||||
|
||||
+33
-46
@@ -10,7 +10,6 @@ import json
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import typing
|
||||
from pathlib import Path
|
||||
|
||||
from mitmproxy import http # type: ignore[import-not-found] # pylint: disable=import-error
|
||||
@@ -231,27 +230,21 @@ class EgressAddon:
|
||||
+ "\n"
|
||||
)
|
||||
|
||||
def _resolve_flow(
|
||||
self, flow: http.HTTPFlow,
|
||||
) -> "tuple[Config, str, typing.Mapping[str, str]]":
|
||||
"""The `(Config, supervise slug, env)` to apply to this request.
|
||||
Single-tenant → the static `self.config`, the env slug, and the process
|
||||
env. Consolidated → the calling bottle's Config + bottle id + auth
|
||||
tokens, resolved by source IP in one round-trip (fail-closed to deny-all
|
||||
+ empty slug if unattributed); `env` is the process env overlaid with
|
||||
the bottle's tokens, so upstream-auth injection (and DLP) use *this*
|
||||
bottle's credentials — exactly what the per-bottle sidecar's env did.
|
||||
The identity token, if the agent injected one, is read then stripped so
|
||||
it never leaks upstream."""
|
||||
def _resolve_flow(self, flow: http.HTTPFlow) -> "tuple[Config, str]":
|
||||
"""The `(Config, supervise slug)` to apply to this request. Single-tenant
|
||||
→ the static `self.config` and the env slug. Consolidated → the calling
|
||||
bottle's Config and its bottle id, resolved by source IP in one
|
||||
round-trip (fail-closed to deny-all + empty slug if unattributed). The
|
||||
identity token, if the agent injected one, is read then stripped so it
|
||||
never leaks upstream. The slug keys both the proposal queue and the
|
||||
per-bottle safelist, so an approval only ever affects its own bottle."""
|
||||
if self._resolver is None:
|
||||
return self.config, self._supervise_slug, os.environ
|
||||
return self.config, self._supervise_slug
|
||||
conn = flow.client_conn
|
||||
client_ip = conn.peername[0] if conn and conn.peername else ""
|
||||
token = flow.request.headers.get(IDENTITY_HEADER, "")
|
||||
flow.request.headers.pop(IDENTITY_HEADER, None)
|
||||
config, slug, tokens = resolve_client_context(self._resolver, client_ip, token)
|
||||
env = {**os.environ, **tokens} if tokens else os.environ
|
||||
return config, slug, env
|
||||
return resolve_client_context(self._resolver, client_ip, token)
|
||||
|
||||
async def request(self, flow: http.HTTPFlow) -> None:
|
||||
request_path, _, query = flow.request.path.partition("?")
|
||||
@@ -260,14 +253,14 @@ class EgressAddon:
|
||||
self._serve_introspection(flow, request_path)
|
||||
return
|
||||
|
||||
config, slug, env = self._resolve_flow(flow)
|
||||
config, slug = self._resolve_flow(flow)
|
||||
|
||||
# DLP outbound scan BEFORE stripping auth — catches tokens the
|
||||
# agent tried to smuggle in any header, path, query param, or body.
|
||||
# Hostname is included to catch DNS-tunnelling exfiltration attempts.
|
||||
route = match_route(config.routes, flow.request.pretty_host)
|
||||
if route is not None:
|
||||
if not await self._handle_outbound_dlp(flow, route, slug, env):
|
||||
if not await self._handle_outbound_dlp(flow, route, slug):
|
||||
return
|
||||
# The redact policy may have rewritten the request line; recompute
|
||||
# the path/query the git checks below rely on.
|
||||
@@ -306,7 +299,7 @@ class EgressAddon:
|
||||
config.routes,
|
||||
flow.request.pretty_host,
|
||||
request_path,
|
||||
env,
|
||||
os.environ,
|
||||
request_method=flow.request.method,
|
||||
request_headers=req_headers,
|
||||
)
|
||||
@@ -332,12 +325,10 @@ class EgressAddon:
|
||||
flow: http.HTTPFlow,
|
||||
route: Route,
|
||||
slug: str,
|
||||
env: "typing.Mapping[str, str]",
|
||||
) -> bool:
|
||||
"""Scan the outbound request and apply the route's on-match policy
|
||||
(PRD 0062). `env` is the per-bottle env overlay (process env + this
|
||||
bottle's tokens) used for DLP detection. Returns True if the request may
|
||||
be forwarded, False if a 403 response has been written to `flow`.
|
||||
(PRD 0062). Returns True if the request may be forwarded, False if a
|
||||
403 response has been written to `flow`.
|
||||
|
||||
Loops so the supervise policy can re-scan after each approval — a
|
||||
second, un-approved token in the same request is still caught."""
|
||||
@@ -354,7 +345,7 @@ class EgressAddon:
|
||||
flow.request.pretty_host, request_path, query, headers, "",
|
||||
)
|
||||
result = scan_outbound(
|
||||
route, scan_text, env,
|
||||
route, scan_text, os.environ,
|
||||
safe_tokens=self._safe_tokens_for(slug), crlf_text=crlf_text,
|
||||
)
|
||||
if result is None or result.severity != "block":
|
||||
@@ -365,7 +356,7 @@ class EgressAddon:
|
||||
# redact scrubs every detection (tokens and structural CRLF) and
|
||||
# forwards; it fails closed only if a match survives the scrub.
|
||||
if policy == ON_MATCH_REDACT:
|
||||
if self._redact_outbound(flow, route, env):
|
||||
if self._redact_outbound(flow, route):
|
||||
if self.config.log >= LOG_BLOCKS:
|
||||
sys.stderr.write(json.dumps({
|
||||
"event": "egress_redacted",
|
||||
@@ -393,31 +384,29 @@ class EgressAddon:
|
||||
if not self._supervise_available(slug):
|
||||
self._block_dlp(flow, result)
|
||||
return False
|
||||
approved = await self._supervise_token_block(flow, request_path, result, slug, env)
|
||||
approved = await self._supervise_token_block(flow, request_path, result, slug)
|
||||
if not approved:
|
||||
return False # _supervise_token_block wrote the 403 response
|
||||
# loop: the approved value is now in safe_tokens; re-scan.
|
||||
|
||||
def _redact_outbound(
|
||||
self, flow: http.HTTPFlow, route: Route, env: "typing.Mapping[str, str]",
|
||||
) -> bool:
|
||||
def _redact_outbound(self, flow: http.HTTPFlow, route: Route) -> bool:
|
||||
"""Scrub detected tokens (and CRLF injection sequences) from the mutable
|
||||
request surfaces (body, headers, path/query) and re-scan. `env` is the
|
||||
per-bottle env overlay. Returns True if the request is now clean; False
|
||||
if a block-severity match remains on a surface redaction cannot rewrite
|
||||
(the hostname) so the caller fails closed."""
|
||||
request surfaces (body, headers, path/query) and re-scan. Returns True
|
||||
if the request is now clean; False if a block-severity match remains on
|
||||
a surface redaction cannot rewrite (the hostname) so the caller fails
|
||||
closed."""
|
||||
body = flow.request.get_text(strict=False)
|
||||
if body:
|
||||
redacted_body = redact_tokens(body, env=env)
|
||||
redacted_body = redact_tokens(body, env=os.environ)
|
||||
if redacted_body != body:
|
||||
flow.request.text = redacted_body
|
||||
for name, value in list(flow.request.headers.items()):
|
||||
if name.lower() == "host":
|
||||
continue # routing-critical; never a legitimate token
|
||||
redacted = strip_crlf(redact_tokens(value, env=env))
|
||||
redacted = strip_crlf(redact_tokens(value, env=os.environ))
|
||||
if redacted != value:
|
||||
flow.request.headers[name] = redacted
|
||||
redacted_path = strip_crlf(redact_tokens(flow.request.path, env=env))
|
||||
redacted_path = strip_crlf(redact_tokens(flow.request.path, env=os.environ))
|
||||
if redacted_path != flow.request.path:
|
||||
flow.request.path = redacted_path
|
||||
|
||||
@@ -430,7 +419,7 @@ class EgressAddon:
|
||||
crlf_text = build_outbound_scan_text(
|
||||
flow.request.pretty_host, request_path, query, headers, "",
|
||||
)
|
||||
result = scan_outbound(route, scan_text, env, crlf_text=crlf_text)
|
||||
result = scan_outbound(route, scan_text, os.environ, crlf_text=crlf_text)
|
||||
return result is None or result.severity != "block"
|
||||
|
||||
async def _supervise_token_block(
|
||||
@@ -439,22 +428,20 @@ class EgressAddon:
|
||||
request_path: str,
|
||||
result: ScanResult,
|
||||
slug: str,
|
||||
env: "typing.Mapping[str, str]",
|
||||
) -> bool:
|
||||
"""Route a token DLP block to the operator's supervisor queue and wait.
|
||||
|
||||
`slug` attributes the proposal to the calling bottle (its own queue +
|
||||
safelist) — in the shared gateway this is what keeps one bottle's
|
||||
approval from unblocking another's request. `env` is the per-bottle env
|
||||
overlay used to redact secrets from the proposal text. Returns True if
|
||||
the operator approved (the matched value is added to that bottle's
|
||||
safelist and the caller re-scans); False if the request must be blocked
|
||||
(a 403 response has been written to `flow`)."""
|
||||
approval from unblocking another's request. Returns True if the operator
|
||||
approved (the matched value is added to that bottle's safelist and the
|
||||
caller re-scans); False if the request must be blocked (a 403 response
|
||||
has been written to `flow`)."""
|
||||
host = flow.request.pretty_host
|
||||
payload = build_token_allow_payload(
|
||||
redact_tokens(host, env=env),
|
||||
redact_tokens(host, env=os.environ),
|
||||
flow.request.method,
|
||||
redact_tokens(request_path, env=env),
|
||||
redact_tokens(request_path, env=os.environ),
|
||||
result,
|
||||
)
|
||||
proposal = _sv.Proposal.new(
|
||||
|
||||
@@ -450,31 +450,28 @@ def resolve_client_config(
|
||||
|
||||
class ContextResolverLike(typing.Protocol):
|
||||
"""The bit of `policy_resolver.PolicyResolver` `resolve_client_context`
|
||||
needs — one round-trip returning policy, bottle id, and auth tokens."""
|
||||
needs — one round-trip returning both policy and bottle id."""
|
||||
|
||||
def resolve_policy_and_bottle_id(
|
||||
self, source_ip: str, identity_token: str = ...,
|
||||
) -> "tuple[str | None, str | None, dict[str, str]]":
|
||||
) -> "tuple[str | None, str | None]":
|
||||
...
|
||||
|
||||
|
||||
def resolve_client_context(
|
||||
resolver: ContextResolverLike, client_ip: str, identity_token: str = "",
|
||||
) -> "tuple[Config, str, dict[str, str]]":
|
||||
"""The calling client's `(Config, bottle_id, tokens)` in one round-trip —
|
||||
) -> "tuple[Config, str]":
|
||||
"""The calling client's `(Config, bottle_id)` in one round-trip —
|
||||
**fail-closed**. The Config follows `resolve_client_config`'s deny-all
|
||||
rules; the bottle id is `""` whenever unattributed or the orchestrator
|
||||
errored (caller treats as "supervise unavailable", never another bottle's
|
||||
queue); `tokens` are the per-bottle upstream auth values the addon injects.
|
||||
One `/resolve` keys the egress policy, the supervise queue + safelist, and
|
||||
auth injection."""
|
||||
errored, which the caller treats as "supervise unavailable for this
|
||||
bottle" (never another bottle's queue). One `/resolve` keys both the
|
||||
per-request egress policy and the per-bottle supervise queue + safelist."""
|
||||
try:
|
||||
policy, bottle_id, tokens = resolver.resolve_policy_and_bottle_id(
|
||||
client_ip, identity_token,
|
||||
)
|
||||
policy, bottle_id = resolver.resolve_policy_and_bottle_id(client_ip, identity_token)
|
||||
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
|
||||
return Config(routes=()), "", {} # orchestrator unreachable/errored → deny
|
||||
return _config_from_policy(policy), (bottle_id or ""), tokens
|
||||
return Config(routes=()), "" # orchestrator unreachable/errored → deny
|
||||
return _config_from_policy(policy), (bottle_id or "")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -90,18 +90,14 @@ class OrchestratorClient:
|
||||
image_ref: str = "",
|
||||
metadata: str = "",
|
||||
policy: str = "",
|
||||
tokens: dict[str, str] | None = None,
|
||||
) -> RegisteredBottle:
|
||||
"""Register a bottle and broker its launch (`POST /bottles`). `tokens`
|
||||
are the per-bottle egress auth values (env_name -> value) the
|
||||
orchestrator holds in memory for the gateway to inject. Returns the
|
||||
minted id + identity token."""
|
||||
"""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,
|
||||
"tokens": tokens or {},
|
||||
})
|
||||
bottle_id = payload.get("bottle_id")
|
||||
token = payload.get("identity_token")
|
||||
|
||||
@@ -81,16 +81,11 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
|
||||
image_ref = data.get("image_ref")
|
||||
metadata = data.get("metadata")
|
||||
policy = data.get("policy")
|
||||
raw_tokens = data.get("tokens")
|
||||
tokens = {
|
||||
k: v for k, v in raw_tokens.items() if isinstance(k, str) and isinstance(v, str)
|
||||
} if isinstance(raw_tokens, dict) else {}
|
||||
rec = orch.launch_bottle(
|
||||
source_ip,
|
||||
image_ref=image_ref if isinstance(image_ref, str) else "",
|
||||
metadata=metadata if isinstance(metadata, str) else "",
|
||||
policy=policy if isinstance(policy, str) else "",
|
||||
tokens=tokens,
|
||||
)
|
||||
return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
|
||||
|
||||
@@ -142,13 +137,7 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
|
||||
rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
|
||||
if rec is None:
|
||||
return 403, {"error": "unattributed"}
|
||||
# tokens are the in-memory per-bottle egress auth values the gateway
|
||||
# injects; served here, never persisted.
|
||||
return 200, {
|
||||
"bottle_id": rec.bottle_id,
|
||||
"policy": rec.policy,
|
||||
"tokens": orch.tokens_for(rec.bottle_id),
|
||||
}
|
||||
return 200, {"bottle_id": rec.bottle_id, "policy": rec.policy}
|
||||
|
||||
return 404, {"error": "not found"}
|
||||
|
||||
|
||||
@@ -19,16 +19,10 @@ from __future__ import annotations
|
||||
|
||||
import abc
|
||||
import os
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from ..docker_cmd import run_docker
|
||||
|
||||
# The gateway's mitmproxy writes its CA a beat after the container starts, so
|
||||
# reads poll for it rather than assuming it's there on a fresh launch.
|
||||
_CA_POLL_SECONDS = 0.5
|
||||
DEFAULT_CA_TIMEOUT_SECONDS = 30.0
|
||||
|
||||
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.
|
||||
@@ -144,18 +138,6 @@ class DockerGateway(Gateway):
|
||||
])
|
||||
return self.name in proc.stdout.split()
|
||||
|
||||
def _running_image_is_current(self) -> bool:
|
||||
"""True iff the running gateway was created from the *current*
|
||||
`image_ref`. When `ensure_built` rebuilds the image (a source change),
|
||||
the running container is still the OLD image running the OLD flat
|
||||
daemons — so this is how a rebuild actually takes effect: a mismatch
|
||||
means recreate."""
|
||||
running = run_docker(["docker", "inspect", "--format", "{{.Image}}", self.name])
|
||||
current = run_docker(["docker", "image", "inspect", "--format", "{{.Id}}", self.image_ref])
|
||||
if running.returncode != 0 or current.returncode != 0:
|
||||
return True # can't compare → don't churn a working container
|
||||
return running.stdout.strip() == current.stdout.strip()
|
||||
|
||||
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').
|
||||
@@ -169,15 +151,11 @@ class DockerGateway(Gateway):
|
||||
)
|
||||
|
||||
def ensure_running(self) -> None:
|
||||
# Recreate when the running container's image is stale (a rebuild),
|
||||
# so source changes to the gateway's flat daemons take effect — not
|
||||
# just when the container is absent.
|
||||
if self.is_running() and self._running_image_is_current():
|
||||
if self.is_running():
|
||||
return
|
||||
self._ensure_network()
|
||||
# Clear any stale (stopped OR outdated-image) container holding the
|
||||
# fixed name, then start fresh. `rm --force` on an absent name is a
|
||||
# tolerated no-op.
|
||||
# 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])
|
||||
argv = [
|
||||
"docker", "run", "--detach",
|
||||
@@ -197,23 +175,17 @@ class DockerGateway(Gateway):
|
||||
if proc.returncode != 0:
|
||||
raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
|
||||
|
||||
def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:
|
||||
def ca_cert_pem(self) -> str:
|
||||
"""The gateway's CA certificate (PEM) that agents install to trust its
|
||||
TLS interception. mitmproxy generates it a moment after the container
|
||||
starts, so this **polls** for it (up to `timeout`) rather than assuming
|
||||
it's already there on a fresh gateway — raising only if it never
|
||||
appears."""
|
||||
deadline = time.monotonic() + timeout
|
||||
while True:
|
||||
proc = run_docker(["docker", "exec", self.name, "cat", GATEWAY_CA_CERT])
|
||||
if proc.returncode == 0 and proc.stdout.strip():
|
||||
return proc.stdout
|
||||
if time.monotonic() >= deadline:
|
||||
raise GatewayError(
|
||||
f"gateway CA cert not available after {timeout:g}s: "
|
||||
f"{proc.stderr.strip() or 'empty'}"
|
||||
)
|
||||
time.sleep(_CA_POLL_SECONDS)
|
||||
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])
|
||||
|
||||
@@ -127,17 +127,13 @@ class OrchestratorService:
|
||||
gateway are left untouched. Raises `OrchestratorStartError` on
|
||||
timeout."""
|
||||
gateway = self._gateway()
|
||||
gateway.ensure_built() # rebuild the bundle image on a source change
|
||||
gateway.ensure_running() # creates the shared network + (re)starts gateway
|
||||
gateway.ensure_built() # build the bundle image if missing (both use it)
|
||||
if self.is_healthy():
|
||||
gateway.ensure_running() # make sure the gateway is up too
|
||||
return self.url
|
||||
|
||||
# Always (re)create the orchestrator container. It runs the repo's code
|
||||
# bind-mounted, but the Python process loaded that code at startup and
|
||||
# won't reload — so reusing a healthy-but-stale container would keep
|
||||
# running OLD control-plane code (e.g. dropping the tokens field). Cheap
|
||||
# (~seconds); the registry DB persists and the current launch
|
||||
# re-registers its own in-memory state. (The dedicated orchestrator
|
||||
# image follow-up replaces this with image-staleness detection.)
|
||||
log.info("starting orchestrator container", context={"name": ORCHESTRATOR_NAME})
|
||||
gateway.ensure_running() # creates the shared network + gateway
|
||||
self._run_orchestrator_container()
|
||||
|
||||
deadline = time.monotonic() + startup_timeout
|
||||
|
||||
@@ -149,15 +149,7 @@ class RegistryStore(DbStore):
|
||||
policy: str = "",
|
||||
) -> BottleRecord:
|
||||
"""Register (or replace) a bottle. Mints a bottle_id / identity token
|
||||
when not supplied. Live — this is the control-plane reload path.
|
||||
|
||||
Supersedes any *other* active bottle at the same source IP first:
|
||||
`by_source_ip` fail-closes on ambiguity (>1 active row → None), so a
|
||||
leftover row at a reused IP — from an untorn-down bottle, crash
|
||||
recovery, or a persisted DB — would otherwise *brick* the new bottle's
|
||||
attribution. The new registration is authoritative for its IP; the
|
||||
stale rows go. INSERT OR REPLACE handles a same-`bottle_id` reload in
|
||||
place (its own row is exempt from the sweep)."""
|
||||
when not supplied. Live — this is the control-plane reload path."""
|
||||
rec = BottleRecord(
|
||||
bottle_id=bottle_id or secrets.token_hex(8),
|
||||
source_ip=source_ip,
|
||||
@@ -168,11 +160,6 @@ class RegistryStore(DbStore):
|
||||
policy=policy,
|
||||
)
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"DELETE FROM orchestrator_bottles "
|
||||
"WHERE source_ip = ? AND state = 'active' AND bottle_id != ?",
|
||||
(rec.source_ip, rec.bottle_id),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT OR REPLACE INTO orchestrator_bottles "
|
||||
"(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "
|
||||
|
||||
@@ -38,12 +38,6 @@ class Orchestrator:
|
||||
self._broker = broker
|
||||
self._secret = sign_secret
|
||||
self._gateway = gateway
|
||||
# Per-bottle egress auth tokens (env_name -> value), keyed by bottle_id.
|
||||
# Held **in memory only** — never written to the registry DB — so the
|
||||
# gateway can inject each bottle's upstream credential without secrets
|
||||
# at rest. Lost on restart (re-launch re-registers them); the future
|
||||
# SecretProvider (#355) replaces this with per-request minting.
|
||||
self._tokens: dict[str, dict[str, str]] = {}
|
||||
|
||||
def launch_bottle(
|
||||
self,
|
||||
@@ -53,14 +47,11 @@ class Orchestrator:
|
||||
slot: int | None = None,
|
||||
metadata: str = "",
|
||||
policy: str = "",
|
||||
tokens: dict[str, str] | None = None,
|
||||
) -> BottleRecord:
|
||||
"""Register a bottle (with its gateway policy + in-memory egress auth
|
||||
tokens) and broker its launch. Rolls the registry entry back if the
|
||||
launch doesn't take, so a failure leaves no orphan."""
|
||||
"""Register a bottle (with its gateway policy) and broker its launch.
|
||||
Rolls the registry entry back if the launch doesn't take, so a
|
||||
failure leaves no orphan."""
|
||||
rec = self.registry.register(source_ip, metadata=metadata, policy=policy)
|
||||
if tokens:
|
||||
self._tokens[rec.bottle_id] = dict(tokens)
|
||||
req = LaunchRequest(
|
||||
op="launch",
|
||||
bottle_id=rec.bottle_id,
|
||||
@@ -75,7 +66,6 @@ class Orchestrator:
|
||||
finally:
|
||||
if not launched:
|
||||
self.registry.deregister(rec.bottle_id)
|
||||
self._tokens.pop(rec.bottle_id, None)
|
||||
return rec
|
||||
|
||||
def teardown_bottle(self, bottle_id: str) -> bool:
|
||||
@@ -86,15 +76,8 @@ class Orchestrator:
|
||||
req = LaunchRequest(op="teardown", bottle_id=bottle_id, source_ip=rec.source_ip)
|
||||
self._broker.submit(sign_request(req, self._secret))
|
||||
self.registry.deregister(bottle_id)
|
||||
self._tokens.pop(bottle_id, None)
|
||||
return True
|
||||
|
||||
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
|
||||
persisted."""
|
||||
return dict(self._tokens.get(bottle_id, {}))
|
||||
|
||||
def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None:
|
||||
"""Fail-closed attribution (delegates to the registry)."""
|
||||
return self.registry.attribute(source_ip, identity_token)
|
||||
|
||||
@@ -102,26 +102,21 @@ class PolicyResolver:
|
||||
|
||||
def resolve_policy_and_bottle_id(
|
||||
self, source_ip: str, identity_token: str = "",
|
||||
) -> tuple[str | None, str | None, dict[str, str]]:
|
||||
"""The policy blob, bottle id, **and per-bottle egress auth tokens** in
|
||||
a single `/resolve` — so the egress addon gets everything it needs
|
||||
(policy for routing, bottle id for the supervise queue/safelist, tokens
|
||||
to inject upstream auth) in one round-trip. Returns `(None, None, {})`
|
||||
when unattributed (a clean `403`). Raises `PolicyResolveError` if the
|
||||
orchestrator can't be reached, so the caller still fails closed."""
|
||||
) -> tuple[str | None, str | None]:
|
||||
"""Both the policy blob and the bottle id in a single `/resolve` — so
|
||||
a caller that needs each (the egress addon: policy for routing, bottle
|
||||
id to key the calling bottle's supervise queue + safelist) makes one
|
||||
round-trip, not two. Returns `(None, None)` when unattributed (a clean
|
||||
`403`). Raises `PolicyResolveError` if the orchestrator can't be
|
||||
reached, so the caller still fails closed."""
|
||||
payload = self._post_resolve(source_ip, identity_token)
|
||||
if payload is None:
|
||||
return None, None, {}
|
||||
return None, None
|
||||
policy = payload.get("policy")
|
||||
bottle_id = payload.get("bottle_id")
|
||||
raw = payload.get("tokens")
|
||||
tokens = {
|
||||
k: v for k, v in raw.items() if isinstance(k, str) and isinstance(v, str)
|
||||
} if isinstance(raw, dict) else {}
|
||||
return (
|
||||
policy if isinstance(policy, str) else "",
|
||||
bottle_id if isinstance(bottle_id, str) and bottle_id else None,
|
||||
tokens,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -15,25 +15,6 @@ the biggest credential risk).
|
||||
|
||||
## Summary
|
||||
|
||||
> **Status — implemented (2026-07-14).** This note's recommendation
|
||||
> shipped, so the "today" / "in-flight" tenses below are now historical
|
||||
> (kept as the reasoning that led here). Current reality: bot-bottle no
|
||||
> longer injects raw provider tokens into the agent. The agent's environ
|
||||
> carries only a **placeholder** (e.g.
|
||||
> `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder`); the real token is held
|
||||
> in the egress sidecar and injected as the upstream `Authorization`
|
||||
> header. Rather than adding a *separate* in-container reverse proxy (as
|
||||
> proposed below), credential injection was folded into the **existing
|
||||
> pipelock/mitmproxy egress firewall**: an `EgressRoute` carries
|
||||
> `auth_scheme` + `token_ref`, the host token is forwarded into the
|
||||
> sidecar addon under an `EGRESS_TOKEN_N` slot, and the addon rewrites
|
||||
> `Authorization` on matching routes — **one MITM, not two** (this
|
||||
> resolves the Infisical-doubling concern noted later). The same
|
||||
> mechanism now covers Codex and Pi provider tokens and git-host PATs.
|
||||
> A `printenv` inside the bottle yields the placeholder, not the secret.
|
||||
> For current wiring see `bot_bottle/egress.py` and
|
||||
> `bot_bottle/contrib/*/agent_provider.py`.
|
||||
|
||||
Today every bot-bottle agent gets `CLAUDE_CODE_OAUTH_TOKEN` (and
|
||||
any `bottle.env` secrets like a Gitea PAT) injected as env vars,
|
||||
which means the agent process can read them with `printenv` or
|
||||
@@ -114,9 +95,7 @@ The remaining credible designs reduce to three:
|
||||
|
||||
### Anthropic / Claude Code
|
||||
|
||||
**Original wiring** (pre-gateway; **superseded 2026-07-14** — see the
|
||||
Status note in the Summary; the token now lives in the egress sidecar
|
||||
and the agent gets a placeholder): the host's
|
||||
**Today's wiring** (`bot_bottle/cli/start.py`): the host's
|
||||
`BOT_BOTTLE_CLAUDE_OAUTH_TOKEN` is forwarded into the bottle as
|
||||
`CLAUDE_CODE_OAUTH_TOKEN` via `docker run -e CLAUDE_CODE_OAUTH_TOKEN`
|
||||
(no `=value`, so the value never lands on argv — good). Inside the
|
||||
@@ -353,12 +332,6 @@ exposed.
|
||||
|
||||
## Recommended path forward
|
||||
|
||||
> **Mostly implemented (2026-07-14).** Steps 1–3 shipped — the token
|
||||
> injection moved into the egress sidecar (folded into pipelock rather
|
||||
> than a standalone reverse proxy) and generalized to Codex/Pi/git-host
|
||||
> tokens. Steps 4–5 (issuance-side scope narrowing, per-route allowlists)
|
||||
> remain good hygiene. See the Status note in the Summary.
|
||||
|
||||
In priority order:
|
||||
|
||||
1. **In-container reverse proxy holding `CLAUDE_CODE_OAUTH_TOKEN`.**
|
||||
|
||||
@@ -103,19 +103,7 @@ manifest merge.
|
||||
has no equivalent. (3) *Trust posture:* OneCLI's managed tier reintroduces a
|
||||
third-party credential custodian, whereas bot-bottle's OSS-runtime +
|
||||
paid-control-plane split keeps custody inside the operator's own boundary —
|
||||
the stronger story for the security-minded self-hoster. (4) *Runs inside your
|
||||
network boundary — local/internal reach.* Because bot-bottle executes the
|
||||
agent on your own host (homelab, corporate LAN, a Tailnet) and egress is a
|
||||
manifest field, giving an agent *scoped* access to **internal** resources — a
|
||||
private Gitea, a LAN database, a Tailscale node — is just another egress-route
|
||||
line, not a networking project (the same move an operator already makes to
|
||||
reach their Tailscale services). OneCLI's OSS core can self-host too, but it's
|
||||
a credential *broker* for outbound API calls, not an agent runtime, and its
|
||||
managed tier + 50+ integrations are oriented at public SaaS — it doesn't put
|
||||
the agent behind your firewall for you. This is a reach advantage, distinct
|
||||
from the isolation ones above, and it's a wedge cloud-first agent products
|
||||
(Devin, Copilot Workspace, OneCLI Cloud) structurally can't match. **Tactical
|
||||
read:**
|
||||
the stronger story for the security-minded self-hoster. **Tactical read:**
|
||||
adopt OneCLI's OSS core for the credential slice if building is undesirable
|
||||
(it's mature now); don't build atop its managed tier (competitor, not
|
||||
dependency); re-position bot-bottle on isolation + fleet + self-hosted custody
|
||||
|
||||
@@ -1,161 +0,0 @@
|
||||
"""Integration: two bottles share one gateway; source-IP attribution keeps
|
||||
their egress tokens *and* allowlists isolated (PRD 0070 multi-tenancy).
|
||||
|
||||
The whole premise of the consolidation is one shared gateway for every
|
||||
bottle, isolated only by the unspoofable source IP. A single-bottle test
|
||||
can't catch cross-bottle token bleed or an allowlist leak — this brings up
|
||||
the real orchestrator + gateway and drives two bottles through the one
|
||||
gateway to prove each gets exactly its own token and its own routes.
|
||||
|
||||
Gated on a reachable Docker daemon and builds the bundle image, so it runs
|
||||
with the other docker integration tests, not in unit CI. Uses the real
|
||||
fixed gateway/orchestrator names (that's what it's testing), so it can't run
|
||||
concurrently with a live per-host gateway; it points the orchestrator at a
|
||||
throwaway BOT_BOTTLE_ROOT for a clean registry and tears everything down.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from bot_bottle.backend.docker.consolidated_launch import (
|
||||
_network_cidr,
|
||||
_network_container_ips,
|
||||
)
|
||||
from bot_bottle.backend.docker.egress import EGRESS_PORT
|
||||
from bot_bottle.backend.docker.gateway_net import next_free_ip
|
||||
from bot_bottle.orchestrator.client import OrchestratorClient
|
||||
from bot_bottle.orchestrator.gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK
|
||||
from bot_bottle.orchestrator.lifecycle import OrchestratorService
|
||||
from tests._docker import skip_unless_docker
|
||||
|
||||
# One upstream reachable under two names; reflects the Authorization header so
|
||||
# the probe can see exactly what the gateway injected (or didn't).
|
||||
_ECHO_NAME = "bot-bottle-mtitest-echo"
|
||||
_ECHO_ALIASES = ("echo-shared", "echo-bonly")
|
||||
_ECHO_SRC = (
|
||||
"from http.server import BaseHTTPRequestHandler, HTTPServer\n"
|
||||
"class H(BaseHTTPRequestHandler):\n"
|
||||
" def do_GET(s):\n"
|
||||
" a=s.headers.get('Authorization','NONE'); s.send_response(200); s.end_headers()\n"
|
||||
" s.wfile.write(('AUTH='+a).encode())\n"
|
||||
" def log_message(s,*a): pass\n"
|
||||
"HTTPServer(('0.0.0.0',80),H).serve_forever()\n"
|
||||
)
|
||||
|
||||
# A: echo-shared only, authed. B: echo-shared authed + echo-bonly open.
|
||||
_POLICY_A = (
|
||||
'routes:\n'
|
||||
' - host: echo-shared\n'
|
||||
' auth_scheme: "Bearer"\n'
|
||||
' token_env: "EGRESS_TOKEN_0"\n'
|
||||
)
|
||||
_POLICY_B = _POLICY_A + ' - host: echo-bonly\n'
|
||||
_TOKEN_A = "sk-AAA-alice"
|
||||
_TOKEN_B = "sk-BBB-bob"
|
||||
|
||||
# Client probe: open http://<host>/ through the gateway proxy from a container
|
||||
# pinned to the bottle's source IP, printing "<status> <body>". Uses only the
|
||||
# bundle image's python3 — no dependency on the agent image.
|
||||
_PROBE_SRC = (
|
||||
"import sys,urllib.request,urllib.error\n"
|
||||
"op=urllib.request.build_opener(urllib.request.ProxyHandler({'http':sys.argv[1]}))\n"
|
||||
"try:\n"
|
||||
" r=op.open('http://'+sys.argv[2]+'/',timeout=8)\n"
|
||||
" sys.stdout.write(str(r.status)+' '+r.read().decode())\n"
|
||||
"except urllib.error.HTTPError as e:\n"
|
||||
" sys.stdout.write(str(e.code)+' '+e.read().decode())\n"
|
||||
)
|
||||
|
||||
|
||||
@skip_unless_docker()
|
||||
@unittest.skipIf(
|
||||
os.environ.get("GITEA_ACTIONS") == "true",
|
||||
"skipped under act_runner: the orchestrator container bind-mounts the repo "
|
||||
"path into a container on the socket-shared host daemon, which can't see the "
|
||||
"runner's /workspace — same host-bind-mount constraint as the other "
|
||||
"bottle-bringup integration tests",
|
||||
)
|
||||
class TestMultitenantIsolation(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
# Throwaway root → a clean registry DB, independent of the host's.
|
||||
self.svc = OrchestratorService(host_root=Path(self._tmp.name))
|
||||
self.addCleanup(self._teardown_docker)
|
||||
# ensure_running builds the bundle image (slow on a cold cache) and
|
||||
# brings up the shared network + gateway + orchestrator.
|
||||
url = self.svc.ensure_running(startup_timeout=300)
|
||||
self.client = OrchestratorClient(url)
|
||||
self.gw_ip = self._container_ip(GATEWAY_NAME)
|
||||
self._run_echo()
|
||||
|
||||
def _teardown_docker(self) -> None:
|
||||
subprocess.run(["docker", "rm", "--force", _ECHO_NAME],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
|
||||
self.svc.stop()
|
||||
subprocess.run(["docker", "network", "rm", GATEWAY_NETWORK],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
|
||||
# The orchestrator container wrote the registry DB as root into the
|
||||
# throwaway root; chown it back so the (non-root) tempdir cleanup can
|
||||
# remove it.
|
||||
subprocess.run(
|
||||
["docker", "run", "--rm", "-v", f"{self._tmp.name}:/r",
|
||||
"--entrypoint", "chown", GATEWAY_IMAGE, "-R",
|
||||
f"{os.getuid()}:{os.getgid()}", "/r"],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
|
||||
|
||||
@staticmethod
|
||||
def _container_ip(name: str) -> str:
|
||||
proc = subprocess.run(
|
||||
["docker", "inspect", "-f",
|
||||
'{{(index .NetworkSettings.Networks "' + GATEWAY_NETWORK + '").IPAddress}}', name],
|
||||
stdout=subprocess.PIPE, text=True, check=True,
|
||||
)
|
||||
return proc.stdout.strip()
|
||||
|
||||
def _run_echo(self) -> None:
|
||||
argv = ["docker", "run", "--detach", "--name", _ECHO_NAME,
|
||||
"--network", GATEWAY_NETWORK]
|
||||
for alias in _ECHO_ALIASES:
|
||||
argv += ["--network-alias", alias]
|
||||
argv += ["--entrypoint", "python3", GATEWAY_IMAGE, "-c", _ECHO_SRC]
|
||||
proc = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
text=True, check=False)
|
||||
self.assertEqual(0, proc.returncode, f"echo upstream failed: {proc.stderr}")
|
||||
|
||||
def _free_ip(self, extra_taken: list[str]) -> str:
|
||||
taken = _network_container_ips(GATEWAY_NETWORK) + extra_taken
|
||||
return next_free_ip(_network_cidr(GATEWAY_NETWORK), taken)
|
||||
|
||||
def _probe(self, source_ip: str, host: str) -> str:
|
||||
proc = subprocess.run(
|
||||
["docker", "run", "--rm", "--network", GATEWAY_NETWORK, "--ip", source_ip,
|
||||
"--entrypoint", "python3", GATEWAY_IMAGE, "-c", _PROBE_SRC,
|
||||
f"http://{self.gw_ip}:{EGRESS_PORT}", host],
|
||||
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, timeout=90,
|
||||
)
|
||||
return proc.stdout.strip()
|
||||
|
||||
def test_two_bottles_share_gateway_with_isolated_tokens_and_allowlists(self) -> None:
|
||||
ip_a = self._free_ip([])
|
||||
ip_b = self._free_ip([ip_a])
|
||||
self.client.register_bottle(ip_a, policy=_POLICY_A, tokens={"EGRESS_TOKEN_0": _TOKEN_A})
|
||||
self.client.register_bottle(ip_b, policy=_POLICY_B, tokens={"EGRESS_TOKEN_0": _TOKEN_B})
|
||||
|
||||
# Each bottle gets its OWN token injected on the shared route — no bleed.
|
||||
self.assertEqual(f"200 AUTH=Bearer {_TOKEN_A}", self._probe(ip_a, "echo-shared"))
|
||||
self.assertEqual(f"200 AUTH=Bearer {_TOKEN_B}", self._probe(ip_b, "echo-shared"))
|
||||
|
||||
# Allowlist is per-bottle: echo-bonly is only in B's policy.
|
||||
self.assertTrue(self._probe(ip_a, "echo-bonly").startswith("403"), # fail-closed for A
|
||||
"A reached a host outside its allowlist")
|
||||
self.assertEqual("200 AUTH=NONE", self._probe(ip_b, "echo-bonly")) # allowed, unauthed for B
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -40,7 +40,7 @@ class TestGetBottleBackend(unittest.TestCase):
|
||||
return True
|
||||
|
||||
with patch.dict(os.environ, {}, clear=True), \
|
||||
patch.object(backend_mod, "_backends", {
|
||||
patch.object(backend_mod, "_BACKENDS", {
|
||||
"macos-container": _FakeBackend(),
|
||||
"docker": _FakeBackend(),
|
||||
}):
|
||||
@@ -61,7 +61,7 @@ class TestGetBottleBackend(unittest.TestCase):
|
||||
with patch.dict(os.environ, {}, clear=True), \
|
||||
patch.object(backend_mod.FirecrackerBottleBackend,
|
||||
"is_host_capable", classmethod(lambda cls: False)), \
|
||||
patch.object(backend_mod, "_backends", {
|
||||
patch.object(backend_mod, "_BACKENDS", {
|
||||
"macos-container": _FakeBackend("macos-container", False),
|
||||
"docker": _FakeBackend("docker", True),
|
||||
}):
|
||||
@@ -83,7 +83,7 @@ class TestGetBottleBackend(unittest.TestCase):
|
||||
with patch.dict(os.environ, {}, clear=True), \
|
||||
patch.object(backend_mod.FirecrackerBottleBackend,
|
||||
"is_host_capable", classmethod(lambda cls: True)), \
|
||||
patch.object(backend_mod, "_backends", {
|
||||
patch.object(backend_mod, "_BACKENDS", {
|
||||
"macos-container": _FakeBackend("macos-container", False),
|
||||
"firecracker": _FakeBackend("firecracker", False),
|
||||
"docker": _FakeBackend("docker", True),
|
||||
@@ -133,7 +133,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
|
||||
return self._items
|
||||
|
||||
with patch.object(
|
||||
backend_mod, "_backends",
|
||||
backend_mod, "_BACKENDS",
|
||||
{"docker": _FakeBackend([a]), "firecracker": _FakeBackend([b])},
|
||||
):
|
||||
self.assertEqual([a, b], enumerate_active_agents())
|
||||
@@ -167,7 +167,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
|
||||
return self._items
|
||||
|
||||
with patch.object(
|
||||
backend_mod, "_backends",
|
||||
backend_mod, "_BACKENDS",
|
||||
{
|
||||
"docker": _FakeBackend([newer, tie_b]),
|
||||
"firecracker": _FakeBackend([missing_metadata, tie_a]),
|
||||
@@ -187,7 +187,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
|
||||
return []
|
||||
|
||||
with patch.object(
|
||||
backend_mod, "_backends",
|
||||
backend_mod, "_BACKENDS",
|
||||
{"docker": _FakeBackend(), "firecracker": _FakeBackend()},
|
||||
):
|
||||
self.assertEqual([], enumerate_active_agents())
|
||||
@@ -218,7 +218,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
|
||||
return self._items
|
||||
|
||||
with patch.object(
|
||||
backend_mod, "_backends",
|
||||
backend_mod, "_BACKENDS",
|
||||
{
|
||||
"docker": _FakeBackend([present], available=True),
|
||||
"firecracker": _FakeBackend([hidden], available=False),
|
||||
@@ -234,7 +234,7 @@ class TestHasBackend(unittest.TestCase):
|
||||
return False
|
||||
|
||||
with patch.object(
|
||||
backend_mod, "_backends", {"docker": _FakeBackend()},
|
||||
backend_mod, "_BACKENDS", {"docker": _FakeBackend()},
|
||||
):
|
||||
from bot_bottle.backend import has_backend
|
||||
self.assertFalse(has_backend("docker"))
|
||||
|
||||
@@ -29,7 +29,7 @@ def _fail(stderr: str = "boom") -> subprocess.CompletedProcess: # type: ignore
|
||||
class TestCommitContainer(unittest.TestCase):
|
||||
def test_runs_docker_commit(self):
|
||||
with patch.object(
|
||||
docker_mod, "run_docker", return_value=_ok(),
|
||||
docker_mod.subprocess, "run", return_value=_ok(),
|
||||
) as run, patch.object(docker_mod, "info"):
|
||||
docker_mod.commit_container(
|
||||
"bot-bottle-dev-abc12",
|
||||
@@ -47,7 +47,7 @@ class TestCommitContainer(unittest.TestCase):
|
||||
|
||||
def test_dies_on_docker_commit_failure(self):
|
||||
with patch.object(
|
||||
docker_mod, "run_docker", return_value=_fail("No such container"),
|
||||
docker_mod.subprocess, "run", return_value=_fail("No such container"),
|
||||
), patch.object(
|
||||
docker_mod, "die", side_effect=SystemExit("die"),
|
||||
) as die:
|
||||
@@ -58,7 +58,7 @@ class TestCommitContainer(unittest.TestCase):
|
||||
|
||||
def test_die_message_includes_image_tag(self):
|
||||
with patch.object(
|
||||
docker_mod, "run_docker", return_value=_fail("boom"),
|
||||
docker_mod.subprocess, "run", return_value=_fail("boom"),
|
||||
), patch.object(
|
||||
docker_mod, "die", side_effect=SystemExit("die"),
|
||||
) as die:
|
||||
|
||||
@@ -233,19 +233,16 @@ class _CtxResolver:
|
||||
"""Fake orchestrator resolver: maps source IP -> bottle id, and grants the
|
||||
same allow-list to any attributed bottle (unattributed -> deny)."""
|
||||
|
||||
def __init__(
|
||||
self, ip_to_bottle: dict[str, str], tokens: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
def __init__(self, ip_to_bottle: dict[str, str]) -> None:
|
||||
self._map = ip_to_bottle
|
||||
self._tokens = tokens or {}
|
||||
|
||||
def resolve_policy_and_bottle_id(
|
||||
self, source_ip: str, identity_token: str = "",
|
||||
) -> tuple[str | None, str | None, dict[str, str]]:
|
||||
) -> tuple[str | None, str | None]:
|
||||
del identity_token
|
||||
bottle_id = self._map.get(source_ip)
|
||||
policy = "routes:\n - host: api.example.com\n" if bottle_id else None
|
||||
return policy, bottle_id, (self._tokens if bottle_id else {})
|
||||
return policy, bottle_id
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -806,46 +803,6 @@ class TestSuperviseMultiTenant(unittest.TestCase):
|
||||
_run_request(addon, flow)
|
||||
self.assertEqual(["bottle-b"], seen) # proposal keyed by the resolved bottle
|
||||
|
||||
def test_auth_token_injected_from_resolved_tokens(self) -> None:
|
||||
# The bottle's upstream token comes from /resolve (in-memory on the
|
||||
# orchestrator), NOT the gateway's env — the request is forwarded with
|
||||
# Authorization injected. This is the token-injection cut-over fix.
|
||||
policy = (
|
||||
'routes:\n - host: api.example.com\n'
|
||||
' auth_scheme: "Bearer"\n token_env: "EGRESS_TOKEN_0"\n'
|
||||
)
|
||||
|
||||
class _AuthResolver:
|
||||
def resolve_policy_and_bottle_id(self, ip: str, identity_token: str = ""):
|
||||
del ip, identity_token
|
||||
return policy, "bottle-a", {"EGRESS_TOKEN_0": "sk-secret"}
|
||||
|
||||
addon = _addon(Config(routes=()))
|
||||
addon._resolver = cast(Any, _AuthResolver())
|
||||
flow = _with_client_ip(_Flow(_Request(host="api.example.com", method="GET")), "10.0.0.1")
|
||||
_run_request(addon, flow)
|
||||
self.assertIsNone(flow.response) # forwarded, not blocked
|
||||
self.assertEqual("Bearer sk-secret", flow.request.headers.get("authorization"))
|
||||
|
||||
def test_authed_route_without_token_is_blocked(self) -> None:
|
||||
# No token resolved for the bottle → the authed route can't inject, so
|
||||
# the request is blocked (the error the cut-over originally produced).
|
||||
policy = (
|
||||
'routes:\n - host: api.example.com\n'
|
||||
' auth_scheme: "Bearer"\n token_env: "EGRESS_TOKEN_0"\n'
|
||||
)
|
||||
|
||||
class _NoTokenResolver:
|
||||
def resolve_policy_and_bottle_id(self, ip: str, identity_token: str = ""):
|
||||
del ip, identity_token
|
||||
return policy, "bottle-a", {} # no tokens
|
||||
|
||||
addon = _addon(Config(routes=()))
|
||||
addon._resolver = cast(Any, _NoTokenResolver())
|
||||
flow = _with_client_ip(_Flow(_Request(host="api.example.com", method="GET")), "10.0.0.1")
|
||||
_run_request(addon, flow)
|
||||
self.assertIsNotNone(flow.response) # blocked — token unset
|
||||
|
||||
def test_unattributed_source_ip_cannot_supervise(self) -> None:
|
||||
addon = self._consolidated_addon()
|
||||
# 10.9.9.9 is not in the resolver map -> deny-all config, empty slug.
|
||||
|
||||
@@ -51,55 +51,45 @@ class TestResolveClientConfig(unittest.TestCase):
|
||||
|
||||
class _FakeContextResolver:
|
||||
def __init__(
|
||||
self, policy: str | None = None, bottle_id: str | None = None,
|
||||
raises: bool = False, tokens: dict[str, str] | None = None,
|
||||
self, policy: str | None = None, bottle_id: str | None = None, raises: bool = False,
|
||||
) -> None:
|
||||
self._policy = policy
|
||||
self._bottle_id = bottle_id
|
||||
self._raises = raises
|
||||
self._tokens = tokens or {}
|
||||
|
||||
def resolve_policy_and_bottle_id(
|
||||
self, source_ip: str, identity_token: str = "",
|
||||
) -> tuple[str | None, str | None, dict[str, str]]:
|
||||
) -> tuple[str | None, str | None]:
|
||||
if self._raises:
|
||||
raise PolicyResolveError("orchestrator down")
|
||||
return self._policy, self._bottle_id, self._tokens
|
||||
return self._policy, self._bottle_id
|
||||
|
||||
|
||||
class TestResolveClientContext(unittest.TestCase):
|
||||
def test_returns_config_bottle_id_and_tokens(self) -> None:
|
||||
cfg, slug, tokens = resolve_client_context(
|
||||
_FakeContextResolver(
|
||||
policy="routes:\n - host: example.com\n", bottle_id="b1",
|
||||
tokens={"EGRESS_TOKEN_0": "sekret"},
|
||||
),
|
||||
def test_returns_config_and_bottle_id(self) -> None:
|
||||
cfg, slug = resolve_client_context(
|
||||
_FakeContextResolver(policy="routes:\n - host: example.com\n", bottle_id="b1"),
|
||||
"10.243.0.1",
|
||||
)
|
||||
self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
|
||||
self.assertEqual("b1", slug)
|
||||
self.assertEqual({"EGRESS_TOKEN_0": "sekret"}, tokens) # for auth injection
|
||||
|
||||
def test_unattributed_denies_and_empty_slug(self) -> None:
|
||||
cfg, slug, tokens = resolve_client_context(
|
||||
cfg, slug = resolve_client_context(
|
||||
_FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
|
||||
)
|
||||
self.assertEqual((), cfg.routes)
|
||||
self.assertEqual("", slug) # no bottle → supervise unavailable
|
||||
self.assertEqual({}, tokens)
|
||||
|
||||
def test_resolver_error_denies_empty_slug_no_tokens(self) -> None:
|
||||
cfg, slug, tokens = resolve_client_context(
|
||||
_FakeContextResolver(raises=True), "10.243.0.1",
|
||||
)
|
||||
def test_resolver_error_denies_and_empty_slug(self) -> None:
|
||||
cfg, slug = resolve_client_context(_FakeContextResolver(raises=True), "10.243.0.1")
|
||||
self.assertEqual((), cfg.routes)
|
||||
self.assertEqual("", slug)
|
||||
self.assertEqual({}, tokens)
|
||||
|
||||
def test_unparseable_policy_denies_but_keeps_slug(self) -> None:
|
||||
# A bad policy denies egress, but the bottle is still attributed (its
|
||||
# supervise queue is keyed by the id, independent of route parsing).
|
||||
cfg, slug, _tokens = resolve_client_context(
|
||||
cfg, slug = resolve_client_context(
|
||||
_FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
|
||||
)
|
||||
self.assertEqual((), cfg.routes)
|
||||
|
||||
@@ -35,43 +35,11 @@ class TestDockerGateway(unittest.TestCase):
|
||||
with patch(_RUN_DOCKER, return_value=_proc(stdout="")):
|
||||
self.assertFalse(self.sc.is_running())
|
||||
|
||||
def test_ensure_running_noop_when_up_and_image_current(self) -> None:
|
||||
calls: list[list[str]] = []
|
||||
|
||||
def fake(argv: list[str]) -> Mock:
|
||||
calls.append(argv)
|
||||
if argv[:2] == ["docker", "ps"]:
|
||||
return _proc(stdout=self.sc.name) # running
|
||||
if argv[:3] == ["docker", "image", "inspect"]:
|
||||
return _proc(stdout="img-A") # current image id
|
||||
if argv[:2] == ["docker", "inspect"]:
|
||||
return _proc(stdout="img-A") # container's image (same)
|
||||
return _proc()
|
||||
|
||||
with patch(_RUN_DOCKER, side_effect=fake):
|
||||
def test_ensure_running_is_noop_when_already_up(self) -> None:
|
||||
with patch(_RUN_DOCKER, return_value=_proc(stdout=self.sc.name)) as m:
|
||||
self.sc.ensure_running()
|
||||
self.assertEqual([], [c for c in calls if c[:2] == ["docker", "run"]])
|
||||
self.assertEqual([], [c for c in calls if c[:2] == ["docker", "rm"]])
|
||||
|
||||
def test_ensure_running_recreates_when_image_is_stale(self) -> None:
|
||||
# Running, but the container was built from an OLD image → recreate so
|
||||
# a rebuild's new flat daemons take effect.
|
||||
calls: list[list[str]] = []
|
||||
|
||||
def fake(argv: list[str]) -> Mock:
|
||||
calls.append(argv)
|
||||
if argv[:2] == ["docker", "ps"]:
|
||||
return _proc(stdout=self.sc.name)
|
||||
if argv[:3] == ["docker", "image", "inspect"]:
|
||||
return _proc(stdout="img-NEW")
|
||||
if argv[:2] == ["docker", "inspect"]:
|
||||
return _proc(stdout="img-OLD")
|
||||
return _proc()
|
||||
|
||||
with patch(_RUN_DOCKER, side_effect=fake):
|
||||
self.sc.ensure_running()
|
||||
self.assertEqual(1, len([c for c in calls if c[:2] == ["docker", "run"]]))
|
||||
self.assertTrue(any(c[:2] == ["docker", "rm"] for c in calls))
|
||||
# Only the is_running() ps probe — no rm / run.
|
||||
self.assertEqual(1, m.call_count)
|
||||
|
||||
def test_ensure_running_starts_the_singleton_when_absent(self) -> None:
|
||||
calls: list[list[str]] = []
|
||||
@@ -113,17 +81,9 @@ class TestDockerGateway(unittest.TestCase):
|
||||
self.assertEqual(["docker", "exec", self.sc.name, "cat", GATEWAY_CA_CERT], argv)
|
||||
|
||||
def test_ca_cert_pem_raises_when_absent(self) -> None:
|
||||
# timeout=0 → one probe then give up (no polling delay in the test).
|
||||
with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="No such file")):
|
||||
with self.assertRaises(GatewayError):
|
||||
self.sc.ca_cert_pem(timeout=0)
|
||||
|
||||
def test_ca_cert_pem_polls_until_mitmproxy_writes_it(self) -> None:
|
||||
# First read: CA not there yet; second read: present.
|
||||
seq = [_proc(returncode=1, stderr="No such file"), _proc(stdout=_CA_PEM)]
|
||||
with patch(_RUN_DOCKER, side_effect=seq), \
|
||||
patch("bot_bottle.orchestrator.gateway.time.sleep"):
|
||||
self.assertEqual(_CA_PEM, self.sc.ca_cert_pem(timeout=5))
|
||||
self.sc.ca_cert_pem()
|
||||
|
||||
def test_ensure_running_reuses_existing_network(self) -> None:
|
||||
calls: list[list[str]] = []
|
||||
|
||||
@@ -46,18 +46,13 @@ class TestOrchestratorService(unittest.TestCase):
|
||||
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
||||
self.assertFalse(self.svc.is_healthy())
|
||||
|
||||
def test_ensure_running_always_recreates_orchestrator(self) -> None:
|
||||
# Even when a control plane is already healthy, the orchestrator is
|
||||
# recreated so bind-mounted code changes take effect (its process
|
||||
# won't reload). The gateway is ensured too.
|
||||
run = Mock(return_value=Mock(returncode=0, stderr=""))
|
||||
def test_ensure_running_healthy_still_ensures_gateway_but_not_orchestrator(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_health(200)), \
|
||||
patch(_GATEWAY) as gw_cls, patch(_RUN, run), patch(_SLEEP):
|
||||
patch(_GATEWAY) as gw_cls, patch(_RUN) as run:
|
||||
self.assertEqual(self.svc.url, self.svc.ensure_running())
|
||||
gw_cls.return_value.ensure_running.assert_called() # gateway kept up
|
||||
runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
|
||||
self.assertEqual(1, len(runs)) # orchestrator recreated
|
||||
self.assertIn(ORCHESTRATOR_NAME, runs[0])
|
||||
gw = gw_cls.return_value
|
||||
gw.ensure_running.assert_called() # gateway kept up
|
||||
run.assert_not_called() # no orchestrator container run
|
||||
|
||||
def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None:
|
||||
run = Mock(return_value=Mock(returncode=0, stderr=""))
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
@@ -24,20 +23,6 @@ class TestRegistryStore(unittest.TestCase):
|
||||
def tearDown(self) -> None:
|
||||
self._tmp.cleanup()
|
||||
|
||||
def _force_active_row(self, source_ip: str, bottle_id: str) -> None:
|
||||
"""Insert a raw active row, bypassing `register`'s same-IP supersede —
|
||||
the only way to manufacture the ambiguity the fail-closed guard exists
|
||||
for (crash-orphaned / externally-injected duplicate)."""
|
||||
conn = sqlite3.connect(self.db)
|
||||
conn.execute(
|
||||
"INSERT INTO orchestrator_bottles "
|
||||
"(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "
|
||||
"VALUES (?, ?, ?, 'active', 0.0, '', '')",
|
||||
(bottle_id, source_ip, "tok-" + bottle_id),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def test_register_mints_id_and_token(self) -> None:
|
||||
rec = self.store.register("10.243.0.1")
|
||||
self.assertTrue(rec.bottle_id)
|
||||
@@ -91,29 +76,11 @@ class TestRegistryStore(unittest.TestCase):
|
||||
|
||||
def test_attribute_ambiguous_ip_denied(self) -> None:
|
||||
# Two active bottles on one source IP is a misconfiguration — deny
|
||||
# rather than guess (fail-closed), even with a valid token. (Forced
|
||||
# directly: `register` now supersedes, so this can only arise from an
|
||||
# orphaned/injected row.)
|
||||
# rather than guess (fail-closed), even with a valid token.
|
||||
a = self.store.register("10.243.0.1", bottle_id="a")
|
||||
self._force_active_row("10.243.0.1", "b")
|
||||
self.store.register("10.243.0.1", bottle_id="b")
|
||||
self.assertIsNone(self.store.attribute("10.243.0.1", a.identity_token))
|
||||
|
||||
def test_reregister_same_source_ip_supersedes(self) -> None:
|
||||
# Re-registering a source IP (reused IP / crash recovery / persisted
|
||||
# DB) must supersede the prior active bottle, not add a second active
|
||||
# row that would fail-closed the whole IP to 403.
|
||||
a = self.store.register("10.243.0.1", bottle_id="a", policy="A")
|
||||
b = self.store.register("10.243.0.1", bottle_id="b", policy="B")
|
||||
# exactly one active row at that IP, and it's the new bottle
|
||||
got = self.store.by_source_ip("10.243.0.1")
|
||||
assert got is not None
|
||||
self.assertEqual("b", got.bottle_id)
|
||||
self.assertEqual("B", got.policy)
|
||||
# the superseded bottle is gone and its token no longer attributes
|
||||
self.assertIsNone(self.store.get("a"))
|
||||
self.assertIsNone(self.store.attribute("10.243.0.1", a.identity_token))
|
||||
self.assertIsNotNone(self.store.attribute("10.243.0.1", b.identity_token))
|
||||
|
||||
def test_state_persists_across_reopen(self) -> None:
|
||||
rec = self.store.register("10.243.0.1")
|
||||
reopened = RegistryStore(self.db)
|
||||
@@ -163,7 +130,7 @@ class TestRegistryStore(unittest.TestCase):
|
||||
def test_by_source_ip_unknown_or_ambiguous_denied(self) -> None:
|
||||
self.assertIsNone(self.store.by_source_ip("10.243.9.9")) # unknown
|
||||
self.store.register("10.243.0.1", bottle_id="a")
|
||||
self._force_active_row("10.243.0.1", "b") # orphaned duplicate
|
||||
self.store.register("10.243.0.1", bottle_id="b")
|
||||
self.assertIsNone(self.store.by_source_ip("10.243.0.1")) # ambiguous
|
||||
|
||||
|
||||
|
||||
@@ -92,18 +92,6 @@ class TestOrchestrator(unittest.TestCase):
|
||||
assert got is not None
|
||||
self.assertEqual('{"routes":[]}', got.policy)
|
||||
|
||||
def test_tokens_held_in_memory_and_cleared_on_teardown(self) -> None:
|
||||
rec = self.orch.launch_bottle("10.243.0.5", tokens={"EGRESS_TOKEN_0": "sk"})
|
||||
self.assertEqual({"EGRESS_TOKEN_0": "sk"}, self.orch.tokens_for(rec.bottle_id))
|
||||
# not persisted in the registry (redacted record has no tokens field)
|
||||
self.assertNotIn("tokens", rec.redacted())
|
||||
self.orch.teardown_bottle(rec.bottle_id)
|
||||
self.assertEqual({}, self.orch.tokens_for(rec.bottle_id))
|
||||
|
||||
def test_tokens_default_empty(self) -> None:
|
||||
rec = self.orch.launch_bottle("10.243.0.6")
|
||||
self.assertEqual({}, self.orch.tokens_for(rec.bottle_id))
|
||||
|
||||
def test_set_policy_live_reload(self) -> None:
|
||||
rec = self.orch.launch_bottle("10.243.0.3")
|
||||
self.assertTrue(self.orch.set_policy(rec.bottle_id, '{"x":1}'))
|
||||
|
||||
@@ -89,17 +89,13 @@ class TestPolicyResolver(unittest.TestCase):
|
||||
self.r.resolve_bottle_id("10.243.0.1", "tok")
|
||||
|
||||
def test_resolve_policy_and_bottle_id_one_call(self) -> None:
|
||||
payload = {"bottle_id": "b1", "policy": "P", "tokens": {"EGRESS_TOKEN_0": "s"}}
|
||||
with patch(_URLOPEN, return_value=_resp(payload)) as m:
|
||||
self.assertEqual(
|
||||
("P", "b1", {"EGRESS_TOKEN_0": "s"}),
|
||||
self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"),
|
||||
)
|
||||
self.assertEqual(1, m.call_count) # policy + id + tokens from a single /resolve
|
||||
with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})) as m:
|
||||
self.assertEqual(("P", "b1"), self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"))
|
||||
self.assertEqual(1, m.call_count) # both from a single /resolve
|
||||
|
||||
def test_resolve_policy_and_bottle_id_403_is_none_none_empty(self) -> None:
|
||||
def test_resolve_policy_and_bottle_id_403_is_none_none(self) -> None:
|
||||
with patch(_URLOPEN, side_effect=_http_error(403)):
|
||||
self.assertEqual((None, None, {}), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
|
||||
self.assertEqual((None, None), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
|
||||
|
||||
def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
|
||||
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
||||
|
||||
Reference in New Issue
Block a user