Compare commits

..

2 Commits

Author SHA1 Message Date
didericis-claude 197a60cc56 ci(coverage): install dev requirements on the KVM runner
lint / lint (push) Successful in 41s
The self-hosted KVM runner is a persistent machine, so
--break-system-packages is inappropriate. Use --user instead so
coverage (and pyright/pylint for future jobs) land in ~/.local
and survive between runs without touching the system Python.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-18 09:15:42 +00:00
didericis 1a86e86139 ci(coverage): run the diff-coverage gate on a self-hosted KVM runner
Re-land the coverage gate deferred from #343. The Firecracker VM/SSH
orchestration (~230 lines) is only exercised by the integration suite,
which needs /dev/kvm + the provisioned TAP/nft pool — a container runner
skips it and those lines read uncovered, so the 90% diff gate can't pass
on ubuntu-latest. Move the `coverage` job to a self-hosted `kvm` runner
with a firecracker-readiness preflight (binary + /dev/kvm + `backend
status`) so the integration test actually runs. Unit/lint stay on
ubuntu-latest. README documents the runner prerequisites.

Depends on a registered self-hosted runner labelled `kvm`; until one is
provisioned this gate will not run. See PRD 0069 / #348.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck
2026-07-18 09:12:15 +00:00
19 changed files with 149 additions and 690 deletions
+5 -2
View File
@@ -22,7 +22,10 @@ jobs:
- name: Checkout - name: Checkout
uses: actions/checkout@v4 uses: actions/checkout@v4
# No actions/setup-python: canaries are stdlib unittest on the image's - name: Set up Python
# system Python 3.12 (older act_runner mishandles setup-python's PATH). uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Run canaries - name: Run canaries
run: python3 -m unittest discover -t . -s tests/canaries -v run: python3 -m unittest discover -t . -s tests/canaries -v
+5 -2
View File
@@ -37,8 +37,11 @@ jobs:
fetch-depth: 0 fetch-depth: 0
token: ${{ secrets.GITHUB_TOKEN }} token: ${{ secrets.GITHUB_TOKEN }}
# No actions/setup-python: the inline script is stdlib-only on the - name: Set up Python
# image's system Python 3.12 (older act_runner mishandles its PATH). uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Configure git - name: Configure git
run: | run: |
git config user.name "github-actions[bot]" git config user.name "github-actions[bot]"
+35 -17
View File
@@ -68,30 +68,48 @@ jobs:
- name: Run integration tests - name: Run integration tests
run: python3 -m unittest discover -t . -s tests/integration -v run: python3 -m unittest discover -t . -s tests/integration -v
# Combined unit+integration coverage report (informational). See # Combined unit+integration coverage + the diff-coverage gate (the hard
# docs/decisions/0004-coverage-policy.md. # gate: new/changed lines >= 90%). See docs/decisions/0004-coverage-policy.md.
# #
# The hard diff-coverage gate (changed lines >= 90%) is DEFERRED: the # This runs on a self-hosted KVM runner (label `kvm`), NOT ubuntu-latest,
# Firecracker backend's VM/SSH orchestration is covered by the integration # because the Firecracker backend's subprocess/VM orchestration
# suite, which needs /dev/kvm + the provisioned TAP/nft pool — a # (launch/boot/SSH/isolation-probe) is covered by the integration suite,
# container-based runner skips it and those lines read uncovered, so the # and that suite needs `/dev/kvm` + the provisioned TAP/nft pool — which a
# gate can't pass here. Re-enabling it on a self-hosted KVM runner is # container-based runner doesn't have. On such a runner the firecracker
# tracked separately (see PRD 0069 / #348 and the ci-runner branch). # integration test skips and its ~230 orchestration lines read as
# uncovered, so the gate can't pass there.
#
# Runner prerequisites (provision once on the host; see the README
# "Firecracker on Linux" section): the `firecracker` binary on PATH,
# `/dev/kvm` accessible to the runner user, Docker, the cached guest
# kernel + static dropbear (BOT_BOTTLE_FC_KERNEL / BOT_BOTTLE_FC_DROPBEAR),
# and the network pool installed as the persistent systemd unit
# (`./cli.py backend setup --backend=firecracker`). The preflight step
# below fails fast with instructions if anything is missing.
coverage: coverage:
runs-on: ubuntu-latest runs-on: [self-hosted, kvm]
steps: steps:
- name: Checkout - name: Checkout
uses: actions/checkout@v4 uses: actions/checkout@v4
with: with:
fetch-depth: 0 fetch-depth: 0
# No actions/setup-python: the runner image already ships Python 3.12, - name: Preflight — Firecracker host is ready
# and older act_runner engines mishandle setup-python's PATH (coverage run: |
# lands in one interpreter, `python3` resolves to another). Install command -v firecracker >/dev/null || {
# straight into the ephemeral job container's system Python — echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; }
# --break-system-packages is safe because the container is disposable. test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; }
- name: Install dev requirements # `backend status` exits non-zero unless the TAP pool is up + no
run: python3 -m pip install --break-system-packages -r requirements-dev.txt # range overlap; it prints the exact `backend setup` fix.
python3 cli.py backend status --backend=firecracker
- name: Combined coverage report (unit + integration) - name: Install dev requirements
run: python3 -m pip install --user -r requirements-dev.txt
- name: Combined coverage (unit + integration, incl. firecracker)
run: PYTHON=python3 bash scripts/coverage.sh critical run: PYTHON=python3 bash scripts/coverage.sh critical
- name: Diff-coverage gate (changed lines >= 90%)
run: |
git fetch --no-tags origin main:refs/remotes/origin/main
python3 scripts/diff_coverage.py --base origin/main --min 90
+11 -8
View File
@@ -20,18 +20,21 @@ jobs:
fetch-depth: 0 fetch-depth: 0
token: ${{ secrets.GITHUB_TOKEN }} token: ${{ secrets.GITHUB_TOKEN }}
# No actions/setup-python: the runner image ships Python 3.12 and older - name: Set up Python
# act_runner engines mishandle setup-python's PATH. Install into the uses: actions/setup-python@v4
# ephemeral job container's system Python (--break-system-packages is with:
# safe because the container is disposable). python-version: '3.12'
- name: Install dev dependencies - name: Install dev dependencies
run: python3 -m pip install --break-system-packages -r requirements-dev.txt run: |
python -m pip install --upgrade pip
pip install -r requirements-dev.txt
- name: Run coverage and extract percentage - name: Run coverage and extract percentage
id: coverage id: coverage
run: | run: |
python3 -m coverage run -m unittest discover -t . -s tests/unit > /dev/null 2>&1 || true python -m coverage run -m unittest discover -t . -s tests/unit > /dev/null 2>&1 || true
PERCENT=$(python3 -m coverage report 2>/dev/null | grep '^TOTAL' | grep -oP '\d+(?=%)' | tail -1) PERCENT=$(python -m coverage report 2>/dev/null | grep '^TOTAL' | grep -oP '\d+(?=%)' | tail -1)
echo "percent=$PERCENT" >> $GITHUB_OUTPUT echo "percent=$PERCENT" >> $GITHUB_OUTPUT
echo "Coverage: $PERCENT%" echo "Coverage: $PERCENT%"
@@ -42,7 +45,7 @@ jobs:
# the single source of truth in scripts/critical-modules.txt; every # the single source of truth in scripts/critical-modules.txt; every
# core module is unit-tested, so the unit-only run is accurate for it. # core module is unit-tested, so the unit-only run is accurate for it.
INCLUDE=$(grep -vE '^[[:space:]]*(#|$)' scripts/critical-modules.txt | paste -sd, -) INCLUDE=$(grep -vE '^[[:space:]]*(#|$)' scripts/critical-modules.txt | paste -sd, -)
PERCENT=$(python3 -m coverage report --include="$INCLUDE" 2>/dev/null | grep '^TOTAL' | grep -oP '\d+(?=%)' | tail -1) PERCENT=$(python -m coverage report --include="$INCLUDE" 2>/dev/null | grep '^TOTAL' | grep -oP '\d+(?=%)' | tail -1)
echo "percent=$PERCENT" >> $GITHUB_OUTPUT echo "percent=$PERCENT" >> $GITHUB_OUTPUT
echo "Core coverage: $PERCENT%" echo "Core coverage: $PERCENT%"
+4 -2
View File
@@ -5,8 +5,8 @@
# bot-bottle # bot-bottle
[![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml) [![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
[![coverage](https://img.shields.io/badge/coverage-81%25-brightgreen)](https://coverage.readthedocs.io/) [![coverage](https://img.shields.io/badge/coverage-82%25-brightgreen)](https://coverage.readthedocs.io/)
[![core coverage](https://img.shields.io/badge/core%20coverage-94%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md) [![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
**Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data. **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.
@@ -90,6 +90,8 @@ BOT_BOTTLE_BACKEND=firecracker ./cli.py start <agent>
> **NixOS:** enable `virtualisation.docker`, ensure the KVM module is loaded (`boot.kernelModules = [ "kvm-intel" ];` or `kvm-amd`), and add your user to the `kvm` and `docker` groups. For the network pool, consume the flake module — `imports = [ inputs.bot-bottle.nixosModules.firecracker-netpool ]; services.bot-bottle-firecracker = { enable = true; owner = "you"; };` — then `nixos-rebuild switch` (imperative nft/TAP rules don't survive a rebuild; channel users can `imports = [ <bot-bottle>/nix/firecracker-netpool.nix ]`). `firecracker` isn't in nixpkgs by default as a user binary — install the release binary (pin the version) and put it on `PATH`. > **NixOS:** enable `virtualisation.docker`, ensure the KVM module is loaded (`boot.kernelModules = [ "kvm-intel" ];` or `kvm-amd`), and add your user to the `kvm` and `docker` groups. For the network pool, consume the flake module — `imports = [ inputs.bot-bottle.nixosModules.firecracker-netpool ]; services.bot-bottle-firecracker = { enable = true; owner = "you"; };` — then `nixos-rebuild switch` (imperative nft/TAP rules don't survive a rebuild; channel users can `imports = [ <bot-bottle>/nix/firecracker-netpool.nix ]`). `firecracker` isn't in nixpkgs by default as a user binary — install the release binary (pin the version) and put it on `PATH`.
> **CI:** the coverage gate (`.gitea/workflows/test.yml` → `coverage` job) runs on a self-hosted runner labelled `kvm`, because the Firecracker backend's VM/SSH orchestration is exercised only by the integration suite, which needs `/dev/kvm` + the provisioned pool (a container runner would skip it and read as uncovered). Provision that runner exactly like a normal Firecracker host — `firecracker` on `PATH`, `/dev/kvm`, Docker, the cached guest kernel + static dropbear, and the pool installed as the persistent systemd unit — then register it with the `kvm` label. The unit/lint jobs still run on `ubuntu-latest`.
```sh ```sh
./cli.py start <agent> # builds the image on first run, drops you into claude ./cli.py start <agent> # builds the image on first run, drops you into claude
``` ```
-4
View File
@@ -45,10 +45,6 @@ PROVIDER_TEMPLATES = frozenset({PROVIDER_CLAUDE, PROVIDER_CODEX, PROVIDER_PI})
# forward_host_credentials is enabled. Pipelock must pass these through # forward_host_credentials is enabled. Pipelock must pass these through
# (no TLS MITM) or its header DLP blocks the injected JWT. # (no TLS MITM) or its header DLP blocks the injected JWT.
CODEX_HOST_CREDENTIAL_HOSTS = ("api.openai.com", "chatgpt.com") CODEX_HOST_CREDENTIAL_HOSTS = ("api.openai.com", "chatgpt.com")
# Host that egress injects the host Claude bearer on when Claude
# forward_host_credentials is enabled.
CLAUDE_HOST_CREDENTIAL_HOSTS = ("api.anthropic.com",)
PromptMode = Literal[ PromptMode = Literal[
"append_file", "append_file",
"read_prompt_file", "read_prompt_file",
+31 -73
View File
@@ -40,7 +40,7 @@ from abc import ABC, abstractmethod
from contextlib import AbstractContextManager from contextlib import AbstractContextManager
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, Any, Generic, Sequence, TypeVar from typing import Any, Generic, Sequence, TypeVar
from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
from ..egress import EgressPlan from ..egress import EgressPlan
@@ -54,9 +54,6 @@ from ..workspace import WorkspacePlan, workspace_plan
from .print_util import print_multi, visible_agent_env_names from .print_util import print_multi, visible_agent_env_names
from .util import host_skill_dir from .util import host_skill_dir
if TYPE_CHECKING:
from .freeze import CommitCancelled, Freezer, get_freezer
@dataclass(frozen=True) @dataclass(frozen=True)
class BottleSpec: class BottleSpec:
@@ -587,63 +584,28 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
Not called by the launch path or the test suite.""" Not called by the launch path or the test suite."""
# _backends is None until the first call to _get_backends(), at which # Import concrete backend classes AFTER the base types are defined, so
# point all three concrete backend classes are imported and instantiated. # each backend module can pull BottleSpec / BottlePlan / BottleBackend
# Keeping the imports out of module scope means that importing any # via `from . import ...` without hitting a partially-initialized module.
# backend sub-module (e.g. `backend.docker.util`) no longer drags the from .docker import DockerBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
# firecracker and macos-container implementations into memory. from .firecracker import FirecrackerBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
# from .macos_container import MacosContainerBottleBackend # noqa: E402 # pylint: disable=wrong-import-position
# Tests may replace _backends with a {name: fake} dict via patch.object;
# _get_backends() returns the current module-level value as-is when it # Freezer is imported after the backend classes for the same reason:
# is not None, so test fakes take effect without triggering real imports. # Freezer.commit_slug constructs ActiveAgent, which must be fully
_backends: dict[str, BottleBackend[Any, Any]] | None = None # defined first.
from .freeze import CommitCancelled, Freezer, get_freezer # noqa: E402 # pylint: disable=wrong-import-position
def _get_backends() -> dict[str, BottleBackend[Any, Any]]: # The dict is heterogeneous: each value is a BottleBackend specialized
"""Return the registry of all backend instances, loading lazily on first call.""" # over its own plan type. Concrete plan types are erased here because
global _backends # pylint: disable=global-statement # the registry is selected at runtime and the CLI only needs the
if _backends is None: # unparameterized methods (prepare → plan → launch(plan), cleanup, etc.).
from .docker import DockerBottleBackend _BACKENDS: dict[str, BottleBackend[Any, Any]] = {
from .firecracker import FirecrackerBottleBackend "docker": DockerBottleBackend(),
from .macos_container import MacosContainerBottleBackend "firecracker": FirecrackerBottleBackend(),
_backends = { "macos-container": MacosContainerBottleBackend(),
"docker": DockerBottleBackend(), }
"firecracker": FirecrackerBottleBackend(),
"macos-container": MacosContainerBottleBackend(),
}
return _backends
def __getattr__(name: str) -> Any:
"""Lazily surface concrete backend classes and freeze symbols at the
package level so existing `from bot_bottle.backend import X` and
`patch.object(backend_mod, X, ...)` call-sites keep working without
forcing an import of every backend at module-init time."""
if name == "DockerBottleBackend":
from .docker import DockerBottleBackend
globals()[name] = DockerBottleBackend
return DockerBottleBackend
if name == "FirecrackerBottleBackend":
from .firecracker import FirecrackerBottleBackend
globals()[name] = FirecrackerBottleBackend
return FirecrackerBottleBackend
if name == "MacosContainerBottleBackend":
from .macos_container import MacosContainerBottleBackend
globals()[name] = MacosContainerBottleBackend
return MacosContainerBottleBackend
if name == "CommitCancelled":
from .freeze import CommitCancelled
globals()[name] = CommitCancelled
return CommitCancelled
if name == "Freezer":
from .freeze import Freezer
globals()[name] = Freezer
return Freezer
if name == "get_freezer":
from .freeze import get_freezer
globals()[name] = get_freezer
return get_freezer
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
def get_bottle_backend( def get_bottle_backend(
@@ -661,11 +623,10 @@ def get_bottle_backend(
Dies with a pointer at the known backends if the chosen name Dies with a pointer at the known backends if the chosen name
isn't implemented.""" isn't implemented."""
resolved = name or os.environ.get("BOT_BOTTLE_BACKEND") or _default_backend_name() resolved = name or os.environ.get("BOT_BOTTLE_BACKEND") or _default_backend_name()
backends = _get_backends() if resolved not in _BACKENDS:
if resolved not in backends: known = ", ".join(sorted(_BACKENDS))
known = ", ".join(sorted(backends))
die(f"unknown backend {resolved!r}; known backends: {known}") die(f"unknown backend {resolved!r}; known backends: {known}")
return backends[resolved] return _BACKENDS[resolved]
def _default_backend_name() -> str: def _default_backend_name() -> str:
@@ -675,17 +636,16 @@ def _default_backend_name() -> str:
# `firecracker` binary isn't installed yet: selecting it here routes # `firecracker` binary isn't installed yet: selecting it here routes
# start through firecracker's preflight, which prints an install # start through firecracker's preflight, which prints an install
# pointer, instead of silently falling back to docker. # pointer, instead of silently falling back to docker.
from .firecracker import FirecrackerBottleBackend
if FirecrackerBottleBackend.is_host_capable(): if FirecrackerBottleBackend.is_host_capable():
return "firecracker" return "firecracker"
return "docker" return "docker"
def known_backend_names() -> tuple[str, ...]: def known_backend_names() -> tuple[str, ...]:
"""Sorted tuple of all backend keys in `_get_backends()`. Used by """Sorted tuple of all backend keys in `_BACKENDS`. Used by
argparse (`--backend` choices) and the dashboard's backend argparse (`--backend` choices) and the dashboard's backend
picker.""" picker."""
return tuple(sorted(_get_backends())) return tuple(sorted(_BACKENDS))
def has_backend(name: str) -> bool: def has_backend(name: str) -> bool:
@@ -697,10 +657,9 @@ def has_backend(name: str) -> bool:
Returns False for unknown names so callers can pass Returns False for unknown names so callers can pass
arbitrary input without separate validation.""" arbitrary input without separate validation."""
backends = _get_backends() if name not in _BACKENDS:
if name not in backends:
return False return False
return backends[name].is_available() return _BACKENDS[name].is_available()
def enumerate_active_agents() -> list[ActiveAgent]: def enumerate_active_agents() -> list[ActiveAgent]:
@@ -716,11 +675,10 @@ def enumerate_active_agents() -> list[ActiveAgent]:
deterministic tiebreaker. Agents with missing metadata deterministic tiebreaker. Agents with missing metadata
(`started_at == ""`) sort first.""" (`started_at == ""`) sort first."""
out: list[ActiveAgent] = [] out: list[ActiveAgent] = []
backends = _get_backends() for name in known_backend_names():
for name in sorted(backends): if not has_backend(name):
if not backends[name].is_available():
continue continue
out.extend(backends[name].enumerate_active()) out.extend(_BACKENDS[name].enumerate_active())
out.sort(key=lambda a: (a.started_at, a.slug)) out.sort(key=lambda a: (a.started_at, a.slug))
return out return out
+34 -7
View File
@@ -8,7 +8,7 @@ import os
import re import re
import shutil import shutil
import subprocess import subprocess
from typing import Iterator from typing import Iterable, Iterator
from ...docker_cmd import run_docker from ...docker_cmd import run_docker
from ...log import die, info from ...log import die, info
@@ -32,7 +32,12 @@ def container_name_candidates(base: str) -> Iterator[str]:
def runsc_available() -> bool: def runsc_available() -> bool:
"""Return True if the Docker daemon has the gVisor (`runsc`) runtime """Return True if the Docker daemon has the gVisor (`runsc`) runtime
registered. Called once per prepare; the result lives on the plan.""" registered. Called once per prepare; the result lives on the plan."""
r = run_docker(["docker", "info", "--format", "{{json .Runtimes}}"]) r = subprocess.run(
["docker", "info", "--format", "{{json .Runtimes}}"],
capture_output=True,
text=True,
check=False,
)
return r.returncode == 0 and "runsc" in r.stdout return r.returncode == 0 and "runsc" in r.stdout
@@ -46,15 +51,20 @@ def require_docker() -> None:
def image_exists(ref: str) -> bool: def image_exists(ref: str) -> bool:
return run_docker(["docker", "image", "inspect", ref]).returncode == 0 return _silent_run(["docker", "image", "inspect", ref]) == 0
def container_exists(name: str) -> bool: def container_exists(name: str) -> bool:
"""Returns True if a container (running or stopped) with the given """Returns True if a container (running or stopped) with the given
name exists. Uses `docker ps -a -q -f name=^<name>$` so substring name exists. Uses `docker ps -a -q -f name=^<name>$` so substring
matches don't false-positive.""" matches don't false-positive."""
result = run_docker(["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"]) result = subprocess.run(
return result.returncode == 0 and bool(result.stdout.strip()) ["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"],
capture_output=True,
text=True,
check=True,
)
return bool(result.stdout.strip())
def force_remove_container(name: str) -> None: def force_remove_container(name: str) -> None:
@@ -62,7 +72,12 @@ def force_remove_container(name: str) -> None:
doesn't — and the rm itself is best-effort (errors swallowed) so doesn't — and the rm itself is best-effort (errors swallowed) so
this is safe to register as a teardown callback.""" this is safe to register as a teardown callback."""
if container_exists(name): if container_exists(name):
run_docker(["docker", "rm", "-f", name]) subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
def docker_exec_root(container: str, argv: list[str]) -> None: def docker_exec_root(container: str, argv: list[str]) -> None:
@@ -190,10 +205,22 @@ def verify_agent_image(image: str, argv: tuple[str, ...]) -> None:
def commit_container(container_name: str, image_tag: str) -> None: def commit_container(container_name: str, image_tag: str) -> None:
"""Run `docker commit <container_name> <image_tag>` to snapshot the """Run `docker commit <container_name> <image_tag>` to snapshot the
running container's filesystem state as a local Docker image.""" running container's filesystem state as a local Docker image."""
result = run_docker(["docker", "commit", container_name, image_tag]) result = subprocess.run(
["docker", "commit", container_name, image_tag],
capture_output=True, text=True, check=False,
)
if result.returncode != 0: if result.returncode != 0:
die( die(
f"docker commit {container_name!r}{image_tag!r} failed: " f"docker commit {container_name!r}{image_tag!r} failed: "
f"{(result.stderr or '').strip() or '<no stderr>'}" f"{(result.stderr or '').strip() or '<no stderr>'}"
) )
info(f"committed {container_name!r}{image_tag!r}") info(f"committed {container_name!r}{image_tag!r}")
def _silent_run(cmd: Iterable[str]) -> int:
return subprocess.run(
list(cmd),
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode
+5 -17
View File
@@ -23,9 +23,8 @@ from ...agent_provider import (
provider_startup_args, provider_startup_args,
) )
from ...backend.docker import util as docker_mod from ...backend.docker import util as docker_mod
from ...egress import CLAUDE_HOST_CREDENTIAL_TOKEN_REF, EgressRoute from ...egress import EgressRoute
from ...log import die, info, warn from ...log import die, info, warn
from .claude_auth import claude_host_access_token
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -119,6 +118,7 @@ class ClaudeAgentProvider(AgentProvider):
color: str = "", color: str = "",
provider_settings: dict[str, object] | None = None, provider_settings: dict[str, object] | None = None,
) -> AgentProvisionPlan: ) -> AgentProvisionPlan:
del forward_host_credentials, host_env
resolved_guest_env = dict(guest_env or {}) resolved_guest_env = dict(guest_env or {})
startup_args = provider_startup_args(provider_settings) startup_args = provider_startup_args(provider_settings)
guest_home = self.guest_home guest_home = self.guest_home
@@ -180,24 +180,13 @@ class ClaudeAgentProvider(AgentProvider):
claude_settings, claude_settings,
f"{guest_home}/.claude/settings.json", f"{guest_home}/.claude/settings.json",
)) ))
provisioned_env: dict[str, str] = {}
if forward_host_credentials:
_host_env = host_env or dict(os.environ)
provisioned_env[CLAUDE_HOST_CREDENTIAL_TOKEN_REF] = (
claude_host_access_token(_host_env)
)
cred_token_ref = (
CLAUDE_HOST_CREDENTIAL_TOKEN_REF if forward_host_credentials
else auth_token
)
egress_routes = (EgressRoute( egress_routes = (EgressRoute(
host="api.anthropic.com", host="api.anthropic.com",
auth_scheme="Bearer" if (auth_token or forward_host_credentials) else "", auth_scheme="Bearer" if auth_token else "",
token_ref=cred_token_ref, token_ref=auth_token,
),) ),)
hidden_env_names: frozenset[str] = frozenset() hidden_env_names: frozenset[str] = frozenset()
if auth_token or forward_host_credentials: if auth_token:
env_vars["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-placeholder" env_vars["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-placeholder"
hidden_env_names = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"}) hidden_env_names = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"})
@@ -219,7 +208,6 @@ class ClaudeAgentProvider(AgentProvider):
files=tuple(files), files=tuple(files),
egress_routes=egress_routes, egress_routes=egress_routes,
hidden_env_names=hidden_env_names, hidden_env_names=hidden_env_names,
provisioned_env=provisioned_env,
) )
def provision_skills(self, plan: "BottlePlan", bottle: "Bottle") -> None: def provision_skills(self, plan: "BottlePlan", bottle: "Bottle") -> None:
-114
View File
@@ -1,114 +0,0 @@
"""Host Claude auth helpers.
Reads the host's Claude Code credentials and returns only the access
token needed by egress. Does not expose refresh tokens or raw payloads.
Credential storage by platform:
Linux ~/.claude/.credentials.json
macOS macOS Keychain, service "Claude Code-credentials"
(file path is tried first; Keychain is the fallback)
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from ...log import die
_KEYCHAIN_SERVICE = "Claude Code-credentials"
def claude_auth_path(host_env: dict[str, str] | None = None) -> Path:
env = os.environ if host_env is None else host_env
home = env.get("HOME")
if home:
return Path(home) / ".claude" / ".credentials.json"
return Path.home() / ".claude" / ".credentials.json"
def _read_keychain() -> dict[str, object] | None:
"""Try the macOS Keychain. Returns parsed JSON dict or None."""
if sys.platform != "darwin":
return None
try:
result = subprocess.run(
["security", "find-generic-password", "-s", _KEYCHAIN_SERVICE, "-w"],
capture_output=True,
text=True,
timeout=10,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
if result.returncode != 0 or not result.stdout.strip():
return None
try:
raw = json.loads(result.stdout.strip())
except json.JSONDecodeError:
return None
return raw if isinstance(raw, dict) else None
def claude_host_access_token(
host_env: dict[str, str] | None = None,
*,
now: datetime | None = None,
) -> str:
path = claude_auth_path(host_env)
raw: dict[str, object] | None = None
if path.is_file():
try:
raw = json.loads(path.read_text())
except (OSError, json.JSONDecodeError) as e:
die(f"claude host credentials: could not read valid JSON at {path}: {e}")
if not isinstance(raw, dict):
die(f"claude host credentials: {path} must contain a JSON object")
else:
raw = _read_keychain()
if raw is None:
die(
f"claude host credentials: auth file missing at {path} and "
f"macOS Keychain lookup for '{_KEYCHAIN_SERVICE}' failed. "
"Run `claude login` on the host or disable "
"agent_provider.forward_host_credentials."
)
oauth = raw.get("claudeAiOauth")
if not isinstance(oauth, dict):
die(
"claude host credentials: claudeAiOauth is missing from credentials. "
"Run `claude login` on the host or disable "
"agent_provider.forward_host_credentials."
)
access_token = oauth.get("accessToken")
if not isinstance(access_token, str) or not access_token:
die(
"claude host credentials: claudeAiOauth.accessToken is missing or empty. "
"Run `claude login` on the host and restart the bottle."
)
# expiresAt is in milliseconds
expires_at = oauth.get("expiresAt")
if isinstance(expires_at, (int, float)):
check_now = now or datetime.now(timezone.utc)
exp_dt = datetime.fromtimestamp(float(expires_at) / 1000.0, timezone.utc)
if exp_dt <= check_now:
die(
"claude host credentials: host Claude access token is expired. "
"Run `claude login` on the host and restart the bottle."
)
return access_token
__all__ = [
"claude_auth_path",
"claude_host_access_token",
]
-2
View File
@@ -30,7 +30,6 @@ if TYPE_CHECKING:
from .manifest import ManifestBottle from .manifest import ManifestBottle
CODEX_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CODEX_HOST_ACCESS_TOKEN" CODEX_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CODEX_HOST_ACCESS_TOKEN"
CLAUDE_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN"
EGRESS_HOSTNAME = "egress" EGRESS_HOSTNAME = "egress"
@@ -401,7 +400,6 @@ class Egress(ABC):
) )
__all__ = [ __all__ = [
"CLAUDE_HOST_CREDENTIAL_TOKEN_REF",
"CODEX_HOST_CREDENTIAL_TOKEN_REF", "CODEX_HOST_CREDENTIAL_TOKEN_REF",
"EGRESS_HOSTNAME", "EGRESS_HOSTNAME",
"EGRESS_ROUTES_FILENAME", "EGRESS_ROUTES_FILENAME",
+4 -10
View File
@@ -25,9 +25,8 @@ class ManifestAgentProvider:
header, and sets a placeholder CLAUDE_CODE_OAUTH_TOKEN in the agent header, and sets a placeholder CLAUDE_CODE_OAUTH_TOKEN in the agent
so the Claude Code CLI starts. so the Claude Code CLI starts.
`forward_host_credentials` forwards the host provider auth token into `forward_host_credentials` forwards the host Codex auth token into
the egress sidecar (Codex and Claude). For Codex this reads the egress daemon (Codex only).
`~/.codex/auth.json`; for Claude it reads `~/.claude/.credentials.json`.
""" """
template: str = "claude" template: str = "claude"
@@ -93,15 +92,10 @@ class ManifestAgentProvider:
f"is only supported for built-in templates " f"is only supported for built-in templates "
f"({', '.join(sorted(PROVIDER_TEMPLATES))})" f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
) )
if forward_host_credentials and template not in {"codex", "claude"}: if forward_host_credentials and template != "codex":
raise ManifestError( raise ManifestError(
f"bottle '{bottle_name}' agent_provider.forward_host_credentials " f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
"is only supported for templates 'codex' and 'claude'" "is currently only supported for template 'codex'"
)
if forward_host_credentials and auth_token:
raise ManifestError(
f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
"and auth_token both set; use one or the other"
) )
settings = _parse_provider_settings(bottle_name, template, d.get("settings")) settings = _parse_provider_settings(bottle_name, template, d.get("settings"))
return cls( return cls(
@@ -1,146 +0,0 @@
# PRD prd-new: Claude forward_host_credentials
- **Status:** Draft
- **Author:** claude
- **Created:** 2026-07-01
- **Issue:** #325
## Summary
Add `agent_provider.forward_host_credentials: true` support for the
`claude` template, mirroring the existing Codex flow. When enabled,
bot-bottle reads the host's Claude OAuth session key from
`~/.claude/.credentials.json` at launch, forwards it only to the egress sidecar,
and injects a placeholder `CLAUDE_CODE_OAUTH_TOKEN` into the agent so
Claude Code starts without ever seeing the real credential.
## Problem
Running a Claude agent in a container today requires the operator to
manually extract a long-lived OAuth token (`claude setup-token`), export
it as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`, and reference it explicitly in
the manifest with `agent_provider.auth_token:
"BOT_BOTTLE_CLAUDE_OAUTH_TOKEN"`. This is a two-step manual ceremony
that is easy to skip or do incorrectly.
The host already stores a valid Claude session in `~/.claude/.credentials.json`
after `claude login`. Codex already automates an
equivalent extraction from `~/.codex/auth.json`. There is no reason
Claude bottles cannot do the same.
## Goals / Success Criteria
- A Claude bottle with `forward_host_credentials: true` in the manifest
uses the host's `~/.claude/.credentials.json` session key at launch with no
additional operator steps.
- The agent container receives only `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder`
— never the real token.
- The real session key lives only in the egress sidecar's environment.
- Missing, malformed, or expired host Claude auth fails launch with a
clear operator-facing message.
- Existing `auth_token` behavior is unchanged.
- `forward_host_credentials: true` is rejected in the manifest when both
`auth_token` and `forward_host_credentials` are set, since they serve
the same purpose.
## Non-goals
- Refreshing Claude OAuth tokens in the sidecar.
- Writing a dummy `~/.claude.json` auth state to the agent (unlike the
Codex flow, Claude Code reads its credential from `CLAUDE_CODE_OAUTH_TOKEN`
in env, not from an auth file — no guest-side auth marker is needed).
- Supporting `forward_host_credentials` for providers other than `codex`
and `claude`.
## Design
### Manifest schema
```yaml
agent_provider:
template: claude
forward_host_credentials: true
```
Rejects in manifest validation when:
- Template is not `codex` or `claude`.
- Both `auth_token` and `forward_host_credentials` are set.
### Host auth extraction (`contrib/claude/claude_auth.py`)
Claude Code credential storage varies by platform:
- **Linux**: `~/.claude/.credentials.json`
- **macOS**: macOS Keychain, service `"Claude Code-credentials"`
(the file path is tried first; Keychain is the fallback when the file
is absent)
`~/.claude.json` contains only UI state and profile metadata — no token.
The credentials JSON schema (same whether from file or Keychain):
```json
{
"claudeAiOauth": {
"accessToken": "<access-token>",
"refreshToken": "<refresh-token>",
"expiresAt": 1748276587173,
"scopes": ["user:inference", "user:profile"]
}
}
```
`expiresAt` is in **milliseconds** (not seconds).
At prepare/launch time, when `forward_host_credentials: true`:
1. Try `~/.claude/.credentials.json`; on macOS, if absent, run
`security find-generic-password -s "Claude Code-credentials" -w`
and parse its stdout as JSON.
2. Require a `claudeAiOauth` dict.
3. Require a non-empty `claudeAiOauth.accessToken` string.
4. If `claudeAiOauth.expiresAt` is present, divide by 1000 and require
the result to be in the future.
5. Return only the access token to the launch path.
Errors name the missing or invalid condition and point the operator at
`claude login`, without printing token values.
### Egress route
When `forward_host_credentials: true`:
- Provision the session key in `provisioned_env` under
`BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN` (new constant in `egress.py`).
- Set up the `api.anthropic.com` egress route with `auth_scheme: Bearer`
and `token_ref: BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN`.
- Set `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder` in the agent env and
add it to `hidden_env_names`.
No dummy auth file and no `verify` step are needed — Claude Code reads
the credential from the env var, not from a file.
### Constants
- `CLAUDE_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN"`
in `egress.py` (alongside the existing `CODEX_HOST_CREDENTIAL_TOKEN_REF`).
- `CLAUDE_HOST_CREDENTIAL_HOSTS = ("api.anthropic.com",)` in
`agent_provider.py` (alongside the existing `CODEX_HOST_CREDENTIAL_HOSTS`).
### Data flow
```
Host ~/.claude/.credentials.json → bot-bottle launch
├──► egress sidecar env (real token only)
└──► agent env: CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder
Agent → HTTPS to api.anthropic.com (via egress)
Egress → injects Authorization: Bearer <real token>
Egress → forwards to api.anthropic.com
```
## Open questions
None — the Codex precedent makes the design clear.
+1 -66
View File
@@ -9,15 +9,11 @@ import unittest
from pathlib import Path from pathlib import Path
from bot_bottle.agent_provider import ( from bot_bottle.agent_provider import (
CLAUDE_HOST_CREDENTIAL_HOSTS,
CODEX_HOST_CREDENTIAL_HOSTS, CODEX_HOST_CREDENTIAL_HOSTS,
build_agent_provision_plan, build_agent_provision_plan,
prompt_args, prompt_args,
) )
from bot_bottle.egress import ( from bot_bottle.egress import CODEX_HOST_CREDENTIAL_TOKEN_REF
CLAUDE_HOST_CREDENTIAL_TOKEN_REF,
CODEX_HOST_CREDENTIAL_TOKEN_REF,
)
def _jwt(exp: int) -> str: def _jwt(exp: int) -> str:
@@ -296,67 +292,6 @@ class TestAgentProviderRuntime(unittest.TestCase):
) )
self.assertEqual({}, plan.provisioned_env) self.assertEqual({}, plan.provisioned_env)
def test_claude_forward_host_credentials_populates_egress_route(self):
access_token = "sk-ant-oat01-test-key" # gitleaks:allow
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
home = Path(tmp) / "host-claude"
cred_dir = home / ".claude"
cred_dir.mkdir(parents=True)
(cred_dir / ".credentials.json").write_text(json.dumps({
"claudeAiOauth": {"accessToken": access_token},
}))
plan = build_agent_provision_plan(
template="claude",
dockerfile="",
state_dir=Path(tmp),
instance_name="bot-bottle-test",
prompt_file=Path(tmp) / "prompt.txt",
forward_host_credentials=True,
host_env={"HOME": str(home)},
)
self.assertEqual(1, len(plan.egress_routes))
route = plan.egress_routes[0]
self.assertIn(route.host, CLAUDE_HOST_CREDENTIAL_HOSTS)
self.assertEqual("Bearer", route.auth_scheme)
self.assertEqual(CLAUDE_HOST_CREDENTIAL_TOKEN_REF, route.token_ref)
self.assertEqual("egress-placeholder", plan.env_vars["CLAUDE_CODE_OAUTH_TOKEN"])
self.assertEqual(frozenset({"CLAUDE_CODE_OAUTH_TOKEN"}), plan.hidden_env_names)
def test_claude_forward_host_credentials_populates_provisioned_env(self):
access_token = "sk-ant-oat01-test-key" # gitleaks:allow
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
home = Path(tmp) / "host-claude"
cred_dir = home / ".claude"
cred_dir.mkdir(parents=True)
(cred_dir / ".credentials.json").write_text(json.dumps({
"claudeAiOauth": {"accessToken": access_token},
}))
plan = build_agent_provision_plan(
template="claude",
dockerfile="",
state_dir=Path(tmp),
instance_name="bot-bottle-test",
prompt_file=Path(tmp) / "prompt.txt",
forward_host_credentials=True,
host_env={"HOME": str(home)},
)
self.assertEqual(
{CLAUDE_HOST_CREDENTIAL_TOKEN_REF: access_token},
plan.provisioned_env,
)
def test_claude_without_forward_host_credentials_has_empty_provisioned_env(self):
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
plan = build_agent_provision_plan(
template="claude",
dockerfile="",
state_dir=Path(tmp),
instance_name="bot-bottle-test",
prompt_file=Path(tmp) / "prompt.txt",
forward_host_credentials=False,
)
self.assertEqual({}, plan.provisioned_env)
def test_pi_plan_writes_default_ollama_models(self): def test_pi_plan_writes_default_ollama_models(self):
with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp: with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
plan = build_agent_provision_plan( plan = build_agent_provision_plan(
+8 -8
View File
@@ -40,7 +40,7 @@ class TestGetBottleBackend(unittest.TestCase):
return True return True
with patch.dict(os.environ, {}, clear=True), \ with patch.dict(os.environ, {}, clear=True), \
patch.object(backend_mod, "_backends", { patch.object(backend_mod, "_BACKENDS", {
"macos-container": _FakeBackend(), "macos-container": _FakeBackend(),
"docker": _FakeBackend(), "docker": _FakeBackend(),
}): }):
@@ -61,7 +61,7 @@ class TestGetBottleBackend(unittest.TestCase):
with patch.dict(os.environ, {}, clear=True), \ with patch.dict(os.environ, {}, clear=True), \
patch.object(backend_mod.FirecrackerBottleBackend, patch.object(backend_mod.FirecrackerBottleBackend,
"is_host_capable", classmethod(lambda cls: False)), \ "is_host_capable", classmethod(lambda cls: False)), \
patch.object(backend_mod, "_backends", { patch.object(backend_mod, "_BACKENDS", {
"macos-container": _FakeBackend("macos-container", False), "macos-container": _FakeBackend("macos-container", False),
"docker": _FakeBackend("docker", True), "docker": _FakeBackend("docker", True),
}): }):
@@ -83,7 +83,7 @@ class TestGetBottleBackend(unittest.TestCase):
with patch.dict(os.environ, {}, clear=True), \ with patch.dict(os.environ, {}, clear=True), \
patch.object(backend_mod.FirecrackerBottleBackend, patch.object(backend_mod.FirecrackerBottleBackend,
"is_host_capable", classmethod(lambda cls: True)), \ "is_host_capable", classmethod(lambda cls: True)), \
patch.object(backend_mod, "_backends", { patch.object(backend_mod, "_BACKENDS", {
"macos-container": _FakeBackend("macos-container", False), "macos-container": _FakeBackend("macos-container", False),
"firecracker": _FakeBackend("firecracker", False), "firecracker": _FakeBackend("firecracker", False),
"docker": _FakeBackend("docker", True), "docker": _FakeBackend("docker", True),
@@ -133,7 +133,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
return self._items return self._items
with patch.object( with patch.object(
backend_mod, "_backends", backend_mod, "_BACKENDS",
{"docker": _FakeBackend([a]), "firecracker": _FakeBackend([b])}, {"docker": _FakeBackend([a]), "firecracker": _FakeBackend([b])},
): ):
self.assertEqual([a, b], enumerate_active_agents()) self.assertEqual([a, b], enumerate_active_agents())
@@ -167,7 +167,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
return self._items return self._items
with patch.object( with patch.object(
backend_mod, "_backends", backend_mod, "_BACKENDS",
{ {
"docker": _FakeBackend([newer, tie_b]), "docker": _FakeBackend([newer, tie_b]),
"firecracker": _FakeBackend([missing_metadata, tie_a]), "firecracker": _FakeBackend([missing_metadata, tie_a]),
@@ -187,7 +187,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
return [] return []
with patch.object( with patch.object(
backend_mod, "_backends", backend_mod, "_BACKENDS",
{"docker": _FakeBackend(), "firecracker": _FakeBackend()}, {"docker": _FakeBackend(), "firecracker": _FakeBackend()},
): ):
self.assertEqual([], enumerate_active_agents()) self.assertEqual([], enumerate_active_agents())
@@ -218,7 +218,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
return self._items return self._items
with patch.object( with patch.object(
backend_mod, "_backends", backend_mod, "_BACKENDS",
{ {
"docker": _FakeBackend([present], available=True), "docker": _FakeBackend([present], available=True),
"firecracker": _FakeBackend([hidden], available=False), "firecracker": _FakeBackend([hidden], available=False),
@@ -234,7 +234,7 @@ class TestHasBackend(unittest.TestCase):
return False return False
with patch.object( with patch.object(
backend_mod, "_backends", {"docker": _FakeBackend()}, backend_mod, "_BACKENDS", {"docker": _FakeBackend()},
): ):
from bot_bottle.backend import has_backend from bot_bottle.backend import has_backend
self.assertFalse(has_backend("docker")) self.assertFalse(has_backend("docker"))
-186
View File
@@ -1,186 +0,0 @@
"""Unit: host Claude auth extraction."""
from __future__ import annotations
import json
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle.contrib.claude.claude_auth import (
claude_auth_path,
claude_host_access_token,
)
from bot_bottle.log import Die
def _cred_json(access_token: str, **extra: object) -> str:
payload: dict[str, object] = {"claudeAiOauth": {"accessToken": access_token, **extra}}
return json.dumps(payload)
class TestClaudeHostAccessToken(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory(prefix="bb-claude-auth.")
self.home = Path(self.tmp.name)
self.cred_dir = self.home / ".claude"
self.cred_dir.mkdir()
self.auth_path = self.cred_dir / ".credentials.json"
def tearDown(self):
self.tmp.cleanup()
def _write(self, payload: dict) -> None: # type: ignore[no-untyped-def]
self.auth_path.write_text(json.dumps(payload))
def test_auth_path_uses_home_env(self):
self.assertEqual(
self.auth_path,
claude_auth_path({"HOME": str(self.home)}),
)
# --- file-based (Linux) ---
def test_file_returns_access_token(self):
key = "sk-ant-oat01-real-key" # gitleaks:allow
self._write({"claudeAiOauth": {"accessToken": key}})
out = claude_host_access_token({"HOME": str(self.home)})
self.assertEqual(key, out)
def test_file_missing_claude_ai_oauth_dies(self):
self._write({"hasCompletedOnboarding": True})
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(self.home)})
def test_file_missing_access_token_dies(self):
self._write({"claudeAiOauth": {"expiresAt": 2000000000000}})
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(self.home)})
def test_file_empty_access_token_dies(self):
self._write({"claudeAiOauth": {"accessToken": ""}})
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(self.home)})
def test_file_expired_token_dies(self):
# expiresAt is milliseconds; 1_000_000 ms is year 1970
self._write({
"claudeAiOauth": {"accessToken": "sk-ant-oat01-x", "expiresAt": 1_000_000}, # gitleaks:allow
})
with self.assertRaises(Die):
claude_host_access_token(
{"HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
def test_file_future_expiry_is_accepted(self):
key = "sk-ant-oat01-y" # gitleaks:allow
# 2_000_000_000_000 ms ≈ year 2033
self._write({
"claudeAiOauth": {"accessToken": key, "expiresAt": 2_000_000_000_000},
})
out = claude_host_access_token(
{"HOME": str(self.home)},
now=datetime(2026, 1, 1, tzinfo=timezone.utc),
)
self.assertEqual(key, out)
def test_file_absent_expiry_is_accepted(self):
key = "sk-ant-oat01-z" # gitleaks:allow
self._write({"claudeAiOauth": {"accessToken": key}})
out = claude_host_access_token({"HOME": str(self.home)})
self.assertEqual(key, out)
def test_file_non_json_dies(self):
self.auth_path.write_text("not json {{{")
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(self.home)})
def test_file_json_array_root_dies(self):
self.auth_path.write_text("[]")
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(self.home)})
def test_file_extra_fields_are_ignored(self):
key = "sk-ant-oat01-real" # gitleaks:allow
self._write({
"claudeAiOauth": {
"accessToken": key,
"refreshToken": "sk-ant-ort01-secret", # gitleaks:allow
"scopes": ["user:inference"],
"expiresAt": 2_000_000_000_000,
},
})
out = claude_host_access_token({"HOME": str(self.home)})
self.assertEqual(key, out)
# --- macOS Keychain fallback ---
def _home_without_creds(self) -> Path:
"""A home dir that has .claude/ but no .credentials.json."""
empty = self.home / "no-creds"
(empty / ".claude").mkdir(parents=True)
return empty
def _mock_keychain(self, stdout: str, returncode: int = 0) -> MagicMock:
mock = MagicMock()
mock.returncode = returncode
mock.stdout = stdout
return mock
def test_keychain_used_when_file_absent(self):
key = "sk-ant-oat01-keychain" # gitleaks:allow
home = self._home_without_creds()
with patch(
"bot_bottle.contrib.claude.claude_auth.subprocess.run",
return_value=self._mock_keychain(_cred_json(key)),
), patch(
"bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
):
out = claude_host_access_token({"HOME": str(home)})
self.assertEqual(key, out)
def test_keychain_failure_when_file_absent_dies(self):
home = self._home_without_creds()
with patch(
"bot_bottle.contrib.claude.claude_auth.subprocess.run",
return_value=self._mock_keychain("", returncode=44),
), patch(
"bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
):
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(home)})
def test_no_file_no_keychain_on_linux_dies(self):
home = self._home_without_creds()
with patch("bot_bottle.contrib.claude.claude_auth.sys.platform", "linux"):
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(home)})
def test_keychain_non_json_dies(self):
home = self._home_without_creds()
with patch(
"bot_bottle.contrib.claude.claude_auth.subprocess.run",
return_value=self._mock_keychain("not-json"),
), patch(
"bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
):
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(home)})
def test_keychain_security_not_found_dies(self):
home = self._home_without_creds()
with patch(
"bot_bottle.contrib.claude.claude_auth.subprocess.run",
side_effect=FileNotFoundError,
), patch(
"bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
):
with self.assertRaises(Die):
claude_host_access_token({"HOME": str(home)})
if __name__ == "__main__":
unittest.main()
+3 -3
View File
@@ -29,7 +29,7 @@ def _fail(stderr: str = "boom") -> subprocess.CompletedProcess: # type: ignore
class TestCommitContainer(unittest.TestCase): class TestCommitContainer(unittest.TestCase):
def test_runs_docker_commit(self): def test_runs_docker_commit(self):
with patch.object( with patch.object(
docker_mod, "run_docker", return_value=_ok(), docker_mod.subprocess, "run", return_value=_ok(),
) as run, patch.object(docker_mod, "info"): ) as run, patch.object(docker_mod, "info"):
docker_mod.commit_container( docker_mod.commit_container(
"bot-bottle-dev-abc12", "bot-bottle-dev-abc12",
@@ -47,7 +47,7 @@ class TestCommitContainer(unittest.TestCase):
def test_dies_on_docker_commit_failure(self): def test_dies_on_docker_commit_failure(self):
with patch.object( with patch.object(
docker_mod, "run_docker", return_value=_fail("No such container"), docker_mod.subprocess, "run", return_value=_fail("No such container"),
), patch.object( ), patch.object(
docker_mod, "die", side_effect=SystemExit("die"), docker_mod, "die", side_effect=SystemExit("die"),
) as die: ) as die:
@@ -58,7 +58,7 @@ class TestCommitContainer(unittest.TestCase):
def test_die_message_includes_image_tag(self): def test_die_message_includes_image_tag(self):
with patch.object( with patch.object(
docker_mod, "run_docker", return_value=_fail("boom"), docker_mod.subprocess, "run", return_value=_fail("boom"),
), patch.object( ), patch.object(
docker_mod, "die", side_effect=SystemExit("die"), docker_mod, "die", side_effect=SystemExit("die"),
) as die: ) as die:
+1 -9
View File
@@ -80,19 +80,11 @@ class TestAgentProviderHostCredentials(unittest.TestCase):
"forward_host_credentials": "yes", "forward_host_credentials": "yes",
}) })
def test_forward_host_credentials_allowed_for_claude(self): def test_forward_host_credentials_rejected_for_claude(self):
b = _provider_config_bottle({
"template": "claude",
"forward_host_credentials": True,
})
self.assertTrue(b.agent_provider.forward_host_credentials)
def test_forward_host_credentials_and_auth_token_rejected_together(self):
with self.assertRaises(ManifestError): with self.assertRaises(ManifestError):
_provider_config_bottle({ _provider_config_bottle({
"template": "claude", "template": "claude",
"forward_host_credentials": True, "forward_host_credentials": True,
"auth_token": "SOME_TOKEN",
}) })
def test_auth_token_defaults_empty(self): def test_auth_token_defaults_empty(self):
+2 -14
View File
@@ -86,22 +86,10 @@ class TestAgentProviderValidation(unittest.TestCase):
"b", {"forward_host_credentials": True, "template": "weird"} "b", {"forward_host_credentials": True, "template": "weird"}
) )
def test_forward_creds_pi_template_rejected(self) -> None: def test_forward_creds_non_codex_template(self) -> None:
with self.assertRaises(ManifestError): with self.assertRaises(ManifestError):
ManifestAgentProvider.from_dict( ManifestAgentProvider.from_dict(
"b", {"forward_host_credentials": True, "template": "pi"} "b", {"forward_host_credentials": True, "template": "claude"}
)
def test_forward_creds_claude_allowed(self) -> None:
p = ManifestAgentProvider.from_dict(
"b", {"forward_host_credentials": True, "template": "claude"}
)
self.assertTrue(p.forward_host_credentials)
def test_forward_creds_and_auth_token_rejected(self) -> None:
with self.assertRaises(ManifestError):
ManifestAgentProvider.from_dict(
"b", {"forward_host_credentials": True, "auth_token": "T", "template": "claude"}
) )
def test_valid_claude_auth_token(self) -> None: def test_valid_claude_auth_token(self) -> None: