Files
Unraid-AppOps-Gateway-Public/backend/tests/test_runner_fleet_specialize.py
T

3465 lines
142 KiB
Python

from __future__ import annotations
import base64
import copy
import hashlib
import importlib.util
import io
import json
import os
import subprocess
import sys
import tarfile
from contextlib import contextmanager, nullcontext
from pathlib import Path
from types import SimpleNamespace
import pytest
@pytest.fixture
def controller_only_source(specialize, tmp_path, monkeypatch):
image = "sha256:" + "a" * 64
code = f"{specialize.RELEASE_VOLUME_HOST_ROOT}/controller-artifacts/{'b' * 40}"
model = {"name": specialize.PROJECT, "services": {"deploy-control": {
"container_name": specialize.CONTROLLER_CONTAINER, "image": image,
"volumes": [{"type": "bind", "source": code, "target": "/opt/gitea-deploy",
"read_only": True, "bind": {"propagation": "rslave"}}],
}}}
compose = tmp_path / "docker-compose.yml"
archive_dir = tmp_path / "backups" / "controller-v2-fixed"
archive_dir.mkdir(parents=True, mode=0o700)
archive_dir.parent.chmod(0o700)
archive = archive_dir / "compose-before.bin"
archived = b"reviewed complete legacy v1 source"
archive.write_bytes(archived)
archive.chmod(0o600)
monkeypatch.setattr(specialize, "ROOT", tmp_path)
monkeypatch.setattr(specialize, "COMPOSE_PATH", compose)
monkeypatch.setattr(specialize, "LEGACY_V1_COMPOSE_BACKUP", archive)
monkeypatch.setattr(specialize, "_KNOWN_V1_COMPOSE_SHA256", hashlib.sha256(archived).hexdigest())
monkeypatch.setattr(specialize, "_controller_runtime_identity", lambda: (image, code))
return model, compose, archive, archived
def test_controller_only_source_uses_fixed_pinned_archive_without_rewriting_live_compose(
specialize, controller_only_source,
):
model, compose, _archive, archived = controller_only_source
current = json.dumps(model).encode()
compose.write_bytes(current)
result, current_sha = specialize._read_fleet_base_compose()
assert result == archived
assert current_sha == hashlib.sha256(current).hexdigest()
assert compose.read_bytes() == current
@pytest.mark.parametrize("mutation", ["project", "image", "container", "code", "writable", "shared", "extra_code"])
def test_controller_only_source_rejects_identity_and_code_binding_drift(
specialize, controller_only_source, mutation,
):
model, compose, _archive, _archived = controller_only_source
controller = model["services"]["deploy-control"]
mount = controller["volumes"][0]
if mutation == "project":
model["name"] = "other"
elif mutation == "image":
controller["image"] = "sha256:" + "c" * 64
elif mutation == "container":
controller["container_name"] = "other"
elif mutation == "code":
mount["source"] = "/mnt/shared/code"
elif mutation == "writable":
mount["read_only"] = False
elif mutation == "shared":
mount["bind"]["propagation"] = "rshared"
else:
controller["volumes"].append(copy.deepcopy(mount))
compose.write_text(json.dumps(model))
with pytest.raises(specialize.RunnerFleetSpecializeError, match="controller_compose_untrusted"):
specialize._read_fleet_base_compose()
def test_controller_only_source_rejects_modified_fixed_archive(specialize, controller_only_source):
model, compose, archive, _archived = controller_only_source
compose.write_text(json.dumps(model))
archive.write_bytes(b"different legacy source")
with pytest.raises(specialize.RunnerFleetSpecializeError, match="legacy_backup_changed"):
specialize._read_fleet_base_compose()
def test_legacy_live_source_does_not_open_promotion_backup(specialize, monkeypatch):
current = b"services:\n runner: {}\n"
monkeypatch.setattr(specialize, "_read_source_file", lambda *_args, **_kwargs: current)
monkeypatch.setattr(specialize, "_trusted_private_file", lambda *_args, **_kwargs: pytest.fail("archive must remain unopened"))
assert specialize._read_fleet_base_compose() == (current, None)
@pytest.mark.parametrize("separate_controller_hash", [False, True])
def test_controller_compose_cas_blocks_rollback_before_any_lifecycle_action(
specialize, controller_only_source, monkeypatch, separate_controller_hash,
):
monkeypatch.setattr(specialize, "_rollback_maintenance", _fixture_rollback_maintenance)
model, compose, _archive, _archived = controller_only_source
compose.write_text(json.dumps(model))
original_sha = hashlib.sha256(compose.read_bytes()).hexdigest()
journal = {"baseComposeSha256": original_sha}
if separate_controller_hash:
journal["controllerComposeSha256"] = original_sha
compose.write_bytes(compose.read_bytes() + b"\n")
monkeypatch.setattr(specialize, "_guard_state_unchanged", lambda *_args: pytest.fail("must reject before lifecycle"))
assert specialize._rollback_before_activation(journal) is False
@pytest.mark.parametrize("reviewed,user,image,expected", [
(True, None, True, "rootless"), (False, None, True, None),
(True, "root", True, "root"), (True, None, False, None),
])
def test_only_byte_pinned_v1_pinned_image_default_user_is_normalized(
specialize, monkeypatch, reviewed, user, image, expected,
):
source = b"reviewed v1 source" if reviewed else b"different source"
monkeypatch.setattr(specialize, "_KNOWN_V1_COMPOSE_SHA256", hashlib.sha256(b"reviewed v1 source").hexdigest())
model = {"services": {"runner-dind": {"image": specialize._DIND_IMAGE_REFERENCE if image else "unknown", "user": user}}}
monkeypatch.setattr(specialize, "_run_compose", lambda *_args: json.dumps(model))
assert specialize._compose_model(source)["services"]["runner-dind"]["user"] == expected
ROOT = Path(__file__).resolve().parents[2]
BROKER = ROOT / "ops" / "gitea-deploy-broker"
SPECIALIZE_PATH = BROKER / "runner_fleet_specialize.py"
@pytest.fixture
def specialize(monkeypatch):
monkeypatch.syspath_prepend(str(BROKER))
spec = importlib.util.spec_from_file_location("runner_fleet_specialize_test", SPECIALIZE_PATH)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
return module
@contextmanager
def _fixture_rollback_maintenance():
yield lambda: None
def _valid_task_image_manifest(specialize, revision="a" * 40, tree="b" * 40):
source = "https://gitea.example.com/Jens/Unraid-AppOps-Gateway"
return {
"apiVersion": specialize.IMAGE_MANIFEST_SCHEMA,
"sourceRepository": specialize.SOURCE_REPOSITORY,
"sourceRevision": revision,
"sourceTree": tree,
"workflowPath": specialize.IMAGE_WORKFLOW,
"workflowRunId": "12",
"fullRunId": "13",
"artifactName": specialize.IMAGE_ARTIFACT_PREFIX + revision,
"dockerTarSha256": "c" * 64,
"dockerTarBytes": 1,
"images": [
{
"profile": profile_id,
"reference": f"localhost/itworx-ci-{profile_id}:sha-{revision}",
"imageId": f"sha256:{str(index + 1) * 64}",
"architecture": "amd64",
"dockerfileSha256": "d" * 64,
"labels": {
"org.opencontainers.image.source": source,
"org.opencontainers.image.revision": revision,
"tech.itworx.source-tree": tree,
"tech.itworx.ci.profile": profile_id,
},
}
for index, profile_id in enumerate(specialize.PROFILE_IDS)
],
}
def _build_probe_record(specialize, *, container_id=None, status="intent"):
revision = "a" * 40
tree = "b" * 40
image_labels = {
"org.opencontainers.image.source": "https://gitea.example.com/Jens/Unraid-AppOps-Gateway",
"org.opencontainers.image.revision": revision,
"tech.itworx.source-tree": tree,
"tech.itworx.ci.profile": "build",
}
record = specialize._new_build_probe_record(
{"sourceRevision": revision, "sourceTree": tree},
{
"taskImageId": "sha256:" + "1" * 64,
"dockerfileSha256": "d" * 64,
"imageLabels": image_labels,
},
)
record["status"] = status
record["containerId"] = container_id
return record
def _build_probe_inspect(specialize, record, *, container_id=None, changes=None):
image_environment = ["PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"]
item = {
"Id": container_id or "e" * 64,
"Name": f"/{record['name']}",
"Image": record["taskImageId"],
"Config": {
"Image": record["taskImageId"],
"Entrypoint": [specialize._BUILD_PROBE_EXECUTABLE],
"Cmd": [
record["sourceRevision"],
record["sourceTree"],
"--hold-seconds",
str(specialize._BUILD_PROBE_HOLD_SECONDS),
],
"Env": image_environment,
"Labels": specialize._build_probe_container_labels(record),
"Volumes": {"/home/itworx-builder/.local/share/docker": {}},
},
"HostConfig": {
"NetworkMode": "bridge",
"PidMode": "",
"IpcMode": "private",
"CgroupnsMode": "private",
"CgroupParent": "docker",
"NanoCpus": 4_000_000_000,
"Memory": 6 * 1024**3,
"PidsLimit": 1024,
"BlkioWeight": 100,
"Privileged": False,
"CapAdd": [],
"CapDrop": [],
"SecurityOpt": ["seccomp=unconfined", "systempaths=unconfined"],
"Devices": [{
"PathOnHost": "/dev/net/tun",
"PathInContainer": "/dev/net/tun",
"CgroupPermissions": "rwm",
}],
"DeviceRequests": [],
"DeviceCgroupRules": [],
"ExtraHosts": ["gitea.example.com:198.19.104.1"],
"PortBindings": {},
"Binds": [],
"Mounts": [],
"VolumesFrom": [],
"Tmpfs": {},
"RestartPolicy": {"Name": "no"},
},
"Mounts": [{
"Type": "volume", "Name": "9" * 64,
"Source": f"/var/lib/docker/volumes/{'9' * 64}/_data",
"Destination": "/home/itworx-builder/.local/share/docker",
"Driver": "local", "Mode": "", "RW": True, "Propagation": "",
}],
"State": {"Running": True, "Pid": 77},
}
if changes:
for section, values in changes.items():
item[section].update(values)
return item, image_environment
def _write_task_image_archive(specialize, path: Path, *, extra_label: bool = False):
images = _valid_task_image_manifest(specialize)["images"]
docker_manifest = []
members: dict[str, bytes] = {}
for index, image in enumerate(images):
labels = specialize._runtime_task_image_labels(image)
assert labels is not None
if extra_label and index == 0:
labels["untrusted.extra"] = "value"
config = {
"architecture": "amd64",
"os": "linux",
"config": {"Labels": labels},
}
config_bytes = json.dumps(config, sort_keys=True, separators=(",", ":")).encode()
config_name = f"{image['profile']}.json"
layer_name = f"{image['profile']}/layer.tar"
image["imageId"] = "sha256:" + hashlib.sha256(config_bytes).hexdigest()
members[config_name] = config_bytes
members[layer_name] = b"layer"
docker_manifest.append(
{
"Config": config_name,
"RepoTags": [image["reference"]],
"Layers": [layer_name],
}
)
members["manifest.json"] = json.dumps(docker_manifest).encode()
with tarfile.open(path, mode="w:") as archive:
for name, payload in members.items():
info = tarfile.TarInfo(name)
info.size = len(payload)
archive.addfile(info, io.BytesIO(payload))
return images
def _active_recovery_fixture(specialize):
rows = []
registration_ids = {}
originals = []
containers = {}
for index, profile in enumerate(specialize.RUNNER_PROFILES.values(), start=100):
registration_ids[profile.id] = index
rows.append(
{
"name": profile.container_name,
"displayName": profile.display_name,
"id": index,
"labels": sorted({profile.runner_label, *profile.legacy_labels}),
"disabled": True,
"busy": False,
"status": "offline",
}
)
containers[profile.container_name] = {"State": {"Running": False}}
for index, name in enumerate(specialize._LEGACY_V1_NAMES, start=9):
labels = sorted(specialize._legacy_allowed_labels("v1", name))
legacy_row = {
"name": name,
"id": index,
"labels": labels,
"disabled": True,
"busy": False,
"status": "offline",
}
rows.append(legacy_row)
originals.append({"name": name, "id": index, "labels": labels, "disabled": True})
dind = {"State": {"Running": False, "Health": {"Status": "none"}}}
containers[specialize.DIND_NAME] = dind
journal = {
"legacyTopology": "v1",
"newRegistrationIds": registration_ids,
"originalRegistrations": originals,
}
model = {"services": {"runner-dind": {"privileged": True}}}
return rows, containers, dind, journal, model
def _legacy_dind_mounts(specialize):
return [
{
"type": "bind",
"source": specialize.LEGACY_DIND_STORAGE_ROOT.as_posix(),
"target": "/home/rootless/.local/share/docker",
"read_only": False,
},
{
"type": "bind",
"source": "/mnt/cache/appdata/itworx-security-runner-workspace",
"target": "/workspace",
"read_only": False,
},
{
"type": "bind",
"source": "/mnt/cache/appdata/itworx-validation-runner-2-workspace",
"target": "/workspace-secondary",
"read_only": False,
},
]
def _specialized_dind_mounts(specialize):
return [
{
"type": "bind",
"source": str(specialize._host_volume_path(specialize.DIND_STORAGE_ROOT)),
"target": "/var/lib/docker",
"read_only": False,
},
{
"type": "bind",
"source": str(specialize._host_volume_path(specialize._DIND_CONTROL_ROOT)),
"target": specialize._DIND_CONTROL_CONTAINER_ROOT,
"read_only": False,
},
{
"type": "bind",
"source": str(specialize._host_volume_path(specialize._DIND_ROOTLESS_SHADOW_ROOT)),
"target": "/home/rootless/.local/share/docker",
"read_only": True,
},
*[
{
"type": "bind",
"source": str(
specialize._host_volume_path(
specialize._SPECIALIZED_WORKSPACE_ROOT / profile.container_name
)
),
"target": f"/{profile.container_name}-workspace",
"read_only": False,
}
for profile in specialize.RUNNER_PROFILES.values()
],
]
def _base_v1_compose_model(specialize):
services = {
"deploy-control": {
"container_name": specialize.CONTROLLER_CONTAINER,
"image": "itworx/appops-deploy-control:fixed",
},
"runner-dind": {
"container_name": specialize.DIND_NAME,
"image": "docker@sha256:7c3e797187e43738220462658f4586572cbd3bf009f728b21e34d9c5c06ce431",
"user": "rootless",
"privileged": True,
"cpus": 6,
"mem_limit": "12g",
"restart": "unless-stopped",
"security_opt": ["label=disable"],
"command": ["--default-address-pool=base=198.19.0.0/16,size=27"],
"volumes": _legacy_dind_mounts(specialize),
},
}
for name, service_name in specialize._OLD_SERVICES.items():
if name not in specialize._LEGACY_V1_NAMES:
continue
services[service_name] = {
"container_name": name,
"image": specialize.RUNNER_IMAGE,
"profiles": ["runner"],
"cpus": 1,
"mem_limit": "1g",
"volumes": [
{
"type": "bind",
"source": source,
"target": target,
"read_only": read_only,
}
for source, target, read_only in specialize._expected_legacy_mounts("v1", name)
],
"labels": {"tech.itworx.role": "gitea-actions-runner"},
}
return {
"services": services,
"networks": {
"runner-sandbox": {"name": specialize._RUNNER_NETWORK_NAME}
},
}
def _gitea_28_run(
specialize, revision, *, run_id=21533, attempt=1, status="completed", conclusion="success"
):
filename = specialize.FULL_WORKFLOW.rsplit("/", 1)[-1]
return {
"id": run_id,
"run_attempt": attempt,
"path": f"{filename}@refs/heads/main",
"status": status,
"conclusion": conclusion,
"event": "push",
"head_branch": "main",
"head_sha": revision,
"repository": {"id": 9, "full_name": "Jens/Unraid-AppOps-Gateway"},
"head_repository": None,
}
def _gitea_workflow_api(specialize, rows, *, run_pages=None, calls=None):
run_pages = run_pages or {1: rows}
calls = calls if calls is not None else []
def api(path):
calls.append(path)
if path.endswith("/actions/workflows?limit=100&page=1"):
return {
"workflows": [
{
"id": workflow.rsplit("/", 1)[-1],
"path": workflow,
"state": "active",
}
for workflow in (specialize.IMAGE_WORKFLOW, specialize.FULL_WORKFLOW)
]
}
if "/actions/runs?" in path:
page = int(path.rsplit("page=", 1)[-1])
return {"workflow_runs": run_pages.get(page, [])}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
return api, calls
def test_gitea_workflow_inventory_uses_exact_active_filename_and_path(
specialize, monkeypatch
):
workflow = specialize.FULL_WORKFLOW
filename = workflow.rsplit("/", 1)[-1]
inventory = {
"workflows": [
{"id": filename, "path": workflow, "state": "active"},
{
"id": specialize.IMAGE_WORKFLOW.rsplit("/", 1)[-1],
"path": specialize.IMAGE_WORKFLOW,
"state": "active",
},
]
}
monkeypatch.setattr(specialize, "_api_json", lambda _path: inventory)
assert specialize._workflow_id(workflow) == filename
for bad in (
{"id": 34, "path": workflow, "state": "active"},
{"id": filename, "path": ".gitea/workflows/other.yml", "state": "active"},
{"id": filename, "path": workflow, "state": "enabled"},
):
monkeypatch.setattr(
specialize, "_api_json", lambda _path, row=bad: {"workflows": [row]}
)
with pytest.raises(
specialize.RunnerFleetSpecializeError, match="workflow_unavailable"
):
specialize._workflow_id(workflow)
def test_successful_main_run_accepts_actual_gitea_run_without_workflow_id(
specialize, monkeypatch
):
revision = "a" * 40
run = _gitea_28_run(specialize, revision)
api, calls = _gitea_workflow_api(specialize, [run])
monkeypatch.setattr(specialize, "_api_json", api)
selected = specialize._successful_main_run(specialize.FULL_WORKFLOW, revision)
assert selected is run
assert "workflow_id" not in selected
assert selected["head_repository"] is None
assert calls[0].endswith("/actions/workflows?limit=100&page=1")
assert calls[1].endswith("/actions/runs?branch=main&limit=100&page=1")
def test_source_image_fetch_accepts_actual_gitea_commit_tree_shape(
specialize, monkeypatch, tmp_path
):
revision = "a" * 40
tree = "b" * 40
calls = []
workflow_calls = []
def api(path):
calls.append(path)
if path.endswith("/branches/main"):
return {"commit": {"id": revision}}
if path.endswith(f"/git/commits/{revision}"):
return {"sha": revision, "commit": {"tree": {"sha": tree}}}
if "/actions/runs/" in path and "/artifacts?" in path:
return {
"artifacts": [
{
"name": specialize.IMAGE_ARTIFACT_PREFIX + revision,
"expired": False,
"id": 44,
"size_in_bytes": 1,
}
]
}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
def successful_run(workflow_path, requested_revision):
assert requested_revision == revision
workflow_calls.append(workflow_path)
return {"id": 41 if workflow_path == specialize.IMAGE_WORKFLOW else 42}
class ReachedArtifactDownload(Exception):
pass
def download(_artifact_id, _destination):
raise ReachedArtifactDownload
monkeypatch.setattr(specialize, "_api_json", api)
monkeypatch.setattr(specialize, "_successful_main_run", successful_run)
monkeypatch.setattr(specialize, "_download_artifact_zip", download)
with pytest.raises(ReachedArtifactDownload):
specialize._fetch_source_image_artifact(revision, tree, tmp_path)
assert workflow_calls == [specialize.IMAGE_WORKFLOW, specialize.FULL_WORKFLOW]
assert calls == [
f"/repos/{specialize.SOURCE_REPOSITORY_API}/branches/main",
f"/repos/{specialize.SOURCE_REPOSITORY_API}/git/commits/{revision}",
f"/repos/{specialize.SOURCE_REPOSITORY_API}/actions/runs/41/artifacts?name={specialize.IMAGE_ARTIFACT_PREFIX}{revision}",
]
@pytest.mark.parametrize(
("commit_value", "error"),
[
({"sha": "a" * 40}, "source_tree_mismatch"),
({"sha": "a" * 40, "commit": {}}, "source_tree_mismatch"),
(
{"sha": "a" * 40, "commit": {"tree": {"sha": "c" * 40}}},
"source_tree_mismatch",
),
(
{"sha": "a" * 40, "tree": {"sha": "b" * 40}},
"source_tree_mismatch",
),
(
{"sha": "c" * 40, "commit": {"tree": {"sha": "b" * 40}}},
"source_revision_mismatch",
),
(
{"commit": {"tree": {"sha": "b" * 40}}},
"source_revision_mismatch",
),
],
)
def test_source_image_fetch_rejects_noncanonical_gitea_commit_payload_before_artifact_lookup(
specialize, monkeypatch, tmp_path, commit_value, error
):
revision = "a" * 40
calls = []
workflow_calls = []
def api(path):
calls.append(path)
if path.endswith("/branches/main"):
return {"commit": {"id": revision}}
if path.endswith(f"/git/commits/{revision}"):
return commit_value
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
def unexpected_workflow(*_args):
workflow_calls.append(True)
raise AssertionError("workflow lookup must not follow an untrusted commit response")
monkeypatch.setattr(specialize, "_api_json", api)
monkeypatch.setattr(specialize, "_successful_main_run", unexpected_workflow)
monkeypatch.setattr(
specialize,
"_download_artifact_zip",
lambda *_args: pytest.fail("artifact download must not follow an untrusted commit response"),
)
with pytest.raises(specialize.RunnerFleetSpecializeError, match=error):
specialize._fetch_source_image_artifact(revision, "b" * 40, tmp_path)
assert calls == [
f"/repos/{specialize.SOURCE_REPOSITORY_API}/branches/main",
f"/repos/{specialize.SOURCE_REPOSITORY_API}/git/commits/{revision}",
]
assert workflow_calls == []
@pytest.mark.parametrize(
"changes",
[
{"path": "managed-validation.yml@refs/heads/feature"},
{"repository": {"full_name": "attacker/Unraid-AppOps-Gateway"}},
{"head_repository": {"full_name": "attacker/Unraid-AppOps-Gateway"}},
{"head_sha": "b" * 40},
{"event": "workflow_dispatch"},
{"head_branch": "feature"},
{"pull_requests": [{"number": 123}]},
],
)
def test_successful_main_run_rejects_untrusted_gitea_identity(
specialize, monkeypatch, changes
):
revision = "a" * 40
run = {**_gitea_28_run(specialize, revision), **changes}
api, _calls = _gitea_workflow_api(specialize, [run])
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="protected_validation_missing",
):
specialize._successful_main_run(specialize.FULL_WORKFLOW, revision)
@pytest.mark.parametrize(
("latest", "error"),
[
({"status": "in_progress", "conclusion": None}, "protected_validation_missing"),
({"status": "completed", "conclusion": "failure"}, "protected_validation_missing"),
(
{"run_attempt": 2, "status": "completed", "conclusion": "failure"},
"protected_validation_missing",
),
({"run_attempt": None}, "workflow_runs_invalid"),
],
)
def test_newest_matching_gitea_run_or_attempt_supersedes_prior_success(
specialize, monkeypatch, latest, error
):
revision = "a" * 40
successful = _gitea_28_run(specialize, revision, run_id=21532)
newer = {
**_gitea_28_run(specialize, revision, run_id=21533),
**latest,
}
if latest.get("run_attempt") == 2:
newer["id"] = successful["id"]
api, _calls = _gitea_workflow_api(specialize, [successful, newer])
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(specialize.RunnerFleetSpecializeError, match=error):
specialize._successful_main_run(specialize.FULL_WORKFLOW, revision)
def test_workflow_run_pagination_fails_closed_at_fixed_page_limit(specialize, monkeypatch):
revision = "a" * 40
wrong = {"path": "unrelated.yml@refs/heads/main"}
pages = {page: [wrong] * 100 for page in range(1, 11)}
api, calls = _gitea_workflow_api(specialize, [], run_pages=pages)
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError, match="workflow_runs_invalid"
):
specialize._successful_main_run(specialize.FULL_WORKFLOW, revision)
assert len([path for path in calls if "/actions/runs?" in path]) == 10
def test_archive_verification_hashes_large_tar_without_read_bytes(specialize, tmp_path, monkeypatch):
archive_path = tmp_path / "images.tar"
payload = b"bounded-stream" * 4096
with tarfile.open(archive_path, "w") as archive:
member = tarfile.TarInfo("layer.tar")
member.size = len(payload)
archive.addfile(member, io.BytesIO(payload))
digest = hashlib.sha256(archive_path.open("rb").read()).hexdigest()
def forbidden_read_bytes(_path):
raise AssertionError("archive verifier must stream the tar file")
monkeypatch.setattr(Path, "read_bytes", forbidden_read_bytes)
specialize._verify_docker_archive(archive_path, archive_path.stat().st_size, digest)
def test_archive_verifier_requires_exact_pinned_inherited_labels(
specialize, tmp_path
):
archive_path = tmp_path / "task-images.tar"
images = _write_task_image_archive(specialize, archive_path)
specialize._verify_docker_archive(
archive_path,
archive_path.stat().st_size,
hashlib.sha256(archive_path.read_bytes()).hexdigest(),
images,
)
images = _write_task_image_archive(specialize, archive_path, extra_label=True)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="image_archive_identity_invalid",
):
specialize._verify_docker_archive(
archive_path,
archive_path.stat().st_size,
hashlib.sha256(archive_path.read_bytes()).hexdigest(),
images,
)
def test_installed_image_manifest_requires_complete_source_and_profile_pins(specialize):
manifest = _valid_task_image_manifest(specialize)
assert specialize._valid_installed_manifest(manifest) is True
manifest["images"][0]["reference"] = "localhost/itworx-ci-checks:latest"
assert specialize._valid_installed_manifest(manifest) is False
def test_task_image_runtime_verifier_accepts_only_exact_pinned_inherited_labels(
specialize, monkeypatch
):
expected = _valid_task_image_manifest(specialize)["images"][0]
runtime_labels = specialize._runtime_task_image_labels(expected)
assert runtime_labels is not None
assert len(runtime_labels) == 14
actual = {
"Id": expected["imageId"],
"Os": "linux",
"Architecture": "amd64",
"Config": {"Labels": runtime_labels},
}
monkeypatch.setattr(
specialize,
"_run",
lambda *_args, **_kwargs: json.dumps([actual]),
)
assert specialize._verify_manifest_images_in_dind([expected]) is True
actual["Config"]["Labels"]["untrusted.extra"] = "value"
assert specialize._verify_manifest_images_in_dind([expected]) is False
del actual["Config"]["Labels"]["untrusted.extra"]
del expected["labels"]["tech.itworx.source-tree"]
assert specialize._verify_manifest_images_in_dind([expected]) is False
def test_bootstrap_receipt_is_source_bound_private_and_cas_consumed(
specialize, tmp_path, monkeypatch
):
state_root = tmp_path / "runner-fleet-state"
state_root.mkdir(mode=0o700)
os.chmod(state_root, 0o700)
manifest_path = state_root / "task-image-manifest.json"
marker_path = state_root / "specialized.json"
receipt_path = state_root / "admission-bootstrap.json"
manifest_bytes = json.dumps(
_valid_task_image_manifest(specialize), sort_keys=True, separators=(",", ":")
).encode() + b"\n"
manifest_path.write_bytes(manifest_bytes)
os.chmod(manifest_path, 0o600)
manifest_sha = hashlib.sha256(manifest_bytes).hexdigest()
marker = {
"schemaVersion": 1,
"status": "active",
"sourceScriptSha256": "e" * 64,
"imageManifestSha256": manifest_sha,
"imageSourceRevision": "a" * 40,
"imageSourceTree": "b" * 40,
"verifiedRunners": list(specialize._SPECIALIZED_NAMES),
"resourceEnvelopeDigest": "f" * 64,
"updatedAt": "2026-10-01T12:00:00Z",
}
marker_path.write_text(json.dumps(marker, sort_keys=True, separators=(",", ":")) + "\n")
os.chmod(marker_path, 0o600)
registrations = []
for profile in specialize.RUNNER_PROFILES.values():
registrations.append(
{
"profileId": profile.id,
"displayName": profile.display_name,
"containerName": profile.container_name,
"id": len(registrations) + 10,
"labels": sorted({profile.runner_label, *profile.legacy_labels}),
}
)
receipt = {
"schemaVersion": 1,
"status": "pending",
"sourceScriptSha256": marker["sourceScriptSha256"],
"imageManifestSha256": manifest_sha,
"imageSourceRevision": "a" * 40,
"imageSourceTree": "b" * 40,
"verifiedRunners": list(specialize._SPECIALIZED_NAMES),
"registrations": registrations,
"resourceEnvelopeDigest": marker["resourceEnvelopeDigest"],
"createdAt": "2026-10-01T12:00:00Z",
}
receipt_path.write_text(json.dumps(receipt, sort_keys=True, separators=(",", ":")) + "\n")
os.chmod(receipt_path, 0o600)
assert specialize._valid_admission_bootstrap_receipt(receipt)
validated = specialize.read_admission_bootstrap_receipt(
receipt_path, marker_path=marker_path, image_manifest_path=manifest_path
)
assert validated is not None
pending_sha = validated[1]
assert validated[0]["status"] == "pending"
assert specialize.consume_admission_bootstrap_receipt(
pending_sha,
receipt_path,
marker_path=marker_path,
image_manifest_path=manifest_path,
)
consumed = specialize.read_admission_bootstrap_receipt(
receipt_path, marker_path=marker_path, image_manifest_path=manifest_path
)
assert consumed is not None
assert consumed[0]["status"] == "consumed"
assert consumed[1] == pending_sha
assert specialize.consume_admission_bootstrap_receipt(
pending_sha,
receipt_path,
marker_path=marker_path,
image_manifest_path=manifest_path,
)
assert not specialize.consume_admission_bootstrap_receipt(
"0" * 64,
receipt_path,
marker_path=marker_path,
image_manifest_path=manifest_path,
)
def test_runner_config_is_profile_locked_and_only_backend_frontend_keep_legacy_labels(specialize):
revision = "a" * 40
checks = specialize._render_runner_config("checks", revision).decode()
backend = specialize._render_runner_config("backend", revision).decode()
frontend = specialize._render_runner_config("frontend", revision).decode()
security = specialize._render_runner_config("security", revision).decode()
assert "itworx-ci-checks:docker://localhost/itworx-ci-checks:sha-" in checks
assert "linux-validation" not in checks and "ubuntu-latest" not in checks
assert "linux-validation" in backend and "ubuntu-latest" in backend
assert "linux-validation" in frontend and "ubuntu-latest" in frontend
assert "linux-validation" not in security and "ubuntu-latest" not in security
for profile_id, profile in specialize.RUNNER_PROFILES.items():
config = specialize._render_runner_config(profile_id, revision).decode()
task_image = specialize._task_image_reference(profile_id, revision)
job_cpu_milli, job_memory_mib = specialize._runner_job_limits(profile_id)
cpu = f"{job_cpu_milli / 1000:.3f}".rstrip("0").rstrip(".")
assert f"--cpus={cpu} --memory={job_memory_mib}m" in config
assert f"source=itworx-ci-{profile_id}-cache,target=/opt/itworx-ci-cache" in config
assert f'valid_volumes:\n - "itworx-ci-{profile_id}-cache"' in config
assert "[0-9]*/**" not in config
assert "DOCKER_HOST:" not in config
assert 'docker_host: "-"' in config
assert "host.docker.internal" not in config
assert all(
f"docker://{task_image}" in line
for line in config.splitlines()
if ":docker://" in line
)
if profile_id == "build":
assert "--security-opt=seccomp=unconfined" in config
assert "--security-opt=systempaths=unconfined" in config
assert "--device=/dev/net/tun:/dev/net/tun:rwm" in config
else:
assert "--security-opt=" not in config
assert "--device=" not in config
def test_fixed_workload_reservations_are_subtracted_from_aggregate_profiles(specialize):
for profile_id in ("backend", "frontend", "browser"):
profile = specialize.RUNNER_PROFILES[profile_id]
service_cpu, service_memory = specialize.FIXED_WORKLOAD_RESERVATION[profile_id]
job_cpu, job_memory = specialize._runner_job_limits(profile_id)
assert job_cpu + service_cpu == profile.cpu_milli
assert job_memory + service_memory == profile.memory_mib
config = specialize._render_runner_config(profile_id, "a" * 40).decode()
cpu = f"{job_cpu / 1000:.3f}".rstrip("0").rstrip(".")
assert f"--cpus={cpu} --memory={job_memory}m" in config
for profile_id in ("checks", "security"):
profile = specialize.RUNNER_PROFILES[profile_id]
assert specialize._runner_job_limits(profile_id) == (
profile.cpu_milli,
profile.memory_mib,
)
assert specialize._runner_job_limits("build") == (4000, 6144)
def test_base_compose_model_explicitly_resolves_the_runner_profile(specialize, monkeypatch):
compose = b"captured v1 source"
base = _base_v1_compose_model(specialize)
monkeypatch.setattr(
specialize,
"_KNOWN_V1_COMPOSE_SHA256",
hashlib.sha256(compose).hexdigest(),
)
calls = []
def run_compose(_compose_bytes, *arguments, **_kwargs):
calls.append(arguments)
return json.dumps(base)
monkeypatch.setattr(specialize, "_run_compose", run_compose)
assert specialize._validate_base_composition(compose) == "v1"
assert calls == [("--profile", "runner", "config", "--format", "json")]
def test_rendered_compose_uses_fixed_rootful_dind_and_shared_workspaces(specialize):
base = _base_v1_compose_model(specialize)
rendered = json.loads(specialize._render_specialized_compose(base))
services = rendered["services"]
dind = services["runner-dind"]
assert services["deploy-control"] == base["services"]["deploy-control"]
assert dind["user"] == "0"
assert dind["command"][0] == "dockerd"
assert dind["command"][1] == f"--host={specialize._DIND_DOCKER_HOST}"
assert "--data-root=/var/lib/docker" in dind["command"]
assert "--exec-opt=native.cgroupdriver=cgroupfs" in dind["command"]
assert dind["environment"] == {"DOCKER_HOST": specialize._DIND_DOCKER_HOST}
assert _mounts(specialize, dind) == {
(str(specialize._host_volume_path(specialize.DIND_STORAGE_ROOT)), "/var/lib/docker", False),
(
str(specialize._host_volume_path(specialize._DIND_CONTROL_ROOT)),
specialize._DIND_CONTROL_CONTAINER_ROOT,
False,
),
(
str(specialize._host_volume_path(specialize._DIND_ROOTLESS_SHADOW_ROOT)),
"/home/rootless/.local/share/docker",
True,
),
*{
(
str(specialize._host_volume_path(specialize._SPECIALIZED_WORKSPACE_ROOT / profile.container_name)),
f"/{profile.container_name}-workspace",
False,
)
for profile in specialize.RUNNER_PROFILES.values()
},
}
assert len([volume for volume in dind["volumes"] if volume["type"] == "bind"]) == 9
expected_services = {"deploy-control", "runner-dind"}
expected_services.update(
specialize._OLD_SERVICES[name] for name in specialize._LEGACY_V1_NAMES
)
expected_services.update(f"runner-{profile_id}" for profile_id in specialize.PROFILE_IDS)
assert set(services) == expected_services
for profile_id in specialize.PROFILE_IDS:
service = services[f"runner-{profile_id}"]
assert service["environment"]["DOCKER_HOST"] == specialize._DIND_DOCKER_HOST
assert service["extra_hosts"] == ["gitea.example.com:198.19.104.1"]
assert (
str(specialize._host_volume_path(specialize._DIND_CONTROL_ROOT)),
specialize._DIND_CONTROL_CONTAINER_ROOT,
True,
) in specialize._compose_mounts(service)
@pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="requires POSIX root ownership")
def test_profile_state_creation_uses_effective_runner_image_owners_and_resumes(
specialize, tmp_path, monkeypatch
):
state = tmp_path / "state"
state.mkdir(mode=0o700)
data_root = state / "runner-data"
workspace_root = state / "workspaces"
config_root = state / "runner-configs"
image_root = state / "task-images"
dind_root = state / "dind" / "docker"
shadow_root = state / "dind" / "rootless-unused"
control_root = state / "dind-control"
for name, value in (
("_SPECIALIZED_DATA_ROOT", data_root),
("_SPECIALIZED_WORKSPACE_ROOT", workspace_root),
("RUNNER_CONFIG_ROOT", config_root),
("TASK_IMAGE_ROOT", image_root),
("DIND_STORAGE_ROOT", dind_root),
("_DIND_ROOTLESS_SHADOW_ROOT", shadow_root),
("_DIND_CONTROL_ROOT", control_root),
):
monkeypatch.setattr(specialize, name, value)
control_root.mkdir(mode=0o700)
specialize._prepare_profile_state("a" * 40)
for profile_id, profile in specialize.RUNNER_PROFILES.items():
data_path = data_root / profile_id
workspace_path = workspace_root / profile.container_name
assert data_path.stat().st_uid == 0
assert workspace_path.stat().st_uid == 1000
assert control_root.stat().st_uid == 0
assert os.stat(control_root).st_mode & 0o777 == 0o700
config_path = config_root / "checks.yaml"
expected = specialize._render_runner_config("checks", "a" * 40)
invalid_configs = (
expected.replace(
b'valid_volumes:\n - "itworx-ci-checks-cache"',
b'valid_volumes:\n - "/itworx-ci-checks-workspace/[0-9]*/**"',
1,
),
expected.replace(
b'valid_volumes:\n - "itworx-ci-checks-cache"',
b'valid_volumes:\n - "/run/itworx-ci-control/**"',
1,
),
expected.replace(
b' docker_host: "-"',
b' docker_host: "tcp://runner-dind:2375"',
1,
),
expected.replace(
b" envs:\n",
b" envs:\n DOCKER_HOST: tcp://runner-dind:2375\n",
1,
),
)
for invalid_config in invalid_configs:
config_path.write_bytes(invalid_config)
os.chmod(config_path, 0o600)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="profile_config_adoption_rejected",
):
specialize._prepare_profile_state("a" * 40)
config_path.write_bytes(expected)
os.chmod(config_path, 0o600)
specialize._prepare_profile_state("a" * 40)
def _mounts(specialize, service):
return specialize._compose_mounts(service)
def test_rendered_compose_uses_only_canonical_protected_volume_sources(specialize):
base = _base_v1_compose_model(specialize)
rendered = json.loads(specialize._render_specialized_compose(base))
services = rendered["services"]
assert specialize._compose_mounts(services["runner-dind"]) == {
(str(specialize._host_volume_path(specialize.DIND_STORAGE_ROOT)), "/var/lib/docker", False),
(
str(specialize._host_volume_path(specialize._DIND_CONTROL_ROOT)),
specialize._DIND_CONTROL_CONTAINER_ROOT,
False,
),
(
str(specialize._host_volume_path(specialize._DIND_ROOTLESS_SHADOW_ROOT)),
"/home/rootless/.local/share/docker",
True,
),
*{
(
str(specialize._host_volume_path(specialize._SPECIALIZED_WORKSPACE_ROOT / profile.container_name)),
f"/{profile.container_name}-workspace",
False,
)
for profile in specialize.RUNNER_PROFILES.values()
},
}
assert services["runner-dind"]["restart"] == "no"
for profile_id, profile in specialize.RUNNER_PROFILES.items():
service = services[f"runner-{profile_id}"]
assert specialize._compose_mounts(service) == {
(str(specialize._host_volume_path(specialize._SPECIALIZED_DATA_ROOT / profile_id)), "/data", False),
(str(specialize._host_volume_path(specialize.RUNNER_CONFIG_ROOT / f"{profile_id}.yaml")), "/runner-config.yaml", True),
(str(specialize._host_volume_path(specialize._DIND_CONTROL_ROOT)), specialize._DIND_CONTROL_CONTAINER_ROOT, True),
(str(specialize._host_volume_path(specialize._SPECIALIZED_WORKSPACE_ROOT / profile.container_name)), f"/{profile.container_name}-workspace", False),
}
assert "-f" in specialize._compose_command("up", "-d")
assert specialize._compose_command("up", "-d")[specialize._compose_command("up", "-d").index("-f") + 1] == "-"
def test_specialized_compose_validation_rejects_socket_and_gateway_broadening(
specialize, monkeypatch
):
rendered = json.loads(
specialize._render_specialized_compose(_base_v1_compose_model(specialize))
)
mutations = (
lambda model: model["services"]["runner-dind"]["environment"].update(
DOCKER_HOST="tcp://0.0.0.0:2375"
),
lambda model: next(
volume
for volume in model["services"]["runner-dind"]["volumes"]
if volume["target"] == specialize._DIND_CONTROL_CONTAINER_ROOT
).update(read_only=True),
lambda model: next(
volume
for volume in model["services"]["runner-frontend"]["volumes"]
if volume["target"] == specialize._DIND_CONTROL_CONTAINER_ROOT
).update(read_only=False),
lambda model: model["services"]["runner-frontend"]["extra_hosts"].append(
"host.docker.internal:host-gateway"
),
lambda model: model["services"]["runner-frontend"].update(
network_mode="host"
),
lambda model: model["services"]["runner-frontend"].update(
privileged=True
),
lambda model: model["networks"]["runner-sandbox"].update(
name="runner-sandbox"
),
)
for mutate in mutations:
candidate = copy.deepcopy(rendered)
mutate(candidate)
monkeypatch.setattr(specialize, "_compose_model", lambda _raw, value=candidate: value)
with pytest.raises(specialize.RunnerFleetSpecializeError):
specialize._validate_composition(b"candidate")
def test_specialized_compose_validation_accepts_real_compose_normalization(
specialize, monkeypatch
):
normalized = json.loads(
specialize._render_specialized_compose(_base_v1_compose_model(specialize))
)
service = normalized["services"]["runner-frontend"]
service.pop("privileged")
service["extra_hosts"] = ["gitea.example.com=198.19.104.1"]
service["networks"] = {"runner-sandbox": {"aliases": ["itworx-ci-frontend"]}}
monkeypatch.setattr(specialize, "_compose_model", lambda _raw: normalized)
assert specialize._validate_composition(b"normalized candidate")
def test_fixed_legacy_dind_runtime_preserves_rootless_storage_and_anonymous_volume(specialize):
base = _base_v1_compose_model(specialize)
item = _legacy_dind_fixture(specialize, base, image_id=specialize._DIND_IMAGE_ID,
running=True, restart_policy="unless-stopped")
assert specialize._legacy_dind_observation(item, base)["running"] is True
for key, value in (("User", "0"), ("Cmd", ["--data-root=/var/lib/docker"]),
("Entrypoint", ["sh"])):
changed = copy.deepcopy(item)
changed["Config"][key] = value
with pytest.raises(specialize.RunnerFleetSpecializeError, match="legacy_dind_identity_invalid"):
specialize._legacy_dind_observation(changed, base)
changed = copy.deepcopy(item)
changed["Mounts"][-1]["Source"] = "/untrusted/docker"
with pytest.raises(specialize.RunnerFleetSpecializeError, match="legacy_dind_identity_invalid"):
specialize._legacy_dind_observation(changed, base)
@pytest.mark.parametrize("support", ["2|none|false", "2|cgroupfs|false", "2|cgroupfs|true"])
def test_new_dind_requires_actual_resource_support_and_fixed_shared_workspaces(
specialize, monkeypatch, support
):
model = json.loads(specialize._render_specialized_compose(_base_v1_compose_model(specialize)))
service = model["services"]["runner-dind"]
item = _legacy_dind_fixture(specialize, _base_v1_compose_model(specialize),
image_id=specialize._DIND_IMAGE_ID, running=True, restart_policy="no")
item["Config"].update(User="0", Cmd=service["command"])
item["Config"]["Env"] = [f"DOCKER_HOST={specialize._DIND_DOCKER_HOST}"]
item["HostConfig"].update(
CgroupParent="itworx-ci", PidsLimit=2048, PortBindings={}
)
item["Mounts"] = [{"Type": "bind", "Source": m["source"],
"Destination": m["target"], "RW": not m["read_only"]}
for m in service["volumes"]]
assert specialize._validate_dind_outer(item, model) is True
monkeypatch.setattr(specialize, "_inspect", lambda _name: item)
commands = []
def run(args, **_kwargs):
commands.append(args)
return support
monkeypatch.setattr(specialize, "_run", run)
if support == "2|cgroupfs|true":
specialize._require_dind_resource_support(model)
assert f"DOCKER_HOST={specialize._DIND_DOCKER_HOST}" in commands[-1]
else:
with pytest.raises(specialize.RunnerFleetSpecializeError, match="dind_limits_unsupported"):
specialize._require_dind_resource_support(model)
missing_workspace = copy.deepcopy(item)
missing_workspace["Mounts"].pop()
assert specialize._validate_dind_outer(missing_workspace, model) is False
changed_cases = []
tcp_host = copy.deepcopy(item)
tcp_host["Config"]["Env"] = ["DOCKER_HOST=tcp://0.0.0.0:2375"]
changed_cases.append(tcp_host)
tcp_command = copy.deepcopy(item)
tcp_command["Config"]["Cmd"][1] = "--host=tcp://0.0.0.0:2375"
changed_cases.append(tcp_command)
published_port = copy.deepcopy(item)
published_port["HostConfig"]["PortBindings"] = {
"2375/tcp": [{"HostIp": "0.0.0.0", "HostPort": "2375"}]
}
changed_cases.append(published_port)
for changed in changed_cases:
assert specialize._validate_dind_outer(changed, model) is False
def test_outer_runner_must_match_fixed_compose_mount_and_image_identity(specialize):
profile_id = "frontend"
profile = specialize.RUNNER_PROFILES[profile_id]
name = profile.container_name
image = {
"Name": f"/{name}",
"Image": specialize.RUNNER_IMAGE_ID,
"Config": {
"Image": specialize.RUNNER_IMAGE,
"Labels": {
"com.docker.compose.project": specialize.PROJECT,
"com.docker.compose.service": f"runner-{profile_id}",
"tech.itworx.role": "gitea-actions-runner",
"tech.itworx.ci.profile": profile_id,
},
"Env": [
"CONFIG_FILE=/runner-config.yaml",
f"DOCKER_HOST={specialize._DIND_DOCKER_HOST}",
],
},
"State": {"Running": False},
"HostConfig": {
"CgroupParent": "itworx-ci",
"NanoCpus": 500_000_000,
"Memory": 512 * specialize._MIB,
"PidsLimit": 192,
"PidMode": "",
"NetworkMode": specialize._RUNNER_NETWORK_NAME,
"Privileged": False,
"ExtraHosts": ["gitea.example.com:198.19.104.1"],
"RestartPolicy": {"Name": "no"},
},
"Mounts": [
{"Source": str(specialize._host_volume_path(specialize._SPECIALIZED_DATA_ROOT / profile_id)), "Destination": "/data", "RW": True},
{"Source": str(specialize._host_volume_path(specialize.RUNNER_CONFIG_ROOT / f"{profile_id}.yaml")), "Destination": "/runner-config.yaml", "RW": False},
{"Source": str(specialize._host_volume_path(specialize._DIND_CONTROL_ROOT)), "Destination": specialize._DIND_CONTROL_CONTAINER_ROOT, "RW": False},
{
"Source": str(specialize._host_volume_path(specialize._SPECIALIZED_WORKSPACE_ROOT / name)),
"Destination": f"/{name}-workspace",
"RW": True,
},
],
}
assert specialize._validate_outer_item(name, image, profile_id)
unqualified_network = copy.deepcopy(image)
unqualified_network["HostConfig"]["NetworkMode"] = "runner-sandbox"
assert not specialize._validate_outer_item(name, unqualified_network, profile_id)
tcp_env = copy.deepcopy(image)
tcp_env["Config"]["Env"][-1] = "DOCKER_HOST=tcp://runner-dind:2375"
assert not specialize._validate_outer_item(name, tcp_env, profile_id)
writable_socket = copy.deepcopy(image)
next(
mount for mount in writable_socket["Mounts"]
if mount["Destination"] == specialize._DIND_CONTROL_CONTAINER_ROOT
)["RW"] = True
assert not specialize._validate_outer_item(name, writable_socket, profile_id)
host_gateway = copy.deepcopy(image)
host_gateway["HostConfig"]["ExtraHosts"].append("host.docker.internal:host-gateway")
assert not specialize._validate_outer_item(name, host_gateway, profile_id)
image["Image"] = "sha256:" + "0" * 64
assert not specialize._validate_outer_item(name, image, profile_id)
def test_atomic_guard_replace_uses_expected_hash_as_compare_and_swap(specialize, tmp_path):
path = tmp_path / "guard.sh"
path.write_bytes(b"original guard\n")
original_sha = hashlib.sha256(path.read_bytes()).hexdigest()
specialize._atomic_write_owned(
path,
b"patched guard\n",
mode=0o600,
expected_sha256=original_sha,
)
assert path.read_bytes() == b"patched guard\n"
with pytest.raises(specialize.RunnerFleetSpecializeError, match="compare_and_swap_failed"):
specialize._atomic_write_owned(
path,
b"stale write\n",
mode=0o600,
expected_sha256=original_sha,
)
assert path.read_bytes() == b"patched guard\n"
def test_runner_guard_patch_fences_legacy_and_preserves_exact_old_shape(specialize):
source = (
"stack_recovery=0\n"
"set_managed_runners_disabled false\n"
"unexpected=0\n"
' [ "$name" = "$DEPLOY_RUNNER" ] && continue\n'
)
patched = specialize._patch_runner_guard(
source,
"fixed-module --state",
'fixed-module --guard-allows-container "$name"',
)
assert "--fence-legacy" in patched
assert "--guard-allows-legacy-container" in patched
assert 'if [ "$specialization_state" = "legacy" ]; then' in patched
assert 'elif [ "$specialization_state" = "active" ]' in patched
assert '--guard-allows-container "$name"' in patched
assert " set_managed_runners_disabled false\n" in patched
with pytest.raises(specialize.RunnerFleetSpecializeError, match="runner_guard_shape_invalid"):
specialize._patch_runner_guard(source.replace("unexpected=0\n", ""), "state", "allow")
def test_execute_rejects_modified_script_before_starting_shell(specialize, monkeypatch):
script = (ROOT / "scripts" / "upgrade_ci_runner_fleet_specialized.sh").read_bytes()
monkeypatch.setattr(specialize, "_APPROVED_SCRIPT_SHA256", hashlib.sha256(script).hexdigest())
real_os = os
class RootPosixOS:
name = "posix"
@staticmethod
def geteuid():
return 0
def __getattr__(self, name):
return getattr(real_os, name)
monkeypatch.setattr(specialize, "os", RootPosixOS())
monkeypatch.setattr(specialize, "verified_topology", lambda: {"ready": False})
shell_calls = []
monkeypatch.setattr(specialize, "_run_approved_shell_script", shell_calls.append)
with pytest.raises(specialize.RunnerFleetSpecializeError, match="script_not_approved"):
specialize.execute(script + b"# altered\n")
assert shell_calls == []
def test_guard_scripts_patch_only_source_locked_files_and_reject_later_operator_edits(
specialize, tmp_path, monkeypatch
):
runner = tmp_path / "runner-guard"
host = tmp_path / "host-guard"
runner.write_bytes(
b"stack_recovery=0\n"
b" set_managed_runners_disabled false\n"
b"unexpected=0\n"
b' [ "$name" = "$DEPLOY_RUNNER" ] && continue\n'
)
host.write_bytes(
b"# ITWORX_RUNNER_CAPACITY_V2_BEGIN\n"
b"old capacity writer\n"
b"# ITWORX_RUNNER_CAPACITY_V2_END\n"
b"runner_healthy=1\n"
b' if [ "$sandbox_healthy" -eq 1 ] && \\\n'
b' timeout 20 docker start "$RUNNER" "$SECONDARY_RUNNER" >/dev/null 2>&1; then\n'
b" recover legacy stack\n"
b" fi\n"
)
runner_sha = hashlib.sha256(runner.read_bytes()).hexdigest()
host_sha = hashlib.sha256(host.read_bytes()).hexdigest()
monkeypatch.setattr(specialize, "RUNNER_GUARD_PATH", runner)
monkeypatch.setattr(specialize, "HOST_GUARD_PATH", host)
monkeypatch.setattr(specialize, "RUNNER_GUARD_SOURCE_SHA256", runner_sha)
monkeypatch.setattr(specialize, "HOST_GUARD_SOURCE_SHA256", host_sha)
monkeypatch.setattr(specialize, "_run", lambda *_args, **_kwargs: "")
monkeypatch.setattr(specialize, "_validate_guard_syntax", lambda _encoded: None)
monkeypatch.setattr(specialize, "_write_journal", lambda _journal: None)
backup = tmp_path / "guard-backup"
backup.mkdir(mode=0o700)
for path, name in ((runner, "gitea-runner-guard.sh"), (host, "host-resource-guard.sh")):
target = backup / name
target.write_bytes(path.read_bytes())
target.chmod(0o600)
journal = {
"backupDirectory": str(backup),
"guardFiles": [
{"path": str(runner), "originalSha256": runner_sha, "installedSha256": None},
{"path": str(host), "originalSha256": host_sha, "installedSha256": None},
]
}
specialize._service_guard_scripts(journal)
installed_runner_sha = hashlib.sha256(runner.read_bytes()).hexdigest()
assert journal["guardFiles"][0]["installedSha256"] == installed_runner_sha
assert "--fence-legacy" in runner.read_text(encoding="utf-8")
assert "# ITWORX_RUNNER_CAPACITY_V2_BEGIN" not in host.read_text(encoding="utf-8")
host.write_text("operator-owned guard revision\n", encoding="utf-8")
with pytest.raises(specialize.RunnerFleetSpecializeError, match="guard_source_changed"):
specialize._service_guard_scripts(journal)
assert host.read_text(encoding="utf-8") == "operator-owned guard revision\n"
@pytest.fixture
def guard_transaction(specialize, tmp_path, monkeypatch):
monkeypatch.setattr(specialize, "_rollback_maintenance", _fixture_rollback_maintenance)
runner = tmp_path / "runner-guard"
host = tmp_path / "host-guard"
originals = (
(b'#!/bin/bash\nstack_recovery=0\nset_managed_runners_disabled false\nunexpected=0\n'
b' [ "$name" = "$DEPLOY_RUNNER" ] && continue\n'),
(b'#!/bin/bash\n# ITWORX_RUNNER_CAPACITY_V2_BEGIN\nold capacity writer\n'
b'# ITWORX_RUNNER_CAPACITY_V2_END\nrunner_healthy=1\n'
b' if [ "$sandbox_healthy" -eq 1 ] && \\\n'
b' timeout 20 docker start "$RUNNER" "$SECONDARY_RUNNER" >/dev/null 2>&1; then\n'
b' recover legacy stack\n fi\n'),
)
backup = tmp_path / "backups"
backup.mkdir(mode=0o700)
for path, original, name in zip((runner, host), originals, ("gitea-runner-guard.sh", "host-resource-guard.sh"), strict=True):
path.write_bytes(original)
target = backup / name
target.write_bytes(original)
target.chmod(0o600)
monkeypatch.setattr(specialize, "RUNNER_GUARD_PATH", runner)
monkeypatch.setattr(specialize, "HOST_GUARD_PATH", host)
for name, original in zip(("RUNNER_GUARD_SOURCE_SHA256", "HOST_GUARD_SOURCE_SHA256"), originals, strict=True):
monkeypatch.setattr(specialize, name, hashlib.sha256(original).hexdigest())
journal = {"backupDirectory": str(backup), "guardFiles": [
{"path": str(path), "originalSha256": hashlib.sha256(original).hexdigest(), "installedSha256": None}
for path, original in zip((runner, host), originals, strict=True)
]}
monkeypatch.setattr(specialize, "_write_journal", lambda _journal: None)
monkeypatch.setattr(specialize, "_set_stage", lambda _journal, _stage: None)
monkeypatch.setattr(specialize, "_validate_guard_syntax", lambda _encoded: None)
return journal, runner, host, originals
@pytest.mark.parametrize("invalid", [0, 1])
def test_both_candidate_syntax_checks_precede_any_intent_or_guard_write(specialize, guard_transaction, monkeypatch, invalid):
journal, runner, host, originals = guard_transaction
checked = []
def syntax(encoded):
checked.append(encoded)
if len(checked) == invalid + 1:
raise specialize.RunnerFleetSpecializeError("runner_fleet_specialization_guard_syntax_invalid")
monkeypatch.setattr(specialize, "_validate_guard_syntax", syntax)
monkeypatch.setattr(specialize, "_write_journal", lambda _journal: pytest.fail("invalid candidate published intent"))
monkeypatch.setattr(specialize, "_atomic_write_owned", lambda *_args, **_kwargs: pytest.fail("invalid candidate installed"))
with pytest.raises(specialize.RunnerFleetSpecializeError, match="guard_syntax_invalid"):
specialize._service_guard_scripts(journal)
assert (runner.read_bytes(), host.read_bytes()) == originals
assert all(row["installedSha256"] is None for row in journal["guardFiles"])
@pytest.mark.parametrize("boundary", ["intent_failed", "after_intent", "after_runner", "after_host"])
def test_guard_installation_durable_crash_boundaries_restore_only_owned_bytes(specialize, guard_transaction, monkeypatch, boundary):
journal, runner, host, originals = guard_transaction
persisted = []
real_write = specialize._atomic_write_owned
writes = []
def journal_write(value):
assert all(row["installedSha256"] for row in value["guardFiles"])
assert (runner.read_bytes(), host.read_bytes()) == originals
if boundary == "intent_failed":
raise OSError("private canary write failure")
persisted.append(copy.deepcopy(value))
if boundary == "after_intent":
raise OSError("private canary crash after intent")
def install(path, encoded, **kwargs):
assert persisted and persisted[0]["guardFiles"] == journal["guardFiles"]
real_write(path, encoded, **kwargs)
writes.append(path)
if boundary == "after_runner" or boundary == "after_host" and len(writes) == 2:
raise OSError("private canary crash after replace")
monkeypatch.setattr(specialize, "_write_journal", journal_write)
monkeypatch.setattr(specialize, "_atomic_write_owned", install)
with pytest.raises(OSError):
specialize._service_guard_scripts(journal)
monkeypatch.setattr(specialize, "_atomic_write_owned", real_write)
specialize._restore_guard_scripts(persisted[0] if persisted else journal)
assert (runner.read_bytes(), host.read_bytes()) == originals
specialize._restore_guard_scripts(journal)
assert (runner.read_bytes(), host.read_bytes()) == originals
@pytest.mark.parametrize("mutation", ["duplicate", "foreign_path", "original_hash", "planned_hash", "backup", "foreign_edit"])
def test_guard_restore_rejects_foreign_journal_backup_or_live_bytes_before_any_write(specialize, guard_transaction, monkeypatch, mutation):
journal, runner, host, _originals = guard_transaction
specialize._service_guard_scripts(journal)
if mutation == "duplicate":
journal["guardFiles"][1] = copy.deepcopy(journal["guardFiles"][0])
elif mutation == "foreign_path":
journal["guardFiles"][1]["path"] = str(host.with_name("foreign"))
elif mutation == "original_hash":
journal["guardFiles"][0]["originalSha256"] = "f" * 64
elif mutation == "planned_hash":
journal["guardFiles"][0]["installedSha256"] = "f" * 64
elif mutation == "backup":
(Path(journal["backupDirectory"]) / "host-resource-guard.sh").write_bytes(b"foreign private canary")
else:
host.write_bytes(b"foreign private canary")
before = runner.read_bytes(), host.read_bytes()
monkeypatch.setattr(specialize, "_atomic_write_owned", lambda *_args, **_kwargs: pytest.fail("foreign state restored"))
with pytest.raises(specialize.RunnerFleetSpecializeError):
specialize._restore_guard_scripts(journal)
assert (runner.read_bytes(), host.read_bytes()) == before
def test_guard_syntax_checker_never_exposes_bash_output(specialize, monkeypatch):
def parse(args, **kwargs):
assert args == ["bash", "-n"] and kwargs["input"] == b"candidate"
assert kwargs["timeout"] == 15 and kwargs["capture_output"] is True
return SimpleNamespace(returncode=2, stdout=b"private canary", stderr=b"private canary")
monkeypatch.setattr(specialize.subprocess, "run", parse)
with pytest.raises(specialize.RunnerFleetSpecializeError, match="^runner_fleet_specialization_guard_syntax_invalid$"):
specialize._validate_guard_syntax(b"candidate")
@pytest.mark.parametrize("mutation", [None, "foreign", "begin_only", "end_only", "repeated"])
def test_host_guard_without_capacity_stanza_requires_exact_approved_bytes(specialize, guard_transaction, monkeypatch, mutation):
_journal, _runner, _host, originals = guard_transaction
text = originals[1].decode()
start = text.index("# ITWORX_RUNNER_CAPACITY_V2_BEGIN\n")
end = text.index("# ITWORX_RUNNER_CAPACITY_V2_END\n") + len("# ITWORX_RUNNER_CAPACITY_V2_END\n")
approved = text[:start] + text[end:]
monkeypatch.setattr(specialize, "HOST_GUARD_SOURCE_SHA256", hashlib.sha256(approved.encode()).hexdigest())
if mutation == "foreign":
approved += "# operator edit\n"
elif mutation == "begin_only":
approved += "# ITWORX_RUNNER_CAPACITY_V2_BEGIN\n"
elif mutation == "end_only":
approved += "# ITWORX_RUNNER_CAPACITY_V2_END\n"
elif mutation == "repeated":
approved += "# ITWORX_RUNNER_CAPACITY_V2_BEGIN\n# ITWORX_RUNNER_CAPACITY_V2_END\n" * 2
if mutation is None:
result = specialize._patch_host_guard(approved, "fixed --state")
assert 'specialization_state="$(fixed --state' in result
assert '"$specialization_state" != "legacy"' in result
else:
with pytest.raises(specialize.RunnerFleetSpecializeError, match="capacity_block_invalid"):
specialize._patch_host_guard(approved, "fixed --state")
@pytest.mark.parametrize("mutation", ["path", "backup", "foreign_edit"])
def test_guard_rollback_refusal_preserves_marker_journal_before_lifecycle_effects(specialize, guard_transaction, monkeypatch, mutation):
journal, runner, host, _originals = guard_transaction
specialize._service_guard_scripts(journal)
live_compose = runner.with_name("compose")
live_compose.write_bytes(b"fixed compose")
monkeypatch.setattr(specialize, "COMPOSE_PATH", live_compose)
journal["baseComposeSha256"] = hashlib.sha256(live_compose.read_bytes()).hexdigest()
monkeypatch.setattr(specialize, "_guard_state_unchanged", lambda _journal: True)
monkeypatch.setattr(specialize, "_cleanup_new_containers", lambda _journal: pytest.fail("refusal reached lifecycle effects"))
marker, durable = runner.with_name("marker"), runner.with_name("journal")
marker.write_bytes(b"owned marker")
durable.write_bytes(b"owned journal")
monkeypatch.setattr(specialize, "SPECIALIZED_STATE_PATH", marker)
monkeypatch.setattr(specialize, "MIGRATION_JOURNAL_PATH", durable)
if mutation == "path":
journal["guardFiles"][0]["path"] = str(runner.with_name("foreign"))
elif mutation == "backup":
(Path(journal["backupDirectory"]) / "host-resource-guard.sh").write_bytes(b"foreign canary")
elif mutation == "foreign_edit":
host.write_bytes(b"foreign canary")
assert specialize._rollback_before_activation(journal) is False
assert marker.read_bytes() == b"owned marker" and durable.read_bytes() == b"owned journal"
@pytest.mark.parametrize("failure", ["cleanup", "restore_cas", None])
def test_rollback_retains_fences_until_coherent_lifecycle_recovery(specialize, guard_transaction, monkeypatch, failure):
journal, runner, host, originals = guard_transaction
specialize._service_guard_scripts(journal)
fenced = runner.read_bytes(), host.read_bytes()
compose, marker, durable = runner.with_name("compose"), runner.with_name("marker"), runner.with_name("journal")
compose.write_bytes(b"fixed compose")
marker.write_bytes(b"owned marker")
durable.write_bytes(b"owned journal")
monkeypatch.setattr(specialize, "COMPOSE_PATH", compose)
monkeypatch.setattr(specialize, "SPECIALIZED_STATE_PATH", marker)
monkeypatch.setattr(specialize, "MIGRATION_JOURNAL_PATH", durable)
journal.update(baseComposeSha256=hashlib.sha256(compose.read_bytes()).hexdigest(),
originalDind={"present": False, "running": False, "restartPolicy": "no", "imageId": None},
originalContainers=[], originalRegistrations=[], cgroupCreated=True, cgroupRootSubtree=[])
monkeypatch.setattr(specialize, "_guard_state_unchanged", lambda _journal: True)
monkeypatch.setattr(specialize, "_captured_compose_model", lambda *_args, **_kwargs: b"fixed compose")
monkeypatch.setattr(specialize, "_captured_compose_mapping", lambda *_args, **_kwargs: {})
monkeypatch.setattr(specialize, "_legacy_services_for_restore", lambda _journal: [])
monkeypatch.setattr(specialize, "_inspect", lambda _name: None)
events = []
def cleanup(_journal):
assert (runner.read_bytes(), host.read_bytes()) == fenced
events.append("cleanup")
if failure == "cleanup":
raise specialize.RunnerFleetSpecializeError("runner_fleet_specialization_command_failed")
monkeypatch.setattr(specialize, "_cleanup_new_containers", cleanup)
def cgroup(**_kwargs):
assert (runner.read_bytes(), host.read_bytes()) == fenced
events.append("cgroup")
return {"familyExists": False}
monkeypatch.setattr(specialize, "rollback_host_cgroup_family", cgroup)
write = specialize._atomic_write_owned
def restore(*args, **kwargs):
assert events == ["cleanup", "cgroup"]
assert marker.exists() and durable.exists()
if failure == "restore_cas":
raise specialize.RunnerFleetSpecializeError("runner_fleet_specialization_compare_and_swap_failed")
write(*args, **kwargs)
monkeypatch.setattr(specialize, "_atomic_write_owned", restore)
assert specialize._rollback_before_activation(journal) is (failure is None)
if failure:
assert (runner.read_bytes(), host.read_bytes()) == fenced
assert marker.read_bytes() == b"owned marker" and durable.read_bytes() == b"owned journal"
else:
assert (runner.read_bytes(), host.read_bytes()) == originals
assert not marker.exists() and not durable.exists()
def test_rollback_maintenance_refusal_keeps_all_state_without_any_restore(specialize, tmp_path, monkeypatch):
marker, journal = tmp_path / "marker", tmp_path / "journal"
marker.write_bytes(b"owned marker")
journal.write_bytes(b"owned journal")
@contextmanager
def busy():
raise RuntimeError("private canary lock refusal")
yield
monkeypatch.setattr(specialize, "_rollback_maintenance", busy)
monkeypatch.setattr(specialize, "_rollback_before_activation_locked", lambda *_args: pytest.fail("busy barrier reached effects"))
assert specialize._rollback_before_activation({}) is False
assert marker.read_bytes() == b"owned marker" and journal.read_bytes() == b"owned journal"
def test_rollback_barrier_drift_refuses_before_guards_or_cleanup(specialize, guard_transaction, monkeypatch):
journal, _runner, _host, _originals = guard_transaction
@contextmanager
def changed():
def refuse():
raise RuntimeError("private canary changed inode")
yield refuse
monkeypatch.setattr(specialize, "_rollback_maintenance", changed)
monkeypatch.setattr(specialize, "_restore_guard_scripts", lambda *_args, **_kwargs: pytest.fail("drift restored guard"))
monkeypatch.setattr(specialize, "_cleanup_new_containers", lambda *_args: pytest.fail("drift changed fleet"))
assert specialize._rollback_before_activation(journal) is False
@pytest.mark.skipif(os.name != "posix", reason="real common flock contract requires POSIX")
def test_real_common_maintenance_lock_excludes_process_until_whole_rollback_finishes(specialize, tmp_path, monkeypatch):
if os.geteuid() != 0:
pytest.skip("protected maintenance lock requires root")
import runner_acceptance_quiescence as quiescence
lock = tmp_path / "maintenance.lock"
lock.touch(mode=0o600)
monkeypatch.setattr(quiescence, "MAINTENANCE_LOCK", lock)
monkeypatch.setattr(quiescence, "_LOCK_PROBE", quiescence._LOCK_PROBE.replace(
"/srv/example/appdata/gitea-deploy-runners/state/runner-maintenance.lock", str(lock)))
monkeypatch.setitem(sys.modules, "deploy", SimpleNamespace(load_projects=dict))
probe = (
"import fcntl,os,sys\nfd=os.open(sys.argv[1],os.O_RDWR|os.O_NOFOLLOW)\n"
"try: fcntl.flock(fd,fcntl.LOCK_EX|fcntl.LOCK_NB)\n"
"except BlockingIOError: sys.exit(0)\n"
"sys.exit(1)\n"
)
events = []
def held(phase):
result = subprocess.run([sys.executable, "-c", probe, str(lock)], capture_output=True, timeout=5, check=False)
assert result.returncode == 0 and not result.stdout and not result.stderr
events.append(phase)
compose, marker, durable = tmp_path / "compose", tmp_path / "marker", tmp_path / "journal"
compose.write_bytes(b"captured compose")
marker.write_bytes(b"owned marker")
durable.write_bytes(b"owned journal")
monkeypatch.setattr(specialize, "COMPOSE_PATH", compose)
monkeypatch.setattr(specialize, "SPECIALIZED_STATE_PATH", marker)
monkeypatch.setattr(specialize, "MIGRATION_JOURNAL_PATH", durable)
monkeypatch.setattr(specialize, "_guard_state_unchanged", lambda _journal: True)
monkeypatch.setattr(specialize, "_guard_script_plan", lambda _journal: [])
monkeypatch.setattr(specialize, "_restore_guard_scripts", lambda *_args, **_kwargs: held("guards-restored"))
monkeypatch.setattr(specialize, "_cleanup_new_containers", lambda _journal: held("cleanup"))
monkeypatch.setattr(specialize, "_captured_compose_model", lambda *_args, **_kwargs: b"captured compose")
monkeypatch.setattr(specialize, "_captured_compose_mapping", lambda *_args, **_kwargs: {})
monkeypatch.setattr(specialize, "_legacy_services_for_restore", lambda _journal: [])
monkeypatch.setattr(specialize, "_inspect", lambda _name: None)
def cgroup(**_kwargs):
held("cgroup")
assert marker.exists() and durable.exists()
return {"familyExists": False}
monkeypatch.setattr(specialize, "rollback_host_cgroup_family", cgroup)
journal = {"baseComposeSha256": hashlib.sha256(compose.read_bytes()).hexdigest(), "guardFiles": [{"owned": True}],
"originalDind": {"present": False, "running": False, "restartPolicy": "no", "imageId": None},
"originalContainers": [], "originalRegistrations": [], "cgroupCreated": True, "cgroupRootSubtree": []}
assert specialize._rollback_before_activation(journal) is True
assert events == ["cleanup", "cgroup", "guards-restored"]
assert not marker.exists() and not durable.exists()
# The independent participant can acquire only after the wrapper returns.
result = subprocess.run([sys.executable, "-c", probe, str(lock)], capture_output=True, timeout=5, check=False)
assert result.returncode == 1 and not result.stdout and not result.stderr
@pytest.fixture
def owned_orphan(specialize, guard_transaction, tmp_path, monkeypatch):
import runner_acceptance_quiescence as quiescence
journal, runner, host, originals = guard_transaction
corrected = specialize._guard_script_plan(journal)[0][3]
closing = f'\nfi\n{specialize.SPECIALIZATION_GUARD_END}\n'.encode()
orphan = corrected.replace(closing, f'\n{specialize.SPECIALIZATION_GUARD_END}\n'.encode(), 1)
runner.write_bytes(orphan)
monkeypatch.setattr(specialize, "_OWNED_MALFORMED_RUNNER_GUARD_SHA256", hashlib.sha256(orphan).hexdigest())
receipt = {
"schema": quiescence.SCHEMA, "revision": "0ea6f267c28d51ac7113e6d943c89857d3a189ab",
"manifestDigest": quiescence._digest(quiescence.OBSERVATION), "phase": "complete",
"progress": {row["id"]: "quiesced" for row in quiescence._fixed_rows()},
"registrationProgress": {"9": "restored", "22": "restored"},
"guardProgress": {"runner": "restored", "host": "restored"},
"guards": {name: {
"original": base64.b64encode(original).decode(), "originalSha256": hashlib.sha256(original).hexdigest(),
"wrappedSha256": hashlib.sha256(quiescence.ControllerDriver._wrapped_guard(original)).hexdigest(),
"mode": path.stat().st_mode & 0o777,
} for name, path, original in zip(("runner", "host"), (runner, host), originals, strict=True)},
"baseline": {"rows": [{"state": {"Running": False}} for _row in quiescence._fixed_rows()]},
}
events = []
for name in ("SPECIALIZED_STATE_PATH", "MIGRATION_JOURNAL_PATH"):
monkeypatch.setattr(specialize, name, tmp_path / name)
monkeypatch.setattr(specialize, "_specialized_storage_present", lambda: False)
monkeypatch.setattr(specialize, "_cgroup_family_snapshot", lambda: {"exists": False})
monkeypatch.setattr(specialize, "_read_fleet_base_compose", lambda: (b"source", None))
monkeypatch.setattr(specialize, "_validate_base_composition", lambda _source: "v1")
monkeypatch.setattr(specialize, "_runner_rows", lambda: [{"id": 9}])
monkeypatch.setattr(specialize, "_validate_legacy_registrations", lambda *_args: [{"id": 9}])
monkeypatch.setattr(specialize, "_validate_legacy_containers", lambda *_args: ("v1", [{"id": "fixed"}]))
monkeypatch.setitem(sys.modules, "deploy", SimpleNamespace(load_projects=lambda: {"fixed": "registry"}))
monkeypatch.setattr(quiescence, "_read_receipt", lambda: copy.deepcopy(receipt))
monkeypatch.setattr(quiescence, "_idle", lambda *_args: events.append("idle"))
def attest(snapshot, *, baseline, progress):
assert baseline == receipt["baseline"] and progress == receipt["progress"]
assert snapshot["rows"] == baseline["rows"]
events.append("attested")
monkeypatch.setattr(quiescence, "attest_snapshot", attest)
class Driver:
_wrapped_guard = staticmethod(quiescence.ControllerDriver._wrapped_guard)
def __init__(self, projects):
assert projects == {"fixed": "registry"}
@contextmanager
def maintenance(self):
with (tmp_path / "fixture-lock").open("wb") as lock:
self._maintenance_fd = lock.fileno()
events.append("locked")
yield
events.append("unlocked")
def assert_barrier(self):
events.append("barrier")
def snapshot(self):
self._guard_binding()
events.append("snapshot")
return copy.deepcopy(receipt["baseline"])
monkeypatch.setattr(quiescence, "ControllerDriver", Driver)
return quiescence, receipt, runner, host, originals, orphan, events, Driver
def test_owned_orphan_recovery_uses_full_repeated_admission_and_cas_then_is_idempotent(specialize, owned_orphan, monkeypatch):
_q, receipt, runner, host, originals, orphan, events, _driver = owned_orphan
before = copy.deepcopy(receipt)
write = specialize._atomic_write_owned
def restore(path, data, **kwargs):
assert events.count("attested") == 2 and events.count("idle") == 3
assert path == runner and data == originals[0]
assert kwargs["expected_sha256"] == hashlib.sha256(orphan).hexdigest()
events.append("restore")
write(path, data, **kwargs)
monkeypatch.setattr(specialize, "_atomic_write_owned", restore)
specialize._recover_owned_orphan_guard()
assert (runner.read_bytes(), host.read_bytes()) == originals and receipt == before
assert events.index("locked") < events.index("restore") < events.index("unlocked")
specialize._recover_owned_orphan_guard()
assert events.count("restore") == 1
@pytest.mark.parametrize("mutation", ["revision", "manifest", "phase", "progress", "registration", "guard_progress",
"original", "wrapped", "foreign_runner", "foreign_host", "marker", "journal",
"storage", "family", "idle", "lock", "origin", "changed_receipt", "container", "cas"])
def test_orphan_recovery_refuses_unowned_or_changed_state_without_write(specialize, owned_orphan, monkeypatch, mutation):
q, receipt, runner, host, _originals, _orphan, _events, driver = owned_orphan
if mutation in {"revision", "manifest", "phase"}:
receipt[{"revision": "revision", "manifest": "manifestDigest", "phase": "phase"}[mutation]] = "foreign"
elif mutation == "progress":
receipt["progress"].pop(next(iter(receipt["progress"])))
elif mutation == "registration":
receipt["registrationProgress"]["9"] = "disabled"
elif mutation == "guard_progress":
receipt["guardProgress"]["host"] = "installed"
elif mutation in {"original", "wrapped"}:
receipt["guards"]["runner"]["originalSha256" if mutation == "original" else "wrappedSha256"] = "f" * 64
elif mutation in {"foreign_runner", "foreign_host"}:
(runner if mutation == "foreign_runner" else host).write_bytes(b"private foreign canary")
elif mutation in {"marker", "journal"}:
(specialize.SPECIALIZED_STATE_PATH if mutation == "marker" else specialize.MIGRATION_JOURNAL_PATH).touch()
elif mutation == "storage":
monkeypatch.setattr(specialize, "_specialized_storage_present", lambda: True)
elif mutation == "family":
monkeypatch.setattr(specialize, "_cgroup_family_snapshot", lambda: {"exists": True})
elif mutation == "idle":
monkeypatch.setattr(q, "_idle", lambda *_args: (_ for _ in ()).throw(q.QuiescenceError("global_ci_work_present")))
elif mutation == "lock":
monkeypatch.setattr(driver, "assert_barrier", lambda _self: (_ for _ in ()).throw(q.QuiescenceError("dispatch_maintenance_busy")))
elif mutation == "origin":
monkeypatch.setattr(q, "attest_snapshot", lambda *_args, **_kwargs: (_ for _ in ()).throw(q.QuiescenceError("origin_changed")))
elif mutation == "changed_receipt":
reads = []
def changed():
reads.append(True)
result = copy.deepcopy(receipt)
if len(reads) > 1:
result["revision"] = "foreign"
return result
monkeypatch.setattr(q, "_read_receipt", changed)
elif mutation == "container":
reads = []
monkeypatch.setattr(specialize, "_validate_legacy_containers", lambda *_args: reads.append(True) or ("v1", [{"id": str(len(reads))}]))
else:
monkeypatch.setattr(specialize, "_atomic_write_owned", lambda *_args, **_kwargs: (_ for _ in ()).throw(
specialize.RunnerFleetSpecializeError("runner_fleet_specialization_compare_and_swap_failed")))
before = runner.read_bytes(), host.read_bytes()
if mutation != "cas":
monkeypatch.setattr(specialize, "_atomic_write_owned", lambda *_args, **_kwargs: pytest.fail("unowned orphan restored"))
with pytest.raises(specialize.RunnerFleetSpecializeError):
specialize._recover_owned_orphan_guard()
assert (runner.read_bytes(), host.read_bytes()) == before
def test_execute_runs_only_approved_shell_then_requires_verified_postcondition(specialize, monkeypatch):
script = (ROOT / "scripts" / "upgrade_ci_runner_fleet_specialized.sh").read_bytes()
digest = hashlib.sha256(script).hexdigest()
monkeypatch.setattr(specialize, "_APPROVED_SCRIPT_SHA256", digest)
real_os = os
class RootPosixOS:
name = "posix"
@staticmethod
def geteuid():
return 0
def __getattr__(self, name):
return getattr(real_os, name)
monkeypatch.setattr(specialize, "os", RootPosixOS())
monkeypatch.setattr(specialize, "_migration_lock", nullcontext)
monkeypatch.setattr(specialize, "specialization_state", lambda: "legacy")
monkeypatch.setitem(
sys.modules,
"status_server",
type("StatusServerStub", (), {"_require_controller_release": staticmethod(lambda: {
"sourceRepository": specialize.SOURCE_REPOSITORY,
"sourceRevision": "a" * 40,
"sourceTree": "b" * 40,
})}),
)
shell_calls: list[bytes] = []
state = {"active": False}
def run_approved(received):
shell_calls.append(received)
state["active"] = True
monkeypatch.setattr(specialize, "_run_approved_shell_script", run_approved)
monkeypatch.setattr(
specialize,
"verified_topology",
lambda: {
"ready": state["active"],
"verifiedRunners": list(specialize._SPECIALIZED_NAMES) if state["active"] else [],
"resourceEnvelope": {"ready": state["active"], "digest": "c" * 64},
},
)
result = specialize.execute(script)
assert result["status"] == "specialized"
assert result["verifiedRunners"] == list(specialize._SPECIALIZED_NAMES)
assert shell_calls == [script]
def test_run_with_input_keeps_registration_secret_out_of_command_arguments(specialize, monkeypatch):
observed = {}
def fake_run(args, **kwargs):
observed["args"] = args
observed["input"] = kwargs["input"]
observed["capture_output"] = kwargs["capture_output"]
return type("Result", (), {"returncode": 0, "stdout": b"registered"})()
monkeypatch.setattr(specialize.subprocess, "run", fake_run)
result = specialize._run_with_input(["docker", "run", "fixed"], b"sensitive-runner-token\n")
assert result == "registered"
assert observed["args"] == ["docker", "run", "fixed"]
assert observed["input"] == b"sensitive-runner-token\n"
assert "sensitive-runner-token" not in " ".join(observed["args"])
def test_build_probe_create_command_is_fixed_and_has_no_mount_or_shared_socket(specialize):
record = _build_probe_record(specialize)
args = specialize._build_probe_create_args(record)
assert args[:8] == [
"create",
"--name",
record["name"],
"--restart=no",
"--cgroup-parent=docker",
"--network=bridge",
"--cgroupns=private",
"--label",
]
assert record["taskImageId"] in args
assert "--cpus=4" in args
assert "--memory=6144m" in args
assert "--pids-limit=1024" in args
assert "--blkio-weight=100" in args
assert "--security-opt=seccomp=unconfined" in args
assert "--security-opt=systempaths=unconfined" in args
assert "--device=/dev/net/tun:/dev/net/tun:rwm" in args
assert "--entrypoint" in args
assert specialize._BUILD_PROBE_EXECUTABLE in args
assert record["sourceRevision"] in args
assert record["sourceTree"] in args
assert not any(
value == "--privileged"
or value == "-e"
or value.startswith(("--cap-add", "--volume", "--mount", "--env"))
for value in args
)
assert not any("docker.sock" in value or "host.docker.internal" in value for value in args)
def test_build_probe_creation_inspect_requires_stopped_zero_pid_and_exact_inherited_labels(
specialize,
):
record = _build_probe_record(specialize, container_id="e" * 64, status="created")
item, environment = _build_probe_inspect(specialize, record)
item["State"] = {"Running": False, "Pid": 0}
assert specialize._validate_build_probe_container(
item,
record,
image_environment=environment,
require_running=False,
require_created_stopped=True,
)
labels = specialize._build_probe_container_labels(record)
assert set(specialize.PINNED_TASK_BASE_INHERITED_LABELS).issubset(labels)
assert record["imageLabels"].items() <= labels.items()
assert item["Config"]["Labels"] == labels
running, _ = _build_probe_inspect(specialize, record)
assert not specialize._validate_build_probe_container(
running,
record,
image_environment=environment,
require_running=False,
require_created_stopped=True,
)
bad_pid = copy.deepcopy(item)
bad_pid["State"]["Pid"] = 9
assert not specialize._validate_build_probe_container(
bad_pid,
record,
image_environment=environment,
require_running=False,
require_created_stopped=True,
)
foreign_label = copy.deepcopy(item)
foreign_label["Config"]["Labels"]["untrusted.extra"] = "true"
assert not specialize._validate_build_probe_container(
foreign_label,
record,
image_environment=environment,
require_running=False,
require_created_stopped=True,
)
def test_build_probe_container_inspect_accepts_only_fixed_private_build_job(specialize):
record = _build_probe_record(specialize, container_id="e" * 64, status="created")
item, environment = _build_probe_inspect(specialize, record)
assert specialize._validate_build_probe_container(item, record, image_environment=environment)
for section, changes in (
("HostConfig", {"Privileged": True}),
("HostConfig", {"PidMode": "host"}),
("HostConfig", {"NetworkMode": "host"}),
("HostConfig", {"CapAdd": ["SYS_ADMIN"]}),
("HostConfig", {"Devices": [
*item["HostConfig"]["Devices"],
{"PathOnHost": "/dev/sda", "PathInContainer": "/dev/sda", "CgroupPermissions": "rwm"},
]}),
("HostConfig", {"Mounts": [{"Type": "bind", "Source": "/host", "Target": "/host"}]}),
("HostConfig", {"SecurityOpt": ["seccomp=unconfined"]}),
("Config", {"Env": [*environment, "DOCKER_HOST=tcp://shared-dind:2375"]}),
("Config", {"Image": "localhost/itworx-ci-build:latest"}),
):
changed, expected_environment = _build_probe_inspect(specialize, record)
changed[section].update(changes)
assert not specialize._validate_build_probe_container(
changed,
record,
image_environment=expected_environment,
)
@pytest.mark.parametrize("field,value", [
("Type", "bind"), ("Name", "persistent-builder-cache"),
("Source", "/var/run/docker.sock"), ("Destination", "/host"),
("Driver", "nfs"), ("RW", False), ("Propagation", "rshared"),
])
def test_build_probe_rejects_foreign_or_shared_data_mount(specialize, field, value):
record = _build_probe_record(specialize, container_id="e" * 64, status="created")
item, environment = _build_probe_inspect(specialize, record)
item["Mounts"][0][field] = value
assert not specialize._validate_build_probe_container(item, record, image_environment=environment)
def test_build_probe_volume_identity_detects_same_name_recreation(specialize, monkeypatch):
name = "9" * 64
volume = {
"Name": name, "Driver": "local", "Scope": "local",
"Mountpoint": f"/var/lib/docker/volumes/{name}/_data",
"Labels": {"com.docker.volume.anonymous": ""}, "Options": None, "CreatedAt": "2026-10-03T00:00:00Z",
}
monkeypatch.setattr(specialize, "_run_dind_docker", lambda *_args, **_kwargs: json.dumps([volume]))
identity = specialize._attest_build_probe_data_volume(name)
volume["Labels"]["foreign.owner"] = "another-job"
with pytest.raises(specialize.RunnerFleetSpecializeError, match="volume_untrusted"):
specialize._attest_build_probe_data_volume(name, identity)
volume["Labels"] = {"com.docker.volume.anonymous": ""}
volume["CreatedAt"] = "2026-10-03T00:00:01Z"
with pytest.raises(specialize.RunnerFleetSpecializeError, match="volume_untrusted"):
specialize._attest_build_probe_data_volume(name, identity)
def test_build_probe_volume_recovery_never_removes_volume_used_by_another_container(specialize, monkeypatch):
identity = {"name": "9" * 64, "createdAt": "2026-10-03T00:00:00Z"}
mutations = []
def run(*args, **kwargs):
if args[:2] == ("volume", "ls"):
return identity["name"]
if args[:2] == ("ps", "-a"):
return "f" * 64
mutations.append(args)
raise AssertionError("unexpected volume mutation")
monkeypatch.setattr(specialize, "_run_dind_docker", run)
monkeypatch.setattr(specialize, "_attest_build_probe_data_volume", lambda *_args: identity)
with pytest.raises(specialize.RunnerFleetSpecializeError, match="volume_in_use"):
specialize._remove_recorded_build_probe_volume({"dataVolume": identity})
assert mutations == []
def test_build_probe_volume_recovery_removes_only_exact_journal_owned_volume(specialize, monkeypatch):
identity = {"name": "9" * 64, "createdAt": "2026-10-03T00:00:00Z"}
mutations = []
present = True
def run(*args, **kwargs):
nonlocal present
if args[:2] == ("volume", "ls"):
assert args == ("volume", "ls", "--quiet", "--filter", f"name={identity['name']}")
return identity["name"] if present else ""
if args[:2] == ("ps", "-a"):
return ""
assert args == ("volume", "rm", identity["name"])
mutations.append(args)
present = False
return identity["name"]
monkeypatch.setattr(specialize, "_run_dind_docker", run)
monkeypatch.setattr(specialize, "_attest_build_probe_data_volume", lambda *_args: identity)
specialize._remove_recorded_build_probe_volume({"dataVolume": identity})
assert mutations == [("volume", "rm", identity["name"])]
def test_build_probe_result_requires_exact_marker_and_every_rootless_tool_check(specialize):
record = _build_probe_record(specialize)
payload = {
"schema": specialize._BUILD_PROBE_SCHEMA,
"status": "passed",
"sourceRevision": record["sourceRevision"],
"sourceTree": record["sourceTree"],
"engineVersion": "29.8.2",
"engineRootless": True,
"engineStorageDriver": "overlay2",
"engineCgroupDriver": "none",
"engineSocket": "unix:///run/itworx-job-docker/docker.sock",
"engineSocketUid": 1000,
"composeVersion": "5.5.1",
"buildRun": True,
"composeHealthy": True,
"composeDnsResolved": True,
"archiveNormalized": True,
"archiveLoaded": True,
"imageId": "sha256:" + "2" * 64,
"imageLabels": record["imageLabels"],
"archiveSha256": "sha256:" + "3" * 64,
"holdSeconds": 30,
}
line = specialize._BUILD_PROBE_RESULT_PREFIX + json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
)
assert specialize._validate_build_probe_result(line, record=record) == payload
with pytest.raises(specialize.RunnerFleetSpecializeError, match="build_probe_result_missing"):
specialize._validate_build_probe_result(line + "\n" + line, record=record)
for key, value in (
("engineRootless", False),
("engineSocket", "tcp://0.0.0.0:2375"),
("composeHealthy", False),
("holdSeconds", 0),
):
changed = {**payload, key: value}
changed_line = specialize._BUILD_PROBE_RESULT_PREFIX + json.dumps(
changed,
sort_keys=True,
separators=(",", ":"),
)
with pytest.raises(specialize.RunnerFleetSpecializeError, match="build_probe_result_untrusted"):
specialize._validate_build_probe_result(changed_line, record=record)
def test_build_probe_live_process_inventory_requires_daemon_and_compose_children(specialize):
output = "\n".join(
(
"PID PPID COMMAND",
"1 0 python3 /usr/local/bin/itworx-ci-job-docker-probe",
"10 1 rootlesskit --net=slirp4netns",
"11 10 dockerd --host=unix:///run/itworx-job-docker/docker.sock --storage-driver=overlay2",
"12 11 containerd --config /run/containerd/config.toml",
"21 20 sleep 300",
)
)
result = specialize._validate_build_probe_processes(output)
assert result == {
"processCount": 5,
"probe": True,
"rootlesskit": True,
"dockerd": True,
"containerd": True,
"composeService": True,
}
with pytest.raises(specialize.RunnerFleetSpecializeError, match="build_probe_processes_untrusted"):
specialize._validate_build_probe_processes(
output.replace("21 20 sleep 300", "21 20 cat /etc/os-release")
)
with pytest.raises(specialize.RunnerFleetSpecializeError, match="build_probe_processes_untrusted"):
specialize._validate_build_probe_processes(output.replace("containerd --config", "unknown --config"))
def test_build_probe_intent_binds_revision_tree_image_nonce_and_exact_container_id(specialize):
record = _build_probe_record(specialize)
assert specialize._validated_build_probe_record(
record,
source_revision="a" * 40,
source_tree="b" * 40,
) == record
invalid = dict(record, name=record["name"] + "-foreign")
with pytest.raises(specialize.RunnerFleetSpecializeError, match="journal_invalid"):
specialize._validated_build_probe_record(
invalid,
source_revision="a" * 40,
source_tree="b" * 40,
)
def _install_build_probe_cleanup_fakes(specialize, monkeypatch, record, item, ids):
image_environment = item["Config"]["Env"]
present = list(ids)
removed = []
monkeypatch.setattr(
specialize,
"_inspect",
lambda name: {"State": {"Running": True}} if name == specialize.DIND_NAME else None,
)
monkeypatch.setattr(specialize, "_captured_compose_mapping", lambda *_args, **_kwargs: {})
monkeypatch.setattr(specialize, "_validate_dind_outer", lambda *_args: True)
monkeypatch.setattr(
specialize,
"_build_probe_ids",
lambda *, label=True, name=None: list(present)
if label or name == record["name"] and present
else [],
)
monkeypatch.setattr(specialize, "_inspect_dind_container", lambda _cid: item)
monkeypatch.setattr(
specialize,
"_build_probe_image_environment",
lambda _record: image_environment,
)
def run_dind(*args, **_kwargs):
if args[:3] == ("rm", "--force", "--volumes"):
removed.append(args[3])
present.clear()
return args[3]
if args[:2] == ("volume", "ls"):
return ""
raise AssertionError(f"unexpected private-DinD command: {args}")
monkeypatch.setattr(specialize, "_run_dind_docker", run_dind)
monkeypatch.setattr(
specialize, "_attest_build_probe_data_volume",
lambda name, expected=None: {"name": name, "createdAt": "2026-10-03T00:00:00Z"},
)
monkeypatch.setattr(specialize, "_write_journal", lambda _journal: None)
return removed
def test_build_probe_recovery_cleans_only_intent_owned_container_after_create_crash(
specialize,
monkeypatch,
):
record = _build_probe_record(specialize, status="intent")
cid = "e" * 64
item, _environment = _build_probe_inspect(specialize, record, container_id=cid)
removed = _install_build_probe_cleanup_fakes(specialize, monkeypatch, record, item, [cid])
journal = {
"sourceRevision": record["sourceRevision"],
"sourceTree": record["sourceTree"],
"buildProbe": record,
}
specialize._cleanup_build_probe(journal)
assert removed == [cid]
assert journal["buildProbe"]["containerId"] == cid
assert journal["buildProbe"]["status"] == "cleaned"
def test_build_probe_recovery_refuses_foreign_cid_even_with_matching_name_and_nonce(
specialize,
monkeypatch,
):
expected_cid = "d" * 64
foreign_cid = "e" * 64
record = _build_probe_record(
specialize,
container_id=expected_cid,
status="started",
)
item, _environment = _build_probe_inspect(specialize, record, container_id=foreign_cid)
removed = _install_build_probe_cleanup_fakes(
specialize,
monkeypatch,
record,
item,
[foreign_cid],
)
journal = {
"sourceRevision": record["sourceRevision"],
"sourceTree": record["sourceTree"],
"buildProbe": record,
}
with pytest.raises(specialize.RunnerFleetSpecializeError, match="build_probe_container_untrusted"):
specialize._cleanup_build_probe(journal)
assert removed == []
def test_build_probe_recovery_refuses_ambiguous_owned_label_inventory(
specialize,
monkeypatch,
):
record = _build_probe_record(specialize, status="intent")
cid = "e" * 64
item, _environment = _build_probe_inspect(specialize, record, container_id=cid)
removed = _install_build_probe_cleanup_fakes(
specialize,
monkeypatch,
record,
item,
[cid, "f" * 64],
)
journal = {
"sourceRevision": record["sourceRevision"],
"sourceTree": record["sourceTree"],
"buildProbe": record,
}
with pytest.raises(specialize.RunnerFleetSpecializeError, match="build_probe_ambiguous_containers"):
specialize._cleanup_build_probe(journal)
assert removed == []
invalid = dict(record, status="started")
with pytest.raises(specialize.RunnerFleetSpecializeError, match="journal_invalid"):
specialize._validated_build_probe_record(
invalid,
source_revision="a" * 40,
source_tree="b" * 40,
)
@pytest.mark.parametrize("recovering_cgroup_intent", [False, True])
@pytest.mark.parametrize("probe_fails", [False, True])
def test_specialization_stage_fixture_quiesces_old_before_registering_and_never_runs_over_two(
specialize, tmp_path, monkeypatch, recovering_cgroup_intent, probe_fails
):
revision = "a" * 40
tree = "b" * 40
manifest_sha = "d" * 64
manifest = {
"images": [
{"profile": profile_id, "reference": specialize._task_image_reference(profile_id, revision)}
for profile_id in specialize.PROFILE_IDS
]
}
archive = tmp_path / "task-images.tar"
archive.write_bytes(b"fixture")
journal = {
"status": "in_progress",
"stage": "prepared",
"legacyTopology": "v1",
"sourceRevision": revision,
"sourceTree": tree,
"cgroupCreated": False,
"cgroupEnsurePending": recovering_cgroup_intent,
"baseComposeSha256": hashlib.sha256(b"captured-base").hexdigest(),
"baseComposeModelSha256": hashlib.sha256(b"base-model").hexdigest(),
"specializedComposeSha256": hashlib.sha256(b"specialized-model").hexdigest(),
"backupDirectory": str(tmp_path / "specialized-test"),
"newRegistrationIds": {},
}
events: list[str] = []
running: set[str] = set()
maximum_running = 0
marker_active = {"value": False}
def rows():
return [
{
"name": specialize.RUNNER_PROFILES[profile_id].container_name,
"id": index + 1,
"labels": sorted(
{
specialize.RUNNER_PROFILES[profile_id].runner_label,
*specialize.RUNNER_PROFILES[profile_id].legacy_labels,
}
),
"disabled": True,
"busy": False,
}
for index, profile_id in enumerate(specialize.PROFILE_IDS)
]
def inventory():
return (
{
name: {"name": name, "State": {"Running": name in running}}
for name in (*specialize._SPECIALIZED_NAMES, specialize.DIND_NAME)
},
[],
)
def resource_proof(_claims, _containers, _legacy):
return {
"ready": True,
"digest": "e" * 64,
"runningRunnerContainers": len(running),
}
monkeypatch.setattr(specialize, "_captured_base_source", lambda _journal: b"captured-base")
monkeypatch.setattr(
specialize,
"_captured_compose_model",
lambda _journal, *, specialized: b"specialized-model" if specialized else b"base-model",
)
monkeypatch.setattr(specialize, "_read_source_file", lambda *_args, **_kwargs: b"captured-base")
monkeypatch.setattr(specialize, "_validate_base_composition", lambda _bytes=None: "v1")
monkeypatch.setattr(specialize, "_validate_composition", lambda raw: raw)
monkeypatch.setattr(specialize, "_run_specialized_compose", lambda _journal, *args, **_kwargs: events.append(f"compose:{args[-1]}"))
monkeypatch.setattr(specialize, "_wait_guards_quiet", lambda: events.append("guards-quiet"))
monkeypatch.setattr(specialize, "_service_guard_scripts", lambda _journal: events.append("guards-fenced"))
monkeypatch.setattr(specialize, "_stop_legacy_capacity_daemon", lambda: events.append("capacity-stopped"))
monkeypatch.setattr(specialize, "_set_legacy_disabled", lambda _journal: events.append("legacy-disabled"))
monkeypatch.setattr(specialize, "_check_legacy_queued_jobs", lambda: None)
monkeypatch.setattr(specialize, "_stop_remove_legacy_containers", lambda _journal: events.append("legacy-removed"))
monkeypatch.setattr(
specialize,
"_cgroup_family_snapshot",
lambda: {
"exists": recovering_cgroup_intent,
"rootSubtreeControllers": ["cpu", "memory", "pids"],
},
)
def ensure_cgroup_family(**kwargs):
assert kwargs["allow_existing"] is recovering_cgroup_intent
events.append("cgroup-ensured")
return not recovering_cgroup_intent
monkeypatch.setattr(
specialize,
"_ensure_cgroup_family",
ensure_cgroup_family,
)
monkeypatch.setattr(specialize, "_prepare_profile_state", lambda _revision: None)
monkeypatch.setattr(specialize, "_write_private_atomic", lambda *_args, **_kwargs: None)
def register(current_journal):
assert "legacy-disabled" in events
events.append("registrations-created")
current_journal["newRegistrationIds"] = {
profile_id: index + 1 for index, profile_id in enumerate(specialize.PROFILE_IDS)
}
specialize._set_stage(current_journal, "registrations_created")
monkeypatch.setattr(specialize, "_register_profiles", register)
monkeypatch.setattr(specialize, "_run", lambda *_args, **_kwargs: "")
monkeypatch.setattr(specialize, "_wait_healthy", lambda *_args, **_kwargs: None)
monkeypatch.setattr(specialize, "_inspect", lambda _name: {"State": {"Running": True}})
monkeypatch.setattr(specialize, "_require_dind_resource_support", lambda _model: None)
monkeypatch.setattr(specialize, "_captured_compose_mapping", lambda *_args, **_kwargs: {"services": {}})
monkeypatch.setattr(specialize, "_load_images_into_dind", lambda *_args: None)
monkeypatch.setattr(specialize, "_verify_manifest_images_in_dind", lambda _items: True)
monkeypatch.setattr(specialize, "_docker_outer_inventory", inventory)
monkeypatch.setattr(specialize, "_resource_proof", resource_proof)
monkeypatch.setattr(specialize, "_validate_outer_item", lambda *_args: True)
monkeypatch.setattr(
specialize,
"_trusted_private_file",
lambda path, **_kwargs: specialize._render_runner_config("build", revision)
if str(path).endswith("build.yaml")
else b"fixture",
)
monkeypatch.setattr(specialize, "_cleanup_build_probe", lambda _journal: None)
def run_build_probe(_journal, _manifest):
assert not running
events.append("build-probed")
if probe_fails:
raise specialize.RunnerFleetSpecializeError(
"runner_fleet_specialization_build_probe_failed"
)
monkeypatch.setattr(specialize, "_run_build_probe", run_build_probe)
monkeypatch.setattr(specialize, "_runner_rows", rows)
def write_journal(current_journal):
if current_journal.get("cgroupEnsurePending") is True:
events.append("cgroup-intent-durable")
if current_journal.get("cgroupCreated") is True:
events.append("cgroup-result-durable")
if current_journal.get("status") == "active":
events.append("journal-active-durable")
monkeypatch.setattr(specialize, "_write_journal", write_journal)
batches: list[tuple[str, ...]] = []
def set_containers(profiles, start):
nonlocal maximum_running
names = {specialize.RUNNER_PROFILES[item].container_name for item in profiles}
if start:
assert len(running | names) <= 2
running.update(names)
maximum_running = max(maximum_running, len(running))
batches.append(tuple(profiles))
events.append("batch-started")
else:
running.difference_update(names)
events.append("batch-stopped")
monkeypatch.setattr(specialize, "_set_runner_containers_running", set_containers)
monkeypatch.setattr(specialize, "_bootstrap_registrations", lambda: [{"name": name} for name in specialize._SPECIALIZED_NAMES])
monkeypatch.setattr(specialize, "_read_installed_manifest_sha256", lambda: manifest_sha)
monkeypatch.setattr(specialize, "_write_pending_bootstrap_receipt", lambda *_args: events.append("receipt-written"))
crash = {"next_active_write": True}
class SimulatedPowerLoss(BaseException):
pass
def write_marker(**kwargs):
assert kwargs["status"] == "active"
assert not running
assert "receipt-written" in events
events.append("marker-active-attempt")
if crash["next_active_write"]:
crash["next_active_write"] = False
raise SimulatedPowerLoss
marker_active["value"] = True
events.append("marker-active")
monkeypatch.setattr(specialize, "_write_marker", write_marker)
monkeypatch.setattr(specialize, "specialization_state", lambda: "active" if marker_active["value"] else "blocked")
if probe_fails:
monkeypatch.setattr(specialize, "_rollback_before_activation", lambda _journal: True)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="failed_rolled_back",
):
specialize._perform_migration_stages(journal, manifest, archive)
assert "build-probed" in events
assert "receipt-written" not in events
assert "marker-active-attempt" not in events
assert not marker_active["value"]
return
with pytest.raises(SimulatedPowerLoss):
specialize._perform_migration_stages(journal, manifest, archive)
assert journal["status"] == "active"
assert not marker_active["value"]
assert events.index("journal-active-durable") < events.index("marker-active-attempt")
marker = {
"status": "in_progress",
"sourceScriptSha256": specialize._APPROVED_SCRIPT_SHA256,
"imageSourceRevision": revision,
"imageSourceTree": tree,
}
script = (ROOT / "scripts" / "upgrade_ci_runner_fleet_specialized.sh").read_bytes()
monkeypatch.setattr(specialize, "_read_json_file", lambda *_args, **_kwargs: marker)
monkeypatch.setattr(specialize, "_valid_marker", lambda _marker: True)
monkeypatch.setattr(specialize, "_migration_lock", lambda: nullcontext())
monkeypatch.setattr(
specialize,
"verified_topology",
lambda: {
"ready": marker_active["value"],
"verifiedRunners": list(specialize._SPECIALIZED_NAMES) if marker_active["value"] else [],
"resourceEnvelope": {"ready": marker_active["value"], "digest": "e" * 64},
},
)
monkeypatch.setitem(
sys.modules,
"status_server",
type(
"StatusServerStub",
(),
{
"_require_controller_release": staticmethod(
lambda: {
"sourceRepository": specialize.SOURCE_REPOSITORY,
"sourceRevision": revision,
"sourceTree": tree,
}
)
},
),
)
monkeypatch.setattr(
specialize,
"_run_approved_shell_script",
lambda _script: specialize._perform_migration_stages(journal, manifest, archive),
)
result = specialize.execute(script)
assert result["status"] == "specialized"
assert maximum_running == 2
assert events.index("cgroup-intent-durable") < events.index("cgroup-ensured")
assert events.index("cgroup-ensured") < events.index("cgroup-result-durable")
assert batches == [
tuple(specialize.PROFILE_IDS[0:2]),
tuple(specialize.PROFILE_IDS[2:4]),
tuple(specialize.PROFILE_IDS[4:6]),
]
assert events.index("legacy-disabled") < events.index("registrations-created")
assert events.index("batch-stopped") < events.index("build-probed")
assert events.index("build-probed") < events.index("receipt-written")
assert events.index("receipt-written") < events.index("marker-active")
assert journal["cgroupCreated"] is True
assert journal["status"] == "active"
assert not running
@pytest.mark.parametrize("original_running", [False, True])
def test_migration_failure_restores_captured_dind_state_before_cgroup_cleanup(
specialize, tmp_path, monkeypatch, original_running
):
monkeypatch.setattr(specialize, "_rollback_maintenance", _fixture_rollback_maintenance)
image_id = specialize._DIND_IMAGE_ID
base_model = _base_v1_compose_model(specialize)
specialized_model = {
"services": {
"runner-dind": {
**base_model["services"]["runner-dind"],
"cgroup_parent": "itworx-ci",
"restart": "no",
"volumes": [
{
"type": "bind",
"source": str(specialize._host_volume_path(specialize.DIND_STORAGE_ROOT)),
"target": "/var/lib/docker",
"read_only": False,
}
],
}
}
}
journal = {
"originalDind": {
"present": True,
"running": original_running,
"restartPolicy": "unless-stopped",
"imageId": image_id,
},
"originalContainers": [],
"originalRegistrations": [],
"newRegistrationIds": {},
"guardFiles": [],
"cgroupCreated": True,
"cgroupEnsurePending": False,
"cgroupRootSubtree": ["cpu", "memory", "pids"],
"backupDirectory": str(tmp_path / "specialized-test"),
}
events = []
current = {"kind": "specialized", "State": {"Running": True}}
current_marker = {"value": current}
def inspect(name):
assert name == specialize.DIND_NAME or name in specialize._SPECIALIZED_NAMES
return current_marker["value"] if name == specialize.DIND_NAME else None
def run(args, **_kwargs):
events.append(tuple(args))
if args[:2] == ["docker", "update"] and args[-1] == specialize.DIND_NAME:
if args[2] == "--restart=no":
return ""
current_marker["value"]["HostConfig"]["RestartPolicy"]["Name"] = args[2].partition("=")[2]
return ""
if args[:2] == ["docker", "stop"]:
current_marker["value"]["State"]["Running"] = False
return ""
if args[:2] == ["docker", "rm"]:
current_marker["value"] = None
return ""
return ""
def run_compose(_compose, *args, **_kwargs):
events.append(("compose", *args))
assert args[-1] == "runner-dind"
restored_running = original_running if args[0] == "up" else False
current_marker["value"] = _legacy_dind_fixture(
specialize,
base_model,
image_id=image_id,
running=restored_running,
restart_policy="no",
)
base_bytes = json.dumps(base_model, sort_keys=True, separators=(",", ":")).encode()
specialized_bytes = json.dumps(specialized_model, sort_keys=True, separators=(",", ":")).encode()
live_compose = tmp_path / "docker-compose.yml"
live_compose.write_bytes(base_bytes)
monkeypatch.setattr(specialize, "COMPOSE_PATH", live_compose)
journal["baseComposeSha256"] = hashlib.sha256(base_bytes).hexdigest()
monkeypatch.setattr(specialize, "_guard_state_unchanged", lambda _journal: True)
monkeypatch.setattr(
specialize,
"_captured_compose_model",
lambda _journal, *, specialized: specialized_bytes if specialized else base_bytes,
)
monkeypatch.setattr(
specialize,
"_validate_dind_outer",
lambda item, _model: item.get("kind") == "specialized",
)
monkeypatch.setattr(specialize, "_inspect", inspect)
monkeypatch.setattr(specialize, "_run", run)
monkeypatch.setattr(specialize, "_run_compose", run_compose)
monkeypatch.setattr(specialize, "_runner_rows", lambda: [])
monkeypatch.setattr(
specialize,
"rollback_host_cgroup_family",
lambda **_kwargs: events.append("cgroup-cleanup") or {"familyExists": False},
)
monkeypatch.setattr(specialize, "SPECIALIZED_COMPOSE_PATH", tmp_path / "specialized-compose.yml")
monkeypatch.setattr(specialize, "TASK_IMAGE_MANIFEST_PATH", tmp_path / "task-image-manifest.json")
monkeypatch.setattr(specialize, "ADMISSION_BOOTSTRAP_RECEIPT_PATH", tmp_path / "receipt.json")
monkeypatch.setattr(specialize, "RUNNER_CONFIG_ROOT", tmp_path / "runner-configs")
monkeypatch.setattr(specialize, "_SPECIALIZED_DATA_ROOT", tmp_path / "runner-data")
monkeypatch.setattr(specialize, "_SPECIALIZED_WORKSPACE_ROOT", tmp_path / "workspaces")
monkeypatch.setattr(specialize, "TASK_IMAGE_ROOT", tmp_path / "task-images")
monkeypatch.setattr(specialize, "SPECIALIZED_STATE_PATH", tmp_path / "specialized.json")
monkeypatch.setattr(specialize, "MIGRATION_JOURNAL_PATH", tmp_path / "journal.json")
assert specialize._rollback_before_activation(journal) is True
restored = current_marker["value"]
assert restored["Image"] == image_id
assert restored["State"]["Running"] is original_running
assert restored["HostConfig"]["RestartPolicy"]["Name"] == "unless-stopped"
compose = next(event for event in events if isinstance(event, tuple) and event[0] == "compose")
assert compose[1] == ("up" if original_running else "create")
remove = next(index for index, event in enumerate(events) if isinstance(event, tuple) and event[:2] == ("docker", "rm"))
assert remove < events.index(compose) < events.index("cgroup-cleanup")
def _legacy_dind_fixture(
specialize, base_model, *, image_id, running, restart_policy
):
service = base_model["services"]["runner-dind"]
return {
"Name": f"/{specialize.DIND_NAME}",
"Image": image_id,
"Config": {
"Image": service["image"],
"User": "rootless",
"Entrypoint": ["dockerd-entrypoint.sh"],
"Cmd": ["--default-address-pool=base=198.19.0.0/16,size=27"],
"Volumes": {"/home/rootless/.local/share/docker": {}, "/var/lib/docker": {}},
"Labels": {
"com.docker.compose.project": specialize.PROJECT,
"com.docker.compose.service": "runner-dind",
},
},
"State": {"Running": running},
"HostConfig": {
"NanoCpus": 6_000_000_000,
"Memory": 12 * 1024**3,
"PidMode": "",
"Privileged": True,
"RestartPolicy": {"Name": restart_policy},
},
"Mounts": [
{
"Type": "bind",
"Source": mount["source"],
"Destination": mount["target"],
"RW": not mount.get("read_only", False),
}
for mount in service["volumes"]
] + [{
"Type": "volume", "Driver": "local", "Name": "f" * 64,
"Source": "/var/lib/docker/volumes/" + "f" * 64 + "/_data",
"Destination": "/var/lib/docker", "RW": True,
}],
}
def test_known_v1_and_v2_topologies_are_explicit_and_v1_does_not_require_absent_v2_services(
specialize, monkeypatch
):
compose = b"known Tower v1 compose fixture"
monkeypatch.setattr(specialize, "_KNOWN_V1_COMPOSE_SHA256", hashlib.sha256(compose).hexdigest())
v1_services = {
service_name: {"container_name": name}
for name, service_name in specialize._OLD_SERVICES.items()
if name in specialize._LEGACY_V1_NAMES
}
assert specialize._classify_legacy_topology(compose, v1_services) == "v1"
v2_services = {}
for name in specialize._LEGACY_V2_NAMES:
service_name = specialize._OLD_SERVICES[name]
burst = name.endswith("burst-1")
v2_services[service_name] = {
"container_name": name,
"image": specialize.RUNNER_IMAGE,
"profiles": ["runner"],
"labels": {
"tech.itworx.role": "gitea-actions-burst-runner" if burst else "gitea-actions-runner"
},
**({"cpus": 0.5, "mem_limit": "512m"} if name in specialize._LEGACY_V2_NAMES[2:] else {}),
}
assert specialize._classify_legacy_topology(b"known v2 compose", v2_services) == "v2"
partial_v2 = dict(v1_services)
partial_v2["runner-3"] = v2_services["runner-3"]
with pytest.raises(specialize.RunnerFleetSpecializeError, match="legacy_compose_invalid"):
specialize._classify_legacy_topology(compose, partial_v2)
def test_exact_live_v1_registration_and_container_contract_accepts_only_two_known_runners(specialize):
rows = [
{
"name": name,
"id": runner_id,
"labels": sorted(specialize._OLD_LABELS[name]),
"disabled": False,
"busy": False,
}
for runner_id, name in zip((9, 22), specialize._LEGACY_V1_NAMES, strict=True)
]
rows.extend(
[
{"name": "itworx-unraid-deploy", "id": 13, "labels": ["unraid-deploy"], "disabled": False, "busy": False},
{"name": "itworx-windows11-native", "id": 14, "labels": ["windows-native"], "disabled": False, "busy": False},
]
)
assert len(specialize._validate_legacy_registrations(rows, "v1")) == 2
name = "itworx-security-runner"
mount_entries = [
{"Source": source, "Destination": target, "RW": True}
for source, target, _read_only in specialize._V1_RUNNER_VOLUMES[name]
]
item = {
"Image": specialize.RUNNER_IMAGE_ID,
"Config": {
"Image": specialize.RUNNER_IMAGE,
"Labels": {
"com.docker.compose.project": specialize.PROJECT,
"com.docker.compose.service": "runner",
"tech.itworx.role": "gitea-actions-runner",
},
},
"State": {"Running": True},
"HostConfig": {
"NanoCpus": 1_000_000_000,
"Memory": 1024 * specialize._MIB,
"PidMode": "",
"RestartPolicy": {"Name": "unless-stopped"},
},
"Mounts": mount_entries,
}
assert specialize._validate_legacy_container_item("v1", name, item)["running"] is True
item["HostConfig"]["PidMode"] = "host"
with pytest.raises(specialize.RunnerFleetSpecializeError, match="v1_container_budget_invalid"):
specialize._validate_legacy_container_item("v1", name, item)
extra_known_runner = [*rows, {"name": "itworx-validation-runner-3", "id": 30}]
with pytest.raises(specialize.RunnerFleetSpecializeError, match="legacy_registration_invalid"):
specialize._validate_legacy_registrations(extra_known_runner, "v1")
def test_queue_gate_allows_only_legacy_labels_covered_by_new_profiles(specialize):
assert specialize._queued_job_is_covered(["self-hosted", "linux", "linux-validation", "ubuntu-latest"])
assert specialize._queued_job_is_covered(["itworx-ci-backend"])
assert specialize._queued_job_is_covered(["docker-validation"])
assert not specialize._queued_job_is_covered(["mcp-validation", "unraid"])
def _legacy_queue_job(job_id, status, labels, *, run_id=21500):
return {"id": job_id, "run_id": run_id, "status": status, "labels": labels}
def test_legacy_queue_gate_reads_queued_and_requested_admin_inventories(specialize, monkeypatch):
calls = []
def api(path):
calls.append(path)
if "status=pending" in path:
return {
"total_count": 1,
"jobs": [_legacy_queue_job(102, "requested", ["linux-validation", "ubuntu-latest"])],
}
if "status=queued" in path:
return {
"total_count": 1,
"jobs": [_legacy_queue_job(101, "queued", ["itworx-ci-backend"])],
}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
specialize._check_legacy_queued_jobs()
assert len(calls) == 2
assert "status=pending&limit=50&page=1&sort=id&order=asc" in calls[0]
assert "status=queued&limit=50&page=1&sort=id&order=asc" in calls[1]
def test_legacy_queue_gate_blocks_unsupported_requested_dependency_job(specialize, monkeypatch):
def api(path):
if "status=queued" in path:
return {"total_count": 0, "jobs": []}
if "status=pending" in path:
return {
"total_count": 1,
"jobs": [_legacy_queue_job(201, "requested", ["mcp-validation", "unraid"])],
}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_present",
):
specialize._check_legacy_queued_jobs()
def test_legacy_queue_gate_catches_requested_to_queued_transition_between_filters(
specialize, monkeypatch
):
state = "requested"
calls = []
def api(path):
nonlocal state
calls.append(path)
if "status=pending" in path:
# The job changes status immediately before this filtered read.
state = "queued"
return {"total_count": 0, "jobs": []}
if "status=queued" in path:
jobs = (
[_legacy_queue_job(250, "queued", ["mcp-validation", "unraid"])]
if state == "queued"
else []
)
return {"total_count": len(jobs), "jobs": jobs}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_present",
):
specialize._check_legacy_queued_jobs()
assert "status=pending" in calls[0]
assert "status=queued" in calls[1]
def test_legacy_queue_gate_paginates_51_jobs_and_blocks_unsupported_page_two_job(specialize, monkeypatch):
calls = []
def api(path):
calls.append(path)
if "status=queued" in path:
page = int(path.rsplit("page=", 1)[1].split("&", 1)[0])
if page == 1:
return {
"total_count": 51,
"jobs": [
_legacy_queue_job(index + 1, "queued", ["itworx-ci-backend"])
for index in range(50)
],
}
return {
"total_count": 51,
"jobs": [_legacy_queue_job(51, "queued", ["mcp-validation", "unraid"])],
}
if "status=pending" in path:
return {"total_count": 0, "jobs": []}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_present",
):
specialize._check_legacy_queued_jobs()
assert any("status=queued&limit=50&page=2" in path for path in calls)
def test_legacy_queue_gate_paginates_101_jobs_and_checks_unsupported_last_page(specialize, monkeypatch):
calls = []
def api(path):
calls.append(path)
if "status=queued" in path:
page = int(path.rsplit("page=", 1)[1].split("&", 1)[0])
start_id = (page - 1) * 50 + 1
page_size = min(50, 101 - (page - 1) * 50)
labels = ["mcp-validation", "unraid"] if page == 3 else ["itworx-ci-backend"]
return {
"total_count": 101,
"jobs": [
_legacy_queue_job(start_id + offset, "queued", labels)
for offset in range(page_size)
],
}
if "status=pending" in path:
return {"total_count": 0, "jobs": []}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_present",
):
specialize._check_legacy_queued_jobs()
assert any("status=queued&limit=50&page=3" in path for path in calls)
def test_legacy_queue_gate_rejects_total_count_changes_between_pages(specialize, monkeypatch):
def api(path):
if "status=queued" in path:
page = int(path.rsplit("page=", 1)[1].split("&", 1)[0])
if page == 1:
return {
"total_count": 51,
"jobs": [
_legacy_queue_job(index + 1, "queued", ["itworx-ci-backend"])
for index in range(50)
],
}
return {
"total_count": 102,
"jobs": [_legacy_queue_job(51, "queued", ["itworx-ci-backend"])],
}
if "status=pending" in path:
return {"total_count": 0, "jobs": []}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_untrusted",
):
specialize._check_legacy_queued_jobs()
def test_legacy_queue_gate_accepts_maximum_1000_job_inventory(specialize, monkeypatch):
calls = []
def api(path):
calls.append(path)
page = int(path.rsplit("page=", 1)[1].split("&", 1)[0])
start_id = (page - 1) * 50 + 1
return {
"total_count": 1000,
"jobs": [
_legacy_queue_job(start_id + offset, "queued", ["itworx-ci-backend"])
for offset in range(50)
],
}
monkeypatch.setattr(specialize, "_api_json", api)
rows = specialize._legacy_action_jobs("queued")
assert len(rows) == 1000
assert len(calls) == 20
assert "limit=50&page=20" in calls[-1]
def test_legacy_queue_gate_rejects_inventory_over_1000_jobs(specialize, monkeypatch):
calls = []
def api(path):
calls.append(path)
return {"total_count": 1001, "jobs": []}
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_untrusted",
):
specialize._legacy_action_jobs("queued")
assert len(calls) == 1
@pytest.mark.parametrize(
"payload",
[
{"total_count": 1, "jobs": []},
{"total_count": 1, "jobs": [_legacy_queue_job(1, "requested", ["itworx-ci-backend"])]},
{"total_count": 1, "jobs": [_legacy_queue_job(1, "queued", [""])]},
{"total_count": 1, "jobs": [_legacy_queue_job(1, "queued", ["itworx-ci-backend"], run_id=0)]},
{"total_count": 2, "jobs": [_legacy_queue_job(1, "queued", ["itworx-ci-backend"])]},
{
"total_count": 2,
"jobs": [
_legacy_queue_job(1, "queued", ["itworx-ci-backend"]),
_legacy_queue_job(1, "queued", ["itworx-ci-backend"]),
],
},
{
"total_count": 2,
"jobs": [
_legacy_queue_job(2, "queued", ["itworx-ci-backend"]),
_legacy_queue_job(1, "queued", ["itworx-ci-backend"]),
],
},
],
)
def test_legacy_queue_gate_rejects_malformed_or_truncated_inventory(specialize, monkeypatch, payload):
def api(path):
if "status=queued" in path:
return payload
if "status=pending" in path:
return {"total_count": 0, "jobs": []}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_untrusted",
):
specialize._check_legacy_queued_jobs()
def test_legacy_queue_gate_rejects_job_duplicated_across_status_snapshots(specialize, monkeypatch):
row = _legacy_queue_job(301, "queued", ["itworx-ci-backend"])
def api(path):
if "status=queued" in path:
return {"total_count": 1, "jobs": [row]}
if "status=pending" in path:
return {
"total_count": 1,
"jobs": [{**row, "status": "requested"}],
}
raise AssertionError(f"unexpected fixed Gitea API path: {path}")
monkeypatch.setattr(specialize, "_api_json", api)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_legacy_queue_untrusted",
):
specialize._check_legacy_queued_jobs()
def test_image_workflow_builds_from_protected_appops_main_with_gitea_checkout_url():
workflow = (ROOT / ".gitea" / "workflows" / "ci-task-images.yml").read_text(encoding="utf-8")
build = (ROOT / "scripts" / "build_ci_task_images.sh").read_text(encoding="utf-8")
frontend = (ROOT / "ops" / "runner-task-images" / "frontend" / "Dockerfile").read_text(encoding="utf-8")
assert "github-server-url: http://gitea-runner-internal:3000" in workflow
assert "runs-on: linux-validation" in workflow
assert 'bash scripts/build_ci_task_images.sh "$GITHUB_SHA" "$GITHUB_RUN_ID" "$output"' in workflow
assert "runs-on: itworx-ci-build" not in workflow
assert "--job-local" not in workflow
assert "jens/unraid-appops-gateway" in build
assert "Jens/Unraid-AppOps-Gateway" in build
assert "corepack prepare pnpm@11.15.0 --activate" in frontend
assert "test \"$(pnpm --version)\" = 11.15.0" in frontend
def test_specialization_shell_entrypoint_is_fixed_and_has_no_caller_selected_target():
script = ROOT / "scripts" / "upgrade_ci_runner_fleet_specialized.sh"
content = script.read_text(encoding="utf-8")
assert content.startswith("#!/usr/bin/env bash\n")
assert "SPECIALIZE_CI_RUNNER_FLEET" in content
assert "--apply-fixed" in content
assert "eval " not in content
assert "docker run" not in content
assert "docker compose" not in content
def test_gitea_pretty_registration_names_map_only_to_fixed_container_identity(
specialize, monkeypatch
):
registrations = [
{
"name": profile.display_name,
"id": index,
"labels": [profile.runner_label, *profile.legacy_labels],
"disabled": True,
"busy": False,
}
for index, profile in enumerate(specialize.RUNNER_PROFILES.values(), start=10)
]
monkeypatch.setattr(specialize, "_api_json", lambda _path: {"runners": registrations})
normalized = specialize._runner_rows()
assert [row["name"] for row in normalized] == list(specialize._SPECIALIZED_NAMES)
assert [row["displayName"] for row in normalized] == [
profile.display_name for profile in specialize.RUNNER_PROFILES.values()
]
registrations[0]["name"] = specialize.RUNNER_PROFILES["checks"].container_name
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="registration_name_invalid",
):
specialize._runner_rows()
registrations[0]["name"] = specialize.RUNNER_PROFILES["checks"].display_name
registrations[1]["name"] = specialize.RUNNER_PROFILES["checks"].display_name
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="registration_duplicate",
):
specialize._runner_rows()
def _stub_active_recovery(specialize, monkeypatch, *, family_exists=True):
rows, containers, dind, journal, model = _active_recovery_fixture(specialize)
events = []
class Volume:
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def assert_current(self):
events.append("volume-current")
monkeypatch.setattr(specialize, "_migration_lock", lambda **_kwargs: nullcontext())
monkeypatch.setattr(specialize, "_open_fixed_release_volume", lambda: Volume())
monkeypatch.setattr(
specialize,
"_active_release_bundle",
lambda: ({"status": "active"}, journal, model, b"captured-compose"),
)
monkeypatch.setattr(specialize, "_runner_rows", lambda: rows)
monkeypatch.setattr(specialize, "_docker_outer_inventory", lambda: (containers, []))
monkeypatch.setattr(specialize, "_validate_outer_item", lambda *_args: True)
monkeypatch.setattr(specialize, "_validate_dind_outer", lambda *_args: True)
monkeypatch.setattr(
specialize,
"_cgroup_family_snapshot",
lambda: {
"exists": family_exists,
"safeToStartDind": True,
"rootSubtreeControllers": ["cpu", "memory", "pids"],
},
)
monkeypatch.setattr(
specialize,
"_fresh_recovery_host_gate",
lambda: events.append("fresh-host-gate"),
)
monkeypatch.setattr(
specialize,
"_ensure_cgroup_family_details",
lambda **_kwargs: events.append("cgroup-create")
or {"created": True, "rootSubtreeBefore": ["cpu", "memory", "pids"]},
)
monkeypatch.setattr(
specialize,
"_run_compose",
lambda compose, *args, **kwargs: events.append(("compose", compose, args, kwargs)),
)
monkeypatch.setattr(specialize, "_wait_healthy", lambda *_args, **_kwargs: events.append("healthy"))
monkeypatch.setattr(specialize, "verify_profile_task_images", lambda _profiles: True)
monkeypatch.setattr(
specialize,
"_resource_proof",
lambda *_args: {"ready": True, "runningRunnerContainers": 0},
)
def inspect(name):
if name == specialize.DIND_NAME:
dind["State"] = {"Running": True, "Health": {"Status": "healthy"}}
return dind
return None
def inventory():
return containers, []
monkeypatch.setattr(specialize, "_inspect", inspect)
monkeypatch.setattr(specialize, "_docker_outer_inventory", inventory)
return events, rows, containers, dind
def test_active_recovery_restarts_only_fixed_dind_after_two_fresh_gates(
specialize, monkeypatch
):
events, _rows, _containers, _dind = _stub_active_recovery(
specialize, monkeypatch, family_exists=True
)
result = specialize.ensure_active_fleet_ready()
assert result == {"ready": True, "reason": None}
compose_events = [item for item in events if isinstance(item, tuple) and item[0] == "compose"]
assert len(compose_events) == 1
_, compose_bytes, args, kwargs = compose_events[0]
assert compose_bytes == b"captured-compose"
assert args == ("up", "-d", "--no-deps", "runner-dind")
assert kwargs["timeout"] <= 30
assert events.count("fresh-host-gate") == 2
assert events.index("fresh-host-gate") < events.index("healthy")
def test_active_recovery_rolls_back_empty_cgroup_if_host_guard_pauses_before_dind(
specialize, monkeypatch
):
events, _rows, _containers, _dind = _stub_active_recovery(
specialize, monkeypatch, family_exists=False
)
gate_count = {"value": 0}
def fresh_gate():
gate_count["value"] += 1
events.append(f"fresh-host-gate-{gate_count['value']}")
if gate_count["value"] == 2:
raise specialize.RunnerFleetSpecializeError(
"runner_fleet_specialization_host_guard_paused"
)
rollback = []
monkeypatch.setattr(specialize, "_fresh_recovery_host_gate", fresh_gate)
monkeypatch.setattr(
specialize,
"rollback_host_cgroup_family",
lambda **kwargs: rollback.append(kwargs) or {"removed": True, "familyExists": False},
)
result = specialize.ensure_active_fleet_ready()
assert result == {
"ready": False,
"reason": "runner_fleet_specialization_host_guard_paused",
}
assert rollback == [
{"created": True, "initial_controllers": ["cpu", "memory", "pids"]}
]
assert not any(isinstance(item, tuple) and item[0] == "compose" for item in events)
def test_active_recovery_accepts_only_disabled_legacy_registrations_after_containers_removed(
specialize, monkeypatch
):
events, rows, _containers, dind = _stub_active_recovery(
specialize, monkeypatch, family_exists=True
)
rows[-1]["disabled"] = False
result = specialize.ensure_active_fleet_ready()
assert result["ready"] is False
assert result["reason"] == "runner_fleet_specialization_legacy_registration_untrusted"
assert not any(isinstance(item, tuple) and item[0] == "compose" for item in events)
assert dind["State"]["Running"] is False
def test_active_recovery_cli_reports_only_sanitized_ready_or_reason(
specialize, monkeypatch, capsys
):
monkeypatch.setattr(
specialize,
"ensure_active_fleet_ready",
lambda: {"ready": False, "reason": "runner_fleet_specialization_host_guard_paused"},
)
assert specialize._main(("--ensure-active-fleet-ready",)) == 1
assert capsys.readouterr().err.strip() == "runner_fleet_specialization_host_guard_paused"
monkeypatch.setattr(
specialize,
"ensure_active_fleet_ready",
lambda: {"ready": True, "reason": None},
)
assert specialize._main(("--ensure-active-fleet-ready",)) == 0
assert capsys.readouterr().out.strip() == "ready"
def test_active_recovery_host_gate_rechecks_guard_after_fresh_sample(
specialize, monkeypatch
):
now = specialize.time.time()
sample = {
"loadRatio": 0.2,
"availableMemoryBytes": 16 * 1024**3,
"cacheFreeBytes": 40 * 1024**3,
"swapSampleAt": now,
"swapCritical": False,
}
guard_states = iter([("clear", ""), ("paused", "guard marker appeared")])
class HostSampler:
def host_snapshot(self):
return {"status": "observed", "generatedAt": "2026-10-02T00:00:00Z", "host": sample}
monkeypatch.setitem(
sys.modules,
"runner_fleet",
type("FleetStub", (), {"_host_guard_state": staticmethod(lambda: next(guard_states))}),
)
monkeypatch.setitem(
sys.modules,
"runner_capacity_controller",
type("CapacityStub", (), {"RunnerCapacityController": HostSampler}),
)
with pytest.raises(
specialize.RunnerFleetSpecializeError,
match="runner_fleet_specialization_host_guard_paused",
):
specialize._fresh_recovery_host_gate()
@pytest.mark.parametrize(
"sample,expected_reason",
[
(None, "runner_fleet_specialization_host_telemetry_unavailable"),
(
{
"loadRatio": 0.2,
"availableMemoryBytes": 16 * 1024**3,
"cacheFreeBytes": 40 * 1024**3,
"swapSampleAt": 1,
"swapCritical": False,
},
"runner_fleet_specialization_host_headroom_insufficient",
),
],
)
def test_active_recovery_host_gate_fails_closed_on_missing_or_stale_sample(
specialize, monkeypatch, sample, expected_reason
):
class HostSampler:
def host_snapshot(self):
if sample is None:
return {"status": "unavailable", "host": None}
return {"status": "observed", "generatedAt": "2026-10-02T00:00:00Z", "host": sample}
monkeypatch.setitem(
sys.modules,
"runner_fleet",
type("FleetStub", (), {"_host_guard_state": staticmethod(lambda: ("clear", ""))}),
)
monkeypatch.setitem(
sys.modules,
"runner_capacity_controller",
type("CapacityStub", (), {"RunnerCapacityController": HostSampler}),
)
with pytest.raises(specialize.RunnerFleetSpecializeError, match=expected_reason):
specialize._fresh_recovery_host_gate()