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

test: drop ssh-gate suites and shadow-route assertions (PRD 0009)

Browse Source 
- Delete tests/unit/test_ssh_gate.py and the fixture_with_ssh helpers.
- test_pipelock_yaml: drop the ssh-leak guard (structurally
  impossible now); the remaining tests switch to fixture_minimal.
- test_pipelock_allowlist: rewrite the union/dedup test to
  exercise an egress.allowlist that duplicates a baked default
  (the property the ssh-leak assertion was hitching onto).
- test_manifest_git: shadow-route assertion becomes a legacy-ssh-
  dies-with-hint assertion, since bottle.ssh is now parse-fail.
- test_orphan_cleanup: drop the SSHGate.stop idempotency check;
  pipelock equivalent stays.
- test_dry_run_plan: drop assertions on the removed ssh_hosts /
  ssh_gate keys.

52 unit tests pass.
This commit is contained in:
didericis
2026-05-12 23:54:22 -04:00
parent 3d66ad2a86
commit 249e8cc15e
7 changed files with 21 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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
+11 -16
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()