259 lines
13 KiB
Python
259 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
import builtins
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from test_runner_acceptance_quiescence import module, proven # noqa: F401
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
@pytest.fixture
|
|
def helper():
|
|
spec = importlib.util.spec_from_file_location("api_child_recovery_test",
|
|
ROOT / "scripts/recover_ci_api_term_once.py")
|
|
result = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(result)
|
|
return result
|
|
|
|
|
|
@pytest.fixture
|
|
def prepared(helper, module, proven, monkeypatch, tmp_path):
|
|
from contextlib import nullcontext
|
|
driver, holder = proven
|
|
baseline = driver.snapshot()
|
|
holder["value"] = {
|
|
"schema": module.SCHEMA, "revision": module.REVIEWED_RECEIPT_PREDECESSOR,
|
|
"manifestDigest": module._digest(module.OBSERVATION), "phase": "quiescing",
|
|
"baseline": baseline, "registrations": copy.deepcopy(driver.runners[:2]),
|
|
"registrationProgress": {"9": "disabled", "22": "disabled"},
|
|
"progress": {row["id"]: "term_intent" for row in baseline["rows"]},
|
|
"guards": driver.capture_guards(), "guardProgress": {"runner": "installed", "host": "installed"},
|
|
}
|
|
for row in driver.state["rows"]:
|
|
row["restartPolicy"] = {"Name": "no", "MaximumRetryCount": 0}
|
|
for row in driver.runners[:2]:
|
|
row["disabled"] = True
|
|
driver._effect_admission = module.ControllerDriver._effect_admission.__get__(driver)
|
|
base = SimpleNamespace(ORIGIN=module.REVIEWED_RECEIPT_PREDECESSOR,
|
|
CANDIDATE="07c75a3aae4d0881f93e4bef2aa0275235c4253a",
|
|
CANDIDATE_SHA="236af955566bd2713684977008045c7b9f1390b83fff4b160da83c6473bf4246",
|
|
MARKER=tmp_path / "original-marker.json")
|
|
def original_scope(module, receipt):
|
|
assert receipt is not None and receipt["revision"] == base.ORIGIN
|
|
assert receipt["phase"] == "quiescing"
|
|
assert receipt["manifestDigest"] == module._digest(module.OBSERVATION)
|
|
base._scope = original_scope
|
|
base._load = lambda: (module, {})
|
|
base._installed_proof = lambda: None
|
|
base._read = lambda *args, **kwargs: json.dumps({"candidateRevision": base.CANDIDATE,
|
|
"candidateSha256": base.CANDIDATE_SHA, "receiptOrigin": base.ORIGIN,
|
|
"receiptDigestBefore": "a" * 64}).encode()
|
|
monkeypatch.setattr(helper, "_load_base", lambda: base)
|
|
monkeypatch.setattr(module, "ControllerDriver", lambda projects: driver)
|
|
bundle = tmp_path / "bundle"
|
|
bundle.mkdir()
|
|
monkeypatch.setattr(helper, "BUNDLE", bundle)
|
|
monkeypatch.setattr(helper, "__file__", str(bundle / "recover_ci_api_term_once.py"))
|
|
monkeypatch.setattr(helper, "os", SimpleNamespace(name="posix", geteuid=lambda: 0))
|
|
monkeypatch.setattr(helper, "sys", SimpleNamespace(flags=SimpleNamespace(isolated=1), dont_write_bytecode=True))
|
|
monkeypatch.setattr(helper, "stat", SimpleNamespace(S_IMODE=lambda mode: 0o700))
|
|
claims, effects = [], []
|
|
monkeypatch.setattr(helper, "_claim", lambda *args: claims.append(args))
|
|
def probe(driver, module, cid, ticket=None):
|
|
if ticket is None:
|
|
return {"pid": 11, "parent": 1, "start": 123, "initStart": 10, "namespace": "pid:[111]"}
|
|
effects.append(cid)
|
|
monkeypatch.setattr(helper, "_probe", probe)
|
|
monkeypatch.setattr(module.fleet, "_migration_lock", lambda **kwargs: nullcontext())
|
|
return driver, holder, base, claims, effects
|
|
|
|
|
|
def test_readonly_default_never_claims_or_signals(helper, prepared):
|
|
driver, holder, _, claims, effects = prepared
|
|
receipt = copy.deepcopy(holder["value"])
|
|
result = helper.run()
|
|
assert result["status"] == "ready" and result["effectsPerformed"] is False
|
|
assert result["existingMaintenanceEffects"] is True and result["normalDeployment"] is False
|
|
assert not claims and not effects and holder["value"] == receipt
|
|
assert not driver.effects
|
|
|
|
|
|
def test_three_exact_children_only_and_original_receipt_unchanged(helper, prepared):
|
|
_, holder, _, claims, effects = prepared
|
|
before = copy.deepcopy(holder["value"])
|
|
result = helper.run(execute=True, confirmation=helper.CONFIRMATION)
|
|
assert result["status"] == "term_sent_inspect_exits" and result["resolvedChildren"] == 3
|
|
assert result["confirmedSignalAcks"] == 3
|
|
assert len(claims) == 1 and effects == list(helper.API_IDS)
|
|
assert holder["value"] == before and result["retryAllowed"] is False
|
|
|
|
|
|
@pytest.mark.parametrize("fault", ["missing", "foreign", "complete", "missing-intent", "foreign-intent",
|
|
"queue", "busy-fleet", "claimed", "container", "previous-marker"])
|
|
def test_failed_admission_never_signals(helper, prepared, module, monkeypatch, fault):
|
|
from contextlib import contextmanager
|
|
driver, holder, base, claims, effects = prepared
|
|
if fault == "missing":
|
|
holder["value"] = None
|
|
elif fault == "foreign":
|
|
holder["value"]["revision"] = "f" * 40
|
|
elif fault == "complete":
|
|
holder["value"]["phase"] = "complete"
|
|
elif fault == "missing-intent":
|
|
holder["value"]["progress"].pop(helper.API_IDS[0])
|
|
elif fault == "foreign-intent":
|
|
holder["value"]["progress"][helper.API_IDS[0]] = "restart_intent"
|
|
elif fault == "queue":
|
|
driver.pending_jobs = lambda: [{"status": "queued"}]
|
|
elif fault == "busy-fleet":
|
|
@contextmanager
|
|
def busy(**kwargs):
|
|
raise RuntimeError("busy")
|
|
yield
|
|
monkeypatch.setattr(module.fleet, "_migration_lock", busy)
|
|
elif fault == "claimed":
|
|
monkeypatch.setattr(helper, "_claim", lambda *args: (_ for _ in ()).throw(RuntimeError("exists")))
|
|
elif fault == "container":
|
|
driver.state["rows"][0]["imageId"] = "foreign"
|
|
else:
|
|
base._read = lambda *args, **kwargs: b"{}"
|
|
result = helper.run(execute=True, confirmation=helper.CONFIRMATION)
|
|
assert result["status"] == "blocked" and not effects and not claims
|
|
|
|
|
|
@pytest.mark.parametrize("exit_attested", [True, False])
|
|
def test_missing_ack_advances_only_on_exact_owned_exit(helper, prepared, monkeypatch, exit_attested):
|
|
driver, holder, _, claims, effects = prepared
|
|
before = copy.deepcopy(holder["value"])
|
|
normal = helper._probe
|
|
def probe(driver, module, cid, ticket=None):
|
|
if ticket is not None and cid == helper.API_IDS[0]:
|
|
effects.append(cid)
|
|
if exit_attested:
|
|
row = next(row for row in driver.state["rows"] if row["id"] == cid)
|
|
row["state"].update(Running=False, Status="exited", Pid=0, FinishedAt="2026-10-10T00:00:00Z")
|
|
raise RuntimeError("namespace teardown, token-secret-not-output")
|
|
return normal(driver, module, cid, ticket)
|
|
monkeypatch.setattr(helper, "_probe", probe)
|
|
result = helper.run(execute=True, confirmation=helper.CONFIRMATION)
|
|
assert len(claims) == 1 and effects.count(helper.API_IDS[0]) == 1
|
|
assert result["status"] == ("term_sent_inspect_exits" if exit_attested else "unknown")
|
|
assert len(effects) == (3 if exit_attested else 1)
|
|
assert "token-secret" not in json.dumps(result) and holder["value"] == before
|
|
|
|
|
|
@pytest.mark.skipif(os.name != "posix" or os.geteuid() != 0, reason="native root durable marker")
|
|
@pytest.mark.parametrize("fault", [None, "before-link", "after-link"])
|
|
def test_actual_atomic_marker_once_under_write_failures(helper, monkeypatch, tmp_path, fault):
|
|
marker = tmp_path / "private" / "once.json"
|
|
marker.parent.mkdir(mode=0o700)
|
|
monkeypatch.setattr(helper, "MARKER", marker)
|
|
base = SimpleNamespace(_trusted_parents=lambda path: None,
|
|
ORIGIN="0ea6f267c28d51ac7113e6d943c89857d3a189ab",
|
|
CANDIDATE_SHA="a" * 64,
|
|
_read=lambda *args, **kwargs: b"reviewed source")
|
|
calls = 0
|
|
def fsync(fd):
|
|
nonlocal calls
|
|
calls += 1
|
|
if (fault == "before-link" and calls == 1) or (fault == "after-link" and calls == 2):
|
|
raise OSError("injected actual fsync failure")
|
|
os.fsync(fd)
|
|
proxy = SimpleNamespace(**{name: getattr(os, name) for name in dir(os)})
|
|
proxy.fsync = fsync
|
|
monkeypatch.setattr(helper, "os", proxy)
|
|
if fault:
|
|
with pytest.raises(OSError):
|
|
helper._claim(base, "b" * 64, {})
|
|
else:
|
|
helper._claim(base, "b" * 64, {})
|
|
assert marker.exists() is (fault != "before-link")
|
|
assert not list(marker.parent.glob(".api-term-invocation-*"))
|
|
if marker.exists():
|
|
payload = json.loads(marker.read_bytes())
|
|
assert payload["receiptOrigin"] == base.ORIGIN and payload["kind"] == "ci_api_child_term_once"
|
|
for digest in ("b" * 64, "c" * 64):
|
|
with pytest.raises(RuntimeError, match="already_invoked"):
|
|
helper._claim(base, digest, {})
|
|
|
|
|
|
def test_explicit_confirmation_required(helper, prepared):
|
|
with pytest.raises(RuntimeError, match="separate_fixed_confirmation_required"):
|
|
helper.run(execute=True)
|
|
assert not prepared[-1] and not prepared[-2]
|
|
|
|
|
|
@pytest.mark.parametrize("fault", [None, "argv", "parent", "extra-child", "namespace", "exe", "pid-reuse", "esrch", "shell-term"])
|
|
def test_actual_child_script_pidfd_and_process_races(helper, monkeypatch, capsys, fault):
|
|
signals, closes = [], []
|
|
start = 123
|
|
def procstat(parent, tick):
|
|
return "11 (python) S " + str(parent) + " " + " ".join(["0"] * 17) + " " + str(tick)
|
|
class FakePath:
|
|
def __init__(self, path):
|
|
self.path = path
|
|
self.name = path.rsplit("/", 1)[-1]
|
|
def iterdir(self):
|
|
return [FakePath("/proc/" + str(pid)) for pid in ([1, 11, 99, 12] if fault == "extra-child" else [1, 11, 99])]
|
|
def read_bytes(self):
|
|
args = (["sh", "-c", helper.SHELL_COMMAND] if self.path == "/proc/1/cmdline" else
|
|
["foreign"] if fault == "argv" else helper.CHILD_ARGV)
|
|
return ("\0".join(args) + "\0").encode()
|
|
def read_text(self):
|
|
if self.path.endswith("/status"):
|
|
return "SigCgt:\t00014002" if fault == "shell-term" else "SigCgt:\t00010002"
|
|
if self.path == "/proc/1/stat":
|
|
return procstat(0, 10)
|
|
return procstat(0 if self.path == "/proc/99/stat" else 2 if fault == "parent" else 1, start)
|
|
def pidfd_open(pid, flags):
|
|
nonlocal start
|
|
assert pid == 11 and flags == 0
|
|
if fault == "pid-reuse":
|
|
start += 1
|
|
if fault == "esrch":
|
|
raise ProcessLookupError()
|
|
return 77
|
|
def readlink(path):
|
|
if path.endswith("/exe"):
|
|
return "/foreign" if fault == "exe" else "/usr/local/bin/python3.12"
|
|
return "pid:[222]" if fault == "namespace" and "/11/" in path else "pid:[111]"
|
|
fake_os = SimpleNamespace(geteuid=lambda: 0, getpid=lambda: 99, pidfd_open=pidfd_open,
|
|
readlink=readlink, close=lambda fd: closes.append(fd))
|
|
fake_signal = SimpleNamespace(SIGTERM=15, pidfd_send_signal=lambda fd, sig, info, flags:
|
|
signals.append((fd, sig, info, flags)))
|
|
ticket = {"pid": 11, "parent": 1, "start": 123, "initStart": 10, "namespace": "pid:[111]"}
|
|
fake_sys = SimpleNamespace(flags=SimpleNamespace(isolated=1), dont_write_bytecode=True,
|
|
argv=["-c", json.dumps(ticket)], exit=lambda code: (_ for _ in ()).throw(SystemExit(code)))
|
|
actual_import = builtins.__import__
|
|
modules = {"os": fake_os, "signal": fake_signal, "sys": fake_sys,
|
|
"pathlib": SimpleNamespace(Path=FakePath)}
|
|
namespace = {"__builtins__": {**vars(builtins), "__import__": lambda name, *args, **kwargs:
|
|
modules.get(name) or actual_import(name, *args, **kwargs)}}
|
|
if fault is None:
|
|
exec(helper.CHILD_SOURCE, namespace)
|
|
assert signals == [(77, 15, None, 0)] and closes == [77]
|
|
assert json.loads(capsys.readouterr().out) == {"status": "term_sent"}
|
|
else:
|
|
with pytest.raises(SystemExit):
|
|
exec(helper.CHILD_SOURCE, namespace)
|
|
assert not signals
|
|
assert json.loads(capsys.readouterr().out) == {"status": "rejected"}
|
|
|
|
|
|
def test_original_c185_pin_and_no_stronger_signal_or_host_pid(helper):
|
|
original = (ROOT / "scripts/recover_ci_acceptance_once.py").read_bytes().replace(b"\r\n", b"\n")
|
|
assert hashlib.sha256(original).hexdigest() == helper.BASE_SHA
|
|
assert len(set(helper.API_IDS)) == 3
|
|
assert "os.kill" not in helper.CHILD_SOURCE and "SIGKILL" not in helper.CHILD_SOURCE
|
|
assert "pidfd_send_signal(fd,signal.SIGTERM,None,0)" in helper.CHILD_SOURCE
|