from __future__ import annotations import subprocess import sys import tempfile from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import k3s_dispatch import k3s_result import remote_dispatch IMAGE = "overdeck.k3s.local/overdeck-agent-sandbox@sha256:" + "a" * 64 GHCR_IMAGE = "ghcr.io/alexcodeplace/overdeck-agent-sandbox@sha256:" + "b" * 64 APPROVED_GIT_URL = "git@github.com:alexcodeplace/overdeck.git" INPUT = "b" * 40 TARGET = k3s_result.UploadTarget( "https://100.126.128.50:24443/v1/cdx-results/" + "1" * 24, "t" * 43, "-----BEGIN CERTIFICATE-----\nfixture\n-----END CERTIFICATE-----\n", ) def make_manifest(): return k3s_dispatch.manifest( job_name="cdx-test-111111111111", namespace="overdeck", image=IMAGE, git_url=APPROVED_GIT_URL, input_ref="refs/cdx/demo/1-" + "c" * 32 + "/in", input_commit=INPUT, exec_name="codex", forwarded_argv=["exec", "hello"], upload_secret="cdx-upload-1111111111111111", ) def test_manifest_is_suspended_restricted_and_has_no_result_sidecar_or_write_key(): job = make_manifest() pod = job["spec"]["template"]["spec"] assert job["spec"]["suspend"] is True assert job["spec"]["backoffLimit"] == 0 assert pod["automountServiceAccountToken"] is False node_requirement = pod["affinity"]["nodeAffinity"]["requiredDuringSchedulingIgnoredDuringExecution"]["nodeSelectorTerms"][0]["matchExpressions"][0] assert node_requirement == { "key": "kubernetes.io/hostname", "operator": "In", "values": ["debian2", "debian3"], } assert pod["imagePullSecrets"] == [{"name": k3s_dispatch.IMAGE_PULL_SECRET}] assert pod["securityContext"]["runAsNonRoot"] is True assert pod["securityContext"]["seccompProfile"] == {"type": "RuntimeDefault"} assert all(container["securityContext"]["capabilities"] == {"drop": ["ALL"]} for container in [*pod["initContainers"], *pod["containers"]]) assert [container["name"] for container in pod["containers"]] == ["agent"] agent = pod["containers"][0] assert agent["resources"] == { "requests": {"cpu": "2", "memory": "8Gi", "ephemeral-storage": "128Mi"}, "limits": {"cpu": "4", "memory": "12Gi", "ephemeral-storage": "512Mi"}, } assert "git push" not in " ".join(agent["command"]) assert agent["command"][:2] == ["/bin/sh", "-c"] assert agent["command"][3:] == [ "cdx-k3s-worker", INPUT, "--", "codex", "exec", "hello", ] assert "HTTPS_PROXY=\"http://$proxy_host:31888\"" in agent["command"][2] assert "exec /usr/local/bin/cdx-k3s-worker \"$@\"" in agent["command"][2] upload = next(item for item in agent["env"] if item["name"] == "CDX_RESULT_UPLOAD_JSON") assert upload["valueFrom"]["secretKeyRef"] == { "name": "cdx-upload-1111111111111111", "key": "upload-json", } fetch = next(item for item in pod["initContainers"] if item["name"] == "fetch-workspace") ssh_command = next(item["value"] for item in fetch["env"] if item["name"] == "GIT_SSH_COMMAND") assert "HostName=ssh.github.com -p 443 -o HostKeyAlias=github.com" in ssh_command assert "$K3S_NODE_PROXY:31888" in ssh_command fetch_script = fetch["args"][0] assert "K3S_NODE_PROXY=\"${K3S_POD_IP%.*}.1\"" in fetch_script assert "while [ \"$attempt\" -lt 20 ]" in fetch_script assert "/bin/nc -z -w 1 \"$K3S_NODE_PROXY\" 31888" in fetch_script assert "egress proxy did not become ready" in fetch_script assert fetch_script.index("/bin/nc") < fetch_script.index("git clone") pod_ip = next(item for item in fetch["env"] if item["name"] == "K3S_POD_IP") assert pod_ip["valueFrom"]["fieldRef"] == {"fieldPath": "status.podIP"} assert "hostAliases" not in pod assert {mount["name"] for mount in fetch["volumeMounts"]} >= {"workspace", "git-read"} workspace_git = "git -c safe.directory=/workspace -C /workspace" assert f'test "$({workspace_git} rev-parse FETCH_HEAD)" = {INPUT}' in fetch["args"][0] assert f"{workspace_git} checkout --detach {INPUT}" in fetch["args"][0] assert f"{workspace_git} fetch origin" in fetch["args"][0] assert "checkout --detach FETCH_HEAD" not in fetch["args"][0] assert "git-read" not in {mount["name"] for mount in agent["volumeMounts"]} credential = next(item for item in pod["initContainers"] if item["name"] == "prepare-credential") assert credential["args"] == ["mkdir -p $HOME/.codex && cp /credential/auth.json $HOME/.codex/auth.json"] assert "${" not in credential["args"][0] assert all("hostPath" not in volume for volume in pod["volumes"]) secrets = {volume["name"]: volume["secret"]["secretName"] for volume in pod["volumes"] if "secret" in volume} assert secrets == {"credential": k3s_dispatch.AUTH_SECRET, "git-read": k3s_dispatch.GIT_READ_SECRET} def test_upload_secret_and_network_policy_are_attempt_bound_to_suspended_job(): secret = k3s_dispatch.upload_secret_manifest( namespace="overdeck", name="cdx-upload-1111111111111111", job_name="cdx-test-111111111111", job_uid="12345678-1234-1234-1234-123456789abc", target=TARGET, ) assert secret["immutable"] is True assert set(secret["data"]) == {"upload-json"} assert secret["metadata"]["ownerReferences"][0]["name"] == "cdx-test-111111111111" policy = k3s_dispatch.network_policy_manifest( namespace="overdeck", name="cdx-egress-1111111111111111", job_name="cdx-test-111111111111", job_uid="12345678-1234-1234-1234-123456789abc", target=TARGET, proxy_url=k3s_dispatch.DEFAULT_PROXY_URL, ) egress = policy["spec"]["egress"] assert {item["to"][0]["ipBlock"]["cidr"] for item in egress if "ipBlock" in item["to"][0]} == { "10.42.0.0/16", "100.126.128.50/32", } assert { "namespaceSelector": { "matchLabels": {"kubernetes.io/metadata.name": "overdeck-factory"}, }, "podSelector": { "matchLabels": {"app.kubernetes.io/name": "overdeck-cdx-egress"}, }, } in [item["to"][0] for item in egress] assert {item["ports"][0]["port"] for item in egress} == {31888, 8888, 24443} assert policy["spec"]["podSelector"]["matchLabels"] == { "overdeck.dev/cdx-job": "cdx-test-111111111111", } def test_image_lane_digest_identity_is_accepted_by_the_submitter(): assert k3s_dispatch._DIGEST_IMAGE.fullmatch(IMAGE) assert k3s_dispatch._DIGEST_IMAGE.fullmatch(GHCR_IMAGE) assert not k3s_dispatch._DIGEST_IMAGE.fullmatch( "ghcr.io/another-owner/overdeck-agent-sandbox@sha256:" + "a" * 64 ) def test_k3s_config_is_a_complete_non_secret_contract(): config = { "namespace": "overdeck", "image": GHCR_IMAGE, "git_url": APPROVED_GIT_URL, "result_host": "100.126.128.50", "proxy_url": "http://10.42.0.1:31888", } assert k3s_dispatch._config({"k3s": config}) == config @pytest.mark.parametrize("field, value, message", [ ("git_url", "https://user:token@github.com/alexcodeplace/overdeck.git", "credential-free"), ("result_host", "127.0.0.1", "result mailbox host"), ("result_host", "0.0.0.0", "result mailbox host"), ("proxy_url", "http://127.0.0.1:8888", "proxy host"), ("proxy_url", "http://0.0.0.0:8888", "proxy host"), ]) def test_k3s_config_refuses_secrets_and_unreachable_endpoints(field, value, message): config = { "namespace": "overdeck", "image": GHCR_IMAGE, "git_url": APPROVED_GIT_URL, "result_host": "100.126.128.50", "proxy_url": "http://10.42.0.1:31888", } config[field] = value with pytest.raises(k3s_dispatch.PreAgentFailure, match=message): k3s_dispatch._config({"k3s": config}) @pytest.mark.parametrize("config, message", [ ({}, "absent"), ({"namespace": "overdeck"}, "missing keys"), ({"namespace": "overdeck", "image": GHCR_IMAGE, "git_url": APPROVED_GIT_URL, "result_host": "100.126.128.50", "proxy_url": "http://10.42.0.1:31888", "secret": "no"}, "unknown keys"), ]) def test_k3s_config_refuses_incomplete_or_secret_bearing_shape(config, message): registry = {} if not config else {"k3s": config} with pytest.raises(k3s_dispatch.PreAgentFailure, match=message): k3s_dispatch._config(registry) def test_kubernetes_create_requests_json_for_non_job_resources(): calls = [] def fake_run(argv, **_kwargs): calls.append(argv) output = '{"kind":"Secret"}\n' if argv[-2:] == ["-o", "json"] else "secret/test created\n" return subprocess.CompletedProcess(argv, 0, output, "") api = k3s_dispatch.KubernetesApi("overdeck", None, run=fake_run) api.create({"apiVersion": "v1", "kind": "Secret", "metadata": {"name": "test"}}) assert calls[0][-4:] == ["-f", "-", "-o", "json"] class FakeMailbox: instances = [] def __init__(self, host, attempt, output): self.host, self.attempt, self.output = host, attempt, output self.closed = False self.instances.append(self) def start(self): return k3s_result.UploadTarget( f"https://{self.host}:24443/v1/cdx-results/{self.attempt}", "t" * 43, "-----BEGIN CERTIFICATE-----\nfixture\n-----END CERTIFICATE-----\n", ) def wait(self, timeout): raise RuntimeError("result upload is absent") def close(self): self.closed = True class FakeApi: def __init__(self, *, auth=True, git=True, pull=True, reject=False, ambiguous_start=False): self.auth, self.git, self.pull = auth, git, pull self.reject, self.ambiguous_start = reject, ambiguous_start self.created = [] self.unsuspended = [] self.deleted = [] def secret_exists(self, name): if name == k3s_dispatch.AUTH_SECRET: return self.auth if name == k3s_dispatch.GIT_READ_SECRET: return self.git if name == k3s_dispatch.IMAGE_PULL_SECRET: return self.pull raise AssertionError(f"unexpected Secret lookup: {name}") def create_job(self, document): self.created.append(document) if self.reject: raise k3s_dispatch.PreAgentFailure("Job rejected: policy denied") return "12345678-1234-1234-1234-123456789abc" def create(self, document): self.created.append(document) def unsuspend(self, name): self.unsuspended.append(name) if self.ambiguous_start: raise k3s_dispatch.PostAgentFailure("API response was lost") def wait_argv(self, name): return ["kubectl", "wait", name] def delete_job(self, name): self.deleted.append(name) return None def _configured(monkeypatch, tmp_path, **overrides): config = { "image": IMAGE, "git_url": APPROVED_GIT_URL, "namespace": "overdeck", "result_host": "100.126.128.50", "proxy_url": "http://10.42.0.1:31888", } config.update(overrides) monkeypatch.setattr(remote_dispatch, "load_registry", lambda _: {"k3s": config}) monkeypatch.setattr(remote_dispatch, "repo_root", lambda cwd=None: tmp_path) monkeypatch.setattr(remote_dispatch, "snapshot_commit", lambda root, run: INPUT) monkeypatch.setattr(remote_dispatch, "sandbox_id", lambda root: "dispatch") monkeypatch.setattr(remote_dispatch, "new_dispatch_id", lambda: "1-" + "c" * 32) monkeypatch.setattr(k3s_result, "ResultMailbox", FakeMailbox) def _successful_run(argv, **kwargs): return subprocess.CompletedProcess(argv, 0, "", "") @pytest.mark.parametrize("api", [ FakeApi(reject=True), FakeApi(auth=False), FakeApi(git=False), FakeApi(pull=False), ]) def test_proven_pre_agent_failure_falls_back_and_names_reason(tmp_path, monkeypatch, capsys, api): _configured(monkeypatch, tmp_path) fallback = object() result = k3s_dispatch.open_session( exec_name="codex", forwarded_argv=["exec", "hi"], cwd=None, registry_path=tmp_path / "hosts", credential=remote_dispatch.Credential("codex", "acct", tmp_path), run=_successful_run, fallback=lambda: fallback, api=api, ) assert result is fallback output = capsys.readouterr().err assert "cdx: k3s dispatch failed" in output assert "ran on podman" in output if not (api.auth and api.git and api.pull): assert api.created == [] @pytest.mark.parametrize("overrides", [ {"git_url": "https://user:token@github.com/alexcodeplace/overdeck.git"}, {"result_host": "127.0.0.1"}, {"proxy_url": "http://0.0.0.0:8888"}, ]) def test_invalid_transport_config_falls_back_before_job_creation(tmp_path, monkeypatch, overrides): _configured(monkeypatch, tmp_path, **overrides) api = FakeApi() fallback = object() result = k3s_dispatch.open_session( exec_name="codex", forwarded_argv=["exec", "hi"], cwd=None, registry_path=tmp_path / "hosts", credential=remote_dispatch.Credential("codex", "acct", tmp_path), run=_successful_run, fallback=lambda: fallback, api=api, ) assert result is fallback assert api.created == [] assert api.unsuspended == [] def test_ambiguous_unsuspend_failure_never_falls_back(tmp_path, monkeypatch): _configured(monkeypatch, tmp_path) api = FakeApi(ambiguous_start=True) fallback_calls = [] with pytest.raises(remote_dispatch.OffloadUnavailable, match="may have started and will not be retried"): k3s_dispatch.open_session( exec_name="codex", forwarded_argv=["exec", "hi"], cwd=None, registry_path=tmp_path / "hosts", credential=remote_dispatch.Credential("codex", "acct", tmp_path), run=_successful_run, fallback=lambda: fallback_calls.append(True), api=api, ) assert fallback_calls == [] assert api.deleted def test_missing_mailbox_result_after_job_may_have_run_is_loud_and_never_falls_back(tmp_path, monkeypatch): _configured(monkeypatch, tmp_path) api = FakeApi() session = k3s_dispatch.open_session( exec_name="codex", forwarded_argv=["exec", "hi"], cwd=None, registry_path=tmp_path / "hosts", credential=remote_dispatch.Credential("codex", "acct", tmp_path), run=_successful_run, fallback=lambda: pytest.fail("successful k3s setup must not fall back"), api=api, ) assert isinstance(session, k3s_dispatch.Session) assert session.workspace == "/workspace" calls = [] def forbidden(argv, **kwargs): calls.append(argv) pytest.fail("missing mailbox result must not attempt a result-ref fallback") error = session.pull_back(run=forbidden) assert error is not None and "result upload is absent" in error assert calls == [] session.release() def test_quarantined_bundle_is_validated_and_applied_as_uncommitted_delta(tmp_path): root = tmp_path / "controller" worker = tmp_path / "worker" subprocess.run(["/usr/bin/git", "init", "-q", str(root)], check=True) subprocess.run(["/usr/bin/git", "-C", str(root), "config", "user.name", "fixture"], check=True) subprocess.run(["/usr/bin/git", "-C", str(root), "config", "user.email", "fixture@example.invalid"], check=True) (root / "value.txt").write_text("before\n") subprocess.run(["/usr/bin/git", "-C", str(root), "add", "value.txt"], check=True) subprocess.run(["/usr/bin/git", "-C", str(root), "commit", "-qm", "base"], check=True) base = subprocess.check_output(["/usr/bin/git", "-C", str(root), "rev-parse", "HEAD"], text=True).strip() subprocess.run(["/usr/bin/git", "clone", "-q", str(root), str(worker)], check=True) subprocess.run(["/usr/bin/git", "-C", str(worker), "config", "user.name", "fixture"], check=True) subprocess.run(["/usr/bin/git", "-C", str(worker), "config", "user.email", "fixture@example.invalid"], check=True) (worker / "value.txt").write_text("after\n") subprocess.run(["/usr/bin/git", "-C", str(worker), "commit", "-qam", "result"], check=True) bundle = tmp_path / "result.bundle" subprocess.run(["/usr/bin/git", "-C", str(worker), "bundle", "create", str(bundle), "HEAD", f"^{base}"], check=True) result_ref = k3s_dispatch.import_result(root, bundle, base, "fixture") error = k3s_dispatch.apply_result(root, base, result_ref, tmp_path / "result.patch") assert error is None assert (root / "value.txt").read_text() == "after\n" assert subprocess.check_output( ["/usr/bin/git", "-C", str(root), "status", "--short"], text=True, ) == " M value.txt\n" def test_unset_flag_uses_incumbent_argv_without_k3s_call(monkeypatch): expected = ["agent-sandbox", "--host", "debian1", "--", "codex", "exec", "hi"] incumbent = remote_dispatch.Session("debian1", {}, Path("/repo"), "rel", expected) monkeypatch.delenv(remote_dispatch.K3S_DISPATCH_ENV, raising=False) monkeypatch.setattr(remote_dispatch, "_open_podman_session", lambda *args: incumbent) monkeypatch.setattr(k3s_dispatch, "open_session", lambda **kwargs: pytest.fail("k3s must not run")) actual = remote_dispatch.open_session("codex", ["exec", "hi"]) assert actual.argv == expected