feat(pipelock): allow route tls passthrough policy
test / unit (pull_request) Successful in 37s
test / integration (pull_request) Successful in 58s

This commit is contained in:
2026-05-28 19:19:40 -04:00
parent 3299674c30
commit bcadc07d09
11 changed files with 164 additions and 78 deletions
+28
View File
@@ -215,6 +215,34 @@ class TestRole(unittest.TestCase):
}])
class TestPipelockPolicy(unittest.TestCase):
def test_tls_passthrough_route_policy(self):
b = _bottle([{
"host": "api.openai.com",
"pipelock": {"tls_passthrough": True},
}])
self.assertTrue(b.egress.routes[0].Pipelock.TlsPassthrough)
def test_tls_passthrough_defaults_false(self):
b = _bottle([{"host": "api.openai.com"}])
self.assertFalse(b.egress.routes[0].Pipelock.TlsPassthrough)
def test_pipelock_policy_must_be_object(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "pipelock": True}])
def test_tls_passthrough_must_be_bool(self):
with self.assertRaises(Die):
_bottle([{
"host": "x.example",
"pipelock": {"tls_passthrough": "yes"},
}])
def test_unknown_pipelock_key_rejected(self):
with self.assertRaises(Die):
_bottle([{"host": "x.example", "pipelock": {"wat": True}}])
class TestRouteValidation(unittest.TestCase):
def test_duplicate_hosts_rejected(self):
# Routes match by exact host; duplicates leave the choice
+14 -4
View File
@@ -89,18 +89,28 @@ class TestAllowlistWithRoutes(unittest.TestCase):
class TestTlsPassthrough(unittest.TestCase):
def test_default_includes_api_anthropic(self):
def test_default_empty(self):
passthrough = pipelock_effective_tls_passthrough(_bottle({}))
self.assertEqual(["api.anthropic.com"], passthrough)
self.assertEqual([], passthrough)
def test_route_hosts_NOT_added_to_passthrough(self):
def test_route_hosts_not_added_to_passthrough_by_default(self):
passthrough = pipelock_effective_tls_passthrough(_bottle(_routes([
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "G"}},
{"host": "registry.npmjs.org",
"auth": {"scheme": "Bearer", "token_ref": "N"}},
])))
self.assertEqual(["api.anthropic.com"], passthrough)
self.assertEqual([], passthrough)
def test_route_policy_adds_tls_passthrough(self):
passthrough = pipelock_effective_tls_passthrough(_bottle(_routes([
{"host": "api.openai.com",
"auth": {"scheme": "Bearer", "token_ref": "O"},
"pipelock": {"tls_passthrough": True}},
{"host": "api.github.com",
"auth": {"scheme": "Bearer", "token_ref": "G"}},
])))
self.assertEqual(["api.openai.com"], passthrough)
if __name__ == "__main__":
+30 -12
View File
@@ -54,11 +54,7 @@ class TestBuildConfig(unittest.TestCase):
def test_tls_interception_block_emitted_when_paths_supplied(self):
# PRD 0006: paths flow in via the platform-neutral in-container
# constants; this directly pins the dict shape. passthrough_domains
# is baked in so LLM provider endpoints (api.anthropic.com) skip
# MITM — pipelock's docs explicitly recommend this for LLM hosts,
# and without it the BIP-39 body scanner false-positives on
# Claude conversation traffic.
# constants; this directly pins the dict shape.
cfg = pipelock_build_config(
fixture_minimal().bottles["dev"],
ca_cert_path="/etc/pipelock-ca.pem",
@@ -69,11 +65,28 @@ class TestBuildConfig(unittest.TestCase):
"enabled": True,
"ca_cert": "/etc/pipelock-ca.pem",
"ca_key": "/etc/pipelock-ca-key.pem",
"passthrough_domains": list(DEFAULT_TLS_PASSTHROUGH),
"passthrough_domains": [],
},
cfg["tls_interception"],
)
self.assertIn("api.anthropic.com", DEFAULT_TLS_PASSTHROUGH)
self.assertEqual((), DEFAULT_TLS_PASSTHROUGH)
def test_tls_passthrough_route_policy_emits_domain(self):
bottle = Manifest.from_json_obj({
"bottles": {"dev": {"egress": {"routes": [
{"host": "api.openai.com",
"auth": {"scheme": "Bearer", "token_ref": "T"},
"pipelock": {"tls_passthrough": True}},
]}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}).bottles["dev"]
cfg = pipelock_build_config(
bottle,
ca_cert_path="/etc/pipelock-ca.pem",
ca_key_path="/etc/pipelock-ca-key.pem",
)
tls = cast(dict[str, object], cfg["tls_interception"])
self.assertEqual(["api.openai.com"], tls["passthrough_domains"])
def test_tls_interception_requires_both_paths(self):
# Half-set is a programmer error, not a silent omission.
@@ -179,19 +192,24 @@ class TestRenderAndWrite(unittest.TestCase):
"""`PipelockProxy.prepare` plumbs the module-level in-container
CA constants through to the YAML. The block should land in the
rendered output with `enabled: true`, the configured paths,
and the baked LLM-provider passthrough list. The actual
and any route-owned passthrough domains. The actual
host-side CA generation happens in launch (not prepare), so
this test exercises only the YAML rendering."""
plan = PipelockProxy().prepare(
fixture_minimal().bottles["dev"], "demo", self.out_dir
)
bottle = Manifest.from_json_obj({
"bottles": {"dev": {"egress": {"routes": [
{"host": "api.openai.com",
"pipelock": {"tls_passthrough": True}},
]}}},
"agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
}).bottles["dev"]
plan = PipelockProxy().prepare(bottle, "demo", self.out_dir)
content = plan.yaml_path.read_text()
self.assertIn("tls_interception:", content)
self.assertIn("enabled: true", content)
self.assertIn('ca_cert: "/etc/pipelock-ca.pem"', content)
self.assertIn('ca_key: "/etc/pipelock-ca-key.pem"', content)
self.assertIn("passthrough_domains:", content)
self.assertIn('- "api.anthropic.com"', content)
self.assertIn('- "api.openai.com"', content)
def test_render_emits_ssrf_block_when_allowlist_given(self):
cfg = pipelock_build_config(