tests/test_principles.py
back to source
"""Executable governance.
docs/PRINCIPLES.md states the commitments this project holds itself to. A
principle that nothing enforces is decoration, so each one that CAN be
tested is tested here, against the real shell and a real HTTP server.
The three enforced here:
no-surveillance a boot requests exactly the artifacts it needs from the
base it was told to use, and nothing else - no beacon,
no identity ping, no usage report
subsidiarity a node with no network still supervises itself: local
remediation runs, and escalation is queued rather than
depended upon
truthfulness the status line stating the boot is unverified cannot be
quietly removed
"""
import ast
import hashlib
import http.server
import os
import re
import shutil
import subprocess
import tempfile
import threading
import unittest
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("\\", "/")
NUMINIT = os.path.join(ROOT, "boot", "numinit.sh").replace("\\", "/")
LIB = os.path.join(ROOT, "boot", "lib").replace("\\", "/")
def bash():
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")
class RecordingServer:
"""Serves a tree and records every path requested of it."""
def __init__(self, root):
self.paths = []
recorder = self.paths
class Handler(http.server.SimpleHTTPRequestHandler):
def __init__(self, *a, **kw):
super().__init__(*a, directory=root, **kw)
def do_GET(self):
recorder.append(self.path)
super().do_GET()
def do_POST(self):
recorder.append("POST " + self.path)
self.send_response(204)
self.end_headers()
def log_message(self, *a):
pass
self.httpd = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
self.port = self.httpd.server_address[1]
threading.Thread(target=self.httpd.serve_forever, daemon=True).start()
@property
def base(self):
return "http://127.0.0.1:%d" % self.port
def stop(self):
self.httpd.shutdown()
self.httpd.server_close()
class PrincipleCase(unittest.TestCase):
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_INIT\n"
self.publish("artifacts/numinit.sh", self.numinit)
manifest = "V 1\nF numinit.sh %s %d\n" % (
hashlib.sha256(self.numinit).hexdigest(), len(self.numinit))
self.publish("artifacts/manifest.txt", manifest.encode())
self.server = RecordingServer(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 state_file(self, name="numos.state"):
body = "V 1\n"
text = "V 1\nC %s\n" % hashlib.sha256(body.encode()).hexdigest()
path = os.path.join(self.work, name)
with open(path, "w", newline="\n") as handle:
handle.write(text)
return path.replace("\\", "/")
class TestNoSurveillance(PrincipleCase):
"""CCC 1907: the common good presupposes respect for the person, including
the safeguarding of privacy. A machine's owner is not this project's data
source."""
def test_a_boot_requests_only_the_artifacts_it_needs(self):
state = self.state_file()
env = dict(os.environ)
env.update({
"NUMOS_BASE": self.server.base,
"NUMOS_FAKE_UNAME_M": "x86_64",
"NUMOS_LIB": LIB,
"NUMOS_STATE": state,
"NUMOS_PREFIX": os.path.join(self.work, "p").replace("\\", "/"),
})
env.pop("NUMOS_SOURCE_ONLY", None)
proc = subprocess.run([bash(), BOOTSTRAP], capture_output=True,
text=True, env=env, timeout=120)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertEqual(
sorted(set(self.server.paths)),
["/artifacts/manifest.txt", "/artifacts/numinit.sh"],
"the boot contacted something it was not asked to: %s"
% self.server.paths)
def test_no_request_carries_identifying_query_parameters(self):
state = self.state_file()
env = dict(os.environ)
env.update({
"NUMOS_BASE": self.server.base,
"NUMOS_FAKE_UNAME_M": "x86_64",
"NUMOS_LIB": LIB,
"NUMOS_STATE": state,
"NUMOS_PREFIX": os.path.join(self.work, "p2").replace("\\", "/"),
})
env.pop("NUMOS_SOURCE_ONLY", None)
subprocess.run([bash(), BOOTSTRAP], capture_output=True, text=True,
env=env, timeout=120)
for path in self.server.paths:
self.assertNotIn("?", path,
"a request carried query parameters: %s" % path)
self.assertFalse(path.startswith("POST "),
"the boot path made a POST: %s" % path)
def test_the_shell_reaches_the_network_only_through_one_function(self):
"""Every outbound call goes through numos_fetch, so there is exactly
one place to audit. A curl or wget anywhere else would be an
unreviewed channel."""
for script in (BOOTSTRAP, NUMINIT):
with open(script) as handle:
code_lines = [line for line in handle.read().split("\n")
if not line.lstrip().startswith("#")]
body = "\n".join(code_lines)
for tool in ("curl", "wget", "nc ", "telnet"):
occurrences = body.count(tool)
if occurrences and os.path.basename(script) != "bootstrap.sh":
self.fail("%s reaches the network directly via %r"
% (os.path.basename(script), tool))
# In bootstrap.sh the only permitted callers are inside numos_fetch.
if os.path.basename(script) == "bootstrap.sh":
fetch = body[body.index("numos_fetch()"):]
fetch = fetch[:fetch.index("\n}")]
self.assertEqual(body.count("curl "), fetch.count("curl "),
"curl is used outside numos_fetch")
self.assertEqual(body.count("wget "), fetch.count("wget "),
"wget is used outside numos_fetch")
class TestSubsidiarity(PrincipleCase):
"""CCC 1883: a higher-order community should not deprive a lower one of
its functions, but support it. A node keeps its own house with no network
and escalates only as help, never as a dependency."""
def test_a_node_with_no_network_still_applies_local_remediation(self):
units = [{"name": "a", "kind": "oneshot", "restart": "never",
"backoff_ms": 0, "backoff_max_ms": 0, "arch_mask": [],
"health_probe": None, "requires": [], "after": [],
"exec": "true"}]
phases = [{"ordinal": 10, "name": "p", "on_failure": "halt",
"required_units": ["a"]}]
healths = [{"name": "disk", "interval_s": 1, "threshold": 1,
"local_action": "degrade-node", "on_fire": "swarm-x",
"probe": "false"}]
text = render({"version": 1, "arch_targets": [], "units": units,
"boot_phases": phases, "health": healths})
path = os.path.join(self.work, "s.state")
with open(path, "w", newline="\n") as handle:
handle.write(text)
rundir = os.path.join(self.work, "run")
env = dict(os.environ)
env.update({
"NUMOS_STATE": path.replace("\\", "/"),
"NUMOS_RUNDIR": rundir.replace("\\", "/"),
"NUMOS_MAX_TICKS": "2",
"NUMOS_NO_SLEEP": "1",
# No reachable base at all.
"NUMOS_BASE": "http://127.0.0.1:1",
"NUMOS_OFFLINE": "1",
})
env.pop("NUMOS_DRY_RUN", None)
env.pop("NUMOS_SOURCE_ONLY", None)
proc = subprocess.run([bash(), NUMINIT], capture_output=True,
text=True, env=env, timeout=120)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("HEALTH-FIRE disk", proc.stdout,
"an offline node stopped watching itself")
self.assertIn("degraded=1", proc.stdout,
"local remediation did not run without a network")
self.assertIn("queued escalation", proc.stdout,
"escalation was dropped rather than queued")
self.assertTrue(
os.path.exists(os.path.join(rundir, "escalations.queue")),
"the queue is the record that help was asked for; it must persist")
class TestTruthfulness(PrincipleCase):
"""CCC 2469: people cannot live together without mutual confidence that
they are being truthful. The status line is the project's central factual
claim about itself and must not be quietly softened into unscoped victory."""
# Scoped honesty: self-init complete on kernel+initrd AND hybrid ISO/GRUB
# serial; fleet degraded on bare QEMU; metal / other arches still open.
# Continuous one-line form (no newline splits) for dual-claim STATUS.
STATUS = (
"bootstrap logic tested; x86_64 QEMU self-init PASS (degraded=0) "
"on kernel+initrd and hybrid ISO/GRUB serial; fleet bare-QEMU "
"PASS-degraded; metal and non-x86_64 unverified"
)
def test_the_readme_states_the_scoped_boot_status(self):
with open(os.path.join(ROOT, "README.md"), encoding="utf-8") as handle:
self.assertIn(self.STATUS, handle.read())
def test_the_published_status_page_states_it_too(self):
page = os.path.join(ROOT, "site", "content", "status.html")
with open(page, encoding="utf-8") as handle:
self.assertIn(self.STATUS, handle.read())
def test_no_source_file_claims_unscoped_boot_victory(self):
claims = re.compile(
r"(successfully boots|boots successfully|verified on real hardware"
r"|known to boot|proven to boot|boots on all arches"
r"|metal verified)", re.I)
for folder in ("boot", "numos", "skills", "site/content"):
base = os.path.join(ROOT, *folder.split("/"))
for root, _dirs, files in os.walk(base):
for name in files:
path = os.path.join(root, name)
with open(path, encoding="utf-8", errors="ignore") as handle:
found = claims.search(handle.read())
self.assertIsNone(
found,
"%s claims a verified boot: %r"
% (path, found.group(0) if found else ""))
class TestWhatWePublish(PrincipleCase):
"""The repository is published as a cloneable remote, so what it tracks is
what every user downloads and is asked to trust. It should contain the
source someone would review, not generated output they cannot.
This guard exists because it failed: the ignore rule read `dist-site/`
while builds used dist-site2..5 to sidestep a locked output tree, so 917
files of generated site output were committed and republished - each build
embedding the previous one's copy of itself.
"""
def test_no_build_output_is_tracked(self):
tracked = subprocess.run(
["git", "-C", ROOT, "ls-files"],
capture_output=True, text=True, check=True).stdout.split()
generated = [p for p in tracked
if p.startswith(("dist", "public", "node_modules"))
or "/__pycache__/" in p
or p.endswith(".pyc")]
self.assertEqual(generated, [],
"generated output is tracked and would be published")
def test_no_tooling_cache_is_tracked(self):
""".wrangler/cache carries the Cloudflare account id and an
email-derived account name. It leaked once."""
tracked = subprocess.run(
["git", "-C", ROOT, "ls-files"],
capture_output=True, text=True, check=True).stdout.split()
caches = [p for p in tracked
if p.startswith((".wrangler/", ".env", ".dev.vars"))]
self.assertEqual(caches, [],
"a tooling cache is tracked and would be published")
if __name__ == "__main__":
unittest.main()
class TestPublishedSurfacesAreNotEmpty(PrincipleCase):
"""Every surface the project advertises must actually contain something.
`skills/AGENTS.md` was 0 bytes and served publicly as the advertised
"multi-agent protocol" across several deploys. Every test was green the
whole time: the suite covered code BEHAVIOUR, and the published artifact
sat outside the evidence boundary, so an empty file satisfied everything.
The house rule -- a principle that can be tested is tested -- held for
logic and silently exempted output.
Root cause of the truncation was a read-before-write hazard:
`open(f, "w").write(open(f).read().replace(...))` opens for writing, which
truncates, BEFORE the argument expression reads the now-empty file. The
second test below forbids that shape outright.
"""
# Small enough that no honest surface trips it, large enough that a
# truncated or stub file cannot pass.
FLOOR_BYTES = 200
def advertised(self):
paths = [os.path.join(ROOT, "skills", "AGENTS.md"),
os.path.join(ROOT, "site", "boot.sh"),
os.path.join(ROOT, "README.md"),
os.path.join(ROOT, "docs", "PRINCIPLES.md")]
skills = os.path.join(ROOT, "skills")
for name in sorted(os.listdir(skills)):
skill = os.path.join(skills, name, "SKILL.md")
if os.path.isfile(skill):
paths.append(skill)
content = os.path.join(ROOT, "site", "content")
for name in sorted(os.listdir(content)):
if name.endswith(".html"):
paths.append(os.path.join(content, name))
return paths
def test_no_advertised_surface_is_empty_or_a_stub(self):
undersized = []
for path in self.advertised():
self.assertTrue(os.path.isfile(path),
"advertised surface is missing: %s" % path)
size = os.path.getsize(path)
if size < self.FLOOR_BYTES:
undersized.append((os.path.relpath(path, ROOT), size))
self.assertEqual(undersized, [],
"advertised surfaces are empty or stubs: %s"
% undersized)
def test_no_source_file_truncates_a_file_it_then_reads(self):
"""Forbid the exact shape that caused it.
Uses the AST rather than a regex: the first version of this lint
matched the docstring above, which describes the hazard, and reported
the explanation as an offence. A test that fires on prose is noise.
"""
offenders = []
def opened_name(node):
"""Return the variable name a call to open()/io.open() targets."""
if not isinstance(node, ast.Call):
return None
func = node.func
name = getattr(func, "id", None) or getattr(func, "attr", None)
if name != "open" or not node.args:
return None
target = node.args[0]
return getattr(target, "id", None)
def write_mode(node):
if len(node.args) < 2:
return False
mode = node.args[1]
return isinstance(mode, ast.Constant) and isinstance(mode.value, str) and mode.value.startswith("w")
for folder in ("numos", "site", "tests"):
base = os.path.join(ROOT, folder)
for root, _dirs, files in os.walk(base):
for filename in files:
if not filename.endswith(".py"):
continue
path = os.path.join(root, filename)
with open(path, encoding="utf-8") as handle:
try:
tree = ast.parse(handle.read(), filename=path)
except SyntaxError:
continue
for node in ast.walk(tree):
# Looking for open(f, "w").write( ... open(f) ... )
if not isinstance(node, ast.Call):
continue
func = node.func
if not isinstance(func, ast.Attribute):
continue
writer = opened_name(func.value)
if writer is None or not write_mode(func.value):
continue
for arg in node.args:
for inner in ast.walk(arg):
if opened_name(inner) == writer:
offenders.append("%s:%d" % (
os.path.relpath(path, ROOT),
node.lineno))
self.assertEqual(sorted(set(offenders)), [],
"a file is opened for writing and read in the same "
"expression; the write truncates first: %s"
% sorted(set(offenders)))
class TestUniversalDestinationOfGoods(PrincipleCase):
"""CCC 2452: the goods of creation are destined for the entire human race.
This project publishes source rather than binaries and argues people
should be able to inspect and build what will run as PID 1 on their own
machine. For a period there was no licence at all, so a clone granted its
recipient nothing: readable but not usable. The grant is now the thing
being enforced, not merely asserted.
"""
def test_a_licence_exists_and_grants_the_rights_it_claims(self):
path = os.path.join(ROOT, "LICENSE")
self.assertTrue(os.path.isfile(path), "no LICENSE file")
with open(path, encoding="utf-8") as handle:
text = handle.read()
self.assertIn("MIT License", text)
for right in ("use", "copy", "modify", "merge", "publish",
"distribute", "sublicense"):
self.assertIn(right, text,
"the licence does not grant the right to %s" % right)
def test_redistributed_files_carry_an_spdx_identifier(self):
"""These are the files that end up inside somebody's initramfs or
container image, where the repository's LICENSE does not travel."""
redistributed = [
os.path.join(ROOT, "boot", "bootstrap.sh"),
os.path.join(ROOT, "boot", "numinit.sh"),
os.path.join(ROOT, "boot", "numctl"),
os.path.join(ROOT, "site", "boot.sh"),
]
missing = []
for path in redistributed:
with open(path, encoding="utf-8") as handle:
if "SPDX-License-Identifier: MIT" not in handle.read():
missing.append(os.path.relpath(path, ROOT))
self.assertEqual(missing, [],
"redistributed files carry no SPDX identifier: %s"
% missing)
def test_the_gplv2_obligation_for_busybox_is_stated_where_it_is_incurred(self):
"""Solidarity, CCC 1939/1941: an obligation to someone else's commons
is not the owner's to waive, so it is stated at the point of use."""
skill = os.path.join(ROOT, "skills",
"numericalos-build-initramfs", "SKILL.md")
with open(skill, encoding="utf-8") as handle:
text = handle.read()
self.assertIn("GPLv2", text)
for obligation in ("source", "notice"):
self.assertIn(obligation, text.lower(),
"the GPLv2 %s obligation is not stated" % obligation)