311 lines
14 KiB
Python
311 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import stat
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from test_runner_acceptance_quiescence import module, proven, smart_database # noqa: F401
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
@pytest.fixture
|
|
def helper():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"root_recovery_review_test", ROOT / "scripts/recover_ci_acceptance_once.py")
|
|
helper = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(helper)
|
|
return helper
|
|
|
|
|
|
@pytest.fixture
|
|
def partial(helper, module, smart_database, monkeypatch, tmp_path):
|
|
driver, holder, pg, clients, waits = smart_database
|
|
baseline = driver.snapshot()
|
|
registrations = copy.deepcopy(driver.runners[:2])
|
|
driver.disable_restart(pg["id"])
|
|
driver.term(pg["id"])
|
|
for row in driver.runners[:2]:
|
|
row["disabled"] = True
|
|
holder["value"] = {
|
|
"schema": module.SCHEMA, "revision": helper.ORIGIN,
|
|
"manifestDigest": module._digest(module.OBSERVATION), "phase": "quiescing",
|
|
"baseline": baseline, "registrations": registrations,
|
|
"registrationProgress": {"9": "disabled", "22": "disabled"},
|
|
"progress": {pg["id"]: "term_intent"}, "guards": driver.capture_guards(),
|
|
"guardProgress": {"runner": "installed", "host": "installed"},
|
|
}
|
|
|
|
class Adapter:
|
|
def __init__(self, projects):
|
|
pass
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(driver, name)
|
|
|
|
def maintenance(self):
|
|
return driver.maintenance()
|
|
|
|
def snapshot(self):
|
|
return driver.snapshot()
|
|
|
|
monkeypatch.setattr(module, "ControllerDriver", Adapter)
|
|
monkeypatch.setattr(helper, "_load", lambda: (module, {}))
|
|
monkeypatch.setattr(helper, "_installed_proof", lambda: None)
|
|
monkeypatch.setattr(helper, "_read", lambda *args, **kwargs: b"private reviewed helper")
|
|
bundle = tmp_path / "bundle"
|
|
bundle.mkdir(mode=0o700)
|
|
monkeypatch.setattr(helper, "BUNDLE", bundle)
|
|
monkeypatch.setattr(helper, "__file__", str(bundle / "recover_ci_acceptance_once.py"))
|
|
monkeypatch.setattr(helper, "sys", SimpleNamespace(flags=SimpleNamespace(isolated=1),
|
|
dont_write_bytecode=True))
|
|
monkeypatch.setattr(helper, "os", SimpleNamespace(name="posix", geteuid=lambda: 0))
|
|
monkeypatch.setattr(helper, "stat", SimpleNamespace(
|
|
S_IMODE=lambda mode: 0o700))
|
|
claims = []
|
|
monkeypatch.setattr(helper, "_claim", lambda digest: claims.append(digest))
|
|
return driver, holder, pg, clients, claims
|
|
|
|
|
|
def test_default_preflight_is_readonly_and_preserves_receipt(helper, partial):
|
|
driver, holder, _, _, claims = partial
|
|
before = copy.deepcopy(holder["value"])
|
|
effects = list(driver.effects)
|
|
result = helper.run()
|
|
assert result["status"] == "partial" and result["effectsPerformed"] is False
|
|
assert result["candidateRevision"] == helper.CANDIDATE
|
|
assert result["receiptOrigin"] == helper.ORIGIN and result["normalDeployment"] is False
|
|
assert not claims and driver.effects == effects and holder["value"] == before
|
|
|
|
|
|
def test_reviewed_execution_resumes_same_receipt_and_two_clients(helper, module, partial):
|
|
driver, holder, pg, clients, claims = partial
|
|
before = holder["value"]
|
|
identity = (before["revision"], before["manifestDigest"], module._digest(before["baseline"]))
|
|
result = helper.run(execute=True, confirmation=helper.CONFIRMATION)
|
|
assert result["status"] == "quiesced" and result["invocationClaimed"] is True
|
|
assert result["retryAllowed"] is False and result["normalDeployment"] is False
|
|
assert len(claims) == 1
|
|
assert all(not row["state"]["Running"] for row in [pg, *clients])
|
|
after = holder["value"]
|
|
assert identity == (after["revision"], after["manifestDigest"], module._digest(after["baseline"]))
|
|
assert after["phase"] == "complete"
|
|
assert all(effect[0] in {"TERM", "restart", "registration"} for effect in driver.effects)
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["missing", "origin", "manifest", "complete", "guarding",
|
|
"restoring", "guards", "registrations", "cid", "stage"])
|
|
def test_foreign_or_new_maintenance_never_executes(helper, partial, change):
|
|
driver, holder, _, _, claims = partial
|
|
if change == "missing":
|
|
holder["value"] = None
|
|
else:
|
|
receipt = holder["value"]
|
|
if change == "origin":
|
|
receipt["revision"] = "a" * 40
|
|
elif change == "manifest":
|
|
receipt["manifestDigest"] = "a" * 64
|
|
elif change in {"complete", "guarding", "restoring"}:
|
|
receipt["phase"] = change
|
|
elif change == "guards":
|
|
receipt["guardProgress"]["host"] = "restored"
|
|
elif change == "registrations":
|
|
receipt["registrationProgress"]["22"] = "restored"
|
|
elif change == "cid":
|
|
receipt["progress"]["a" * 64] = "term_intent"
|
|
else:
|
|
receipt["progress"][next(iter(receipt["progress"]))] = "foreign"
|
|
effects = list(driver.effects)
|
|
result = helper.run(execute=True, confirmation=helper.CONFIRMATION)
|
|
assert result["status"] == "blocked" and result["effectsPerformed"] is False
|
|
assert not claims and driver.effects == effects
|
|
|
|
|
|
@pytest.mark.parametrize("fault", ["queued", "changed-container", "busy-lock", "busy-fleet-lock", "claimed-marker"])
|
|
def test_admission_races_before_claim_refuse(helper, partial, monkeypatch, fault):
|
|
from contextlib import contextmanager
|
|
driver, _, _, _, claims = partial
|
|
if fault == "queued":
|
|
driver.pending_jobs = lambda: [{"status": "queued"}]
|
|
elif fault == "changed-container":
|
|
driver.state["rows"][0]["imageId"] = "sha256:" + "0" * 64
|
|
elif fault == "busy-lock":
|
|
@contextmanager
|
|
def busy():
|
|
raise RuntimeError("sensitive unexpected detail")
|
|
yield
|
|
driver.maintenance = busy
|
|
elif fault == "busy-fleet-lock":
|
|
@contextmanager
|
|
def busy_fleet(**kwargs):
|
|
raise RuntimeError("fleet lock busy")
|
|
yield
|
|
module = driver.module
|
|
monkeypatch.setattr(module.fleet, "_migration_lock", busy_fleet)
|
|
else:
|
|
def claimed(digest):
|
|
raise helper.RecoveryRejected("invocation_already_claimed_inspect_only")
|
|
monkeypatch.setattr(helper, "_claim", claimed)
|
|
effects = list(driver.effects)
|
|
result = helper.run(execute=True, confirmation=helper.CONFIRMATION)
|
|
assert result["status"] == "blocked" and not claims and driver.effects == effects
|
|
assert "sensitive" not in json.dumps(result)
|
|
|
|
|
|
def test_failure_after_claim_is_unknown_never_blind_retry(helper, partial, monkeypatch):
|
|
driver, _, _, _, claims = partial
|
|
driver.term = lambda cid: (_ for _ in ()).throw(RuntimeError("token-secret-value"))
|
|
result = helper.run(execute=True, confirmation=helper.CONFIRMATION)
|
|
assert len(claims) == 1 and result["status"] == "unknown"
|
|
assert result["effectsPerformed"] is None and result["retryAllowed"] is False
|
|
assert "token-secret-value" not in json.dumps(result)
|
|
|
|
|
|
@pytest.mark.parametrize("execute,confirmation", [(True, None), (True, "other"), (False, "other")])
|
|
def test_confirmation_cannot_be_selected(helper, partial, execute, confirmation):
|
|
with pytest.raises(helper.RecoveryRejected, match="explicit_fixed_confirmation_required"):
|
|
helper.run(execute=execute, confirmation=confirmation)
|
|
assert not partial[-1]
|
|
|
|
|
|
def test_fixed_artifact_inventory_and_candidate_hash_match_immutable_code(helper):
|
|
source = ROOT / "ops/gitea-deploy-broker"
|
|
tree = ast.parse((source / "status_server.py").read_text())
|
|
files = next(ast.literal_eval(node.value) for node in tree.body
|
|
if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name)
|
|
and target.id == "_CONTROLLER_ARTIFACT_FILES" for target in node.targets))
|
|
assert helper.FILES == files
|
|
# The private bundle must contain Git's LF bytes, not a Windows CRLF checkout.
|
|
canonical = (source / helper.MODULE_NAME).read_bytes().replace(b"\r\n", b"\n")
|
|
assert hashlib.sha256(canonical).hexdigest() == helper.CANDIDATE_SHA
|
|
assert helper.ORIGIN != helper.CANDIDATE
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["module", "artifact", "revision", "tree", "version"])
|
|
def test_installed_identity_pin_rejects_drift(helper, monkeypatch, change):
|
|
raws = {name: name.encode() for name in helper.FILES}
|
|
original_hash = hashlib.sha256(raws[helper.MODULE_NAME]).hexdigest()
|
|
digest = hashlib.sha256()
|
|
for name, raw in raws.items():
|
|
digest.update(name.encode() + b"\0" + raw + b"\0")
|
|
monkeypatch.setattr(helper, "ORIGINAL_MODULE_SHA", original_hash)
|
|
monkeypatch.setattr(helper, "ARTIFACT_SHA", digest.hexdigest())
|
|
release = {"apiVersion": "itworx.controller-artifact/v1", "controllerVersion": "2.1.21",
|
|
"sourceRepository": "jens/unraid-appops-gateway", "sourceRevision": helper.ORIGIN,
|
|
"sourceTree": helper.ORIGIN_TREE, "artifactSha256": helper.ARTIFACT_SHA,
|
|
"builtAt": "2026-10-10T00:00:00Z"}
|
|
if change in {"module", "artifact"}:
|
|
raws[helper.MODULE_NAME if change == "module" else helper.FILES[0]] += b"altered"
|
|
else:
|
|
release[{"revision": "sourceRevision", "tree": "sourceTree",
|
|
"version": "controllerVersion"}[change]] = "different"
|
|
monkeypatch.setattr(helper, "_read", lambda path, **kwargs:
|
|
json.dumps(release).encode() if path.name == "controller-release.json"
|
|
else raws[path.name])
|
|
with pytest.raises(helper.RecoveryRejected):
|
|
helper._installed_proof()
|
|
|
|
|
|
def test_candidate_hash_refuses_before_any_import(helper, monkeypatch):
|
|
monkeypatch.setattr(helper, "_installed_proof", lambda: None)
|
|
monkeypatch.setattr(helper, "_read", lambda *args, **kwargs: b"foreign candidate")
|
|
with pytest.raises(helper.RecoveryRejected, match="candidate_bytes_changed"):
|
|
helper._load()
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["root", "isolated", "bytecode"])
|
|
def test_untrusted_interpreter_refuses(helper, partial, monkeypatch, field):
|
|
if field == "root":
|
|
helper.os.geteuid = lambda: 1000
|
|
elif field == "isolated":
|
|
helper.sys.flags.isolated = 0
|
|
else:
|
|
helper.sys.dont_write_bytecode = False
|
|
with pytest.raises(helper.RecoveryRejected, match="isolated_root_python_required"):
|
|
helper.run()
|
|
assert not partial[-1]
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="native root POSIX file safety")
|
|
@pytest.mark.parametrize("boundary", ["before-link", "after-link"])
|
|
def test_claim_publication_failure_is_atomic_and_never_replayed(helper, monkeypatch, tmp_path, boundary):
|
|
bundle = tmp_path / "bundle"
|
|
bundle.mkdir(mode=0o700)
|
|
monkeypatch.setattr(helper, "BUNDLE", bundle)
|
|
monkeypatch.setattr(helper, "MARKER", bundle / "invoked.json")
|
|
monkeypatch.setattr(helper, "_trusted_parents", lambda path: None)
|
|
calls = 0
|
|
|
|
def fsync(fd):
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == (1 if boundary == "before-link" else 2):
|
|
raise OSError("simulated actual durability failure")
|
|
os.fsync(fd)
|
|
|
|
proxy = SimpleNamespace(**{name: getattr(os, name) for name in dir(os)})
|
|
proxy.fsync = fsync
|
|
monkeypatch.setattr(helper, "os", proxy)
|
|
with pytest.raises(OSError):
|
|
helper._claim("a" * 64)
|
|
marker = bundle / "invoked.json"
|
|
assert marker.exists() is (boundary == "after-link")
|
|
if marker.exists():
|
|
assert json.loads(helper._read(marker, private=True))["candidateRevision"] == helper.CANDIDATE
|
|
with pytest.raises(helper.RecoveryRejected, match="invocation_already_claimed"):
|
|
helper._claim("a" * 64)
|
|
assert not list(bundle.glob(".acceptance-recovery-invocation-*"))
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="native root POSIX file safety")
|
|
def test_private_marker_atomic_once_and_reader_metadata(helper, monkeypatch, tmp_path):
|
|
bundle = tmp_path / "bundle"
|
|
bundle.mkdir(mode=0o700)
|
|
monkeypatch.setattr(helper, "BUNDLE", bundle)
|
|
monkeypatch.setattr(helper, "MARKER", bundle / "invoked.json")
|
|
# pytest's sandbox ancestors are not the deployment bundle; test leaf metadata here.
|
|
monkeypatch.setattr(helper, "_trusted_parents", lambda path: None)
|
|
helper._claim("a" * 64)
|
|
marker = bundle / "invoked.json"
|
|
assert stat.S_IMODE(marker.stat().st_mode) == 0o600
|
|
assert json.loads(helper._read(marker, private=True))["receiptOrigin"] == helper.ORIGIN
|
|
with pytest.raises(helper.RecoveryRejected, match="invocation_already_claimed"):
|
|
helper._claim("a" * 64)
|
|
with pytest.raises(helper.RecoveryRejected, match="invocation_already_claimed"):
|
|
helper._claim("b" * 64)
|
|
for mode in (0o644, 0o640, 0o664, 0o777):
|
|
marker.chmod(mode)
|
|
with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"):
|
|
helper._read(marker, private=True)
|
|
marker.chmod(0o600)
|
|
os.chown(marker, 1000, 0)
|
|
with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"):
|
|
helper._read(marker, private=True)
|
|
os.chown(marker, 0, 0)
|
|
link = bundle / "extra-link"
|
|
os.link(marker, link)
|
|
with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"):
|
|
helper._read(marker, private=True)
|
|
link.unlink()
|
|
proxy = SimpleNamespace(**{name: getattr(os, name) for name in dir(os)})
|
|
proxy.fstat = lambda fd: SimpleNamespace(**{
|
|
key: getattr(os.fstat(fd), key) + (1 if key == "st_ino" else 0)
|
|
for key in ("st_dev", "st_ino", "st_uid", "st_gid", "st_mode", "st_nlink", "st_size",
|
|
"st_mtime_ns", "st_ctime_ns")})
|
|
with monkeypatch.context() as scope:
|
|
scope.setattr(helper, "os", proxy)
|
|
with pytest.raises(helper.RecoveryRejected, match="recovery_file_changed"):
|
|
helper._read(marker, private=True)
|
|
marker.unlink()
|
|
marker.symlink_to(bundle / "absent")
|
|
with pytest.raises(helper.RecoveryRejected, match="recovery_file_untrusted"):
|
|
helper._read(marker, private=True)
|