"""k3s Job transport for cdx dispatches. The module owns no placement state: Kubernetes schedules the Job. It is loaded only when OD_DISPATCH_K3S=1, so the incumbent dispatcher remains the default. """ from __future__ import annotations import base64 import ipaddress import json import os import re import shlex import subprocess import sys import tempfile import uuid from dataclasses import dataclass from pathlib import Path from typing import Any, Callable from urllib.parse import urlsplit import k3s_result import remote_dispatch NAMESPACE = "overdeck" AUTH_SECRET = "cdx-codex-auth" GIT_READ_SECRET = "cdx-git-read" IMAGE_PULL_SECRET = "overdeck-ghcr-pull" KUBECONFIG_ENV = "OD_DISPATCH_K3S_KUBECONFIG" JOB_TIMEOUT_SECONDS = 7200 RESULT_PROCESSING_TIMEOUT_SECONDS = 120 RESULT_MAILBOX_GRACE_SECONDS = 10 MAX_RESULT_OBJECTS = 100_000 MAX_RESULT_OBJECT_BYTES = 256 * 1024 * 1024 MAX_RESULT_EXPANDED_BYTES = 1024 * 1024 * 1024 MAX_RESULT_PATCH_BYTES = 256 * 1024 * 1024 MAX_OBJECT_LIST_BYTES = 16 * 1024 * 1024 PROCESS_ADDRESS_SPACE_BYTES = 2 * 1024 * 1024 * 1024 DEFAULT_PROXY_URL = "http://10.42.0.1:31888" _PROXY_POD_CIDR = "10.42.0.0/16" _PROXY_CONFIG_HOST = "10.42.0.1" _PROXY_NAMESPACE = "overdeck-factory" _PROXY_APP = "overdeck-cdx-egress" _PROXY_NODE_PORT = 31888 _PROXY_CONTAINER_PORT = 8888 _WORKER_NODES = ("debian2", "debian3") _DIGEST_IMAGE = re.compile( r"^(?:overdeck\.k3s\.local|ghcr\.io/alexcodeplace)/overdeck-agent-sandbox@sha256:[0-9a-f]{64}$" ) _DNS_LABEL = re.compile(r"^[a-z0-9]([-a-z0-9]*[a-z0-9])?$") _K3S_CONFIG_KEYS = frozenset({"namespace", "image", "git_url", "result_host", "proxy_url"}) _APPROVED_GIT_URLS = frozenset({ "git@github.com:alexcodeplace/overdeck.git", "ssh://git@github.com/alexcodeplace/overdeck.git", }) class PreAgentFailure(RuntimeError): """The Job cannot have run, so the podman path remains safe.""" class PostAgentFailure(RuntimeError): """The Job may have run, so retrying on podman is forbidden.""" def _label(value: str, name: str) -> str: if len(value) > 63 or not _DNS_LABEL.fullmatch(value): raise PreAgentFailure(f"{name} is not a Kubernetes DNS label") return value def _usable_ipv4(value: str, name: str) -> ipaddress.IPv4Address: try: address = ipaddress.ip_address(value) except ValueError as exc: raise PreAgentFailure(f"{name} is not configured") from exc if ( address.version != 4 or address.is_unspecified or address.is_loopback or address.is_link_local or address.is_multicast or address.is_reserved ): raise PreAgentFailure(f"{name} is not configured") return address class KubernetesApi: """Small injectable Kubernetes API seam; tests use it without a cluster.""" def __init__(self, namespace: str, kubeconfig: str | None, run=subprocess.run) -> None: self.namespace = namespace self.kubeconfig = kubeconfig self.run = run def _argv(self, *args: str) -> list[str]: argv = ["kubectl"] if self.kubeconfig: argv += ["--kubeconfig", self.kubeconfig] return [*argv, "--namespace", self.namespace, *args] def _run_json(self, *args: str, document: dict[str, Any] | None = None) -> dict[str, Any]: try: done = self.run( self._argv(*args), input=(json.dumps(document) + "\n") if document is not None else None, capture_output=True, text=True, timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise PreAgentFailure(f"API unreachable: {exc}") from exc if done.returncode: raise PreAgentFailure((done.stderr or done.stdout).strip() or "Kubernetes API request failed") if not done.stdout.strip(): return {} try: value = json.loads(done.stdout) except json.JSONDecodeError as exc: raise PreAgentFailure("Kubernetes API returned invalid JSON") from exc if not isinstance(value, dict): raise PreAgentFailure("Kubernetes API returned an invalid object") return value def secret_exists(self, name: str) -> bool: try: done = self.run( self._argv("get", "secret", name), capture_output=True, text=True, timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise PreAgentFailure(f"API unreachable: {exc}") from exc if done.returncode == 0: return True detail = (done.stderr or done.stdout).strip() if "notfound" in detail.lower() or "not found" in detail.lower(): return False raise PreAgentFailure(f"API unreachable: {detail or 'secret lookup failed'}") def create_job(self, document: dict[str, Any]) -> str: value = self._run_json("create", "-f", "-", "-o", "json", document=document) uid = value.get("metadata", {}).get("uid", "") if not isinstance(uid, str) or not re.fullmatch(r"[0-9a-f-]{36}", uid): raise PreAgentFailure("created Job has no valid UID") return uid def create(self, document: dict[str, Any]) -> None: self._run_json("create", "-f", "-", "-o", "json", document=document) def unsuspend(self, job_name: str) -> None: try: done = self.run( self._argv("patch", "job", job_name, "--type=merge", "-p", '{"spec":{"suspend":false}}'), capture_output=True, text=True, timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: raise PostAgentFailure(f"could not establish whether Job started: {exc}") from exc if done.returncode: detail = (done.stderr or done.stdout).strip() raise PostAgentFailure(f"could not establish whether Job started: {detail or 'unsuspend failed'}") def wait_argv(self, job_name: str) -> list[str]: runner = Path(__file__).with_name("k3s_job_runner.py") command = [sys.executable, str(runner), "--namespace", self.namespace, "--job", job_name, "--timeout", str(JOB_TIMEOUT_SECONDS)] if self.kubeconfig: command += ["--kubeconfig", self.kubeconfig] return command def delete_job(self, job_name: str) -> str | None: try: done = self.run( self._argv("delete", "job", job_name, "--ignore-not-found", "--cascade=foreground", "--wait=true"), capture_output=True, text=True, timeout=30, ) except (OSError, subprocess.SubprocessError) as exc: return f"could not delete Kubernetes Job {job_name}: {exc}" if done.returncode: return f"could not delete Kubernetes Job {job_name}: {(done.stderr or done.stdout).strip()}" return None def _job_name(sandbox: str, attempt_id: str) -> str: safe = re.sub(r"[^a-z0-9-]", "-", sandbox.lower()).strip("-") or "workspace" return f"cdx-{safe[:30]}-{attempt_id[:12]}" def _owner_reference(job_name: str, job_uid: str) -> list[dict[str, Any]]: if not re.fullmatch(r"[0-9a-f-]{36}", job_uid): raise PreAgentFailure("created Job has no valid UID") return [{ "apiVersion": "batch/v1", "kind": "Job", "name": job_name, "uid": job_uid, "controller": True, "blockOwnerDeletion": True, }] def upload_secret_manifest(*, namespace: str, name: str, job_name: str, job_uid: str, target: k3s_result.UploadTarget) -> dict[str, Any]: _label(namespace, "namespace") _label(name, "upload Secret") value = k3s_result.serialize_target(target) if len(value.encode()) > 131072: raise PreAgentFailure("result upload target exceeds the Kubernetes control-plane limit") return { "apiVersion": "v1", "kind": "Secret", "metadata": {"name": name, "namespace": namespace, "labels": {"overdeck.dev/cdx-job": job_name}, "ownerReferences": _owner_reference(job_name, job_uid)}, "immutable": True, "type": "Opaque", "data": {"upload-json": base64.b64encode(value.encode()).decode()}, } def network_policy_manifest(*, namespace: str, name: str, job_name: str, job_uid: str, target: k3s_result.UploadTarget, proxy_url: str) -> dict[str, Any]: _label(namespace, "namespace") _label(name, "network policy") parsed_target = k3s_result.validate_target(target) parsed_proxy = urlsplit(proxy_url) try: proxy_address = ipaddress.ip_address(parsed_proxy.hostname or "") except ValueError as exc: raise PreAgentFailure("k3s egress proxy URL is invalid") from exc if parsed_proxy.scheme != "http" or parsed_proxy.username or parsed_proxy.password or parsed_proxy.path not in {"", "/"} or parsed_proxy.query or parsed_proxy.fragment or proxy_address.version != 4 or str(proxy_address) != _PROXY_CONFIG_HOST or parsed_proxy.port != _PROXY_NODE_PORT: raise PreAgentFailure("k3s egress proxy URL is invalid") return { "apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicy", "metadata": {"name": name, "namespace": namespace, "labels": {"overdeck.dev/cdx-job": job_name}, "ownerReferences": _owner_reference(job_name, job_uid)}, "spec": { "podSelector": {"matchLabels": {"overdeck.dev/cdx-job": job_name}}, "policyTypes": ["Egress"], "egress": [ {"to": [{"ipBlock": {"cidr": _PROXY_POD_CIDR}}], "ports": [{"protocol": "TCP", "port": parsed_proxy.port}]}, {"to": [{ "namespaceSelector": {"matchLabels": { "kubernetes.io/metadata.name": _PROXY_NAMESPACE, }}, "podSelector": {"matchLabels": { "app.kubernetes.io/name": _PROXY_APP, }}, }], "ports": [{"protocol": "TCP", "port": _PROXY_CONTAINER_PORT}]}, {"to": [{"ipBlock": {"cidr": f"{parsed_target.hostname}/32"}}], "ports": [{"protocol": "TCP", "port": parsed_target.port}]}, ], }, } def manifest(*, job_name: str, namespace: str, image: str, git_url: str, input_ref: str, input_commit: str, exec_name: str, forwarded_argv: list[str], upload_secret: str, proxy_url: str = DEFAULT_PROXY_URL) -> dict[str, Any]: """Build a suspended restricted Job; dependent resources exist before it starts.""" _label(job_name, "job name") _label(namespace, "namespace") _label(upload_secret, "upload Secret") if not _DIGEST_IMAGE.fullmatch(image): raise PreAgentFailure("image digest is missing or not from the approved sandbox repository") if not git_url or not input_ref.startswith("refs/cdx/") or not re.fullmatch(r"[0-9a-f]{40,64}", input_commit): raise PreAgentFailure("in-cluster git transport is not configured") proxy = urlsplit(proxy_url) try: proxy_ip = ipaddress.ip_address(proxy.hostname or "") except ValueError as exc: raise PreAgentFailure("k3s egress proxy URL is invalid") from exc if proxy.scheme != "http" or proxy_ip.version != 4 or str(proxy_ip) != _PROXY_CONFIG_HOST or proxy.port != _PROXY_NODE_PORT: raise PreAgentFailure("k3s egress proxy URL is invalid") locked = {"allowPrivilegeEscalation": False, "readOnlyRootFilesystem": True, "capabilities": {"drop": ["ALL"]}} pod_security = {"runAsNonRoot": True, "runAsUser": 1000, "runAsGroup": 1000, "fsGroup": 1000, "seccompProfile": {"type": "RuntimeDefault"}} resources = {"requests": {"cpu": "2", "memory": "8Gi", "ephemeral-storage": "128Mi"}, "limits": {"cpu": "4", "memory": "12Gi", "ephemeral-storage": "512Mi"}} ssh_command = f"/usr/bin/ssh -o HostName=ssh.github.com -p 443 -o HostKeyAlias=github.com -o ProxyCommand='/bin/nc -X connect -x $K3S_NODE_PROXY:{proxy.port} %h %p' -i /git-read/id_ed25519 -o IdentitiesOnly=yes -o StrictHostKeyChecking=yes -o UserKnownHostsFile=/git-read/known_hosts" proxy_ready = ( f"K3S_NODE_PROXY=\"${{K3S_POD_IP%.*}}.1\"; export K3S_NODE_PROXY; " f"attempt=0; while [ \"$attempt\" -lt 20 ]; do " f"attempt=$((attempt + 1)); " f"if /bin/nc -z -w 1 \"$K3S_NODE_PROXY\" {proxy.port}; then break; fi; " f"if [ \"$attempt\" -eq 20 ]; then " f"echo 'cdx-k3s-job: egress proxy did not become ready' >&2; exit 69; fi; " f"sleep 1; done" ) workspace_git = "git -c safe.directory=/workspace -C /workspace" fetch = ( f"{proxy_ready} && " f"git clone --no-checkout {shlex.quote(git_url)} /workspace && " f"{workspace_git} fetch origin {shlex.quote(input_ref)} && " f"test \"$({workspace_git} rev-parse FETCH_HEAD)\" = {shlex.quote(input_commit)} && " f"{workspace_git} checkout --detach {shlex.quote(input_commit)}" ) credential_init = "mkdir -p $HOME/.codex && cp /credential/auth.json $HOME/.codex/auth.json" agent_script = ( f"proxy_host=\"${{K3S_POD_IP%.*}}.1\"; " f"export HTTPS_PROXY=\"http://$proxy_host:{proxy.port}\" " f"HTTP_PROXY=\"http://$proxy_host:{proxy.port}\"; " f"exec /usr/local/bin/cdx-k3s-worker \"$@\"" ) agent_command = ["/bin/sh", "-c", agent_script, "cdx-k3s-worker", input_commit, "--", exec_name, *forwarded_argv] volumes = [ {"name": "workspace", "emptyDir": {"sizeLimit": "8Gi"}}, {"name": "home", "emptyDir": {"sizeLimit": "2Gi"}}, {"name": "tmp", "emptyDir": {"sizeLimit": "256Mi"}}, {"name": "result", "emptyDir": {"sizeLimit": "256Mi"}}, {"name": "credential", "secret": {"secretName": AUTH_SECRET, "defaultMode": 288}}, {"name": "git-read", "secret": {"secretName": GIT_READ_SECRET, "defaultMode": 288}}, ] workspace_mount = {"name": "workspace", "mountPath": "/workspace"} labels = {"app.kubernetes.io/name": "overdeck-cdx", "app.kubernetes.io/component": "worker", "overdeck.dev/cdx-job": job_name} pod_ip_env = {"name": "K3S_POD_IP", "valueFrom": { "fieldRef": {"fieldPath": "status.podIP"}, }} return { "apiVersion": "batch/v1", "kind": "Job", "metadata": {"name": job_name, "namespace": namespace, "labels": labels}, "spec": {"suspend": True, "backoffLimit": 0, "activeDeadlineSeconds": JOB_TIMEOUT_SECONDS, "ttlSecondsAfterFinished": 3600, "template": {"metadata": {"labels": labels}, "spec": { "restartPolicy": "Never", "automountServiceAccountToken": False, "affinity": {"nodeAffinity": { "requiredDuringSchedulingIgnoredDuringExecution": { "nodeSelectorTerms": [{"matchExpressions": [{ "key": "kubernetes.io/hostname", "operator": "In", "values": list(_WORKER_NODES), }]}], }, }}, "imagePullSecrets": [{"name": IMAGE_PULL_SECRET}], "securityContext": pod_security, "initContainers": [ {"name": "fetch-workspace", "image": image, "imagePullPolicy": "Always", "command": ["/bin/sh", "-c"], "args": [fetch], "securityContext": locked, "env": [pod_ip_env, {"name": "GIT_SSH_COMMAND", "value": ssh_command}], "volumeMounts": [workspace_mount, {"name": "home", "mountPath": "/home/cdx"}, {"name": "git-read", "mountPath": "/git-read", "readOnly": True}]}, {"name": "prepare-credential", "image": image, "imagePullPolicy": "Always", "command": ["/bin/sh", "-c"], "args": [credential_init], "securityContext": locked, "env": [{"name": "HOME", "value": "/home/cdx"}], "volumeMounts": [{"name": "home", "mountPath": "/home/cdx"}, {"name": "credential", "mountPath": "/credential", "readOnly": True}]}, ], "containers": [{ "name": "agent", "image": image, "imagePullPolicy": "Always", "command": agent_command, "workingDir": "/workspace", "env": [{"name": "HOME", "value": "/home/cdx"}, pod_ip_env, {"name": "NODE_USE_ENV_PROXY", "value": "1"}, {"name": "CDX_RESULT_UPLOAD_JSON", "valueFrom": { "secretKeyRef": {"name": upload_secret, "key": "upload-json"}}}], "resources": resources, "securityContext": locked, "volumeMounts": [workspace_mount, {"name": "home", "mountPath": "/home/cdx"}, {"name": "tmp", "mountPath": "/tmp"}, {"name": "result", "mountPath": "/result"}], }], "volumes": volumes, }}}, } def _run_limited_git(root: Path, *args: str, env: dict[str, str] | None = None, input_path: Path | None = None, output_path: Path | None = None, file_limit: int = k3s_result.MAX_BUNDLE_BYTES * 2, check: bool = True) -> subprocess.CompletedProcess: argv = [ "/usr/bin/prlimit", f"--as={PROCESS_ADDRESS_SPACE_BYTES}", f"--fsize={file_limit}", f"--cpu={RESULT_PROCESSING_TIMEOUT_SECONDS}", "--", "/usr/bin/git", *args, ] source = input_path.open("rb") if input_path is not None else None destination = output_path.open("wb") if output_path is not None else subprocess.PIPE errors = tempfile.TemporaryFile() try: completed = subprocess.run( argv, cwd=root, env=env, stdin=source, stdout=destination, stderr=errors, check=False, timeout=RESULT_PROCESSING_TIMEOUT_SECONDS + 5, ) errors.seek(0, os.SEEK_END) size = errors.tell() errors.seek(max(0, size - 1200)) error = errors.read().decode(errors="replace").strip() finally: errors.close() if source is not None: source.close() if output_path is not None: destination.close() if check and completed.returncode: raise RuntimeError(f"git exited {completed.returncode}: {error}") return completed def _validate_objects(root: Path, quarantine: Path, result_commit: str, input_commit: str, env: dict[str, str], directory: Path) -> None: objects = directory / "objects" _run_limited_git(root, "--git-dir", str(quarantine), "rev-list", "--objects", "--no-object-names", result_commit, f"^{input_commit}", env=env, output_path=objects, file_limit=MAX_OBJECT_LIST_BYTES) count = 0 with objects.open() as source: for line in source: count += 1 if count > MAX_RESULT_OBJECTS or not re.fullmatch(r"[0-9a-f]{40,64}", line.strip()): raise RuntimeError("result bundle object count or identity is not approved") sizes = directory / "object-sizes" _run_limited_git(root, "--git-dir", str(quarantine), "cat-file", "--batch-check=%(objectname) %(objecttype) %(objectsize)", env=env, input_path=objects, output_path=sizes, file_limit=MAX_OBJECT_LIST_BYTES) total = 0 measured = 0 with sizes.open() as source: for line in source: fields = line.split() if len(fields) != 3 or not re.fullmatch(r"[0-9a-f]{40,64}", fields[0]) or fields[1] not in {"blob", "commit", "tag", "tree"}: raise RuntimeError("result bundle object metadata is invalid") try: size = int(fields[2]) except ValueError as exc: raise RuntimeError("result bundle object size is invalid") from exc if size < 0 or size > MAX_RESULT_OBJECT_BYTES: raise RuntimeError("result bundle contains an oversized object") total += size measured += 1 if total > MAX_RESULT_EXPANDED_BYTES: raise RuntimeError("result bundle expanded size is not approved") if measured != count: raise RuntimeError("result bundle object inventory is incomplete") def import_result(root: Path, bundle: Path, input_commit: str, sandbox: str) -> str: """Validate untrusted objects in quarantine before importing one local result ref.""" if not bundle.is_file() or not 0 < bundle.stat().st_size <= k3s_result.MAX_BUNDLE_BYTES: raise RuntimeError("result bundle size is not approved") common = Path(subprocess.check_output( ["/usr/bin/git", "-C", str(root), "rev-parse", "--path-format=absolute", "--git-common-dir"], text=True, timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ).strip()) object_directory = common / "objects" if not object_directory.is_dir(): raise RuntimeError("controller Git object directory is unavailable") with tempfile.TemporaryDirectory(prefix="cdx-result-quarantine-", dir=bundle.parent) as temporary: directory = Path(temporary) _run_limited_git(root, "bundle", "verify", str(bundle), output_path=directory / "bundle-verification", file_limit=MAX_OBJECT_LIST_BYTES) quarantine = directory / "repository.git" _run_limited_git(root, "init", "--bare", str(quarantine)) alternate = str(object_directory) if "\n" in alternate: raise RuntimeError("controller Git object directory is invalid") (quarantine / "objects/info/alternates").write_text(alternate + "\n") env = {**os.environ, "GIT_ALTERNATE_OBJECT_DIRECTORIES": alternate} _run_limited_git(root, "--git-dir", str(quarantine), "-c", "fetch.unpackLimit=0", "-c", "fetch.fsckObjects=true", "fetch", "--no-tags", str(bundle), "HEAD", env=env) result_commit = _run_limited_git( root, "--git-dir", str(quarantine), "rev-parse", "FETCH_HEAD^{commit}", env=env, ).stdout.decode().strip() if _run_limited_git(root, "--git-dir", str(quarantine), "merge-base", "--is-ancestor", input_commit, result_commit, env=env, check=False).returncode: raise RuntimeError("result bundle does not descend from the submitted workspace") _validate_objects(root, quarantine, result_commit, input_commit, env, directory) _run_limited_git(root, "--git-dir", str(quarantine), "fsck", "--strict", "--connectivity-only", "--no-reflogs", "--no-dangling", result_commit, env=env) quarantine_ref = "refs/cdx/result" _run_limited_git(root, "--git-dir", str(quarantine), "update-ref", quarantine_ref, result_commit, env=env) fetch_ref = f"refs/cdx-fetch/{sandbox}" _run_limited_git(root, "fetch", "--no-tags", str(quarantine), f"+{quarantine_ref}:{fetch_ref}", env=env) imported = _run_limited_git(root, "rev-parse", f"{fetch_ref}^{{commit}}").stdout.decode().strip() if imported != result_commit: raise RuntimeError("validated result commit changed during import") return fetch_ref def apply_result(root: Path, pre_sha: str, result_ref: str, patch: Path, run=subprocess.run) -> str | None: """Apply one validated delta unless the owner's checkout changed during dispatch.""" try: current_tree, _ = remote_dispatch._snapshot_tree(root, run=run) expected_tree = remote_dispatch.tree_of(root, pre_sha, run=run) except remote_dispatch.OffloadUnavailable as exc: return f"checking the local checkout of {root} failed: {exc}" if current_tree != expected_tree: return ( f"local checkout {root} changed during the k3s dispatch; not overwriting it — " f"the validated result is safe at {result_ref} in this repository" ) try: _run_limited_git(root, "diff", "--binary", pre_sha, result_ref, "--", output_path=patch, file_limit=MAX_RESULT_PATCH_BYTES) if patch.stat().st_size: _run_limited_git(root, "apply", "--check", "--binary", "-", input_path=patch) latest_tree, _ = remote_dispatch._snapshot_tree(root, run=run) if latest_tree != expected_tree: return ( f"local checkout {root} changed during the k3s dispatch; not overwriting it — " f"the validated result is safe at {result_ref} in this repository" ) if patch.stat().st_size: _run_limited_git(root, "apply", "--binary", "-", input_path=patch) except (OSError, RuntimeError, subprocess.SubprocessError) as exc: return f"applying the validated k3s result failed: {exc}" finally: patch.unlink(missing_ok=True) return None def _delete_input_ref(root: Path, git_url: str, input_ref: str, run=subprocess.run) -> str | None: try: deleted = run( ["/usr/bin/git", "-C", str(root), "push", "-q", "--no-verify", git_url, f":{input_ref}"], capture_output=True, text=True, timeout=remote_dispatch.SYNC_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError) as exc: return f"could not remove cdx input ref: {exc}" if deleted.returncode: return f"could not remove cdx input ref: {deleted.stderr.strip()}" return None @dataclass class Session: host_name: str access: dict root: Path rel_dir: str workspace: str argv: list[str] sandbox_id: str git_url: str pre_sha: str input_ref: str job_name: str api: KubernetesApi mailbox: k3s_result.ResultMailbox temporary: tempfile.TemporaryDirectory git_warning: str | None = None def release(self) -> None: errors = [self.api.delete_job(self.job_name), _delete_input_ref(self.root, self.git_url, self.input_ref)] try: self.mailbox.close() except Exception as exc: errors.append(f"could not close cdx result mailbox: {exc}") self.temporary.cleanup() for error in errors: if error: print(f"cdx: {error}", file=sys.stderr) def pull_back(self, run=subprocess.run) -> str | None: """Receive, quarantine, validate, and locally apply the one attempt result.""" try: bundle = self.mailbox.wait(RESULT_MAILBOX_GRACE_SECONDS) result_ref = import_result(self.root, bundle, self.pre_sha, self.sandbox_id) return apply_result(self.root, self.pre_sha, result_ref, Path(self.temporary.name) / "result.patch", run=run) except (OSError, RuntimeError, TimeoutError, subprocess.SubprocessError) as exc: return f"result retrieval from k3s failed: {exc}" def _config(registry: dict) -> dict: value = registry.get("k3s") if not isinstance(value, dict): raise PreAgentFailure("k3s dispatch configuration is absent") unknown = set(value) - _K3S_CONFIG_KEYS missing = _K3S_CONFIG_KEYS - set(value) if unknown or missing: details = [] if unknown: details.append(f"unknown keys: {', '.join(sorted(unknown))}") if missing: details.append(f"missing keys: {', '.join(sorted(missing))}") raise PreAgentFailure(f"k3s dispatch configuration is invalid ({'; '.join(details)})") if any(not isinstance(value[key], str) or not value[key] for key in _K3S_CONFIG_KEYS): raise PreAgentFailure("k3s dispatch configuration values must be non-empty strings") _label(value["namespace"], "namespace") if not _DIGEST_IMAGE.fullmatch(value["image"]): raise PreAgentFailure("image digest is missing or not from the approved sandbox repository") if value["git_url"] not in _APPROVED_GIT_URLS: raise PreAgentFailure("in-cluster Git URL is not an approved credential-free repository") _usable_ipv4(value["result_host"], "k3s result mailbox host") try: proxy = urlsplit(value["proxy_url"]) proxy_address = _usable_ipv4(proxy.hostname or "", "k3s egress proxy host") proxy_port = proxy.port except ValueError as exc: raise PreAgentFailure("k3s egress proxy URL is invalid") from exc if ( proxy.scheme != "http" or proxy.username or proxy.password or proxy.path not in {"", "/"} or proxy.query or proxy.fragment or proxy_address.version != 4 or proxy_port is None ): raise PreAgentFailure("k3s egress proxy URL is invalid") return value def open_session(*, exec_name: str, forwarded_argv: list[str], cwd: Path | None, registry_path: Path, credential: remote_dispatch.Credential | None, run, fallback: Callable[[], remote_dispatch.Session], api: KubernetesApi | None = None) -> Session | remote_dispatch.Session: """Prepare a suspended Job; only failures proven pre-agent may use podman.""" client: KubernetesApi | None = None mailbox: k3s_result.ResultMailbox | None = None temporary: tempfile.TemporaryDirectory | None = None root: Path | None = None git_url = "" input_ref = "" job_name = "" job_created = False input_pushed = False transferred = False try: if credential is None: raise PreAgentFailure("no credential resolved") registry = remote_dispatch.load_registry(registry_path) config = _config(registry) namespace = config["namespace"] image = config["image"] git_url = config["git_url"] result_host = config["result_host"] proxy_url = config["proxy_url"] client = api or KubernetesApi(namespace, os.environ.get(KUBECONFIG_ENV), run=run) for secret in (AUTH_SECRET, GIT_READ_SECRET, IMAGE_PULL_SECRET): if not client.secret_exists(secret): raise PreAgentFailure(f"Secret {secret} is absent") root = remote_dispatch.repo_root(cwd) pre_sha = remote_dispatch.snapshot_commit(root, run=run) if pre_sha is None: raise PreAgentFailure("git snapshot is unavailable for k3s dispatch") sandbox = remote_dispatch.sandbox_id(root) attempt_id = uuid.uuid4().hex[:24] input_ref = remote_dispatch.dispatch_ref(sandbox, remote_dispatch.new_dispatch_id()) if not git_url: raise PreAgentFailure("in-cluster git transport is not configured") if not result_host: raise PreAgentFailure("k3s result mailbox host is not configured") temporary = tempfile.TemporaryDirectory(prefix="cdx-k3s-result-") mailbox = k3s_result.ResultMailbox(result_host, attempt_id, Path(temporary.name)) target = mailbox.start() job_name = _job_name(sandbox, attempt_id) upload_secret = f"cdx-upload-{attempt_id[:16]}" policy_name = f"cdx-egress-{attempt_id[:16]}" job = manifest(job_name=job_name, namespace=namespace, image=image, git_url=git_url, input_ref=input_ref, input_commit=pre_sha, exec_name=exec_name, forwarded_argv=forwarded_argv, upload_secret=upload_secret, proxy_url=proxy_url) pushed = run(remote_dispatch.push_commit_argv(root, git_url, pre_sha, input_ref), capture_output=True, text=True, timeout=remote_dispatch.SYNC_TIMEOUT_SEC) if pushed.returncode: raise PreAgentFailure(f"input git push failed: {pushed.stderr.strip()}") input_pushed = True job_created = True job_uid = client.create_job(job) client.create(upload_secret_manifest(namespace=namespace, name=upload_secret, job_name=job_name, job_uid=job_uid, target=target)) client.create(network_policy_manifest(namespace=namespace, name=policy_name, job_name=job_name, job_uid=job_uid, target=target, proxy_url=proxy_url)) client.unsuspend(job_name) session = Session( host_name="k3s", access={}, root=root, rel_dir="/workspace", workspace="/workspace", argv=client.wait_argv(job_name), sandbox_id=sandbox, git_url=git_url, pre_sha=pre_sha, input_ref=input_ref, job_name=job_name, api=client, mailbox=mailbox, temporary=temporary, ) mailbox = None temporary = None transferred = True return session except PostAgentFailure as exc: raise remote_dispatch.OffloadUnavailable( f"k3s dispatch may have started and will not be retried: {exc}", code=remote_dispatch.EXIT_MIRROR, ) from exc except (PreAgentFailure, remote_dispatch.OffloadUnavailable, OSError, subprocess.SubprocessError, ValueError) as exc: print(f"cdx: k3s dispatch failed ({exc}); ran on podman", file=sys.stderr) return fallback() finally: if mailbox is not None: if job_created and client is not None: error = client.delete_job(job_name) if error: print(f"cdx: {error}", file=sys.stderr) mailbox.close() if temporary is not None: temporary.cleanup() if input_pushed and not transferred and root is not None and git_url and input_ref: error = _delete_input_ref(root, git_url, input_ref, run=run) if error: print(f"cdx: {error}", file=sys.stderr)