From 249e8cc15ee2c9a7c65887585ae409c3f00a436b Mon Sep 17 00:00:00 2001 From: didericis Date: Tue, 12 May 2026 23:54:22 -0400 Subject: [PATCH] test: drop ssh-gate suites and shadow-route assertions (PRD 0009) - 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. --- tests/fixtures.py | 32 ------ tests/integration/test_dry_run_plan.py | 2 - tests/integration/test_orphan_cleanup.py | 15 +-- tests/unit/test_manifest_git.py | 37 +----- tests/unit/test_pipelock_allowlist.py | 27 ++--- tests/unit/test_pipelock_yaml.py | 22 +--- tests/unit/test_ssh_gate.py | 137 ----------------------- 7 files changed, 21 insertions(+), 251 deletions(-) delete mode 100644 tests/unit/test_ssh_gate.py diff --git a/tests/fixtures.py b/tests/fixtures.py index 49fc04d..639dd0c 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -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()) diff --git a/tests/integration/test_dry_run_plan.py b/tests/integration/test_dry_run_plan.py index 09e8a1c..9a82b76 100644 --- a/tests/integration/test_dry_run_plan.py +++ b/tests/integration/test_dry_run_plan.py @@ -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"]) diff --git a/tests/integration/test_orphan_cleanup.py b/tests/integration/test_orphan_cleanup.py index 2462120..4aff79b 100644 --- a/tests/integration/test_orphan_cleanup.py +++ b/tests/integration/test_orphan_cleanup.py @@ -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() diff --git a/tests/unit/test_manifest_git.py b/tests/unit/test_manifest_git.py index cf22fc8..639ead0 100644 --- a/tests/unit/test_manifest_git.py +++ b/tests/unit/test_manifest_git.py @@ -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): diff --git a/tests/unit/test_pipelock_allowlist.py b/tests/unit/test_pipelock_allowlist.py index 3cd6626..e50d9d6 100644 --- a/tests/unit/test_pipelock_allowlist.py +++ b/tests/unit/test_pipelock_allowlist.py @@ -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") diff --git a/tests/unit/test_pipelock_yaml.py b/tests/unit/test_pipelock_yaml.py index b846be2..caa1105 100644 --- a/tests/unit/test_pipelock_yaml.py +++ b/tests/unit/test_pipelock_yaml.py @@ -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) diff --git a/tests/unit/test_ssh_gate.py b/tests/unit/test_ssh_gate.py deleted file mode 100644 index c972a86..0000000 --- a/tests/unit/test_ssh_gate.py +++ /dev/null @@ -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()