feat(bottle): add exec method to the bottle abstraction

Bottle.exec(script) -> ExecResult runs a POSIX shell script inside a
running bottle and returns captured stdout/stderr/returncode. The
Docker impl pipes the script via stdin to `docker exec -i ... sh -s`
so the source never crosses argv.

Two integration tests exercise it end-to-end through the pipelock
sidecar: a Node request to a non-allowlisted host (example.com)
returns 403 from pipelock; a Node CONNECT to an allowlisted host
(raw.githubusercontent.com) is tunneled with 200 Connection
Established. The 200/403 split on each verb is decided by pipelock
itself, isolating the allowlist decision from whatever the remote
might return.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

claude_bottle/backend/__init__.py

@@ -94,12 +94,26 @@ class BottleCleanupPlan(ABC):
         short-circuit before showing the y/N."""
 
 
+@dataclass(frozen=True)
+class ExecResult:
+    """Captured result of `Bottle.exec`. Backend-neutral: the Docker
+    impl populates it from a `subprocess.CompletedProcess`, but a
+    future fly/smolmachines backend could populate it from any source
+    that produces a returncode + captured streams."""
+
+    returncode: int
+    stdout: str
+    stderr: str
+
+
 class Bottle(ABC):
     """Handle to a running bottle. Yielded by a backend's launch step.
 
     `exec_claude` runs `claude` inside the bottle and blocks until the
