admin
c9fa2e717b
gates / gates (push) Successful in 18s
felhom-os-apply gains mode 'bundle' (signed, verified by the wrapper itself against the root-owned signers file — or, when that file is missing, only the installer's pinned key, which it then creates) and --install-bundle (the installer's root entry). BUNDLE_FILES is the one table of paths; every check (visudo, sh/bash -n, python, unit sections, RuntimeDirectory guard, nft -c, the route itself) runs before the first write; a failed write or self-check puts every previous copy back. The trust root is never a bundle path (R17). scripts/build-config-bundle.py builds it reproducibly; release-agent.sh publishes it beside the binary. The agent reports the bundle record in system.config_bundle. felhom-opsign signs agent_config_update. 43 wrapper tests (22 mutants red), Go executor tests. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
546 lines
24 KiB
Python
546 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""Tests for the config bundle (R-840, `11` §5.4.2): felhom-os-apply's `bundle` mode and `--install-bundle`, and
|
|
scripts/build-config-bundle.py. An in-memory host plays the files; nothing real is written or run. Each refusal has a
|
|
test; each test names the rule it pins. Red-proof: `audits/r840-config-bundle-2026-10-04/partB/redproof.txt`.
|
|
|
|
Run: python3 configs/test_felhom_config_bundle.py (also run by internal/osupdate's Go test)
|
|
"""
|
|
import sys
|
|
sys.dont_write_bytecode = True # importing the builder must not leave scripts/__pycache__ behind
|
|
import base64
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import stat as statmod
|
|
import unittest
|
|
|
|
HERE = pathlib.Path(__file__).resolve().parent
|
|
REPO = HERE.parent
|
|
_loader = importlib.machinery.SourceFileLoader("osapply", os.environ.get("OSAPPLY_UNDER_TEST", str(HERE / "felhom-os-apply")))
|
|
_spec = importlib.util.spec_from_loader("osapply", _loader)
|
|
osapply = importlib.util.module_from_spec(_spec)
|
|
_loader.exec_module(osapply)
|
|
_bl = importlib.machinery.SourceFileLoader("bundlebuild", str(REPO / "scripts" / "build-config-bundle.py"))
|
|
_bs = importlib.util.spec_from_loader("bundlebuild", _bl)
|
|
builder = importlib.util.module_from_spec(_bs)
|
|
_bl.exec_module(builder)
|
|
|
|
PLAN = "/var/lib/felhom-agent/os/plan-b1.json"
|
|
BUNDLE = "/var/lib/felhom-agent/os/bundle-0.143.0.json"
|
|
HOST = "demo-hp-bb76ea"
|
|
INSTALLER = REPO.parent / "felhom.eu" / "scripts" / "felhom-host-install.sh"
|
|
|
|
|
|
class St:
|
|
def __init__(self, mode, uid):
|
|
self.st_mode, self.st_uid, self.st_size = mode, uid, 100
|
|
|
|
|
|
class Box:
|
|
"""An in-memory host. files: path -> bytes; modes/uids per path; dirs: a set."""
|
|
|
|
def __init__(self, bundle_bytes, signed, oob=False, signers=True):
|
|
self.files, self.modes, self.uids = {}, {}, {}
|
|
self.dirs = {osapply.OOB_DIR} if oob else set()
|
|
self.put(osapply.TRUST_FILE, json.dumps({"host_id": HOST, "ring0_slow_lane": False}).encode(), 0o644)
|
|
if signers:
|
|
self.put(osapply.TRUST_SIGNERS, b'felhom-op-1 namespaces="felhom-op-v1" ssh-ed25519 AAAA felhom-op-1\n', 0o644)
|
|
self.put("/proc/sys/kernel/panic", b"0\n", 0o644)
|
|
self.plan = {"release_id": "bundle-0.143.0", "layer": "host", "mode": "bundle", "bundle": BUNDLE, "signed": signed}
|
|
self.put(PLAN, json.dumps(self.plan).encode(), 0o600, uid=999)
|
|
self.put(BUNDLE, bundle_bytes, 0o600, uid=999)
|
|
self.sig_rc, self.nonces, self.clock = 0, {}, 1791115200.0 # 2026-10-04T12:00:00Z
|
|
self.calls, self.logs, self.writes = [], [], []
|
|
self.visudo_fail = False # the WHOLE sudoers (`visudo -c`) after install
|
|
self.sudo_l = " (root) NOPASSWD: /usr/local/sbin/felhom-os-apply --plan /var/lib/felhom-agent/os/plan-*.json\n"
|
|
self.guard = {"armed": True, "kernel_panic": 10}
|
|
|
|
def put(self, p, data, mode, uid=0):
|
|
self.files[p], self.modes[p], self.uids[p] = data, mode, uid
|
|
|
|
# Runner interface
|
|
def now(self):
|
|
return self.clock
|
|
|
|
def log(self, line):
|
|
self.logs.append(line)
|
|
|
|
def agent_uid(self):
|
|
return 999
|
|
|
|
def verify_sig(self, signers, key_id, ns, blob, sig):
|
|
self.verified = (signers, key_id, ns)
|
|
return self.sig_rc
|
|
|
|
def read_nonces(self):
|
|
return dict(self.nonces)
|
|
|
|
def write_nonces(self, d):
|
|
self.nonces = dict(d)
|
|
|
|
def read_file(self, p):
|
|
if p not in self.files:
|
|
raise OSError("no such file")
|
|
return self.files[p].decode()
|
|
|
|
def read_bytes(self, p):
|
|
if p not in self.files:
|
|
raise OSError("no such file")
|
|
return self.files[p]
|
|
|
|
def stat(self, p):
|
|
if p not in self.files:
|
|
raise OSError("no such file")
|
|
return St(statmod.S_IFREG | self.modes[p], self.uids[p])
|
|
|
|
def lexists(self, p):
|
|
return p in self.files
|
|
|
|
def isdir(self, p):
|
|
return p in self.dirs
|
|
|
|
def put_file(self, p, data, mode):
|
|
self.writes.append(p)
|
|
self.put(p, data, mode)
|
|
|
|
def remove(self, p):
|
|
self.writes.append("rm " + p)
|
|
del self.files[p]
|
|
|
|
def list_dir(self, p):
|
|
return sorted({k[len(p) + 1:].split("/")[0] for k in self.files if k.startswith(p + "/")})
|
|
|
|
def rmtree(self, p):
|
|
for k in [k for k in self.files if k.startswith(p + "/")]:
|
|
del self.files[k]
|
|
|
|
def check_content(self, kind, data):
|
|
return (1, f"{kind}: syntax error") if b"BROKEN-SYNTAX" in data else (0, "")
|
|
|
|
def host(self, argv, timeout=600, stdin=None):
|
|
self.calls.append(argv)
|
|
if argv[:2] == ["visudo", "-c"]:
|
|
return (1, "", "parse error") if self.visudo_fail else (0, "ok", "")
|
|
if argv[:2] == ["sudo", "-n"]:
|
|
return 0, self.sudo_l, ""
|
|
if argv[-1] == "--self-check":
|
|
body = self.files.get("/usr/local/sbin/felhom-os-apply", b"")
|
|
return (0, "felhom-os-apply ok bundle-format=1 files=22\n", "") if b"BUNDLE_OP" in body else (1, "", "boom")
|
|
if argv[-1] == "/usr/local/sbin/felhom-selfupdate-guarded":
|
|
return 2, "", "felhom-selfupdate-guarded: usage: ...\n"
|
|
if argv[-1] == "status" and argv[0].endswith("felhom-crash-guard"):
|
|
return 0, json.dumps(self.guard), ""
|
|
if argv[:3] == ["systemctl", "enable", "--now"] and "felhom-crash-guard.service" in argv:
|
|
self.put("/proc/sys/kernel/panic", f"{self.guard['kernel_panic']}\n".encode(), 0o644)
|
|
return 0, "", ""
|
|
|
|
|
|
def signed_job(sha, version="0.143.0", host=HOST, op="agent_config_update", nonce="b1",
|
|
issued="2026-10-04T11:50:00Z", expires="2026-10-04T12:30:00Z"):
|
|
blob = json.dumps({"expires_at": expires, "issued_at": issued, "key_id": "felhom-op-1", "nonce": nonce, "op": op,
|
|
"params": {"agent_version": version, "bundle_sha256": sha},
|
|
"target": {"guest_id": "", "host_id": host}}, sort_keys=True).encode()
|
|
return {"blob_b64": base64.b64encode(blob).decode(), "sig": "-----BEGIN SSH SIGNATURE-----\nx\n-----END SSH SIGNATURE-----\n"}
|
|
|
|
|
|
def real_bundle(version="0.143.0"):
|
|
data = builder.build(version)
|
|
return data, hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def edited_bundle(edit):
|
|
"""The real bundle, with edit(files_list) applied and every sha recomputed — a SIGNED bundle with bad content."""
|
|
b = json.loads(builder.build("0.143.0"))
|
|
edit(b["files"])
|
|
for e in b["files"]:
|
|
e["sha256"] = hashlib.sha256(base64.b64decode(e["content_b64"])).hexdigest()
|
|
data = (json.dumps(b, indent=1, sort_keys=True) + "\n").encode()
|
|
return data, hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def replace_content(files, path, fn):
|
|
for e in files:
|
|
if e["path"] == path:
|
|
e["content_b64"] = base64.b64encode(fn(base64.b64decode(e["content_b64"]))).decode()
|
|
|
|
|
|
def run(box, argv=None, environ=None):
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = osapply.main(argv or ["felhom-os-apply", "--plan", PLAN], runner=box, environ=environ or {})
|
|
line = [l for l in buf.getvalue().splitlines() if l.startswith("OSAPPLY-REPORT ")][-1]
|
|
return rc, json.loads(line[len("OSAPPLY-REPORT "):])
|
|
|
|
|
|
def box_files(box):
|
|
return {p: box.files[p] for p in box.files if p in osapply.BUNDLE_DESTS}
|
|
|
|
|
|
class Install(unittest.TestCase):
|
|
def test_fresh_box_gets_every_file_and_a_record(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
b = rep["bundle"]
|
|
# every path but the four OOB ones (no belt on this box)
|
|
self.assertEqual(len(b["written"]), len(osapply.BUNDLE_FILES) - 4, b)
|
|
self.assertEqual(len(b["skipped"]), 4)
|
|
self.assertEqual(box.files["/etc/sudoers.d/felhom-agent"], (REPO / "configs" / "felhom-agent.sudoers").read_bytes())
|
|
self.assertEqual(box.modes["/etc/sudoers.d/felhom-agent"], 0o440)
|
|
rec = json.loads(box.files[osapply.BUNDLE_RECORD])
|
|
self.assertEqual((rec["agent_version"], rec["bundle_sha256"], rec["authority"]), ("0.143.0", sha, "signed"))
|
|
self.assertIn("b1", box.nonces, "the job is consumed")
|
|
self.assertEqual(b["self_check"]["crash_guard"], {"armed": True, "kernel_panic": 10})
|
|
self.assertIn(["systemctl", "daemon-reload"], box.calls)
|
|
|
|
def test_sudoers_is_written_after_every_wrapper(self):
|
|
"""Order: a referenced wrapper is in place before the sudoers line that allows it."""
|
|
data, sha = edited_bundle(lambda f: f.reverse()) # the bundle lists the sudoers FIRST; the wrapper must reorder
|
|
box = Box(data, signed_job(sha))
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
w = [p for p in box.writes if p in osapply.BUNDLE_DESTS]
|
|
self.assertEqual(w[-1], "/etc/sudoers.d/felhom-agent")
|
|
self.assertLess(w.index("/usr/local/sbin/felhom-os-apply"), w.index("/etc/sudoers.d/felhom-agent"))
|
|
|
|
def test_identical_box_writes_nothing(self):
|
|
"""The demo boxes' case: hand-copied files equal to the release → 0 written, all 'same'."""
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
for dest, src, mode, _, policy in osapply.BUNDLE_FILES:
|
|
if policy != "oob":
|
|
box.put(dest, (REPO / "configs" / src).read_bytes(), mode)
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
self.assertEqual(rep["bundle"]["written"], [])
|
|
self.assertEqual(rep["bundle"]["same"], len(osapply.BUNDLE_FILES) - 5) # 4 oob skipped + crash-guard.conf kept
|
|
self.assertEqual(rep["bundle"]["kept"], ["/etc/felhom/crash-guard.conf"])
|
|
|
|
def test_a_wrong_mode_is_rewritten(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.put("/etc/sudoers.d/felhom-agent", (REPO / "configs" / "felhom-agent.sudoers").read_bytes(), 0o644)
|
|
rc, rep = run(box)
|
|
self.assertIn("/etc/sudoers.d/felhom-agent", rep["bundle"]["written"])
|
|
self.assertEqual(box.modes["/etc/sudoers.d/felhom-agent"], 0o440)
|
|
|
|
def test_tuned_crash_guard_conf_is_kept(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.put("/etc/felhom/crash-guard.conf", b"LIMIT=5\n", 0o644)
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
self.assertEqual(box.files["/etc/felhom/crash-guard.conf"], b"LIMIT=5\n")
|
|
|
|
def test_oob_files_only_on_a_box_with_the_belt(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha), oob=True)
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
self.assertIn("/etc/sudoers.d/felhom-op", rep["bundle"]["written"])
|
|
self.assertEqual(rep["bundle"]["skipped"], [])
|
|
|
|
def test_previous_copies_are_kept(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.put("/usr/local/sbin/felhom-pbs-apply", b"#!/bin/bash\necho old\n", 0o755)
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
self.assertEqual(box.files[rep["bundle"]["prev_dir"] + "/usr/local/sbin/felhom-pbs-apply"], b"#!/bin/bash\necho old\n")
|
|
|
|
|
|
class Refusals(unittest.TestCase):
|
|
"""Each: refused, and NOTHING on the box changed."""
|
|
|
|
def refused(self, box, code):
|
|
before = dict(box_files(box))
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 2, rep)
|
|
self.assertEqual(rep["refused"]["code"], code, rep)
|
|
self.assertEqual(box_files(box), before, "a refusal changed a file")
|
|
self.assertNotIn(osapply.BUNDLE_RECORD, box.files)
|
|
return rep
|
|
|
|
def test_wrong_sha_is_refused(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job("0" * 64))
|
|
self.refused(box, "R18")
|
|
self.assertEqual(box.nonces, {}, "a wrong sha must not burn the job")
|
|
|
|
def test_bad_signature_is_refused(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.sig_rc = 255
|
|
self.refused(box, "R3")
|
|
|
|
def test_other_op_is_refused(self):
|
|
data, sha = real_bundle()
|
|
self.refused(Box(data, signed_job(sha, op="agent_update")), "R3")
|
|
|
|
def test_other_host_is_refused(self):
|
|
data, sha = real_bundle()
|
|
self.refused(Box(data, signed_job(sha, host="demo-felhom-8363b5")), "R3")
|
|
|
|
def test_expired_job_is_refused(self):
|
|
data, sha = real_bundle()
|
|
self.refused(Box(data, signed_job(sha, expires="2026-10-04T11:55:00Z")), "R3")
|
|
|
|
def test_replayed_job_is_refused(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.nonces = {"b1": box.clock + 600}
|
|
self.refused(box, "R3")
|
|
|
|
def test_version_mismatch_is_refused(self):
|
|
data, sha = real_bundle()
|
|
self.refused(Box(data, signed_job(sha, version="0.142.1")), "R18")
|
|
|
|
def test_sudoers_failing_visudo_is_refused(self):
|
|
data, sha = edited_bundle(lambda f: replace_content(f, "/etc/sudoers.d/felhom-agent", lambda c: c + b"BROKEN-SYNTAX\n"))
|
|
self.refused(Box(data, signed_job(sha)), "R18")
|
|
|
|
def test_sudoers_dropping_the_route_is_refused(self):
|
|
data, sha = edited_bundle(lambda f: replace_content(f, "/etc/sudoers.d/felhom-agent",
|
|
lambda c: c.replace(b"/usr/local/sbin/felhom-os-apply --plan", b"/bin/true --plan")))
|
|
self.refused(Box(data, signed_job(sha)), "R18")
|
|
|
|
def test_wrapper_without_bundle_mode_is_refused(self):
|
|
data, sha = edited_bundle(lambda f: replace_content(f, "/usr/local/sbin/felhom-os-apply",
|
|
lambda c: c.replace(b'BUNDLE_OP = "agent_config_update"', b'BUNDLE_OPX = 1')))
|
|
self.refused(Box(data, signed_job(sha)), "R18")
|
|
|
|
def test_python_syntax_error_is_refused(self):
|
|
data, sha = edited_bundle(lambda f: replace_content(f, "/usr/local/sbin/felhom-crash-guard", lambda c: c + b"\ndef (\n"))
|
|
self.refused(Box(data, signed_job(sha)), "R18")
|
|
|
|
def test_unit_with_runtime_directory_is_refused(self):
|
|
data, sha = edited_bundle(lambda f: replace_content(f, "/etc/systemd/system/felhom-mgmt-watchdog.service",
|
|
lambda c: c + b"RuntimeDirectory=sshd\n"))
|
|
self.refused(Box(data, signed_job(sha)), "R18")
|
|
|
|
def test_agent_unit_not_as_the_agent_user_is_refused(self):
|
|
data, sha = edited_bundle(lambda f: replace_content(f, "/etc/systemd/system/felhom-agent.service",
|
|
lambda c: c.replace(b"User=felhom-agent", b"User=root")))
|
|
self.refused(Box(data, signed_job(sha)), "R18")
|
|
|
|
def test_a_bundle_that_changes_a_signer_is_refused(self):
|
|
"""R17: the trust root is not a bundle's to change — not even a signed one."""
|
|
def add(f):
|
|
f.append({"path": osapply.TRUST_SIGNERS, "content_b64": base64.b64encode(b"evil-key\n").decode()})
|
|
data, sha = edited_bundle(add)
|
|
box = Box(data, signed_job(sha))
|
|
self.refused(box, "R17")
|
|
self.assertIn(b"felhom-op-1", box.files[osapply.TRUST_SIGNERS])
|
|
|
|
def test_a_path_outside_the_table_is_refused(self):
|
|
def add(f):
|
|
f.append({"path": "/etc/shadow", "content_b64": base64.b64encode(b"root::0:0\n").decode()})
|
|
data, sha = edited_bundle(add)
|
|
self.refused(Box(data, signed_job(sha)), "R16")
|
|
|
|
def test_content_not_matching_its_sha_is_refused(self):
|
|
b = json.loads(builder.build("0.143.0"))
|
|
b["files"][0]["content_b64"] = base64.b64encode(b"#!/bin/bash\nexit 0\n").decode()
|
|
data = json.dumps(b).encode()
|
|
self.refused(Box(data, signed_job(hashlib.sha256(data).hexdigest())), "R18")
|
|
|
|
def test_bundle_outside_the_plan_dir_is_refused(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.plan["bundle"] = "/tmp/bundle-0.143.0.json"
|
|
box.files[PLAN] = json.dumps(box.plan).encode()
|
|
self.refused(box, "R1")
|
|
|
|
def test_bundle_not_owned_by_the_agent_is_refused(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.uids[BUNDLE] = 0
|
|
self.refused(box, "R1")
|
|
|
|
def test_no_trust_file_is_refused(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
del box.files[osapply.TRUST_FILE]
|
|
self.refused(box, "R3")
|
|
|
|
|
|
class SelfCheckUndo(unittest.TestCase):
|
|
def assert_restored(self, box, before, rep):
|
|
self.assertTrue(rep["bundle"]["rolled_back"], rep)
|
|
self.assertEqual(box_files(box), before, "the previous files must be back, byte for byte")
|
|
self.assertNotIn(osapply.BUNDLE_RECORD, box.files)
|
|
|
|
def with_old_files(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.put("/usr/local/sbin/felhom-pbs-apply", b"#!/bin/bash\necho old\n", 0o755)
|
|
box.put("/etc/sudoers.d/felhom-agent", b"# old sudoers\n", 0o440)
|
|
return box
|
|
|
|
def test_route_missing_after_install_puts_everything_back(self):
|
|
box = self.with_old_files()
|
|
before = dict(box_files(box))
|
|
box.sudo_l = " (root) NOPASSWD: /bin/true\n"
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 3, rep)
|
|
self.assert_restored(box, before, rep)
|
|
self.assertEqual(box.calls[-1], ["visudo", "-c"], "the undo re-checks the whole sudoers")
|
|
|
|
def test_visudo_failing_after_install_puts_everything_back(self):
|
|
box = self.with_old_files()
|
|
before = dict(box_files(box))
|
|
box.visudo_fail = True
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 3, rep)
|
|
self.assert_restored(box, before, rep)
|
|
|
|
def test_crash_guard_disagreeing_with_kernel_panic_puts_everything_back(self):
|
|
box = self.with_old_files()
|
|
before = dict(box_files(box))
|
|
box.guard = {"armed": True, "kernel_panic": 10}
|
|
box.host_orig = box.host
|
|
|
|
def host(argv, timeout=600, stdin=None):
|
|
if argv[:3] == ["systemctl", "enable", "--now"]:
|
|
box.calls.append(argv)
|
|
return 0, "", "" # the unit "started" but kernel.panic stayed 0
|
|
return box.host_orig(argv, timeout, stdin)
|
|
box.host = host
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 3, rep)
|
|
self.assert_restored(box, before, rep)
|
|
|
|
|
|
class TrustBootstrap(unittest.TestCase):
|
|
def test_missing_signers_verifies_against_the_pinned_key_and_creates_it(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha), signers=False)
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
self.assertEqual(box.verified[0], osapply.PINNED_SIGNERS, "verified against the pinned key, nothing else")
|
|
self.assertTrue(rep["bundle"]["signers_created"])
|
|
self.assertEqual(box.files[osapply.TRUST_SIGNERS], osapply.pinned_signers_line().encode())
|
|
self.assertEqual(box.modes[osapply.TRUST_SIGNERS], 0o644)
|
|
|
|
def test_missing_signers_and_a_job_the_pinned_key_did_not_sign(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha), signers=False)
|
|
box.sig_rc = 255
|
|
rc, rep = run(box)
|
|
self.assertEqual((rc, rep["refused"]["code"]), (2, "R3"))
|
|
self.assertNotIn(osapply.TRUST_SIGNERS, box.files)
|
|
|
|
def test_present_signers_are_never_touched(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.put(osapply.TRUST_SIGNERS, b'felhom-op-2 namespaces="felhom-op-v1" ssh-ed25519 BBBB felhom-op-2\n', 0o644)
|
|
rc, rep = run(box)
|
|
self.assertEqual(rc, 0, rep)
|
|
self.assertEqual(box.verified[0], osapply.TRUST_SIGNERS)
|
|
self.assertFalse(rep["bundle"]["signers_created"])
|
|
self.assertIn(b"felhom-op-2", box.files[osapply.TRUST_SIGNERS])
|
|
|
|
def test_agent_writable_signers_are_refused(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
box.uids[osapply.TRUST_SIGNERS] = 999
|
|
rc, rep = run(box)
|
|
self.assertEqual((rc, rep["refused"]["code"]), (2, "R3"))
|
|
|
|
def test_pinned_operator_key_equals_the_installers(self):
|
|
"""The bootstrap key is exactly the one felhom-host-install.sh pins (OPERATOR_KEY_OPERATIONAL_*)."""
|
|
text = INSTALLER.read_text()
|
|
kid = re.search(r'^OPERATOR_KEY_OPERATIONAL_ID="([^"]+)"', text, re.M).group(1)
|
|
line = re.search(r'^OPERATOR_KEY_OPERATIONAL_LINE="([^"]+)"', text, re.M).group(1)
|
|
self.assertEqual((osapply.PINNED_OPERATOR_KEY_ID, osapply.PINNED_OPERATOR_KEY_LINE), (kid, line))
|
|
# and the file format is the installer's printf, byte for byte
|
|
self.assertIn("printf '%s namespaces=\"felhom-op-v1\" %s\\n'", text)
|
|
|
|
|
|
class InstallerEntry(unittest.TestCase):
|
|
def test_installer_installs_without_a_signature(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, None)
|
|
box.put("/root/bundle.json", data, 0o600)
|
|
rc, rep = run(box, ["felhom-os-apply", "--install-bundle", "/root/bundle.json", "--sha256", sha])
|
|
self.assertEqual(rc, 0, rep)
|
|
self.assertEqual(json.loads(box.files[osapply.BUNDLE_RECORD])["authority"], "installer")
|
|
|
|
def test_installer_entry_checks_the_sha(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, None)
|
|
box.put("/root/bundle.json", data, 0o600)
|
|
rc, rep = run(box, ["felhom-os-apply", "--install-bundle", "/root/bundle.json", "--sha256", "1" * 64])
|
|
self.assertEqual((rc, rep["refused"]["code"]), (2, "R18"))
|
|
|
|
def test_installer_entry_is_refused_through_sudo(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, None)
|
|
box.put("/root/bundle.json", data, 0o600)
|
|
rc, rep = run(box, ["felhom-os-apply", "--install-bundle", "/root/bundle.json", "--sha256", sha], {"SUDO_UID": "999"})
|
|
self.assertEqual((rc, rep["refused"]["code"]), (2, "R1"))
|
|
self.assertNotIn(osapply.BUNDLE_RECORD, box.files)
|
|
|
|
def test_self_check_answers(self):
|
|
buf = io.StringIO()
|
|
with contextlib.redirect_stdout(buf):
|
|
rc = osapply.main(["felhom-os-apply", "--self-check"], runner=Box(b"", None))
|
|
self.assertEqual(rc, 0)
|
|
self.assertIn("bundle-format=1", buf.getvalue())
|
|
|
|
|
|
class Facts(unittest.TestCase):
|
|
def test_state_reports_drift_against_the_record(self):
|
|
data, sha = real_bundle()
|
|
box = Box(data, signed_job(sha))
|
|
run(box)
|
|
box.put("/usr/local/sbin/felhom-pbs-apply", b"#!/bin/bash\necho by hand\n", 0o755)
|
|
a = osapply.Apply(box, PLAN)
|
|
st = osapply.Bundle(a).state()
|
|
self.assertEqual(st["version"], "0.143.0")
|
|
self.assertEqual(st["drift"], ["/usr/local/sbin/felhom-pbs-apply"])
|
|
self.assertTrue(st["signers_present"])
|
|
|
|
def test_state_without_a_record_says_none(self):
|
|
box = Box(b"", None)
|
|
st = osapply.Bundle(osapply.Apply(box, PLAN)).state()
|
|
self.assertEqual(st["version"], "none")
|
|
self.assertNotIn("drift", st)
|
|
self.assertEqual(st["live"]["/etc/sudoers.d/felhom-agent"], "absent")
|
|
|
|
|
|
class Builder(unittest.TestCase):
|
|
def test_reproducible(self):
|
|
self.assertEqual(builder.build("0.143.0"), builder.build("0.143.0"))
|
|
|
|
def test_every_source_exists_and_every_dest_is_unique(self):
|
|
dests = [e[0] for e in osapply.BUNDLE_FILES]
|
|
self.assertEqual(len(dests), len(set(dests)))
|
|
for _, src, *_ in osapply.BUNDLE_FILES:
|
|
self.assertTrue((REPO / "configs" / src).is_file(), src)
|
|
|
|
def test_every_root_file_the_installer_writes_is_in_the_bundle(self):
|
|
"""One source of truth: a felhom root-owned path the installer names must be a bundle path, a trust file, or a
|
|
path the AGENT itself writes at run time (named here, with why). Scope is a regex over the WHOLE installer."""
|
|
text = INSTALLER.read_text()
|
|
found = set(re.findall(r"(/usr/local/sbin/felhom-[a-z-]+|/etc/systemd/system/felhom-[a-z.-]+|"
|
|
r"/etc/sudoers\.d/felhom-[a-z-]+|/etc/felhom-oob\.nft|/etc/tmpfiles\.d/felhom-[a-z.-]+|"
|
|
r"/etc/felhom/[a-z.-]+)", text))
|
|
agent_writes = {"/usr/local/sbin/felhom-shared-parent", "/etc/systemd/system/felhom-shared-parent.service"}
|
|
trust = {osapply.TRUST_FILE, osapply.TRUST_SIGNERS, osapply.TRUST_SIGNERS + ".tmp", osapply.BUNDLE_RECORD}
|
|
missing = sorted(p for p in found if p not in osapply.BUNDLE_DESTS and p not in agent_writes | trust)
|
|
self.assertEqual(missing, [], "the installer writes these root files, but the bundle does not carry them")
|
|
# the limits drop-in is named through $AGENT_UNIT in the installer
|
|
self.assertIn("/etc/systemd/system/felhom-agent.service.d/felhom-agent-limits.conf", osapply.BUNDLE_DESTS)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|