Compare commits

...

6 Commits

Author SHA1 Message Date
didericis-codex 571ea4dbf0 chore: tighten upkeep boundaries and static checks
tracker-policy-pr / check-pr (pull_request) Successful in 10s
test / integration-docker (pull_request) Successful in 19s
test / unit (pull_request) Successful in 53s
test / coverage (pull_request) Successful in 16s
2026-07-26 06:42:34 +00:00
didericis-codex b8893d4c0c fix(diagnostics): make optional failures observable and safe 2026-07-26 06:42:34 +00:00
didericis-codex 9c3dd003af refactor(backend): extract shared bottle preparation planner 2026-07-26 06:42:34 +00:00
didericis-codex f081fdf7dd refactor(egress): make addon core a compatibility facade 2026-07-26 06:42:34 +00:00
didericis-codex 14827bc151 refactor(egress): separate matching, DLP, and context concerns 2026-07-26 06:42:34 +00:00
didericis-codex 97bd7caa76 refactor: validate reconciliation inputs and neutralize CLI helpers 2026-07-26 06:42:34 +00:00
30 changed files with 1263 additions and 1117 deletions
+11 -75
View File
@@ -23,14 +23,14 @@ from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Generator, Generic, Sequence, TypeVar from typing import Generator, Generic, Sequence, TypeVar
from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan from ..agent_provider import AgentProvisionPlan, get_provider
from ..egress import EgressPlan from ..egress import EgressPlan
from ..git_gate import GitGatePlan from ..git_gate import GitGatePlan
from ..log import die, info from ..log import die, info
from ..util import expand_tilde from ..util import expand_tilde
from ..manifest import Manifest, ManifestIndex from ..manifest import Manifest, ManifestIndex
from ..supervisor.plan import SupervisePlan from ..supervisor.plan import SupervisePlan
from ..env import resolve_env, ResolvedEnv from ..env import ResolvedEnv
from ..workspace import WorkspacePlan, workspace_plan from ..workspace import WorkspacePlan, workspace_plan
from .print_util import print_multi, visible_agent_env_names from .print_util import print_multi, visible_agent_env_names
from .util import host_skill_dir from .util import host_skill_dir
@@ -296,82 +296,18 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
backend-specific resolution (names, scratch files, etc.). The backend-specific resolution (names, scratch files, etc.). The
validation step is enforced here so a future backend cannot validation step is enforced here so a future backend cannot
accidentally skip it. No remote/runtime resources are created.""" accidentally skip it. No remote/runtime resources are created."""
from .resolve_common import ( from .preparation import BottlePreparationPlanner
merge_provision_env_vars, prepared = BottlePreparationPlanner(self).prepare(spec)
mint_slug,
prepare_agent_state_dir,
prepare_egress,
prepare_git_gate,
prepare_supervise,
reject_nested_containers,
resolve_manifest_dockerfile,
write_launch_metadata,
)
manifest = self._validate(spec)
if not self.supports_nested_containers:
reject_nested_containers(self.name, manifest)
self._preflight()
from ..git_gate import GitGate
manifest = GitGate().preflight_host_keys(
manifest,
headless=spec.headless,
home_md=spec.manifest.home_md,
)
manifest_bottle = manifest.bottle
manifest_agent_provider = manifest_bottle.agent_provider
agent_provider = get_provider(manifest_agent_provider.template)
resolved_env = resolve_env(manifest)
workspace = workspace_plan(spec, guest_home=agent_provider.guest_home)
slug = mint_slug(spec)
write_launch_metadata(slug, spec, compose_project="", backend=self.name)
# Manifest may override the Dockerfile per-bottle; otherwise fall
# back to the provider plugin's bundled Dockerfile (next to its
# agent_provider.py module).
if manifest_agent_provider.dockerfile:
agent_dockerfile_path = resolve_manifest_dockerfile(
manifest_agent_provider.dockerfile, spec,
)
else:
agent_dockerfile_path = str(agent_provider.dockerfile)
agent_dir, prompt_file = prepare_agent_state_dir(slug, manifest)
agent_provision_plan = build_agent_provision_plan(
template=manifest_agent_provider.template,
dockerfile=agent_dockerfile_path,
state_dir=agent_dir,
instance_name=f"bot-bottle-{slug}",
prompt_file=prompt_file,
guest_env=self._build_guest_env(resolved_env),
forward_host_credentials=manifest_agent_provider.forward_host_credentials,
auth_token=manifest_agent_provider.auth_token,
host_env=dict(os.environ),
trusted_project_path=workspace.workdir,
label=spec.label,
color=spec.color,
provider_settings=manifest_agent_provider.settings,
)
agent_provision_plan = merge_provision_env_vars(agent_provision_plan)
egress_plan = prepare_egress(manifest_bottle, slug, agent_provision_plan)
supervise_plan = prepare_supervise(manifest_bottle, slug)
git_gate_plan = prepare_git_gate(manifest_bottle, slug)
return self._resolve_plan( return self._resolve_plan(
spec, spec,
manifest=manifest, manifest=prepared.manifest,
slug=slug, slug=prepared.slug,
resolved_env=resolved_env, resolved_env=prepared.resolved_env,
agent_provision_plan=agent_provision_plan, agent_provision_plan=prepared.agent_provision_plan,
egress_plan=egress_plan, egress_plan=prepared.egress_plan,
supervise_plan=supervise_plan, supervise_plan=prepared.supervise_plan,
git_gate_plan=git_gate_plan, git_gate_plan=prepared.git_gate_plan,
stage_dir=stage_dir, stage_dir=stage_dir,
) )
+6 -14
View File
@@ -6,12 +6,17 @@ from __future__ import annotations
import os import os
from datetime import datetime, timezone from datetime import datetime, timezone
import re
import shutil import shutil
import subprocess import subprocess
from typing import Iterator from typing import Iterator
from ...log import die, info from ...log import die, info
from ...util import slugify as _slugify
def slugify(name: str) -> str:
"""Compatibility wrapper; new generic callers import ``bot_bottle.util``."""
return _slugify(name)
def run_docker( def run_docker(
@@ -114,19 +119,6 @@ def docker_cp(src: str, dest: str) -> None:
f"{(result.stderr or '').strip() or '<no stderr>'}") f"{(result.stderr or '').strip() or '<no stderr>'}")
_SLUG_RE = re.compile(r"[^a-z0-9]+")
def slugify(name: str) -> str:
"""Lowercase, non-alnum runs → '-', trimmed. Dies on empty result."""
if not name:
die("slugify: missing name")
slug = _SLUG_RE.sub("-", name.lower()).strip("-")
if not slug:
die(f"name '{name}' produced an empty slug; use alphanumeric characters")
return slug
def build_image(ref: str, context: str, *, dockerfile: str = "") -> None: def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
"""Invokes `docker build` every call. Layer cache makes no-change """Invokes `docker build` every call. Layer cache makes no-change
rebuilds cheap; running every time means Dockerfile edits land rebuilds cheap; running every time means Dockerfile edits land
+2 -1
View File
@@ -11,7 +11,8 @@ from pathlib import Path
from ..bottle_state import egress_state_dir from ..bottle_state import egress_state_dir
from ..egress import EGRESS_ROUTES_FILENAME from ..egress import EGRESS_ROUTES_FILENAME
from ..gateway.egress.addon_core import LOG_OFF, load_config from ..gateway.egress.schema import load_config
from ..gateway.egress.types import LOG_OFF
class EgressApplyError(RuntimeError): class EgressApplyError(RuntimeError):
+124
View File
@@ -0,0 +1,124 @@
"""Backend-neutral preparation planner.
This module owns the shared transformation from a CLI ``BottleSpec`` to the
typed inputs consumed by a concrete backend's ``_resolve_plan``. Backend
classes retain only their validation/preflight/env hooks and their
backend-specific final resolution.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import TYPE_CHECKING, Protocol
from ..agent_provider import AgentProvisionPlan, build_agent_provision_plan, get_provider
from ..egress import EgressPlan
from ..env import ResolvedEnv, resolve_env
from ..git_gate import GitGate, GitGatePlan
from ..manifest import Manifest
from ..supervisor.plan import SupervisePlan
from ..workspace import workspace_plan
from .resolve_common import (
merge_provision_env_vars,
mint_slug,
prepare_agent_state_dir,
prepare_egress,
prepare_git_gate,
prepare_supervise,
reject_nested_containers,
resolve_manifest_dockerfile,
write_launch_metadata,
)
if TYPE_CHECKING:
from .base import BottleSpec
class PreparationBackend(Protocol):
"""Backend hooks needed by the shared planner."""
name: str
supports_nested_containers: bool
def _validate(self, spec: BottleSpec) -> Manifest: ...
def _preflight(self) -> None: ...
def _build_guest_env(self, resolved_env: ResolvedEnv) -> dict[str, str]: ...
@dataclass(frozen=True)
class PreparedBottle:
"""Typed, backend-neutral result of shared launch preparation."""
manifest: Manifest
slug: str
resolved_env: ResolvedEnv
agent_provision_plan: AgentProvisionPlan
egress_plan: EgressPlan
git_gate_plan: GitGatePlan
supervise_plan: SupervisePlan | None
class BottlePreparationPlanner:
"""Run the common, side-effect-limited part of bottle preparation."""
def __init__(self, backend: PreparationBackend) -> None:
self._backend = backend
def prepare(self, spec: BottleSpec) -> PreparedBottle:
backend = self._backend
# These are deliberately protected backend hooks: only this shared
# planner orchestrates them, while concrete backends provide the
# implementation.
manifest = backend._validate(spec) # pylint: disable=protected-access
if not backend.supports_nested_containers:
reject_nested_containers(backend.name, manifest)
backend._preflight() # pylint: disable=protected-access
manifest = GitGate().preflight_host_keys(
manifest,
headless=spec.headless,
home_md=spec.manifest.home_md,
)
bottle = manifest.bottle
provider_config = bottle.agent_provider
provider = get_provider(provider_config.template)
resolved_env = resolve_env(manifest)
workspace = workspace_plan(spec, guest_home=provider.guest_home)
slug = mint_slug(spec)
write_launch_metadata(slug, spec, compose_project="", backend=backend.name)
dockerfile = (
resolve_manifest_dockerfile(provider_config.dockerfile, spec)
if provider_config.dockerfile
else str(provider.dockerfile)
)
agent_dir, prompt_file = prepare_agent_state_dir(slug, manifest)
provision = build_agent_provision_plan(
template=provider_config.template,
dockerfile=dockerfile,
state_dir=agent_dir,
instance_name=f"bot-bottle-{slug}",
prompt_file=prompt_file,
guest_env=backend._build_guest_env( # pylint: disable=protected-access
resolved_env
),
forward_host_credentials=provider_config.forward_host_credentials,
auth_token=provider_config.auth_token,
host_env=dict(os.environ),
trusted_project_path=workspace.workdir,
label=spec.label,
color=spec.color,
provider_settings=provider_config.settings,
)
provision = merge_provision_env_vars(provision)
return PreparedBottle(
manifest=manifest,
slug=slug,
resolved_env=resolved_env,
agent_provision_plan=provision,
egress_plan=prepare_egress(bottle, slug, provision),
git_gate_plan=prepare_git_gate(bottle, slug),
supervise_plan=prepare_supervise(bottle, slug),
)
+2 -2
View File
@@ -30,6 +30,7 @@ from ..log import die
from ..manifest import Manifest, ManifestBottle from ..manifest import Manifest, ManifestBottle
from ..supervisor.plan import SupervisePlan from ..supervisor.plan import SupervisePlan
from ..orchestrator.supervisor import Supervisor from ..orchestrator.supervisor import Supervisor
from ..util import slugify
from . import BottleSpec from . import BottleSpec
@@ -44,8 +45,7 @@ def mint_slug(spec: BottleSpec) -> str:
if spec.identity: if spec.identity:
return spec.identity return spec.identity
if spec.label: if spec.label:
from .docker import util as docker_mod return slugify(spec.label)
return docker_mod.slugify(spec.label)
return bottle_identity(spec.agent_name) return bottle_identity(spec.agent_name)
+8 -9
View File
@@ -25,12 +25,11 @@ from typing import Callable
from ...agent_provider import get_provider, runtime_for from ...agent_provider import get_provider, runtime_for
from ...backend import ( from ...backend import (
Bottle, Bottle,
BottlePlan,
BottleSpec, BottleSpec,
enumerate_active_agents, enumerate_active_agents,
get_bottle_backend, get_bottle_backend,
) )
from ...backend.docker import util as docker_mod
from ...backend.docker.bottle_plan import DockerBottlePlan
from ...bottle_state import ( from ...bottle_state import (
cleanup_state, cleanup_state,
is_preserved, is_preserved,
@@ -40,7 +39,7 @@ from ...image_cache import StaleImageError
from ...log import info, die from ...log import info, die
from ...manifest import Manifest, ManifestIndex from ...manifest import Manifest, ManifestIndex
from ..constants import PROG from ..constants import PROG
from ...util import read_tty_line from ...util import read_tty_line, slugify
from .. import tui from .. import tui
@@ -257,10 +256,10 @@ def _uniquify_label_headless(label: str) -> str:
logging the chosen label. Orchestrators fire-and-forget many bottles, logging the chosen label. Orchestrators fire-and-forget many bottles,
so silently picking a free name beats erroring on every collision.""" so silently picking a free name beats erroring on every collision."""
active_slugs = {a.slug for a in enumerate_active_agents()} active_slugs = {a.slug for a in enumerate_active_agents()}
if docker_mod.slugify(label) not in active_slugs: if slugify(label) not in active_slugs:
return label return label
n = 2 n = 2
while docker_mod.slugify(f"{label}-{n}") in active_slugs: while slugify(f"{label}-{n}") in active_slugs:
n += 1 n += 1
chosen = f"{label}-{n}" chosen = f"{label}-{n}"
info(f"label '{label}' already in use; using '{chosen}'") info(f"label '{label}' already in use; using '{chosen}'")
@@ -274,11 +273,11 @@ def prepare_with_preflight(
spec: BottleSpec, spec: BottleSpec,
*, *,
stage_dir: Path, stage_dir: Path,
render_preflight: Callable[[DockerBottlePlan, str], None], render_preflight: Callable[[BottlePlan, str], None],
prompt_yes: Callable[[], bool], prompt_yes: Callable[[], bool],
dry_run: bool = False, dry_run: bool = False,
backend_name: str | None = None, backend_name: str | None = None,
) -> tuple[DockerBottlePlan | None, str]: ) -> tuple[BottlePlan | None, str]:
"""Run `backend.prepare`, render the preflight summary via the """Run `backend.prepare`, render the preflight summary via the
injected callable, prompt y/N via the injected callable. injected callable, prompt y/N via the injected callable.
@@ -405,7 +404,7 @@ def _resolve_unique_label(label: str, color: str) -> tuple[str, str]:
in use among running bottles. Passes through unchanged when no in use among running bottles. Passes through unchanged when no
collision is found on the first check.""" collision is found on the first check."""
while True: while True:
slug_candidate = docker_mod.slugify(label) slug_candidate = slugify(label)
active_slugs = {a.slug for a in enumerate_active_agents()} active_slugs = {a.slug for a in enumerate_active_agents()}
if slug_candidate not in active_slugs: if slug_candidate not in active_slugs:
return label, color return label, color
@@ -432,7 +431,7 @@ def _select_image_policy() -> str | None:
def _text_render_preflight(): def _text_render_preflight():
def _render(plan: DockerBottlePlan, backend_name: str) -> None: def _render(plan: BottlePlan, backend_name: str) -> None:
print(file=sys.stderr) print(file=sys.stderr)
print(f"backend: {backend_name}", file=sys.stderr) print(f"backend: {backend_name}", file=sys.stderr)
print(_manifest_to_yaml(plan.manifest), file=sys.stderr) print(_manifest_to_yaml(plan.manifest), file=sys.stderr)
+22 -19
View File
@@ -368,7 +368,7 @@ def _main_loop(stdscr: "curses._CursesWindow") -> None: # type: ignore # pragm
elif key in (curses.KEY_UP, ord("k")): elif key in (curses.KEY_UP, ord("k")):
selected = max(selected - 1, 0) selected = max(selected - 1, 0)
elif key in (curses.KEY_ENTER, 10, 13): elif key in (curses.KEY_ENTER, 10, 13):
_detail_view(stdscr, qp, green_attr=green_attr) status_line = _detail_view(stdscr, qp, green_attr=green_attr)
elif key == ord("a"): elif key == ord("a"):
try: try:
status_line = _approve_from_tui(stdscr, qp) status_line = _approve_from_tui(stdscr, qp)
@@ -456,7 +456,7 @@ def _detail_view(
qp: QueuedProposal, qp: QueuedProposal,
*, *,
green_attr: int = 0, green_attr: int = 0,
) -> None: # pragma: no cover ) -> str: # pragma: no cover
"""Render the full proposal. Scrollable. Press q to return.""" """Render the full proposal. Scrollable. Press q to return."""
lines = _detail_lines(qp, green_attr=green_attr) lines = _detail_lines(qp, green_attr=green_attr)
offset = 0 offset = 0
@@ -473,7 +473,7 @@ def _detail_view(
stdscr.refresh() stdscr.refresh()
key = stdscr.getch() key = stdscr.getch()
if key in (ord("q"), 27): if key in (ord("q"), 27):
return return ""
if key in (curses.KEY_DOWN, ord("j")): if key in (curses.KEY_DOWN, ord("j")):
offset = min(offset + 1, max(0, len(lines) - 1)) offset = min(offset + 1, max(0, len(lines) - 1))
elif key in (curses.KEY_UP, ord("k")): elif key in (curses.KEY_UP, ord("k")):
@@ -484,31 +484,34 @@ def _detail_view(
offset = max(0, len(lines) - 1) offset = max(0, len(lines) - 1)
elif key == ord("a"): elif key == ord("a"):
try: try:
_approve_from_tui(stdscr, qp) return _approve_from_tui(stdscr, qp)
except ApplyError: except ApplyError as exc:
pass return f"apply failed: {exc}"
return
elif key == ord("m"): elif key == ord("m"):
if qp.proposal.tool in _REPORT_ONLY_TOOLS: if qp.proposal.tool in _REPORT_ONLY_TOOLS:
return return f"modify unavailable for {qp.proposal.tool}"
edited = _modify(stdscr, qp) edited = _modify(stdscr, qp)
if edited is not None: if edited is None:
try: return "modify aborted (no change)"
_approve_from_tui( try:
stdscr, qp, final_file=edited, return _approve_from_tui(
notes="operator modified before approving", stdscr, qp, final_file=edited,
) notes="operator modified before approving",
except ApplyError: )
pass except ApplyError as exc:
return return f"apply failed: {exc}"
elif key == ord("r"): elif key == ord("r"):
reason = _prompt(stdscr, "reject reason: ") reason = _prompt(stdscr, "reject reason: ")
if reason: if reason:
reject(qp, reason=reason) reject(qp, reason=reason)
return return f"rejected {qp.proposal.tool} for [{qp.label}]"
return "reject aborted (empty reason)"
def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None: # type: ignore # pragma: no cover def _modify(
stdscr: "curses._CursesWindow", # type: ignore
qp: QueuedProposal,
) -> str | None: # pragma: no cover
"""Suspend curses, open $EDITOR on the proposed file, return edited content.""" """Suspend curses, open $EDITOR on the proposed file, return edited content."""
suffix = _suffix_for_tool(qp.proposal.tool) suffix = _suffix_for_tool(qp.proposal.tool)
curses.endwin() curses.endwin()
+32 -6
View File
@@ -16,6 +16,8 @@ import os
import sys import sys
from typing import Any, Optional from typing import Any, Optional
from ..log import debug
def filter_multiselect( def filter_multiselect(
items: list[str], items: list[str],
@@ -42,7 +44,11 @@ def filter_multiselect(
try: try:
tty_fd = open(tty_path, "r+b", buffering=0) tty_fd = open(tty_path, "r+b", buffering=0)
except OSError: except OSError as exc:
debug(
"multi-select unavailable; treating it as cancellation",
context={"error_type": type(exc).__name__, "tty": tty_path},
)
return None return None
try: try:
@@ -73,7 +79,11 @@ def filter_select(
try: try:
tty_fd = open(tty_path, "r+b", buffering=0) tty_fd = open(tty_path, "r+b", buffering=0)
except OSError: except OSError as exc:
debug(
"filter-select unavailable; treating it as cancellation",
context={"error_type": type(exc).__name__, "tty": tty_path},
)
return None return None
try: try:
@@ -129,7 +139,11 @@ def _run_picker(items: list[str], *, title: str, tty_fd: int) -> Optional[str]:
curses.nocbreak() curses.nocbreak()
curses.echo() curses.echo()
curses.endwin() curses.endwin()
except Exception: # noqa: W0718 — curses can raise many error types except Exception as exc: # noqa: W0718 — curses can raise many error types
debug(
"filter-select display failed; treating it as cancellation",
context={"error_type": type(exc).__name__},
)
return None return None
finally: finally:
sys.__stdin__ = orig_stdin # type: ignore[assignment] sys.__stdin__ = orig_stdin # type: ignore[assignment]
@@ -292,7 +306,11 @@ def _run_multiselect(
curses.nocbreak() curses.nocbreak()
curses.echo() curses.echo()
curses.endwin() curses.endwin()
except Exception: # noqa: W0718 except Exception as exc: # noqa: W0718
debug(
"multi-select display failed; treating it as cancellation",
context={"error_type": type(exc).__name__},
)
return None return None
finally: finally:
sys.__stdin__ = orig_stdin # type: ignore[assignment] sys.__stdin__ = orig_stdin # type: ignore[assignment]
@@ -558,13 +576,21 @@ def name_color_modal(
""" """
try: try:
tty_fd = open(tty_path, "r+b", buffering=0) # pylint: disable=consider-using-with tty_fd = open(tty_path, "r+b", buffering=0) # pylint: disable=consider-using-with
except OSError: except OSError as exc:
debug(
"name/color picker unavailable; using defaults",
context={"error_type": type(exc).__name__, "tty": tty_path},
)
return default_label, "" return default_label, ""
try: try:
fd_dup = os.dup(tty_fd.fileno()) fd_dup = os.dup(tty_fd.fileno())
return _run_name_color(default_label, tty_fd=fd_dup, disclaimer=disclaimer) return _run_name_color(default_label, tty_fd=fd_dup, disclaimer=disclaimer)
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught except Exception as exc: # noqa: BLE001 # pylint: disable=broad-exception-caught
debug(
"name/color picker failed; using defaults",
context={"error_type": type(exc).__name__},
)
return default_label, "" return default_label, ""
finally: finally:
tty_fd.close() tty_fd.close()
+2 -2
View File
@@ -11,7 +11,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from ..gateway.egress.addon_core import Route from ..gateway.egress.types import Route
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -19,7 +19,7 @@ class EgressRoute(Route):
"""Host-side extension of the addon's `Route`. """Host-side extension of the addon's `Route`.
Inherits `host`, `matches`, `auth_scheme`, and `token_env` Inherits `host`, `matches`, `auth_scheme`, and `token_env`
from `egress_addon_core.Route` those are the fields that cross the from the gateway's wire `Route` — those are the fields that cross the
YAML wire into the gateway. The fields below are host-only and YAML wire into the gateway. The fields below are host-only and
are never serialised to the addon. are never serialised to the addon.
+2 -2
View File
@@ -14,8 +14,8 @@ import secrets
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from ..gateway.egress.addon_core import ( from ..gateway.egress.dlp_config import ON_MATCH_REDACT
ON_MATCH_REDACT, from ..gateway.egress.types import (
HeaderMatch as CoreHeaderMatch, HeaderMatch as CoreHeaderMatch,
MatchEntry as CoreMatchEntry, MatchEntry as CoreMatchEntry,
PathMatch as CorePathMatch, PathMatch as CorePathMatch,
+17 -11
View File
@@ -17,28 +17,34 @@ from mitmproxy import http # type: ignore[import-not-found] # pylint: disable=
from bot_bottle.constants import IDENTITY_HEADER from bot_bottle.constants import IDENTITY_HEADER
from bot_bottle.gateway.egress.dlp_detectors import redact_tokens, strip_crlf from bot_bottle.gateway.egress.dlp_detectors import redact_tokens, strip_crlf
from bot_bottle.gateway.egress.addon_core import ( from bot_bottle.gateway.egress.dlp_config import (
LOG_BLOCKS,
LOG_FULL,
DEFAULT_OUTBOUND_ON_MATCH, DEFAULT_OUTBOUND_ON_MATCH,
ON_MATCH_BLOCK, ON_MATCH_BLOCK,
ON_MATCH_REDACT, ON_MATCH_REDACT,
Config, )
Route, from bot_bottle.gateway.egress.context import resolve_client_context
ScanResult, from bot_bottle.gateway.egress.dlp import (
build_inbound_scan_text, build_inbound_scan_text,
build_outbound_scan_text, build_outbound_scan_text,
build_token_allow_payload, build_token_allow_payload,
outbound_scan_headers,
scan_inbound,
scan_outbound,
)
from bot_bottle.gateway.egress.matching import (
decide, decide,
decide_git_fetch, decide_git_fetch,
is_git_fetch_request, is_git_fetch_request,
is_git_push_request, is_git_push_request,
match_route, match_route,
resolve_client_context, )
outbound_scan_headers, from bot_bottle.gateway.egress.schema import route_to_yaml_dict
route_to_yaml_dict, from bot_bottle.gateway.egress.types import (
scan_inbound, LOG_BLOCKS,
scan_outbound, LOG_FULL,
Config,
Route,
ScanResult,
) )
from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
from bot_bottle.supervisor.types import ( from bot_bottle.supervisor.types import (
File diff suppressed because it is too large Load Diff
+77
View File
@@ -0,0 +1,77 @@
"""Fail-closed resolution of a client's policy and egress credentials."""
from __future__ import annotations
import typing
from ...log import debug
from .types import Config
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."
)
class PolicyResolverLike(typing.Protocol):
def resolve(self, source_ip: str, identity_token: str = ...) -> str | None: ...
class ContextResolverLike(typing.Protocol):
def resolve_policy_and_bottle_id(
self, source_ip: str, identity_token: str = ...,
) -> tuple[str | None, str | None, dict[str, str]]: ...
def _config_from_policy(policy: str | None) -> Config:
# Local import keeps schema parsing independent of resolver protocols.
from .schema import load_config
if not policy:
return Config(routes=(), deny_reason=DENY_UNATTRIBUTED)
try:
return load_config(policy)
except ValueError:
return Config(routes=(), deny_reason=DENY_UNPARSEABLE)
def resolve_client_config(
resolver: PolicyResolverLike, client_ip: str, identity_token: str = "",
) -> Config:
try:
policy = resolver.resolve(client_ip, identity_token)
except Exception as exc: # noqa: BLE001 - a policy lookup failure must deny
debug(
"egress policy resolution failed; applying deny-all",
context={"error_type": type(exc).__name__},
)
return Config(routes=(), deny_reason=DENY_RESOLVER_ERROR)
return _config_from_policy(policy)
def resolve_client_context(
resolver: ContextResolverLike, client_ip: str, identity_token: str = "",
) -> tuple[Config, str, dict[str, str]]:
try:
policy, bottle_id, tokens = resolver.resolve_policy_and_bottle_id(
client_ip, identity_token)
except Exception as exc: # noqa: BLE001 - a policy lookup failure must deny
debug(
"egress context resolution failed; applying deny-all",
context={"error_type": type(exc).__name__},
)
return Config(routes=(), deny_reason=DENY_RESOLVER_ERROR), "", {}
return _config_from_policy(policy), (bottle_id or ""), tokens
+99
View File
@@ -0,0 +1,99 @@
"""DLP scan dispatch and safe proposal rendering for egress requests."""
from __future__ import annotations
import typing
from .types import Route, ScanResult
def build_outbound_scan_text(host: str, path: str, query: str,
headers: typing.Mapping[str, str], body: str) -> str:
parts = [host, path]
if query:
parts.append(query)
parts.extend(f"{name}: {value}" for name, value in headers.items())
if body:
parts.append(body)
return "\n".join(parts)
def outbound_scan_headers(route: Route, headers: typing.Mapping[str, str]) -> dict[str, str]:
"""Drop agent Authorization when the route injects gateway-owned auth."""
skip_auth = bool(route.auth_scheme and route.token_env)
return {name: value for name, value in headers.items()
if not (skip_auth and name.lower() == "authorization")}
def build_inbound_scan_text(headers: typing.Mapping[str, str], body: str) -> str:
parts = [f"{name}: {value}" for name, value in headers.items()]
if body:
parts.append(body)
return "\n".join(parts)
def _enabled(configured: tuple[str, ...] | None, name: str) -> bool:
return configured is None or name in configured
def scan_outbound(route: Route, body: str | bytes, environ: typing.Mapping[str, str], *,
safe_tokens: typing.AbstractSet[str] | None = None,
crlf_text: str | None = None) -> ScanResult | None:
if not route.inspect:
return None
try:
from dlp_detectors import ( # type: ignore[import-not-found]
scan_crlf_injection, scan_entropy, scan_known_secrets, scan_token_patterns)
except ImportError: # pragma: no cover - gateway's flat module path
from .dlp_detectors import (
scan_crlf_injection, scan_entropy, scan_known_secrets, scan_token_patterns)
if isinstance(body, bytes):
try:
text = body.decode("utf-8")
except UnicodeDecodeError:
text = body.decode("latin-1")
else:
text = body
result = scan_crlf_injection(text if crlf_text is None else crlf_text)
if result is not None:
return result
if _enabled(route.outbound_detectors, "token_patterns"):
result = scan_token_patterns(text, location="body", safe_tokens=safe_tokens)
if result is not None:
return result
if _enabled(route.outbound_detectors, "known_secrets"):
extra = tuple(prefix for prefix in environ.get(
"BOT_BOTTLE_SENSITIVE_PREFIXES", "").split(",") if prefix)
result = scan_known_secrets(text, location="body", env=environ,
sensitive_prefixes=("EGRESS_TOKEN_",) + extra,
safe_tokens=safe_tokens)
if result is not None:
return result
if route.outbound_detectors is not None and "entropy" in route.outbound_detectors:
return scan_entropy(text, location="body")
return None
def build_token_allow_payload(host: str, method: str, path: str, result: ScanResult) -> str:
"""Render redacted operator context; the raw matched secret is excluded."""
lines = [
"egress blocked an outbound request carrying a detected token",
f"host: {host}", f"method: {method}", f"path: {path}",
f"detector: {result.reason}",
]
if result.context:
lines.append(f"context: {result.context}")
return "\n".join(lines) + "\n"
def scan_inbound(route: Route, body: str | bytes) -> ScanResult | None:
if not route.inspect:
return None
try:
from dlp_detectors import scan_naive_injection # type: ignore[import-not-found]
except ImportError: # pragma: no cover - gateway's flat module path
from .dlp_detectors import scan_naive_injection
text = body if isinstance(body, str) else body.decode("utf-8", errors="replace")
if _enabled(route.inbound_detectors, "naive_injection_detection"):
return scan_naive_injection(text)
return None
+1 -1
View File
@@ -19,7 +19,7 @@ from math import log2
from collections import Counter from collections import Counter
from urllib.parse import quote as url_quote from urllib.parse import quote as url_quote
from .addon_core import ScanResult from .types import ScanResult
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+112
View File
@@ -0,0 +1,112 @@
"""Route matching and request-policy decisions for the egress gateway."""
from __future__ import annotations
import typing
from .types import Decision, MatchEntry, PathMatch, Route
def _path_matches(pm: PathMatch, request_path: str) -> bool:
if pm.type == "exact":
return request_path == pm.value
if pm.type == "prefix":
if request_path == pm.value:
return True
if not pm.value.endswith("/"):
return request_path.startswith(pm.value + "/")
return request_path.startswith(pm.value)
return (
pm.type == "regex"
and pm.compiled is not None
and pm.compiled.search(request_path) is not None
)
def _entry_matches(
entry: MatchEntry, request_path: str, request_method: str,
request_headers: typing.Mapping[str, str],
) -> bool:
if entry.paths and not any(_path_matches(pm, request_path) for pm in entry.paths):
return False
if entry.methods and request_method.upper() not in entry.methods:
return False
for match in entry.headers:
value = request_headers.get(match.name.lower())
if value is None:
return False
if match.type == "exact" and value != match.value:
return False
if match.type == "regex" and (
match.compiled is None or match.compiled.search(value) is None
):
return False
return True
def evaluate_matches(
route: Route, request_path: str, request_method: str = "GET",
request_headers: typing.Mapping[str, str] | None = None,
) -> bool:
"""Return whether a request satisfies a route's optional match entries."""
if not route.matches:
return True
return any(_entry_matches(entry, request_path, request_method, request_headers or {})
for entry in route.matches)
def is_git_push_request(path: str, query: str) -> bool:
return path.endswith("/git-receive-pack") or (
path.endswith("/info/refs") and any(
pair.partition("=") == ("service", "=", "git-receive-pack")
for pair in query.split("&")
)
)
def is_git_fetch_request(path: str, query: str) -> bool:
return path.endswith("/git-upload-pack") or (
path.endswith("/info/refs") and any(
pair.partition("=") == ("service", "=", "git-upload-pack")
for pair in query.split("&")
)
)
def match_route(routes: typing.Sequence[Route], request_host: str) -> Route | None:
target = request_host.lower()
return next((route for route in routes if route.host.lower() == target), None)
def decide(
routes: typing.Sequence[Route], request_host: str, request_path: str,
environ: typing.Mapping[str, str], *, request_method: str = "GET",
request_headers: typing.Mapping[str, str] | None = None, deny_reason: str = "",
) -> Decision:
route = match_route(routes, request_host)
if route is None:
return Decision("block", deny_reason or (
f"egress: host {request_host!r} is not in the bottle's egress.routes "
"allowlist. Declare a route for it or remove the request."))
if not evaluate_matches(route, request_path, request_method, request_headers):
return Decision("block", (
f"egress: request {request_method} {request_path!r} does not match any "
f"entry in matches for {route.host!r}"))
if route.auth_scheme and route.token_env:
token = environ.get(route.token_env, "")
if not token:
return Decision("block", (
f"egress: route for {route.host!r} declared auth but env var "
f"{route.token_env!r} is unset"))
return Decision("forward", inject_authorization=f"{route.auth_scheme} {token}")
return Decision("forward")
def decide_git_fetch(routes: typing.Sequence[Route], request_host: str) -> Decision:
route = match_route(routes, request_host)
if route is not None and route.git_fetch:
return Decision("forward")
return Decision("block", (
"egress: git fetch/clone over HTTPS is not allowed by default; use git-gate "
"for declared repos or set egress.routes[].git.fetch=true for explicit "
"read-only HTTPS Git access."))
+349
View File
@@ -0,0 +1,349 @@
"""Egress policy schema parsing and serialization (PRD 0017 / 0053)."""
from __future__ import annotations
import re
import typing
from ...yaml_subset import YamlSubsetError, parse_yaml_subset
from .dlp_config import parse_inspect_block
from .types import (
HEADER_MATCH_TYPES,
LOG_BLOCKS,
LOG_FULL,
LOG_OFF,
PATH_MATCH_TYPES,
VALID_METHODS,
Config,
HeaderMatch,
MatchEntry,
PathMatch,
Route,
)
# Parsing
# ---------------------------------------------------------------------------
def _parse_path_match(idx: int, j: int, raw: object) -> PathMatch:
label = f"route[{idx}] matches paths[{j}]"
if not isinstance(raw, dict):
raise ValueError(f"{label}: must be an object")
raw_dict: dict[str, object] = typing.cast(dict[str, object], raw)
ptype = raw_dict.get("type", "prefix")
if not isinstance(ptype, str) or ptype not in PATH_MATCH_TYPES:
raise ValueError(
f"{label}: 'type' must be one of {', '.join(PATH_MATCH_TYPES)} "
f"(got {ptype!r})"
)
value = raw_dict.get("value")
if not isinstance(value, str) or not value:
raise ValueError(f"{label}: 'value' must be a non-empty string")
if ptype in ("exact", "prefix") and not value.startswith("/"):
raise ValueError(
f"{label}: value {value!r} must start with '/' for "
f"type {ptype!r}"
)
compiled: re.Pattern[str] | None = None
if ptype == "regex":
try:
compiled = re.compile(value)
except re.error as e:
raise ValueError(
f"{label}: regex {value!r} failed to compile: {e}"
) from e
for k in raw_dict:
if k not in ("type", "value"):
raise ValueError(f"{label}: unknown key {k!r}")
return PathMatch(type=ptype, value=value, compiled=compiled)
def _parse_header_match(idx: int, j: int, raw: object) -> HeaderMatch:
label = f"route[{idx}] matches headers[{j}]"
if not isinstance(raw, dict):
raise ValueError(f"{label}: must be an object")
raw_dict: dict[str, object] = typing.cast(dict[str, object], raw)
name = raw_dict.get("name")
if not isinstance(name, str) or not name:
raise ValueError(f"{label}: 'name' must be a non-empty string")
value = raw_dict.get("value")
if not isinstance(value, str):
raise ValueError(f"{label}: 'value' must be a string")
htype = raw_dict.get("type", "exact")
if not isinstance(htype, str) or htype not in HEADER_MATCH_TYPES:
raise ValueError(
f"{label}: 'type' must be one of {', '.join(HEADER_MATCH_TYPES)} "
f"(got {htype!r})"
)
compiled: re.Pattern[str] | None = None
if htype == "regex":
try:
compiled = re.compile(value)
except re.error as e:
raise ValueError(
f"{label}: regex {value!r} failed to compile: {e}"
) from e
for k in raw_dict:
if k not in ("name", "value", "type"):
raise ValueError(f"{label}: unknown key {k!r}")
return HeaderMatch(name=name, value=value, type=htype, compiled=compiled)
def _parse_match_entry(idx: int, k: int, raw: object) -> MatchEntry:
label = f"route[{idx}] matches[{k}]"
if not isinstance(raw, dict):
raise ValueError(f"{label}: must be an object")
raw_dict: dict[str, object] = typing.cast(dict[str, object], raw)
paths: tuple[PathMatch, ...] = ()
paths_raw = raw_dict.get("paths")
if paths_raw is not None:
if not isinstance(paths_raw, list):
raise ValueError(f"{label}: 'paths' must be a list")
paths_list = typing.cast(list[object], paths_raw)
paths = tuple(_parse_path_match(idx, j, p) for j, p in enumerate(paths_list))
methods: tuple[str, ...] = ()
methods_raw = raw_dict.get("methods")
if methods_raw is not None:
if not isinstance(methods_raw, list):
raise ValueError(f"{label}: 'methods' must be a list")
methods_list = typing.cast(list[object], methods_raw)
normalised: list[str] = []
for j, m in enumerate(methods_list):
if not isinstance(m, str):
raise ValueError(f"{label}: methods[{j}] must be a string")
upper = m.upper()
if upper not in VALID_METHODS:
raise ValueError(
f"{label}: methods[{j}] {m!r} is not a valid HTTP method"
)
normalised.append(upper)
methods = tuple(normalised)
headers: tuple[HeaderMatch, ...] = ()
headers_raw = raw_dict.get("headers")
if headers_raw is not None:
if not isinstance(headers_raw, list):
raise ValueError(f"{label}: 'headers' must be a list")
headers_list = typing.cast(list[object], headers_raw)
headers = tuple(
_parse_header_match(idx, j, h) for j, h in enumerate(headers_list)
)
for key in raw_dict:
if key not in ("paths", "methods", "headers"):
raise ValueError(f"{label}: unknown key {key!r}")
return MatchEntry(paths=paths, methods=methods, headers=headers)
def parse_routes(payload: object) -> tuple[Route, ...]:
if not isinstance(payload, dict):
raise ValueError("routes payload: top-level must be an object")
payload_dict: dict[str, object] = typing.cast(dict[str, object], payload)
raw: object = payload_dict.get("routes")
if not isinstance(raw, list):
raise ValueError("routes payload: 'routes' must be a list")
raw_list: list[object] = typing.cast(list[object], raw)
out: list[Route] = []
for i, r in enumerate(raw_list):
out.append(_parse_one(i, r))
return tuple(out)
def _parse_one(idx: int, raw: object) -> Route:
label = f"route[{idx}]"
if not isinstance(raw, dict):
raise ValueError(f"{label}: must be an object (got {type(raw).__name__})")
raw_dict: dict[str, object] = typing.cast(dict[str, object], raw)
host: object = raw_dict.get("host")
if not isinstance(host, str) or not host:
raise ValueError(f"{label}: 'host' must be a non-empty string")
legacy_flat = "inspect" not in raw_dict
inspect_raw = raw_dict.get("inspect", {})
if inspect_raw is False:
inspect = False
settings: dict[str, object] = {}
elif isinstance(inspect_raw, dict):
inspect = True
settings = (
{k: v for k, v in raw_dict.items() if k != "host"}
if legacy_flat
else typing.cast(dict[str, object], inspect_raw)
)
legacy_dlp = settings.pop("dlp", None)
if isinstance(legacy_dlp, dict):
settings.update(typing.cast(dict[str, object], legacy_dlp))
elif legacy_dlp is not None:
raise ValueError(
f"{label} ({host}): legacy 'dlp' must be an object"
)
else:
raise ValueError(f"{label} ({host}): 'inspect' must be false or an object")
# matches
matches: tuple[MatchEntry, ...] = ()
matches_raw = settings.get("matches")
if matches_raw is not None:
if not isinstance(matches_raw, list):
raise ValueError(f"{label} ({host}): 'matches' must be a list")
matches_list = typing.cast(list[object], matches_raw)
matches = tuple(
_parse_match_entry(idx, k, m) for k, m in enumerate(matches_list)
)
# auth (unchanged wire format)
auth_scheme: object = settings.get("auth_scheme", "")
token_env: object = settings.get("token_env", "")
if not isinstance(auth_scheme, str):
raise ValueError(f"{label} ({host}): 'auth_scheme' must be a string")
if not isinstance(token_env, str):
raise ValueError(f"{label} ({host}): 'token_env' must be a string")
if bool(auth_scheme) != bool(token_env):
raise ValueError(
f"{label} ({host}): 'auth_scheme' and 'token_env' must be both "
f"set or both empty (got auth_scheme={auth_scheme!r}, "
f"token_env={token_env!r})"
)
# git-over-HTTPS policy
git_fetch = False
git_raw = settings.get("git")
if git_raw is not None:
if not isinstance(git_raw, dict):
raise ValueError(f"{label} ({host}): 'git' must be an object")
git_dict: dict[str, object] = typing.cast(dict[str, object], git_raw)
fetch_raw = git_dict.get("fetch", False)
if fetch_raw is True or fetch_raw is False:
git_fetch = fetch_raw
else:
raise ValueError(f"{label} ({host}): 'git.fetch' must be a boolean")
for k in git_dict:
if k != "fetch":
raise ValueError(
f"{label} ({host}): git has unknown key {k!r}; "
"accepted key is 'fetch'"
)
# dlp detectors
outbound_detectors, inbound_detectors, outbound_on_match = parse_inspect_block(
idx, host, settings,
)
preserve_auth_raw = settings.get("preserve_auth", False)
if preserve_auth_raw is not True and preserve_auth_raw is not False:
raise ValueError(
f"{label} ({host}): 'preserve_auth' must be a boolean"
)
preserve_auth: bool = preserve_auth_raw
for k in settings:
if k not in (
"matches", "auth_scheme", "token_env", "git", "preserve_auth",
"outbound_detectors", "inbound_detectors", "outbound_on_match",
):
raise ValueError(
f"{label} ({host}): inspect has unknown key {k!r}"
)
for k in raw_dict:
if not legacy_flat and k not in ("host", "inspect"):
raise ValueError(
f"{label} ({host}): unknown key {k!r}; accepted keys "
f"are 'host' and 'inspect'"
)
return Route(
host=host,
matches=matches,
auth_scheme=auth_scheme,
token_env=token_env,
git_fetch=git_fetch,
outbound_detectors=outbound_detectors,
inbound_detectors=inbound_detectors,
outbound_on_match=outbound_on_match,
preserve_auth=preserve_auth,
inspect=inspect,
)
def _path_match_to_dict(pm: PathMatch) -> dict[str, object]:
d: dict[str, object] = {"value": pm.value}
if pm.type != "prefix":
d["type"] = pm.type
return d
def _header_match_to_dict(hm: HeaderMatch) -> dict[str, object]:
d: dict[str, object] = {"name": hm.name, "value": hm.value}
if hm.type != "exact":
d["type"] = hm.type
return d
def _match_entry_to_dict(me: MatchEntry) -> dict[str, object]:
d: dict[str, object] = {}
if me.paths:
d["paths"] = [_path_match_to_dict(p) for p in me.paths]
if me.methods:
d["methods"] = list(me.methods)
if me.headers:
d["headers"] = [_header_match_to_dict(h) for h in me.headers]
return d
def route_to_yaml_dict(r: Route) -> dict[str, object]:
"""Serialize a Route to YAML-schema-compatible dict.
Uses the same field names the YAML parser accepts, so the output
can be round-tripped directly into an `allow` or `egress-block`
proposal without translation. Fields that are empty/default are
omitted so the agent doesn't copy irrelevant keys."""
d: dict[str, object] = {"host": r.host}
if not r.inspect:
d["inspect"] = False
return d
inspected: dict[str, object] = {}
if r.auth_scheme:
inspected["auth_scheme"] = r.auth_scheme
inspected["token_env"] = r.token_env
if r.matches:
inspected["matches"] = [_match_entry_to_dict(m) for m in r.matches]
if r.git_fetch:
inspected["git"] = {"fetch": True}
if r.outbound_detectors is not None:
inspected["outbound_detectors"] = list(r.outbound_detectors)
if r.inbound_detectors is not None:
inspected["inbound_detectors"] = list(r.inbound_detectors)
if r.outbound_on_match:
inspected["outbound_on_match"] = r.outbound_on_match
if r.preserve_auth:
inspected["preserve_auth"] = True
if inspected:
d["inspect"] = inspected
return d
def parse_config(payload: object) -> "Config":
"""Parse a full egress config payload (top-level log level + routes)."""
if not isinstance(payload, dict):
raise ValueError("routes payload: top-level must be an object")
payload_dict: dict[str, object] = typing.cast(dict[str, object], payload)
log_raw: object = payload_dict.get("log", LOG_OFF)
if log_raw is True or log_raw is False or not isinstance(log_raw, int) \
or log_raw not in (LOG_OFF, LOG_BLOCKS, LOG_FULL):
raise ValueError(
f"routes payload: 'log' must be {LOG_OFF}, {LOG_BLOCKS}, or {LOG_FULL}"
)
routes = parse_routes(payload)
return Config(routes=routes, log=log_raw)
def load_config(text: str) -> "Config":
"""Parse YAML text → Config (routes + log flag)."""
try:
payload = parse_yaml_subset(text)
except YamlSubsetError as e:
raise ValueError(f"routes payload: invalid YAML: {e}") from e
return parse_config(payload)
+81
View File
@@ -0,0 +1,81 @@
"""Shared egress policy value objects.
Kept dependency-free so the schema parser, matcher, DLP scanner, and addon
adapter can use the same immutable public shapes without importing each other.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
PATH_MATCH_TYPES = ("exact", "prefix", "regex")
HEADER_MATCH_TYPES = ("exact", "regex")
VALID_METHODS = frozenset({
"GET", "HEAD", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "TRACE",
"CONNECT",
})
LOG_OFF = 0
LOG_BLOCKS = 1
LOG_FULL = 2
@dataclass(frozen=True)
class PathMatch:
type: str
value: str
compiled: re.Pattern[str] | None = None
@dataclass(frozen=True)
class HeaderMatch:
name: str
value: str
type: str = "exact"
compiled: re.Pattern[str] | None = None
@dataclass(frozen=True)
class MatchEntry:
paths: tuple[PathMatch, ...] = ()
methods: tuple[str, ...] = ()
headers: tuple[HeaderMatch, ...] = ()
@dataclass(frozen=True)
class Route:
host: str
matches: tuple[MatchEntry, ...] = ()
auth_scheme: str = ""
token_env: str = ""
git_fetch: bool = False
outbound_detectors: tuple[str, ...] | None = None
inbound_detectors: tuple[str, ...] | None = None
outbound_on_match: str = ""
preserve_auth: bool = False
inspect: bool = True
@dataclass(frozen=True)
class Config:
routes: tuple[Route, ...]
log: int = LOG_OFF
deny_reason: str = ""
@dataclass(frozen=True)
class Decision:
action: str
reason: str = ""
inject_authorization: str | None = None
@dataclass(frozen=True)
class ScanResult:
severity: str
reason: str
location: str = ""
context: str = ""
matched: str = ""
+3 -3
View File
@@ -58,9 +58,9 @@ import typing
from dataclasses import dataclass from dataclasses import dataclass
from bot_bottle.constants import IDENTITY_HEADER from bot_bottle.constants import IDENTITY_HEADER
from bot_bottle.gateway.egress.addon_core import ( from bot_bottle.gateway.egress.context import resolve_client_context
LOG_OFF, load_config, resolve_client_context, route_to_yaml_dict, from bot_bottle.gateway.egress.schema import load_config, route_to_yaml_dict
) from bot_bottle.gateway.egress.types import LOG_OFF
from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
from bot_bottle.supervisor import types as _sv from bot_bottle.supervisor import types as _sv
+33 -6
View File
@@ -18,6 +18,7 @@ import urllib.request
from collections.abc import Iterable from collections.abc import Iterable
from dataclasses import dataclass from dataclasses import dataclass
from ..log import debug
from ..orchestrator_auth import ROLE_CLI from ..orchestrator_auth import ROLE_CLI
from ..trust_domain import CONTROL_PLANE from ..trust_domain import CONTROL_PLANE
from .server import ORCHESTRATOR_AUTH_HEADER from .server import ORCHESTRATOR_AUTH_HEADER
@@ -53,6 +54,23 @@ class RegisteredBottle:
env_var_secret: str = "" env_var_secret: str = ""
@dataclass(frozen=True)
class BackendProbeFailure:
"""Safe diagnostic for an optional backend discovery probe."""
backend: str
error_type: str
def _probe_failure(backend: str, exc: BaseException) -> BackendProbeFailure:
failure = BackendProbeFailure(backend, type(exc).__name__)
debug(
"orchestrator discovery probe unavailable",
context={"backend": failure.backend, "error_type": failure.error_type},
)
return failure
class OrchestratorClient: class OrchestratorClient:
"""Trusted host-side client for the orchestrator control plane. """Trusted host-side client for the orchestrator control plane.
@@ -245,32 +263,41 @@ def discover_orchestrator_url(*, timeout: float = 2.0) -> str:
orchestrator TAP. Returns the first that answers `/health`; raises if none orchestrator TAP. Returns the first that answers `/health`; raises if none
do (no orchestrator up launch a bottle first).""" do (no orchestrator up launch a bottle first)."""
candidates: list[str] = [] candidates: list[str] = []
failures: list[BackendProbeFailure] = []
try: # docker: loopback-published control plane try: # docker: loopback-published control plane
from .lifecycle import DEFAULT_PORT as _DOCKER_PORT from .lifecycle import DEFAULT_PORT as _DOCKER_PORT
candidates.append(f"http://127.0.0.1:{_DOCKER_PORT}") candidates.append(f"http://127.0.0.1:{_DOCKER_PORT}")
except Exception: # noqa: BLE001 — backend optional except Exception as exc: # noqa: BLE001 — backend optional
failures.append(_probe_failure("docker", exc))
candidates.append("http://127.0.0.1:8099") candidates.append("http://127.0.0.1:8099")
try: # firecracker: infra VM control plane on the orchestrator TAP try: # firecracker: infra VM control plane on the orchestrator TAP
from ..backend.firecracker import netpool from ..backend.firecracker import netpool
from ..backend.firecracker.infra_vm import ORCHESTRATOR_PORT from ..backend.firecracker.infra_vm import ORCHESTRATOR_PORT
candidates.append( candidates.append(
f"http://{netpool.orch_slot().guest_ip}:{ORCHESTRATOR_PORT}") f"http://{netpool.orch_slot().guest_ip}:{ORCHESTRATOR_PORT}")
except Exception: # noqa: BLE001 — backend optional / not firecracker except Exception as exc: # noqa: BLE001 — backend optional / not firecracker
pass failures.append(_probe_failure("firecracker", exc))
try: # macOS: orchestrator container on its host-only address try: # macOS: orchestrator container on its host-only address
from ..backend.macos_container.infra import probe_orchestrator_url from ..backend.macos_container.infra import probe_orchestrator_url
url = probe_orchestrator_url() url = probe_orchestrator_url()
if url: if url:
candidates.append(url) candidates.append(url)
except Exception: # noqa: BLE001 — backend optional / not macOS except Exception as exc: # noqa: BLE001 — backend optional / not macOS
pass failures.append(_probe_failure("macos-container", exc))
for url in candidates: for url in candidates:
if OrchestratorClient(url, timeout=timeout).health(): if OrchestratorClient(url, timeout=timeout).health():
return url return url
detail = ""
if failures:
detail = "; optional probes unavailable: " + ", ".join(
f"{failure.backend} ({failure.error_type})" for failure in failures
)
raise OrchestratorClientError( raise OrchestratorClientError(
"no running orchestrator control plane found (tried " "no running orchestrator control plane found (tried "
+ ", ".join(candidates) + ", ".join(candidates)
+ "); launch a bottle first" + ")"
+ detail
+ "; launch a bottle first"
) )
+9 -1
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
from ..log import debug
from .client import OrchestratorClient, OrchestratorClientError from .client import OrchestratorClient, OrchestratorClientError
@@ -27,7 +28,14 @@ def reprovision_bottles(
try: try:
if client.reprovision_gateway(bottle_id, secret): if client.reprovision_gateway(bottle_id, secret):
restored += 1 restored += 1
except OrchestratorClientError: except OrchestratorClientError as exc:
debug(
"gateway secret reprovision failed; continuing with other bottles",
context={
"bottle_id": bottle_id,
"error_type": type(exc).__name__,
},
)
continue continue
return restored return restored
+20 -8
View File
@@ -57,6 +57,7 @@ from __future__ import annotations
import http.server import http.server
import json import json
import math
import os import os
import socketserver import socketserver
import sys import sys
@@ -217,13 +218,18 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
raw_ips = data.get("live_source_ips") raw_ips = data.get("live_source_ips")
if not isinstance(raw_ips, list): if not isinstance(raw_ips, list):
return 400, {"error": "live_source_ips (list of strings) is required"} return 400, {"error": "live_source_ips (list of strings) is required"}
live = [ip for ip in raw_ips if isinstance(ip, str) and ip] if any(not isinstance(ip, str) or not ip for ip in raw_ips):
return 400, {"error": "live_source_ips must contain non-empty strings"}
live = raw_ips
grace = data.get("grace_seconds") grace = data.get("grace_seconds")
kwargs = ( kwargs: dict[str, float] = {}
{"grace_seconds": float(grace)} if grace is not None:
if isinstance(grace, (int, float)) and not isinstance(grace, bool) if isinstance(grace, bool) or not isinstance(grace, (int, float)):
else {} return 400, {"error": "grace_seconds must be a non-negative finite number"}
) parsed_grace = float(grace)
if not math.isfinite(parsed_grace) or parsed_grace < 0:
return 400, {"error": "grace_seconds must be a non-negative finite number"}
kwargs["grace_seconds"] = parsed_grace
return 200, {"reaped": orch.reconcile(live, **kwargs)} return 200, {"reaped": orch.reconcile(live, **kwargs)}
if method == "POST" and route == "/attribute": if method == "POST" and route == "/attribute":
@@ -373,9 +379,15 @@ class Handler(http.server.BaseHTTPRequestHandler):
status, payload = dispatch( status, payload = dispatch(
server.orchestrator, method, self.path, body, role=role) server.orchestrator, method, self.path, body, role=role)
except Exception as e: # noqa: BLE001 — the control plane must stay up except Exception as e: # noqa: BLE001 — the control plane must stay up
sys.stderr.write(f"orchestrator: {method} {self.path} failed: {e!r}\n") # Do not echo exception messages to the caller or logs: broker and
# persistence exceptions can contain request data. The operation,
# route, and exception type are enough to correlate a traceback.
sys.stderr.write(
f"orchestrator: {method} {self.path} failed "
f"[error_type={type(e).__name__}]\n"
)
sys.stderr.flush() sys.stderr.flush()
status, payload = 500, {"error": f"internal error: {e}"} status, payload = 500, {"error": "internal error"}
data = json.dumps(payload).encode() data = json.dumps(payload).encode()
self.send_response(status) self.send_response(status)
self.send_header("Content-Type", "application/json") self.send_header("Content-Type", "application/json")
+20
View File
@@ -9,8 +9,11 @@ import difflib
import hashlib import hashlib
import ipaddress import ipaddress
import os import os
import re
import sys import sys
from .log import die
def sha256_hex(content: str) -> str: def sha256_hex(content: str) -> str:
"""Hex SHA-256 of a UTF-8 string.""" """Hex SHA-256 of a UTF-8 string."""
@@ -67,3 +70,20 @@ def expand_tilde(path: str) -> str:
home = os.environ.get("HOME", "") home = os.environ.get("HOME", "")
return home + path[1:] return home + path[1:]
return path return path
_SLUG_RE = re.compile(r"[^a-z0-9]+")
def slugify(name: str) -> str:
"""Return a portable bottle identifier from a human-readable name.
This is deliberately a root utility: names are part of the generic CLI
and state model, not a Docker container concern.
"""
if not name:
die("slugify: missing name")
slug = _SLUG_RE.sub("-", name.lower()).strip("-")
if not slug:
die(f"name '{name}' produced an empty slug; use alphanumeric characters")
return slug
@@ -0,0 +1,96 @@
"""Architecture rules that should fail before coupling becomes entrenched."""
from __future__ import annotations
import ast
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
class TestCliBackendBoundaries(unittest.TestCase):
def test_cli_does_not_import_a_concrete_backend(self) -> None:
forbidden = (
"backend.docker", "backend.firecracker", "backend.macos_container",
"bot_bottle.backend.docker", "bot_bottle.backend.firecracker",
"bot_bottle.backend.macos_container",
)
violations: list[str] = []
for path in (ROOT / "bot_bottle" / "cli").rglob("*.py"):
tree = ast.parse(path.read_text(), filename=str(path))
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
module = node.module
if module and module.startswith(forbidden):
violations.append(
f"{path.relative_to(ROOT)}:{node.lineno}: {module}"
)
if isinstance(node, ast.Import):
violations.extend(
f"{path.relative_to(ROOT)}:{node.lineno}: {alias.name}"
for alias in node.names if alias.name.startswith(forbidden)
)
self.assertEqual([], violations, "generic CLI imports concrete backend internals:\n" +
"\n".join(violations))
class TestRuntimeModuleSizes(unittest.TestCase):
def test_no_runtime_module_grows_beyond_global_ceiling(self) -> None:
"""A coarse ceiling catches new monoliths; focused caps stay tighter."""
ceiling = 850
oversized = [
f"{path.relative_to(ROOT)} ({len(path.read_text().splitlines())})"
for path in (ROOT / "bot_bottle").rglob("*.py")
if len(path.read_text().splitlines()) > ceiling
]
self.assertEqual(
[], oversized,
f"runtime modules must stay at or below {ceiling} lines: "
+ ", ".join(oversized),
)
def test_egress_modules_stay_focused(self) -> None:
caps = {
"addon_core.py": 100,
"schema.py": 400,
"types.py": 180,
"matching.py": 180,
"dlp.py": 180,
"context.py": 140,
}
directory = ROOT / "bot_bottle" / "gateway" / "egress"
oversized = [f"{name} ({len((directory / name).read_text().splitlines())}>{cap})"
for name, cap in caps.items()
if len((directory / name).read_text().splitlines()) > cap]
self.assertEqual([], oversized, "split a module rather than raising its cap: " +
", ".join(oversized))
def test_runtime_code_uses_focused_egress_modules(self) -> None:
"""addon_core is compatibility-only, never an internal dependency."""
violations: list[str] = []
package = ROOT / "bot_bottle"
facade = package / "gateway" / "egress" / "addon_core.py"
package_init = package / "gateway" / "egress" / "__init__.py"
for path in package.rglob("*.py"):
if path in (facade, package_init):
continue
text = path.read_text()
if "gateway.egress.addon_core import" in text or \
".addon_core import" in text:
violations.append(str(path.relative_to(ROOT)))
self.assertEqual([], violations)
def test_backend_contract_does_not_absorb_preparation_logic(self) -> None:
caps = {
ROOT / "bot_bottle" / "backend" / "base.py": 580,
ROOT / "bot_bottle" / "backend" / "preparation.py": 160,
}
oversized = [
f"{path.relative_to(ROOT)} "
f"({len(path.read_text().splitlines())}>{cap})"
for path, cap in caps.items()
if len(path.read_text().splitlines()) > cap
]
self.assertEqual([], oversized)
@@ -48,12 +48,13 @@ class TestSharedReprovision(unittest.TestCase):
client.reprovision_gateway.side_effect = [ client.reprovision_gateway.side_effect = [
OrchestratorClientError("bad key"), True, OrchestratorClientError("bad key"), True,
] ]
self.assertEqual( with patch("bot_bottle.orchestrator.reprovision.debug") as debug:
1, count = reprovision_bottles(
reprovision_bottles(
client, {"10.0.0.1": "key-1", "10.0.0.2": "key-2"}, client, {"10.0.0.1": "key-1", "10.0.0.2": "key-2"},
), )
) self.assertEqual(1, count)
self.assertEqual("b1", debug.call_args.kwargs["context"]["bottle_id"])
self.assertNotIn("bad key", repr(debug.call_args))
class TestMacosReprovision(unittest.TestCase): class TestMacosReprovision(unittest.TestCase):
+7 -2
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import unittest import unittest
from typing import Any, Optional from typing import Any, Optional
from unittest.mock import patch
from bot_bottle.cli.tui import _filter_items, _multiselect_loop, filter_multiselect, filter_select from bot_bottle.cli.tui import _filter_items, _multiselect_loop, filter_multiselect, filter_select
@@ -49,8 +50,10 @@ class TestFilterSelectEmptyItems(unittest.TestCase):
def test_returns_none_when_tty_unavailable(self): def test_returns_none_when_tty_unavailable(self):
# /nonexistent is guaranteed to not open. # /nonexistent is guaranteed to not open.
result = filter_select(["a", "b"], tty_path="/nonexistent/tty") with patch("bot_bottle.cli.tui.debug") as debug:
result = filter_select(["a", "b"], tty_path="/nonexistent/tty")
self.assertIsNone(result) self.assertIsNone(result)
self.assertEqual("FileNotFoundError", debug.call_args.kwargs["context"]["error_type"])
class TestFilterMultiselectEmptyItems(unittest.TestCase): class TestFilterMultiselectEmptyItems(unittest.TestCase):
@@ -60,8 +63,10 @@ class TestFilterMultiselectEmptyItems(unittest.TestCase):
self.assertEqual([], result) self.assertEqual([], result)
def test_returns_none_when_tty_unavailable(self): def test_returns_none_when_tty_unavailable(self):
result = filter_multiselect(["a", "b"], tty_path="/nonexistent/tty") with patch("bot_bottle.cli.tui.debug") as debug:
result = filter_multiselect(["a", "b"], tty_path="/nonexistent/tty")
self.assertIsNone(result) self.assertIsNone(result)
self.assertEqual("FileNotFoundError", debug.call_args.kwargs["context"]["error_type"])
class TestMultiselectLoopReordering(unittest.TestCase): class TestMultiselectLoopReordering(unittest.TestCase):
+8 -6
View File
@@ -8,17 +8,19 @@ from __future__ import annotations
import unittest import unittest
from bot_bottle.gateway.egress.addon_core import ( from bot_bottle.gateway.egress.matching import evaluate_matches
HeaderMatch, from bot_bottle.gateway.egress.schema import (
MatchEntry,
PathMatch,
Route,
evaluate_matches,
load_config, load_config,
parse_config, parse_config,
parse_routes, parse_routes,
route_to_yaml_dict, route_to_yaml_dict,
) )
from bot_bottle.gateway.egress.types import (
HeaderMatch,
MatchEntry,
PathMatch,
Route,
)
def _route(d: dict[str, object]) -> Route: def _route(d: dict[str, object]) -> Route:
+10 -3
View File
@@ -3,15 +3,16 @@
from __future__ import annotations from __future__ import annotations
import unittest import unittest
from unittest.mock import patch
from bot_bottle.gateway.egress.addon_core import ( from bot_bottle.gateway.egress.context import (
DENY_RESOLVER_ERROR, DENY_RESOLVER_ERROR,
DENY_UNATTRIBUTED, DENY_UNATTRIBUTED,
DENY_UNPARSEABLE, DENY_UNPARSEABLE,
decide,
resolve_client_config, resolve_client_config,
resolve_client_context, resolve_client_context,
) )
from bot_bottle.gateway.egress.matching import decide
from bot_bottle.gateway.policy_resolver import PolicyResolveError from bot_bottle.gateway.policy_resolver import PolicyResolveError
@@ -44,7 +45,13 @@ class TestResolveClientConfig(unittest.TestCase):
def test_resolver_error_denies_all(self) -> None: def test_resolver_error_denies_all(self) -> None:
# Orchestrator unreachable/errored must never widen egress. # Orchestrator unreachable/errored must never widen egress.
self.assertEqual((), resolve_client_config(_FakeResolver(raises=True), "10.243.0.1").routes) with patch("bot_bottle.gateway.egress.context.debug") as debug:
config = resolve_client_config(_FakeResolver(raises=True), "10.243.0.1")
self.assertEqual((), config.routes)
self.assertEqual(
"PolicyResolveError", debug.call_args.kwargs["context"]["error_type"],
)
self.assertNotIn("orchestrator down", repr(debug.call_args))
def test_unparseable_policy_denies_all(self) -> None: def test_unparseable_policy_denies_all(self) -> None:
cfg = resolve_client_config(_FakeResolver(result="routes: notalist\n"), "10.243.0.1") cfg = resolve_client_config(_FakeResolver(result="routes: notalist\n"), "10.243.0.1")
+11
View File
@@ -12,7 +12,9 @@ from bot_bottle.orchestrator.client import (
OrchestratorClient, OrchestratorClient,
OrchestratorClientError, OrchestratorClientError,
RegisteredBottle, RegisteredBottle,
BackendProbeFailure,
_host_auth_token, _host_auth_token,
_probe_failure,
) )
_URLOPEN = "bot_bottle.orchestrator.client.urllib.request.urlopen" _URLOPEN = "bot_bottle.orchestrator.client.urllib.request.urlopen"
@@ -33,6 +35,15 @@ class TestHostAuthToken(unittest.TestCase):
self.assertEqual("", _host_auth_token()) self.assertEqual("", _host_auth_token())
class TestBackendProbeFailure(unittest.TestCase):
def test_records_safe_typed_diagnostic(self) -> None:
with patch("bot_bottle.orchestrator.client.debug") as debug:
result = _probe_failure("firecracker", RuntimeError("secret detail"))
self.assertEqual(BackendProbeFailure("firecracker", "RuntimeError"), result)
rendered = repr(debug.call_args)
self.assertNotIn("secret detail", rendered)
def _resp(status: int, payload: object) -> MagicMock: def _resp(status: int, payload: object) -> MagicMock:
m = MagicMock() m = MagicMock()
inner = m.__enter__.return_value inner = m.__enter__.return_value
+39 -5
View File
@@ -7,6 +7,7 @@ server tests), plus one real-socket round-trip to prove the handler wiring.
from __future__ import annotations from __future__ import annotations
import base64 import base64
import io
import json import json
import secrets import secrets
import sqlite3 import sqlite3
@@ -17,7 +18,7 @@ import urllib.error
import urllib.request import urllib.request
from contextlib import closing from contextlib import closing
from pathlib import Path from pathlib import Path
from unittest.mock import patch from unittest.mock import MagicMock, patch
from bot_bottle.orchestrator_auth import ROLE_CLI, ROLE_GATEWAY, mint from bot_bottle.orchestrator_auth import ROLE_CLI, ROLE_GATEWAY, mint
from bot_bottle.orchestrator.broker import StubBroker from bot_bottle.orchestrator.broker import StubBroker
@@ -283,6 +284,25 @@ class TestServerRoundTrip(unittest.TestCase):
)) ))
self.assertEqual(reg["bottle_id"], attr["bottle_id"]) self.assertEqual(reg["bottle_id"], attr["bottle_id"])
def test_internal_failure_is_contextual_but_redacted(self) -> None:
orch = MagicMock()
orch.registry.all.side_effect = RuntimeError("SENSITIVE request value")
with patch("sys.stderr", io.StringIO()) as stderr:
server = make_server(orch, "127.0.0.1", 0)
self.addCleanup(server.server_close)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
self.addCleanup(server.shutdown)
host, port = server.server_address[0], server.server_address[1]
with self.assertRaises(urllib.error.HTTPError) as raised:
urllib.request.urlopen(f"http://{host}:{port}/bottles", timeout=5)
payload = json.loads(raised.exception.read())
output = stderr.getvalue()
self.assertEqual({"error": "internal error"}, payload)
self.assertIn("GET /bottles", output)
self.assertIn("RuntimeError", output)
self.assertNotIn("SENSITIVE", output)
class TestOrchestratorAuth(unittest.TestCase): class TestOrchestratorAuth(unittest.TestCase):
"""Role-scoped control-plane tokens (issue #400 / #469 review): every route """Role-scoped control-plane tokens (issue #400 / #469 review): every route
@@ -647,10 +667,24 @@ class TestReconcileRoute(unittest.TestCase):
self.assertEqual(200, status) self.assertEqual(200, status)
self.assertEqual([], payload["reaped"]) self.assertEqual([], payload["reaped"])
def test_non_string_entries_are_ignored(self) -> None: def test_non_string_entries_are_rejected(self) -> None:
dead = self._old("10.0.0.4")
status, payload = dispatch( status, payload = dispatch(
self.orch, "POST", "/reconcile", self.orch, "POST", "/reconcile",
_body({"live_source_ips": [None, 7, "10.0.0.9"]})) _body({"live_source_ips": [None, 7, "10.0.0.9"]}))
self.assertEqual(200, status) self.assertEqual(400, status)
self.assertEqual([dead], payload["reaped"]) self.assertIn("live_source_ips", str(payload["error"]))
def test_empty_live_source_ip_is_rejected(self) -> None:
status, payload = dispatch(
self.orch, "POST", "/reconcile", _body({"live_source_ips": [""]}))
self.assertEqual(400, status)
self.assertIn("live_source_ips", str(payload["error"]))
def test_invalid_grace_seconds_is_rejected(self) -> None:
for value in (True, "30", -1, float("inf"), float("nan")):
with self.subTest(value=value):
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [], "grace_seconds": value}))
self.assertEqual(400, status)
self.assertIn("grace_seconds", str(payload["error"]))