-    session ends. `cp_in` copies a host path into the bottle. `close`
-    is an idempotent alias for context-manager teardown.
+    session ends. `exec` runs a POSIX shell script inside the bottle
+    and returns the captured result. `cp_in` copies a host path into
+    the bottle. `close` is an idempotent alias for context-manager
+    teardown.
     """
 
     name: str
@@ -107,6 +121,14 @@ class Bottle(ABC):
     @abstractmethod
     def exec_claude(self, argv: list[str], *, tty: bool = True) -> int: ...
 
+    @abstractmethod
+    def exec(self, script: str) -> ExecResult:
+        """Run `script` as a POSIX shell script inside the bottle and
+        return the captured stdout/stderr/returncode. The bottle's
+        environment (including HTTPS_PROXY pointing at the pipelock
+        sidecar) is inherited by the child. Non-zero exit does not
+        raise — callers inspect `returncode` themselves."""
+
     @abstractmethod
     def cp_in(self, host_path: str, container_path: str) -> None: ...
 
@@ -270,5 +292,6 @@ __all__ = [
     "BottleCleanupPlan",
     "BottlePlan",
     "BottleSpec",
+    "ExecResult",
     "get_bottle_backend",
 ]

claude_bottle/backend/docker/bottle.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 import subprocess
 from typing import Callable
 
-from .. import Bottle
+from .. import Bottle, ExecResult
 
 
 class DockerBottle(Bottle):
@@ -38,6 +38,23 @@ class DockerBottle(Bottle):
         cmd.extend([self.name, "claude", *full_argv])
         return subprocess.run(cmd, check=False).returncode
 
+    def exec(self, script: str) -> ExecResult:
+        # Pipe via stdin to `sh -s` so the caller never has to worry
+        # about quoting; the script source lands inside the container
+        # without crossing argv.
+        result = subprocess.run(
+            ["docker", "exec", "-i", self.name, "sh", "-s"],
+            input=script,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        return ExecResult(
+            returncode=result.returncode,
+            stdout=result.stdout,
+            stderr=result.stderr,
+        )
+
     def cp_in(self, host_path: str, container_path: str) -> None:
         subprocess.run(
             ["docker", "cp", host_path, f"{self.name}:{container_path}"],

tests/integration/test_pipelock_allow_node.py

@@ -0,0 +1,111 @@
+"""Integration: a Node request to a host on pipelock's allowlist is
+tunneled through.
+
+End-to-end mirror of test_pipelock_block_node: drives `BottleBackend.
+prepare → launch` so the real image build, network plumbing, and
+pipelock sidecar are all in the loop. Inside the bottle, a Node
+script issues an HTTPS CONNECT for raw.githubusercontent.com:443 —
+a host in the baked-in default allowlist — through `$HTTPS_PROXY`.
+Pipelock must answer 200 Connection Established. The 200 vs. 403
+split on CONNECT is decided by pipelock itself (the remote never
+sees the CONNECT verb), so it isolates the allowlist decision from
+anything the remote might return.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from claude_bottle.backend import BottleSpec, get_bottle_backend
+from tests._docker import skip_unless_docker
+from tests.fixtures import fixture_minimal
+
+
+# Output contract (parsed by the test):
+#   - "connect=<code>"  proxy upgraded to a tunnel (CONNECT success path)
+#   - "status=<code>"   proxy answered without tunneling (block path)
+#   - "error=<code> <message>"  transport-level failure
+#   - "timeout"         request hung
+_PROBE_JS = r"""
+const http = require('http');
+const proxy = new URL(process.env.HTTPS_PROXY);
+const req = http.request({
+  host: proxy.hostname,
+  port: proxy.port,
+  method: 'CONNECT',
+  path: 'raw.githubusercontent.com:443',
+});
+req.on('connect', (res, socket) => {
+  console.log('connect=' + res.statusCode);
+  socket.destroy();
+  process.exit(0);
+});
+req.on('response', (res) => {
+  res.resume();
+  res.on('end', () => {
+    console.log('status=' + res.statusCode);
+    process.exit(0);
+  });
+});
+req.on('error', (e) => {
+  console.log('error=' + (e.code || '') + ' ' + e.message);
+  process.exit(0);
+});
+req.setTimeout(5000, () => {
+  console.log('timeout');
+  req.destroy();
+});
+req.end();
+"""
+
+
+@skip_unless_docker()
+class TestPipelockAllowsNode(unittest.TestCase):
+    @unittest.skipIf(
+        os.environ.get("GITEA_ACTIONS") == "true",
+        "skipped under act_runner: docker socket mount topology breaks "
+        "in-process visibility of networks created on the host daemon",
+    )
+    def test_node_request_to_allowed_host_is_tunneled(self):
+        backend = get_bottle_backend()
+        stage_dir = Path(tempfile.mkdtemp(prefix="cb-test-stage."))
+        try:
+            spec = BottleSpec(
+                manifest=fixture_minimal(),
+                agent_name="demo",
+                copy_cwd=False,
+                user_cwd=str(stage_dir),
+                forward_oauth_token=False,
+            )
+            plan = backend.prepare(spec, stage_dir=stage_dir)
+            with backend.launch(plan) as bottle:
+                script = (
+                    "set -e\n"
+                    "cat > /tmp/probe.js <<'PROBE_EOF'\n"
+                    f"{_PROBE_JS}\n"
+                    "PROBE_EOF\n"
+                    "node /tmp/probe.js\n"
+                )
+                result = bottle.exec(script)
+        finally:
+            shutil.rmtree(stage_dir, ignore_errors=True)
+
+        self.assertEqual(
+            0, result.returncode,
+            f"exec wrapper failed: stdout={result.stdout!r} stderr={result.stderr!r}",
+        )
+        # raw.githubusercontent.com IS in fixture_minimal's effective
+        # allowlist (baked-in default). Pipelock must answer the CONNECT
+        # with 200 Connection Established.
+        self.assertIn(
+            "connect=200", result.stdout,
+            f"pipelock should have tunneled to raw.githubusercontent.com; got: {result.stdout!r}",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/integration/test_pipelock_block_node.py

@@ -0,0 +1,115 @@
+"""Integration: a Node script run inside a launched bottle, hitting
+a host outside the pipelock allowlist, is blocked.
+
+End-to-end: drives `BottleBackend.prepare → launch` so the real
+image build, network plumbing, and pipelock sidecar are all in the
+loop. Inside the bottle, a Node script forms an HTTP forward-proxy
+request (absolute-URI path) to `example.com` via `$HTTPS_PROXY`. The
+fixture's effective allowlist contains only the baked-in defaults,
+so pipelock must refuse to forward.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+from claude_bottle.backend import BottleSpec, get_bottle_backend
+from tests._docker import skip_unless_docker
+from tests.fixtures import fixture_minimal
+
+
+# Node's stdlib http does not respect HTTPS_PROXY on its own; this
+# script builds the forward-proxy request shape by hand so the test
+# is asserting on pipelock's allowlist decision, not on whatever
+# proxy-env auto-detection a Node release happens to ship.
+#
+# Output contract (parsed by the test):
+#   - "status=<code>" when the proxy returns an HTTP response
+#   - "error=<code> <message>" on a transport-level failure
+#   - "timeout" on a hung request
+_PROBE_JS = r"""
+const http = require('http');
+const proxy = new URL(process.env.HTTPS_PROXY);
+const req = http.request({
+  host: proxy.hostname,
+  port: proxy.port,
+  method: 'GET',
+  path: 'http://example.com/',
+  headers: { Host: 'example.com' },
+}, (res) => {
+  res.resume();
+  res.on('end', () => {
+    console.log('status=' + res.statusCode);
+    process.exit(0);
+  });
+});
+req.on('error', (e) => {
+  console.log('error=' + (e.code || '') + ' ' + e.message);
+  process.exit(0);
+});
+req.setTimeout(5000, () => {
+  console.log('timeout');
+  req.destroy();
+});
+req.end();
+"""
+
+
+@skip_unless_docker()
+class TestPipelockBlocksNode(unittest.TestCase):
+    @unittest.skipIf(
+        os.environ.get("GITEA_ACTIONS") == "true",
+        "skipped under act_runner: docker socket mount topology breaks "
+        "in-process visibility of networks created on the host daemon",
+    )
+    def test_node_request_to_blocked_host_is_rejected(self):
+        backend = get_bottle_backend()
+        stage_dir = Path(tempfile.mkdtemp(prefix="cb-test-stage."))
+        try:
+            spec = BottleSpec(
+                manifest=fixture_minimal(),
+                agent_name="demo",
+                copy_cwd=False,
+                user_cwd=str(stage_dir),
+                forward_oauth_token=False,
+            )
+            plan = backend.prepare(spec, stage_dir=stage_dir)
+            with backend.launch(plan) as bottle:
+                script = (
+                    "set -e\n"
+                    "cat > /tmp/probe.js <<'PROBE_EOF'\n"
+                    f"{_PROBE_JS}\n"
+                    "PROBE_EOF\n"
+                    "node /tmp/probe.js\n"
+                )
+                result = bottle.exec(script)
+        finally:
+            shutil.rmtree(stage_dir, ignore_errors=True)
+
+        self.assertEqual(
+            0, result.returncode,
+            f"exec wrapper failed: stdout={result.stdout!r} stderr={result.stderr!r}",
+        )
+        # The probe always prints exactly one signal line. If it
+        # doesn't, the script failed in a way the test doesn't
+        # understand and the surrounding assertions would be
+        # ambiguous.
+        self.assertTrue(
+            "status=" in result.stdout or "error=" in result.stdout or "timeout" in result.stdout,
+            f"probe produced no recognized output: {result.stdout!r}",
+        )
+        # The core invariant: example.com is NOT in fixture_minimal's
+        # effective allowlist (only the baked-in defaults), so the
+        # proxy must not have forwarded a successful response.
+        self.assertNotIn(
+            "status=200", result.stdout,
+            "example.com is outside the allowlist; pipelock should not have forwarded a 200",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()