From b2517910c7b39ae4feaaa42b8b44afa621b964c5 Mon Sep 17 00:00:00 2001 From: Matt Owen Date: Mon, 3 Aug 2026 19:09:14 -0400 Subject: [PATCH 1/3] Add Kubernetes sandbox backend for cwpro-managed deployments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduces a sandbox backend factory that dispatches between Docker (default) and Kubernetes, so cwpro deployments without Docker-in-Docker can run hunter agents as K8s Pods. New modules: - sandbox/factory.py: backend dispatcher (CLEARWING_SANDBOX_BACKEND env) - sandbox/kube_sandbox.py: KubeSandboxContainer — same interface as SandboxContainer but backed by K8s Pods with exec-based I/O - sandbox/kube_builder.py: on-cluster image building via Kaniko Jobs with content-hash tagging and registry push HunterSandbox changes: - build_image() short-circuits for k8s: resolves images from CLEARWING_SANDBOX_IMAGE override or builds on-cluster via Kaniko - spawn() uses create_sandbox() factory instead of direct SandboxContainer Also updates the sandbox-unavailable warning to reference k8s as an alternative to Docker. --- clearwing/sandbox/__init__.py | 3 + clearwing/sandbox/factory.py | 41 ++++ clearwing/sandbox/hunter_sandbox.py | 78 +++++- clearwing/sandbox/kube_builder.py | 272 +++++++++++++++++++++ clearwing/sandbox/kube_sandbox.py | 364 ++++++++++++++++++++++++++++ clearwing/sourcehunt/runner.py | 5 +- tests/test_kube_sandbox.py | 132 ++++++++++ 7 files changed, 891 insertions(+), 4 deletions(-) create mode 100644 clearwing/sandbox/factory.py create mode 100644 clearwing/sandbox/kube_builder.py create mode 100644 clearwing/sandbox/kube_sandbox.py create mode 100644 tests/test_kube_sandbox.py diff --git a/clearwing/sandbox/__init__.py b/clearwing/sandbox/__init__.py index 20449fa3..8fcf18fa 100644 --- a/clearwing/sandbox/__init__.py +++ b/clearwing/sandbox/__init__.py @@ -9,6 +9,7 @@ from .builders import BuildRecipe, BuildSystemDetector from .container import ExecResult, SandboxConfig, SandboxContainer from .dind import get_docker_client, get_docker_host, get_subprocess_env +from .factory import create_sandbox, is_kubernetes_backend from .hunter_sandbox import HunterSandbox from .registry import ContainerRegistry @@ -20,6 +21,8 @@ "BuildRecipe", "BuildSystemDetector", "HunterSandbox", + "create_sandbox", + "is_kubernetes_backend", "get_docker_client", "get_docker_host", "get_subprocess_env", diff --git a/clearwing/sandbox/factory.py b/clearwing/sandbox/factory.py new file mode 100644 index 00000000..f938a66c --- /dev/null +++ b/clearwing/sandbox/factory.py @@ -0,0 +1,41 @@ +"""Sandbox backend factory — dispatches between Docker and Kubernetes. + +The backend is selected by the ``CLEARWING_SANDBOX_BACKEND`` env var: + - ``docker`` (default): uses the local Docker daemon via docker-py. + - ``kubernetes``: creates Kubernetes Jobs that run the same sandbox image + as a Pod (for cwpro-managed deployments where DinD isn't available). + +Both backends implement the same SandboxContainer interface so hunter agents +are backend-agnostic. +""" + +from __future__ import annotations + +import logging +import os + +from .container import SandboxConfig, SandboxContainer + +logger = logging.getLogger(__name__) + +_BACKEND_ENV = "CLEARWING_SANDBOX_BACKEND" +_BACKEND_DOCKER = "docker" +_BACKEND_KUBERNETES = "kubernetes" + + +def is_kubernetes_backend() -> bool: + """True when the configured sandbox backend is Kubernetes.""" + return os.environ.get(_BACKEND_ENV, _BACKEND_DOCKER).lower() == _BACKEND_KUBERNETES + + +def create_sandbox(config: SandboxConfig) -> SandboxContainer: + """Create a sandbox container using the configured backend. + + For Docker (default): returns a standard SandboxContainer backed by docker-py. + For Kubernetes: returns a KubeSandboxContainer that runs as a K8s Job/Pod. + """ + if is_kubernetes_backend(): + from .kube_sandbox import KubeSandboxContainer + + return KubeSandboxContainer(config) + return SandboxContainer(config) diff --git a/clearwing/sandbox/hunter_sandbox.py b/clearwing/sandbox/hunter_sandbox.py index ffe5f202..9794bb7d 100644 --- a/clearwing/sandbox/hunter_sandbox.py +++ b/clearwing/sandbox/hunter_sandbox.py @@ -27,6 +27,7 @@ validate_sanitizer_combo, ) from .container import SandboxConfig, SandboxContainer +from .factory import create_sandbox, is_kubernetes_backend from .seccomp_profiles import get_seccomp_profile logger = logging.getLogger(__name__) @@ -187,7 +188,15 @@ def build_image(self) -> str: calls can pick between them without another build pass. MSan is the motivating case: it can't coexist with ASan in a single binary, so the caller declares it as an extra variant. + + When CLEARWING_SANDBOX_BACKEND=kubernetes, image builds are skipped — + images are expected to be pre-built and available in the cluster's + registry. The tag is computed from the Dockerfile content hash so the + same deterministic naming applies. """ + if is_kubernetes_backend(): + return self._resolve_kube_images() + primary_key = self._variant_key(self.sanitizers) primary_tag = self._build_variant_image(self.sanitizers) self._variant_images[primary_key] = primary_tag @@ -202,6 +211,71 @@ def build_image(self) -> str: return primary_tag + def _resolve_kube_images(self) -> str: + """For k8s backend: resolve image tags, building on-cluster if needed. + + Image resolution order: + 1. CLEARWING_SANDBOX_IMAGE env var (explicit override — skips builds) + 2. CLEARWING_SANDBOX_REGISTRY set → compute content-hash tag, build + on-cluster via Kaniko if image doesn't exist yet + 3. Neither set → error (cannot run without knowing which image to use) + """ + override = os.environ.get("CLEARWING_SANDBOX_IMAGE", "") + if override: + # Use the same override image for all variants + primary_key = self._variant_key(self.sanitizers) + self._variant_images[primary_key] = override + self._image_tag = override + for variant in self.extra_variants: + key = self._variant_key(variant) + self._variant_images[key] = override + logger.info("K8s sandbox: using override image %s", override) + return override + + # Production path: build on-cluster via Kaniko + from .kube_builder import compute_registry_tag + + primary_key = self._variant_key(self.sanitizers) + dockerfile = self._render_dockerfile(sanitizers=self.sanitizers) + primary_tag = compute_registry_tag( + dockerfile, self.sanitizers, self.extra_packages, self.post_install_commands + ) + primary_tag = self._kaniko_build_if_needed(dockerfile, primary_tag) + self._variant_images[primary_key] = primary_tag + self._image_tag = primary_tag + + for variant in self.extra_variants: + key = self._variant_key(variant) + if key == primary_key: + continue + df = self._render_dockerfile(sanitizers=variant) + tag = compute_registry_tag( + df, variant, self.extra_packages, self.post_install_commands + ) + tag = self._kaniko_build_if_needed(df, tag) + self._variant_images[key] = tag + + logger.info( + "K8s sandbox: resolved %d image tags (primary=%s)", + len(self._variant_images), + primary_tag, + ) + return primary_tag + + def _kaniko_build_if_needed(self, dockerfile_content: str, image_tag: str) -> str: + """Submit a Kaniko Job to build the image if it doesn't already exist. + + Returns the image_tag on success. + """ + from .kube_builder import build_image_on_cluster, image_exists_in_registry + + if image_exists_in_registry(image_tag): + logger.info("K8s sandbox: image %s already exists, skipping build", image_tag) + return image_tag + + logger.info("K8s sandbox: building image %s on-cluster via Kaniko", image_tag) + return build_image_on_cluster(dockerfile_content, image_tag) + def build_variant_images(self) -> dict[str, str]: """Build every declared variant. Returns {variant_key: image_tag}. @@ -277,7 +351,7 @@ def _build_variant_image(self, sanitizers: list[str]) -> str: if proc.returncode != 0: raise RuntimeError(proc.stderr[-2000:] or proc.stdout[-2000:]) except subprocess.TimeoutExpired as e: - raise RuntimeError(f"Sandbox image build timed out after 300s") from e + raise RuntimeError("Sandbox image build timed out after 300s") from e except RuntimeError: raise except Exception as e: @@ -379,7 +453,7 @@ def spawn( runtime=runtime, ) - sb = SandboxContainer(cfg) + sb = create_sandbox(cfg) sb.start() if writable_workspace: diff --git a/clearwing/sandbox/kube_builder.py b/clearwing/sandbox/kube_builder.py new file mode 100644 index 00000000..bc57ded3 --- /dev/null +++ b/clearwing/sandbox/kube_builder.py @@ -0,0 +1,272 @@ +"""On-cluster image building via Kaniko for Kubernetes sandbox backend. + +When CLEARWING_SANDBOX_BACKEND=kubernetes, sandbox images can't be built via +the local Docker daemon (there isn't one). Instead we submit a Kaniko Job that +builds + pushes the image to the cluster's registry. + +Environment variables: + CLEARWING_SANDBOX_REGISTRY — required registry prefix (e.g. "ghcr.io/org/clearwing-sandbox") + CLEARWING_SANDBOX_NAMESPACE — k8s namespace for build jobs (default: current pod namespace) + CLEARWING_KANIKO_IMAGE — kaniko executor image (default: gcr.io/kaniko-project/executor:latest) + CLEARWING_KANIKO_SERVICE_ACCOUNT — SA for build jobs (default: "default") +""" + +from __future__ import annotations + +import hashlib +import logging +import os +import time + +logger = logging.getLogger(__name__) + +_REGISTRY_ENV = "CLEARWING_SANDBOX_REGISTRY" +_NAMESPACE_ENV = "CLEARWING_SANDBOX_NAMESPACE" +_KANIKO_IMAGE_ENV = "CLEARWING_KANIKO_IMAGE" +_KANIKO_SA_ENV = "CLEARWING_KANIKO_SERVICE_ACCOUNT" + +_DEFAULT_KANIKO_IMAGE = "gcr.io/kaniko-project/executor:latest" +_DEFAULT_SA = "default" +_BUILD_TIMEOUT_SECONDS = 600 +_POLL_INTERVAL_SECONDS = 5 + + +def _registry() -> str: + registry = os.environ.get(_REGISTRY_ENV, "") + if not registry: + raise RuntimeError( + f"{_REGISTRY_ENV} must be set when using Kubernetes sandbox backend " + "without CLEARWING_SANDBOX_IMAGE override" + ) + return registry.rstrip("/") + + +def _namespace() -> str: + ns = os.environ.get(_NAMESPACE_ENV) + if ns: + return ns + # Fall back to the current pod's namespace + try: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f: + return f.read().strip() + except OSError: + return "default" + + +def compute_registry_tag( + dockerfile: str, + sanitizers: list[str], + extra_packages: list[str], + post_install_commands: list[str], +) -> str: + """Compute a deterministic registry image tag from content hash. + + Same hashing logic as HunterSandbox._compute_tag but prefixed with the + configured registry so it's a fully qualified image reference. + """ + h = hashlib.sha256() + h.update(dockerfile.encode("utf-8")) + h.update(",".join(sorted(sanitizers)).encode("utf-8")) + h.update(",".join(sorted(extra_packages)).encode("utf-8")) + h.update("\n".join(post_install_commands).encode("utf-8")) + digest = h.hexdigest()[:12] + return f"{_registry()}:{digest}" + + +def image_exists_in_registry(image_tag: str) -> bool: + """Check if the image tag already exists in the registry. + + Uses the Kubernetes Python client to attempt a dry-run pull via a + short-lived Job, or queries the registry API directly if credentials + are available. For simplicity, we use `skopeo inspect` semantics via + a lightweight crane check when available, falling back to assuming + the image doesn't exist (which just triggers a rebuild). + """ + try: + import subprocess + + result = subprocess.run( + ["crane", "manifest", image_tag], + capture_output=True, + timeout=30, + ) + return result.returncode == 0 + except (FileNotFoundError, subprocess.TimeoutExpired): + # crane not available or timed out — check via k8s API + pass + + try: + from kubernetes import config + + try: + config.load_incluster_config() + except config.ConfigException: + config.load_kube_config() + + # Try to create a short-lived pod that just pulls the image + # This is expensive; prefer crane. For now, assume not cached. + logger.debug( + "crane not available; assuming image %s needs building", image_tag + ) + return False + except Exception: + return False + + +def build_image_on_cluster(dockerfile_content: str, image_tag: str) -> str: + """Submit a Kaniko Job to build and push the image on-cluster. + + Creates a ConfigMap with the Dockerfile, then a Job that mounts it and + runs Kaniko to build + push to the registry. Blocks until the Job + completes or times out. + + Returns the image_tag on success; raises RuntimeError on failure. + """ + from kubernetes import client, config + + try: + config.load_incluster_config() + except config.ConfigException: + config.load_kube_config() + + namespace = _namespace() + kaniko_image = os.environ.get(_KANIKO_IMAGE_ENV, _DEFAULT_KANIKO_IMAGE) + service_account = os.environ.get(_KANIKO_SA_ENV, _DEFAULT_SA) + + # Unique name for this build + tag_hash = image_tag.rsplit(":", 1)[-1] if ":" in image_tag else "build" + job_name = f"clearwing-build-{tag_hash}" + configmap_name = f"clearwing-dockerfile-{tag_hash}" + + core_v1 = client.CoreV1Api() + batch_v1 = client.BatchV1Api() + + # Create ConfigMap with the Dockerfile + configmap = client.V1ConfigMap( + metadata=client.V1ObjectMeta( + name=configmap_name, + namespace=namespace, + labels={"managed-by": "clearwing", "purpose": "sandbox-build"}, + ), + data={"Dockerfile": dockerfile_content}, + ) + try: + core_v1.create_namespaced_config_map(namespace, configmap) + except client.ApiException as e: + if e.status == 409: + # Already exists — replace it + core_v1.replace_namespaced_config_map(configmap_name, namespace, configmap) + else: + raise + + # Build the Kaniko Job + job = client.V1Job( + metadata=client.V1ObjectMeta( + name=job_name, + namespace=namespace, + labels={"managed-by": "clearwing", "purpose": "sandbox-build"}, + ), + spec=client.V1JobSpec( + ttl_seconds_after_finished=300, + backoff_limit=0, + template=client.V1PodTemplateSpec( + spec=client.V1PodSpec( + service_account_name=service_account, + restart_policy="Never", + containers=[ + client.V1Container( + name="kaniko", + image=kaniko_image, + args=[ + "--dockerfile=/workspace/Dockerfile", + "--context=dir:///workspace", + f"--destination={image_tag}", + "--cache=true", + "--single-snapshot", + ], + volume_mounts=[ + client.V1VolumeMount( + name="dockerfile", + mount_path="/workspace", + read_only=True, + ), + ], + ), + ], + volumes=[ + client.V1Volume( + name="dockerfile", + config_map=client.V1ConfigMapVolumeSource( + name=configmap_name, + ), + ), + ], + ), + ), + ), + ) + + try: + batch_v1.create_namespaced_job(namespace, job) + except client.ApiException as e: + if e.status == 409: + # Job already exists — delete and recreate + batch_v1.delete_namespaced_job( + job_name, namespace, propagation_policy="Background" + ) + time.sleep(2) + batch_v1.create_namespaced_job(namespace, job) + else: + raise + + # Poll until completion + logger.info("Waiting for Kaniko build job %s/%s", namespace, job_name) + deadline = time.monotonic() + _BUILD_TIMEOUT_SECONDS + while time.monotonic() < deadline: + status = batch_v1.read_namespaced_job_status(job_name, namespace).status + if status.succeeded and status.succeeded > 0: + logger.info("Kaniko build succeeded: %s", image_tag) + _cleanup_build_resources(core_v1, batch_v1, namespace, job_name, configmap_name) + return image_tag + if status.failed and status.failed > 0: + # Try to get logs for debugging + logs = _get_build_logs(core_v1, namespace, job_name) + _cleanup_build_resources(core_v1, batch_v1, namespace, job_name, configmap_name) + raise RuntimeError( + f"Kaniko build failed for {image_tag}. Logs:\n{logs[-2000:]}" + ) + time.sleep(_POLL_INTERVAL_SECONDS) + + _cleanup_build_resources(core_v1, batch_v1, namespace, job_name, configmap_name) + raise RuntimeError( + f"Kaniko build timed out after {_BUILD_TIMEOUT_SECONDS}s for {image_tag}" + ) + + +def _get_build_logs(core_v1, namespace: str, job_name: str) -> str: + """Best-effort retrieval of build pod logs.""" + try: + pods = core_v1.list_namespaced_pod( + namespace, label_selector=f"job-name={job_name}" + ) + if pods.items: + return core_v1.read_namespaced_pod_log( + pods.items[0].metadata.name, namespace, container="kaniko" + ) + except Exception: + pass + return "(logs unavailable)" + + +def _cleanup_build_resources(core_v1, batch_v1, namespace: str, job_name: str, configmap_name: str) -> None: + """Best-effort cleanup of build Job and ConfigMap.""" + try: + batch_v1.delete_namespaced_job( + job_name, namespace, propagation_policy="Background" + ) + except Exception: + pass + try: + core_v1.delete_namespaced_config_map(configmap_name, namespace) + except Exception: + pass diff --git a/clearwing/sandbox/kube_sandbox.py b/clearwing/sandbox/kube_sandbox.py new file mode 100644 index 00000000..01f6cc42 --- /dev/null +++ b/clearwing/sandbox/kube_sandbox.py @@ -0,0 +1,364 @@ +"""Kubernetes-backed SandboxContainer implementation. + +Runs sandbox workloads as Kubernetes Jobs/Pods instead of local Docker +containers. Implements the same interface as SandboxContainer so hunter +agents are backend-agnostic. + +Requires: CLEARWING_SANDBOX_BACKEND=kubernetes +""" + +from __future__ import annotations + +import io +import logging +import os +import tarfile +import time +import uuid + +from .container import ExecResult, SandboxConfig, SandboxContainer + +logger = logging.getLogger(__name__) + +_NAMESPACE_ENV = "CLEARWING_SANDBOX_NAMESPACE" +_EXEC_TIMEOUT_SECONDS = 600 +_POD_STARTUP_TIMEOUT_SECONDS = 120 + + +def _namespace() -> str: + ns = os.environ.get(_NAMESPACE_ENV) + if ns: + return ns + try: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f: + return f.read().strip() + except OSError: + return "default" + + +class KubeSandboxContainer(SandboxContainer): + """SandboxContainer backed by a Kubernetes Pod. + + Creates a long-running Pod (not a Job) so we can exec into it multiple + times, matching the Docker container lifecycle model. + """ + + def __init__(self, config: SandboxConfig): + # Don't call super().__init__ — we override the full lifecycle + self._config = config + self._pod_name: str | None = None + self._namespace = _namespace() + self._core_v1 = None + self._started = False + # Public attributes expected by HunterSandbox + self.scratch_host_dir: str | None = None + self.variant: list[str] = [] + + def _api(self): + if self._core_v1 is None: + from kubernetes import client, config + + try: + config.load_incluster_config() + except config.ConfigException: + config.load_kube_config() + self._core_v1 = client.CoreV1Api() + return self._core_v1 + + @property + def container_id(self) -> str: + return self._pod_name or "" + + @property + def short_id(self) -> str: + return (self._pod_name or "")[:12] + + @property + def is_running(self) -> bool: + if not self._pod_name: + return False + try: + pod = self._api().read_namespaced_pod_status(self._pod_name, self._namespace) + return pod.status.phase == "Running" + except Exception: + return False + + def start(self) -> str: + """Create and start a Pod matching the SandboxConfig.""" + from kubernetes import client + + api = self._api() + cfg = self._config + pod_id = uuid.uuid4().hex[:8] + self._pod_name = f"clearwing-sandbox-{pod_id}" + + # Build env vars + env_vars = [ + client.V1EnvVar(name=k, value=v) for k, v in cfg.env.items() + ] + + # Resource limits + resources = client.V1ResourceRequirements( + limits={ + "memory": f"{cfg.memory_mb}Mi", + }, + requests={ + "memory": f"{min(cfg.memory_mb, 512)}Mi", + }, + ) + if cfg.cpus > 0: + resources.limits["cpu"] = str(cfg.cpus) + resources.requests["cpu"] = str(min(cfg.cpus, 0.5)) + + # Security context matching Docker config + security_context = client.V1SecurityContext( + run_as_non_root=False, + read_only_root_filesystem=cfg.read_only_rootfs, + capabilities=client.V1Capabilities( + drop=cfg.cap_drop, + add=cfg.cap_add, + ), + ) + + container = client.V1Container( + name="sandbox", + image=cfg.image, + command=["sleep", "infinity"], + working_dir=cfg.working_dir, + env=env_vars, + resources=resources, + security_context=security_context, + ) + + # Pod-level security and network policy + pod_spec = client.V1PodSpec( + containers=[container], + restart_policy="Never", + # Network isolation: use a NetworkPolicy on the namespace rather + # than Docker's network_mode=none. The pod itself is created + # without hostNetwork so default namespace policies apply. + host_network=False, + automount_service_account_token=False, + ) + + pod = client.V1Pod( + metadata=client.V1ObjectMeta( + name=self._pod_name, + namespace=self._namespace, + labels={ + "managed-by": "clearwing", + "purpose": "sandbox", + "clearwing-session": cfg.env.get("CLEARWING_SESSION_ID", "unknown"), + }, + ), + spec=pod_spec, + ) + + api.create_namespaced_pod(self._namespace, pod) + self._wait_for_running() + self._started = True + logger.debug("K8s sandbox pod %s running", self._pod_name) + return self._pod_name + + def _wait_for_running(self) -> None: + """Block until the Pod reaches Running phase.""" + from kubernetes import watch + + api = self._api() + deadline = time.monotonic() + _POD_STARTUP_TIMEOUT_SECONDS + w = watch.Watch() + try: + for event in w.stream( + api.list_namespaced_pod, + self._namespace, + field_selector=f"metadata.name={self._pod_name}", + timeout_seconds=_POD_STARTUP_TIMEOUT_SECONDS, + ): + pod = event["object"] + if pod.status.phase == "Running": + return + if pod.status.phase in ("Failed", "Succeeded"): + raise RuntimeError( + f"Sandbox pod {self._pod_name} entered {pod.status.phase}" + ) + if time.monotonic() > deadline: + break + finally: + w.stop() + raise RuntimeError( + f"Sandbox pod {self._pod_name} did not reach Running within " + f"{_POD_STARTUP_TIMEOUT_SECONDS}s" + ) + + def exec( + self, + command: str | list[str], + timeout: int | None = None, + env: dict[str, str] | None = None, + workdir: str | None = None, + ) -> ExecResult: + """Execute a command inside the sandbox Pod via kubectl exec.""" + from kubernetes.stream import stream + + if timeout is None: + timeout = self._config.timeout_seconds + + if isinstance(command, str): + exec_command = ["/bin/sh", "-c", command] + else: + exec_command = list(command) + + # Prepend env vars and workdir if specified + if env or workdir: + shell_prefix = "" + if workdir: + shell_prefix += f"cd {workdir} && " + if env: + exports = " ".join(f"{k}={v}" for k, v in env.items()) + shell_prefix += f"export {exports} && " + if isinstance(command, str): + exec_command = ["/bin/sh", "-c", f"{shell_prefix}{command}"] + else: + joined = " ".join(command) + exec_command = ["/bin/sh", "-c", f"{shell_prefix}{joined}"] + + started = time.monotonic() + timed_out = False + try: + # Use the kubernetes python client's exec + resp = stream( + self._api().connect_get_namespaced_pod_exec, + self._pod_name, + self._namespace, + container="sandbox", + command=exec_command, + stderr=True, + stdout=True, + stdin=False, + tty=False, + _preload_content=True, + _request_timeout=timeout, + ) + # stream() returns the combined output as a string for _preload_content=True + stdout = resp if isinstance(resp, str) else "" + stderr = "" + exit_code = 0 + except Exception as exc: + elapsed = time.monotonic() - started + if elapsed >= timeout: + timed_out = True + stdout = "" + stderr = f"Command timed out after {timeout}s" + exit_code = 124 + else: + stdout = "" + stderr = str(exc) + exit_code = 1 + + duration = time.monotonic() - started + return ExecResult( + exit_code=exit_code, + stdout=stdout, + stderr=stderr, + duration_seconds=duration, + timed_out=timed_out, + ) + + def write_file(self, container_path: str, content: bytes) -> None: + """Write a file into the sandbox Pod.""" + # Create a tar archive in memory and pipe it via exec + buf = io.BytesIO() + with tarfile.open(fileobj=buf, mode="w") as tar: + info = tarfile.TarInfo(name=os.path.basename(container_path)) + info.size = len(content) + tar.addfile(info, io.BytesIO(content)) + buf.seek(0) + + dest_dir = os.path.dirname(container_path) + self.exec(f"mkdir -p {dest_dir}") + + # Write via base64 encoding to avoid binary transfer issues + import base64 + + encoded = base64.b64encode(buf.getvalue()).decode("ascii") + self.exec( + f"echo '{encoded}' | base64 -d | tar -xf - -C {dest_dir}" + ) + + def read_file(self, container_path: str) -> bytes: + """Read a file from the sandbox Pod.""" + import base64 + + result = self.exec(f"base64 < {container_path}") + if result.exit_code != 0: + raise FileNotFoundError( + f"Cannot read {container_path}: {result.stderr}" + ) + return base64.b64decode(result.stdout.strip()) + + def copy_tree_into(self, host_path: str, container_path: str) -> None: + """Copy a directory tree into the Pod. + + Creates a tar of the host path and streams it into the container. + """ + import subprocess + + self.exec(f"mkdir -p {container_path}") + + # Create tar locally, base64 encode, exec into pod + result = subprocess.run( + ["tar", "-cf", "-", "-C", host_path, "."], + capture_output=True, + timeout=120, + ) + if result.returncode != 0: + raise RuntimeError(f"Failed to tar {host_path}: {result.stderr.decode()}") + + import base64 + + # For large trees, chunk the transfer + encoded = base64.b64encode(result.stdout).decode("ascii") + chunk_size = 500_000 # ~375KB decoded per chunk + if len(encoded) <= chunk_size: + self.exec( + f"echo '{encoded}' | base64 -d | tar -xf - --no-same-owner -C {container_path}" + ) + else: + # Write chunks to a temp file in the container + self.exec("rm -f /tmp/_transfer.tar.b64") + for i in range(0, len(encoded), chunk_size): + chunk = encoded[i : i + chunk_size] + self.exec(f"echo -n '{chunk}' >> /tmp/_transfer.tar.b64") + self.exec( + f"base64 -d /tmp/_transfer.tar.b64 | tar -xf - --no-same-owner -C {container_path} && " + f"rm -f /tmp/_transfer.tar.b64" + ) + + def stop(self) -> None: + """Delete the sandbox Pod.""" + if not self._pod_name: + return + try: + self._api().delete_namespaced_pod( + self._pod_name, + self._namespace, + grace_period_seconds=5, + ) + logger.debug("K8s sandbox pod %s deleted", self._pod_name) + except Exception: + logger.debug("Failed to delete sandbox pod %s", self._pod_name, exc_info=True) + self._started = False + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.stop() + return False + + def __del__(self): + if self._started: + try: + self.stop() + except Exception: + pass diff --git a/clearwing/sourcehunt/runner.py b/clearwing/sourcehunt/runner.py index 92659f7c..1d884d82 100644 --- a/clearwing/sourcehunt/runner.py +++ b/clearwing/sourcehunt/runner.py @@ -2656,8 +2656,9 @@ def _ensure_sandbox_factory(self, repo_path: str, files: list[FileTarget]) -> No image_tag = manager.build_image() except Exception as exc: logger.warning( - "HunterSandbox unavailable (%s); falling back to host mode. " - "Start Docker to enable sanitizer-backed containers.", + "HunterSandbox unavailable (%s); falling back to constrained " + "host-source tools. Start Docker (or configure Kubernetes " + "sandbox backend) to enable sanitizer-backed containers.", exc, ) logger.debug("HunterSandbox initialization failed", exc_info=True) diff --git a/tests/test_kube_sandbox.py b/tests/test_kube_sandbox.py new file mode 100644 index 00000000..f25b2530 --- /dev/null +++ b/tests/test_kube_sandbox.py @@ -0,0 +1,132 @@ +"""Unit tests for Kubernetes sandbox backend support.""" + +import os +import unittest +from unittest.mock import MagicMock, patch + +from clearwing.sandbox.container import SandboxConfig +from clearwing.sandbox.factory import create_sandbox, is_kubernetes_backend + + +class IsKubernetesBackendTests(unittest.TestCase): + def test_default_is_docker(self): + with patch.dict(os.environ, {}, clear=True): + os.environ.pop("CLEARWING_SANDBOX_BACKEND", None) + self.assertFalse(is_kubernetes_backend()) + + def test_explicit_docker(self): + with patch.dict(os.environ, {"CLEARWING_SANDBOX_BACKEND": "docker"}): + self.assertFalse(is_kubernetes_backend()) + + def test_kubernetes_detected(self): + with patch.dict(os.environ, {"CLEARWING_SANDBOX_BACKEND": "kubernetes"}): + self.assertTrue(is_kubernetes_backend()) + + def test_case_insensitive(self): + with patch.dict(os.environ, {"CLEARWING_SANDBOX_BACKEND": "Kubernetes"}): + self.assertTrue(is_kubernetes_backend()) + + +class CreateSandboxFactoryTests(unittest.TestCase): + def test_docker_backend_returns_sandbox_container(self): + from clearwing.sandbox.container import SandboxContainer + + with patch.dict(os.environ, {"CLEARWING_SANDBOX_BACKEND": "docker"}): + cfg = SandboxConfig(image="test:latest") + # Don't actually start it — just verify the type + with patch.object(SandboxContainer, "start"): + sb = create_sandbox(cfg) + self.assertIsInstance(sb, SandboxContainer) + + def test_kubernetes_backend_returns_kube_container(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + with patch.dict(os.environ, {"CLEARWING_SANDBOX_BACKEND": "kubernetes"}): + cfg = SandboxConfig(image="test:latest") + sb = create_sandbox(cfg) + self.assertIsInstance(sb, KubeSandboxContainer) + + +class HunterSandboxKubeImageResolutionTests(unittest.TestCase): + """Test the k8s image resolution paths in HunterSandbox.""" + + @patch.dict(os.environ, { + "CLEARWING_SANDBOX_BACKEND": "kubernetes", + "CLEARWING_SANDBOX_IMAGE": "registry.example/sandbox:override", + }) + def test_override_image_used_for_all_variants(self): + from clearwing.sandbox.hunter_sandbox import HunterSandbox + + with patch.object(HunterSandbox, "__init__", lambda self, **kw: None): + sandbox = HunterSandbox.__new__(HunterSandbox) + sandbox.sanitizers = ["asan", "ubsan"] + sandbox.extra_variants = [["msan"]] + sandbox.extra_packages = [] + sandbox.post_install_commands = [] + sandbox._variant_images = {} + sandbox._image_tag = None + + tag = sandbox._resolve_kube_images() + + self.assertEqual(tag, "registry.example/sandbox:override") + self.assertEqual(sandbox._image_tag, "registry.example/sandbox:override") + # All variants should map to the override image + for key, img in sandbox._variant_images.items(): + self.assertEqual(img, "registry.example/sandbox:override") + + @patch.dict(os.environ, { + "CLEARWING_SANDBOX_BACKEND": "kubernetes", + "CLEARWING_SANDBOX_REGISTRY": "ghcr.io/org/clearwing-sandbox", + }) + @patch("clearwing.sandbox.hunter_sandbox.HunterSandbox._kaniko_build_if_needed") + def test_registry_builds_via_kaniko(self, mock_build): + from clearwing.sandbox.hunter_sandbox import HunterSandbox + + mock_build.side_effect = lambda df, tag: tag + + with patch.object(HunterSandbox, "__init__", lambda self, **kw: None): + sandbox = HunterSandbox.__new__(HunterSandbox) + sandbox.sanitizers = ["asan", "ubsan"] + sandbox.extra_variants = [] + sandbox.extra_packages = ["python3"] + sandbox.post_install_commands = [] + sandbox._variant_images = {} + sandbox._image_tag = None + sandbox.build_recipe = MagicMock() + sandbox.build_recipe.base_image = "debian:11-slim" + sandbox.build_recipe.apt_packages = ["build-essential"] + sandbox._optional_packages = [] + + # Need _render_dockerfile to work + with patch.object(sandbox, "_render_dockerfile", return_value="FROM debian:11\n"): + # Pop the CLEARWING_SANDBOX_IMAGE to avoid override path + os.environ.pop("CLEARWING_SANDBOX_IMAGE", None) + tag = sandbox._resolve_kube_images() + + self.assertTrue(tag.startswith("ghcr.io/org/clearwing-sandbox:")) + self.assertEqual(mock_build.call_count, 1) + + +class KubeBuilderTagTests(unittest.TestCase): + """Test the content-hash tag computation.""" + + @patch.dict(os.environ, {"CLEARWING_SANDBOX_REGISTRY": "ghcr.io/org/sandbox"}) + def test_deterministic_tag(self): + from clearwing.sandbox.kube_builder import compute_registry_tag + + tag1 = compute_registry_tag("FROM debian\n", ["asan"], ["git"], ["echo hello"]) + tag2 = compute_registry_tag("FROM debian\n", ["asan"], ["git"], ["echo hello"]) + self.assertEqual(tag1, tag2) + self.assertTrue(tag1.startswith("ghcr.io/org/sandbox:")) + + @patch.dict(os.environ, {"CLEARWING_SANDBOX_REGISTRY": "ghcr.io/org/sandbox"}) + def test_different_content_different_tag(self): + from clearwing.sandbox.kube_builder import compute_registry_tag + + tag1 = compute_registry_tag("FROM debian\n", ["asan"], ["git"], []) + tag2 = compute_registry_tag("FROM ubuntu\n", ["asan"], ["git"], []) + self.assertNotEqual(tag1, tag2) + + +if __name__ == "__main__": + unittest.main() From 525d865c6eb7cd8878e896070e93db5eac29699f Mon Sep 17 00:00:00 2001 From: Matt Owen Date: Mon, 3 Aug 2026 19:38:25 -0400 Subject: [PATCH 2/3] Harden k8s sandbox: proper exec exit codes, stdin file transfer, GC safety MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes critical issues identified during code review: - exec() now uses WSClient with separate channels + parses the real exit code from channel 3 status JSON (was silently returning 0 for everything) - write_file/copy_tree_into pipe tar directly via stdin instead of base64-in-shell (no arg size limits, no injection risk, single round-trip) - Sandbox pods get activeDeadlineSeconds (3x timeout, floor 15min) so zombies self-terminate if our process crashes - ownerReference set to parent pod when running in-cluster — k8s GC cascades deletion on parent crash - Extracted shared kube_client.py (cached ApiClient, namespace(), owner ref helper) — eliminates duplicated config loading and repeated token file reads --- clearwing/sandbox/kube_builder.py | 105 ++++------- clearwing/sandbox/kube_client.py | 98 +++++++++++ clearwing/sandbox/kube_sandbox.py | 284 ++++++++++++++++++------------ tests/test_kube_sandbox.py | 131 +++++++++++++- 4 files changed, 427 insertions(+), 191 deletions(-) create mode 100644 clearwing/sandbox/kube_client.py diff --git a/clearwing/sandbox/kube_builder.py b/clearwing/sandbox/kube_builder.py index bc57ded3..b208f558 100644 --- a/clearwing/sandbox/kube_builder.py +++ b/clearwing/sandbox/kube_builder.py @@ -6,7 +6,6 @@ Environment variables: CLEARWING_SANDBOX_REGISTRY — required registry prefix (e.g. "ghcr.io/org/clearwing-sandbox") - CLEARWING_SANDBOX_NAMESPACE — k8s namespace for build jobs (default: current pod namespace) CLEARWING_KANIKO_IMAGE — kaniko executor image (default: gcr.io/kaniko-project/executor:latest) CLEARWING_KANIKO_SERVICE_ACCOUNT — SA for build jobs (default: "default") """ @@ -18,10 +17,11 @@ import os import time +from .kube_client import batch_v1_api, core_v1_api, namespace + logger = logging.getLogger(__name__) _REGISTRY_ENV = "CLEARWING_SANDBOX_REGISTRY" -_NAMESPACE_ENV = "CLEARWING_SANDBOX_NAMESPACE" _KANIKO_IMAGE_ENV = "CLEARWING_KANIKO_IMAGE" _KANIKO_SA_ENV = "CLEARWING_KANIKO_SERVICE_ACCOUNT" @@ -41,18 +41,6 @@ def _registry() -> str: return registry.rstrip("/") -def _namespace() -> str: - ns = os.environ.get(_NAMESPACE_ENV) - if ns: - return ns - # Fall back to the current pod's namespace - try: - with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f: - return f.read().strip() - except OSError: - return "default" - - def compute_registry_tag( dockerfile: str, sanitizers: list[str], @@ -76,11 +64,9 @@ def compute_registry_tag( def image_exists_in_registry(image_tag: str) -> bool: """Check if the image tag already exists in the registry. - Uses the Kubernetes Python client to attempt a dry-run pull via a - short-lived Job, or queries the registry API directly if credentials - are available. For simplicity, we use `skopeo inspect` semantics via - a lightweight crane check when available, falling back to assuming - the image doesn't exist (which just triggers a rebuild). + Tries `crane manifest` first (fast, no k8s API needed). Falls back to + assuming the image doesn't exist (triggering a rebuild) when crane is + not available. """ try: import subprocess @@ -92,25 +78,10 @@ def image_exists_in_registry(image_tag: str) -> bool: ) return result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired): - # crane not available or timed out — check via k8s API pass - try: - from kubernetes import config - - try: - config.load_incluster_config() - except config.ConfigException: - config.load_kube_config() - - # Try to create a short-lived pod that just pulls the image - # This is expensive; prefer crane. For now, assume not cached. - logger.debug( - "crane not available; assuming image %s needs building", image_tag - ) - return False - except Exception: - return False + logger.debug("crane not available; assuming image %s needs building", image_tag) + return False def build_image_on_cluster(dockerfile_content: str, image_tag: str) -> str: @@ -122,14 +93,9 @@ def build_image_on_cluster(dockerfile_content: str, image_tag: str) -> str: Returns the image_tag on success; raises RuntimeError on failure. """ - from kubernetes import client, config - - try: - config.load_incluster_config() - except config.ConfigException: - config.load_kube_config() + from kubernetes import client - namespace = _namespace() + ns = namespace() kaniko_image = os.environ.get(_KANIKO_IMAGE_ENV, _DEFAULT_KANIKO_IMAGE) service_account = os.environ.get(_KANIKO_SA_ENV, _DEFAULT_SA) @@ -138,24 +104,23 @@ def build_image_on_cluster(dockerfile_content: str, image_tag: str) -> str: job_name = f"clearwing-build-{tag_hash}" configmap_name = f"clearwing-dockerfile-{tag_hash}" - core_v1 = client.CoreV1Api() - batch_v1 = client.BatchV1Api() + core = core_v1_api() + batch = batch_v1_api() # Create ConfigMap with the Dockerfile configmap = client.V1ConfigMap( metadata=client.V1ObjectMeta( name=configmap_name, - namespace=namespace, + namespace=ns, labels={"managed-by": "clearwing", "purpose": "sandbox-build"}, ), data={"Dockerfile": dockerfile_content}, ) try: - core_v1.create_namespaced_config_map(namespace, configmap) + core.create_namespaced_config_map(ns, configmap) except client.ApiException as e: if e.status == 409: - # Already exists — replace it - core_v1.replace_namespaced_config_map(configmap_name, namespace, configmap) + core.replace_namespaced_config_map(configmap_name, ns, configmap) else: raise @@ -163,12 +128,13 @@ def build_image_on_cluster(dockerfile_content: str, image_tag: str) -> str: job = client.V1Job( metadata=client.V1ObjectMeta( name=job_name, - namespace=namespace, + namespace=ns, labels={"managed-by": "clearwing", "purpose": "sandbox-build"}, ), spec=client.V1JobSpec( ttl_seconds_after_finished=300, backoff_limit=0, + active_deadline_seconds=_BUILD_TIMEOUT_SECONDS, template=client.V1PodTemplateSpec( spec=client.V1PodSpec( service_account_name=service_account, @@ -207,66 +173,61 @@ def build_image_on_cluster(dockerfile_content: str, image_tag: str) -> str: ) try: - batch_v1.create_namespaced_job(namespace, job) + batch.create_namespaced_job(ns, job) except client.ApiException as e: if e.status == 409: # Job already exists — delete and recreate - batch_v1.delete_namespaced_job( - job_name, namespace, propagation_policy="Background" + batch.delete_namespaced_job( + job_name, ns, propagation_policy="Background" ) time.sleep(2) - batch_v1.create_namespaced_job(namespace, job) + batch.create_namespaced_job(ns, job) else: raise # Poll until completion - logger.info("Waiting for Kaniko build job %s/%s", namespace, job_name) + logger.info("Waiting for Kaniko build job %s/%s", ns, job_name) deadline = time.monotonic() + _BUILD_TIMEOUT_SECONDS while time.monotonic() < deadline: - status = batch_v1.read_namespaced_job_status(job_name, namespace).status + status = batch.read_namespaced_job_status(job_name, ns).status if status.succeeded and status.succeeded > 0: logger.info("Kaniko build succeeded: %s", image_tag) - _cleanup_build_resources(core_v1, batch_v1, namespace, job_name, configmap_name) + _cleanup_build_resources(core, batch, ns, job_name, configmap_name) return image_tag if status.failed and status.failed > 0: - # Try to get logs for debugging - logs = _get_build_logs(core_v1, namespace, job_name) - _cleanup_build_resources(core_v1, batch_v1, namespace, job_name, configmap_name) + logs = _get_build_logs(core, ns, job_name) + _cleanup_build_resources(core, batch, ns, job_name, configmap_name) raise RuntimeError( f"Kaniko build failed for {image_tag}. Logs:\n{logs[-2000:]}" ) time.sleep(_POLL_INTERVAL_SECONDS) - _cleanup_build_resources(core_v1, batch_v1, namespace, job_name, configmap_name) + _cleanup_build_resources(core, batch, ns, job_name, configmap_name) raise RuntimeError( f"Kaniko build timed out after {_BUILD_TIMEOUT_SECONDS}s for {image_tag}" ) -def _get_build_logs(core_v1, namespace: str, job_name: str) -> str: +def _get_build_logs(core, ns: str, job_name: str) -> str: """Best-effort retrieval of build pod logs.""" try: - pods = core_v1.list_namespaced_pod( - namespace, label_selector=f"job-name={job_name}" - ) + pods = core.list_namespaced_pod(ns, label_selector=f"job-name={job_name}") if pods.items: - return core_v1.read_namespaced_pod_log( - pods.items[0].metadata.name, namespace, container="kaniko" + return core.read_namespaced_pod_log( + pods.items[0].metadata.name, ns, container="kaniko" ) except Exception: pass return "(logs unavailable)" -def _cleanup_build_resources(core_v1, batch_v1, namespace: str, job_name: str, configmap_name: str) -> None: +def _cleanup_build_resources(core, batch, ns: str, job_name: str, configmap_name: str) -> None: """Best-effort cleanup of build Job and ConfigMap.""" try: - batch_v1.delete_namespaced_job( - job_name, namespace, propagation_policy="Background" - ) + batch.delete_namespaced_job(job_name, ns, propagation_policy="Background") except Exception: pass try: - core_v1.delete_namespaced_config_map(configmap_name, namespace) + core.delete_namespaced_config_map(configmap_name, ns) except Exception: pass diff --git a/clearwing/sandbox/kube_client.py b/clearwing/sandbox/kube_client.py new file mode 100644 index 00000000..1798ad6c --- /dev/null +++ b/clearwing/sandbox/kube_client.py @@ -0,0 +1,98 @@ +"""Shared Kubernetes client utilities for the sandbox backend. + +Provides a cached API client and common helpers used by both +kube_sandbox.py (pod lifecycle) and kube_builder.py (Kaniko builds). +""" + +from __future__ import annotations + +import logging +import os + +logger = logging.getLogger(__name__) + +_NAMESPACE_ENV = "CLEARWING_SANDBOX_NAMESPACE" + +# Module-level cached client — initialized once on first use. +_api_client = None + + +def namespace() -> str: + """Resolve the Kubernetes namespace for sandbox resources. + + Priority: + 1. CLEARWING_SANDBOX_NAMESPACE env var + 2. In-cluster service account namespace file + 3. "default" + """ + ns = os.environ.get(_NAMESPACE_ENV) + if ns: + return ns + try: + with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f: + return f.read().strip() + except OSError: + return "default" + + +def get_api_client(): + """Return a cached kubernetes ApiClient, loading config once. + + Tries in-cluster config first (running as a pod), falls back to + kubeconfig (local development / CI). + """ + global _api_client + if _api_client is not None: + return _api_client + + from kubernetes import client, config + + try: + config.load_incluster_config() + except config.ConfigException: + config.load_kube_config() + + _api_client = client.ApiClient() + return _api_client + + +def core_v1_api(): + """Return a CoreV1Api using the cached client.""" + from kubernetes import client + + return client.CoreV1Api(api_client=get_api_client()) + + +def batch_v1_api(): + """Return a BatchV1Api using the cached client.""" + from kubernetes import client + + return client.BatchV1Api(api_client=get_api_client()) + + +def parent_pod_owner_reference(): + """Build an ownerReference pointing to the current (parent) pod. + + Returns None if we can't determine the current pod identity (e.g. + running outside a cluster, or HOSTNAME/pod UID not available). + """ + from kubernetes import client + + hostname = os.environ.get("HOSTNAME") + if not hostname: + return None + + try: + api = core_v1_api() + ns = namespace() + pod = api.read_namespaced_pod(hostname, ns) + return client.V1OwnerReference( + api_version="v1", + kind="Pod", + name=pod.metadata.name, + uid=pod.metadata.uid, + block_owner_deletion=False, + ) + except Exception: + logger.debug("Could not resolve parent pod owner reference", exc_info=True) + return None diff --git a/clearwing/sandbox/kube_sandbox.py b/clearwing/sandbox/kube_sandbox.py index 01f6cc42..c7e1ac49 100644 --- a/clearwing/sandbox/kube_sandbox.py +++ b/clearwing/sandbox/kube_sandbox.py @@ -1,6 +1,6 @@ """Kubernetes-backed SandboxContainer implementation. -Runs sandbox workloads as Kubernetes Jobs/Pods instead of local Docker +Runs sandbox workloads as Kubernetes Pods instead of local Docker containers. Implements the same interface as SandboxContainer so hunter agents are backend-agnostic. @@ -10,6 +10,7 @@ from __future__ import annotations import io +import json import logging import os import tarfile @@ -17,23 +18,16 @@ import uuid from .container import ExecResult, SandboxConfig, SandboxContainer +from .kube_client import core_v1_api, namespace, parent_pod_owner_reference logger = logging.getLogger(__name__) -_NAMESPACE_ENV = "CLEARWING_SANDBOX_NAMESPACE" -_EXEC_TIMEOUT_SECONDS = 600 _POD_STARTUP_TIMEOUT_SECONDS = 120 - - -def _namespace() -> str: - ns = os.environ.get(_NAMESPACE_ENV) - if ns: - return ns - try: - with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace") as f: - return f.read().strip() - except OSError: - return "default" +# Hard ceiling multiplier — activeDeadlineSeconds is set to this factor +# times the configured exec timeout so a zombie pod self-terminates even +# if our process crashes. +_DEADLINE_MULTIPLIER = 3 +_MIN_DEADLINE_SECONDS = 900 # 15 minutes floor class KubeSandboxContainer(SandboxContainer): @@ -47,24 +41,12 @@ def __init__(self, config: SandboxConfig): # Don't call super().__init__ — we override the full lifecycle self._config = config self._pod_name: str | None = None - self._namespace = _namespace() - self._core_v1 = None + self._namespace = namespace() self._started = False # Public attributes expected by HunterSandbox self.scratch_host_dir: str | None = None self.variant: list[str] = [] - def _api(self): - if self._core_v1 is None: - from kubernetes import client, config - - try: - config.load_incluster_config() - except config.ConfigException: - config.load_kube_config() - self._core_v1 = client.CoreV1Api() - return self._core_v1 - @property def container_id(self) -> str: return self._pod_name or "" @@ -78,7 +60,7 @@ def is_running(self) -> bool: if not self._pod_name: return False try: - pod = self._api().read_namespaced_pod_status(self._pod_name, self._namespace) + pod = core_v1_api().read_namespaced_pod_status(self._pod_name, self._namespace) return pod.status.phase == "Running" except Exception: return False @@ -87,7 +69,7 @@ def start(self) -> str: """Create and start a Pod matching the SandboxConfig.""" from kubernetes import client - api = self._api() + api = core_v1_api() cfg = self._config pod_id = uuid.uuid4().hex[:8] self._pod_name = f"clearwing-sandbox-{pod_id}" @@ -99,12 +81,8 @@ def start(self) -> str: # Resource limits resources = client.V1ResourceRequirements( - limits={ - "memory": f"{cfg.memory_mb}Mi", - }, - requests={ - "memory": f"{min(cfg.memory_mb, 512)}Mi", - }, + limits={"memory": f"{cfg.memory_mb}Mi"}, + requests={"memory": f"{min(cfg.memory_mb, 512)}Mi"}, ) if cfg.cpus > 0: resources.limits["cpu"] = str(cfg.cpus) @@ -130,17 +108,24 @@ def start(self) -> str: security_context=security_context, ) - # Pod-level security and network policy + # Compute activeDeadlineSeconds so k8s kills zombie pods + deadline = max( + _MIN_DEADLINE_SECONDS, + cfg.timeout_seconds * _DEADLINE_MULTIPLIER, + ) + pod_spec = client.V1PodSpec( containers=[container], restart_policy="Never", - # Network isolation: use a NetworkPolicy on the namespace rather - # than Docker's network_mode=none. The pod itself is created - # without hostNetwork so default namespace policies apply. host_network=False, automount_service_account_token=False, + active_deadline_seconds=deadline, ) + # Set ownerReference to parent pod for cascading GC + owner_ref = parent_pod_owner_reference() + owner_references = [owner_ref] if owner_ref else None + pod = client.V1Pod( metadata=client.V1ObjectMeta( name=self._pod_name, @@ -150,6 +135,7 @@ def start(self) -> str: "purpose": "sandbox", "clearwing-session": cfg.env.get("CLEARWING_SESSION_ID", "unknown"), }, + owner_references=owner_references, ), spec=pod_spec, ) @@ -157,14 +143,14 @@ def start(self) -> str: api.create_namespaced_pod(self._namespace, pod) self._wait_for_running() self._started = True - logger.debug("K8s sandbox pod %s running", self._pod_name) + logger.debug("K8s sandbox pod %s running (deadline=%ds)", self._pod_name, deadline) return self._pod_name def _wait_for_running(self) -> None: """Block until the Pod reaches Running phase.""" from kubernetes import watch - api = self._api() + api = core_v1_api() deadline = time.monotonic() + _POD_STARTUP_TIMEOUT_SECONDS w = watch.Watch() try: @@ -197,37 +183,25 @@ def exec( env: dict[str, str] | None = None, workdir: str | None = None, ) -> ExecResult: - """Execute a command inside the sandbox Pod via kubectl exec.""" + """Execute a command inside the sandbox Pod. + + Uses the Kubernetes exec websocket with separate stdout/stderr + channels and parses the real exit code from the status channel. + """ from kubernetes.stream import stream if timeout is None: timeout = self._config.timeout_seconds - if isinstance(command, str): - exec_command = ["/bin/sh", "-c", command] - else: - exec_command = list(command) - - # Prepend env vars and workdir if specified - if env or workdir: - shell_prefix = "" - if workdir: - shell_prefix += f"cd {workdir} && " - if env: - exports = " ".join(f"{k}={v}" for k, v in env.items()) - shell_prefix += f"export {exports} && " - if isinstance(command, str): - exec_command = ["/bin/sh", "-c", f"{shell_prefix}{command}"] - else: - joined = " ".join(command) - exec_command = ["/bin/sh", "-c", f"{shell_prefix}{joined}"] + # Build the shell command with optional env/workdir prefix + shell_cmd = self._build_shell_command(command, env, workdir) + exec_command = ["/bin/sh", "-c", shell_cmd] started = time.monotonic() timed_out = False try: - # Use the kubernetes python client's exec resp = stream( - self._api().connect_get_namespaced_pod_exec, + core_v1_api().connect_get_namespaced_pod_exec, self._pod_name, self._namespace, container="sandbox", @@ -236,16 +210,18 @@ def exec( stdout=True, stdin=False, tty=False, - _preload_content=True, - _request_timeout=timeout, + _preload_content=False, ) - # stream() returns the combined output as a string for _preload_content=True - stdout = resp if isinstance(resp, str) else "" - stderr = "" - exit_code = 0 + # Read until completion or timeout + resp.run_forever(timeout=timeout) + + stdout = resp.read_stdout() or "" + stderr = resp.read_stderr() or "" + exit_code = self._parse_exit_code(resp) + except Exception as exc: elapsed = time.monotonic() - started - if elapsed >= timeout: + if elapsed >= (timeout - 1): # within 1s of timeout timed_out = True stdout = "" stderr = f"Command timed out after {timeout}s" @@ -264,48 +240,125 @@ def exec( timed_out=timed_out, ) + @staticmethod + def _build_shell_command( + command: str | list[str], + env: dict[str, str] | None, + workdir: str | None, + ) -> str: + """Compose the shell command string with env/workdir prefix.""" + prefix = "" + if workdir: + prefix += f"cd {workdir} && " + if env: + exports = " ".join(f"{k}={v}" for k, v in env.items()) + prefix += f"export {exports} && " + + if isinstance(command, str): + return f"{prefix}{command}" + return f"{prefix}{' '.join(command)}" + + @staticmethod + def _parse_exit_code(resp) -> int: + """Extract the process exit code from the websocket status channel. + + Channel 3 carries a JSON status message: + {"status": "Success"} → exit 0 + {"status": "Failure", "message": "...", "reason": "NonZeroExitCode", + "details": {"causes": [{"reason": "ExitCode", "message": "N"}]}} + """ + try: + err_channel = resp.read_channel(3) + if not err_channel: + return 0 + status = json.loads(err_channel) + if status.get("status") == "Success": + return 0 + # Extract exit code from details.causes + details = status.get("details", {}) + for cause in details.get("causes", []): + if cause.get("reason") == "ExitCode": + return int(cause.get("message", "1")) + # Generic failure without explicit code + return 1 + except (json.JSONDecodeError, ValueError, TypeError): + return 1 + def write_file(self, container_path: str, content: bytes) -> None: - """Write a file into the sandbox Pod.""" - # Create a tar archive in memory and pipe it via exec + """Write a file into the sandbox Pod via stdin tar pipe.""" + from kubernetes.stream import stream + + dest_dir = os.path.dirname(container_path) + filename = os.path.basename(container_path) + + # Build tar archive in memory buf = io.BytesIO() with tarfile.open(fileobj=buf, mode="w") as tar: - info = tarfile.TarInfo(name=os.path.basename(container_path)) + info = tarfile.TarInfo(name=filename) info.size = len(content) tar.addfile(info, io.BytesIO(content)) - buf.seek(0) - - dest_dir = os.path.dirname(container_path) - self.exec(f"mkdir -p {dest_dir}") - - # Write via base64 encoding to avoid binary transfer issues - import base64 - - encoded = base64.b64encode(buf.getvalue()).decode("ascii") - self.exec( - f"echo '{encoded}' | base64 -d | tar -xf - -C {dest_dir}" + tar_bytes = buf.getvalue() + + # Exec tar extraction with stdin pipe + resp = stream( + core_v1_api().connect_get_namespaced_pod_exec, + self._pod_name, + self._namespace, + container="sandbox", + command=["/bin/sh", "-c", f"mkdir -p {dest_dir} && tar -xf - -C {dest_dir}"], + stderr=True, + stdout=True, + stdin=True, + tty=False, + _preload_content=False, ) + resp.write_stdin(tar_bytes) + resp.close() def read_file(self, container_path: str) -> bytes: - """Read a file from the sandbox Pod.""" - import base64 + """Read a file from the sandbox Pod via stdout tar pipe.""" + from kubernetes.stream import stream - result = self.exec(f"base64 < {container_path}") - if result.exit_code != 0: - raise FileNotFoundError( - f"Cannot read {container_path}: {result.stderr}" - ) - return base64.b64decode(result.stdout.strip()) + resp = stream( + core_v1_api().connect_get_namespaced_pod_exec, + self._pod_name, + self._namespace, + container="sandbox", + command=["/bin/sh", "-c", f"tar -cf - -C / {container_path.lstrip('/')}"], + stderr=True, + stdout=True, + stdin=False, + tty=False, + _preload_content=False, + ) + resp.run_forever(timeout=30) + stdout_data = resp.read_stdout(timeout=0) + + if not stdout_data: + raise FileNotFoundError(f"Cannot read {container_path}") + + # Extract file content from tar + tar_buf = io.BytesIO(stdout_data.encode("latin-1") if isinstance(stdout_data, str) else stdout_data) + with tarfile.open(fileobj=tar_buf, mode="r") as tar: + members = tar.getmembers() + if not members: + raise FileNotFoundError(f"Cannot read {container_path}") + f = tar.extractfile(members[0]) + if f is None: + raise FileNotFoundError(f"Cannot read {container_path}: not a regular file") + return f.read() def copy_tree_into(self, host_path: str, container_path: str) -> None: - """Copy a directory tree into the Pod. + """Copy a directory tree into the Pod via stdin tar pipe. - Creates a tar of the host path and streams it into the container. + Pipes a tar archive directly into the container's stdin — + no base64 encoding, no shell arg limits, single round-trip. """ import subprocess - self.exec(f"mkdir -p {container_path}") + from kubernetes.stream import stream - # Create tar locally, base64 encode, exec into pod + # Create tar locally result = subprocess.run( ["tar", "-cf", "-", "-C", host_path, "."], capture_output=True, @@ -314,32 +367,33 @@ def copy_tree_into(self, host_path: str, container_path: str) -> None: if result.returncode != 0: raise RuntimeError(f"Failed to tar {host_path}: {result.stderr.decode()}") - import base64 - - # For large trees, chunk the transfer - encoded = base64.b64encode(result.stdout).decode("ascii") - chunk_size = 500_000 # ~375KB decoded per chunk - if len(encoded) <= chunk_size: - self.exec( - f"echo '{encoded}' | base64 -d | tar -xf - --no-same-owner -C {container_path}" - ) - else: - # Write chunks to a temp file in the container - self.exec("rm -f /tmp/_transfer.tar.b64") - for i in range(0, len(encoded), chunk_size): - chunk = encoded[i : i + chunk_size] - self.exec(f"echo -n '{chunk}' >> /tmp/_transfer.tar.b64") - self.exec( - f"base64 -d /tmp/_transfer.tar.b64 | tar -xf - --no-same-owner -C {container_path} && " - f"rm -f /tmp/_transfer.tar.b64" - ) + tar_bytes = result.stdout + + # Pipe into container via exec stdin + resp = stream( + core_v1_api().connect_get_namespaced_pod_exec, + self._pod_name, + self._namespace, + container="sandbox", + command=["/bin/sh", "-c", f"mkdir -p {container_path} && tar -xf - --no-same-owner -C {container_path}"], + stderr=True, + stdout=True, + stdin=True, + tty=False, + _preload_content=False, + ) + # Write in chunks to avoid websocket frame size limits + chunk_size = 1024 * 1024 # 1MB chunks + for i in range(0, len(tar_bytes), chunk_size): + resp.write_stdin(tar_bytes[i:i + chunk_size]) + resp.close() def stop(self) -> None: """Delete the sandbox Pod.""" if not self._pod_name: return try: - self._api().delete_namespaced_pod( + core_v1_api().delete_namespaced_pod( self._pod_name, self._namespace, grace_period_seconds=5, diff --git a/tests/test_kube_sandbox.py b/tests/test_kube_sandbox.py index f25b2530..2c1efe74 100644 --- a/tests/test_kube_sandbox.py +++ b/tests/test_kube_sandbox.py @@ -1,5 +1,6 @@ """Unit tests for Kubernetes sandbox backend support.""" +import json import os import unittest from unittest.mock import MagicMock, patch @@ -33,7 +34,6 @@ def test_docker_backend_returns_sandbox_container(self): with patch.dict(os.environ, {"CLEARWING_SANDBOX_BACKEND": "docker"}): cfg = SandboxConfig(image="test:latest") - # Don't actually start it — just verify the type with patch.object(SandboxContainer, "start"): sb = create_sandbox(cfg) self.assertIsInstance(sb, SandboxContainer) @@ -70,7 +70,6 @@ def test_override_image_used_for_all_variants(self): self.assertEqual(tag, "registry.example/sandbox:override") self.assertEqual(sandbox._image_tag, "registry.example/sandbox:override") - # All variants should map to the override image for key, img in sandbox._variant_images.items(): self.assertEqual(img, "registry.example/sandbox:override") @@ -97,9 +96,7 @@ def test_registry_builds_via_kaniko(self, mock_build): sandbox.build_recipe.apt_packages = ["build-essential"] sandbox._optional_packages = [] - # Need _render_dockerfile to work with patch.object(sandbox, "_render_dockerfile", return_value="FROM debian:11\n"): - # Pop the CLEARWING_SANDBOX_IMAGE to avoid override path os.environ.pop("CLEARWING_SANDBOX_IMAGE", None) tag = sandbox._resolve_kube_images() @@ -128,5 +125,131 @@ def test_different_content_different_tag(self): self.assertNotEqual(tag1, tag2) +class ExitCodeParsingTests(unittest.TestCase): + """Test the WSClient status channel exit code parser.""" + + def test_success_status(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + resp = MagicMock() + resp.read_channel.return_value = json.dumps({"status": "Success"}) + self.assertEqual(KubeSandboxContainer._parse_exit_code(resp), 0) + + def test_failure_with_exit_code(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + resp = MagicMock() + resp.read_channel.return_value = json.dumps({ + "status": "Failure", + "reason": "NonZeroExitCode", + "details": { + "causes": [{"reason": "ExitCode", "message": "137"}] + }, + }) + self.assertEqual(KubeSandboxContainer._parse_exit_code(resp), 137) + + def test_failure_without_details(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + resp = MagicMock() + resp.read_channel.return_value = json.dumps({ + "status": "Failure", + "message": "command terminated with exit code 2", + }) + self.assertEqual(KubeSandboxContainer._parse_exit_code(resp), 1) + + def test_empty_channel_is_success(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + resp = MagicMock() + resp.read_channel.return_value = "" + self.assertEqual(KubeSandboxContainer._parse_exit_code(resp), 0) + + def test_garbage_channel_is_failure(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + resp = MagicMock() + resp.read_channel.return_value = "not json at all" + self.assertEqual(KubeSandboxContainer._parse_exit_code(resp), 1) + + +class ShellCommandBuildTests(unittest.TestCase): + """Test the command composition helper.""" + + def test_plain_string_command(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + cmd = KubeSandboxContainer._build_shell_command("ls -la", None, None) + self.assertEqual(cmd, "ls -la") + + def test_with_workdir(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + cmd = KubeSandboxContainer._build_shell_command("make", None, "/src") + self.assertEqual(cmd, "cd /src && make") + + def test_with_env_and_workdir(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + cmd = KubeSandboxContainer._build_shell_command( + "gcc test.c", {"CC": "gcc-12"}, "/workspace" + ) + self.assertEqual(cmd, "cd /workspace && export CC=gcc-12 && gcc test.c") + + def test_list_command(self): + from clearwing.sandbox.kube_sandbox import KubeSandboxContainer + + cmd = KubeSandboxContainer._build_shell_command(["echo", "hello"], None, None) + self.assertEqual(cmd, "echo hello") + + +class NamespaceResolutionTests(unittest.TestCase): + """Test the shared namespace() helper.""" + + def test_env_var_takes_precedence(self): + from clearwing.sandbox.kube_client import namespace + + with patch.dict(os.environ, {"CLEARWING_SANDBOX_NAMESPACE": "my-ns"}): + self.assertEqual(namespace(), "my-ns") + + def test_falls_back_to_sa_file(self): + from clearwing.sandbox.kube_client import namespace + + with patch.dict(os.environ, {}, clear=True): + os.environ.pop("CLEARWING_SANDBOX_NAMESPACE", None) + with patch("builtins.open", unittest.mock.mock_open(read_data="prod-ns\n")): + self.assertEqual(namespace(), "prod-ns") + + def test_falls_back_to_default(self): + from clearwing.sandbox.kube_client import namespace + + with patch.dict(os.environ, {}, clear=True): + os.environ.pop("CLEARWING_SANDBOX_NAMESPACE", None) + with patch("builtins.open", side_effect=OSError): + self.assertEqual(namespace(), "default") + + +class ActiveDeadlineTests(unittest.TestCase): + """Test that sandbox pods get an activeDeadlineSeconds.""" + + def test_deadline_computed_from_timeout(self): + from clearwing.sandbox.kube_sandbox import _DEADLINE_MULTIPLIER, _MIN_DEADLINE_SECONDS + + expected = max(_MIN_DEADLINE_SECONDS, 600 * _DEADLINE_MULTIPLIER) + # The deadline is applied in start() — just verify the math + self.assertEqual(expected, 1800) + + def test_deadline_floor_applies(self): + from clearwing.sandbox.kube_sandbox import ( + _DEADLINE_MULTIPLIER, + _MIN_DEADLINE_SECONDS, + ) + + # With a very short timeout, the floor should apply + short_timeout = 60 + deadline = max(_MIN_DEADLINE_SECONDS, short_timeout * _DEADLINE_MULTIPLIER) + self.assertEqual(deadline, _MIN_DEADLINE_SECONDS) + + if __name__ == "__main__": unittest.main() From b86e547cb96b1d3d9649240916162cb621c1f6e9 Mon Sep 17 00:00:00 2001 From: Matt Owen Date: Wed, 5 Aug 2026 15:17:01 -0400 Subject: [PATCH 3/3] Add debug logging to kube sandbox exec --- clearwing/sandbox/kube_sandbox.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/clearwing/sandbox/kube_sandbox.py b/clearwing/sandbox/kube_sandbox.py index c7e1ac49..f46ac00f 100644 --- a/clearwing/sandbox/kube_sandbox.py +++ b/clearwing/sandbox/kube_sandbox.py @@ -195,6 +195,11 @@ def exec( # Build the shell command with optional env/workdir prefix shell_cmd = self._build_shell_command(command, env, workdir) + logger.info( + "sandbox exec pod=%s cmd=%s", + self._pod_name, + shell_cmd[:200], + ) exec_command = ["/bin/sh", "-c", shell_cmd] started = time.monotonic()