tests/test_bootstrap_chain.py
back to source
"""The boot chain's fetch/verify/exec steps - spec section 7, steps 3, 4 and 6.
These were specified from the start and left unimplemented. Until they exist,
bootstrap.sh and numinit.sh are two scripts that never connect to each other.
Every test here runs the real shell against a real local HTTP server; nothing
is mocked at the shell boundary, because the failure modes being guarded
against (a fetch that half-succeeds, a hash that is never checked, an exec
that silently does not happen) all live exactly at that boundary.
"""
import hashlib
import http.server
import os
import shutil
import subprocess
import tempfile
import threading
import unittest
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
BOOTSTRAP = os.path.join(ROOT, "boot", "bootstrap.sh").replace("\\", "/")
LIB = os.path.join(ROOT, "boot", "lib").replace("\\", "/")
def sha256(data):
return hashlib.sha256(data).hexdigest()
class Server:
"""A throwaway static file server rooted at a temp directory."""
def __init__(self, root):
self.root = root
handler = self._handler(root)
self.httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
self.port = self.httpd.server_address[1]
self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
self.thread.start()
@staticmethod
def _handler(root):
class Handler(http.server.SimpleHTTPRequestHandler):
def __init__(self, *a, **kw):
super().__init__(*a, directory=root, **kw)
def log_message(self, *a):
pass
return Handler
@property
def base(self):
return "http://127.0.0.1:%d" % self.port
def stop(self):
self.httpd.shutdown()
self.httpd.server_close()
def sh(snippet, env=None, cwd=None):
"""Source bootstrap.sh with main suppressed, then run snippet."""
full = 'NUMOS_SOURCE_ONLY=1 . "%s"; %s' % (BOOTSTRAP, snippet)
merged = dict(os.environ)
merged["NUMOS_SOURCE_ONLY"] = "1"
merged["NUMOS_LIB"] = LIB
merged.pop("NUMOS_STATE", None)
if env:
merged.update(env)
proc = subprocess.run(["bash", "-c", full], capture_output=True, text=True,
env=merged, cwd=cwd)
return proc.returncode, proc.stdout.strip(), proc.stderr.strip()
def run_main(env, cwd=None):
"""Run bootstrap.sh end to end."""
merged = dict(os.environ)
merged["NUMOS_LIB"] = LIB
merged.pop("NUMOS_SOURCE_ONLY", None)
merged.pop("NUMOS_STATE", None)
merged.update(env)
proc = subprocess.run(["bash", BOOTSTRAP], capture_output=True, text=True,
env=merged, cwd=cwd)
return proc.returncode, proc.stdout.strip(), proc.stderr.strip()
class ChainCase(unittest.TestCase):
"""Shared fixture: a served artifact tree plus a local work directory."""
def setUp(self):
self.served = tempfile.mkdtemp()
self.work = tempfile.mkdtemp()
os.makedirs(os.path.join(self.served, "artifacts"))
self.numinit = b"#!/bin/sh\necho FAKE_NUMINIT_RAN state=$NUMOS_STATE\n"
self.state = b"V 1\nC " + b"0" * 64 + b"\n"
self.publish("artifacts/numinit.sh", self.numinit)
self.manifest = (
"V 1\n"
"F numinit.sh %s %d\n" % (sha256(self.numinit), len(self.numinit))
)
self.publish("artifacts/manifest.txt", self.manifest.encode())
self.server = Server(self.served)
def tearDown(self):
self.server.stop()
shutil.rmtree(self.served, ignore_errors=True)
shutil.rmtree(self.work, ignore_errors=True)
def publish(self, rel, data):
path = os.path.join(self.served, rel)
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as handle:
handle.write(data)
def local(self, rel, data):
path = os.path.join(self.work, rel)
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "wb") as handle:
handle.write(data)
return path.replace("\\", "/")
def env(self, **extra):
base = {"NUMOS_BASE": self.server.base, "NUMOS_FAKE_UNAME_M": "x86_64"}
base.update(extra)
return base
class TestManifestLookup(ChainCase):
def test_returns_hash_and_size_for_a_known_artifact(self):
path = self.local("manifest.txt", self.manifest.encode())
code, out, err = sh('numos_manifest_lookup "%s" numinit.sh' % path)
self.assertEqual(code, 0, err)
self.assertEqual(out, "%s %d" % (sha256(self.numinit), len(self.numinit)))
def test_unknown_artifact_halts_with_its_name(self):
path = self.local("manifest.txt", self.manifest.encode())
code, _, err = sh('numos_manifest_lookup "%s" ghost.sh' % path)
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("ghost.sh", err)
def test_missing_manifest_halts(self):
code, _, err = sh('numos_manifest_lookup "/nonexistent/manifest.txt" numinit.sh')
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
class TestFetch(ChainCase):
def test_fetches_a_published_artifact(self):
dest = os.path.join(self.work, "got.sh").replace("\\", "/")
code, _, err = sh('numos_fetch "$NUMOS_BASE/artifacts/numinit.sh" "%s"' % dest,
env=self.env())
self.assertEqual(code, 0, err)
with open(dest, "rb") as handle:
self.assertEqual(handle.read(), self.numinit)
def test_missing_url_halts_and_leaves_no_partial_file(self):
dest = os.path.join(self.work, "got.sh").replace("\\", "/")
code, _, err = sh('numos_fetch "$NUMOS_BASE/artifacts/ghost.sh" "%s"' % dest,
env=self.env())
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertFalse(os.path.exists(dest),
"a failed fetch must not leave a partial artifact behind")
def test_unreachable_host_halts(self):
dest = os.path.join(self.work, "got.sh").replace("\\", "/")
code, _, err = sh('numos_fetch "http://127.0.0.1:1/x" "%s"' % dest,
env=self.env())
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
class TestAcquire(ChainCase):
def test_uses_a_local_artifact_without_touching_the_network(self):
# Point the base at a dead port: if acquire reaches for the network at
# all, this fails. Offline boot is the contract, not an optimization.
self.local("numinit.sh", self.numinit)
manifest = self.local("manifest.txt", self.manifest.encode())
code, out, err = sh(
'numos_acquire numinit.sh "%s/numinit.sh" "%s"' % (self.work.replace("\\", "/"), manifest),
env=self.env(NUMOS_BASE="http://127.0.0.1:1"))
self.assertEqual(code, 0, err)
self.assertIn("on-disk", out)
def test_fetches_when_the_artifact_is_absent(self):
manifest = self.local("manifest.txt", self.manifest.encode())
dest = os.path.join(self.work, "numinit.sh").replace("\\", "/")
code, out, err = sh('numos_acquire numinit.sh "%s" "%s"' % (dest, manifest),
env=self.env())
self.assertEqual(code, 0, err)
self.assertIn("fetched", out)
self.assertTrue(os.path.exists(dest))
def test_local_artifact_with_wrong_hash_halts_rather_than_refetching(self):
# Silently replacing a corrupt local artifact would mask tampering.
self.local("numinit.sh", b"#!/bin/sh\necho TAMPERED\n")
manifest = self.local("manifest.txt", self.manifest.encode())
dest = os.path.join(self.work, "numinit.sh").replace("\\", "/")
code, _, err = sh('numos_acquire numinit.sh "%s" "%s"' % (dest, manifest),
env=self.env())
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("hash mismatch", err)
def test_fetched_artifact_with_wrong_hash_halts(self):
self.publish("artifacts/numinit.sh", b"#!/bin/sh\necho SWAPPED\n")
manifest = self.local("manifest.txt", self.manifest.encode())
dest = os.path.join(self.work, "numinit.sh").replace("\\", "/")
code, _, err = sh('numos_acquire numinit.sh "%s" "%s"' % (dest, manifest),
env=self.env())
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("hash mismatch", err)
def test_offline_mode_refuses_to_fetch(self):
manifest = self.local("manifest.txt", self.manifest.encode())
dest = os.path.join(self.work, "numinit.sh").replace("\\", "/")
code, _, err = sh('numos_acquire numinit.sh "%s" "%s"' % (dest, manifest),
env=self.env(NUMOS_OFFLINE="1"))
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertIn("offline", err)
class TestResolveInit(ChainCase):
"""Which artifact gets used is decided by the manifest, not by the arch table.
numos_arch_has_static reports whether an arch COULD have a static build
(it has a cross-compilation triple). Only the manifest knows whether one
was actually published. Conflating the two made the bootstrap announce a
static binary and then fetch the shell script.
"""
def test_falls_back_to_the_shell_floor_when_no_static_is_published(self):
manifest = self.local("manifest.txt", self.manifest.encode())
# x86_64 has a zig triple, so the arch table says a static build is
# possible - but the manifest publishes no numinit-x86_64.
code, out, err = sh('numos_arch_has_static x86_64 && echo possible',
env=self.env())
self.assertEqual(out, "possible", "fixture assumption changed")
code, out, err = sh('numos_resolve_init x86_64 "%s"' % manifest)
self.assertEqual(code, 0, err)
self.assertEqual(out, "numinit.sh")
def test_uses_the_static_binary_when_the_manifest_publishes_one(self):
static = b"\x7fELF fake static numinit\n"
self.publish("artifacts/numinit-x86_64", static)
text = self.manifest + "F numinit-x86_64 %s %d\n" % (
sha256(static), len(static))
manifest = self.local("manifest.txt", text.encode())
code, out, err = sh('numos_resolve_init x86_64 "%s"' % manifest)
self.assertEqual(code, 0, err)
self.assertEqual(out, "numinit-x86_64")
def test_an_arch_with_no_static_target_at_all_uses_the_floor(self):
manifest = self.local("manifest.txt", self.manifest.encode())
code, out, _ = sh('numos_resolve_init loongarch64 "%s"' % manifest)
self.assertEqual(out, "numinit.sh")
class TestFullChain(ChainCase):
def _state_file(self):
body = "V 1\n"
digest = sha256(body.encode())
return self.local("numos.state", ("V 1\nC %s\n" % digest).encode())
def test_chain_verifies_then_execs_numinit(self):
state = self._state_file()
code, out, err = run_main(self.env(
NUMOS_STATE=state,
NUMOS_PREFIX=self.work.replace("\\", "/"),
))
self.assertEqual(code, 0, err)
self.assertIn("FAKE_NUMINIT_RAN", out)
self.assertIn("state=%s" % state, out)
def test_chain_halts_before_exec_on_a_bad_state(self):
bad = self.local("numos.state", b"V 1\nC " + b"a" * 64 + b"\n")
code, out, err = run_main(self.env(
NUMOS_STATE=bad,
NUMOS_PREFIX=self.work.replace("\\", "/"),
))
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertNotIn("FAKE_NUMINIT_RAN", out,
"an unverified state must never reach exec")
def test_a_tampered_local_artifact_stops_the_boot(self):
"""The whole point of the chain: a modified init must never execute.
This caught a real defect. `echo "numos: $(numos_acquire ...)"` looks
harmless, but a command substitution's exit status only propagates
under `set -e` in an ASSIGNMENT - inside echo it is discarded. So
numos_die killed the subshell, echo printed an empty line, main
continued, and the tampered artifact was exec'd. Asserting on the
HALT message alone would NOT have caught it: the message was printed.
The exit status and the absence of the artifact's own output are what
make this test bite.
"""
state = self._state_file()
prefix = os.path.join(self.work, "prefix")
os.makedirs(prefix)
# A local copy that does not match the manifest.
with open(os.path.join(prefix, "numinit.sh"), "wb") as handle:
handle.write(b"#!/bin/sh\necho TAMPERED_NUMINIT_RAN\n")
with open(os.path.join(prefix, "manifest.txt"), "w", newline="\n") as handle:
handle.write(self.manifest)
code, out, err = run_main(self.env(
NUMOS_STATE=state,
NUMOS_PREFIX=prefix.replace("\\", "/"),
))
self.assertNotEqual(code, 0, "a tampered artifact must halt the boot")
self.assertIn("hash mismatch", err)
self.assertNotIn("TAMPERED_NUMINIT_RAN", out,
"the tampered artifact was executed")
self.assertNotIn("state verified", out,
"the boot continued past a failed artifact check")
def test_a_stale_on_disk_manifest_is_refreshed_when_online(self):
"""A persistent prefix must not pin an old manifest forever.
Found in production: the edge served a cached manifest alongside a
freshly published artifact, so the fetch brought down a current file
that the old manifest rejected. The halt was correct behaviour on a
mismatch that was not real tampering. Refreshing the manifest on
every online boot removes the whole class.
"""
state = self._state_file()
prefix = os.path.join(self.work, "prefix")
os.makedirs(prefix)
# A manifest naming the right artifact with a hash from some earlier
# publish. If it is not refreshed, acquisition mismatches and halts.
with open(os.path.join(prefix, "manifest.txt"), "w", newline="\n") as handle:
handle.write("V 1\nF numinit.sh %s 1\n" % ("b" * 64))
code, out, err = run_main(self.env(
NUMOS_STATE=state,
NUMOS_PREFIX=prefix.replace("\\", "/"),
))
self.assertEqual(code, 0,
"a stale manifest was not refreshed: %s %s" % (out, err))
self.assertIn("FAKE_NUMINIT_RAN", out)
def test_offline_keeps_the_on_disk_manifest(self):
"""Refreshing must not become a network dependency: offline is the
floor, so whatever is on disk is what gets used."""
state = self._state_file()
prefix = os.path.join(self.work, "prefix")
os.makedirs(prefix)
with open(os.path.join(prefix, "manifest.txt"), "w", newline="\n") as handle:
handle.write(self.manifest)
# Correct artifact already present, so nothing needs fetching.
with open(os.path.join(prefix, "numinit.sh"), "wb") as handle:
handle.write(self.numinit)
code, out, err = run_main(self.env(
NUMOS_STATE=state,
NUMOS_PREFIX=prefix.replace("\\", "/"),
NUMOS_BASE="http://127.0.0.1:1",
NUMOS_OFFLINE="1",
))
self.assertEqual(code, 0, "offline boot reached for the network: %s" % err)
self.assertIn("on-disk", out)
self.assertIn("FAKE_NUMINIT_RAN", out)
def test_a_manifest_missing_the_artifact_stops_the_boot(self):
"""Same masking risk one layer down: numos_manifest_lookup dies inside
a substitution that used to be piped through cut, which swallowed its
status and left the expected hash empty.
Runs offline so the deliberately-incomplete manifest survives. Online,
it would be refreshed from the server and the premise would vanish -
which is itself the correct behaviour, just not what this tests.
"""
state = self._state_file()
prefix = os.path.join(self.work, "prefix")
os.makedirs(prefix)
with open(os.path.join(prefix, "manifest.txt"), "w", newline="\n") as handle:
handle.write("V 1\nF something-else.sh %s 1\n" % ("0" * 64))
code, out, err = run_main(self.env(
NUMOS_STATE=state,
NUMOS_PREFIX=prefix.replace("\\", "/"),
NUMOS_OFFLINE="1",
))
self.assertNotEqual(code, 0)
self.assertIn("numos: HALT:", err)
self.assertNotIn("state verified", out)
def test_chain_halts_on_unknown_arch_before_fetching_anything(self):
state = self._state_file()
code, out, err = run_main(self.env(
NUMOS_FAKE_UNAME_M="vax",
NUMOS_STATE=state,
NUMOS_PREFIX=self.work.replace("\\", "/"),
))
self.assertNotEqual(code, 0)
self.assertIn("vax", err)
self.assertNotIn("FAKE_NUMINIT_RAN", out)
if __name__ == "__main__":
unittest.main()