bot-bottle/claude_bottle/backend/docker/mitmproxy.py

"""DockerMitmproxyProxy — the Docker-specific lifecycle for the
mitmproxy sidecar. Inherits the addon-bundling from MitmproxyProxy.

The sidecar runs `mitmdump -s /addon/addon.py`, listens on
MITMPROXY_PORT inside the per-bottle internal network, and generates
its own ephemeral CA on first launch (extracted by provision_ca,
installed into the agent's trust store)."""

from __future__ import annotations

import os
import subprocess
import time
from pathlib import Path

from ...log import die, info, warn
from ...mitmproxy import MitmproxyProxy, MitmproxyProxyPlan


# mitmproxy/mitmproxy:12.2.3 (mitmproxy v12 release line). The digest
# is the multi-arch image index — pulls resolve to the right per-arch
# child digest. Bumped deliberately; see PRD 0005.
MITMPROXY_IMAGE = os.environ.get(
    "CLAUDE_BOTTLE_MITMPROXY_IMAGE",
    "mitmproxy/mitmproxy@sha256:00b77b5d8804c8ad18cb6caefbf9d5849e895e8986c5ce011f4ae30f4385962f",
)

# Listening port for mitmproxy's forward proxy (agent-facing).
MITMPROXY_PORT = os.environ.get("CLAUDE_BOTTLE_MITMPROXY_PORT", "8080")

# Path inside the sidecar where the addon is dropped by docker cp.
MITMPROXY_ADDON_PATH = "/addon/addon.py"

# Path inside the sidecar where mitmproxy generates its CA.
_CA_PATH_IN_SIDECAR = "/home/mitmproxy/.mitmproxy/mitmproxy-ca-cert.pem"


def mitmproxy_container_name(slug: str) -> str:
    return f"claude-bottle-mitm-{slug}"


def mitmproxy_proxy_url(slug: str) -> str:
    return f"http://{mitmproxy_container_name(slug)}:{MITMPROXY_PORT}"


class DockerMitmproxyProxy(MitmproxyProxy):
    """Brings the mitmproxy sidecar up and down via Docker."""

    def start(self, plan: MitmproxyProxyPlan, *, pipelock_url: str) -> str:
        """Boot the mitmproxy sidecar:
          1. `docker create` on the internal network with mitmdump
             argv: `--listen-port <port> -s <addon path>` plus the
             pipelock URL injected as an env var.
          2. `docker cp` the vendored addon to the sidecar.
          3. Attach to the per-agent egress network so mitmproxy
             can reach real upstreams.
          4. `docker start`.
        Returns the container name (the proxy_target passed to .stop
        and .extract_ca_cert)."""
        name = mitmproxy_container_name(plan.slug)
        if not plan.addon_src.is_file():
            die(f"mitmproxy addon not found at {plan.addon_src}")

        info(f"starting mitmproxy sidecar {name} on network {plan.internal_network}")

        create_args = [
            "docker", "create",
            "--name", name,
            "--network", plan.internal_network,
            "-e", f"CLAUDE_BOTTLE_PIPELOCK_URL={pipelock_url}",
            MITMPROXY_IMAGE,
            "mitmdump",
            "--listen-port", MITMPROXY_PORT,
            "-s", MITMPROXY_ADDON_PATH,
        ]
        if subprocess.run(
            create_args,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        ).returncode != 0:
            die(f"failed to create mitmproxy sidecar {name}")

        cp_result = subprocess.run(
            ["docker", "cp", str(plan.addon_src), f"{name}:{MITMPROXY_ADDON_PATH}"],
            capture_output=True,
            text=True,
            check=False,
        )
        if cp_result.returncode != 0:
            subprocess.run(
                ["docker", "rm", "-f", name],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )
            die(f"failed to copy mitmproxy addon into {name}: {cp_result.stderr.strip()}")

        if subprocess.run(
            ["docker", "network", "connect", plan.egress_network, name],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        ).returncode != 0:
            subprocess.run(
                ["docker", "rm", "-f", name],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )
            die(f"failed to attach mitmproxy sidecar {name} to egress "
                f"network {plan.egress_network}")

        if subprocess.run(
            ["docker", "start", name],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        ).returncode != 0:
            subprocess.run(
                ["docker", "rm", "-f", name],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )
            die(f"failed to start mitmproxy sidecar {name}")

        return name

    def stop(self, proxy_target: str) -> None:
        """Idempotent: missing container is success. Mirrors
        DockerPipelockProxy.stop."""
        if subprocess.run(
            ["docker", "inspect", proxy_target],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        ).returncode == 0:
            if subprocess.run(
                ["docker", "rm", "-f", proxy_target],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            ).returncode != 0:
                warn(
                    f"failed to remove mitmproxy sidecar {proxy_target}; "
                    f"clean up with 'docker rm -f {proxy_target}'"
                )

    def extract_ca_cert(self, proxy_target: str, dest_path: Path) -> None:
        """Poll the running sidecar for the CA cert (mitmproxy
        generates it on first launch, typically <1s after start),
        then `docker cp` the public half to `dest_path`. The private
        key never leaves the container."""
        deadline = time.monotonic() + 15
        while time.monotonic() < deadline:
            check = subprocess.run(
                ["docker", "exec", proxy_target, "test", "-f", _CA_PATH_IN_SIDECAR],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )
            if check.returncode == 0:
                break
            time.sleep(0.5)
        else:
            die(f"mitmproxy CA cert did not appear at {_CA_PATH_IN_SIDECAR} "
                f"after 15s — sidecar {proxy_target} may have failed to start")

        cp_result = subprocess.run(
            ["docker", "cp", f"{proxy_target}:{_CA_PATH_IN_SIDECAR}", str(dest_path)],
            capture_output=True,
            text=True,
            check=False,
        )
        if cp_result.returncode != 0:
            die(f"failed to extract mitmproxy CA cert from {proxy_target}: "
                f"{cp_result.stderr.strip()}")