R-840: the config bundle — a signed agent_config_update brings a box's root-owned files (sudoers, wrappers, units)
gates / gates (push) Successful in 18s
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
This commit is contained in:
+500
-15
@@ -93,6 +93,75 @@ DAEMON_JSON = "/etc/docker/daemon.json"
|
||||
DOCKER_SOCKETS = ("/var/run/docker.sock", "/run/docker.sock")
|
||||
CRASH_GUARD_STATE = "/var/lib/felhom-crash-guard/state.json"
|
||||
|
||||
# ---------- the config bundle (R-840, agent v0.143.0, `11` §5.4.2) ----------
|
||||
# A signed `agent_config_update` job carries {agent_version, bundle_sha256}; the bundle is ONE JSON file built from this
|
||||
# repo's configs/ at the agent tag (scripts/build-config-bundle.py) and published beside the binary. The installer
|
||||
# installs the SAME bundle through this same code (--install-bundle, root only, never reachable through sudo), so a new
|
||||
# box and an updated box cannot drift. This table is the ONLY set of paths a bundle may write — a signed bundle naming
|
||||
# any other path is refused (R16) — and it is also the builder's list (one table). Each entry:
|
||||
# dest: (source file under configs/, mode, check, policy)
|
||||
# check: the content check BEFORE anything is written (R18): sudoers → visudo -cf; sh / bash → -n; python → compile;
|
||||
# unit → a [Unit]/[Service]/[Timer] section; unit-nort → that, and never RuntimeDirectory= (the G1 incident);
|
||||
# agent-unit → User=felhom-agent; dropin → a [Unit] section (SF-3); nft → nft -c -f; plain → none.
|
||||
# policy: replace → always written when it differs; if-absent → only when the box has none (a setting the operator may
|
||||
# have tuned); oob → only on a box with the OOB belt (/etc/felhom-sshd exists — the installer's --no-oob leaves none).
|
||||
# Order matters: the sudoers files come LAST, so a referenced wrapper is in place before the line that allows it.
|
||||
BUNDLE_FORMAT = 1
|
||||
BUNDLE_OP = "agent_config_update"
|
||||
BUNDLE_RECORD = "/etc/felhom/config-bundle.json" # what the box runs (0644 root; the non-root agent reports it)
|
||||
BUNDLE_PREV_DIR = "/var/lib/felhom-os-apply/bundle-prev" # the previous copies of every file a bundle replaced
|
||||
BUNDLE_KEEP = 3
|
||||
BUNDLE_MAX_BYTES = 4 * 1024 * 1024
|
||||
BUNDLE_RE = re.compile(r"^bundle-[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.]+)?\.json$")
|
||||
OOB_DIR = "/etc/felhom-sshd"
|
||||
BUNDLE_FILES = [
|
||||
("/usr/local/sbin/felhom-mkfs-guarded", "felhom-mkfs-guarded.sh", 0o755, "bash", "replace"),
|
||||
("/usr/local/sbin/felhom-selfupdate-guarded", "felhom-selfupdate-guarded", 0o755, "sh", "replace"),
|
||||
("/usr/local/sbin/felhom-pbs-apply", "felhom-pbs-apply", 0o755, "bash", "replace"),
|
||||
("/usr/local/sbin/felhom-backup-target-apply", "felhom-backup-target-apply", 0o755, "bash", "replace"),
|
||||
("/usr/local/sbin/felhom-os-apply", "felhom-os-apply", 0o755, "python", "replace"),
|
||||
("/usr/local/sbin/felhom-crash-guard", "felhom-crash-guard", 0o755, "python", "replace"),
|
||||
("/etc/systemd/system/felhom-crash-guard.service", "felhom-crash-guard.service", 0o644, "unit", "replace"),
|
||||
("/etc/systemd/system/felhom-crash-guard-check.service", "felhom-crash-guard-check.service", 0o644, "unit", "replace"),
|
||||
("/etc/systemd/system/felhom-crash-guard-check.timer", "felhom-crash-guard-check.timer", 0o644, "unit", "replace"),
|
||||
("/etc/felhom/crash-guard.conf", "crash-guard.conf", 0o644, "plain", "if-absent"),
|
||||
("/etc/systemd/system/felhom-agent.service", "felhom-agent.service", 0o644, "agent-unit", "replace"),
|
||||
("/etc/systemd/system/felhom-agent-rollback.service", "felhom-agent-rollback.service", 0o644, "unit", "replace"),
|
||||
("/etc/systemd/system/felhom-agent.service.d/felhom-agent-limits.conf", "felhom-agent-limits.conf", 0o644, "dropin", "replace"),
|
||||
("/usr/local/sbin/felhom-mgmt-watchdog", "felhom-mgmt-watchdog.sh", 0o755, "sh", "replace"),
|
||||
("/etc/tmpfiles.d/felhom-privsep.conf", "felhom-privsep.tmpfiles", 0o644, "plain", "replace"),
|
||||
("/etc/systemd/system/felhom-mgmt-watchdog.service", "felhom-mgmt-watchdog.service", 0o644, "unit-nort", "replace"),
|
||||
("/etc/systemd/system/felhom-mgmt-watchdog.timer", "felhom-mgmt-watchdog.timer", 0o644, "unit-nort", "replace"),
|
||||
("/etc/systemd/system/felhom-sshd.service", "felhom-sshd.service", 0o644, "unit-nort", "oob"),
|
||||
("/etc/felhom-oob.nft", "felhom-oob.nft", 0o644, "nft", "oob"),
|
||||
("/etc/systemd/system/felhom-oob-nft.service", "felhom-oob-nft.service", 0o644, "unit", "oob"),
|
||||
("/etc/sudoers.d/felhom-op", "felhom-op.sudoers", 0o440, "sudoers", "oob"),
|
||||
("/etc/sudoers.d/felhom-agent", "felhom-agent.sudoers", 0o440, "sudoers", "replace"),
|
||||
]
|
||||
BUNDLE_DESTS = {e[0]: e for e in BUNDLE_FILES}
|
||||
# The trust root is never a bundle's to change (R17): who may sign is decided by these files, so no bundle may carry them.
|
||||
TRUST_PATHS = (TRUST_FILE, TRUST_SIGNERS, BUNDLE_RECORD)
|
||||
# When /etc/felhom/operator-signers is MISSING (a box installed before installer 1.30.0), the job is verified against —
|
||||
# and the file is created with — exactly this key: the operational key felhom-host-install.sh pins
|
||||
# (OPERATOR_KEY_OPERATIONAL_*). Never any other. Pinned equal to the installer by
|
||||
# test_pinned_operator_key_equals_the_installers. Signer rotation is a separate, later act (`04` §3).
|
||||
PINNED_OPERATOR_KEY_ID = "felhom-op-1"
|
||||
PINNED_OPERATOR_KEY_LINE = ("ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIL8z0qCNgA3x2xxAB0Qj5ro8waFjGZ8Ta/sWB63tlLw+ felhom-op-1")
|
||||
# The self-check (after install, before the record): the route itself must survive — a bundle whose sudoers no longer
|
||||
# lets the agent call this wrapper would cut the box off from every later bundle.
|
||||
SELF_CHECK_SUDO_LINE = "/usr/local/sbin/felhom-os-apply --plan /var/lib/felhom-agent/os/plan-*.json"
|
||||
PINNED_SIGNERS = "<pinned operator key>" # verify_sig: check against the pinned key, not a file on the box
|
||||
|
||||
|
||||
def pinned_signers_line():
|
||||
"""The ssh allowed_signers line the installer writes (felhom-host-install.sh _write_slow_lane_trust), byte for byte."""
|
||||
return f'{PINNED_OPERATOR_KEY_ID} namespaces="{SIG_NAMESPACE}" {PINNED_OPERATOR_KEY_LINE}\n'
|
||||
|
||||
|
||||
def sha256_hex(data):
|
||||
import hashlib
|
||||
return hashlib.sha256(data).hexdigest()
|
||||
|
||||
|
||||
class Refused(Exception):
|
||||
def __init__(self, code, reason):
|
||||
@@ -144,6 +213,10 @@ class Runner:
|
||||
sp = os.path.join(d, "sig")
|
||||
with open(sp, "w") as f:
|
||||
f.write(sig)
|
||||
if signers == PINNED_SIGNERS:
|
||||
signers = os.path.join(d, "allowed_signers")
|
||||
with open(signers, "w") as f:
|
||||
f.write(pinned_signers_line())
|
||||
p = subprocess.run(["ssh-keygen", "-Y", "verify", "-f", signers, "-I", key_id, "-n", namespace, "-s", sp],
|
||||
input=blob, capture_output=True, timeout=30)
|
||||
return p.returncode
|
||||
@@ -170,6 +243,62 @@ class Runner:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ---------- host files, for the config bundle (R-840). Tests replace these with an in-memory tree. ----------
|
||||
def read_bytes(self, path):
|
||||
with open(path, "rb") as f:
|
||||
return f.read()
|
||||
|
||||
def lexists(self, path):
|
||||
return os.path.lexists(path)
|
||||
|
||||
def isdir(self, path):
|
||||
return os.path.isdir(path)
|
||||
|
||||
def put_file(self, path, data, mode):
|
||||
"""Atomic: a root-owned temp file beside the target (same filesystem), fsync, then rename over it."""
|
||||
d = os.path.dirname(path)
|
||||
os.makedirs(d, mode=0o755, exist_ok=True)
|
||||
tmp = os.path.join(d, f".{os.path.basename(path)}.felhom-new.{os.getpid()}")
|
||||
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as f:
|
||||
f.write(data)
|
||||
f.flush()
|
||||
os.fchown(f.fileno(), 0, 0)
|
||||
os.fchmod(f.fileno(), mode)
|
||||
os.fsync(f.fileno())
|
||||
os.replace(tmp, path)
|
||||
except BaseException:
|
||||
try:
|
||||
os.remove(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
|
||||
def remove(self, path):
|
||||
os.remove(path)
|
||||
|
||||
def list_dir(self, path):
|
||||
try:
|
||||
return sorted(os.listdir(path))
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
def rmtree(self, path):
|
||||
import shutil
|
||||
shutil.rmtree(path, ignore_errors=True)
|
||||
|
||||
def check_content(self, kind, data):
|
||||
"""Run an external syntax check on a root-only temp copy. Returns (rc, message)."""
|
||||
import tempfile
|
||||
cmd = {"sudoers": ["visudo", "-cf"], "sh": ["sh", "-n"], "bash": ["bash", "-n"], "nft": ["nft", "-c", "-f"]}[kind]
|
||||
with tempfile.TemporaryDirectory(prefix="felhom-bundle-") as d:
|
||||
p = os.path.join(d, "f")
|
||||
with open(p, "wb") as f:
|
||||
f.write(data)
|
||||
r = subprocess.run(cmd + [p], capture_output=True, text=True, timeout=60)
|
||||
return r.returncode, (r.stdout + r.stderr).strip()[-300:]
|
||||
|
||||
|
||||
class Apply:
|
||||
def __init__(self, runner, plan_path):
|
||||
@@ -203,8 +332,12 @@ class Apply:
|
||||
|
||||
def check_plan(self, plan):
|
||||
mode = plan.get("mode", "apply")
|
||||
if mode not in ("apply", "inventory", "health", "facts", "live-restore-on"):
|
||||
if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle"):
|
||||
raise Refused("R11", f"unknown mode {mode!r}")
|
||||
if mode == "bundle":
|
||||
if plan.get("layer") != "host":
|
||||
raise Refused("R11", "bundle is a host-layer mode")
|
||||
return mode, "host", 0, "bundle"
|
||||
layer = plan.get("layer")
|
||||
if layer not in ("guest", "host", "docker"):
|
||||
raise Refused("R12", f"layer {layer!r} is not guest, host or docker")
|
||||
@@ -298,9 +431,12 @@ class Apply:
|
||||
raise Refused("R3", f"{TRUST_FILE} names no host_id")
|
||||
return t
|
||||
|
||||
def verify_signed(self, signed, trust):
|
||||
"""R3: an operator-signed os_docker_step, checked HERE (not by the agent): signature against the root-owned
|
||||
signers file, op, host binding, time window, and a root-owned nonce record (no replay)."""
|
||||
def verify_signed(self, signed, trust, op_name=SIGNED_OP, signers=None, burn=True):
|
||||
"""R3: an operator-signed job, checked HERE (not by the agent): signature against the root-owned signers file (or,
|
||||
for a bundle on a box that has none, the PINNED key — `signers` names that temp file), op, host binding, time
|
||||
window, and a root-owned nonce record (no replay). burn=False leaves the nonce for the caller to burn after its
|
||||
own checks (a bundle refused for a wrong sha keeps its job usable — the operator fixes the sha, not the key).
|
||||
Returns the params; with burn=False, (params, nonce, expiry)."""
|
||||
import base64
|
||||
if not isinstance(signed, dict) or not isinstance(signed.get("blob_b64"), str) or not isinstance(signed.get("sig"), str):
|
||||
raise Refused("R3", "the signed job is malformed")
|
||||
@@ -312,17 +448,19 @@ class Apply:
|
||||
key_id = op.get("key_id", "")
|
||||
if not isinstance(key_id, str) or not re.match(r"^[A-Za-z0-9._-]{1,64}$", key_id):
|
||||
raise Refused("R3", "the signed blob names no plain key_id")
|
||||
try:
|
||||
st = self.r.stat(TRUST_SIGNERS)
|
||||
except OSError:
|
||||
raise Refused("R3", f"no {TRUST_SIGNERS} — no operator key to check a signed job against")
|
||||
if st.st_uid != 0 or (st.st_mode & 0o022):
|
||||
raise Refused("R3", f"{TRUST_SIGNERS} is not root-owned and root-only-writable")
|
||||
rc = self.r.verify_sig(TRUST_SIGNERS, key_id, SIG_NAMESPACE, blob, signed["sig"])
|
||||
if signers is None:
|
||||
signers = TRUST_SIGNERS
|
||||
try:
|
||||
st = self.r.stat(TRUST_SIGNERS)
|
||||
except OSError:
|
||||
raise Refused("R3", f"no {TRUST_SIGNERS} — no operator key to check a signed job against")
|
||||
if st.st_uid != 0 or (st.st_mode & 0o022):
|
||||
raise Refused("R3", f"{TRUST_SIGNERS} is not root-owned and root-only-writable")
|
||||
rc = self.r.verify_sig(signers, key_id, SIG_NAMESPACE, blob, signed["sig"])
|
||||
if rc != 0:
|
||||
raise Refused("R3", f"the operator signature does not verify (ssh-keygen rc={rc})")
|
||||
if op.get("op") != SIGNED_OP:
|
||||
raise Refused("R3", f"the signed op is {op.get('op')!r}, not {SIGNED_OP}")
|
||||
if op.get("op") != op_name:
|
||||
raise Refused("R3", f"the signed op is {op.get('op')!r}, not {op_name}")
|
||||
if (op.get("target") or {}).get("host_id") != trust["host_id"]:
|
||||
raise Refused("R3", "the signed job is for another host")
|
||||
now = self.r.now()
|
||||
@@ -336,12 +474,20 @@ class Apply:
|
||||
nonce = op.get("nonce")
|
||||
if not isinstance(nonce, str) or not nonce:
|
||||
raise Refused("R3", "the signed job has no nonce")
|
||||
if nonce in self.r.read_nonces():
|
||||
raise Refused("R3", "the signed job was already used (replay)")
|
||||
if not burn:
|
||||
return op.get("params") or {}, nonce, exp
|
||||
self.burn_nonce(nonce, exp)
|
||||
return op.get("params") or {}
|
||||
|
||||
def burn_nonce(self, nonce, exp):
|
||||
seen = self.r.read_nonces()
|
||||
if nonce in seen:
|
||||
raise Refused("R3", "the signed job was already used (replay)")
|
||||
seen[nonce] = exp
|
||||
now = self.r.now()
|
||||
self.r.write_nonces({k: v for k, v in seen.items() if v > now})
|
||||
return op.get("params") or {}
|
||||
|
||||
def docker_authority(self, plan):
|
||||
"""R3 for the docker layer: returns (who, undo). A signed job binds the EXACT package list and the undo flag."""
|
||||
@@ -481,6 +627,10 @@ class Apply:
|
||||
for k, src in (("debian", "debian"), ("docker_engine", "engine"), ("containerd", "containerd")):
|
||||
g[k] = kv.get(src, "").strip() or "unknown"
|
||||
g["live_restore"] = {"true": "on", "false": "off"}.get(kv.get("live", "").strip(), "unknown")
|
||||
try:
|
||||
h["config_bundle"] = Bundle(self).state()
|
||||
except Exception as e: # a read problem must never cost the System page its other facts
|
||||
h["config_bundle"] = {"version": "unknown", "error": str(e)[:200]}
|
||||
self.report["facts"] = {"host": h, "guest": g}
|
||||
return 0
|
||||
|
||||
@@ -660,6 +810,8 @@ class Apply:
|
||||
self.report.update(mode=self.mode, layer=self.layer, release_id=plan.get("release_id"), vmid=self.vmid)
|
||||
if self.mode == "facts":
|
||||
return self.facts()
|
||||
if self.mode == "bundle":
|
||||
return Bundle(self).from_plan(plan)
|
||||
if self.layer == "host":
|
||||
self.check_appliance()
|
||||
self.check_guest(self.vmid)
|
||||
@@ -902,8 +1054,341 @@ class Apply:
|
||||
self.x(APT_ENV + ["apt-get", "-q", "update"], timeout=600)
|
||||
|
||||
|
||||
def main(argv, runner=None):
|
||||
class Bundle:
|
||||
"""The config bundle (R-840, `11` §5.4.2): every root-owned file the installer's step 5 writes, installed as ONE
|
||||
signed unit. Every check runs before the first write; a failed write or a failed self-check puts every previous
|
||||
copy back. Refusal codes: R1 (the bundle file), R3 (authority), R16 (a path outside BUNDLE_FILES), R17 (a trust
|
||||
file), R18 (a content check or a wrong sha)."""
|
||||
|
||||
def __init__(self, apply):
|
||||
self.a, self.r = apply, apply.r
|
||||
self.report = apply.report
|
||||
|
||||
# ---------- reading and checking ----------
|
||||
def load_file(self, path):
|
||||
d, base = os.path.dirname(path or ""), os.path.basename(path or "")
|
||||
if d != PLAN_DIR or not BUNDLE_RE.match(base) or ".." in path:
|
||||
raise Refused("R1", f"the bundle must be {PLAN_DIR}/bundle-<version>.json, got {path!r}")
|
||||
try:
|
||||
st = self.r.stat(path)
|
||||
except OSError as e:
|
||||
raise Refused("R1", f"cannot stat the bundle: {e}")
|
||||
if not stat.S_ISREG(st.st_mode):
|
||||
raise Refused("R1", "the bundle is not a regular file")
|
||||
if st.st_uid != self.r.agent_uid():
|
||||
raise Refused("R1", f"the bundle is not owned by {AGENT_USER}")
|
||||
if st.st_size > BUNDLE_MAX_BYTES:
|
||||
raise Refused("R1", "the bundle is larger than 4 MB")
|
||||
return self.r.read_bytes(path)
|
||||
|
||||
def parse(self, data, want_sha, want_version=None):
|
||||
"""The bundle bytes → [(dest, content, mode, check, policy)], every content check passed. Nothing is written."""
|
||||
import base64
|
||||
got = sha256_hex(data)
|
||||
if got != want_sha:
|
||||
raise Refused("R18", f"the bundle's sha256 is {got}, not the pinned {want_sha}")
|
||||
try:
|
||||
b = json.loads(data)
|
||||
except ValueError as e:
|
||||
raise Refused("R18", f"the bundle is not JSON: {e}")
|
||||
if not isinstance(b, dict) or b.get("format") != BUNDLE_FORMAT:
|
||||
raise Refused("R18", f"the bundle format is not {BUNDLE_FORMAT}")
|
||||
ver = b.get("agent_version")
|
||||
if not isinstance(ver, str) or not re.match(r"^[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.]+)?$", ver):
|
||||
raise Refused("R18", f"the bundle names no agent version ({ver!r})")
|
||||
if want_version is not None and ver != want_version:
|
||||
raise Refused("R18", f"the bundle is for agent {ver}, the signed job pins {want_version}")
|
||||
files = b.get("files")
|
||||
if not isinstance(files, list) or not files:
|
||||
raise Refused("R18", "the bundle lists no files")
|
||||
out, seen = [], set()
|
||||
for e in files:
|
||||
if not isinstance(e, dict) or not isinstance(e.get("path"), str):
|
||||
raise Refused("R18", "a bundle entry has no path")
|
||||
dest = e["path"]
|
||||
if dest in TRUST_PATHS:
|
||||
raise Refused("R17", f"{dest} is the trust root — no bundle may add, remove or change a signer")
|
||||
if dest not in BUNDLE_DESTS:
|
||||
raise Refused("R16", f"{dest} is not a path a bundle may write")
|
||||
if dest in seen:
|
||||
raise Refused("R18", f"{dest} is named twice")
|
||||
seen.add(dest)
|
||||
try:
|
||||
content = base64.b64decode(e.get("content_b64", ""), validate=True)
|
||||
except (ValueError, TypeError):
|
||||
raise Refused("R18", f"{dest}: the content is not base64")
|
||||
if sha256_hex(content) != e.get("sha256"):
|
||||
raise Refused("R18", f"{dest}: the content does not match its sha256")
|
||||
_, _, mode, check, policy = BUNDLE_DESTS[dest]
|
||||
self.check_content(dest, check, content)
|
||||
out.append((dest, content, mode, check, policy))
|
||||
order = [x[0] for x in BUNDLE_FILES]
|
||||
out.sort(key=lambda x: order.index(x[0])) # the table's order: the sudoers files last
|
||||
return ver, out
|
||||
|
||||
def check_content(self, dest, check, content):
|
||||
try:
|
||||
text = content.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
raise Refused("R18", f"{dest} is not UTF-8 text")
|
||||
if check in ("sudoers", "sh", "bash", "nft"):
|
||||
rc, msg = self.r.check_content(check, content)
|
||||
if rc != 0:
|
||||
raise Refused("R18", f"{dest} fails its {check} check: {msg}")
|
||||
elif check == "python":
|
||||
try:
|
||||
compile(text, dest, "exec")
|
||||
except SyntaxError as e:
|
||||
raise Refused("R18", f"{dest} is not valid Python: {e}")
|
||||
elif check in ("unit", "unit-nort", "agent-unit", "dropin"):
|
||||
if not re.search(r"^\[(Unit|Service|Timer)\]\s*$", text, re.M):
|
||||
raise Refused("R18", f"{dest} has no [Unit]/[Service]/[Timer] section")
|
||||
if check == "unit-nort" and re.search(r"^\s*RuntimeDirectory\s*=", text, re.M | re.I):
|
||||
raise Refused("R18", f"{dest} declares RuntimeDirectory= — the G1 incident; refused")
|
||||
if check == "agent-unit" and not re.search(r"^User=" + AGENT_USER + r"\s*$", text, re.M):
|
||||
raise Refused("R18", f"{dest} does not run the agent as {AGENT_USER}")
|
||||
if check == "dropin" and not re.search(r"^\[Unit\]\s*$", text, re.M):
|
||||
raise Refused("R18", f"{dest} must keep its start limits in [Unit] (SF-3)")
|
||||
# The route must survive the bundle: a sudoers without this wrapper's line, or a wrapper without the bundle
|
||||
# mode, would cut the box off from every later bundle.
|
||||
if dest == "/etc/sudoers.d/felhom-agent" and SELF_CHECK_SUDO_LINE not in text:
|
||||
raise Refused("R18", "the new sudoers no longer lets the agent call felhom-os-apply — the bundle route would end")
|
||||
if dest == "/usr/local/sbin/felhom-os-apply" and f'BUNDLE_OP = "{BUNDLE_OP}"' not in text:
|
||||
raise Refused("R18", "the new felhom-os-apply has no bundle mode — the bundle route would end")
|
||||
|
||||
def live(self, dest):
|
||||
"""(bytes or None, mode or None, uid or None) of what the box has now."""
|
||||
if not self.r.lexists(dest):
|
||||
return None, None, None
|
||||
st = self.r.stat(dest)
|
||||
if not stat.S_ISREG(st.st_mode):
|
||||
return b"", None, None # a symlink or a directory: always replaced by the real file
|
||||
return self.r.read_bytes(dest), stat.S_IMODE(st.st_mode), st.st_uid
|
||||
|
||||
def plan_writes(self, files):
|
||||
"""Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt)."""
|
||||
oob = self.r.isdir(OOB_DIR)
|
||||
plan = []
|
||||
for dest, content, mode, check, policy in files:
|
||||
if policy == "oob" and not oob:
|
||||
plan.append((dest, content, mode, "skipped", None, None))
|
||||
continue
|
||||
old, old_mode, old_uid = self.live(dest)
|
||||
if policy == "if-absent" and old is not None:
|
||||
plan.append((dest, content, mode, "kept", old, old_mode))
|
||||
elif old == content and old_mode == mode and old_uid == 0:
|
||||
plan.append((dest, content, mode, "same", old, old_mode))
|
||||
else:
|
||||
plan.append((dest, content, mode, "write", old, old_mode))
|
||||
return plan
|
||||
|
||||
# ---------- the two entries ----------
|
||||
def from_plan(self, plan):
|
||||
"""A signed agent_config_update, from the agent (sudo, the --plan line)."""
|
||||
data = self.load_file(plan.get("bundle"))
|
||||
trust = self.a.load_trust()
|
||||
signers, bootstrap = TRUST_SIGNERS, False
|
||||
if not self.r.lexists(TRUST_SIGNERS):
|
||||
signers, bootstrap = PINNED_SIGNERS, True
|
||||
self.r.log(f"os-apply: BUNDLE {TRUST_SIGNERS} is missing — verifying against the installer's pinned key "
|
||||
f"{PINNED_OPERATOR_KEY_ID} only")
|
||||
params, nonce, exp = self.a.verify_signed(plan.get("signed"), trust, op_name=BUNDLE_OP,
|
||||
signers=None if not bootstrap else signers, burn=False)
|
||||
sha, ver = params.get("bundle_sha256"), params.get("agent_version")
|
||||
if not isinstance(sha, str) or not re.match(r"^[0-9a-f]{64}$", sha) or not isinstance(ver, str):
|
||||
raise Refused("R3", "the signed job does not pin agent_version and bundle_sha256")
|
||||
ver, files = self.parse(data, sha, ver)
|
||||
self.a.burn_nonce(nonce, exp)
|
||||
return self.install(ver, sha, files, "signed", bootstrap)
|
||||
|
||||
def from_installer(self, path, sha):
|
||||
"""The installer (root, never through sudo): the hub manifest pinned the sha; no signature."""
|
||||
try:
|
||||
data = self.r.read_bytes(path)
|
||||
except OSError as e:
|
||||
raise Refused("R1", f"cannot read the bundle: {e}")
|
||||
ver, files = self.parse(data, sha)
|
||||
return self.install(ver, sha, files, "installer", False)
|
||||
|
||||
# ---------- install, self-check, undo ----------
|
||||
def install(self, ver, sha, files, authority, bootstrap):
|
||||
log = self.r.log
|
||||
plan = self.plan_writes(files)
|
||||
counts = {k: sum(1 for p in plan if p[3] == k) for k in ("write", "same", "kept", "skipped")}
|
||||
log(f"os-apply: BUNDLE START agent={ver} sha={sha[:16]} authority={authority} files={len(plan)} "
|
||||
f"write={counts['write']} same={counts['same']} kept={counts['kept']} skipped={counts['skipped']}")
|
||||
stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(self.r.now()))
|
||||
prev = os.path.join(BUNDLE_PREV_DIR, f"{stamp}-before-{ver}")
|
||||
done = [] # (dest, old bytes or None, old mode)
|
||||
rep = {"agent_version": ver, "sha256": sha, "authority": authority, "prev_dir": prev,
|
||||
"written": [p[0] for p in plan if p[3] == "write"], "kept": [p[0] for p in plan if p[3] == "kept"],
|
||||
"skipped": [p[0] for p in plan if p[3] == "skipped"], "same": counts["same"]}
|
||||
self.report["bundle"] = rep
|
||||
try:
|
||||
for dest, content, mode, action, old, old_mode in plan:
|
||||
if action != "write":
|
||||
continue
|
||||
if old is not None:
|
||||
self.r.put_file(prev + dest, old, 0o600)
|
||||
done.append((dest, old, old_mode))
|
||||
self.r.put_file(dest, content, mode)
|
||||
log(f"os-apply: BUNDLE WROTE {dest} ({'replaced' if old is not None else 'new'})")
|
||||
self.after_install(plan)
|
||||
rep["self_check"] = self.self_check(plan)
|
||||
except (Refused, OSError, subprocess.SubprocessError) as e:
|
||||
reason = e.reason if isinstance(e, Refused) else str(e)
|
||||
log(f"os-apply: BUNDLE FAILED — {reason}; putting {len(done)} previous file(s) back")
|
||||
rep["rolled_back"] = self.undo(done)
|
||||
rep["failed"] = reason[:300]
|
||||
self.report["failed"] = {"rc": 3, "step": "bundle", "reason": reason[:300]}
|
||||
return 3
|
||||
rep["signers_created"] = False
|
||||
if bootstrap and not self.r.lexists(TRUST_SIGNERS):
|
||||
self.r.put_file(TRUST_SIGNERS, pinned_signers_line().encode(), 0o644)
|
||||
rep["signers_created"] = True
|
||||
log(f"os-apply: BUNDLE created {TRUST_SIGNERS} with exactly the pinned key {PINNED_OPERATOR_KEY_ID}")
|
||||
record = {"format": BUNDLE_FORMAT, "agent_version": ver, "bundle_sha256": sha, "authority": authority,
|
||||
"installed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(self.r.now())),
|
||||
"files": {p[0]: (sha256_hex(p[1]) if p[3] in ("write", "same") else
|
||||
(sha256_hex(p[4]) if p[3] == "kept" else "skipped")) for p in plan}}
|
||||
self.r.put_file(BUNDLE_RECORD, (json.dumps(record, indent=2, sort_keys=True) + "\n").encode(), 0o644)
|
||||
self.prune()
|
||||
log(f"os-apply: BUNDLE DONE agent={ver} written={counts['write']} same={counts['same']} "
|
||||
f"self-check=ok signers-created={rep['signers_created']}")
|
||||
return 0
|
||||
|
||||
def after_install(self, plan):
|
||||
written = {p[0] for p in plan if p[3] == "write"}
|
||||
dests = {p[0] for p in plan if p[3] != "skipped"}
|
||||
if any(d.startswith("/etc/systemd/system/") for d in written):
|
||||
self.must(["systemctl", "daemon-reload"], "systemctl daemon-reload")
|
||||
if "/etc/tmpfiles.d/felhom-privsep.conf" in written:
|
||||
self.must(["systemd-tmpfiles", "--create", "/etc/tmpfiles.d/felhom-privsep.conf"], "systemd-tmpfiles")
|
||||
# `enable --now` starts a unit that is not running and leaves a running one alone (no restart). For the crash
|
||||
# guard that runs its boot step once on a box that never had it: kernel.panic for THIS boot, as the boot unit
|
||||
# would set it (`11` §5.9); with no earlier state it is the "first" boot, never counted as an unclean one.
|
||||
if "/etc/systemd/system/felhom-crash-guard.service" in dests:
|
||||
self.must(["systemctl", "enable", "--now", "felhom-crash-guard.service", "felhom-crash-guard-check.timer"],
|
||||
"enable the crash guard")
|
||||
if "/etc/systemd/system/felhom-mgmt-watchdog.timer" in dests:
|
||||
self.must(["systemctl", "enable", "--now", "felhom-mgmt-watchdog.timer"], "enable the mgmt watchdog timer")
|
||||
|
||||
def must(self, argv, what):
|
||||
rc, out, err = self.r.host(argv, 120)
|
||||
if rc != 0:
|
||||
raise Refused("R18", f"{what} failed (rc={rc}): {(out + err).strip()[-200:]}")
|
||||
|
||||
def self_check(self, plan):
|
||||
"""After the install, before the record: the box still has a working route and working wrappers."""
|
||||
sc = {}
|
||||
self.must(["visudo", "-c"], "visudo -c (the whole sudoers)")
|
||||
sc["visudo"] = "ok"
|
||||
rc, out, err = self.r.host(["sudo", "-n", "-l", "-U", AGENT_USER], 60)
|
||||
if rc != 0 or "/usr/local/sbin/felhom-os-apply --plan" not in out:
|
||||
raise Refused("R18", f"sudo -l for {AGENT_USER} no longer lists felhom-os-apply (rc={rc})")
|
||||
sc["sudo_l"] = "felhom-os-apply listed"
|
||||
rc, out, err = self.r.host(["/usr/bin/python3", "/usr/local/sbin/felhom-os-apply", "--self-check"], 60)
|
||||
if rc != 0 or f"bundle-format={BUNDLE_FORMAT}" not in out:
|
||||
raise Refused("R18", f"the installed felhom-os-apply does not answer its self-check (rc={rc})")
|
||||
sc["os_apply"] = out.strip()[:120]
|
||||
rc, out, err = self.r.host(["/bin/sh", "/usr/local/sbin/felhom-selfupdate-guarded"], 60)
|
||||
if rc != 2 or "usage" not in (out + err):
|
||||
raise Refused("R18", f"the installed felhom-selfupdate-guarded does not answer with its usage (rc={rc})")
|
||||
sc["selfupdate"] = "usage ok"
|
||||
dests = {p[0] for p in plan if p[3] != "skipped"}
|
||||
if "/usr/local/sbin/felhom-crash-guard" in dests:
|
||||
rc, out, err = self.r.host(["/usr/local/sbin/felhom-crash-guard", "status"], 60)
|
||||
try:
|
||||
st = json.loads(out)
|
||||
except ValueError:
|
||||
st = None
|
||||
if rc != 0 or not isinstance(st, dict):
|
||||
raise Refused("R18", f"the crash guard does not answer its status (rc={rc})")
|
||||
try:
|
||||
panic = int(self.r.read_file("/proc/sys/kernel/panic").strip())
|
||||
except (OSError, ValueError):
|
||||
panic = None
|
||||
if panic != st.get("kernel_panic"):
|
||||
raise Refused("R18", f"kernel.panic is {panic}, the crash guard set {st.get('kernel_panic')}")
|
||||
sc["crash_guard"] = {"armed": st.get("armed"), "kernel_panic": panic}
|
||||
return sc
|
||||
|
||||
def undo(self, done):
|
||||
"""Put every previous copy back (newest write first); remove files the bundle created."""
|
||||
back = []
|
||||
for dest, old, old_mode in reversed(done):
|
||||
try:
|
||||
if old is None:
|
||||
self.r.remove(dest)
|
||||
else:
|
||||
self.r.put_file(dest, old, old_mode if old_mode is not None else 0o644)
|
||||
back.append(dest)
|
||||
except OSError as e:
|
||||
self.r.log(f"os-apply: BUNDLE UNDO could not restore {dest}: {e}")
|
||||
if any(d.startswith("/etc/systemd/system/") for d in back):
|
||||
self.r.host(["systemctl", "daemon-reload"], 120)
|
||||
rc, _, _ = self.r.host(["visudo", "-c"], 60)
|
||||
self.r.log(f"os-apply: BUNDLE UNDO restored={len(back)} visudo-after={'ok' if rc == 0 else 'FAILED'}")
|
||||
return back
|
||||
|
||||
def prune(self):
|
||||
names = [n for n in self.r.list_dir(BUNDLE_PREV_DIR)]
|
||||
for n in names[:-BUNDLE_KEEP]:
|
||||
self.r.rmtree(os.path.join(BUNDLE_PREV_DIR, n))
|
||||
|
||||
# ---------- the facts mode's view ----------
|
||||
def state(self):
|
||||
"""What the box runs: the record, and every bundle path's live sha256 against it (drift = changed by hand)."""
|
||||
rec = None
|
||||
try:
|
||||
rec = json.loads(self.r.read_file(BUNDLE_RECORD))
|
||||
except (OSError, ValueError):
|
||||
rec = None
|
||||
live = {}
|
||||
for dest, *_ in BUNDLE_FILES:
|
||||
try:
|
||||
data, _, _ = self.live(dest)
|
||||
except OSError:
|
||||
data = None
|
||||
live[dest] = sha256_hex(data) if data is not None else "absent"
|
||||
out = {"version": (rec or {}).get("agent_version", "none"), "live": live}
|
||||
if rec:
|
||||
out["installed_at"] = rec.get("installed_at")
|
||||
out["bundle_sha256"] = rec.get("bundle_sha256")
|
||||
want = rec.get("files") or {}
|
||||
out["drift"] = sorted(d for d, h in want.items() if h != "skipped" and live.get(d) != h)
|
||||
out["signers_present"] = self.r.lexists(TRUST_SIGNERS)
|
||||
return out
|
||||
|
||||
|
||||
def main(argv, runner=None, environ=None):
|
||||
r = runner or Runner()
|
||||
env = os.environ if environ is None else environ
|
||||
if argv[1:] == ["--self-check"]:
|
||||
# The bundle's self-check runs the NEW wrapper this way: it parses and starts. Reads nothing, changes nothing.
|
||||
print(f"felhom-os-apply ok bundle-format={BUNDLE_FORMAT} files={len(BUNDLE_FILES)}")
|
||||
return 0
|
||||
if len(argv) == 5 and argv[1] == "--install-bundle" and argv[3] == "--sha256":
|
||||
# The INSTALLER's entry (root, a fresh box). Unreachable through sudo: the sudoers line pins argv[1] to --plan,
|
||||
# and a sudo caller is refused here as well (sudo always sets SUDO_UID).
|
||||
a = Apply(r, "")
|
||||
if "SUDO_UID" in env:
|
||||
r.log("os-apply: REFUSED: R1 --install-bundle is the installer's entry, never through sudo")
|
||||
print("OSAPPLY-REPORT " + json.dumps({"refused": {"code": "R1", "reason": "install-bundle via sudo"}}))
|
||||
return 2
|
||||
sha = argv[4]
|
||||
if not re.match(r"^[0-9a-f]{64}$", sha):
|
||||
print("OSAPPLY-REPORT " + json.dumps({"refused": {"code": "R18", "reason": "sha256 is not 64 lowercase hex"}}))
|
||||
return 2
|
||||
a.report["mode"] = "bundle"
|
||||
try:
|
||||
rc = Bundle(a).from_installer(argv[2], sha)
|
||||
except Refused as e:
|
||||
r.log(f"os-apply: REFUSED: {e.code} {e.reason}")
|
||||
a.report["refused"] = {"code": e.code, "reason": e.reason}
|
||||
rc = 2
|
||||
print("OSAPPLY-REPORT " + json.dumps(a.report, sort_keys=True))
|
||||
return rc
|
||||
if len(argv) != 3 or argv[1] != "--plan":
|
||||
r.log("os-apply: REFUSED: R1 usage: felhom-os-apply --plan /var/lib/felhom-agent/os/plan-<id>.json")
|
||||
print("OSAPPLY-REPORT " + json.dumps({"refused": {"code": "R1", "reason": "usage"}}))
|
||||
|
||||
@@ -0,0 +1,545 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user