didericis/bot-bottle
1
0
Fork 0
Code Issues 7 Pull Requests 15 Actions Packages Projects Releases Wiki Activity

Merge pull request 'PRD 0009: Remove ssh-gate and bottle.ssh' (#13) from deprecate-ssh-gate into main
test / unit (push) Successful in 16s
Details
test / integration (push) Successful in 23s
Details

Browse Source
This commit was merged in pull request #13.
This commit is contained in:
didericis
2026-05-13 00:00:59 -04:00
parent 00649d27e9 30d92bef48
commit 3d9103d5b5
23 changed files with 280 additions and 962 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+7 -19
View File
@@ -78,13 +78,7 @@ that enforces the manifest before it leaves the host.
│ │ built locally) │ │ (TLS bump, DLP,│ │ hosts
│ │ │ │ allowlist) │ │
│ │ skills, env, │ └────────────────┘ │
│ │ ~/.ssh/config, │ │
│ │ ~/.gitconfig │ ssh ┌────────────────┐ │ TCP to
│ │ │ ───────────────► │ socat/ssh image│──┼──► bottle.ssh
│ │ │ │ (alpine/socat, │ │ upstreams
│ │ │ │ L4 forwarder) │ │
│ │ │ └────────────────┘ │
│ │ │ │
│ │ ~/.gitconfig │ │
│ │ │ git ops ┌────────────────┐ │ SSH (push/
│ │ │ ───────────────► │ git-gate image │──┼──► fetch) to
│ │ │ │ (gitleaks + │ │ bottle.git
@@ -98,16 +92,12 @@ that enforces the manifest before it leaves the host.
- **agent image** — built from the repo `Dockerfile` (`node:22-slim`
base) on first run; runs `claude` with the manifest-granted skills,
env vars, `~/.ssh/config`, and `~/.gitconfig` (the latter for the
git-gate's `pushInsteadOf` rules when `bottle.git` is set).
env vars, and `~/.gitconfig` (the latter for the git-gate's
`insteadOf` rules when `bottle.git` is set).
- **pipelock image** — per-agent sidecar. Terminates the agent's
outbound HTTP/HTTPS, enforces the resolved allowlist, runs DLP
scanning. Design in `docs/prds/0001-per-agent-egress-proxy-via-pipelock.md`
and `docs/prds/0006-pipelock-tls-interception.md`.
- **socat/ssh image** — per-agent sidecar built on `alpine/socat`.
One container, one socat listener per `bottle.ssh` entry, each
forwarding TCP to the upstream `Hostname:Port`. SSH does *not* go
through pipelock. Design in `docs/prds/0007-ssh-egress-gate.md`.
- **git-gate image** — per-agent sidecar built on `zricethezav/gitleaks`
(alpine + gitleaks + git-daemon + openssh-client). Runs
`git daemon` over `git://` as a bidirectional mirror of each
@@ -160,14 +150,12 @@ project entries overriding home entries on key conflict).
"GIT_AUTHOR_NAME": "didericis"
},
"ssh": [
"git": [
{
"Host": "gitea",
"Hostname": "gitea.dideric.is",
"User": "git",
"Port": 30009,
"Name": "claude-bottle",
"Upstream": "ssh://git@gitea.dideric.is:30009/didericis/claude-bottle.git",
"IdentityFile": "/Users/didericis/.ssh/id_ed25519_gitea",
"KnownHostKey": "gitea.dideric.is ssh-ed25519 AAAA..."
"KnownHostKey": "ssh-ed25519 AAAA..."
}
],
claude-bottle.example.json
-1
View File
@@ -2,7 +2,6 @@
"bottles": {
"default": {
"env": {},
"ssh": [],
"egress": {
"allowlist": [
"github.com",
claude_bottle/backend/__init__.py
+13 -30
View File
@@ -37,7 +37,7 @@ from pathlib import Path
from typing import Any, Generic, Sequence, TypeVar
from ..log import die
from ..manifest import GitEntry, Manifest, SshEntry
from ..manifest import GitEntry, Manifest
from ..util import expand_tilde
from .util import host_skill_dir
@@ -162,7 +162,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
def _validate(self, spec: BottleSpec) -> None:
"""Cross-backend pre-launch checks. Confirms the agent exists,
the named skills are present on the host, and every SSH
the named skills are present on the host, and every git
IdentityFile resolves. Subclasses with additional preconditions
should override and call `super()._validate(spec)` first."""
manifest = spec.manifest
@@ -170,7 +170,6 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
agent = manifest.agents[spec.agent_name]
bottle = manifest.bottle_for(spec.agent_name)
self._validate_skills(agent.skills)
self._validate_ssh_entries(bottle.ssh)
self._validate_git_entries(bottle.git)
def _validate_skills(self, skills: Sequence[str]) -> None:
@@ -185,15 +184,6 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
f"Create it under ~/.claude/skills/, then re-run."
)
def _validate_ssh_entries(self, entries: Sequence[SshEntry]) -> None:
"""Each entry's IdentityFile must exist on the host (after
expanding leading ~). Shape is already enforced by Manifest
validation; this only checks file presence."""
for entry in entries:
key = expand_tilde(entry.IdentityFile)
if not os.path.isfile(key):
die(f"ssh key file not found for host '{entry.Host}': {key}")
def _validate_git_entries(self, entries: Sequence[GitEntry]) -> None:
"""Each entry's IdentityFile must exist on the host (after
expanding leading ~) — the git-gate copies it in at start time
@@ -215,24 +205,23 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
"""Build/run the bottle and yield a handle; tear down on exit."""
def provision(self, plan: PlanT, target: str) -> str | None:
"""Copy host-side files (CA cert, prompt, skills, SSH keys,
.git) into the running bottle. Called from `launch` after the
container/machine is up. `target` identifies the running
instance in backend-specific terms (Docker: resolved
container name; fly: machine id). Returns the in-container
prompt path if a prompt was provisioned, else None — the
Bottle handle uses it to decide whether to add
--append-system-prompt-file to claude's argv.
"""Copy host-side files (CA cert, prompt, skills, .git) into
the running bottle. Called from `launch` after the container
/ machine is up. `target` identifies the running instance in
backend-specific terms (Docker: resolved container name; fly:
machine id). Returns the in-container prompt path if a prompt
was provisioned, else None — the Bottle handle uses it to
decide whether to add --append-system-prompt-file to claude's
argv.
Default orchestration: ca → prompt → skills → ssh → git.
CA install runs first so the agent's trust store is rebuilt
before anything inside the agent makes a TLS call. Subclasses
Default orchestration: ca → prompt → skills → git. CA install
runs first so the agent's trust store is rebuilt before
anything inside the agent makes a TLS call. Subclasses
typically don't override this; they implement the sub-methods
below."""
self.provision_ca(plan, target)
prompt_path = self.provision_prompt(plan, target)
self.provision_skills(plan, target)
self.provision_ssh(plan, target)
self.provision_git(plan, target)
return prompt_path
@@ -257,12 +246,6 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
"""Copy the agent's named skills from the host into the
running bottle. No-op when the agent has no skills."""
@abstractmethod
def provision_ssh(self, plan: PlanT, target: str) -> None:
"""Set up SSH in the running bottle (config, agent, keys)
so the bottle can reach the manifest's declared SSH hosts.
No-op when the bottle has no SSH entries."""
@abstractmethod
def provision_git(self, plan: PlanT, target: str) -> None:
"""Copy the host's cwd `.git` directory into the running
claude_bottle/backend/docker/backend.py
-8
View File
@@ -29,8 +29,6 @@ from .provision import ca as _ca
from .provision import git as _git
from .provision import prompt as _prompt
from .provision import skills as _skills
from .provision import ssh as _ssh
from .ssh_gate import DockerSSHGate
class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanupPlan"]):
@@ -41,7 +39,6 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
def __init__(self) -> None:
self._proxy = DockerPipelockProxy()
self._gate = DockerSSHGate()
self._git_gate = DockerGitGate()
def _resolve_plan(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan:
@@ -49,7 +46,6 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
spec,
stage_dir=stage_dir,
proxy=self._proxy,
gate=self._gate,
git_gate=self._git_gate,
)
@@ -58,7 +54,6 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
with _launch.launch(
plan,
proxy=self._proxy,
gate=self._gate,
git_gate=self._git_gate,
provision=self.provision,
) as bottle:
@@ -73,9 +68,6 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
def provision_skills(self, plan: DockerBottlePlan, target: str) -> None:
_skills.provision_skills(plan, target)
def provision_ssh(self, plan: DockerBottlePlan, target: str) -> None:
_ssh.provision_ssh(plan, target)
def provision_git(self, plan: DockerBottlePlan, target: str) -> None:
_git.provision_git(plan, target)
claude_bottle/backend/docker/bottle_plan.py
-23
View File
@@ -15,7 +15,6 @@ from ...git_gate import GitGatePlan
from ...log import info
from ...manifest import Agent, Bottle
from ...pipelock import PipelockProxyPlan, pipelock_effective_allowlist
from ...ssh_gate import SSHGatePlan
from .. import BottlePlan
@@ -27,7 +26,6 @@ class _PlanView:
agent: Agent
bottle: Bottle
env_names: list[str]
ssh_hosts: list[str]
git_names: list[str]
prompt_first_line: str
@@ -52,7 +50,6 @@ class DockerBottlePlan(BottlePlan):
forwarded_env: dict[str, str] = field(repr=False)
prompt_file: Path
proxy_plan: PipelockProxyPlan
gate_plan: SSHGatePlan
git_gate_plan: GitGatePlan
allowlist_summary: str
use_runsc: bool
@@ -69,7 +66,6 @@ class DockerBottlePlan(BottlePlan):
agent=agent,
bottle=bottle,
env_names=env_names,
ssh_hosts=[e.Host for e in bottle.ssh],
git_names=[e.Name for e in bottle.git],
prompt_first_line=agent.prompt.splitlines()[0] if agent.prompt else "",
)
@@ -94,16 +90,6 @@ class DockerBottlePlan(BottlePlan):
info("skills : " + (" ".join(v.agent.skills) if v.agent.skills else "(none)"))
info(f"docker runtime : {runtime_label}")
info(f"bottle : {v.agent.bottle}")
if v.ssh_hosts:
info(f" ssh hosts : {', '.join(v.ssh_hosts)}")
gate_lines = [
f"{u.bottle_host_alias} -> {u.upstream_host}:{u.upstream_port} "
f"(listen {u.listen_port})"
for u in self.gate_plan.upstreams
]
info(f" ssh gate : {'; '.join(gate_lines)}")
else:
info(" ssh hosts : (none)")
if v.git_names:
info(f" git remotes : {', '.join(v.git_names)}")
git_lines = [
@@ -136,15 +122,6 @@ class DockerBottlePlan(BottlePlan):
"runtime": "runsc" if self.use_runsc else "runc",
"env_names": v.env_names,
"skills": list(v.agent.skills),
"ssh_hosts": v.ssh_hosts,
"ssh_gate": [
{
"host": u.bottle_host_alias,
"upstream": f"{u.upstream_host}:{u.upstream_port}",
"listen_port": u.listen_port,
}
for u in self.gate_plan.upstreams
],
"git_remotes": v.git_names,
"git_gate": [
{
claude_bottle/backend/docker/launch.py
-17
View File
@@ -25,7 +25,6 @@ from .bottle_plan import DockerBottlePlan
from .git_gate import DockerGitGate
from .pipelock import DockerPipelockProxy, pipelock_proxy_url, pipelock_tls_init
from .provision.ca import AGENT_CA_BUNDLE, AGENT_CA_PATH
from .ssh_gate import DockerSSHGate
# Where the repo root lives, for `docker build` context. Computed once.
@@ -37,7 +36,6 @@ def launch(
plan: DockerBottlePlan,
*,
proxy: DockerPipelockProxy,
gate: DockerSSHGate,
git_gate: DockerGitGate,
provision: Callable[[DockerBottlePlan, str], str | None],
) -> Generator[DockerBottle, None, None]:
@@ -89,21 +87,6 @@ def launch(
pipelock_name = proxy.start(plan.proxy_plan)
stack.callback(proxy.stop, pipelock_name)
# SSH egress gate (PRD 0007). One sidecar per agent, only
# brought up when the bottle has ssh entries. Lives on the
# same internal + egress networks pipelock straddles; the
# agent dials it by container name (DNS works on --internal,
# confirmed by the PRD 0007 spike).
if plan.gate_plan.upstreams:
gate_plan = dataclasses.replace(
plan.gate_plan,
internal_network=internal_network,
egress_network=egress_network,
)
plan = dataclasses.replace(plan, gate_plan=gate_plan)
gate_name = gate.start(plan.gate_plan)
stack.callback(gate.stop, gate_name)
# Git gate (PRD 0008). One sidecar per agent, only brought up
# when the bottle has git entries. Same internal + egress
# network attachment as the other sidecars; agent dials it as
claude_bottle/backend/docker/prepare.py
+2 -6
View File
@@ -21,7 +21,6 @@ from . import util as docker_mod
from .bottle_plan import DockerBottlePlan
from .git_gate import DockerGitGate
from .pipelock import DockerPipelockProxy
from .ssh_gate import DockerSSHGate
def resolve_plan(
@@ -29,12 +28,11 @@ def resolve_plan(
*,
stage_dir: Path,
proxy: DockerPipelockProxy,
gate: DockerSSHGate,
git_gate: DockerGitGate,
) -> DockerBottlePlan:
"""Resolve Docker-specific names and write scratch files. Trusts
that the agent and its skills/SSH keys are present — validation
already ran in the base class."""
that the agent and its skills/git-gate keys are present —
validation already ran in the base class."""
docker_mod.require_docker()
manifest = spec.manifest
@@ -82,7 +80,6 @@ def resolve_plan(
prompt_file.chmod(0o600)
proxy_plan = proxy.prepare(bottle, slug, stage_dir)
gate_plan = gate.prepare(bottle, slug, stage_dir)
git_gate_plan = git_gate.prepare(bottle, slug, stage_dir)
resolved = resolve_env(manifest, spec.agent_name)
# Everything that should reach the bottle by-name (so its value
@@ -111,7 +108,6 @@ def resolve_plan(
forwarded_env=forwarded_env,
prompt_file=prompt_file,
proxy_plan=proxy_plan,
gate_plan=gate_plan,
git_gate_plan=git_gate_plan,
allowlist_summary=allowlist_summary,
use_runsc=use_runsc,
claude_bottle/backend/docker/provision/ssh.py
-199
View File
@@ -1,199 +0,0 @@
"""Set up SSH inside a running Docker bottle.
This is the most involved provisioner. The end state in the container:
- ~/.ssh/config + ~/.ssh/known_hosts owned by node, mode 600
- ssh-agent running as root with each key loaded; agent socket at
/run/claude-bottle-agent.sock
- socat forwarder (also root) bridging the agent socket to
/run/claude-bottle-agent-public.sock (mode 666) so node can talk
to the agent despite ssh-agent's SO_PEERCRED UID match
- on-disk key files deleted after `ssh-add`; the bytes only live in
the agent process's memory thereafter
See the `provision_ssh` docstring for the full isolation rationale."""
from __future__ import annotations
import os
import subprocess
from ....log import die, info
from ....util import expand_tilde
from .. import util as docker_mod
from ..bottle_plan import DockerBottlePlan
from ..ssh_gate import ssh_gate_host
def provision_ssh(plan: DockerBottlePlan, target: str) -> None:
"""Set up SSH in the container so node can authenticate using
each entry's key without the key file being readable by node.
No-op when the bottle has no SSH entries.
Isolation strategy:
- Keys live at /root/.claude-bottle-keys/ (mode 700,
root-owned). /root is mode 700 in node:22-slim, so node
(uid 1000) can't even traverse in.
- ssh-agent runs as root, listening on
/run/claude-bottle-agent.sock. Each key is loaded with
ssh-add, then deleted; the bytes now live only in the
agent process's memory.
- ssh-agent's SO_PEERCRED-based UID match rejects every
connection whose peer euid is neither 0 nor the agent's.
To bridge that, a root-owned socat forwarder listens on
/run/claude-bottle-agent-public.sock (mode 666) and
proxies bytes to the real agent socket.
- node can't ptrace root-owned agent or socat, so
/proc/<pid>/mem is off-limits and key bytes never leave
root-owned memory.
- ~/.ssh/config in node's home points each Host at the
public socket via IdentityAgent.
Why an in-container agent (not bind-mounted from host):
Docker Desktop on macOS does not forward Unix-domain socket
connect() across the VM boundary — connect() returns
ENOTSUP. Running ssh-agent inside the container sidesteps
that entirely.
Limitation: keys must be passphrase-less. ssh-add prompts on
/dev/tty for passphrases, but our docker exec has no TTY."""
bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
if not bottle.ssh:
return
container = target
gate_target = ssh_gate_host(plan.slug)
container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
container_ssh = f"{container_home}/.ssh"
agent_socket = "/run/claude-bottle-agent.sock"
public_socket = "/run/claude-bottle-agent-public.sock"
keys_dir = "/root/.claude-bottle-keys"
# Per-entry listen ports come off the gate plan (PRD 0007).
# Indexed by the bottle.ssh entry's Host alias so each ssh_config
# block knows which port its forwarder lives on.
upstreams_by_alias = {u.bottle_host_alias: u for u in plan.gate_plan.upstreams}
if set(upstreams_by_alias) != {e.Host for e in bottle.ssh}:
die(
"ssh-gate upstream table is out of sync with bottle.ssh; "
"this is an internal bug"
)
# ~/.ssh for node (700, owned by node).
docker_mod.docker_exec_root(container, ["mkdir", "-p", container_ssh])
docker_mod.docker_exec_root(container, ["chown", "node:node", container_ssh])
docker_mod.docker_exec_root(container, ["chmod", "700", container_ssh])
# /root/.claude-bottle-keys for root (700, root-owned).
docker_mod.docker_exec_root(container, ["mkdir", "-p", keys_dir])
docker_mod.docker_exec_root(container, ["chown", "root:root", keys_dir])
docker_mod.docker_exec_root(container, ["chmod", "700", keys_dir])
config_file = plan.stage_dir / "ssh_config"
known_hosts_file = plan.stage_dir / "ssh_known_hosts"
config_file.write_text("")
config_file.chmod(0o600)
known_hosts_file.write_text("")
known_hosts_file.chmod(0o600)
container_key_paths: list[str] = []
for entry in bottle.ssh:
name = entry.Host
key = expand_tilde(entry.IdentityFile)
hostname = entry.Hostname
user = entry.User
known_host_key = entry.KnownHostKey
upstream = upstreams_by_alias[name]
listen_port = upstream.listen_port
key_basename = os.path.basename(key)
container_key_path = f"{keys_dir}/{key_basename}"
info(f"copying ssh key for '{name}' -> {container} (root-only staging)")
subprocess.run(
["docker", "cp", key, f"{container}:{container_key_path}"],
stdout=subprocess.DEVNULL,
check=True,
)
docker_mod.docker_exec_root(container, ["chown", "root:root", container_key_path])
docker_mod.docker_exec_root(container, ["chmod", "600", container_key_path])
container_key_paths.append(container_key_path)
# Each Host block points at the gate container + its
# per-entry listen port. HostKeyAlias makes ssh validate
# the host key against `hostname` (the real upstream
# name) instead of the gate container; CheckHostIP=no
# skips the resolved-IP lookup, which would also point at
# the gate.
block = (
f"Host {name}\n"
f" HostName {gate_target}\n"
f" User {user}\n"
f" Port {listen_port}\n"
f" IdentityAgent {public_socket}\n"
f" HostKeyAlias {hostname}\n"
f" CheckHostIP no\n"
f"\n"
)
with config_file.open("a") as f:
f.write(block)
if known_host_key:
# HostKeyAlias makes ssh look up known_hosts under
# `hostname` (the upstream's real name / IP literal),
# not the gate container. One unambiguous entry per
# ssh entry.
with known_hosts_file.open("a") as f:
f.write(f"{hostname} {known_host_key}\n")
# Boot the agent, load each key, delete the key files, then
# start the root-owned socat forwarder. One docker exec so the
# whole sequence is atomic.
info(f"starting in-container ssh-agent at {agent_socket} (forwarded via {public_socket})")
setup_lines = [
"set -eu",
f"ssh-agent -a {agent_socket} >/dev/null",
]
for kp in container_key_paths:
setup_lines.append(f"SSH_AUTH_SOCK={agent_socket} ssh-add {kp}")
setup_lines.append(f"rm -f {kp}")
setup_lines.append(f"rmdir {keys_dir} 2>/dev/null || true")
# Forwarder: socat (uid 0) connects to the agent on node's behalf.
setup_lines.append(
f"nohup socat UNIX-LISTEN:{public_socket},fork,reuseaddr,mode=666 "
f"UNIX-CONNECT:{agent_socket} </dev/null >/dev/null 2>&1 &"
)
# Wait briefly for the forwarder to bind.
setup_lines.extend([
"i=0",
"while [ $i -lt 20 ]; do",
f" [ -S {public_socket} ] && break",
" i=$((i + 1))",
" sleep 0.1",
"done",
f"[ -S {public_socket} ] || {{ echo 'claude-bottle: socat forwarder failed to bind {public_socket}' >&2; exit 1; }}",
])
setup_script = "\n".join(setup_lines) + "\n"
subprocess.run(
["docker", "exec", "-u", "0", container, "sh", "-c", setup_script],
check=True,
)
info(f"writing {container_ssh}/config")
subprocess.run(
["docker", "cp", str(config_file), f"{container}:{container_ssh}/config"],
stdout=subprocess.DEVNULL,
check=True,
)
docker_mod.docker_exec_root(container, ["chown", "node:node", f"{container_ssh}/config"])
docker_mod.docker_exec_root(container, ["chmod", "600", f"{container_ssh}/config"])
if known_hosts_file.stat().st_size > 0:
info(f"writing {container_ssh}/known_hosts")
subprocess.run(
["docker", "cp", str(known_hosts_file), f"{container}:{container_ssh}/known_hosts"],
stdout=subprocess.DEVNULL,
check=True,
)
docker_mod.docker_exec_root(container, ["chown", "node:node", f"{container_ssh}/known_hosts"])
docker_mod.docker_exec_root(container, ["chmod", "600", f"{container_ssh}/known_hosts"])
claude_bottle/backend/docker/ssh_gate.py
-159
View File
@@ -1,159 +0,0 @@
"""DockerSSHGate — the Docker-specific lifecycle for the per-agent
SSH egress gate sidecar (PRD 0007). Inherits the platform-agnostic
prepare step (upstream allocation + entrypoint render) from
`SSHGate`."""
from __future__ import annotations
import os
import subprocess
from ...log import die, info, warn
from ...ssh_gate import SSHGate, SSHGatePlan
# alpine/socat pinned by digest. The image is `alpine` + `socat`
# pre-installed; PRD 0007 requires the gate image to be
# self-sufficient at boot (no apk pulls) because the agent-facing
# leg sits on the `--internal` network.
SSH_GATE_IMAGE = os.environ.get(
"CLAUDE_BOTTLE_SSH_GATE_IMAGE",
"alpine/socat@sha256:a26f4bcee25ad4a4096ce91e596c0a2fffcbb51f7fd198dd87a5c86eae66f0e1",
)
# In-container path the entrypoint script lands at after `docker cp`.
# Root path keeps the cp simple — no intermediate directories to
# create.
SSH_GATE_ENTRYPOINT_IN_CONTAINER = "/ssh-gate-entrypoint.sh"
def ssh_gate_container_name(slug: str) -> str:
return f"claude-bottle-ssh-gate-{slug}"
def ssh_gate_host(slug: str) -> str:
"""The hostname the agent's ssh client should connect to. Same as
the container name — Docker's embedded DNS resolves it on the
`--internal` network (verified by the PRD 0007 DNS spike)."""
return ssh_gate_container_name(slug)
class DockerSSHGate(SSHGate):
"""Brings the SSH gate sidecar up and down via Docker."""
def start(self, plan: SSHGatePlan) -> str:
"""Boot the gate sidecar:
1. `docker create` on the internal network with the
canonical name, `--entrypoint /bin/sh`, and the
in-container entrypoint path as the CMD.
2. `docker cp` the entrypoint script in.
3. Attach to the per-agent egress network so socat can dial
upstream.
4. `docker start`.
Returns the container name (the target passed to `.stop`)."""
if not plan.upstreams:
die("DockerSSHGate.start called with no upstreams; caller should skip")
if not plan.internal_network or not plan.egress_network:
die(
"DockerSSHGate.start: internal_network / egress_network must be "
"populated on the plan before start"
)
if not plan.entrypoint_script.is_file():
die(
f"ssh-gate entrypoint script missing at {plan.entrypoint_script}; "
f"SSHGate.prepare must run first"
)
name = ssh_gate_container_name(plan.slug)
info(f"starting ssh-gate sidecar {name} on network {plan.internal_network}")
create_args = [
"docker", "create",
"--name", name,
"--network", plan.internal_network,
"--entrypoint", "/bin/sh",
SSH_GATE_IMAGE,
SSH_GATE_ENTRYPOINT_IN_CONTAINER,
]
if subprocess.run(
create_args,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
die(f"failed to create ssh-gate sidecar {name}")
cp_result = subprocess.run(
[
"docker", "cp",
str(plan.entrypoint_script),
f"{name}:{SSH_GATE_ENTRYPOINT_IN_CONTAINER}",
],
capture_output=True,
text=True,
check=False,
)
if cp_result.returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to copy ssh-gate entrypoint into {name}: "
f"{cp_result.stderr.strip()}"
)
if subprocess.run(
["docker", "network", "connect", plan.egress_network, name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(
f"failed to attach ssh-gate sidecar {name} to egress network "
f"{plan.egress_network}"
)
if subprocess.run(
["docker", "start", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
subprocess.run(
["docker", "rm", "-f", name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
)
die(f"failed to start ssh-gate sidecar {name}")
return name
def stop(self, target: str) -> None:
"""Idempotent: missing container is success. `target` is the
container name returned by `.start`."""
if subprocess.run(
["docker", "inspect", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode == 0:
if subprocess.run(
["docker", "rm", "-f", target],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
check=False,
).returncode != 0:
warn(
f"failed to remove ssh-gate sidecar {target}; "
f"clean up with 'docker rm -f {target}'"
)
claude_bottle/cli/info.py
+5 -6
View File
@@ -31,16 +31,15 @@ def cmd_info(argv: list[str]) -> int:
f"first line: {prompt_first_line or '(empty)'}"
)
info(f"bottle : {agent.bottle}")
if bottle.ssh:
for e in bottle.ssh:
if bottle.git:
for e in bottle.git:
info(
f" ssh host : {e.Host} "
f"(Hostname={e.Hostname}, User={e.User}, "
f"Port={e.Port}, IdentityFile={e.IdentityFile})"
f" git remote : {e.Name} -> {e.Upstream} "
f"(IdentityFile={e.IdentityFile})"
)
if e.KnownHostKey:
info(f" KnownHostKey: {e.KnownHostKey}")
else:
info(" ssh hosts : (none)")
info(" git remotes : (none)")
print()
return 0
claude_bottle/git_gate.py
+2 -3
View File
@@ -93,9 +93,8 @@ class GitGatePlan:
def git_gate_upstreams_for_bottle(bottle: Bottle) -> tuple[GitGateUpstream, ...]:
"""Lift each `bottle.git` entry into a GitGateUpstream. Cross-entry
validation (unique Names, no shadow route with bottle.ssh) already
ran in `manifest.Bottle.from_dict`."""
"""Lift each `bottle.git` entry into a GitGateUpstream. Unique-Name
validation already ran in `manifest.Bottle.from_dict`."""
return tuple(
GitGateUpstream(
name=e.Name,
claude_bottle/manifest.py
+13 -92
View File
@@ -6,7 +6,6 @@ Schema (see CLAUDE.md "Intended design"):
"bottles": {
"<bottle-name>": {
"env": { "<NAME>": <env-entry>, ... },
"ssh": [ <ssh-entry>, ... ],
"git": [ <git-entry>, ... ],
"egress": { "allowlist": [ "<hostname>", ... ] }
}
@@ -20,8 +19,9 @@ Schema (see CLAUDE.md "Intended design"):
}
}
Bottles group shared infrastructure (SSH keys, known hosts, egress allowlist)
that multiple agents can reference. Every agent must reference a bottle.
Bottles group shared infrastructure (git upstreams + their gate credentials,
egress allowlist) that multiple agents can reference. Every agent must
reference a bottle.
Validation runs once at construction (Manifest.from_json_obj) so getters
can trust the shape.
@@ -42,44 +42,6 @@ def _empty_str_dict() -> dict[str, str]:
return {}
@dataclass(frozen=True)
class SshEntry:
Host: str
IdentityFile: str
Hostname: str = ""
User: str = ""
Port: str = ""
KnownHostKey: str = ""
@classmethod
def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "SshEntry":
d = _as_json_object(raw, f"bottle '{bottle_name}' ssh[{idx}]")
host = d.get("Host")
if not isinstance(host, str) or not host:
die(f"bottle '{bottle_name}' ssh[{idx}] missing required string field 'Host'")
ident = d.get("IdentityFile")
if not isinstance(ident, str) or not ident:
die(
f"bottle '{bottle_name}' ssh '{host}' missing required string field "
f"'IdentityFile'"
)
hostname = _opt_str(d.get("Hostname"), f"bottle '{bottle_name}' ssh '{host}' Hostname")
user = _opt_str(d.get("User"), f"bottle '{bottle_name}' ssh '{host}' User")
port = _opt_port(d.get("Port"), f"bottle '{bottle_name}' ssh '{host}' Port")
khk = _opt_str(
d.get("KnownHostKey"),
f"bottle '{bottle_name}' ssh '{host}' KnownHostKey",
)
return cls(
Host=host,
IdentityFile=ident,
Hostname=hostname,
User=user,
Port=port,
KnownHostKey=khk,
)
@dataclass(frozen=True)
class GitEntry:
"""One upstream the per-agent git-gate (PRD 0008) is allowed to
@@ -205,7 +167,6 @@ class BottleEgress:
@dataclass(frozen=True)
class Bottle:
env: Mapping[str, str] = field(default_factory=_empty_str_dict)
ssh: tuple[SshEntry, ...] = ()
git: tuple[GitEntry, ...] = ()
egress: BottleEgress = field(default_factory=BottleEgress)
@@ -221,6 +182,15 @@ class Bottle:
f"definition."
)
if "ssh" in d:
die(
f"bottle '{name}' has an 'ssh' field, which has been removed "
f"(PRD 0009). Move each entry to 'git': declare the upstream "
f"as a git remote with Name + Upstream URL + IdentityFile, "
f"and the per-bottle git-gate (PRD 0008) will hold the "
f"credential and gitleaks-scan pushes."
)
env: dict[str, str] = {}
env_raw = d.get("env")
if env_raw is not None:
@@ -233,17 +203,6 @@ class Bottle:
)
env[var] = value
ssh: tuple[SshEntry, ...] = ()
ssh_raw = d.get("ssh")
if ssh_raw is not None:
if not isinstance(ssh_raw, list):
die(f"bottle '{name}' ssh must be an array (was {type(ssh_raw).__name__})")
ssh_list = cast(list[object], ssh_raw)
ssh = tuple(
SshEntry.from_dict(name, i, entry)
for i, entry in enumerate(ssh_list)
)
git: tuple[GitEntry, ...] = ()
git_raw = d.get("git")
if git_raw is not None:
@@ -255,7 +214,6 @@ class Bottle:
for i, entry in enumerate(git_list)
)
_validate_unique_git_names(name, git)
_validate_no_shadow_route(name, ssh, git)
egress_raw = d.get("egress")
egress = (
@@ -264,7 +222,7 @@ class Bottle:
else BottleEgress()
)
return cls(env=env, ssh=ssh, git=git, egress=egress)
return cls(env=env, git=git, egress=egress)
@dataclass(frozen=True)
@@ -434,19 +392,6 @@ def _opt_str(value: object, label: str) -> str:
return value
def _opt_port(value: object, label: str) -> str:
"""Port accepts string or int (JSON-friendly) and is normalized to str."""
if value is None:
return ""
if isinstance(value, bool):
die(f"{label} must be a string or number (was boolean)")
if isinstance(value, int):
return str(value)
if isinstance(value, str):
return value
die(f"{label} must be a string or number (was {type(value).__name__})")
def _opt_extra_hosts(value: object, label: str) -> dict[str, str]:
"""Validate a `{hostname: ip}` object and return a plain dict. None
yields an empty dict so callers can treat ExtraHosts as always
@@ -507,27 +452,3 @@ def _validate_unique_git_names(bottle_name: str, git: tuple[GitEntry, ...]) -> N
seen[g.Name] = None
def _validate_no_shadow_route(
bottle_name: str,
ssh: tuple[SshEntry, ...],
git: tuple[GitEntry, ...],
) -> None:
"""Reject if any git entry's (host, port) matches an ssh entry's
(Hostname, Port). The same upstream reachable two ways — once through
the L4 ssh-gate, once through the gitleaks-bearing git-gate — defeats
the git-gate."""
ssh_targets: dict[tuple[str, str], str] = {}
for e in ssh:
if not e.Hostname:
continue
port = e.Port or "22"
ssh_targets[(e.Hostname, port)] = e.Host
for g in git:
ssh_host = ssh_targets.get((g.UpstreamHost, g.UpstreamPort))
if ssh_host is not None:
die(
f"bottle '{bottle_name}' has ssh entry '{ssh_host}' "
f"({g.UpstreamHost}:{g.UpstreamPort}) and git entry '{g.Name}' "
f"pointing at the same upstream. The same remote reachable two "
f"ways defeats the git-gate; remove one."
)
claude_bottle/pipelock.py
+3 -3
View File
@@ -57,9 +57,9 @@ def pipelock_bottle_allowlist(bottle: Bottle) -> list[str]:
def pipelock_effective_allowlist(bottle: Bottle) -> list[str]:
"""Deduplicated union of: baked-in defaults, bottle.egress.allowlist.
Sorted for stability. Per PRD 0007, bottle.ssh entries do NOT
contribute here — SSH traffic flows through the per-agent ssh-gate
sidecar, not pipelock."""
Sorted for stability. Git upstreams declared in `bottle.git` do NOT
contribute here — git traffic flows through the per-agent git-gate
sidecar (PRD 0008), not pipelock."""
seen: dict[str, None] = {}
for h in DEFAULT_ALLOWLIST:
seen.setdefault(h, None)
claude_bottle/ssh_gate.py
-144
View File
@@ -1,144 +0,0 @@
"""Per-agent SSH egress gate (PRD 0007).
A second per-agent sidecar that does plain TCP forwarding from a set
of static listen ports to the SSH hosts declared in `bottle.ssh`.
The agent's ssh client points each `Host` block at the gate
container + a per-entry listen port; pipelock stops seeing SSH
traffic entirely.
This module defines the abstract gate (`SSHGate`) and the plan
dataclass (`SSHGatePlan`) consumed by its `start`. The sidecar's
start/stop lifecycle is backend-specific and lives on concrete
subclasses (see `claude_bottle/backend/docker/ssh_gate.py`)."""
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from .log import die
from .manifest import Bottle
# Default port when an ssh entry has no `Port` field. Matches OpenSSH.
_DEFAULT_SSH_PORT = 22
@dataclass(frozen=True)
class SSHGateUpstream:
"""One forwarder rule on the gate: listen locally on `listen_port`,
forward each connection to `upstream_host:upstream_port`. The
`bottle_host_alias` is the `Host` value from the manifest entry,
kept for diagnostics + so the ssh provisioner can correlate
upstreams with their alias.
`listen_port` mirrors the upstream port. That choice lets git
URLs that bake the upstream port into the remote (e.g.
`ssh://git@host:30009/repo.git`) work without rewriting: OpenSSH
treats a URL-supplied port as overriding the config's `Port`
directive, so the gate must be reachable on the same port the URL
names. Two ssh entries that share an upstream port are a config
error and rejected at prepare time."""
listen_port: int
upstream_host: str
upstream_port: str
bottle_host_alias: str
@dataclass(frozen=True)
class SSHGatePlan:
"""Output of SSHGate.prepare; consumed by .start when the sidecar
needs to be brought up.
`upstreams` + `slug` + `entrypoint_script` are filled in at
prepare time (host-side, side-effect-free on docker). The network
fields are populated by the backend's launch step via
`dataclasses.replace` once those networks exist. Empty defaults
are sentinels meaning "not yet set"; `.start` validates that
they are populated."""
slug: str
entrypoint_script: Path
upstreams: tuple[SSHGateUpstream, ...]
internal_network: str = ""
egress_network: str = ""
def ssh_gate_upstreams_for_bottle(bottle: Bottle) -> tuple[SSHGateUpstream, ...]:
"""Build the gate's upstream table. Each ssh entry's listen port
equals its upstream port so URL-supplied ports (which override
`~/.ssh/config`'s `Port` directive) still reach the gate.
Dies on two entries sharing an upstream port — the gate is a
single container with a flat port space, so each listener has to
be unique."""
seen_ports: dict[int, str] = {}
upstreams: list[SSHGateUpstream] = []
for e in bottle.ssh:
port = int(e.Port) if e.Port else _DEFAULT_SSH_PORT
if port in seen_ports:
die(
f"ssh entries '{seen_ports[port]}' and '{e.Host}' share upstream port "
f"{port}; the per-agent ssh gate can only forward one upstream "
f"per port. Change one of the upstream Ports in claude-bottle.json."
)
seen_ports[port] = e.Host
upstreams.append(
SSHGateUpstream(
listen_port=port,
upstream_host=e.Hostname,
upstream_port=e.Port,
bottle_host_alias=e.Host,
)
)
return tuple(upstreams)
def ssh_gate_render_entrypoint(upstreams: tuple[SSHGateUpstream, ...]) -> str:
"""Render the gate's entrypoint script: one `socat TCP-LISTEN`
per upstream, all backgrounded, then `wait`. Posix sh, no bash-isms
(alpine's sh is busybox ash). If any one socat dies, the others
keep running until the container is removed — matches the v1
no-restart policy from the PRD."""
lines = ["#!/bin/sh", "set -eu"]
for u in upstreams:
lines.append(
f"socat TCP-LISTEN:{u.listen_port},reuseaddr,fork "
f"TCP:{u.upstream_host}:{u.upstream_port} &"
)
lines.append("wait")
return "\n".join(lines) + "\n"
class SSHGate(ABC):
"""The per-agent SSH egress gate. Encapsulates the host-side
prepare step (upstream allocation + entrypoint render); the
sidecar's start/stop lifecycle is backend-specific and lives on
concrete subclasses."""
def prepare(self, bottle: Bottle, slug: str, stage_dir: Path) -> SSHGatePlan:
"""Compute the upstream table from `bottle.ssh` and write the
entrypoint script (mode 600) under `stage_dir`. Pure host-side,
no docker subprocess.
Returned plan is incomplete: the launch step must fill
`internal_network` / `egress_network` via `dataclasses.replace`
before passing the plan to `.start`."""
upstreams = ssh_gate_upstreams_for_bottle(bottle)
script = stage_dir / "ssh_gate_entrypoint.sh"
script.write_text(ssh_gate_render_entrypoint(upstreams))
script.chmod(0o600)
return SSHGatePlan(slug=slug, entrypoint_script=script, upstreams=upstreams)
@abstractmethod
def start(self, plan: SSHGatePlan) -> str:
"""Bring up the gate sidecar according to `plan`. Returns the
target string identifying the running instance — the same
value to pass to `.stop`. Backend-specific."""
@abstractmethod
def stop(self, target: str) -> None:
"""Tear down the gate sidecar identified by `target` (the
value `.start` returned). Idempotent: a missing target is
success. Backend-specific."""
docs/prds/0007-ssh-egress-gate.md
+8 -1
View File
@@ -1,9 +1,16 @@
# PRD 0007: SSH egress gate
- **Status:** Draft
- **Status:** Superseded by PRD 0009 (2026-05-13)
- **Author:** didericis
- **Created:** 2026-05-12
> **Superseded.** The ssh-gate sidecar and `bottle.ssh` manifest field
> described below were removed in PRD 0009. Every upstream this PRD
> targeted has since been folded into PRD 0008's git-gate, which
> covers the same use case with credential isolation and gitleaks
> scanning instead of bare L4 forwarding. Kept in-tree for the
> history of intent.
## Summary
Per-agent TCP-forwarder sidecar built from `bottle.ssh` entries; SSH stops
docs/prds/0009-remove-ssh-gate.md
+206
View File
@@ -0,0 +1,206 @@
# PRD 0009: Remove ssh-gate and bottle.ssh
- **Status:** Draft
- **Author:** didericis
- **Created:** 2026-05-13
## Summary
Delete the ssh-gate sidecar and the `bottle.ssh` manifest field.
Git-gate (PRD 0008) covers every current SSH use case in
claude-bottle: each declared upstream gets a per-bottle gate
with gitleaks scanning, an `insteadOf` rewrite that captures
push / fetch / clone / pull / ls-remote, and credential
isolation from the agent. ssh-gate is now redundant L4
forwarding with no gating value, and the only remaining users
of `bottle.ssh` were git remotes that git-gate handles better.
## Problem
PRD 0007 introduced ssh-gate as an L4 SSH forwarder so an agent
could reach declared SSH upstreams without a default route
off-box. At the time that meant git. With PRD 0008 every git
upstream now flows through git-gate, which:
- Holds the upstream credential in the gate, not in the agent
- Runs gitleaks against every incoming ref
- Refreshes from upstream on every fetch (fail-closed when the
upstream is unreachable)
- Rewrites the agent's URLs via `insteadOf`, so push, fetch,
clone, pull, and ls-remote all route through the gate
ssh-gate does none of those — it's transport-only — and offers
no benefit over git-gate for git, which is the only upstream
class currently used. Carrying it means a dead manifest field,
a redundant sidecar lifecycle, a shadow-route validator between
`bottle.ssh` and `bottle.git`, and a third place to keep an SSH
identity in sync. The project's stated goal is minimum
credentials and minimum egress; the simpler answer is to drop
the unused path.
## Goals / Success Criteria
- `bottle.ssh` is no longer a valid manifest field. Manifests
carrying it parse-fail with an error message pointing at
`bottle.git`.
- `SSHGate` / `DockerSSHGate`, `provision_ssh`, and the socat
sidecar lifecycle are gone from the codebase.
- Unit + integration tests pass without the ssh-gate suites;
no test references `SshEntry` or `bottle.ssh`.
- The y/N preflight no longer mentions the ssh sidecar; the
README's architecture diagram drops the socat box; the
example manifest drops `ssh:`.
- PRD 0007 carries a "Superseded by PRD 0009" header so the
history of intent stays in the tree.
## Non-goals
- Removing pipelock or git-gate. Both keep their current roles.
- Removing the SSH egress *capability* in general — git-gate
uses SSH inside the gate to reach the real upstream.
- Building a generic L4 egress proxy as a replacement. If
non-git SSH ever returns, design a fresh gate for the use
case; don't resurrect ssh-gate's L4-only design.
- Preserving a "ssh: []" compatibility shim. If the field is
dead it should error, not silently parse.
- Auto-migrating user manifests. The replacement (`bottle.git`)
needs a `Name` and full upstream URL that `bottle.ssh` does
not carry; a hard error with a migration hint is cleaner than
a partial rewrite.
## Scope
### In scope
- **Manifest.** Delete `SshEntry`, `Bottle.ssh`, and
`_validate_no_shadow_route`. Add an explicit branch in
`Bottle.from_dict` that dies on a `ssh` key with a one-line
"move this to `bottle.git` (see PRD 0008)" hint.
- **Sidecar.** Delete `claude_bottle/ssh_gate.py` and
`claude_bottle/backend/docker/ssh_gate.py`. Drop the socat
image build path.
- **Provisioner.** Delete
`claude_bottle/backend/docker/provision/ssh.py` and its
`~/.ssh/config` render.
- **Docker backend wiring.** Drop `DockerSSHGate` from
`backend.py`; drop its start / stop from `launch.py`'s
`ExitStack`; drop the plan field from `bottle_plan.py`.
- **Pipelock interaction.** Drop the SSH-derived branch from
pipelock's `ssrf.ip_allowlist` build. With no `bottle.ssh`
there is no per-upstream IP carve-out to render; git-gate
has its own egress network and pulls in upstream resolution
via `ExtraHosts` plus DNS.
- **Tests.** Delete the ssh-gate unit + integration suites,
the ssh fixtures in `tests/fixtures.py`, and the
shadow-route assertions in `test_manifest_git.py`. Adjust
any tests that asserted on a "git + ssh" combined manifest
to be git-only.
- **README.** Drop the socat / ssh image box from the
architecture diagram and its bullet; drop `ssh:` from the
manifest example.
- **Example manifest.** Drop `ssh:` from `claude-bottle.example.json`.
- **PRD 0007.** Add a `Status: Superseded by PRD 0009` header
at the top of the document. Do not delete the file; the
history of intent matters for the audit trail.
- **Migration note.** A short paragraph in the README explaining
the swap (drop the `ssh` entry, add a `git` entry pointing
at the same upstream with a `Name` and full URL).
### Out of scope
- A generic non-git SSH gate. If/when the use case returns,
design a fresh sidecar with its own threat model.
- Migrating user manifests automatically. Parse-fail with a
clear error is the path.
- Reworking git-gate to absorb non-git protocols. git-gate
remains git-only.
## Proposed Design
Most of the work is deletion. The substantive decisions are at
the seams between ssh-gate and the rest of the system:
1. **Manifest parse-error shape.** `Bottle.from_dict` adds an
explicit `if "ssh" in d:` branch that dies with a hint
naming `bottle.git` and PRD 0008. Better than silent ignore
— the migration is visible and one-shot — and better than a
warning, which agents wouldn't see anyway.
2. **Pipelock allowlist build.** Today pipelock pulls SSH
upstream IPs into its `ssrf.ip_allowlist` so ssh-gate can
dial out. With ssh-gate gone, that branch goes away; verify
nothing else in the pipelock render touches `bottle.ssh`.
3. **Plan rendering.** `bottle_plan.py` and the y/N preflight
currently list the socat sidecar. Remove the field and any
rendering branch; the dry-run output simplifies.
4. **PRD 0007 marker.** A header line at the top, not a delete.
The PRD's rationale (why ssh-gate was built) is still
useful context when reading PRD 0008 and PRD 0009.
### Existing code touched
- `claude_bottle/manifest.py` — delete `SshEntry`,
`Bottle.ssh`, `_validate_no_shadow_route`; add the
parse-fail branch.
- `claude_bottle/ssh_gate.py` — delete.
- `claude_bottle/backend/docker/ssh_gate.py` — delete.
- `claude_bottle/backend/docker/provision/ssh.py` — delete.
- `claude_bottle/backend/docker/backend.py` — drop
`DockerSSHGate` instantiation.
- `claude_bottle/backend/docker/launch.py` — drop the
ssh-gate start / stop from the `ExitStack`.
- `claude_bottle/backend/docker/bottle_plan.py` — drop the
ssh-gate plan field.
- `claude_bottle/pipelock.py` — drop the `bottle.ssh`-derived
branch in the allowlist render.
- `tests/unit/test_ssh_gate.py` — delete.
- `tests/integration/` — delete any ssh-gate-specific tests.
- `tests/unit/test_manifest_git.py` — drop the shadow-route
assertions.
- `tests/fixtures.py` — drop `fixture_with_ssh` and its dict
helper.
- `README.md` — drop the socat image box from the diagram and
the matching bullet; drop `ssh:` from the manifest example.
- `claude-bottle.example.json` — drop the `ssh` field.
- `docs/prds/0007-ssh-egress-gate.md` — add a
`Status: Superseded by PRD 0009` header at the top.
### Data model changes
- `SshEntry` removed.
- `Bottle.ssh: tuple[SshEntry, ...] = ()` removed.
- `_validate_no_shadow_route` removed.
### External dependencies
Nothing added. The `alpine/socat` image is no longer pulled
by claude-bottle; the cleanup of any existing local image is
the user's choice (a single `docker image rm` if they care).
## Future work
If a non-git SSH use case returns (deployment, rsync, remote
management), build it as its own gate with its own threat
model. Don't resurrect ssh-gate's L4-only design. The
git-gate pattern (gate holds credentials, agent gets a
rewritten URL, gate makes the upstream connection) is the
template.
## Open questions
- **Error vs migrate on legacy `bottle.ssh`.** Default: hard
error with a one-line "move this to `bottle.git`" hint.
Migrating in-place is brittle — the entry has no `Name`
field and no upstream URL, so we'd be inventing both.
- **One PR or two.** The removal touches enough files that a
single big PR may be hard to review; splitting into
"deprecate manifest field" and "remove sidecar code" reads
cleaner, but the field can't be removed cleanly while the
sidecar still references it. Default: single PR, deletion
commit per layer (manifest, sidecar, provisioner, tests,
docs).
## References
- PRD 0007: SSH egress gate — the design being superseded.
- PRD 0008: Git gate — the design that subsumes ssh-gate's
only current use case.
tests/fixtures.py
-32
View File
@@ -37,34 +37,6 @@ def fixture_with_egress_dict() -> dict[str, Any]:
}
def fixture_with_ssh_dict() -> dict[str, Any]:
"""Bottle has both an IPv4-literal SSH host (CGNAT) and a hostname host,
exercising both ssrf.ip_allowlist and trusted_domains code paths. JSON shape."""
return {
"bottles": {
"dev": {
"ssh": [
{
"Host": "tailscale-gitea",
"IdentityFile": "/dev/null",
"Hostname": "100.78.141.42",
"User": "git",
"Port": 30009,
},
{
"Host": "github",
"IdentityFile": "/dev/null",
"Hostname": "github.com",
"User": "git",
"Port": 22,
},
]
}
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}
def fixture_with_git_dict() -> dict[str, Any]:
"""Bottle declares a git-gate upstream. JSON shape."""
return {
@@ -98,10 +70,6 @@ def fixture_with_egress() -> Manifest:
return Manifest.from_json_obj(fixture_with_egress_dict())
def fixture_with_ssh() -> Manifest:
return Manifest.from_json_obj(fixture_with_ssh_dict())
def fixture_with_git() -> Manifest:
return Manifest.from_json_obj(fixture_with_git_dict())
tests/integration/test_dry_run_plan.py
-2
View File
@@ -79,8 +79,6 @@ class TestDryRunPlan(unittest.TestCase):
self.assertEqual("runc", plan["runtime"],
"runsc isn't available on the CI runner")
self.assertEqual([], plan["skills"])
self.assertEqual([], plan["ssh_hosts"])
self.assertEqual([], plan["ssh_gate"])
self.assertEqual([], plan["git_remotes"])
self.assertEqual([], plan["git_gate"])
self.assertEqual(False, plan["remote_control"])
tests/integration/test_orphan_cleanup.py
+2 -13
View File
@@ -1,8 +1,8 @@
"""Integration: the cleanup primitives the start-flow trap depends on
are idempotent. The original orphan-network bug was a trap-ordering
issue; the fix moved the install earlier. The trap is only safe if
network_remove, PipelockProxy.stop, and SSHGate.stop are no-ops
against missing resources."""
network_remove and PipelockProxy.stop are no-ops against missing
resources."""
import os
import subprocess
@@ -17,10 +17,6 @@ from claude_bottle.backend.docker.pipelock import (
DockerPipelockProxy,
pipelock_container_name,
)
from claude_bottle.backend.docker.ssh_gate import (
DockerSSHGate,
ssh_gate_container_name,
)
from tests._docker import skip_unless_docker
@@ -79,13 +75,6 @@ class TestOrphanCleanup(unittest.TestCase):
# Should not raise.
DockerPipelockProxy().stop(pipelock_container_name(f"missing-{self.slug}"))
def test_ssh_gate_stop_missing_sidecar(self):
# Same trap-safety requirement for the gate (PRD 0007). The
# launch ExitStack calls gate.stop on every error path; if
# the container was never created (early failure), stop must
# still no-op.
DockerSSHGate().stop(ssh_gate_container_name(f"missing-{self.slug}"))
if __name__ == "__main__":
unittest.main()
tests/unit/test_manifest_git.py
+3 -34
View File
@@ -168,11 +168,9 @@ class TestGitEntryCrossValidation(unittest.TestCase):
"IdentityFile": "/dev/null"},
]))
def test_shadow_route_with_ssh_entry_dies(self):
# An ssh entry pointing at gitea.dideric.is:30009 AND a git
# entry pointing at ssh://git@gitea.dideric.is:30009/... is a
# bypass: agents could route around the gate by using the
# ssh-gate. Manifest construction must reject.
def test_legacy_ssh_field_dies_with_hint(self):
# PRD 0009: bottle.ssh is removed; manifests carrying it must
# fail loudly with a hint pointing at bottle.git.
with self.assertRaises(Die):
Manifest.from_json_obj({
"bottles": {
@@ -184,40 +182,11 @@ class TestGitEntryCrossValidation(unittest.TestCase):
"User": "git",
"Port": 30009,
}],
"git": [{
"Name": "claude-bottle",
"Upstream": "ssh://git@gitea.dideric.is:30009/didericis/claude-bottle.git",
"IdentityFile": "/dev/null",
}],
},
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
def test_independent_ssh_and_git_targets_allowed(self):
# Same hostname but different ports are independent targets.
m = Manifest.from_json_obj({
"bottles": {
"dev": {
"ssh": [{
"Host": "gitea-ssh",
"IdentityFile": "/dev/null",
"Hostname": "gitea.dideric.is",
"User": "git",
"Port": 22,
}],
"git": [{
"Name": "claude-bottle",
"Upstream": "ssh://git@gitea.dideric.is:30009/didericis/claude-bottle.git",
"IdentityFile": "/dev/null",
}],
},
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
self.assertEqual(1, len(m.bottles["dev"].ssh))
self.assertEqual(1, len(m.bottles["dev"].git))
class TestEmptyGitField(unittest.TestCase):
def test_no_git_field_yields_empty_tuple(self):
tests/unit/test_pipelock_allowlist.py
+10 -15
View File
@@ -1,7 +1,6 @@
"""Unit: pipelock_effective_allowlist — the union of baked-in defaults
and bottle.egress.allowlist. Per PRD 0007, bottle.ssh entries do NOT
contribute (SSH traffic goes through the per-agent ssh-gate, not
pipelock)."""
and bottle.egress.allowlist. Git upstreams declared in bottle.git do not
contribute here; they flow through the per-agent git-gate (PRD 0008)."""
import unittest
@@ -14,25 +13,21 @@ class TestEffectiveAllowlist(unittest.TestCase):
manifest = Manifest.from_json_obj({
"bottles": {
"dev": {
"egress": {"allowlist": ["registry.npmjs.org"]},
"ssh": [
{"Host": "ts", "IdentityFile": "/dev/null",
"Hostname": "100.78.141.42", "User": "git", "Port": 30009},
{"Host": "gh", "IdentityFile": "/dev/null",
"Hostname": "github.com", "User": "git", "Port": 22},
"egress": {
"allowlist": [
"registry.npmjs.org",
# Duplicate of a baked default; the union
# must dedupe.
"api.anthropic.com",
],
}
},
},
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
eff = pipelock_effective_allowlist(manifest.bottles["dev"])
self.assertIn("api.anthropic.com", eff, "baked default present")
self.assertIn("registry.npmjs.org", eff, "egress.allowlist present")
# PRD 0007: ssh hostnames must not contribute to pipelock's
# allowlist anymore — they're routed through the ssh-gate
# sidecar, which is on its own egress path.
self.assertNotIn("100.78.141.42", eff)
self.assertNotIn("github.com", eff)
self.assertEqual(len(eff), len(set(eff)), "deduplicated")
self.assertEqual(eff, sorted(eff), "sorted")
tests/unit/test_pipelock_yaml.py
+5 -17
View File
@@ -19,7 +19,7 @@ from claude_bottle.pipelock import (
pipelock_build_config,
pipelock_render_yaml,
)
from tests.fixtures import fixture_minimal, fixture_with_ssh
from tests.fixtures import fixture_minimal
class TestBuildConfig(unittest.TestCase):
@@ -38,26 +38,14 @@ class TestBuildConfig(unittest.TestCase):
# Baked defaults always present.
self.assertIn("api.anthropic.com", cast(list[str], cfg["api_allowlist"]))
self.assertIn("raw.githubusercontent.com", cast(list[str], cfg["api_allowlist"]))
# PRD 0007: pipelock has no SSH carve-outs at all — neither
# trusted_domains nor ssrf are ever emitted from bottle data
# in v1.
# pipelock has no SSH carve-outs at all — neither
# trusted_domains nor ssrf are emitted from bottle data.
self.assertNotIn("trusted_domains", cfg)
self.assertNotIn("ssrf", cfg)
# Without CA paths, the tls_interception block is omitted —
# pipelock falls back to its built-in default of `enabled: false`.
self.assertNotIn("tls_interception", cfg)
def test_ssh_entries_do_not_leak_into_pipelock(self):
# PRD 0007: bottle.ssh routes through the ssh-gate sidecar,
# so pipelock's config must not reflect those hostnames or
# IPs in any of its blocks.
cfg = pipelock_build_config(fixture_with_ssh().bottles["dev"])
allow = cast(list[str], cfg["api_allowlist"])
self.assertNotIn("github.com", allow)
self.assertNotIn("100.78.141.42", allow)
self.assertNotIn("trusted_domains", cfg)
self.assertNotIn("ssrf", cfg)
def test_tls_interception_block_emitted_when_paths_supplied(self):
# PRD 0006: paths flow in via DockerPipelockProxy's in-container
# constants; this directly pins the dict shape. passthrough_domains
@@ -102,7 +90,7 @@ class TestRenderAndWrite(unittest.TestCase):
"""One render-level smoke check: the serialized YAML is plausibly
the shape pipelock expects. We don't grep every key here — that's
what TestBuildConfig is for."""
cfg = pipelock_build_config(fixture_with_ssh().bottles["dev"])
cfg = pipelock_build_config(fixture_minimal().bottles["dev"])
text = pipelock_render_yaml(cfg)
for required in (
"api_allowlist:",
@@ -111,7 +99,7 @@ class TestRenderAndWrite(unittest.TestCase):
"request_body_scanning:",
):
self.assertIn(required, text)
# PRD 0007: no ssh carve-outs in the rendered yaml.
# No ssh carve-outs in the rendered yaml.
self.assertNotIn("trusted_domains:", text)
self.assertNotIn("ssrf:", text)
tests/unit/test_ssh_gate.py
-137
View File
@@ -1,137 +0,0 @@
"""Unit: SSHGate prepare shape + entrypoint render."""
import os
import stat
import tempfile
import unittest
from pathlib import Path
from claude_bottle.manifest import Manifest
from claude_bottle.ssh_gate import (
SSHGate,
SSHGatePlan,
SSHGateUpstream,
ssh_gate_render_entrypoint,
ssh_gate_upstreams_for_bottle,
)
from tests.fixtures import fixture_minimal, fixture_with_ssh
class _StubGate(SSHGate):
"""Concrete subclass for testing the abstract `prepare`. The
backend-specific start/stop aren't exercised here."""
def start(self, plan: SSHGatePlan) -> str:
raise NotImplementedError
def stop(self, target: str) -> None:
raise NotImplementedError
class TestUpstreamAssignment(unittest.TestCase):
def test_listen_port_matches_upstream_port(self):
# Critical: URLs like ssh://git@host:30009/... override the
# config Port directive, so the gate must listen on the same
# port the URL names.
bottle = fixture_with_ssh().bottles["dev"]
upstreams = ssh_gate_upstreams_for_bottle(bottle)
self.assertEqual(2, len(upstreams))
# Fixture: tailscale-gitea -> 100.78.141.42:30009, github -> github.com:22.
self.assertEqual(30009, upstreams[0].listen_port)
self.assertEqual(22, upstreams[1].listen_port)
def test_upstream_fields_mirror_ssh_entry(self):
bottle = fixture_with_ssh().bottles["dev"]
first = ssh_gate_upstreams_for_bottle(bottle)[0]
self.assertEqual("tailscale-gitea", first.bottle_host_alias)
self.assertEqual("100.78.141.42", first.upstream_host)
self.assertEqual("30009", first.upstream_port)
def test_empty_bottle_yields_empty_upstreams(self):
bottle = fixture_minimal().bottles["dev"]
self.assertEqual((), ssh_gate_upstreams_for_bottle(bottle))
def test_duplicate_upstream_port_is_rejected(self):
# Two entries on the same upstream port can't both have a
# listener — the gate is one container with a flat port
# space. Surface as a clear config error.
manifest = Manifest.from_json_obj({
"bottles": {
"dev": {
"ssh": [
{"Host": "a", "IdentityFile": "/dev/null",
"Hostname": "host-a.example", "User": "git", "Port": 22},
{"Host": "b", "IdentityFile": "/dev/null",
"Hostname": "host-b.example", "User": "git", "Port": 22},
],
}
},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
})
with self.assertRaises(SystemExit):
ssh_gate_upstreams_for_bottle(manifest.bottles["dev"])
class TestEntrypointRender(unittest.TestCase):
def test_one_socat_line_per_upstream(self):
upstreams = (
SSHGateUpstream(30009, "gitea.example", "30009", "gitea"),
SSHGateUpstream(22, "github.com", "22", "gh"),
)
script = ssh_gate_render_entrypoint(upstreams)
self.assertIn("#!/bin/sh", script)
self.assertIn(
"socat TCP-LISTEN:30009,reuseaddr,fork TCP:gitea.example:30009 &", script
)
self.assertIn(
"socat TCP-LISTEN:22,reuseaddr,fork TCP:github.com:22 &", script
)
# wait blocks the entrypoint so PID 1 stays alive while sockets
# are open.
self.assertTrue(script.rstrip().endswith("wait"))
def test_empty_upstreams_still_has_wait(self):
# Defensive: a no-upstream gate is a no-op, but render must
# still produce a valid shell script.
script = ssh_gate_render_entrypoint(())
self.assertIn("#!/bin/sh", script)
self.assertIn("wait", script)
class TestPrepare(unittest.TestCase):
def setUp(self):
self.stage = Path(tempfile.mkdtemp())
def tearDown(self):
import shutil
shutil.rmtree(self.stage, ignore_errors=True)
def test_prepare_writes_entrypoint_mode_600(self):
plan = _StubGate().prepare(
fixture_with_ssh().bottles["dev"], "demo", self.stage
)
self.assertEqual(self.stage / "ssh_gate_entrypoint.sh", plan.entrypoint_script)
self.assertEqual(0o600, os.stat(plan.entrypoint_script).st_mode & 0o777)
def test_prepare_plan_carries_upstreams_and_slug(self):
plan = _StubGate().prepare(
fixture_with_ssh().bottles["dev"], "demo", self.stage
)
self.assertEqual("demo", plan.slug)
self.assertEqual(2, len(plan.upstreams))
self.assertEqual("", plan.internal_network)
self.assertEqual("", plan.egress_network)
def test_prepare_with_no_ssh_writes_minimal_script(self):
plan = _StubGate().prepare(
fixture_minimal().bottles["dev"], "demo", self.stage
)
self.assertEqual((), plan.upstreams)
content = plan.entrypoint_script.read_text()
self.assertNotIn("socat", content)
self.assertIn("wait", content)
if __name__ == "__main__":
unittest.main()