279 lines
14 KiB
Python
279 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
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_tail_test", BROKER / "runner_dind_rollback_tail_recovery.py")
|
|
|
|
|
|
@pytest.fixture
|
|
def owned(recovery, monkeypatch, tmp_path):
|
|
shared = load("tail_shared", 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,
|
|
cgroupRootSubtree=list(recovery.CONTROLLERS), originalContainers=[])
|
|
for row in rows:
|
|
row.update(displayName=None, status="online")
|
|
rows.extend([{"id": rid, "name": name, "labels": [label], "disabled": False, "busy": False,
|
|
"displayName": None, "status": "online"}
|
|
for rid, name, label in ((13, "itworx-unraid-deploy", "unraid-deploy"),
|
|
(14, "itworx-windows11-native", "windows-native"))])
|
|
fleet.TASK_IMAGE_MANIFEST_PATH.unlink()
|
|
fleet.TASK_IMAGE_ROOT.rmdir()
|
|
monkeypatch.setattr(fleet, "ADMISSION_BOOTSTRAP_RECEIPT_PATH", tmp_path / "absent")
|
|
dind["Id"] = recovery.RUNTIME_IDS[fleet.DIND_NAME]
|
|
legacy = {name: {**copy.deepcopy(dind), "Id": cid} for name, cid in recovery.RUNTIME_IDS.items()
|
|
if name != fleet.DIND_NAME}
|
|
monkeypatch.setattr(fleet, "_validate_legacy_containers", lambda _: (copy.deepcopy(legacy), []))
|
|
monkeypatch.setattr(fleet, "_inspect", lambda name: copy.deepcopy(dind) if name == fleet.DIND_NAME else copy.deepcopy(legacy.get(name)))
|
|
family = {"exists": True, "controllers": list(recovery.CONTROLLERS),
|
|
"rootSubtreeControllers": list(recovery.CONTROLLERS), "familySubtreeControllers": list(recovery.CONTROLLERS),
|
|
"groupCount": 1, "hasProcesses": False, "familyCpuMax": "max 100000", "familyMemoryMax": "max"}
|
|
monkeypatch.setattr(fleet, "_cgroup_family_snapshot", lambda: copy.deepcopy(family))
|
|
monkeypatch.setattr(recovery, "_probe_absent", lambda: True)
|
|
monkeypatch.setattr(recovery, "_prior_receipt", lambda: b"pinned")
|
|
monkeypatch.setattr(recovery, "MARKER_SHA", hashlib.sha256(json.dumps(marker).encode()).hexdigest())
|
|
monkeypatch.setattr(recovery, "JOURNAL_SHA", hashlib.sha256(json.dumps(journal).encode()).hexdigest())
|
|
monkeypatch.setattr(recovery, "__file__", str(recovery.BUNDLE / "runner_dind_rollback_tail_recovery.py"))
|
|
def legacy_private_inventory(self, args, **kwargs):
|
|
assert args == ["docker", "exec", "-e", "DOCKER_HOST=tcp://127.0.0.1:2375",
|
|
fleet.DIND_NAME, "docker", "ps", "-a", "--no-trunc", "-q"]
|
|
assert kwargs == {"timeout": 20, "max_output": 128 * 1024}
|
|
return ""
|
|
monkeypatch.setattr(driver, "_run", legacy_private_inventory)
|
|
monkeypatch.setattr(recovery, "_load", lambda: (fleet, fleet, q, {}))
|
|
def ensure(*, allow_existing):
|
|
assert allow_existing is True
|
|
events.append("ensure")
|
|
family.update(familyCpuMax="800000 100000", familyMemoryMax=str(16 * 1024**3))
|
|
return False
|
|
monkeypatch.setattr(fleet, "_ensure_cgroup_family", ensure)
|
|
return values, family, legacy
|
|
|
|
|
|
def test_readonly_discloses_probe_and_never_claims_or_ensures(recovery, owned, monkeypatch):
|
|
values, _, _ = owned
|
|
monkeypatch.setattr(recovery, "_claim", lambda _: pytest.fail("claim"))
|
|
monkeypatch.setattr(values[0], "_ensure_cgroup_family", lambda **kw: pytest.fail("ensure"))
|
|
result = recovery.run()
|
|
assert result["status"] == "ready"
|
|
assert result["diagnosticProbeLifecycle"] is True
|
|
assert result["recoveryEffectsPerformed"] is False
|
|
assert values[9].count("guard-cas-checked") == 2
|
|
|
|
|
|
def test_claim_ensure_native_rollback_all_under_both_locks(recovery, owned, monkeypatch):
|
|
values, family, _ = owned
|
|
fleet, _, _, _, _, _, _, _, _, events, _ = values
|
|
monkeypatch.setattr(recovery, "_claim", lambda _: events.append("claim"))
|
|
def tail(journal, barrier):
|
|
barrier()
|
|
assert events.index("claim") < events.index("ensure")
|
|
assert "maintenance-exit" not in events and "migration-exit" not in events
|
|
assert family["familyCpuMax"] == "800000 100000"
|
|
events.append("native-tail")
|
|
return True
|
|
monkeypatch.setattr(fleet, "_rollback_before_activation_locked", tail)
|
|
result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION)
|
|
assert result["status"] == "rolled_back" and result["invocationConsumed"] is True
|
|
assert events.index("native-tail") < events.index("maintenance-exit") < events.index("migration-exit")
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["cid", "unhealthy", "busy", "extra-registration", "specialized-file", "guards", "private-work",
|
|
"family-process", "family-child", "family-cpu", "family-memory", "family-subtree", "journal", "claim"])
|
|
def test_mutations_rejected_before_claim_and_effect(recovery, owned, monkeypatch, mutation):
|
|
values, family, legacy = owned
|
|
fleet, _, _, _, _, journal, _, rows, dind, events, pending = values
|
|
if mutation == "cid":
|
|
dind["Id"] = "f" * 64
|
|
elif mutation == "unhealthy":
|
|
dind["State"]["Health"]["Status"] = "unhealthy"
|
|
elif mutation == "busy":
|
|
rows[0]["busy"] = True
|
|
elif mutation == "extra-registration":
|
|
rows.append({"id": 23, "busy": False})
|
|
elif mutation == "specialized-file":
|
|
fleet.RUNNER_CONFIG_ROOT.write_text("private-canary")
|
|
elif mutation == "guards":
|
|
monkeypatch.setattr(fleet, "_guard_state_unchanged", lambda _: False)
|
|
elif mutation == "private-work":
|
|
monkeypatch.setattr(values[3], "_run", lambda *a, **kw: "private-canary")
|
|
elif mutation == "family-process":
|
|
family["hasProcesses"] = True
|
|
elif mutation == "family-child":
|
|
family["groupCount"] = 2
|
|
elif mutation == "family-cpu":
|
|
family["familyCpuMax"] = "800000 100000"
|
|
elif mutation == "family-memory":
|
|
family["familyMemoryMax"] = "1"
|
|
elif mutation == "family-subtree":
|
|
family["familySubtreeControllers"].pop()
|
|
elif mutation == "journal":
|
|
journal["backupDirectory"] += "foreign"
|
|
elif mutation == "claim":
|
|
recovery.CLAIM.write_text("consumed")
|
|
monkeypatch.setattr(recovery, "_claim", lambda _: pytest.fail("claim"))
|
|
result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION)
|
|
assert result["status"] == "blocked"
|
|
assert result["recoveryEffectsPerformed"] is False
|
|
assert "ensure" not in events and "private-canary" not in json.dumps(result)
|
|
|
|
|
|
def test_admission_drift_refuses_before_claim(recovery, owned, monkeypatch):
|
|
original = recovery._admit
|
|
calls = []
|
|
def drift(*args, **kwargs):
|
|
journal, proof = original(*args, **kwargs)
|
|
calls.append(1)
|
|
return journal, {**proof, "runtimeDigest": str(len(calls))}
|
|
monkeypatch.setattr(recovery, "_admit", drift)
|
|
monkeypatch.setattr(recovery, "_claim", lambda _: pytest.fail("claim"))
|
|
assert recovery.run(execute=True, confirmation=recovery.CONFIRMATION)["reason"] == "admission_changed"
|
|
|
|
|
|
def test_ensure_failure_consumes_and_reports_unknown_redacted(recovery, owned, monkeypatch):
|
|
monkeypatch.setattr(recovery, "_claim", lambda _: None)
|
|
def fail(**kwargs):
|
|
raise RuntimeError("token=private-canary")
|
|
monkeypatch.setattr(owned[0][0], "_ensure_cgroup_family", fail)
|
|
result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION)
|
|
assert result["status"] == "unknown" and result["invocationConsumed"] is True
|
|
assert result["recoveryEffectsPerformed"] is None and "private-canary" not in json.dumps(result)
|
|
|
|
|
|
def test_prior_receipt_closed_schema_and_raw_hash(recovery, monkeypatch):
|
|
raw = json.dumps({"schema": "appops.failed-cutover-recovery/v1", "phase": "consumed",
|
|
"incidentJobId": recovery.INCIDENT_JOB, "incidentBackupDirectory": recovery.BACKUP,
|
|
"sourceRevision": recovery.ORIGIN, "sourceTree": recovery.TREE, "artifactSha256": recovery.ARTIFACT_SHA,
|
|
"candidateSha256": recovery.FLEET_SHA, "proof": recovery.PRIOR_PROOF}).encode()
|
|
monkeypatch.setattr(recovery, "_read", lambda *a, **kw: raw)
|
|
monkeypatch.setattr(recovery, "PRIOR_SHA", hashlib.sha256(raw).hexdigest())
|
|
assert recovery._prior_receipt() == raw
|
|
for changed in ({"extra": "private-canary"}, {"incidentJobId": "other"}, {"proof": {}}):
|
|
altered = json.dumps({**json.loads(raw), **changed}).encode()
|
|
monkeypatch.setattr(recovery, "_read", lambda *a, **kw: altered)
|
|
monkeypatch.setattr(recovery, "PRIOR_SHA", hashlib.sha256(altered).hexdigest())
|
|
with pytest.raises(recovery.RecoveryRejected, match="prior_receipt_changed"):
|
|
recovery._prior_receipt()
|
|
|
|
|
|
def test_postclaim_runtime_change_refuses_before_ensure(recovery, owned, monkeypatch):
|
|
values, _, _ = owned
|
|
def consume(_):
|
|
values[8]["Id"] = "f" * 64
|
|
monkeypatch.setattr(recovery, "_claim", consume)
|
|
result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION)
|
|
assert result["status"] == "unknown" and result["invocationConsumed"] is True
|
|
assert "ensure" not in values[9]
|
|
|
|
|
|
def test_once_claim_cannot_be_replayed(recovery, owned, monkeypatch):
|
|
values, _, _ = owned
|
|
def consume(_):
|
|
recovery.CLAIM.write_text("consumed")
|
|
monkeypatch.setattr(recovery, "_claim", consume)
|
|
monkeypatch.setattr(values[0], "_rollback_before_activation_locked", lambda journal, barrier: False)
|
|
assert recovery.run(execute=True, confirmation=recovery.CONFIRMATION)["status"] == "unknown"
|
|
result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION)
|
|
assert result["status"] == "blocked" and result["reason"] == "invocation_consumed_inspect_only"
|
|
|
|
|
|
def test_claim_write_uncertainty_is_unknown(recovery, owned, monkeypatch):
|
|
def uncertain(_):
|
|
raise OSError("secret-canary")
|
|
monkeypatch.setattr(recovery, "_claim", uncertain)
|
|
result = recovery.run(execute=True, confirmation=recovery.CONFIRMATION)
|
|
assert result["status"] == "unknown" and result["invocationConsumed"] is None
|
|
assert result["recoveryEffectsPerformed"] is None and "secret-canary" not in json.dumps(result)
|
|
assert "ensure" not in owned[0][9]
|
|
|
|
|
|
def test_all_runtime_ids_are_full_immutable_ids(recovery):
|
|
assert all(len(cid) == 64 and set(cid) <= set("0123456789abcdef") for cid in recovery.RUNTIME_IDS.values())
|
|
assert recovery.CANDIDATE_SHA == recovery.FLEET_SHA
|
|
assert recovery.CLAIM != recovery.PRIOR_CLAIM
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform == "win32", reason="native POSIX ownership required")
|
|
def test_native_existing_migration_lock_rejects_missing_and_retains_lock(recovery):
|
|
import os
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
if os.geteuid() != 0:
|
|
pytest.skip("native root ownership required")
|
|
parent = Path("/data") if Path("/data").is_dir() else Path("/root")
|
|
recovery._parents(parent / "native-test-root")
|
|
with tempfile.TemporaryDirectory(prefix="appops-tail-lock-test-", dir=parent) as temporary:
|
|
root = Path(temporary)
|
|
root.chmod(0o700)
|
|
fleet = SimpleNamespace(MIGRATION_JOURNAL_PATH=root / "journal.json")
|
|
with pytest.raises(FileNotFoundError):
|
|
with recovery._migration_existing_lock(fleet):
|
|
pytest.fail("missing lock admitted")
|
|
lock = root / "specialization.lock"
|
|
lock.write_bytes(b"")
|
|
lock.chmod(0o600)
|
|
with recovery._migration_existing_lock(fleet) as barrier:
|
|
barrier()
|
|
with pytest.raises(recovery.RecoveryRejected, match="migration_lock_busy"):
|
|
with recovery._migration_existing_lock(fleet):
|
|
pytest.fail("second lock acquired")
|
|
assert lock.exists()
|
|
|
|
|
|
@pytest.mark.skipif(sys.platform == "win32", reason="native POSIX ownership required")
|
|
def test_native_untrusted_parent_refused_before_missing_lock(recovery):
|
|
import os
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
if os.geteuid() != 0:
|
|
pytest.skip("native root ownership required")
|
|
parent = Path("/data") if Path("/data").is_dir() else Path("/root")
|
|
recovery._parents(parent / "native-test-root")
|
|
with tempfile.TemporaryDirectory(prefix="appops-tail-untrusted-test-", dir=parent) as temporary:
|
|
root = Path(temporary)
|
|
root.chmod(0o777)
|
|
fleet = SimpleNamespace(MIGRATION_JOURNAL_PATH=root / "journal.json")
|
|
with pytest.raises(recovery.RecoveryRejected, match="root_path_untrusted"):
|
|
with recovery._migration_existing_lock(fleet):
|
|
pytest.fail("untrusted parent admitted")
|
|
|
|
|
|
|
|
def test_tail_uses_captured_legacy_tcp_not_specialized_unix(recovery, owned, monkeypatch):
|
|
values, _, _ = owned
|
|
fleet, _, _, driver, *_ = values
|
|
seen = []
|
|
def inventory(self, args, **kwargs):
|
|
assert args[3] == "DOCKER_HOST=tcp://127.0.0.1:2375"
|
|
assert "unix://" not in " ".join(args)
|
|
seen.append(args[3])
|
|
return ""
|
|
monkeypatch.setattr(driver, "_run", inventory)
|
|
monkeypatch.setattr(fleet, "_DIND_DOCKER_HOST", "unix:///run/user/1000/docker.sock")
|
|
assert recovery.run()["status"] == "ready"
|
|
assert seen == ["DOCKER_HOST=tcp://127.0.0.1:2375"] * 2
|