372 lines
16 KiB
Python
372 lines
16 KiB
Python
import hashlib
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = ROOT / "scripts" / "upgrade-knot.sh"
|
|
BASH = shutil.which("bash") or "/run/current-system/sw/bin/bash"
|
|
|
|
|
|
DISPATCHER = r'''#!/usr/bin/env python3
|
|
import hashlib
|
|
import os
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
name = Path(sys.argv[0]).name
|
|
args = sys.argv[1:]
|
|
log = Path(os.environ["FAKE_LOG"])
|
|
with log.open("a") as f:
|
|
f.write(name + " " + " ".join(args) + "\n")
|
|
if os.environ.get("KNOT_MASTER_KEY") and name not in {"knot-migrate", "knot-server"}:
|
|
Path(os.environ["KEY_LEAK_LOG"]).write_text(name + "\n")
|
|
|
|
def fail(flag):
|
|
return os.environ.get(flag) == "1"
|
|
|
|
if name == "id":
|
|
print("0")
|
|
elif name == "hostname":
|
|
print("asusmininix")
|
|
elif name == "flock":
|
|
pass
|
|
elif name == "systemctl":
|
|
if args[:2] == ["is-active", "--quiet"]:
|
|
unit = args[2]
|
|
if unit == "auto-update.service":
|
|
raise SystemExit(0 if fail("AUTO_UPDATE_ACTIVE") else 1)
|
|
if unit == "knot-rs.service" and fail("RUST_INACTIVE"):
|
|
raise SystemExit(1)
|
|
raise SystemExit(0)
|
|
if args[:1] == ["start"] and "auto-update.timer" in args and fail("TIMER_START_FAIL"):
|
|
raise SystemExit(1)
|
|
elif name == "nix-store":
|
|
if args[:1] == ["--check-validity"]:
|
|
raise SystemExit(1 if fail("INVALID_STORE") else 0)
|
|
if args[:1] == ["--import"]:
|
|
sys.stdin.buffer.read()
|
|
elif name == "sha256sum":
|
|
data = Path(args[0]).read_bytes()
|
|
print(hashlib.sha256(data).hexdigest() + " " + args[0])
|
|
elif name == "gzip":
|
|
try:
|
|
os.write(1, b"fixture export")
|
|
except OSError:
|
|
pass
|
|
elif name == "grep":
|
|
path = Path(args[-1])
|
|
if not path.is_file() or args[1] not in path.read_text():
|
|
raise SystemExit(1)
|
|
if fail("LIVE_GATE_FAIL") and path.name in {"auto-update.timer", "auto-update.service"}:
|
|
raise SystemExit(1)
|
|
elif name == "git":
|
|
if fail("GIT_PREFLIGHT_FAIL"):
|
|
raise SystemExit(1)
|
|
elif name == "curl":
|
|
if "--write-out" in args:
|
|
print("101")
|
|
elif "requestCrawl" in " ".join(args) and fail("CRAWL_FAIL"):
|
|
raise SystemExit(1)
|
|
elif "5555" in " ".join(args) and fail("HEALTH_FAIL"):
|
|
print("{}")
|
|
else:
|
|
print("{}")
|
|
elif name == "jq":
|
|
sys.stdin.read()
|
|
if fail("HEALTH_FAIL") and "version" in " ".join(args):
|
|
raise SystemExit(1)
|
|
elif name == "ssh-keygen":
|
|
try:
|
|
os.write(1, b"256 TESTKEY fixture\n")
|
|
except OSError:
|
|
pass
|
|
os._exit(0)
|
|
elif name == "ssh-keyscan":
|
|
try:
|
|
os.write(1, b"127.0.0.1 ssh-ed25519 TESTKEY\n")
|
|
except OSError:
|
|
pass
|
|
os._exit(0)
|
|
elif name == "nix-env":
|
|
pass
|
|
elif name == "agenix":
|
|
rules = os.environ.get("AGENIX_RULES", "")
|
|
if not Path(rules).is_file() or args[:2] != ["-d", "knot-env.age"]:
|
|
raise SystemExit("agenix needs explicit rules and a rules-relative secret name")
|
|
value = (b"KNOT_MASTER_KEY=" + (b"B" * 43 + b"=\n" if not fail("BAD_KEY") else b"bad\n"))
|
|
sys.stdout.buffer.write(value)
|
|
target = os.environ.get("AGENIX_TARGET")
|
|
if target:
|
|
Path(target).write_bytes(value)
|
|
elif name == "knot-migrate":
|
|
if os.environ.get("KNOT_MASTER_KEY") != "B" * 43 + "=":
|
|
raise SystemExit(1)
|
|
if "--dry-run" in args:
|
|
raise SystemExit(1 if fail("DRYRUN_FAIL") else 0)
|
|
target = Path(args[args.index("--target") + 1])
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
(target / "ssh_host_key").write_text("fixture-key\n")
|
|
(target / "sealed-keys").write_text("fixture-sealed\n")
|
|
(target / "repo-signing-keys.json").write_text("{}\n")
|
|
if fail("MIGRATION_FAIL"):
|
|
raise SystemExit(1)
|
|
elif name == "sqlite3" or name == "tar" or name == "chown" or name == "chmod":
|
|
pass
|
|
elif name == "cmp":
|
|
if Path(args[-2]).read_bytes() != Path(args[-1]).read_bytes():
|
|
raise SystemExit(1)
|
|
elif name == "mktemp":
|
|
template = args[-1]
|
|
path = Path(template.replace("XXXXXX", "fixture"))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if "-d" in args:
|
|
path.mkdir(exist_ok=True)
|
|
else:
|
|
path.touch()
|
|
print(path)
|
|
elif name == "readlink":
|
|
print(os.environ["OLD_SYSTEM"])
|
|
elif name == "rm":
|
|
for arg in args:
|
|
if arg.startswith("-"):
|
|
continue
|
|
p = Path(arg)
|
|
if p.is_dir():
|
|
shutil.rmtree(p, ignore_errors=True)
|
|
else:
|
|
p.unlink(missing_ok=True)
|
|
elif name == "rmdir":
|
|
Path(args[-1]).rmdir()
|
|
elif name == "mv":
|
|
shutil.move(args[-2], args[-1])
|
|
elif name == "touch":
|
|
Path(args[-1]).touch()
|
|
else:
|
|
Path(os.environ["UNKNOWN_LOG"]).open("a").write(name + "\n")
|
|
raise SystemExit(127)
|
|
'''
|
|
|
|
|
|
class UpgradeKnotTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = Path(tempfile.mkdtemp(prefix="upgrade-knot-test-"))
|
|
self.fakebin = self.tmp / "bin"
|
|
self.fakebin.mkdir()
|
|
self.log = self.tmp / "commands.log"
|
|
self.unknown = self.tmp / "unknown.log"
|
|
self.key_scope = self.tmp / "key-scope.log"
|
|
dispatcher = self.fakebin / "fake-command.py"
|
|
dispatcher.write_text(DISPATCHER)
|
|
dispatcher.chmod(dispatcher.stat().st_mode | stat.S_IXUSR)
|
|
names = "id hostname flock systemctl nix-store sha256sum gzip grep git curl jq ssh-keygen ssh-keyscan nix-env agenix knot-migrate sqlite3 tar chown chmod cmp mktemp readlink rm rmdir mv touch".split()
|
|
for name in names:
|
|
(self.fakebin / name).symlink_to(dispatcher)
|
|
|
|
self.stage = self.tmp / "stage"
|
|
self.stage.mkdir()
|
|
self.store = self.tmp / "nix" / "store"
|
|
self.system = self.store / "fixture-nixos-system-asusmininix-test"
|
|
self.tools = self.store / "fixture-knot-migration-tools"
|
|
(self.system / "bin").mkdir(parents=True)
|
|
(self.system / "sw" / "bin").mkdir(parents=True)
|
|
(self.system / "etc" / "systemd" / "system").mkdir(parents=True)
|
|
(self.system / "etc" / "knot").mkdir()
|
|
(self.system / "sw" / "bin" / "knot-server").write_text(
|
|
f"#!/bin/sh\ntest \"$KNOT_MASTER_KEY\" = \"{'B' * 43}=\" || exit 1\nprintf validated > {self.key_scope}\n"
|
|
)
|
|
(self.system / "sw" / "bin" / "knot-server").chmod(0o755)
|
|
switch = self.system / "bin" / "switch-to-configuration"
|
|
switch.write_text("#!/bin/sh\nif [ \"${FAIL_ACTIVATION:-0}\" = 1 ]; then exit 1; fi\nexit 0\n")
|
|
switch.chmod(0o755)
|
|
gate = f"ConditionPathExists={self.state if hasattr(self, 'state') else self.tmp / 'state'}/.upgrade-verified\n"
|
|
(self.system / "etc" / "systemd" / "system" / "auto-update.timer").write_text(gate)
|
|
(self.system / "etc" / "systemd" / "system" / "auto-update.service").write_text(gate)
|
|
(self.tools / "bin").mkdir(parents=True)
|
|
(self.stage / "system-path").write_text(str(self.system) + "\n")
|
|
(self.stage / "tools-path").write_text(str(self.tools) + "\n")
|
|
export = self.stage / "upgrade.export.gz"
|
|
export.write_bytes(b"fixture export")
|
|
(self.stage / "upgrade.export.gz.sha256").write_text(hashlib.sha256(export.read_bytes()).hexdigest() + " upgrade.export.gz\n")
|
|
|
|
self.repo = self.tmp / "repo"
|
|
(self.repo / "secrets").mkdir(parents=True)
|
|
(self.repo / "secrets" / "knot-env.age").write_text("not a secret used by fixture\n")
|
|
(self.repo / "secrets" / "secrets.nix").write_text("{}\n")
|
|
self.source = self.tmp / "git"
|
|
self.source.mkdir()
|
|
(self.source / "knotserver.db").write_text("fixture db\n")
|
|
self.ssh = self.tmp / "ssh_host_ed25519_key"
|
|
self.ssh.write_text("fixture private key\n")
|
|
(self.ssh.parent / (self.ssh.name + ".pub")).write_text("fixture public key\n")
|
|
self.state = self.tmp / "state"
|
|
self.agenix_target = self.tmp / "agenix" / "knot-env"
|
|
self.agenix_target.parent.mkdir()
|
|
self.agenix_target.write_text("KNOT_MASTER_KEY=" + "B" * 43 + "=\n")
|
|
self.old = self.store / "old-system"
|
|
(self.old / "bin").mkdir(parents=True)
|
|
(self.old / "bin" / "switch-to-configuration").write_text("#!/bin/sh\nexit 0\n")
|
|
(self.old / "bin" / "switch-to-configuration").chmod(0o755)
|
|
(self.old / "etc" / "systemd" / "system").mkdir(parents=True)
|
|
(self.old / "etc" / "systemd" / "system" / "auto-update.timer").write_text(
|
|
f"ConditionPathExists={self.state}/.upgrade-verified\n"
|
|
)
|
|
(self.old / "etc" / "systemd" / "system" / "auto-update.service").write_text(
|
|
f"ConditionPathExists={self.state}/.upgrade-verified\n"
|
|
)
|
|
current = self.tmp / "current-system"
|
|
current.symlink_to(self.old)
|
|
|
|
replacements = {
|
|
"/nix/store": self.store,
|
|
"/home/anish/tangled-v1.20.0-preflight": self.stage,
|
|
"/etc/commonscomputing-nix": self.repo,
|
|
"/var/lib/knot": self.state,
|
|
"/home/git": self.source,
|
|
"/var/backups": self.tmp / "backups",
|
|
"/nix/var/nix/profiles/system": self.tmp / "profile",
|
|
"/run/current-system": current,
|
|
"/etc/ssh/ssh_host_ed25519_key": self.ssh,
|
|
"/run/lock/commonscomputing-update.lock": self.tmp / "lock",
|
|
"/run/knot-upgrade.XXXXXX": self.tmp / "runtime.XXXXXX",
|
|
"/var/lib/knot-upgrade-system": self.tmp / "system-root",
|
|
"/var/lib/knot-upgrade-tools": self.tmp / "tools-root",
|
|
"/run/agenix/knot-env": self.agenix_target,
|
|
}
|
|
text = SCRIPT.read_text()
|
|
for old, new in replacements.items():
|
|
text = text.replace(old, str(new))
|
|
self.script = self.tmp / "upgrade-knot.sh"
|
|
self.script.write_text(text)
|
|
self.env = os.environ.copy()
|
|
self.key_leak = self.tmp / "key-leak.log"
|
|
self.env.update({"PATH": str(self.fakebin) + ":" + self.env["PATH"], "FAKE_LOG": str(self.log), "UNKNOWN_LOG": str(self.unknown), "KEY_LEAK_LOG": str(self.key_leak), "OLD_SYSTEM": str(self.old), "AGENIX_TARGET": str(self.agenix_target)})
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
def run_upgrade(self, *args, **flags):
|
|
env = self.env.copy()
|
|
env.update({k: str(v) for k, v in flags.items()})
|
|
return subprocess.run([BASH, str(self.script), *args], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env)
|
|
|
|
def commands(self):
|
|
return self.log.read_text().splitlines() if self.log.exists() else []
|
|
|
|
def assert_no_unknown_tools(self):
|
|
self.assertFalse(self.unknown.exists(), self.unknown.read_text() if self.unknown.exists() else "")
|
|
|
|
def test_happy_path_order_and_secret_cleanup(self):
|
|
result = self.run_upgrade("--yes")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
commands = self.commands()
|
|
self.assertLess(next(i for i, x in enumerate(commands) if x.startswith("systemctl stop auto-update.timer")), next(i for i, x in enumerate(commands) if x.startswith("knot-migrate ") and "--dry-run" not in x))
|
|
timer_start = next(i for i, x in enumerate(commands) if x == "systemctl start auto-update.timer")
|
|
health_end = max(i for i, x in enumerate(commands) if x.startswith("curl ") and i < timer_start)
|
|
self.assertGreater(timer_start, health_end)
|
|
self.assertTrue((self.state / ".migration-complete").exists())
|
|
self.assertFalse(self.tmp.joinpath("runtime.fixture", "knot.env").exists())
|
|
self.assertFalse(self.key_leak.exists())
|
|
self.assertTrue(self.key_scope.exists())
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_dryrun_failure_does_not_stop_go_or_timer(self):
|
|
result = self.run_upgrade("--yes", DRYRUN_FAIL=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
commands = self.commands()
|
|
self.assertFalse(any(x == "systemctl stop knot.service" for x in commands))
|
|
self.assertFalse(any(x == "systemctl stop auto-update.timer" for x in commands))
|
|
self.assertFalse(self.tmp.joinpath("runtime.fixture", "knot.env").exists())
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_migration_failure_quarantines_target_and_restarts_go(self):
|
|
result = self.run_upgrade("--yes", MIGRATION_FAIL=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertTrue((self.tmp / "backups" / "knot-cutover.fixture" / "failed-target").exists())
|
|
self.assertFalse(self.tmp.joinpath("runtime.fixture", "knot.env").exists())
|
|
commands = self.commands()
|
|
self.assertIn("systemctl start knot.service", commands)
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_activation_failure_keeps_new_state_and_stops_services(self):
|
|
result = self.run_upgrade("--yes", FAIL_ACTIVATION=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
commands = self.commands()
|
|
sets = [x for x in commands if x.startswith("nix-env --profile")]
|
|
self.assertEqual(len(sets), 1)
|
|
self.assertFalse(any(x == "systemctl start knot.service" for x in commands))
|
|
self.assertIn("systemctl stop auto-update.timer", commands)
|
|
self.assertTrue(self.state.exists())
|
|
self.assertFalse((self.state / ".migration-complete").exists())
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_health_failure_keeps_new_state_and_stops_services(self):
|
|
result = self.run_upgrade("--yes", HEALTH_FAIL=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
commands = self.commands()
|
|
self.assertEqual(sum(x.startswith("nix-env --profile") for x in commands), 1)
|
|
self.assertNotIn("systemctl start knot.service", commands)
|
|
self.assertIn("systemctl stop knot-rs.service", commands)
|
|
self.assertTrue(self.state.exists())
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_completed_marker_skips_migration(self):
|
|
self.state.mkdir()
|
|
(self.state / ".migration-complete").touch()
|
|
result = self.run_upgrade()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertFalse(any(x.startswith("knot-migrate") for x in self.commands()))
|
|
self.assertTrue(any(x == "systemctl start auto-update.timer" for x in self.commands()))
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_completed_marker_rejects_mismatched_live_gate_without_touching_state(self):
|
|
self.state.mkdir()
|
|
(self.state / ".migration-complete").touch()
|
|
before = (self.state / ".upgrade-verified").exists()
|
|
result = self.run_upgrade(LIVE_GATE_FAIL=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual((self.state / ".upgrade-verified").exists(), before)
|
|
self.assertFalse(any(x == "systemctl start auto-update.timer" for x in self.commands()))
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_invalid_key_fails_before_downtime(self):
|
|
result = self.run_upgrade("--yes", BAD_KEY=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
commands = self.commands()
|
|
self.assertFalse(any(x == "systemctl stop knot.service" for x in commands))
|
|
self.assertFalse(any(x == "systemctl stop auto-update.timer" for x in commands))
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_missing_staged_service_gate_fails_before_downtime(self):
|
|
service = self.system / "etc" / "systemd" / "system" / "auto-update.service"
|
|
service.write_text("[Unit]\n")
|
|
result = self.run_upgrade("--yes")
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertFalse(any(x == "systemctl stop knot.service" for x in self.commands()))
|
|
self.assertFalse(self.state.exists())
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_git_preflight_failure_has_no_downtime(self):
|
|
result = self.run_upgrade("--yes", GIT_PREFLIGHT_FAIL=1)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
commands = self.commands()
|
|
self.assertFalse(any(x == "systemctl stop knot.service" for x in commands))
|
|
self.assertFalse(any(x == "systemctl stop auto-update.timer" for x in commands))
|
|
self.assert_no_unknown_tools()
|
|
|
|
def test_invalid_store_imports_export(self):
|
|
result = self.run_upgrade("--yes", INVALID_STORE=1)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue(any(x.startswith("nix-store --import") for x in self.commands()))
|
|
self.assert_no_unknown_tools()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|