NumericalOS

tests/test_bootstrap_verify.py

back to source

import hashlib
import os
import shutil
import subprocess
import tempfile
import unittest

from numos import seed
from numos.state import render

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
BOOTSTRAP = os.path.join(ROOT, "boot", "bootstrap.sh").replace("\\", "/")


def _bash():
    """Resolve bash for Windows hosts where Git bash is installed but not on PATH."""
    found = shutil.which("bash")
    if found:
        return found
    for candidate in (
            r"C:\Program Files\Git\bin\bash.exe",
            r"C:\Program Files\Git\usr\bin\bash.exe",
            "/usr/bin/bash",
            "/bin/bash"):
        if os.path.isfile(candidate):
            return candidate
    raise unittest.SkipTest("bash not found (install Git for Windows or put bash on PATH)")


def sh(snippet):
    full = 'NUMOS_SOURCE_ONLY=1 . "%s"; %s' % (BOOTSTRAP, snippet)
    env = dict(os.environ)
    env["NUMOS_SOURCE_ONLY"] = "1"
    env["NUMOS_LIB"] = os.path.dirname(BOOTSTRAP) + "/lib"
    proc = subprocess.run([_bash(), "-c", full], capture_output=True, text=True, env=env)
    return proc.returncode, proc.stdout.strip(), proc.stderr.strip()


def write_temp(text):
    handle = tempfile.NamedTemporaryFile("w", suffix=".numos", delete=False, newline="\n")
    handle.write(text)
    handle.close()
    return handle.name.replace("\\", "/")


def good_state():
    return render({"version": 1, "arch_targets": seed.ARCH_TARGETS,
                   "boot_phases": seed.BOOT_PHASES, "units": seed.INFRA_UNITS,
                   "health": []})


class TestSha256(unittest.TestCase):
    def test_matches_python_hashlib(self):
        path = write_temp("hello\n")
        try:
            _, out, _ = sh('numos_sha256 "%s"' % path)
            self.assertEqual(out, hashlib.sha256(b"hello\n").hexdigest())
        finally:
            os.unlink(path)


class TestVerifySha256(unittest.TestCase):
    def test_matching_hash_succeeds_silently(self):
        path = write_temp("payload\n")
        want = hashlib.sha256(b"payload\n").hexdigest()
        try:
            code, _, err = sh('numos_verify_sha256 "%s" %s' % (path, want))
            self.assertEqual(code, 0, err)
        finally:
            os.unlink(path)

    def test_mismatched_hash_halts_with_both_values(self):
        path = write_temp("payload\n")
        try:
            code, _, err = sh('numos_verify_sha256 "%s" %s' % (path, "00" * 32))
            self.assertNotEqual(code, 0)
            self.assertIn("numos: HALT:", err)
            self.assertIn("hash mismatch", err)
        finally:
            os.unlink(path)

    def test_missing_file_halts(self):
        code, _, err = sh('numos_verify_sha256 "/nonexistent/path" %s' % ("00" * 32))
        self.assertNotEqual(code, 0)
        self.assertIn("numos: HALT:", err)


class TestVerifyState(unittest.TestCase):
    def test_good_state_verifies(self):
        path = write_temp(good_state())
        try:
            code, _, err = sh('numos_verify_state "%s"' % path)
            self.assertEqual(code, 0, err)
        finally:
            os.unlink(path)

    def test_tampered_state_halts(self):
        path = write_temp(good_state().replace("busybox-riscv64", "busybox-EVIL"))
        try:
            code, _, err = sh('numos_verify_state "%s"' % path)
            self.assertNotEqual(code, 0)
            self.assertIn("numos: HALT:", err)
            self.assertIn("state hash mismatch", err)
        finally:
            os.unlink(path)

    def test_state_without_c_record_halts(self):
        body = "\n".join(l for l in good_state().split("\n") if not l.startswith("C "))
        path = write_temp(body + "\n")
        try:
            code, _, err = sh('numos_verify_state "%s"' % path)
            self.assertNotEqual(code, 0)
            self.assertIn("numos: HALT:", err)
        finally:
            os.unlink(path)

    def test_state_with_multiple_c_records_halts(self):
        """State with two C records should be rejected with specific count."""
        body = good_state()
        # Insert a bogus second C record
        lines = body.split("\n")
        # Find the C record and add a duplicate
        c_line = [l for l in lines if l.startswith("C ")][0]
        lines.insert(1, c_line)  # Insert after the C record
        bad_state = "\n".join(lines)
        path = write_temp(bad_state)
        try:
            code, _, err = sh('numos_verify_state "%s"' % path)
            self.assertNotEqual(code, 0)
            self.assertIn("numos: HALT:", err)
            self.assertIn("2 C records", err)  # Verify the count is in the message
        finally:
            os.unlink(path)

    def test_temp_file_cleaned_up_on_halt_in_numos_sha256(self):
        """Temp file must be cleaned up even if numos_sha256 halts inside numos_verify_state.

        After sourcing bootstrap, override numos_sha256 to numos_die. That forces
        the halt path after numos_verify_state has written its body temp file,
        without depending on a curated PATH: Git Bash always injects /usr/bin
        (with sha256sum), so hiding hash tools via PATH is not portable.
        The EXIT trap installed around the temp file must still remove it.
        """
        tmpdir = tempfile.mkdtemp()
        state_path = write_temp(good_state())
        try:
            # Override after source so the real verify_state still creates the
            # temp body, then dies inside got="$(numos_sha256 "$tmp")".
            full = (
                'NUMOS_SOURCE_ONLY=1 . "%s"; '
                'numos_sha256() { numos_die "no sha256 implementation available"; }; '
                'numos_verify_state "%s"'
            ) % (BOOTSTRAP, state_path)
            env = dict(os.environ)
            env["NUMOS_SOURCE_ONLY"] = "1"
            env["NUMOS_LIB"] = os.path.dirname(BOOTSTRAP) + "/lib"
            # Forward slashes so POSIX sh path join works on Windows/Git Bash.
            env["TMPDIR"] = tmpdir.replace("\\", "/")

            proc = subprocess.run(
                [_bash(), "-c", full], capture_output=True, text=True, env=env)

            self.assertNotEqual(proc.returncode, 0, proc.stderr)
            self.assertIn("numos: HALT:", proc.stderr)
            self.assertIn("no sha256 implementation available", proc.stderr)

            remaining_files = [
                f for f in os.listdir(tmpdir) if f.startswith("numos-verify")]
            self.assertEqual(
                len(remaining_files), 0,
                "Temp files not cleaned up: %s" % remaining_files)
        finally:
            os.unlink(state_path)
            try:
                os.rmdir(tmpdir)
            except OSError:
                # Leave evidence if cleanup failed so the assert message is clear.
                for f in os.listdir(tmpdir):
                    try:
                        os.unlink(os.path.join(tmpdir, f))
                    except OSError:
                        pass
                os.rmdir(tmpdir)


if __name__ == "__main__":
    unittest.main()