bot-bottle/tests/unit/test_pipelock_yaml.py

"""Unit: pipelock config building and YAML rendering.

`pipelock_build_config` produces the structured config dict pipelock
will load; tests assert on that dict so they don't break on cosmetic
YAML changes. A small set of tests still hit the rendered output for
properties that only make sense on disk (file mode, no-secret-leakage).
"""

import os
import tempfile
import unittest
from pathlib import Path
from typing import Any, cast

from claude_bottle.backend.docker.pipelock import DockerPipelockProxy
from claude_bottle.manifest import Manifest
from claude_bottle.pipelock import pipelock_build_config, pipelock_render_yaml
from tests.fixtures import fixture_minimal, fixture_with_ssh


class TestBuildConfig(unittest.TestCase):
    def test_minimal_shape(self):
        cfg = pipelock_build_config(fixture_minimal().bottles["dev"])
        self.assertEqual("strict", cfg["mode"])
        self.assertEqual(True, cfg["enforce"])
        self.assertEqual({"enabled": True}, cfg["forward_proxy"])
        self.assertEqual(
            {"include_defaults": True, "scan_env": True}, cfg["dlp"]
        )
        # Default body-scan action is "block" — see BottleEgress.dlp_action.
        self.assertEqual(
            {"action": "block"}, cfg["request_body_scanning"]
        )
        # 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.
        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.
        cfg = pipelock_build_config(
            fixture_minimal().bottles["dev"],
            ca_cert_path="/etc/pipelock-ca.pem",
            ca_key_path="/etc/pipelock-ca-key.pem",
        )
        self.assertEqual(
            {
                "enabled": True,
                "ca_cert": "/etc/pipelock-ca.pem",
                "ca_key": "/etc/pipelock-ca-key.pem",
            },
            cfg["tls_interception"],
        )

    def test_tls_interception_requires_both_paths(self):
        # Half-set is a programmer error, not a silent omission.
        with self.assertRaises(ValueError):
            pipelock_build_config(
                fixture_minimal().bottles["dev"],
                ca_cert_path="/etc/pipelock-ca.pem",
            )


class TestRenderAndWrite(unittest.TestCase):
    def setUp(self):
        self.out_dir = Path(tempfile.mkdtemp())

    def tearDown(self):
        import shutil
        shutil.rmtree(self.out_dir, ignore_errors=True)

    def test_render_emits_required_top_level_keys(self):
        """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"])
        text = pipelock_render_yaml(cfg)
        for required in (
            "api_allowlist:",
            "forward_proxy:",
            "dlp:",
            "request_body_scanning:",
        ):
            self.assertIn(required, text)
        # PRD 0007: no ssh carve-outs in the rendered yaml.
        self.assertNotIn("trusted_domains:", text)
        self.assertNotIn("ssrf:", text)

    def test_prepare_writes_file_at_mode_600(self):
        plan = DockerPipelockProxy().prepare(
            fixture_minimal().bottles["dev"], "demo", self.out_dir
        )
        self.assertEqual(0o600, os.stat(plan.yaml_path).st_mode & 0o777)

    def test_prepare_does_not_leak_env_names_or_values(self):
        manifest = Manifest.from_json_obj({
            "bottles": {
                "dev": {
                    "env": {
                        "MY_SECRET": "literal-value-should-not-appear",
                        "ANOTHER": "?prompt-message",
                    },
                    "egress": {"allowlist": ["github.com"]},
                }
            },
            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
        })
        plan = DockerPipelockProxy().prepare(
            manifest.bottles["dev"], "demo", self.out_dir
        )
        content = plan.yaml_path.read_text()
        self.assertNotIn("literal-value-should-not-appear", content)
        self.assertNotIn("MY_SECRET", content)
        self.assertNotIn("prompt-message", content)

    def test_render_emits_tls_interception_via_prepare(self):
        """`DockerPipelockProxy.prepare` plumbs its in-container CA
        constants through to the YAML. The block should land in the
        rendered output with `enabled: true` and the configured paths.
        The actual host-side CA generation happens in launch (not
        prepare), so this test exercises only the YAML rendering."""
        plan = DockerPipelockProxy().prepare(
            fixture_minimal().bottles["dev"], "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)


if __name__ == "__main__":
    unittest.main()