#!/usr/bin/env python3 """Exercise private bootstrap verification without network, root, or a VM.""" import json import os from pathlib import Path import shlex import shutil import subprocess import tempfile import unittest ROOT = Path(__file__).resolve().parent.parent BOOTSTRAP = ROOT / "scripts/micromanager/dev-container-bootstrap.sh" HELPERS = ROOT / "scripts/firecrab-release.sh" INSTALLER = r'''#!/bin/bash set -euo pipefail [ "$#" -eq 2 ] [ -f "$TEST_ROOT/dnsmasq-masked" ] printf '{"command":"installer","args":["%s","%s"]}\n' "$1" "$2" >> "$TEST_ROOT/events" [ "${INSTALL_FAIL:-0}" = 0 ] || exit 47 printf 'verified release installed\n' > "$TEST_ROOT/installed-api" chmod 0755 "$TEST_ROOT/installed-api" ''' COMMAND = r'''#!/usr/bin/env python3 import hashlib, json, os, pathlib, sys root = pathlib.Path(os.environ['TEST_ROOT']) name = pathlib.Path(sys.argv[0]).name args = sys.argv[1:] with (root / 'events').open('a') as events: events.write(json.dumps({'command': name, 'args': args}) + '\n') if name == 'systemctl': if args == ['is-system-running']: state = os.environ.get('SYSTEM_STATE', 'running') print(state) sys.exit(0 if state == 'running' else 1) if args == ['mask', '--now', 'dnsmasq.service']: if os.environ.get('MASK_FAIL') == '1': sys.exit(1) (root / 'dnsmasq-masked').touch() sys.exit(0) if args == ['cat', 'firecrab-api']: assert (root / 'dnsmasq-masked').is_file(), 'service checked before dnsmasq mask' sys.exit(0 if os.environ.get('HAS_UNIT') == '1' else 1) if args == ['start', 'firecrab-dev-publish.service']: assert (root / 'installed-api').is_file(), 'relay started before installation' sys.exit(0) raise AssertionError(('unexpected systemctl', args)) assert name == 'curl', name assert (root / 'dnsmasq-masked').is_file(), 'release downloaded before dnsmasq mask' assert args[:4] == ['--proto', '=https', '--tlsv1.2', '-fsSL'], args assert args[-2] == '-o', args url, output = args[-3], pathlib.Path(args[-1]) assert output.is_relative_to(root), output prefix = 'https://github.com/SteelCrab/firecrab/releases/latest/download/' assert url.startswith(prefix), url asset = url[len(prefix):] if os.environ.get('DOWNLOAD_FAIL') == asset: sys.exit(22) installer = (root / 'release-installer').read_bytes() if asset == 'install.sh': output.write_bytes(installer) elif asset == 'SHA256SUMS': digest = hashlib.sha256(installer).hexdigest() mode = os.environ.get('CHECKSUM_MODE', 'correct') if mode == 'mismatch': digest = '0' * 64 filename = 'unrelated.sh' if mode == 'missing' else 'install.sh' output.write_text(f'{digest} ./{filename}\n') else: raise AssertionError(('unexpected asset', asset)) ''' class PrivateBootstrapTest(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="firecrab-bootstrap-test-") self.addCleanup(temporary.cleanup) self.root = Path(temporary.name).resolve() self.commands = self.root / "commands" self.commands.mkdir() self.downloads = self.root / "downloads" self.downloads.mkdir() for name in ["curl", "systemctl"]: command = self.commands / name command.write_text(COMMAND) command.chmod(0o755) self.installed = self.root / "installed-api" self.helper = self.root / "release-helpers.sh" shutil.copyfile(HELPERS, self.helper) (self.root / "release-installer").write_text(INSTALLER) script = BOOTSTRAP.read_text() script = script.replace("/dev/kvm", "/dev/null") script = script.replace("/usr/local/lib/firecrab/firecrab-api", shlex.quote(str(self.installed))) script = script.replace("/opt/firecrab-bootstrap/scripts/firecrab-release.sh", shlex.quote(str(self.helper))) script = script.replace("seq 1 60", "seq 1 2").replace("sleep 1", "sleep 0.01") self.script = self.root / "bootstrap.sh" self.script.write_text(script) def bootstrap(self, **flags): (self.root / "events").unlink(missing_ok=True) (self.root / "dnsmasq-masked").unlink(missing_ok=True) environment = dict(os.environ) environment.update( TEST_ROOT=str(self.root), TMPDIR=str(self.downloads), PATH=str(self.commands) + os.pathsep + environment["PATH"], FIRECRAB_RELEASE_BASE="https://github.com/SteelCrab/firecrab/releases", CHECKSUM_MODE="correct", HAS_UNIT="0", SYSTEM_STATE="running", INSTALL_FAIL="0", DOWNLOAD_FAIL="", MASK_FAIL="0", ) environment.update({name: str(value) for name, value in flags.items()}) return subprocess.run(["bash", str(self.script)], env=environment, text=True, capture_output=True, timeout=5) def events(self): path = self.root / "events" return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] def mutations(self): return [event for event in self.events() if event["command"] == "installer" or (event["command"] == "systemctl" and event["args"][0] in ["mask", "start"])] def mask_event(self): return {"command": "systemctl", "args": ["mask", "--now", "dnsmasq.service"]} def assert_clean_downloads(self): self.assertEqual(list(self.downloads.iterdir()), []) def test_verified_release_installer_runs_before_its_relay(self): result = self.bootstrap() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(self.mutations(), [ self.mask_event(), {"command": "installer", "args": ["--libc", "gnu"]}, {"command": "systemctl", "args": ["start", "firecrab-dev-publish.service"]}, ]) self.assertEqual(len([event for event in self.events() if event["command"] == "curl"]), 2) self.assert_clean_downloads() def test_bad_or_missing_checksum_never_executes_the_installer(self): for mode in ["mismatch", "missing"]: with self.subTest(mode=mode): result = self.bootstrap(CHECKSUM_MODE=mode) self.assertNotEqual(result.returncode, 0) self.assertIn("checksum mismatch", result.stderr) self.assertEqual(self.mutations(), [self.mask_event()]) self.assertFalse(self.installed.exists()) self.assert_clean_downloads() def test_either_download_failure_stops_before_installation(self): for asset in ["install.sh", "SHA256SUMS"]: with self.subTest(asset=asset): result = self.bootstrap(DOWNLOAD_FAIL=asset) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.mutations(), [self.mask_event()]) self.assertFalse(self.installed.exists()) self.assert_clean_downloads() def test_existing_complete_installation_skips_downloads(self): self.installed.write_text("keep existing installed binary\n") self.installed.chmod(0o755) result = self.bootstrap(HAS_UNIT=1) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertFalse(any(event["command"] in ["curl", "installer"] for event in self.events())) self.assertEqual(self.installed.read_text(), "keep existing installed binary\n") self.assertEqual(self.mutations(), [ self.mask_event(), {"command": "systemctl", "args": ["start", "firecrab-dev-publish.service"]}, ]) def test_missing_installed_binary_repairs_only_the_private_installation(self): result = self.bootstrap(HAS_UNIT=1) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(self.installed.read_text(), "verified release installed\n") self.assertEqual(len(self.mutations()), 3) def test_installer_failure_never_starts_the_relay(self): result = self.bootstrap(INSTALL_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.mutations(), [ self.mask_event(), {"command": "installer", "args": ["--libc", "gnu"]}, ]) self.assertFalse(self.installed.exists()) self.assert_clean_downloads() def test_unready_systemd_stops_before_downloads_or_mutations(self): result = self.bootstrap(SYSTEM_STATE="starting") self.assertNotEqual(result.returncode, 0) self.assertIn("private systemd did not become ready", result.stderr) self.assertTrue(all(event["command"] == "systemctl" and event["args"] == ["is-system-running"] for event in self.events())) self.assertEqual(self.mutations(), []) self.assert_clean_downloads() def test_degraded_systemd_is_a_usable_state(self): result = self.bootstrap(SYSTEM_STATE="degraded") self.assertEqual(result.returncode, 0, result.stdout + result.stderr) self.assertEqual(len(self.mutations()), 3) def test_dnsmasq_mask_failure_stops_before_installation_or_relay(self): result = self.bootstrap(MASK_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertEqual(self.mutations(), [self.mask_event()]) self.assertFalse(any(event["command"] in ["curl", "installer"] for event in self.events())) self.assertFalse(self.installed.exists()) self.assert_clean_downloads() if __name__ == "__main__": unittest.main()