bot-bottle/tests/integration/test_smolmachines_launch.py

"""Integration: end-to-end smolmachines launch + exec round trip.

The smoke confirms the launch flow (sidecar bundle smolVM →
host-loopback forwarders → agent smolVM with TSI allowlist → exec)
plumbs together end to end. The probes confirm the security
properties the design pivot was about:

  - **localhost-reach probe** — guest tries to dial a service
    bound on the host's `127.0.0.1`. TSI's per-bottle loopback
    alias allowlist must refuse the connect.

  - **egress proxy probe** — guest reaches the egress proxy through
    the injected `HTTPS_PROXY`/`HTTP_PROXY` URL on the per-bottle
    loopback alias, while direct egress with proxy vars unset fails.

Gated on macOS/Linux + smolvm + docker + not GITEA_ACTIONS — the
runner can't host libkrun-backed VMs."""

from __future__ import annotations

import os
import platform
import shutil
import tempfile
import unittest
from pathlib import Path

from bot_bottle.backend import BottleSpec, get_bottle_backend
from bot_bottle.backend.smolmachines.smolvm import is_available as _smolvm_available
from bot_bottle.manifest import ManifestIndex
from tests._docker import skip_unless_docker


_AGENT_PROMPT = "You are demo. Be brief."


def _minimal_manifest() -> ManifestIndex:
    return ManifestIndex.from_json_obj({
        "bottles": {
            "dev": {
                "egress": {
                    "routes": [
                        {"host": "example.com"},
                    ],
                },
            },
        },
        "agents": {
            "demo": {
                "skills": [],
                "prompt": _AGENT_PROMPT,
                "bottle": "dev",
            },
        },
    })


@skip_unless_docker()
@unittest.skipUnless(
    platform.system() in ("Darwin", "Linux"),
    "smolvm requires macOS or Linux",
)
@unittest.skipUnless(
    _smolvm_available(),
    "smolvm not on PATH; install via "
    "curl -sSL https://smolmachines.com/install.sh | sh",
)
@unittest.skipIf(
    os.environ.get("GITEA_ACTIONS") == "true",
    "skipped under act_runner: cannot host libkrun-backed VMs",
)
class TestSmolmachinesLaunch(unittest.TestCase):
    """The full smoke + the two acceptance probes share one
    bottle bringup to amortize the ~10s cold-start cost across
    three assertions."""

    @classmethod
    def setUpClass(cls) -> None:
        cls.stage = Path(tempfile.mkdtemp(prefix="cb-smol-launch."))
        os.environ["BOT_BOTTLE_BACKEND"] = "smolmachines"
        backend = get_bottle_backend()
        spec = BottleSpec(
            manifest=_minimal_manifest(),
            agent_name="demo",
            copy_cwd=False,
            user_cwd=str(cls.stage),
        )
        cls.plan = backend.prepare(spec, stage_dir=cls.stage)
        cls._launch = backend.launch(cls.plan)
        cls.bottle = cls._launch.__enter__()

    @classmethod
    def tearDownClass(cls) -> None:
        try:
            cls._launch.__exit__(None, None, None)
        finally:
            shutil.rmtree(cls.stage, ignore_errors=True)
            os.environ.pop("BOT_BOTTLE_BACKEND", None)

    def test_smoke_exec_echo(self):
        # The plumbing-verifies-end-to-end smoke: a shell command
        # round-trips through smolvm machine exec.
        r = self.bottle.exec("echo hello-from-vm")
        self.assertEqual(0, r.returncode, msg=r.stderr)
        self.assertIn("hello-from-vm", r.stdout)

    def test_localhost_reach_probe(self):
        # Agent dials a 127.0.0.1 service on the host. TSI's
        # allowlist contains only the per-bottle loopback alias, so
        # this must refuse. We use a port unlikely to be bound on the host
        # (high-numbered) so we're confirming TSI refusal, not
        # just "no service listening."
        r = self.bottle.exec(
            "curl -s --show-error --max-time 3 http://127.0.0.1:9 2>&1 || true"
        )
        # `curl` to a denied destination produces a connect error.
        # The exact phrasing varies by curl version; we assert
        # the response is NOT the body of any real service.
        self.assertNotIn("hello-from-vm", r.stdout)
        self.assertTrue(
            "refused" in r.stdout.lower()
            or "timed out" in r.stdout.lower()
            or "unreachable" in r.stdout.lower()
            or "failed" in r.stdout.lower(),
            f"expected a connect-refusal message; got: {r.stdout!r}",
        )

    def test_egress_proxy_reachable_through_tsi_loopback_alias(self):
        r = self.bottle.exec(
            "printf '%s\n' \"$HTTPS_PROXY\" \"$HTTP_PROXY\""
        )
        self.assertEqual(0, r.returncode, msg=r.stderr)
        proxies = [line.strip() for line in r.stdout.splitlines()]
        self.assertEqual(2, len(proxies), proxies)
        self.assertEqual(proxies[0], proxies[1], proxies)
        # macOS: proxy binds to the per-bottle loopback alias (127.x.x.x) so
        # TSI can intercept guest connections to it.  Linux: the guest kernel
        # routes 127.0.0.0/8 to its own loopback (TSI never sees those), so
        # the proxy instead binds to the per-bottle bridge gateway (192.168.x.1)
        # which routes via eth0 and is intercepted by TSI normally.
        self.assertRegex(
            proxies[0], r"^http://\d+\.\d+\.\d+\.\d+:\d+$",
            "expected proxy URL to be an http://IP:port address",
        )

        r = self.bottle.exec(
            "curl -fsS --max-time 20 https://example.com >/dev/null && echo OK"
        )
        self.assertEqual(0, r.returncode, msg=r.stderr + r.stdout)
        self.assertIn("OK", r.stdout)

    def test_direct_egress_bypass_without_proxy_fails(self):
        r = self.bottle.exec(
            "env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
            "curl -s --show-error --max-time 5 https://example.com 2>&1 || true"
        )
        self.assertTrue(
            "refused" in r.stdout.lower()
            or "timed out" in r.stdout.lower()
            or "unreachable" in r.stdout.lower()
            or "failed" in r.stdout.lower()
            or "could not resolve" in r.stdout.lower()
            or "connection reset" in r.stdout.lower(),
            f"expected direct egress to fail; got: {r.stdout!r}",
        )

    def test_non_allowlisted_host_fails_through_proxy(self):
        r = self.bottle.exec(
            "curl -s --show-error --max-time 10 https://iana.org 2>&1 || true"
        )
        self.assertTrue(
            "403" in r.stdout
            or "502" in r.stdout
            or "blocked" in r.stdout.lower()
            or "not allowed" in r.stdout.lower()
            or "not in the bottle's egress.routes allowlist" in r.stdout.lower()
            or "forbidden" in r.stdout.lower()
            or "failed" in r.stdout.lower(),
            f"expected non-allowlisted proxy request to fail; got: {r.stdout!r}",
        )

    def test_prompt_file_lands_in_guest(self):
        # provision_prompt copies the host-side prompt.txt into the
        # guest at /home/node/.bot-bottle-prompt.txt. The content
        # must match what the manifest declared so claude-code's
        # --append-system-prompt-file reads the right text.
        r = self.bottle.exec("cat /home/node/.bot-bottle-prompt.txt")
        self.assertEqual(0, r.returncode, msg=r.stderr)
        self.assertEqual(_AGENT_PROMPT, r.stdout.rstrip("\n"))


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