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..b208f558 --- /dev/null +++ b/clearwing/sandbox/kube_builder.py @@ -0,0 +1,233 @@ +"""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_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 + +from .kube_client import batch_v1_api, core_v1_api, namespace + +logger = logging.getLogger(__name__) + +_REGISTRY_ENV = "CLEARWING_SANDBOX_REGISTRY" +_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 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. + + 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 + + result = subprocess.run( + ["crane", "manifest", image_tag], + capture_output=True, + timeout=30, + ) + return result.returncode == 0 + except (FileNotFoundError, subprocess.TimeoutExpired): + pass + + 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: + """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 + + ns = 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 = core_v1_api() + batch = batch_v1_api() + + # Create ConfigMap with the Dockerfile + configmap = client.V1ConfigMap( + metadata=client.V1ObjectMeta( + name=configmap_name, + namespace=ns, + labels={"managed-by": "clearwing", "purpose": "sandbox-build"}, + ), + data={"Dockerfile": dockerfile_content}, + ) + try: + core.create_namespaced_config_map(ns, configmap) + except client.ApiException as e: + if e.status == 409: + core.replace_namespaced_config_map(configmap_name, ns, configmap) + else: + raise + + # Build the Kaniko Job + job = client.V1Job( + metadata=client.V1ObjectMeta( + name=job_name, + 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, + 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.create_namespaced_job(ns, job) + except client.ApiException as e: + if e.status == 409: + # Job already exists — delete and recreate + batch.delete_namespaced_job( + job_name, ns, propagation_policy="Background" + ) + time.sleep(2) + batch.create_namespaced_job(ns, job) + else: + raise + + # Poll until completion + logger.info("Waiting for Kaniko build job %s/%s", ns, job_name) + deadline = time.monotonic() + _BUILD_TIMEOUT_SECONDS + while time.monotonic() < deadline: + 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, batch, ns, job_name, configmap_name) + return image_tag + if status.failed and status.failed > 0: + 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, 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, ns: str, job_name: str) -> str: + """Best-effort retrieval of build pod logs.""" + try: + pods = core.list_namespaced_pod(ns, label_selector=f"job-name={job_name}") + if pods.items: + 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, batch, ns: str, job_name: str, configmap_name: str) -> None: + """Best-effort cleanup of build Job and ConfigMap.""" + try: + batch.delete_namespaced_job(job_name, ns, propagation_policy="Background") + except Exception: + pass + try: + 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 new file mode 100644 index 00000000..f46ac00f --- /dev/null +++ b/clearwing/sandbox/kube_sandbox.py @@ -0,0 +1,423 @@ +"""Kubernetes-backed SandboxContainer implementation. + +Runs sandbox workloads as Kubernetes 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 json +import logging +import os +import tarfile +import time +import uuid + +from .container import ExecResult, SandboxConfig, SandboxContainer +from .kube_client import core_v1_api, namespace, parent_pod_owner_reference + +logger = logging.getLogger(__name__) + +_POD_STARTUP_TIMEOUT_SECONDS = 120 +# 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): + """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._started = False + # Public attributes expected by HunterSandbox + self.scratch_host_dir: str | None = None + self.variant: list[str] = [] + + @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 = core_v1_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 = core_v1_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, + ) + + # 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", + 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, + namespace=self._namespace, + labels={ + "managed-by": "clearwing", + "purpose": "sandbox", + "clearwing-session": cfg.env.get("CLEARWING_SESSION_ID", "unknown"), + }, + owner_references=owner_references, + ), + spec=pod_spec, + ) + + api.create_namespaced_pod(self._namespace, pod) + self._wait_for_running() + self._started = True + 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 = core_v1_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. + + 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 + + # 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() + timed_out = False + try: + resp = stream( + core_v1_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=False, + ) + # 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 - 1): # within 1s of 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, + ) + + @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 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=filename) + info.size = len(content) + tar.addfile(info, io.BytesIO(content)) + 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 via stdout tar pipe.""" + from kubernetes.stream import stream + + 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 via stdin tar pipe. + + Pipes a tar archive directly into the container's stdin — + no base64 encoding, no shell arg limits, single round-trip. + """ + import subprocess + + from kubernetes.stream import stream + + # Create tar locally + 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()}") + + 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: + core_v1_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..2c1efe74 --- /dev/null +++ b/tests/test_kube_sandbox.py @@ -0,0 +1,255 @@ +"""Unit tests for Kubernetes sandbox backend support.""" + +import json +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") + 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") + 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 = [] + + with patch.object(sandbox, "_render_dockerfile", return_value="FROM debian:11\n"): + 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) + + +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()