from __future__ import annotations import copy import hashlib import importlib.util import json import os import sys from pathlib import Path import pytest BROKER = Path(__file__).resolve().parents[2] / "ops" / "gitea-deploy-broker" def load(name, path): spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) sys.modules[name] = module spec.loader.exec_module(module) return module @pytest.fixture def recovery(monkeypatch): monkeypatch.syspath_prepend(str(BROKER)) return load("dind_health_recovery_test", BROKER / "runner_dind_health_cutover_recovery.py") @pytest.fixture def owned(recovery, monkeypatch, tmp_path): shared = load("dind_recovery_fixture", Path(__file__).with_name("test_runner_failed_cutover_recovery.py")) monkeypatch.setattr(recovery, "PROBE_ID", "a" * 64, raising=False) values = shared.owned.__wrapped__(recovery, monkeypatch, tmp_path) fleet, _, q, driver, marker, journal, _, rows, dind, events, pending = values journal.update(stage="registrations_created", cgroupCreated=True, cgroupEnsurePending=False, newRegistrationIds=dict(recovery.REGISTRATIONS), backupDirectory=recovery.BACKUP) for key, profile in fleet.RUNNER_PROFILES.items(): rows.append({"id": recovery.REGISTRATIONS[key], "name": profile.container_name, "displayName": profile.display_name, "labels": sorted({profile.runner_label, *profile.legacy_labels}), "disabled": True, "busy": False}) monkeypatch.setattr(fleet, "_validate_legacy_registrations", lambda current, _: [ {k: v for k, v in row.items() if k != "busy"} for row in current if row["id"] in (9, 22)]) manifest = fleet.TASK_IMAGE_MANIFEST_PATH.read_bytes() sha = hashlib.sha256(manifest).hexdigest() monkeypatch.setattr(recovery, "MANIFEST_SHA", sha) marker["imageManifestSha256"] = journal["imageManifestSha256"] = sha monkeypatch.setattr(recovery, "MARKER_SHA", hashlib.sha256(json.dumps(marker).encode()).hexdigest()) monkeypatch.setattr(recovery, "JOURNAL_SHA", hashlib.sha256(json.dumps(journal).encode()).hexdigest()) health = ["CMD-SHELL", "docker -H tcp://127.0.0.1:2375 info >/dev/null 2>&1"] dind["Id"] = recovery.DIND_ID dind["HostConfig"]["RestartPolicy"] = {"Name": "unless-stopped", "MaximumRetryCount": 0} dind["Config"]["Healthcheck"] = {"Test": health} dind["State"]["Health"]["Status"] = "unhealthy" model = {"services": {"runner-dind": {"healthcheck": {"test": health}}}} monkeypatch.setattr(fleet, "_captured_compose_mapping", lambda *a, **kw: copy.deepcopy(model)) expected = copy.deepcopy(dind) expected["HostConfig"]["RestartPolicy"]["Name"] = "no" def validate(item, _model): return all(item.get(key) == expected[key] for key in ("Id", "Image", "Config", "HostConfig", "Mounts", "NetworkSettings")) monkeypatch.setattr(fleet, "_validate_dind_outer", validate) monkeypatch.setattr(fleet, "ADMISSION_BOOTSTRAP_RECEIPT_PATH", tmp_path / "absent") monkeypatch.setattr(recovery, "_workspace_proof", lambda _: {"owned": "six-empty"}) monkeypatch.setattr(recovery, "_profile_state_proof", lambda _: {"private": "hash-only"}) monkeypatch.setattr(recovery, "_probe_absent", lambda: True) monkeypatch.setattr(recovery, "__file__", str(recovery.BUNDLE / "runner_dind_health_cutover_recovery.py")) def native(args, **kw): assert args[3].startswith("DOCKER_HOST=unix://") if args[6:] == ["info", "--format", "{{json .}}"]: return json.dumps({"ID": "fixed-inner-engine", "ServerVersion": "28.5.2"}) assert args[6:] == ["ps", "-a", "--no-trunc", "-q"] return "private-secret-canary" if pending else "" monkeypatch.setattr(driver, "_run", lambda self, *a, **kw: native(*a, **kw)) def effect(args, **kw): assert args == ["docker", "update", "--restart=no", recovery.DIND_ID] assert kw == {"timeout": 20, "max_output": 16 * 1024} events.append("exact-policy-update") dind["HostConfig"]["RestartPolicy"]["Name"] = "no" monkeypatch.setattr(fleet, "_run", effect) monkeypatch.setattr(recovery, "_load", lambda: (fleet, fleet, q, {})) return values def test_readonly_double_admission_never_claims_or_updates(recovery, owned, monkeypatch): fleet, *rest = owned monkeypatch.setattr(recovery, "_claim", lambda _: pytest.fail("read-only claim")) monkeypatch.setattr(fleet, "_run", lambda *a, **kw: pytest.fail("read-only effect")) assert recovery.run()["status"] == "ready" assert rest[-2].count("guard-cas-checked") == 2 def test_claim_precedes_exact_update_and_existing_locked_rollback(recovery, owned, monkeypatch): fleet, _, _, _, _, _, _, _, dind, events, _ = owned monkeypatch.setattr(recovery, "_claim", lambda _: events.append("claim")) def rollback(journal, barrier): barrier() assert dind["HostConfig"]["RestartPolicy"]["Name"] == "no" events.append("existing-rollback") return True monkeypatch.setattr(fleet, "_rollback_before_activation_locked", rollback) result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION) assert result["status"] == "rolled_back" and result["invocationConsumed"] is True assert events.index("claim") < events.index("exact-policy-update") < events.index("existing-rollback") < events.index("maintenance-exit") @pytest.mark.parametrize("change", ["cid", "image", "mount", "config", "health", "policy", "retry", "queue", "busy", "registration", "guard", "claim", "marker", "journal", "backup", "manifest", "workspace", "profile"]) def test_changed_incident_refuses_before_claim_or_effect(recovery, owned, monkeypatch, change): fleet, _, _, _, marker, journal, _, rows, dind, _, pending = owned if change in {"cid", "image"}: dind["Id" if change == "cid" else "Image"] = "f" * 64 elif change == "mount": dind["Mounts"].append({"Source": "foreign"}) elif change == "config": dind["Config"]["foreign"] = True elif change == "health": dind["Config"]["Healthcheck"]["Test"] = ["NONE"] elif change == "policy": dind["HostConfig"]["RestartPolicy"]["Name"] = "always" elif change == "retry": dind["HostConfig"]["RestartPolicy"]["MaximumRetryCount"] = 1 elif change == "queue": pending.append({"id": 1}) elif change == "busy": rows[-1]["busy"] = True elif change == "registration": rows[-1]["id"] = 999 elif change == "guard": monkeypatch.setattr(fleet, "_guard_script_plan", lambda _: (_ for _ in ()).throw(RuntimeError("private-secret-canary"))) elif change == "claim": recovery.CLAIM.write_text("consumed") elif change == "marker": marker["updatedAt"] = "foreign" elif change == "journal": journal["stage"] = "receipt_pending" elif change == "backup": journal["backupDirectory"] += "-foreign" monkeypatch.setattr(recovery, "JOURNAL_SHA", hashlib.sha256(json.dumps(journal).encode()).hexdigest()) elif change == "manifest": fleet.TASK_IMAGE_MANIFEST_PATH.write_text("{}") else: monkeypatch.setattr(recovery, "_workspace_proof" if change == "workspace" else "_profile_state_proof", lambda _: (_ for _ in ()).throw(recovery.RecoveryRejected("profile_state_changed"))) monkeypatch.setattr(recovery, "_claim", lambda _: pytest.fail("refusal must not claim")) monkeypatch.setattr(fleet, "_run", lambda *a, **kw: pytest.fail("refusal must not update")) result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION) assert result["status"] == "blocked" and result["effectsPerformed"] is False assert "private-secret-canary" not in str(result) def test_postclaim_policy_failure_is_unknown_and_claim_not_replayed(recovery, owned, monkeypatch): fleet = owned[0] monkeypatch.setattr(recovery, "_claim", lambda _: recovery.CLAIM.write_text("consumed")) monkeypatch.setattr(fleet, "_run", lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("private-secret-canary"))) first = recovery.run(execute=True, confirmation=recovery.CONFIRMATION) assert first["status"] == "unknown" and first["invocationConsumed"] is True and first["effectsPerformed"] is None second = recovery.run(execute=True, confirmation=recovery.CONFIRMATION) assert second["status"] == "blocked" and second["reason"] == "invocation_consumed_inspect_only" assert "private-secret-canary" not in str(first) + str(second) @pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="native root metadata") @pytest.mark.parametrize("change", [None, "extra_file", "config", "id", "labels", "ephemeral", "lock_content", "lock_mode", "lock_symlink"]) def test_native_profile_state_hashes_secret_without_exposing_it(recovery, monkeypatch, tmp_path, change): from types import SimpleNamespace root = tmp_path / "state" root.mkdir(mode=0o700) config, data = root / "config", root / "data" config.mkdir(mode=0o700) data.mkdir(mode=0o700) for key in ("checks", "build"): file = config / f"{key}.yaml" file.write_text("rendered-config") file.chmod(0o600) directory = data / key directory.mkdir(mode=0o700) file = directory / ".runner" file.write_text(json.dumps({"WARNING": "private metadata", "ephemeral": False, "id": recovery.REGISTRATIONS[key], "uuid": "private-uuid", "name": key, "token": "private-canary", "address": "https://gitea.example.com", "labels": [key + ":docker://fixed"]})) file.chmod(0o644) (directory / ".runner.lock").touch(mode=0o600) compose = root / "compose" compose.write_bytes(b"captured") compose.chmod(0o600) fleet = SimpleNamespace(RUNNER_PROFILES={key: SimpleNamespace(display_name=key, runner_label=key, legacy_labels=()) for key in ("checks", "build")}, _render_runner_config=lambda *a: b"rendered-config", _task_image_reference=lambda *a: "fixed", RUNNER_CONFIG_ROOT=config, _SPECIALIZED_DATA_ROOT=data, SPECIALIZED_COMPOSE_PATH=compose, _captured_compose_model=lambda *a, **kw: b"captured", _read_journal=lambda: {}) monkeypatch.setattr(recovery, "_parents", lambda _: None) if change == "extra_file": (data / "build" / "foreign").write_text("private-canary") elif change == "config": (config / "build.yaml").write_text("foreign") elif change in {"id", "labels", "ephemeral"}: path = data / "build" / ".runner" metadata = json.loads(path.read_text()) metadata[change] = {"id": 999, "labels": ["foreign"], "ephemeral": True}[change] path.write_text(json.dumps(metadata)) elif change == "lock_content": (data / "build" / ".runner.lock").write_text("foreign") elif change == "lock_mode": (data / "build" / ".runner.lock").chmod(0o644) elif change == "lock_symlink": path = data / "build" / ".runner.lock" path.unlink() path.symlink_to(compose) if change is not None: with pytest.raises(recovery.RecoveryRejected): recovery._profile_state_proof(fleet) else: proof = recovery._profile_state_proof(fleet) assert "secret-canary" not in str(proof) and "private-canary" not in str(proof) @pytest.mark.parametrize("change", ["cid", "registration", "journal", "private_work", "workspace"]) def test_postclaim_drift_never_reaches_policy_update(recovery, owned, monkeypatch, change): fleet, _, _, _, _, journal, _, rows, dind, _, pending = owned def claim(_): if change == "cid": dind["Id"] = "f" * 64 elif change == "registration": rows[-1]["disabled"] = False elif change == "journal": journal["foreign"] = True elif change == "private_work": pending.append({"id": 1}) else: monkeypatch.setattr(recovery, "_workspace_proof", lambda _: {"changed": True}) monkeypatch.setattr(recovery, "_claim", claim) monkeypatch.setattr(fleet, "_run", lambda *a, **kw: pytest.fail("changed admission must not update")) result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION) assert result["status"] == "unknown" and result["effectsPerformed"] is None assert result["invocationConsumed"] is True def test_inherited_health_counter_changes_only_do_not_change_process_identity(recovery, owned): fleet, _, q, _, _, _, _, _, dind, *_ = owned before = recovery._dind_identity(fleet, dind, q) dind["State"]["Health"].update(FailingStreak=99, Log=[{"Output": "secret-canary"}]) assert recovery._dind_identity(fleet, dind, q) == before dind["State"]["Pid"] += 1 assert recovery._dind_identity(fleet, dind, q) != before