Compare commits

..

6 Commits

Author SHA1 Message Date
didericis bd663196dc docs: reposition README around provider-neutral secure substrate
test / unit (pull_request) Successful in 36s
test / integration (pull_request) Successful in 17s
Lead with the agnostic + security story instead of the single-user
security framing. New hero positions bot-bottle as a neutral control
plane that runs any agent (Claude, Codex, or a drop-in contrib plugin)
inside an isolation boundary the agent can't touch.

Restructure Features into three pillars — neutral substrate, isolation
boundary, host-matched isolation — promoting provider-agnosticism (PRD
0053 user plugins) from a buried bullet to a headline. No capability
claims changed; per-provider auth/image detail preserved as a note
linking to Manifest.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YcU7nerbg8cVj9R4EkpfLJ
2026-06-24 01:21:05 -04:00
didericis-codex 6b0de88be6 docs: activate install script prd
lint / lint (push) Successful in 1m39s
test / unit (pull_request) Successful in 31s
test / integration (pull_request) Successful in 16s
2026-06-23 21:47:12 -04:00
didericis-codex 9a941e59be feat: add install script packaging 2026-06-23 21:47:12 -04:00
didericis d7a3539755 ci(prd): rename PRD to prd-new placeholder per new convention 2026-06-23 21:46:44 -04:00
didericis cfe57a50d0 docs(prd): renumber PRD 0054 → 0057 (0054 slot taken by named-labelled-agents) 2026-06-23 21:46:44 -04:00
didericis e5d551861c docs(prd): PRD 0054 - install script 2026-06-23 21:46:44 -04:00
27 changed files with 529 additions and 987 deletions
+41 -5
View File
@@ -8,25 +8,40 @@
[![pylint](https://img.shields.io/badge/pylint-9.93%2F10-brightgreen)](https://github.com/PyCQA/pylint) [![pylint](https://img.shields.io/badge/pylint-9.93%2F10-brightgreen)](https://github.com/PyCQA/pylint)
[![pyright](https://img.shields.io/badge/pyright-0%20errors-brightgreen)](https://github.com/microsoft/pyright) [![pyright](https://img.shields.io/badge/pyright-0%20errors-brightgreen)](https://github.com/microsoft/pyright)
**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. **Run any coding agent like it might be compromised — and lose nothing when it is.**
**Solution:** Ephemeral, per agent "bottles" the agent cannot modify that scan all traffic for data exfiltration and limit capabilities and egress to only what the agent needs. bot-bottle is a provider-neutral, security-first substrate for autonomous agents. Bring Claude Code, Codex, or your own harness; each one runs in an ephemeral, per-agent "bottle" it cannot modify, where every byte of egress is scanned for exfiltration and capabilities are narrowed to exactly what the task declares.
## Features **Problem:** You want to let a coding agent run unsupervised, but a prompt-injected or misbehaving agent — or a poisoned repo, MCP server, or skill — can wreck your environment or exfiltrate your secrets. Locking yourself to one vendor's cloud doesn't fix that; it just moves the blast radius.
**Solution:** A neutral control plane that runs *whatever agent you choose* inside an isolation boundary the agent can't touch: TLS-bumped egress allowlisting, outbound/inbound DLP, gitleaks-gated pushes, and host secrets the agent never sees. Swap the agent; keep the guarantees.
## Why bot-bottle
### A neutral substrate — bring your own agent
- **Provider-agnostic by design** — Claude and Codex ship built in; any other agent (Gemini, Aider, a local-model wrapper) is a drop-in plugin at `~/.bot-bottle/contrib/<name>/` — no fork, no PR against this repo. The manifest accepts any provider template, and the isolation, egress, and git guarantees are identical across all of them.
- **One control plane, every harness** — the same bottle, egress policy, and supervise flow wrap whichever agent you run, so switching or mixing providers doesn't change your security posture.
- **Composable bottles (`extends:`)** — keep provider/runtime policy in one base bottle (e.g. `claude.md`) and overlay task bottles on top.
### An isolation boundary the agent can't touch
- **Per-bottle egress allowlist** — TLS-bumped HTTP/HTTPS chokepoint with a per-manifest host allowlist; per-route path/method/header `matches` filtering; outbound DLP scanning for known tokens and secrets, inbound DLP scanning for prompt-injection attempts; DoH and arbitrary hosts blocked by default. - **Per-bottle egress allowlist** — TLS-bumped HTTP/HTTPS chokepoint with a per-manifest host allowlist; per-route path/method/header `matches` filtering; outbound DLP scanning for known tokens and secrets, inbound DLP scanning for prompt-injection attempts; DoH and arbitrary hosts blocked by default.
- **Tokens the agent never sees** — host secrets live in a sidecar; the agent dials `http://sidecar:9099/<path>` and the proxy strips inbound `Authorization` and injects the real token before forwarding. `printenv` in the agent shows proxy URLs only. - **Tokens the agent never sees** — host secrets live in a sidecar; the agent dials `http://sidecar:9099/<path>` and the proxy strips inbound `Authorization` and injects the real token before forwarding. `printenv` in the agent shows proxy URLs only.
- **Gitleaks-scanned push (git-gate)** — `bottle.git` remotes route through a per-bottle `git daemon` that gitleaks-scans incoming refs pre-receive and forwards clean refs upstream over SSH. The agent never holds the upstream credential. - **Gitleaks-scanned push (git-gate)** — `bottle.git` remotes route through a per-bottle `git daemon` that gitleaks-scans incoming refs pre-receive and forwards clean refs upstream over SSH. The agent never holds the upstream credential.
- **Manifest-scoped skills + secrets** — each bottle declares its skills, env, git identity, remotes, and egress routes; unknown keys die at load. - **Manifest-scoped skills + secrets** — each bottle declares its skills, env, git identity, remotes, and egress routes; unknown keys die at load.
- **Trust boundary at `$HOME`** — bottles (credentials, egress, remotes) live only under `~/.bot-bottle/bottles/`. Repos may ship agents but not bottles, so a cloned repo can't redirect an env var to an attacker host. - **Trust boundary at `$HOME`** — bottles (credentials, egress, remotes) live only under `~/.bot-bottle/bottles/`. Repos may ship agents but not bottles, so a cloned repo can't redirect an env var to an attacker host.
- **Composable bottles (`extends:`)** — keep provider/runtime policy in one base bottle (e.g. `claude.md`) and overlay task bottles on top.
### Isolation that matches your host
- **Parallel, isolated bottles** — each bottle runs in its own backend-owned isolation boundary; bottles don't share state or talk to each other. - **Parallel, isolated bottles** — each bottle runs in its own backend-owned isolation boundary; bottles don't share state or talk to each other.
- **Provider templates (Claude, Codex)** — `Dockerfile.claude` / `Dockerfile.codex`, or a bottle-supplied Dockerfile. Claude auth via long-lived OAuth token; Codex via opt-in host device-auth forwarding.
- **gVisor auto-detect** — on Linux hosts where `runsc` is registered with Docker, every bottle launches under it for a userspace syscall barrier; no manifest config required. - **gVisor auto-detect** — on Linux hosts where `runsc` is registered with Docker, every bottle launches under it for a userspace syscall barrier; no manifest config required.
- **Apple Container backend (macOS default when available)** — runs the agent and sidecar bundle with Apple's `container` CLI, using a host-only agent network plus a separate sidecar egress network. - **Apple Container backend (macOS default when available)** — runs the agent and sidecar bundle with Apple's `container` CLI, using a host-only agent network plus a separate sidecar egress network.
- **Smolmachines backend** — runs the agent in a libkrun micro-VM while the sidecar bundle stays in Docker. TSI and smolmachines DNS filtering close the raw DNS exfiltration gap that exists in the legacy Docker backend. - **Smolmachines backend** — runs the agent in a libkrun micro-VM while the sidecar bundle stays in Docker. TSI and smolmachines DNS filtering close the raw DNS exfiltration gap that exists in the legacy Docker backend.
- **Legacy Docker backend** — still available for examples, CI, and hosts without Apple Container via `BOT_BOTTLE_BACKEND=docker` or `--backend=docker`. - **Legacy Docker backend** — still available for examples, CI, and hosts without Apple Container via `BOT_BOTTLE_BACKEND=docker` or `--backend=docker`.
Per-provider auth (Claude long-lived OAuth token; Codex opt-in host device-auth forwarding) and per-provider images (`Dockerfile.claude` / `Dockerfile.codex`, or a bottle-supplied Dockerfile) are configured on the bottle — see [Manifest](#manifest).
## Architecture ## Architecture
On the default macOS Apple Container backend, a bottle is an agent container on a host-only internal network plus a sidecar bundle attached to both that internal network and a NAT egress network. The agent gets HTTP(S)_PROXY and CA bundle env vars pointing at the sidecar's internal-network IP, so HTTP/HTTPS traffic flows through the sidecar instead of direct egress. `bottle.git` / git-gate is intentionally deferred on this backend until a safe Apple Container key-delivery path exists. On the default macOS Apple Container backend, a bottle is an agent container on a host-only internal network plus a sidecar bundle attached to both that internal network and a NAT egress network. The agent gets HTTP(S)_PROXY and CA bundle env vars pointing at the sidecar's internal-network IP, so HTTP/HTTPS traffic flows through the sidecar instead of direct egress. `bottle.git` / git-gate is intentionally deferred on this backend until a safe Apple Container key-delivery path exists.
@@ -68,6 +83,27 @@ The Docker topology looks like this:
When the agent exits, `cli.py` tears down every sidecar and both networks; nothing about a bottle persists between runs. When the agent exits, `cli.py` tears down every sidecar and both networks; nothing about a bottle persists between runs.
## Install
Install the CLI with the bootstrap script:
```sh
curl -fsSL https://gitea.dideric.is/didericis/bot-bottle/raw/branch/main/install.sh | sh
```
The script checks Python 3.11+, checks Docker daemon reachability, creates the `~/.bot-bottle/` config directories, installs the Python package with `pipx` when available or `pip --user` otherwise, then runs:
```sh
bot-bottle doctor
```
Python-native installers can use the package metadata directly:
```sh
pipx install git+https://gitea.dideric.is/didericis/bot-bottle.git
uv tool install git+https://gitea.dideric.is/didericis/bot-bottle.git
```
## Quickstart ## Quickstart
On compatible macOS hosts, the default backend requires Apple's `container` CLI and does not require Docker. The smolmachines backend requires Docker on the host for the sidecar bundle plus smolvm. The legacy Docker backend requires Docker. Claude bottles also need a long-lived Claude Code OAuth token (`claude setup-token`) exported as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`. On compatible macOS hosts, the default backend requires Apple's `container` CLI and does not require Docker. The smolmachines backend requires Docker on the host for the sidecar bundle plus smolvm. The legacy Docker backend requires Docker. Claude bottles also need a long-lived Claude Code OAuth token (`claude setup-token`) exported as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`.
+96
View File
@@ -0,0 +1,96 @@
# Per-bottle sidecar bundle image (PRD 0024).
#
# Collapses the prior per-sidecar images (egress, git-gate,
# supervise) into one. A small stdlib-Python init supervisor at
# /app/sidecar_init.py spawns all daemons, forwards SIGTERM, and
# propagates per-daemon stdout/stderr to the container log with a
# `[name]` prefix. See PRD 0024 for the rationale.
#
# Layout:
#
# /usr/bin/gitleaks gitleaks binary
# /app/egress_addon.py + siblings mitmproxy addon (egress)
# /app/egress-entrypoint.sh mitmdump launcher
# /app/supervise_server.py + .py supervise MCP server
# /app/sidecar_init.py PID 1 supervisor
# /etc/egress/routes.yaml bind-mounted at run time
# /etc/git-gate/pre-receive docker-cp'd at start time
# /git-gate-entrypoint.sh docker-cp'd at start time
# /git-gate/creds/* docker-cp'd at start time
# /git/* bare repos, populated at runtime
# /run/supervise/queue/ bind-mounted at run time
# /home/mitmproxy/.mitmproxy/ mitmproxy CA dir
#
# Exposed ports inside the container:
# 9099 egress (mitmproxy, agent-facing HTTPS proxy)
# 9418 git-gate (git-daemon)
# 9420 git-gate smart HTTP (smolmachines agent-facing transport)
# 9100 supervise (MCP HTTP)
# Stage 1: gitleaks binary. The upstream gitleaks image is alpine
# with the binary at /usr/bin/gitleaks. Pinned by digest in lockstep
# with Dockerfile.git-gate's prior base (now deleted at chunk 3).
FROM zricethezav/gitleaks@sha256:c00b6bd0aeb3071cbcb79009cb16a60dd9e0a7c60e2be9ab65d25e6bc8abbb7f AS gitleaks-src
# Stage 2: assembly. mitmproxy/mitmproxy is debian-slim-based with
# Python + mitmdump pre-installed — heavier than the others, so
# this stage starts there and pulls the standalone binaries in.
FROM mitmproxy/mitmproxy:11.1.3
# Run as root inside the bundle. The bundle is the isolation
# boundary; per-daemon user separation inside it is not load-bearing
# and complicates the supervisor's spawn path.
USER root
# Runtime system deps:
# git supplies the `git daemon` subcommand (no separate package)
# plus the core `git` binary the pre-receive hook invokes.
# openssh-client supplies the upstream SSH transport the
# pre-receive hook uses to forward accepted refs.
# ca-certificates is needed for mitmdump upstream TLS (the
# base image already has it; listed for explicitness).
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
git openssh-client ca-certificates \
&& rm -rf /var/lib/apt/lists/*
# Pull the standalone binaries into the final image.
COPY --from=gitleaks-src /usr/bin/gitleaks /usr/bin/gitleaks
# Project Python: addon + server modules + the init supervisor.
# Kept flat under /app/ so mitmdump's loader resolves them as
# top-level siblings (absolute imports), matching the prior
# Dockerfile.egress / Dockerfile.supervise layout.
COPY bot_bottle/egress_addon_core.py /app/egress_addon_core.py
COPY bot_bottle/egress_addon.py /app/egress_addon.py
COPY bot_bottle/dlp_detectors.py /app/dlp_detectors.py
COPY bot_bottle/yaml_subset.py /app/yaml_subset.py
COPY bot_bottle/supervise.py /app/supervise.py
COPY bot_bottle/supervise_server.py /app/supervise_server.py
COPY bot_bottle/sidecar_init.py /app/sidecar_init.py
COPY bot_bottle/git_http_backend.py /app/git_http_backend.py
COPY bot_bottle/egress_entrypoint.sh /app/egress-entrypoint.sh
RUN chmod +x /app/egress-entrypoint.sh
# Pre-create runtime directories the compose renderer + start
# step expect to exist. `docker cp` does not create intermediate
# dirs, and bind mounts won't either if the parent is missing.
RUN mkdir -p \
/etc/egress \
/etc/git-gate \
/git-gate/creds \
/git \
/run/supervise/queue \
/home/mitmproxy/.mitmproxy
# Documentation only — the compose renderer publishes whichever
# subset the bottle uses.
EXPOSE 8888 9099 9418 9420 9100
# WORKDIR matches Dockerfile.supervise's prior layout so the
# in-app same-dir import in supervise_server.py stays deterministic.
WORKDIR /app
# PID 1 is the supervisor. It owns signal handling and exit-code
# propagation; no `exec` chain in the entrypoint itself.
ENTRYPOINT ["python3", "/app/sidecar_init.py"]
+9 -10
View File
@@ -58,10 +58,17 @@ from .sidecar_bundle import (
) )
# Repo root, used as the build context for the bundle Dockerfile. # Repo root or installed site-packages root, used as the build context for
# Dockerfiles that COPY bot_bottle source files.
_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent) _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
def _sidecar_bundle_dockerfile() -> str:
if (Path(_REPO_DIR) / SIDECAR_BUNDLE_DOCKERFILE).is_file():
return SIDECAR_BUNDLE_DOCKERFILE
return f"bot_bottle/{SIDECAR_BUNDLE_DOCKERFILE}"
def bottle_plan_to_compose(plan: DockerBottlePlan) -> dict[str, Any]: def bottle_plan_to_compose(plan: DockerBottlePlan) -> dict[str, Any]:
"""Render a Compose v2 spec dict from a fully-resolved """Render a Compose v2 spec dict from a fully-resolved
DockerBottlePlan. DockerBottlePlan.
@@ -137,10 +144,6 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
volumes.append(_bind(ep.routes_path.parent, str(Path(EGRESS_ROUTES_IN_CONTAINER).parent))) volumes.append(_bind(ep.routes_path.parent, str(Path(EGRESS_ROUTES_IN_CONTAINER).parent)))
for token_env in sorted(ep.token_env_map.keys()): for token_env in sorted(ep.token_env_map.keys()):
env.append(token_env) env.append(token_env)
if ep.canary:
# Inject canary as a literal NAME=VALUE (not a bare name) — the
# value is a fake secret so it need not be hidden from the compose file.
env.append(f"EGRESS_TOKEN_CANARY={ep.canary}")
# --- git-gate ----------------------------------------------------- # --- git-gate -----------------------------------------------------
gp = plan.git_gate_plan gp = plan.git_gate_plan
@@ -187,7 +190,7 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
"image": SIDECAR_BUNDLE_IMAGE, "image": SIDECAR_BUNDLE_IMAGE,
"build": { "build": {
"context": _REPO_DIR, "context": _REPO_DIR,
"dockerfile": SIDECAR_BUNDLE_DOCKERFILE, "dockerfile": _sidecar_bundle_dockerfile(),
}, },
"container_name": sidecar_bundle_container_name(plan.slug), "container_name": sidecar_bundle_container_name(plan.slug),
"networks": { "networks": {
@@ -224,10 +227,6 @@ def _agent_service(plan: DockerBottlePlan) -> dict[str, Any]:
# never lands on argv or in the compose file. # never lands on argv or in the compose file.
for name in sorted(plan.forwarded_env.keys()): for name in sorted(plan.forwarded_env.keys()):
env.append(name) env.append(name)
# Canary token: visible to the agent as a fake secret so that any
# outbound appearance of this value is a zero-FP exfil signal.
if plan.egress_plan.canary:
env.append(f"BOT_BOTTLE_CANARY={plan.egress_plan.canary}")
service: dict[str, Any] = { service: dict[str, Any] = {
"image": plan.image, "image": plan.image,
+4 -3
View File
@@ -12,9 +12,10 @@ from __future__ import annotations
import os import os
# Bundle image. Defaults to a built-locally tag (built from the # Bundle image. Defaults to a built-locally tag. Source checkouts
# repo's Dockerfile.sidecars via compose `build:`). Operators # build from the repo-root Dockerfile.sidecars; installed packages
# pinning to a published digest can override via env. # build from the packaged copy under bot_bottle/.
# Operators pinning to a published digest can override via env.
SIDECAR_BUNDLE_IMAGE = os.environ.get( SIDECAR_BUNDLE_IMAGE = os.environ.get(
"BOT_BOTTLE_SIDECAR_IMAGE", "BOT_BOTTLE_SIDECAR_IMAGE",
"bot-bottle-sidecars:latest", "bot-bottle-sidecars:latest",
@@ -353,8 +353,6 @@ def _sidecar_env_entries(plan: MacosContainerBottlePlan) -> tuple[str, ...]:
env: list[str] = [] env: list[str] = []
if plan.egress_plan.routes: if plan.egress_plan.routes:
env.extend(sorted(plan.egress_plan.token_env_map.keys())) env.extend(sorted(plan.egress_plan.token_env_map.keys()))
if plan.egress_plan.canary:
env.append(f"EGRESS_TOKEN_CANARY={plan.egress_plan.canary}")
if plan.git_gate_plan.upstreams: if plan.git_gate_plan.upstreams:
env.append(f"BOT_BOTTLE_GIT_GATE_READY_FILE={_GIT_GATE_READY_FILE}") env.append(f"BOT_BOTTLE_GIT_GATE_READY_FILE={_GIT_GATE_READY_FILE}")
if plan.supervise_plan is not None: if plan.supervise_plan is not None:
@@ -422,8 +420,6 @@ def _agent_env_entries(
env.append(f"{name}={value}") env.append(f"{name}={value}")
for name in sorted(plan.forwarded_env.keys()): for name in sorted(plan.forwarded_env.keys()):
env.append(name) env.append(name)
if plan.egress_plan.canary:
env.append(f"BOT_BOTTLE_CANARY={plan.egress_plan.canary}")
return tuple(env) return tuple(env)
+4 -1
View File
@@ -1,6 +1,6 @@
"""Main CLI dispatcher. """Main CLI dispatcher.
Commands: cleanup, commit, edit, info, init, list, resume, start, supervise Commands: cleanup, commit, doctor, edit, info, init, list, resume, start, supervise
""" """
from __future__ import annotations from __future__ import annotations
@@ -13,6 +13,7 @@ from ._common import PROG
from . import list as _list_mod from . import list as _list_mod
from .cleanup import cmd_cleanup from .cleanup import cmd_cleanup
from .commit import cmd_commit from .commit import cmd_commit
from .doctor import cmd_doctor
from .edit import cmd_edit from .edit import cmd_edit
from .info import cmd_info from .info import cmd_info
from .init import cmd_init from .init import cmd_init
@@ -25,6 +26,7 @@ cmd_list = _list_mod.cmd_list
COMMANDS = { COMMANDS = {
"cleanup": cmd_cleanup, "cleanup": cmd_cleanup,
"commit": cmd_commit, "commit": cmd_commit,
"doctor": cmd_doctor,
"edit": cmd_edit, "edit": cmd_edit,
"info": cmd_info, "info": cmd_info,
"init": cmd_init, "init": cmd_init,
@@ -40,6 +42,7 @@ def usage() -> None:
sys.stderr.write("Commands:\n") sys.stderr.write("Commands:\n")
sys.stderr.write(" cleanup stop and remove all active bot-bottle containers\n") sys.stderr.write(" cleanup stop and remove all active bot-bottle containers\n")
sys.stderr.write(" commit snapshot a running bottle's container state to a Docker image\n") sys.stderr.write(" commit snapshot a running bottle's container state to a Docker image\n")
sys.stderr.write(" doctor check Python, Docker, and bot-bottle config prerequisites\n")
sys.stderr.write(" edit open an agent in vim for editing\n") sys.stderr.write(" edit open an agent in vim for editing\n")
sys.stderr.write(" info print env, skills, and prompt details for a named agent\n") sys.stderr.write(" info print env, skills, and prompt details for a named agent\n")
sys.stderr.write(" init interactively create a new agent and add it to bot-bottle.json\n") sys.stderr.write(" init interactively create a new agent and add it to bot-bottle.json\n")
+1 -1
View File
@@ -6,7 +6,7 @@ import os
import sys import sys
from pathlib import Path from pathlib import Path
PROG = "cli.py" PROG = Path(sys.argv[0]).name or "bot-bottle"
USER_CWD = os.getcwd() USER_CWD = os.getcwd()
REPO_DIR = str(Path(__file__).resolve().parent.parent.parent) REPO_DIR = str(Path(__file__).resolve().parent.parent.parent)
+73
View File
@@ -0,0 +1,73 @@
"""doctor: validate host prerequisites for running bot-bottle."""
from __future__ import annotations
import argparse
import shutil
import subprocess
import sys
from pathlib import Path
from ._common import PROG
def _ok(label: str, detail: str) -> None:
print(f"ok: {label}: {detail}")
def _fail(label: str, detail: str) -> None:
print(f"fail: {label}: {detail}")
def _check_python() -> bool:
version = sys.version_info
detail = f"{version.major}.{version.minor}.{version.micro}"
if version >= (3, 11):
_ok("python", detail)
return True
_fail("python", f"{detail}; need 3.11 or newer")
return False
def _check_docker() -> bool:
docker = shutil.which("docker")
if not docker:
_fail("docker", "docker command not found")
return False
try:
result = subprocess.run(
[docker, "info"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
timeout=10,
)
except (OSError, subprocess.TimeoutExpired) as exc:
_fail("docker", f"daemon check failed: {exc}")
return False
if result.returncode == 0:
_ok("docker", "daemon reachable")
return True
_fail("docker", "daemon not reachable")
return False
def _check_config_dir() -> bool:
config = Path.home() / ".bot-bottle"
if config.is_dir():
_ok("config", str(config))
return True
_fail("config", f"{config} does not exist")
return False
def cmd_doctor(argv: list[str]) -> int:
parser = argparse.ArgumentParser(prog=f"{PROG} doctor", add_help=True)
parser.parse_args(argv)
checks = (
_check_python(),
_check_docker(),
_check_config_dir(),
)
return 0 if all(checks) else 1
+1 -4
View File
@@ -292,10 +292,7 @@ def cmd_supervise(argv: list[str]) -> int:
return e.code if isinstance(e.code, int) else 1 return e.code if isinstance(e.code, int) else 1
except Exception as e: # noqa: W0718 — catch supervise crash for logging except Exception as e: # noqa: W0718 — catch supervise crash for logging
log_path = _write_crash_log(e) log_path = _write_crash_log(e)
error( error(f"supervise crashed: {type(e).__name__}: {e}")
f"supervise crashed: {type(e).__name__}: {e}",
context={"error_type": type(e).__name__, "crash_log": str(log_path)},
)
error(f"full traceback written to {log_path}") error(f"full traceback written to {log_path}")
return 1 return 1
return 0 return 0
+4 -154
View File
@@ -1,4 +1,4 @@
"""DLP detectors for the egress proxy (PRD 0053, prd-new). """DLP detectors for the egress proxy (PRD 0053).
Pure Python, no mitmproxy dependency. Each detector is a module-level Pure Python, no mitmproxy dependency. Each detector is a module-level
function returning `ScanResult | None`. function returning `ScanResult | None`.
@@ -15,8 +15,6 @@ import gzip
import re import re
import typing import typing
import unicodedata import unicodedata
from math import log2
from collections import Counter
from urllib.parse import quote as url_quote from urllib.parse import quote as url_quote
try: try:
@@ -98,21 +96,20 @@ def redact_tokens(
text: str, text: str,
*, *,
env: typing.Mapping[str, str] | None = None, env: typing.Mapping[str, str] | None = None,
sensitive_prefixes: tuple[str, ...] = ("EGRESS_TOKEN_",),
) -> str: ) -> str:
"""Replace token pattern matches and (if env given) provisioned secrets with REDACT.""" """Replace token pattern matches and (if env given) provisioned secrets with REDACT."""
for _, pattern in TOKEN_PATTERNS: for _, pattern in TOKEN_PATTERNS:
text = pattern.sub(REDACT, text) text = pattern.sub(REDACT, text)
if env is not None: if env is not None:
for key, value in env.items(): for key, value in env.items():
if any(key.startswith(p) for p in sensitive_prefixes) and value: if key.startswith("EGRESS_TOKEN_") and value:
for variant in _encoded_variants(value): for variant in _encoded_variants(value):
text = text.replace(variant, REDACT) text = text.replace(variant, REDACT)
return text return text
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Known secrets detector (Phase 1b, prd-new) # Known secrets detector (Phase 1b)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _encoded_variants(secret: str) -> list[str]: def _encoded_variants(secret: str) -> list[str]:
@@ -153,63 +150,17 @@ def _encoded_variants(secret: str) -> list[str]:
return variants return variants
# ---------------------------------------------------------------------------
# Fragmentation-resistant helpers (prd-new)
# ---------------------------------------------------------------------------
# Minimum length of alnum projection for projection-based checks to run.
# Short secrets produce too many false positives in projection space.
_ALNUM_MIN_LEN = 8
# Minimum window length for the partial-substring sliding scan.
PARTIAL_MATCH_MIN_LEN = 12
def _alnum_projection(text: str) -> str:
"""Return text with every non-alphanumeric character stripped.
Used for fragmentation-resistant matching: separator-injected secrets
(spaces, hyphens, dots inserted between characters) are identical to
their originals in alnum projection space.
"""
return "".join(c for c in text if c.isalnum())
def _find_partial_window(secret_alnum: str, text_alnum: str, min_len: int) -> int | None:
"""Return the position in text_alnum where any min_len-char window of
secret_alnum first appears, or None.
Slides a window of width min_len across secret_alnum and searches for
each window in text_alnum. The first hit position is returned.
"""
if len(secret_alnum) < min_len or len(text_alnum) < min_len:
return None
for i in range(len(secret_alnum) - min_len + 1):
window = secret_alnum[i:i + min_len]
pos = text_alnum.find(window)
if pos >= 0:
return pos
return None
def scan_known_secrets( def scan_known_secrets(
text: str, text: str,
*, *,
location: str = "body", location: str = "body",
env: typing.Mapping[str, str] | None = None, env: typing.Mapping[str, str] | None = None,
sensitive_prefixes: tuple[str, ...] = ("EGRESS_TOKEN_",),
) -> ScanResult | None: ) -> ScanResult | None:
if env is None: if env is None:
return None return None
# Pre-compute alnum projection of the scan text once; reused per secret.
text_alnum: str | None = None
for key, value in env.items(): for key, value in env.items():
if not any(key.startswith(p) for p in sensitive_prefixes) or not value: if not key.startswith("EGRESS_TOKEN_") or not value:
continue continue
# Pass 1: exact match across encoded variants (original behaviour).
for variant in _encoded_variants(value): for variant in _encoded_variants(value):
pos = text.find(variant) pos = text.find(variant)
if pos >= 0: if pos >= 0:
@@ -219,100 +170,6 @@ def scan_known_secrets(
location=location, location=location,
context=_snippet(text, pos, pos + len(variant)), context=_snippet(text, pos, pos + len(variant)),
) )
# Pass 2 & 3: fragmentation-resistant projection checks.
secret_alnum = _alnum_projection(value)
if len(secret_alnum) < _ALNUM_MIN_LEN:
continue
if text_alnum is None:
text_alnum = _alnum_projection(text)
# Pass 2: full alnum-projection exact match (catches separator injection).
pos2 = text_alnum.find(secret_alnum)
if pos2 >= 0:
return ScanResult(
severity="block",
reason=(
f"provisioned secret from {key} found in {location} "
f"(fragmented match — separator injection)"
),
location=location,
context=_snippet(text_alnum, pos2, pos2 + len(secret_alnum)),
)
# Pass 3: sliding-window partial match (catches chunked-substring leaks).
pos3 = _find_partial_window(secret_alnum, text_alnum, PARTIAL_MATCH_MIN_LEN)
if pos3 is not None:
return ScanResult(
severity="block",
reason=(
f"provisioned secret from {key} found in {location} "
f"(partial match — at least {PARTIAL_MATCH_MIN_LEN} consecutive "
f"alphanumeric chars)"
),
location=location,
context=_snippet(text_alnum, pos3, pos3 + PARTIAL_MATCH_MIN_LEN),
)
return None
# ---------------------------------------------------------------------------
# Entropy detector (warn-only, prd-new)
# ---------------------------------------------------------------------------
# Sliding window size and step for the entropy scan.
ENTROPY_WINDOW = 64
ENTROPY_STEP = 32
# Bits-per-character threshold. Random ASCII printable ≈ 6.6 bits; random
# lowercase hex ≈ 4 bits; random base64url ≈ 6 bits. 5.5 sits above
# typical structured data (JSON, URLs) while staying below truly random
# content.
ENTROPY_BLOCK_THRESHOLD = 5.5
def _shannon_entropy(text: str) -> float:
if not text:
return 0.0
counts = Counter(text)
n = len(text)
return -sum((c / n) * log2(c / n) for c in counts.values())
def scan_entropy(
text: str,
*,
location: str = "body",
window: int = ENTROPY_WINDOW,
threshold: float = ENTROPY_BLOCK_THRESHOLD,
) -> ScanResult | None:
"""Warn-only detector: flag windows of `window` chars with Shannon entropy
above `threshold` bits per character.
Never blocks; always returns severity='warn'. Disabled by default
routes must opt in via dlp.outbound_detectors=['entropy'].
"""
if not text:
return None
step = max(1, window // 2)
end = len(text)
# Scan overlapping windows; also check the final tail if shorter than window.
positions = list(range(0, end - window + 1, step))
if end < window:
positions = [0]
elif (end - window) % step != 0:
positions.append(end - window)
for i in positions:
chunk = text[i:i + window]
if _shannon_entropy(chunk) >= threshold:
return ScanResult(
severity="warn",
reason=f"high-entropy content in {location} (possible encrypted exfil)",
location=location,
context=_snippet(text, i, i + len(chunk)),
)
return None return None
@@ -423,18 +280,11 @@ def scan_crlf_injection(text: str) -> ScanResult | None:
__all__ = [ __all__ = [
"ENTROPY_BLOCK_THRESHOLD",
"ENTROPY_WINDOW",
"ENTROPY_STEP",
"PARTIAL_MATCH_MIN_LEN",
"REDACT", "REDACT",
"SNIPPET_CONTEXT", "SNIPPET_CONTEXT",
"TOKEN_PATTERNS", "TOKEN_PATTERNS",
"_alnum_projection",
"_shannon_entropy",
"redact_tokens", "redact_tokens",
"scan_crlf_injection", "scan_crlf_injection",
"scan_entropy",
"scan_known_secrets", "scan_known_secrets",
"scan_naive_injection", "scan_naive_injection",
"scan_token_patterns", "scan_token_patterns",
-7
View File
@@ -10,7 +10,6 @@ specific and lives on concrete subclasses (see
from __future__ import annotations from __future__ import annotations
import dataclasses import dataclasses
import secrets
from abc import ABC from abc import ABC
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
@@ -65,7 +64,6 @@ class EgressPlan:
mitmproxy_ca_host_path: Path = Path() mitmproxy_ca_host_path: Path = Path()
mitmproxy_ca_cert_only_host_path: Path = Path() mitmproxy_ca_cert_only_host_path: Path = Path()
log: int = 0 log: int = 0
canary: str = ""
def egress_manifest_routes( def egress_manifest_routes(
@@ -301,17 +299,12 @@ class Egress(ABC):
routes_path = stage_dir / EGRESS_ROUTES_FILENAME routes_path = stage_dir / EGRESS_ROUTES_FILENAME
routes_path.write_text(egress_render_routes(routes, log=log)) routes_path.write_text(egress_render_routes(routes, log=log))
routes_path.chmod(0o600) routes_path.chmod(0o600)
# Generate a per-session canary token. The sidecar receives it as
# EGRESS_TOKEN_CANARY (scanned by the existing known-secrets detector);
# the agent receives it as BOT_BOTTLE_CANARY (a visible fake secret).
canary = secrets.token_urlsafe(32)
return EgressPlan( return EgressPlan(
slug=slug, slug=slug,
routes_path=routes_path, routes_path=routes_path,
routes=routes, routes=routes,
token_env_map=egress_token_env_map(routes), token_env_map=egress_token_env_map(routes),
log=log, log=log,
canary=canary,
) )
__all__ = [ __all__ = [
+3 -33
View File
@@ -34,7 +34,7 @@ VALID_METHODS = frozenset({
"CONNECT", "CONNECT",
}) })
OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets", "entropy"}) OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets"})
INBOUND_DETECTOR_NAMES = frozenset({"naive_injection_detection"}) INBOUND_DETECTOR_NAMES = frozenset({"naive_injection_detection"})
@@ -696,28 +696,17 @@ def scan_outbound(
try: try:
from dlp_detectors import ( # type: ignore[import-not-found] from dlp_detectors import ( # type: ignore[import-not-found]
scan_crlf_injection, scan_crlf_injection,
scan_entropy,
scan_known_secrets, scan_known_secrets,
scan_token_patterns, scan_token_patterns,
) )
except ImportError: # pragma: no cover - host-side path except ImportError: # pragma: no cover - host-side path
from .dlp_detectors import ( # type: ignore[import-not-found] from .dlp_detectors import ( # type: ignore[import-not-found]
scan_crlf_injection, scan_crlf_injection,
scan_entropy,
scan_known_secrets, scan_known_secrets,
scan_token_patterns, scan_token_patterns,
) )
# Binary bodies: latin-1 is a bijective byte↔codepoint mapping that text = body if isinstance(body, str) else body.decode("utf-8", errors="replace")
# preserves every byte value, so ASCII-range secret strings remain
# findable by str.find / regex. Prefer strict UTF-8 for valid text bodies.
if isinstance(body, bytes):
try:
text = body.decode("utf-8")
except UnicodeDecodeError:
text = body.decode("latin-1")
else:
text = body
# CRLF injection is never legitimate — runs unconditionally, not gated # CRLF injection is never legitimate — runs unconditionally, not gated
# by outbound_detectors config. # by outbound_detectors config.
@@ -731,26 +720,7 @@ def scan_outbound(
return result return result
if _detector_enabled(route.outbound_detectors, "known_secrets"): if _detector_enabled(route.outbound_detectors, "known_secrets"):
# BOT_BOTTLE_SENSITIVE_PREFIXES lets operators add extra env prefixes result = scan_known_secrets(text, location="body", env=environ)
# beyond EGRESS_TOKEN_* without changing the manifest schema.
extra_raw = environ.get("BOT_BOTTLE_SENSITIVE_PREFIXES", "")
extra = tuple(p for p in extra_raw.split(",") if p)
sensitive_prefixes = ("EGRESS_TOKEN_",) + extra
result = scan_known_secrets(
text, location="body", env=environ, sensitive_prefixes=sensitive_prefixes,
)
if result is not None:
return result
# Entropy scanning requires explicit opt-in: it is NOT part of the
# default "all detectors" set because it produces false positives on
# legitimate base64 / binary payloads. Routes must list "entropy" in
# dlp.outbound_detectors to enable it.
if (
route.outbound_detectors is not None
and "entropy" in route.outbound_detectors
):
result = scan_entropy(text, location="body")
if result is not None: if result is not None:
return result return result
+10 -96
View File
@@ -1,107 +1,21 @@
"""Tiny logging wrappers. All output goes to stderr. """Tiny logging wrappers. All output goes to stderr."""
Two capabilities layer onto the bare wrappers (issue #252):
- **Levels.** `debug` / `info` / `warn` / `error` carry an ordered
severity. Output is gated by `BOT_BOTTLE_LOG_LEVEL` (debug | info |
warn | error; default `info`). A message emits when its severity is
at or above the threshold, so `debug` is silent by default and
`error` always surfaces (nothing sits above it) which keeps the
fatal `die` path visible regardless of the configured level.
- **Context.** Every wrapper takes an optional `context` mapping that
renders as a parseable ` [k=v ...]` suffix (keys sorted; values with
whitespace/quotes are quoted), so failures can be filtered and
correlated instead of being flat strings.
With no `context` and the default level, output is byte-identical to the
original `bot-bottle: <msg>` / `bot-bottle: warning: <msg>` /
`bot-bottle: error: <msg>` lines the 100+ existing call sites are
unaffected.
"""
from __future__ import annotations from __future__ import annotations
import os
import sys import sys
from typing import Mapping, NoReturn from typing import NoReturn
# Ordered severities. Gaps left between values so intermediate levels
# can be added later without renumbering.
DEBUG = 10
INFO = 20
WARN = 30
ERROR = 40
_LEVEL_NAMES: dict[str, int] = {
"debug": DEBUG,
"info": INFO,
"warn": WARN,
"warning": WARN,
"error": ERROR,
}
# Default threshold when BOT_BOTTLE_LOG_LEVEL is unset or unrecognised.
_DEFAULT_THRESHOLD = INFO
_LOG_LEVEL_ENV = "BOT_BOTTLE_LOG_LEVEL"
def _threshold() -> int: def info(msg: str) -> None:
"""Resolve the active level threshold from the environment. print(f"bot-bottle: {msg}", file=sys.stderr)
Read per-call (not cached) so the level can be changed at runtime
and so tests can patch `os.environ` without a reload. Unknown values
fall back to the default rather than raising logging must never be
the thing that crashes the process."""
raw = os.environ.get(_LOG_LEVEL_ENV, "")
return _LEVEL_NAMES.get(raw.strip().lower(), _DEFAULT_THRESHOLD)
def _format_context(context: Mapping[str, object] | None) -> str: def warn(msg: str) -> None:
"""Render a context mapping as a ` [k=v k2=v2]` suffix. print(f"bot-bottle: warning: {msg}", file=sys.stderr)
Keys are sorted for stable, diffable output. Values that are empty or
contain whitespace or a quote are wrapped in double quotes (with inner
quotes escaped) so each `k=v` pair stays parseable. Empty/None context
renders as the empty string."""
if not context:
return ""
parts: list[str] = []
for key in sorted(context):
value = str(context[key])
if value == "" or any(ch.isspace() for ch in value) or '"' in value:
value = '"' + value.replace('"', '\\"') + '"'
parts.append(f"{key}={value}")
return " [" + " ".join(parts) + "]"
def _emit( def error(msg: str) -> None:
level: int, print(f"bot-bottle: error: {msg}", file=sys.stderr)
label: str,
msg: str,
context: Mapping[str, object] | None,
) -> None:
if level < _threshold():
return
prefix = f"{label}: " if label else ""
sys.stderr.write(f"bot-bottle: {prefix}{msg}{_format_context(context)}\n")
def debug(msg: str, *, context: Mapping[str, object] | None = None) -> None:
_emit(DEBUG, "debug", msg, context)
def info(msg: str, *, context: Mapping[str, object] | None = None) -> None:
_emit(INFO, "", msg, context)
def warn(msg: str, *, context: Mapping[str, object] | None = None) -> None:
_emit(WARN, "warning", msg, context)
def error(msg: str, *, context: Mapping[str, object] | None = None) -> None:
_emit(ERROR, "error", msg, context)
class Die(SystemExit): class Die(SystemExit):
@@ -117,6 +31,6 @@ class Die(SystemExit):
self.message = message self.message = message
def die(msg: str, *, context: Mapping[str, object] | None = None) -> NoReturn: def die(msg: str) -> NoReturn:
error(msg, context=context) error(msg)
raise Die(1, msg) raise Die(1, msg)
+75
View File
@@ -0,0 +1,75 @@
# PRD prd-new: Install script
- **Status:** Active
- **Author:** didericis
- **Created:** 2026-06-06
- **Issue:** #197
## Summary
Add a proper Python package distribution and a thin `install.sh` bootstrapper so users can install bot-bottle with a single command without cloning the repo.
## Problem
There is currently no install path for new users. The only way to run bot-bottle is to clone the repo and invoke `cli.py` directly. This blocks any HN-style public demo: readers want `curl | sh` or `pipx install`, not a manual clone-and-configure flow.
## Goals / Success Criteria
- `curl -fsSL <url>/install.sh | sh` (or equivalent) leaves a working `bot-bottle` command on PATH.
- Python-native users can install with `pipx install bot-bottle` or `uv tool install bot-bottle`.
- `install.sh` validates prerequisites (Python ≥ 3.11, Docker) and exits with a clear message if they are missing. It does not silently install Docker.
- `install.sh` runs `bot-bottle doctor` (or equivalent diagnostic) after install to confirm the environment is ready.
- The package has no runtime pip dependencies (stdlib-only, matching the existing constraint).
## Non-goals
- Bundling a Python runtime or producing a standalone binary.
- Automatic Docker installation.
- Plugin architecture changes (out of scope; see issue #197 for future direction).
- Publishing to PyPI in this PR — the package structure is the deliverable; publishing is a separate step.
## Design
### Package structure
Add a minimal `pyproject.toml` at the repo root:
```toml
[project]
name = "bot-bottle"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = []
[project.scripts]
bot-bottle = "bot_bottle.cli:main"
```
The existing `bot_bottle/` package and `cli.py` entry point already contain the logic; this just wires up the standard entry point. `cli.py` may need a small refactor to expose a `main()` callable if it uses `if __name__ == "__main__"` only.
### `install.sh`
A thin bootstrapper that:
1. Checks `python3 --version` ≥ 3.11; exits with instructions if not met.
2. Checks `docker info` exits 0; exits with instructions if Docker is not running.
3. Installs via `pipx` if available, otherwise falls back to `pip install --user`.
4. Runs `bot-bottle doctor` to verify the install.
The script must be idempotent (safe to re-run) and must not require `sudo`.
### `bot-bottle doctor`
A new subcommand that checks and reports:
- Python version.
- Docker daemon reachability.
- Whether `~/.bot-bottle/` config directory exists.
Exits 0 if all checks pass, non-zero otherwise.
## Decisions
- `install.sh` is hosted from the repo's raw Gitea URL for now:
`https://gitea.dideric.is/didericis/bot-bottle/raw/branch/main/install.sh`.
- Should `version` in `pyproject.toml` be driven by a git tag at build time (e.g. via `hatch-vcs`) or kept as a static string? Static is simpler for now.
@@ -1,186 +0,0 @@
# PRD prd-new: Strengthen outbound exfiltration detection
- **Status:** Active
- **Author:** claude
- **Created:** 2026-06-25
- **Issue:** #259
## Summary
The current DLP system has two detection tiers: weak (vendor token regexes)
and strong (exact matching of provisioned `EGRESS_TOKEN_*` values in multiple
encodings). This PRD adds four complementary hardening layers in ranked order
of impact:
1. **Canary tokens** — inject a per-session fake secret into the agent's
environment; any outbound appearance is a zero-false-positive exfil signal.
2. **Broadened known-value scanning** — extend the strong tier beyond
`EGRESS_TOKEN_*` via a `BOT_BOTTLE_SENSITIVE_PREFIXES` sidecar env var.
3. **Fragmentation-resistant matching** — catch separator injection and
long-substring leaks via alnum projection and sliding-window substring
search.
4. **Entropy scoring** — warn-only flag on high-entropy blobs as a
complementary signal for encrypted exfil.
## Problem
After PRD 0056 the outbound scan covers all HTTP request surfaces (host, path,
query, headers, body). But the content-based strong tier only matches
`EGRESS_TOKEN_*` values and their encodings. Three gaps remain:
| Gap | How an agent exploits it |
|-----|--------------------------|
| No canary | There is no "can't be false-positive" signal |
| Narrow prefix filter | Sensitive values provisioned under other env prefixes (MCP keys, API keys injected via the agent provider) are invisible to `scan_known_secrets` |
| Fragmentation blindness | Inserting `-`, space, or `\n` between secret characters turns an exact-match into a miss |
## Goals / Success Criteria
1. Each launched bottle has a unique canary token in the agent's environment
(`BOT_BOTTLE_CANARY`) and the egress sidecar's environment
(`EGRESS_TOKEN_CANARY`). Any outbound appearance of the canary blocks the
request with reason `"canary token"`.
2. `scan_known_secrets` accepts a `sensitive_prefixes` parameter (default:
`("EGRESS_TOKEN_",)`). `scan_outbound` reads
`BOT_BOTTLE_SENSITIVE_PREFIXES` from `environ` and merges those prefixes
in, so operators can mark additional env vars as scanned values without
changing the manifest schema.
3. For every secret that passes exact-match, a secondary alnum-projection pass
checks for the secret with all non-alphanumeric characters stripped. This
catches separator-injection evasion (`MY-SECRET` → body contains
`MY SECRET`).
4. A sliding-window partial-match pass checks for long-enough contiguous
substrings of the secret's alnum projection in the text's alnum projection.
Any match ≥ `PARTIAL_MATCH_MIN_LEN` (12 chars) blocks with reason
`"partial match"`.
5. A new `scan_entropy` detector flags outbound text windows with Shannon
entropy ≥ `ENTROPY_BLOCK_THRESHOLD` (5.5 bits/char) at **warn** severity
only. It is registered under the new detector name `"entropy"` in
`OUTBOUND_DETECTOR_NAMES` and disabled by default (routes must opt in).
6. Binary request bodies are decoded via `latin-1` instead of
`utf-8 errors="replace"`, preserving every byte value and allowing
ASCII-range secrets to be found within binary payloads.
7. All new behaviour is unit-tested; existing tests pass unchanged.
## Non-goals
- Rolling per-host buffer for split-across-requests detection (state in the
stateless addon is complex; deferred).
- Additional vendor regexes.
- ML / embedding-based detection.
- Entropy-based hard blocks (warn only per the issue).
## Design
### Canary token flow
```
Egress.prepare()
canary = secrets.token_urlsafe(32)
EgressPlan(canary=canary, ...)
Docker compose render:
sidecar env: EGRESS_TOKEN_CANARY=<canary> ← scanned by existing known-secrets detector
agent env: BOT_BOTTLE_CANARY=<canary> ← visible to agent as a "secret"
macos-container launch: same literals added to sidecar + agent env entries
```
`EGRESS_TOKEN_CANARY` matches the `EGRESS_TOKEN_` prefix already scanned by
`scan_known_secrets`, so no detector code changes are required for canary
detection — only the injection path.
### Broadened known-value scanning
`scan_known_secrets` gains a `sensitive_prefixes` parameter:
```python
def scan_known_secrets(
text: str,
*,
location: str = "body",
env: Mapping[str, str] | None = None,
sensitive_prefixes: tuple[str, ...] = ("EGRESS_TOKEN_",),
) -> ScanResult | None:
```
`scan_outbound` reads `BOT_BOTTLE_SENSITIVE_PREFIXES` (comma-separated list
of additional prefixes) from `environ` and appends them:
```python
extra = tuple(
p for p in environ.get("BOT_BOTTLE_SENSITIVE_PREFIXES", "").split(",") if p
)
sensitive_prefixes = ("EGRESS_TOKEN_",) + extra
```
`redact_tokens` receives the same treatment for consistent redaction.
### Fragmentation-resistant matching
A new helper `_alnum_projection(text)` strips all non-alphanumeric characters.
`scan_known_secrets` runs two passes per secret:
1. **Exact pass** — existing encoded-variant loop (unchanged).
2. **Alnum-projection pass** — if the secret's alnum projection has ≥ 8 chars,
check if it appears in the text's alnum projection. Match → block with
`"fragmented match (separator injection)"` reason.
3. **Partial-substring pass** — if the secret's alnum projection has ≥
`PARTIAL_MATCH_MIN_LEN` chars (12), slide a window of that length across the
secret's projection and look for each window in the text's alnum projection.
First match → block with `"partial match"` reason.
All three passes run only for the `"known_secrets"` detector; the token-pattern
and entropy detectors are unchanged.
### Entropy scoring
New public function:
```python
def scan_entropy(
text: str,
*,
location: str = "body",
window: int = ENTROPY_WINDOW, # 64
threshold: float = ENTROPY_BLOCK_THRESHOLD, # 5.5
) -> ScanResult | None:
```
Slides a window of `window` characters across `text` in steps of `window // 2`.
If any window's Shannon entropy exceeds `threshold`, returns a **warn**-severity
`ScanResult`. Never blocks.
`OUTBOUND_DETECTOR_NAMES` gains `"entropy"`. Routes opt in via their `dlp`
block; entropy scanning is **off by default** to avoid false-positive noise on
legitimate binary payloads.
### Binary body handling
In `scan_outbound`, the bytes → str decoding changes from:
```python
body.decode("utf-8", errors="replace")
```
to:
```python
body.decode("utf-8") if body is str else body.decode("latin-1")
```
`latin-1` is a bijective byte↔codepoint mapping; every byte value is preserved
as its corresponding Latin-1 code point, so ASCII-range secret strings remain
intact and `str.find` / regex still locate them correctly. The fallback from
strict UTF-8 is tried first so valid UTF-8 bodies are decoded faithfully.
## Implementation
Delivered in three commits on the same branch:
1. **DLP detector changes**`_alnum_projection`, fragmentation passes,
`scan_entropy`, broadened `scan_known_secrets`, updated `scan_outbound` and
`redact_tokens`; all accompanying unit tests.
2. **Canary injection**`EgressPlan.canary`, `Egress.prepare()`,
Docker compose + macos-container backend injection.
3. **PRD flip**`Status: Draft → Active`.
@@ -22,7 +22,7 @@ escapes**, and **whether credentials are short-lived and scoped**.
- Outbound: Docker containers have full internet access by default; no egress monitoring on most home networks - Outbound: Docker containers have full internet access by default; no egress monitoring on most home networks
- Lateral movement: compromised container can reach the LAN — NAS, other machines, internal services - Lateral movement: compromised container can reach the LAN — NAS, other machines, internal services
- Notable: CVE-2025-59536 (CVSS 8.7, Feb 2026) — a poisoned `.claude/settings.json` in a repo gives RCE when Claude Code opens it. `--dangerously-skip-permissions` removes the last gate. - Notable: CVE-2025-59536 (CVSS 8.7, Feb 2026) — a poisoned `.claude/settings.json` in a repo gives RCE when Claude Code opens it. `--dangerously-skip-permissions` removes the last gate.
- Supply chain: MCP servers, skills, and npm packages pulled during agent execution. A Jan 2026 large-scale empirical study of a 98,380-skill snapshot confirmed 157 malicious skills, ~71% of them credential harvesters. Exfiltration was overwhelmingly naive — plaintext HTTP to hardcoded endpoints; under 10% used any code obfuscation, and concealment was mostly at the documentation level, not the code level. ([Malicious Agent Skills in the Wild](https://arxiv.org/html/2602.06547v1), arXiv:2602.06547) - Supply chain: MCP servers, skills, and npm packages pulled during agent execution. ~20% of ClawHub skills were found malicious in early 2026.
**What local topology protects:** **What local topology protects:**
- No inbound attack surface — nothing listening on a public port - No inbound attack surface — nothing listening on a public port
Executable
+50
View File
@@ -0,0 +1,50 @@
#!/bin/sh
set -eu
PACKAGE_SPEC="${BOT_BOTTLE_INSTALL_SPEC:-git+https://gitea.dideric.is/didericis/bot-bottle.git}"
MIN_PYTHON="3.11"
say() {
printf 'bot-bottle install: %s\n' "$*" >&2
}
die() {
say "error: $*"
exit 1
}
command -v python3 >/dev/null 2>&1 || die "python3 is required (version ${MIN_PYTHON} or newer)"
python3 - <<'PY' || die "python3 3.11 or newer is required"
import sys
raise SystemExit(0 if sys.version_info >= (3, 11) else 1)
PY
command -v docker >/dev/null 2>&1 || die "Docker is required; install Docker and start the daemon, then re-run this script"
docker info >/dev/null 2>&1 || die "Docker is installed but the daemon is not reachable; start Docker and re-run this script"
mkdir -p \
"${HOME}/.bot-bottle/agents" \
"${HOME}/.bot-bottle/bottles" \
"${HOME}/.bot-bottle/contrib"
if command -v pipx >/dev/null 2>&1; then
say "installing with pipx"
pipx install --force "${PACKAGE_SPEC}"
else
say "pipx not found; installing with python3 -m pip --user"
python3 -m pip install --user --upgrade "${PACKAGE_SPEC}"
fi
if command -v bot-bottle >/dev/null 2>&1; then
BOT_BOTTLE_BIN="bot-bottle"
elif [ -x "${HOME}/.local/bin/bot-bottle" ]; then
BOT_BOTTLE_BIN="${HOME}/.local/bin/bot-bottle"
say "using ${BOT_BOTTLE_BIN}; add ${HOME}/.local/bin to PATH for future shells"
else
die "bot-bottle was installed but is not on PATH"
fi
say "running bot-bottle doctor"
"${BOT_BOTTLE_BIN}" doctor
+27
View File
@@ -0,0 +1,27 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "bot-bottle"
version = "0.1.0"
description = "Self-hosted sandbox for AI coding agents with egress controls"
readme = "README.md"
requires-python = ">=3.11"
license = { text = "Apache-2.0" }
dependencies = []
[project.scripts]
bot-bottle = "bot_bottle.cli:main"
[tool.setuptools.packages.find]
include = ["bot_bottle*"]
[tool.setuptools.package-data]
bot_bottle = [
"Dockerfile.sidecars",
"egress_entrypoint.sh",
"contrib/claude/Dockerfile",
"contrib/codex/Dockerfile",
"contrib/pi/Dockerfile",
]
+51
View File
@@ -0,0 +1,51 @@
"""Unit: `bot-bottle doctor` host prerequisite checks."""
from __future__ import annotations
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle.cli import doctor
class TestDoctor(unittest.TestCase):
def test_success_when_prerequisites_present(self):
with tempfile.TemporaryDirectory() as tmp, patch.object(
doctor.Path, "home", return_value=Path(tmp),
), patch.object(
doctor.shutil, "which", return_value="/usr/bin/docker",
), patch.object(
doctor.subprocess, "run",
return_value=MagicMock(returncode=0),
):
Path(tmp, ".bot-bottle").mkdir()
self.assertEqual(0, doctor.cmd_doctor([]))
def test_missing_config_fails(self):
with tempfile.TemporaryDirectory() as tmp, patch.object(
doctor.Path, "home", return_value=Path(tmp),
), patch.object(
doctor.shutil, "which", return_value="/usr/bin/docker",
), patch.object(
doctor.subprocess, "run",
return_value=MagicMock(returncode=0),
):
self.assertEqual(1, doctor.cmd_doctor([]))
def test_missing_docker_fails_before_daemon_check(self):
with tempfile.TemporaryDirectory() as tmp, patch.object(
doctor.Path, "home", return_value=Path(tmp),
), patch.object(
doctor.shutil, "which", return_value=None,
), patch.object(
doctor.subprocess, "run",
) as run:
Path(tmp, ".bot-bottle").mkdir()
self.assertEqual(1, doctor.cmd_doctor([]))
run.assert_not_called()
if __name__ == "__main__":
unittest.main()
+13
View File
@@ -301,6 +301,19 @@ class TestSidecarBundleShape(unittest.TestCase):
self.assertEqual("bot-bottle-sidecars:latest", sc["image"]) self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"]) self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"])
def test_bundle_uses_packaged_dockerfile_when_root_missing(self):
from bot_bottle.backend.docker import compose as compose_mod
original = compose_mod._REPO_DIR
try:
compose_mod._REPO_DIR = "/tmp/does-not-exist"
self.assertEqual(
"bot_bottle/Dockerfile.sidecars",
compose_mod._sidecar_bundle_dockerfile(),
)
finally:
compose_mod._REPO_DIR = original
def test_bundle_container_name_uses_sidecars_prefix(self): def test_bundle_container_name_uses_sidecars_prefix(self):
sc = self._render()["services"]["sidecars"] sc = self._render()["services"]["sidecars"]
self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"]) self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])
+3 -191
View File
@@ -1,24 +1,18 @@
"""Unit: DLP detectors (PRD 0053, prd-new). """Unit: DLP detectors (PRD 0053).
Tests for token pattern scanning, known secret detection, fragmentation- Tests for token pattern scanning, known secret detection, and
resistant matching, entropy scoring, and naive prompt injection detection.""" naive prompt injection detection."""
import base64 import base64
import gzip import gzip
import unittest import unittest
from bot_bottle.dlp_detectors import ( from bot_bottle.dlp_detectors import (
ENTROPY_BLOCK_THRESHOLD,
ENTROPY_WINDOW,
PARTIAL_MATCH_MIN_LEN,
REDACT, REDACT,
_alnum_projection,
_encoded_variants, _encoded_variants,
_normalize_text, _normalize_text,
_shannon_entropy,
redact_tokens, redact_tokens,
scan_crlf_injection, scan_crlf_injection,
scan_entropy,
scan_known_secrets, scan_known_secrets,
scan_naive_injection, scan_naive_injection,
scan_token_patterns, scan_token_patterns,
@@ -451,187 +445,5 @@ class TestKnownSecretsNewVariants(unittest.TestCase):
self.assertIsNotNone(result) self.assertIsNotNone(result)
class TestAlnumProjection(unittest.TestCase):
def test_alphanumeric_unchanged(self):
self.assertEqual("abc123XYZ", _alnum_projection("abc123XYZ"))
def test_strips_hyphens(self):
self.assertEqual("mysecretvalue", _alnum_projection("my-secret-value"))
def test_strips_spaces(self):
self.assertEqual("mysecretvalue", _alnum_projection("my secret value"))
def test_strips_dots_and_underscores(self):
self.assertEqual("mysecretvalue", _alnum_projection("my.secret_value"))
def test_empty_string(self):
self.assertEqual("", _alnum_projection(""))
def test_all_special_chars(self):
self.assertEqual("", _alnum_projection("!@#$%^&*()"))
class TestFragmentationResistantMatching(unittest.TestCase):
"""scan_known_secrets catches separator-injection and partial-substring evasion."""
# Secrets long enough that their alnum projections are ≥ 8 chars.
SECRET = "supersecrettoken99"
ENV = {"EGRESS_TOKEN_0": SECRET}
def test_exact_match_still_works(self):
result = scan_known_secrets(f"key={self.SECRET}", env=self.ENV)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("block", result.severity)
def test_separator_injection_blocked(self):
# Hyphens inserted between chars of the secret.
fragmented = "-".join(self.SECRET)
result = scan_known_secrets(f"data={fragmented}", env=self.ENV)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("block", result.severity)
self.assertIn("separator injection", result.reason)
def test_space_separator_blocked(self):
fragmented = " ".join(self.SECRET)
result = scan_known_secrets(f"body: {fragmented}", env=self.ENV)
self.assertIsNotNone(result)
assert result is not None
self.assertIn("separator injection", result.reason)
def test_partial_substring_blocked(self):
# First PARTIAL_MATCH_MIN_LEN alnum chars of the secret, no separators.
partial = _alnum_projection(self.SECRET)[:PARTIAL_MATCH_MIN_LEN]
result = scan_known_secrets(f"x={partial}&y=other", env=self.ENV)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("block", result.severity)
self.assertIn("partial match", result.reason)
def test_short_secret_skips_projection(self):
# Secrets shorter than _ALNUM_MIN_LEN in alnum projection are not
# fragmentation-checked (too many false positives).
short_env = {"EGRESS_TOKEN_0": "abc"}
# "a b c" has alnum projection "abc" (3 chars, < 8); should not block.
self.assertIsNone(scan_known_secrets("a b c", env=short_env))
def test_clean_text_not_blocked(self):
self.assertIsNone(scan_known_secrets("nothing to see here", env=self.ENV))
def test_sensitive_prefixes_param_extra_prefix(self):
env = {"MY_CRED_0": self.SECRET, "IGNORED": "other"}
result = scan_known_secrets(
f"key={self.SECRET}",
env=env,
sensitive_prefixes=("MY_CRED_",),
)
self.assertIsNotNone(result)
assert result is not None
self.assertIn("MY_CRED_0", result.reason)
def test_sensitive_prefixes_default_only_egress_token(self):
# A value under a non-EGRESS_TOKEN_ key is ignored with default prefixes.
env = {"MY_CRED_0": self.SECRET}
self.assertIsNone(scan_known_secrets(f"key={self.SECRET}", env=env))
def test_canary_prefix_detected(self):
canary_value = "canary-fake-secret-value-xyz"
env = {"EGRESS_TOKEN_CANARY": canary_value}
result = scan_known_secrets(f"x={canary_value}", env=env)
self.assertIsNotNone(result)
assert result is not None
self.assertIn("EGRESS_TOKEN_CANARY", result.reason)
class TestRedactTokensBroadenedPrefixes(unittest.TestCase):
SECRET = "my-provisioned-secret"
def test_default_redacts_egress_token(self):
env = {"EGRESS_TOKEN_0": self.SECRET}
out = redact_tokens(f"val={self.SECRET}", env=env)
self.assertNotIn(self.SECRET, out)
self.assertIn(REDACT, out)
def test_extra_prefix_redacted(self):
env = {"MY_SECRET_KEY": self.SECRET}
out = redact_tokens(
f"val={self.SECRET}",
env=env,
sensitive_prefixes=("MY_SECRET_",),
)
self.assertNotIn(self.SECRET, out)
self.assertIn(REDACT, out)
def test_non_matching_prefix_not_redacted(self):
env = {"MY_SECRET_KEY": self.SECRET}
out = redact_tokens(f"val={self.SECRET}", env=env)
# Default prefixes only include EGRESS_TOKEN_ → secret not redacted
self.assertIn(self.SECRET, out)
class TestShannonEntropy(unittest.TestCase):
def test_empty_string_zero(self):
self.assertEqual(0.0, _shannon_entropy(""))
def test_single_char_zero(self):
self.assertEqual(0.0, _shannon_entropy("aaaaaa"))
def test_two_equal_chars_one_bit(self):
self.assertAlmostEqual(1.0, _shannon_entropy("abababab"), places=10)
def test_high_entropy_random_like(self):
# Uniform 64-char string over 64 distinct symbols has entropy 6 bits.
import string
alphabet = (string.ascii_letters + string.digits + "+/")[:64]
text = alphabet # each char appears exactly once
self.assertAlmostEqual(6.0, _shannon_entropy(text), places=10)
class TestScanEntropy(unittest.TestCase):
def test_empty_returns_none(self):
self.assertIsNone(scan_entropy(""))
def test_low_entropy_returns_none(self):
# Highly repetitive text has low entropy.
self.assertIsNone(scan_entropy("a" * 200))
def test_high_entropy_warns(self):
# Build a 64-char string with entropy > ENTROPY_BLOCK_THRESHOLD.
# Use all 64 distinct printable chars to maximise entropy (~6 bits).
import string
alphabet = (string.ascii_letters + string.digits + "+/")[:64]
result = scan_entropy(alphabet, threshold=ENTROPY_BLOCK_THRESHOLD)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("warn", result.severity)
self.assertIn("high-entropy", result.reason)
def test_never_blocks(self):
import string
alphabet = (string.ascii_letters + string.digits + "+/")[:64]
result = scan_entropy(alphabet)
# scan_entropy is warn-only; it must never return severity="block".
if result is not None:
self.assertNotEqual("block", result.severity)
def test_location_in_result(self):
import string
alphabet = (string.ascii_letters + string.digits + "+/")[:64]
result = scan_entropy(alphabet, location="authorization header")
if result is not None:
self.assertIn("authorization header", result.location)
def test_structured_json_no_warn(self):
# Typical JSON has low entropy and should not be flagged.
json_body = '{"status": "ok", "message": "hello world", "count": 42}'
self.assertIsNone(scan_entropy(json_body))
def test_short_text_below_window(self):
# Text shorter than the window: checked as one chunk.
# Use a uniform string to ensure it won't be flagged.
self.assertIsNone(scan_entropy("abcde", threshold=ENTROPY_BLOCK_THRESHOLD))
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+1 -64
View File
@@ -1,14 +1,10 @@
"""Unit: Egress route lift + routes.yaml render + token """Unit: Egress route lift + routes.yaml render + token
resolution (PRD 0017, PRD 0053, prd-new).""" resolution (PRD 0017, PRD 0053)."""
import tempfile
import unittest import unittest
from pathlib import Path
from bot_bottle.egress import ( from bot_bottle.egress import (
CODEX_HOST_CREDENTIAL_TOKEN_REF, CODEX_HOST_CREDENTIAL_TOKEN_REF,
Egress,
EgressPlan,
EgressRoute, EgressRoute,
egress_manifest_routes, egress_manifest_routes,
egress_render_routes, egress_render_routes,
@@ -413,64 +409,5 @@ class TestResolveTokenValues(unittest.TestCase):
self.assertEqual({"EGRESS_TOKEN_0": "codex-access-token"}, out) self.assertEqual({"EGRESS_TOKEN_0": "codex-access-token"}, out)
class TestCanaryGeneration(unittest.TestCase):
"""Egress.prepare() generates a unique canary token per session (prd-new)."""
def _bottle_obj(self):
return ManifestIndex.from_json_obj({
"bottles": {"dev": {"egress": {"routes": []}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}).bottles["dev"]
def _make_plan(self) -> EgressPlan:
# Use a concrete no-op subclass so we can call prepare() without
# a real backend.
class _TestEgress(Egress):
pass
e = _TestEgress()
with tempfile.TemporaryDirectory() as td:
return e.prepare(self._bottle_obj(), "test-slug", Path(td))
def test_canary_is_non_empty(self):
plan = self._make_plan()
self.assertIsInstance(plan.canary, str)
self.assertGreater(len(plan.canary), 0)
def test_canary_is_unique_per_session(self):
with tempfile.TemporaryDirectory() as td:
bottle = self._bottle_obj()
class _TestEgress(Egress):
pass
e = _TestEgress()
plan_a = e.prepare(bottle, "slug-a", Path(td))
plan_b = e.prepare(bottle, "slug-b", Path(td))
self.assertNotEqual(plan_a.canary, plan_b.canary)
def test_canary_detected_by_scan_known_secrets(self):
from bot_bottle.dlp_detectors import scan_known_secrets
plan = self._make_plan()
env = {"EGRESS_TOKEN_CANARY": plan.canary}
result = scan_known_secrets(f"exfil={plan.canary}", env=env)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("block", result.severity)
self.assertIn("EGRESS_TOKEN_CANARY", result.reason)
def test_egress_plan_canary_field_default_empty(self):
# Verify EgressPlan can be constructed with an empty canary (backward compat).
from pathlib import Path
plan = EgressPlan(
slug="s",
routes_path=Path("/tmp/r.yaml"),
routes=(),
token_env_map={},
)
self.assertEqual("", plan.canary)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
-98
View File
@@ -1167,103 +1167,5 @@ class TestScanInbound(unittest.TestCase):
self.assertEqual("block", result.severity) self.assertEqual("block", result.severity)
class TestScanOutboundEnhanced(unittest.TestCase):
"""scan_outbound changes from prd-new: binary decode, entropy detector,
broadened known-value prefixes, fragmentation resistance."""
_ROUTE = Route(host="api.example.com")
_ROUTE_ENTROPY = Route(
host="api.example.com",
outbound_detectors=("entropy",),
)
def test_binary_body_latin1_decode_finds_ascii_secret(self):
# Body contains valid ASCII secret surrounded by non-UTF-8 bytes.
secret = "supersecrettoken99"
env = {"EGRESS_TOKEN_0": secret}
# Wrap the secret in bytes that are invalid UTF-8.
body = b"\x80\x81" + secret.encode("ascii") + b"\xff"
result = scan_outbound(self._ROUTE, body, env)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("block", result.severity)
def test_binary_body_valid_utf8_decoded_correctly(self):
env = {"EGRESS_TOKEN_0": "mysecret"}
# Valid UTF-8 body — should be decoded as UTF-8, not latin-1.
body = "clean body with mysecret".encode("utf-8")
result = scan_outbound(self._ROUTE, body, env)
self.assertIsNotNone(result)
def test_entropy_detector_off_by_default(self):
import string
# High-entropy content should NOT warn if the route has no entropy detector.
alphabet = (string.ascii_letters + string.digits + "+/")[:64]
result = scan_outbound(self._ROUTE, alphabet, {})
self.assertIsNone(result)
def test_entropy_detector_warns_when_enabled(self):
import string
alphabet = (string.ascii_letters + string.digits + "+/")[:64]
result = scan_outbound(self._ROUTE_ENTROPY, alphabet, {})
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("warn", result.severity)
def test_bot_bottle_sensitive_prefixes_env_var(self):
# When the sidecar env contains BOT_BOTTLE_SENSITIVE_PREFIXES,
# scan_outbound should scan those additional prefixes.
secret = "extra-sensitive-value-abc"
env = {
"MY_CRED_KEY": secret,
"BOT_BOTTLE_SENSITIVE_PREFIXES": "MY_CRED_",
}
result = scan_outbound(self._ROUTE, f"x={secret}", env)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("block", result.severity)
def test_bot_bottle_sensitive_prefixes_multiple(self):
secret = "my-api-key-value-xyz"
env = {
"ANTHROPIC_API_0": secret,
"BOT_BOTTLE_SENSITIVE_PREFIXES": "ANTHROPIC_API_,OTHER_",
}
result = scan_outbound(self._ROUTE, f"auth={secret}", env)
self.assertIsNotNone(result)
def test_canary_detected_via_egress_token_canary(self):
# The canary (injected as EGRESS_TOKEN_CANARY) is caught by known_secrets.
canary = "canaryvalue12345abcdef"
env = {"EGRESS_TOKEN_CANARY": canary}
result = scan_outbound(self._ROUTE, f"data={canary}", env)
self.assertIsNotNone(result)
assert result is not None
self.assertEqual("block", result.severity)
self.assertIn("EGRESS_TOKEN_CANARY", result.reason)
def test_fragmented_canary_blocked(self):
# Canary with separators injected is still caught.
canary = "supersecretcanary99"
env = {"EGRESS_TOKEN_CANARY": canary}
fragmented = "-".join(canary)
result = scan_outbound(self._ROUTE, f"x={fragmented}", env)
self.assertIsNotNone(result)
class TestOutboundDetectorNames(unittest.TestCase):
def test_entropy_in_outbound_detector_names(self):
from bot_bottle.egress_addon_core import OUTBOUND_DETECTOR_NAMES
self.assertIn("entropy", OUTBOUND_DETECTOR_NAMES)
def test_known_secrets_in_outbound_detector_names(self):
from bot_bottle.egress_addon_core import OUTBOUND_DETECTOR_NAMES
self.assertIn("known_secrets", OUTBOUND_DETECTOR_NAMES)
def test_token_patterns_in_outbound_detector_names(self):
from bot_bottle.egress_addon_core import OUTBOUND_DETECTOR_NAMES
self.assertIn("token_patterns", OUTBOUND_DETECTOR_NAMES)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+34
View File
@@ -0,0 +1,34 @@
"""Unit: install.sh static contract checks."""
from __future__ import annotations
import subprocess
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
class TestInstallScript(unittest.TestCase):
def test_shell_syntax(self):
result = subprocess.run(
["sh", "-n", str(ROOT / "install.sh")],
check=False,
capture_output=True,
text=True,
)
self.assertEqual("", result.stderr)
self.assertEqual(0, result.returncode)
def test_contract_phrases(self):
script = (ROOT / "install.sh").read_text(encoding="utf-8")
self.assertIn("python3", script)
self.assertIn("docker info", script)
self.assertIn("pipx install --force", script)
self.assertIn("pip install --user --upgrade", script)
self.assertIn('"${BOT_BOTTLE_BIN}" doctor', script)
if __name__ == "__main__":
unittest.main()
-127
View File
@@ -1,127 +0,0 @@
"""Unit: leveled + structured logging wrappers (issue #252).
Locks three properties of bot_bottle.log:
- backward compatibility default output is byte-identical to the
original bare wrappers, so the 100+ existing single-string call
sites are unaffected;
- context rendering an optional mapping becomes a parseable
` [k=v ...]` suffix;
- level gating BOT_BOTTLE_LOG_LEVEL filters by severity, debug is
silent by default, and error always surfaces.
"""
from __future__ import annotations
import contextlib
import io
import unittest
from typing import Callable
from unittest import mock
from bot_bottle import log
def _capture(
fn: Callable[..., None],
*args: object,
env: dict[str, str] | None = None,
**kwargs: object,
) -> str:
buf = io.StringIO()
patched = mock.patch.dict("os.environ", env or {}, clear=False)
with patched, contextlib.redirect_stderr(buf):
fn(*args, **kwargs)
return buf.getvalue()
class TestBackwardCompat(unittest.TestCase):
"""No context + default level → exactly the legacy lines."""
def test_info(self):
self.assertEqual("bot-bottle: hello\n", _capture(log.info, "hello"))
def test_warn(self):
self.assertEqual(
"bot-bottle: warning: careful\n", _capture(log.warn, "careful")
)
def test_error(self):
self.assertEqual(
"bot-bottle: error: boom\n", _capture(log.error, "boom")
)
class TestContext(unittest.TestCase):
def test_appends_sorted_parseable_suffix(self):
out = _capture(
log.error, "rpc failed", context={"slug": "abc123", "code": "-32603"}
)
# keys sorted: code before slug
self.assertEqual(
"bot-bottle: error: rpc failed [code=-32603 slug=abc123]\n", out
)
def test_quotes_values_with_whitespace(self):
out = _capture(
log.info, "did thing", context={"path": "/a b/c", "ok": "yes"}
)
self.assertEqual(
'bot-bottle: did thing [ok=yes path="/a b/c"]\n', out
)
def test_empty_context_is_noop_suffix(self):
self.assertEqual(
"bot-bottle: x\n", _capture(log.info, "x", context={})
)
class TestLevels(unittest.TestCase):
def test_debug_silent_by_default(self):
self.assertEqual("", _capture(log.debug, "trace"))
def test_debug_emits_when_level_lowered(self):
out = _capture(log.debug, "trace", env={"BOT_BOTTLE_LOG_LEVEL": "debug"})
self.assertEqual("bot-bottle: debug: trace\n", out)
def test_error_level_suppresses_info_and_warn(self):
env = {"BOT_BOTTLE_LOG_LEVEL": "error"}
self.assertEqual("", _capture(log.info, "i", env=env))
self.assertEqual("", _capture(log.warn, "w", env=env))
# error still surfaces — nothing sits above it
self.assertEqual(
"bot-bottle: error: e\n", _capture(log.error, "e", env=env)
)
def test_unknown_level_falls_back_to_default(self):
# garbage value → default INFO threshold, so info still prints
out = _capture(log.info, "i", env={"BOT_BOTTLE_LOG_LEVEL": "loud"})
self.assertEqual("bot-bottle: i\n", out)
def test_warning_alias_accepted(self):
env = {"BOT_BOTTLE_LOG_LEVEL": "warning"}
self.assertEqual("", _capture(log.info, "i", env=env))
self.assertEqual(
"bot-bottle: warning: w\n", _capture(log.warn, "w", env=env)
)
class TestDie(unittest.TestCase):
def test_die_still_raises_and_prints_error(self):
buf = io.StringIO()
with contextlib.redirect_stderr(buf):
with self.assertRaises(log.Die) as cm:
log.die("fatal thing")
self.assertEqual("fatal thing", cm.exception.message)
self.assertIn("bot-bottle: error: fatal thing", buf.getvalue())
def test_die_surfaces_even_at_error_level(self):
buf = io.StringIO()
with mock.patch.dict("os.environ", {"BOT_BOTTLE_LOG_LEVEL": "error"}):
with contextlib.redirect_stderr(buf):
with self.assertRaises(log.Die):
log.die("still fatal")
self.assertIn("bot-bottle: error: still fatal", buf.getvalue())
if __name__ == "__main__":
unittest.main()
+1 -2
View File
@@ -42,7 +42,6 @@ def _plan(
routes_path=routes_path, routes_path=routes_path,
routes=("route",), routes=("route",),
token_env_map={"EGRESS_TOKEN_0": "HOST_TOKEN"}, token_env_map={"EGRESS_TOKEN_0": "HOST_TOKEN"},
canary="",
) )
if git: if git:
key_path = stage_dir / "origin-key" key_path = stage_dir / "origin-key"
@@ -272,7 +271,7 @@ def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
manifest=_MANIFEST, manifest=_MANIFEST,
stage_dir=stage_dir, stage_dir=stage_dir,
git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())), git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
egress_plan=cast(EgressPlan, SimpleNamespace(canary="")), egress_plan=cast(EgressPlan, SimpleNamespace()),
supervise_plan=None, supervise_plan=None,
agent_provision=AgentProvisionPlan( agent_provision=AgentProvisionPlan(
template="claude", template="claude",
+27
View File
@@ -0,0 +1,27 @@
"""Unit: Python package metadata for install script PRD."""
from __future__ import annotations
import tomllib
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
class TestPyproject(unittest.TestCase):
def test_console_script_and_no_runtime_dependencies(self):
data = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))
project = data["project"]
self.assertEqual("bot-bottle", project["name"])
self.assertEqual(">=3.11", project["requires-python"])
self.assertEqual([], project["dependencies"])
self.assertEqual(
"bot_bottle.cli:main",
project["scripts"]["bot-bottle"],
)
if __name__ == "__main__":
unittest.main()