From aa9f7d178c14843c61f6bec94d0fd92c01011738 Mon Sep 17 00:00:00 2001 From: Public Source Publisher Date: Sun, 11 Oct 2026 18:56:38 +0000 Subject: [PATCH] Publish curated Unraid-AppOps-Gateway source from ae6b891c48da412de613ce0ed52c0bc1e9908fbf --- PUBLIC_SOURCE_EXPORT.md | 2 +- PUBLIC_SOURCE_MANIFEST.json | 10 +- backend/tests/test_runner_fleet_specialize.py | 9 +- .../tests/test_runner_resource_envelope.py | 266 +++++++++++++++++- 4 files changed, 275 insertions(+), 12 deletions(-) diff --git a/PUBLIC_SOURCE_EXPORT.md b/PUBLIC_SOURCE_EXPORT.md index a128530..e168c6a 100644 --- a/PUBLIC_SOURCE_EXPORT.md +++ b/PUBLIC_SOURCE_EXPORT.md @@ -1,3 +1,3 @@ # Public source export -This source snapshot was generated from the private canonical repository at revision `8e49e71ad0298326f565f57d4962e482e738d6e1`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. +This source snapshot was generated from the private canonical repository at revision `ae6b891c48da412de613ce0ed52c0bc1e9908fbf`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. diff --git a/PUBLIC_SOURCE_MANIFEST.json b/PUBLIC_SOURCE_MANIFEST.json index 6353c0e..ed65a13 100644 --- a/PUBLIC_SOURCE_MANIFEST.json +++ b/PUBLIC_SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "sourceRevision": "8e49e71ad0298326f565f57d4962e482e738d6e1", + "sourceRevision": "ae6b891c48da412de613ce0ed52c0bc1e9908fbf", "files": [ { "path": ".dockerignore", @@ -1659,8 +1659,8 @@ }, { "path": "backend/tests/test_runner_fleet_specialize.py", - "sha256": "e3134a8ad84ab23fa2fef3d377d25b891cc4d854b8bdad7f271048404fbfc249", - "bytes": 205429 + "sha256": "5bd85d16db6ef824a8c83bdafb4e1a70cbb736064611929e538df7224748f885", + "bytes": 205798 }, { "path": "backend/tests/test_runner_fleet_v2_execution_surface.py", @@ -1684,8 +1684,8 @@ }, { "path": "backend/tests/test_runner_resource_envelope.py", - "sha256": "f2a1c2468bdda62acbc91657a35cfc63fe525413ee56dc66b8bd4527597c9aba", - "bytes": 101609 + "sha256": "dfba7e7ba84689205ed677a22aa2f7daca1898def800a9e7b6fa60528a5d0b73", + "bytes": 116563 }, { "path": "backend/tests/test_runner_scheduler_liveness_recovery.py", diff --git a/backend/tests/test_runner_fleet_specialize.py b/backend/tests/test_runner_fleet_specialize.py index cd8ae56..a43e8e5 100644 --- a/backend/tests/test_runner_fleet_specialize.py +++ b/backend/tests/test_runner_fleet_specialize.py @@ -23,10 +23,11 @@ import pytest "resource_cgroup_attestation_failed", "resource_cgroup_v2_unavailable", "Bearer secret-value", None, ]) -def test_build_resource_refusal_keeps_only_allowlisted_private_cause(specialize, monkeypatch, reason): +@pytest.mark.parametrize("sample_stage", ["after_snapshot", "Bearer secret-value", {"secret": "value"}, None]) +def test_build_resource_refusal_keeps_only_allowlisted_private_cause(specialize, monkeypatch, reason, sample_stage): monkeypatch.setattr(specialize, "_docker_outer_inventory", lambda: ([], [])) monkeypatch.setattr(specialize, "attest_build_probe_workload", lambda *_a, **_k: { - "ready": False, "reason": reason, + "ready": False, "reason": reason, "sampleStage": sample_stage, }) with pytest.raises(specialize.RunnerFleetSpecializeError) as error: specialize._build_probe_resource_attestation({"containerId": "a" * 64}, {"State": {"Pid": 77}}) @@ -45,6 +46,10 @@ def test_build_resource_refusal_keeps_only_allowlisted_private_cause(specialize, }, error.value) assert len(records) == 1 assert records[0]["causeCode"] == (reason if reason in specialize._RESOURCE_FAILURE_CODES else str(error.value)) + if reason == "resource_cgroup_process_mapping_invalid" and sample_stage == "after_snapshot": + assert records[0]["sampleStage"] == "after_snapshot" + else: + assert "sampleStage" not in records[0] assert "secret-value" not in json.dumps(records) diff --git a/backend/tests/test_runner_resource_envelope.py b/backend/tests/test_runner_resource_envelope.py index 0467976..61131e0 100644 --- a/backend/tests/test_runner_resource_envelope.py +++ b/backend/tests/test_runner_resource_envelope.py @@ -1286,6 +1286,216 @@ def _owned_family_fixture(monkeypatch, *, exact=False, init=True, docker=True): return envelope, claim, rows, processes, controls, writes, inodes +@pytest.mark.parametrize("case", ["owned", "foreign_zero", "duplicate", "escaped_thread", "missing_thread", "family_process"]) +def test_coherent_namespace_sample_requires_all_threads_contained(monkeypatch, case): + envelope, claim, _rows, processes, _controls, writes, _inodes = _owned_family_fixture(monkeypatch, exact=True) + leaf = f"/itworx-ci/{claim['containerId']}" + if case == "foreign_zero": + processes[leaf + "/init"] = "1 0" + elif case == "duplicate": + processes[leaf + "/init"] = "1 1" + elif case == "family_process": + processes["/itworx-ci"] = "77" + tids = (1, 2, 9) if case == "escaped_thread" else (1,) if case == "missing_thread" else (1, 2) + monkeypatch.setattr(envelope, "_probe_proc_task_ids", lambda _pids: (tids, ((1, 36, 901), (2, 36, 902)))) + if case == "owned": + sample = envelope._probe_namespace_thread_membership(101, leaf) + assert sample["processIds"] == (1, 2) + assert sample["threadIds"] == (1, 2) + else: + reason = "resource_cgroup_probe_namespace_invalid" if case == "family_process" else "resource_cgroup_process_mapping_invalid" + with pytest.raises(RuntimeError, match="^" + reason + "$"): + envelope._probe_namespace_thread_membership(101, leaf) + assert writes == [] + + +def test_build_snapshot_helper_is_fixed_dind_namespace_and_readonly_family(): + envelope = _module() + target = {"name": envelope.DIND_NAME, "id": "a" * 64} + args = envelope._host_probe_create_args("build_snapshot", "sha256:" + "b" * 64, + envelope._HOST_VOLUME_ROOT + "/controller-artifacts/" + "c" * 40, + namespace_target=target) + assert "container:" + target["id"] in args + assert "type=bind,source=/sys/fs/cgroup/itworx-ci,target=/host-cgroup,readonly,bind-propagation=rprivate" in args + assert not any("/proc" in value or "--pid=host" in value or "docker.sock" in value for value in args) + with pytest.raises(RuntimeError, match="resource_cgroup_probe_namespace_invalid"): + envelope._host_probe_create_args("build_snapshot", "sha256:" + "b" * 64, + envelope._HOST_VOLUME_ROOT + "/controller-artifacts/" + "c" * 40, + namespace_target={"name": "itworx-ci-checks", "id": "a" * 64}) + + +@pytest.mark.parametrize("case", ["owned", "thread_race_once", "process_count", "inode_drift", "limit_drift", "wrong_pid", "running_outer", "boolean_pid", "fractional_pid", "fractional_cpu", "fractional_memory", "malformed_name"]) +def test_coherent_build_snapshot_retries_only_inconsistent_membership(monkeypatch, case): + envelope, claim, rows, processes, _controls, writes, inodes = _owned_family_fixture(monkeypatch, exact=True) + leaf = f"/itworx-ci/{claim['containerId']}" + probe_id = "b" * 64 + path = leaf + "/docker/" + probe_id + rows[path] = {"path": path, "cpuMax": "400000 100000", "memoryMax": str(6 * 1024**3)} + processes[leaf + "/init"] = "1" + old_open, old_read, old_stat = envelope._probe_open_relative, envelope._probe_read, envelope.os.fstat + monkeypatch.setattr(envelope, "_probe_open_relative", lambda fd, name: 909 if name == path else old_open(fd, name)) + monkeypatch.setattr(envelope, "_probe_read", lambda fd, name: "2" if fd == 909 and name in {"cgroup.procs", "cgroup.threads"} else old_read(fd, name)) + monkeypatch.setattr(envelope.os, "fstat", lambda fd: SimpleNamespace(st_dev=36, st_ino=909) if fd == 909 else old_stat(fd)) + proc_calls, snapshots = [], [] + + def proc(_pids): + assert _pids == (1, 2) + proc_calls.append(True) + tids = (1, 2, 9) if case == "thread_race_once" and len(proc_calls) == 1 else (1, 2) + return tids, ((1, 36, 901), (2, 36, 902)) + + def snapshot(payload, _fd): + snapshots.append(payload) + result = {"version": 2, "groups": [dict(row) for row in rows.values()]} + if case == "inode_drift": + inodes[leaf + "/init"] += 1 + elif case == "limit_drift": + rows[leaf]["cpuMax"] = "500000 100000" + return result + + monkeypatch.setattr(envelope, "_probe_proc_task_ids", proc) + monkeypatch.setattr(envelope, "_probe_snapshot", snapshot) + monkeypatch.setattr(envelope.time, "sleep", lambda _seconds: None) + outer = [{"name": profile.container_name, "id": format(index, "064x"), "running": False} + for index, profile in enumerate(envelope.RUNNER_PROFILES.values(), 1)] + outer.append({"name": envelope.DIND_NAME, "id": claim["containerId"], "running": True}) + if case == "running_outer": + outer[0]["running"] = True + request = {"op": "build_snapshot", "target": {"name": envelope.DIND_NAME, "id": claim["containerId"]}, + "outer": outer, "processCount": 3 if case == "process_count" else 2, + "workloads": [{"id": probe_id, "nestedPid": 77 if case == "wrong_pid" else 2, + "cpuMilli": 4000, "memoryBytes": 6 * 1024**3}]} + for mutation, key, value in (("boolean_pid", "nestedPid", True), ("fractional_pid", "nestedPid", 2.0), + ("fractional_cpu", "cpuMilli", 4000.0), ("fractional_memory", "memoryBytes", float(6 * 1024**3))): + if case == mutation: + request["workloads"][0][key] = value + if case == "malformed_name": + outer[0]["name"] = [] + if case in {"owned", "thread_race_once"}: + result = envelope._probe_coherent_build_snapshot(request, 101) + assert result["namespaceProof"] == {"allThreadsContained": True, "processCount": 2, "threadCount": 2} + assert len(snapshots) == 1 + assert len(proc_calls) == (3 if case == "thread_race_once" else 2) + else: + expected = {"process_count": "process_mapping_invalid", "inode_drift": "probe_namespace_invalid", + "limit_drift": "probe_namespace_invalid", "wrong_pid": "probe_workload_mismatch", + "running_outer": "probe_request_invalid", "boolean_pid": "probe_request_invalid", + "fractional_pid": "probe_request_invalid", "fractional_cpu": "probe_request_invalid", + "fractional_memory": "probe_request_invalid", "malformed_name": "probe_request_invalid"}[case] + with pytest.raises(RuntimeError, match="^resource_cgroup_" + expected + "$"): + envelope._probe_coherent_build_snapshot(request, 101) + assert len(snapshots) == (1 if case in {"inode_drift", "limit_drift"} else 0) + if case == "process_count": + assert len(proc_calls) == 3 + assert writes == [] + + +@pytest.mark.parametrize("drift", ["none", "threads", "process", "identity"]) +def test_build_coherent_sample_keeps_complete_binding_and_exact_process_ids(monkeypatch, drift): + envelope = _module() + original = ({"Id": "a" * 64, "StartedAt": "start", "allIdentityFields": {"memory": 12 * 1024**3}}, [(10, 2)]) + current = copy.deepcopy(original) + if drift == "threads": + current[1][0] = (10, 3) + elif drift == "process": + current[1][0] = (11, 2) + elif drift == "identity": + current[0]["allIdentityFields"]["memory"] = 0 + monkeypatch.setattr(envelope, "_engine_outer_population", lambda _name: current) + if drift in {"none", "threads"}: + envelope._assert_outer_process_ids_unchanged(envelope.DIND_NAME, original) + else: + reason = "resource_cgroup_probe_namespace_invalid" if drift == "identity" else "resource_cgroup_process_mapping_invalid" + with pytest.raises(RuntimeError, match="^" + reason + "$"): + envelope._assert_outer_process_ids_unchanged(envelope.DIND_NAME, original) + + +@pytest.mark.parametrize("drift", ["none", "host_pid", "foreign_cid", "capability", "writable", "host_proc", "source"]) +def test_build_snapshot_inspection_retains_exact_private_boundary(drift): + envelope = _module() + item, image, source = _host_probe_fixture(envelope) + target = {"name": envelope.DIND_NAME, "id": "a" * 64} + item["HostConfig"]["PidMode"] = "container:" + target["id"] + item["Config"]["Labels"].update({"tech.itworx.probe.namespace-name": target["name"], "tech.itworx.probe.namespace-id": target["id"]}) + if drift == "host_pid": + item["HostConfig"]["PidMode"] = "host" + elif drift == "foreign_cid": + item["HostConfig"]["PidMode"] = "container:" + "f" * 64 + elif drift == "capability": + item["HostConfig"]["CapAdd"] = ["DAC_OVERRIDE"] + elif drift == "writable": + item["Mounts"][0]["RW"] = True + elif drift == "host_proc": + item["Mounts"].append({"Type": "bind", "Source": "/proc", "Destination": "/host-proc", "RW": False}) + elif drift == "source": + item["Mounts"][1]["Source"] += "/foreign" + assert envelope._validate_host_probe_inspect(item, mode="build_snapshot", image_id=image, code_source=source, + namespace_target=target) is (drift == "none") + + +@pytest.mark.parametrize("case", ["owned", "missing_proof", "false_proof", "foreign_count", "boolean_count", "extra_field", "running_outer", "wrong_workload", "identity_drift", "pid_drift", "late_runner_start", "late_runner_replaced"]) +def test_collect_build_snapshot_requires_complete_coherent_proof(monkeypatch, case): + envelope = _module() + snapshot, nested_items, _calls = _install_classic_builder_collection(envelope, monkeypatch, {}) + nested_items[:] = nested_items[:1] + nested_items[0]["HostConfig"].update(NanoCpus=4_000_000_000, Memory=6 * 1024**3) + original_inspect = envelope._inspect_container + helper_completed = [] + + def inspect(name): + item = copy.deepcopy(original_inspect(name)) + if name != envelope.DIND_NAME: + item["State"].update(Running=case == "running_outer" and name == "itworx-ci-build", Pid=0) + if helper_completed and name == "itworx-ci-build": + if case == "late_runner_start": + item["State"].update(Running=True, Pid=123) + elif case == "late_runner_replaced": + item["Id"] = "f" * 64 + return item + + monkeypatch.setattr(envelope, "_inspect_container", inspect) + binding = envelope._engine_outer_binding(inspect(envelope.DIND_NAME)) + populations, probes = [], [] + + def population(_name): + populations.append(True) + current = copy.deepcopy(binding) + if len(populations) > 1 and case == "identity_drift": + current["Id"] = "f" * 64 + return current, [(11 if len(populations) > 1 and case == "pid_drift" else 10, len(populations))] + + def probe(operation, payload): + helper_completed.append(True) + probes.append((operation, payload)) + assert operation == "build_snapshot" + assert payload["target"] == {"name": envelope.DIND_NAME, "id": binding["Id"]} + assert payload["processCount"] == 1 + result = {"snapshot": snapshot, "namespaceProof": {"allThreadsContained": True, "processCount": 1, "threadCount": 2}} + if case == "missing_proof": + del result["namespaceProof"] + elif case == "false_proof": + result["namespaceProof"]["allThreadsContained"] = False + elif case in {"foreign_count", "boolean_count"}: + result["namespaceProof"]["processCount"] = 2 if case == "foreign_count" else True + elif case == "extra_field": + result["namespaceProof"]["extra"] = True + return result + + monkeypatch.setattr(envelope, "_engine_outer_population", population) + monkeypatch.setattr(envelope, "_run_host_probe", probe) + candidate = ("f" * 64 if case == "wrong_workload" else nested_items[0]["Id"], nested_items[0]["State"]["Pid"]) + if case == "owned": + assert envelope._collect_snapshot(envelope.FIXED_RUNNER_AND_DIND_NAMES, coherent_build_probe=candidate) == snapshot + assert len(probes) == 1 and len(populations) == 2 + else: + reason = "resource_cgroup_probe_workload_mismatch" if case in {"running_outer", "wrong_workload"} else ( + "resource_cgroup_probe_namespace_invalid" if case in {"identity_drift", "late_runner_start", "late_runner_replaced"} else + "resource_cgroup_process_mapping_invalid" if case == "pid_drift" else "resource_cgroup_probe_snapshot_invalid") + with pytest.raises(RuntimeError, match="^" + reason + "$"): + envelope._collect_snapshot(envelope.FIXED_RUNNER_AND_DIND_NAMES, coherent_build_probe=candidate) + assert len(probes) == (0 if case in {"running_outer", "wrong_workload"} else 1) + + @pytest.mark.parametrize("init,docker,exact", [(False, False, False), (True, False, False), (True, True, False), (True, True, True)]) def test_owned_dind_family_binds_only_existing_exact_private_child(monkeypatch, init, docker, exact): envelope, claim, _rows, _processes, _controls, writes, _inodes = _owned_family_fixture( @@ -1965,7 +2175,7 @@ def test_build_probe_attestor_binds_exact_cid_nested_pid_and_build_cgroup(monkey attempts = [] waits = [] - def collect(_expected): + def collect(_expected, **_kwargs): attempts.append(True) if len(attempts) <= race_count: raise RuntimeError("resource_cgroup_process_mapping_invalid") @@ -2006,7 +2216,7 @@ def test_build_probe_attestor_does_not_retry_other_resource_refusals(monkeypatch envelope = _module() attempts = [] - def collect(_expected): + def collect(_expected, **_kwargs): attempts.append(True) raise RuntimeError(reason) @@ -2017,7 +2227,36 @@ def test_build_probe_attestor_does_not_retry_other_resource_refusals(monkeypatch assert len(attempts) == 1 and "probeWorkload" not in proof -@pytest.mark.parametrize("drift", [None, "binding_before_snapshot", "binding_after_snapshot", "threads_after_snapshot", "binding_and_threads_after_snapshot"]) +@pytest.mark.parametrize("stage", [ + "before_helper_start", "after_helper_exit", "namespace_counts", + "host_population_counts", "host_snapshot_counts", "outer_population_compare", + "after_snapshot", "Bearer secret-value", {"secret": "value"}, None, +]) +def test_build_population_refusal_retains_only_finite_sample_stage(monkeypatch, stage): + envelope = _module() + attempts = [] + sleeps = [] + + def refuse(_names, **_kwargs): + attempts.append(True) + error = RuntimeError("resource_cgroup_process_mapping_invalid") + error.sample_stage = stage + raise error + + monkeypatch.setattr(envelope, "_collect_snapshot", refuse) + monkeypatch.setattr(envelope.time, "sleep", sleeps.append) + result = envelope.attest_build_probe_workload("a" * 64, 77) + assert result["ready"] is False + assert result["reason"] == "resource_cgroup_process_mapping_invalid" + assert len(attempts) == 3 and sleeps == [0.1, 0.1] + if isinstance(stage, str) and stage in envelope.POPULATION_SAMPLE_STAGES: + assert result["sampleStage"] == stage + else: + assert "sampleStage" not in result + assert "secret-value" not in json.dumps(result) + + +@pytest.mark.parametrize("drift", [None, "binding_before_snapshot", "binding_after_snapshot", "threads_after_snapshot", "binding_and_threads_after_snapshot", "inventory_thread_churn", "inventory_pid_churn", "inventory_identity_drift"]) def test_collect_snapshot_uses_host_helper_and_exports_real_family_counters(monkeypatch, drift): envelope = _module() snapshot = _snapshot() @@ -2064,12 +2303,15 @@ def test_collect_snapshot_uses_host_helper_and_exports_real_family_counters(monk } docker_commands = [] + inventory_collected = False def command(args, **_kwargs): + nonlocal inventory_collected docker_commands.append(args) if "info" in args: return "2|cgroupfs|true" if "exec" in args else "2|cgroupfs" if "docker" in args and "ps" in args: + inventory_collected = True return "" raise AssertionError(f"unexpected fixed Docker command: {args}") @@ -2090,6 +2332,17 @@ def test_collect_snapshot_uses_host_helper_and_exports_real_family_counters(monk namespace_calls[name] = namespace_calls.get(name, 0) + 1 binding, pairs, claim = copy.deepcopy(stable_namespace(name)) if name == envelope.DIND_NAME: + # Docker exec used by inventory can start/retire daemon threads or + # processes. Measure only after our own inventory commands finish; + # a changed container identity must still refuse before sampling. + if inventory_collected and drift == "inventory_thread_churn": + pairs[0] = (pairs[0][0], 2) + claim["threadCount"] = 2 + elif inventory_collected and drift == "inventory_pid_churn": + pairs.append((pairs[0][0] + 1000, 1)) + claim.update(processCount=2, threadCount=2) + elif inventory_collected and drift == "inventory_identity_drift": + binding["State"]["StartedAt"] = "changed-start" phase = "before_snapshot" if namespace_calls[name] == 1 else "after_snapshot" if drift and drift.endswith(phase): if drift.startswith("binding"): @@ -2113,7 +2366,12 @@ def test_collect_snapshot_uses_host_helper_and_exports_real_family_counters(monk expected_host_budget={"cpuMilli": 8000, "memoryMiB": 16384}, ) - if drift: + if drift == "inventory_identity_drift": + assert result["ready"] is False + assert result["reason"] == "resource_cgroup_probe_namespace_invalid" + assert probe_calls == [] + return + if drift and not drift.startswith("inventory_"): expected = "resource_cgroup_probe_namespace_invalid" if drift.startswith("binding") else "resource_cgroup_process_mapping_invalid" assert result["ready"] is False and result["reason"] == expected if drift.endswith("before_snapshot"):