D43 = A (R-1257): the root-key-only sshd drop-in only on an appliance; on BYO nothing is written and our copy is removed; the in-force check reads the LAN addresses (X3); a failing sshd -t undoes only the SSH step (X4). Commit only, not released.
gates / gates (push) Successful in 1m32s
gates / gates (push) Successful in 1m32s
This commit is contained in:
@@ -1,3 +1,22 @@
|
||||
## Unreleased (next: v0.165.0) — D43 = A: BYO hosts keep their own SSH settings (R-1257) (2026-10-24 night) — committed, NOT released (the release + the operator's vouch are Monday's)
|
||||
|
||||
- **The root-key-only sshd drop-in (`/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf`, R-982 / D4) goes only to an
|
||||
appliance** — operator ruling D43 = A (2026-10-24): a BYO host belongs to its owner (`11` §1). The table entry's policy
|
||||
is now `appliance`: written only when the ROOT-OWNED install record (`/var/lib/felhom-install/state.json`) says
|
||||
`appliance`. On a `byo` record nothing is written, and a copy WE wrote (this exact path, beginning with our header
|
||||
line) is REMOVED, its previous copy kept for the undo, then `sshd -t` and a reload. A box whose record is absent or
|
||||
agent-writable proves nothing and is left exactly as it is. The bundle report lists `removed`; the record says
|
||||
`removed` / `skipped`. The installer records the mode before it installs the bundle, so a new appliance still gets it.
|
||||
- **X3: the „in force" check asks sshd about each of the host's own LAN addresses** (`ip -4 -o addr show scope
|
||||
global`), not only 192.0.2.1 — a `Match Address <LAN>` block that re-allows root passwords no longer reads as in force.
|
||||
- **X4: a failing `sshd -t` puts back ONLY the SSH step** (`undo_sshd_step`: the previous copy, or none; sshd is not
|
||||
reloaded, so it never ran the refused state) and the rest of the bundle stays; it used to refuse and undo the whole
|
||||
bundle. The report carries `sshd_step_undone`; the record names what is on disk.
|
||||
- Tests (`configs/test_felhom_config_bundle.py`, class `ByoKeepsItsSSH`, 7 tests; each rule seen failing with it taken
|
||||
out): a BYO host gets no drop-in; our copy comes off a BYO host; the owner's own file stays; no record → left as it
|
||||
is; a LAN `Match` block is not in force; every address key-only passes; a failing `sshd -t` undoes only the SSH step.
|
||||
The fake box now carries an appliance record by default (every earlier test unchanged).
|
||||
|
||||
## 2026-10-24 (review round 2, unattended) — `.claude/rules/unprompted-work.md` aligned with its four siblings
|
||||
|
||||
- The copy lacked item 6's paragraph „A version number is pushed once …" that the felhom.eu and felhom-controller copies
|
||||
|
||||
+103
-16
@@ -168,7 +168,10 @@ KERNEL_ACTIVE = ("staged", "oneshot", "judging", "reverting")
|
||||
# agent-unit → User=felhom-agent; dropin → a [Unit] section (SF-3); nft → nft -c -f; plain → none;
|
||||
# sshd → `sshd -t -f` on the snippet alone (R-982).
|
||||
# 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).
|
||||
# have tuned); oob → only on a box with the OOB belt (/etc/felhom-sshd exists — the installer's --no-oob leaves none);
|
||||
# appliance → only on a box whose ROOT-OWNED install record says appliance (D43 = A, operator 2026-10-24: a BYO
|
||||
# host keeps its own settings, `11` §1). On a BYO box nothing is written and a copy WE wrote (this exact path, our
|
||||
# first line) is removed; a box with no readable record is left exactly as it is (it cannot prove which it is).
|
||||
# Order matters: the sudoers files come LAST, so a referenced wrapper is in place before the line that allows it.
|
||||
# R-861 (agent v0.146.0): the signed agent update, verified here (mode agent_update), then the A/B flip.
|
||||
SELFUPDATE_OP = "agent_update"
|
||||
@@ -187,6 +190,8 @@ BUNDLE_RE = re.compile(r"^bundle-[0-9]+\.[0-9]+\.[0-9]+(-[0-9A-Za-z.]+)?\.json$"
|
||||
OOB_DIR = "/etc/felhom-sshd"
|
||||
# R-982: the host's own sshd drop-in (after_install reloads sshd for it; undo reloads after putting it back).
|
||||
SSHD_DROPIN = "/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf"
|
||||
# D43 = A (R-1257): the first line every copy of the drop-in we ship carries — what proves a file on a BYO host is ours.
|
||||
SSHD_DROPIN_HEADER = b"# MANAGED BY FELHOM (the agent's config bundle) \xe2\x80\x94 R-982"
|
||||
|
||||
BUNDLE_FILES = [
|
||||
("/usr/local/sbin/felhom-mkfs-guarded", "felhom-mkfs-guarded.sh", 0o755, "bash", "replace"),
|
||||
@@ -227,7 +232,8 @@ BUNDLE_FILES = [
|
||||
("/etc/systemd/system/felhom-oob-nft.service", "felhom-oob-nft.service", 0o644, "unit", "oob"),
|
||||
# R-982, operator ruling D4 (2026-10-22): the host's own sshd (port 22, the LAN): root by key only. The console
|
||||
# password stays; felhom-sshd (the operator's door) reads nothing from sshd_config.d. Reloaded in after_install.
|
||||
("/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf", "felhom-root-ssh-key-only.conf", 0o644, "sshd", "replace"),
|
||||
# D43 = A (operator 2026-10-24, R-1257): appliances only — a BYO host keeps its own SSH settings (`11` §1).
|
||||
("/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf", "felhom-root-ssh-key-only.conf", 0o644, "sshd", "appliance"),
|
||||
("/etc/sudoers.d/felhom-op", "felhom-op.sudoers", 0o440, "sudoers", "oob"),
|
||||
("/etc/sudoers.d/felhom-agent", "felhom-agent.sudoers", 0o440, "sudoers", "replace"),
|
||||
]
|
||||
@@ -1873,6 +1879,7 @@ class Bundle:
|
||||
self.a, self.r = apply, apply.r
|
||||
self.report = apply.report
|
||||
self.sshd_dropin_not_in_force = False
|
||||
self.sshd_step_undone = None # X4 (R-1257): why the SSH step alone was put back ("" = not), the rest stays
|
||||
|
||||
# ---------- reading and checking ----------
|
||||
def load_file(self, path):
|
||||
@@ -1982,14 +1989,40 @@ class Bundle:
|
||||
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 install_mode(self):
|
||||
"""The box's ROOT-OWNED install record's mode ("appliance", "byo", …), or None when it proves nothing (absent,
|
||||
agent-writable, unreadable) — the same proof check_appliance asks for the host layer (R12)."""
|
||||
try:
|
||||
st = self.r.stat(INSTALL_STATE)
|
||||
if st.st_uid != 0 or (st.st_mode & 0o022):
|
||||
return None
|
||||
mode = json.loads(self.r.read_file(INSTALL_STATE)).get("mode")
|
||||
except (OSError, ValueError, AttributeError):
|
||||
return None
|
||||
return mode if isinstance(mode, str) and mode else None
|
||||
|
||||
def plan_writes(self, files):
|
||||
"""Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt)."""
|
||||
"""Decide per file: write / same / kept (if-absent and present) / skipped (oob on a box without the belt; an
|
||||
appliance-only file on a box that is not one) / remove (an appliance-only file WE wrote, on a BYO box)."""
|
||||
oob = self.r.isdir(OOB_DIR)
|
||||
install_mode = self.install_mode()
|
||||
plan = []
|
||||
for dest, content, mode, check, policy in files:
|
||||
if policy == "oob" and not oob:
|
||||
plan.append((dest, content, mode, "skipped", None, None))
|
||||
continue
|
||||
if policy == "appliance" and install_mode != "appliance":
|
||||
old, old_mode, _ = self.live(dest)
|
||||
ours = old is not None and old.startswith(SSHD_DROPIN_HEADER)
|
||||
if install_mode == "byo" and ours:
|
||||
self.r.log(f"os-apply: BUNDLE {dest}: a BYO host keeps its own settings (D43) — removing the copy we wrote")
|
||||
plan.append((dest, content, mode, "remove", old, old_mode))
|
||||
else:
|
||||
why = "a BYO host keeps its own settings (D43)" if install_mode == "byo" else \
|
||||
f"the install record proves no appliance (mode={install_mode!r}) — left as it is"
|
||||
self.r.log(f"os-apply: BUNDLE {dest} skipped: {why}")
|
||||
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))
|
||||
@@ -2031,7 +2064,7 @@ class Bundle:
|
||||
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")}
|
||||
counts = {k: sum(1 for p in plan if p[3] == k) for k in ("write", "same", "kept", "skipped", "remove")}
|
||||
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()))
|
||||
@@ -2039,10 +2072,17 @@ class Bundle:
|
||||
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"]}
|
||||
"skipped": [p[0] for p in plan if p[3] == "skipped"], "same": counts["same"],
|
||||
"removed": [p[0] for p in plan if p[3] == "remove"]}
|
||||
self.report["bundle"] = rep
|
||||
try:
|
||||
for dest, content, mode, action, old, old_mode in plan:
|
||||
if action == "remove": # D43 = A: our copy off a BYO host (the previous copy kept, the undo restores it)
|
||||
self.r.put_file(prev + dest, old, 0o600)
|
||||
done.append((dest, old, old_mode))
|
||||
self.r.remove(dest)
|
||||
log(f"os-apply: BUNDLE REMOVED {dest} (ours, on a BYO host — D43)")
|
||||
continue
|
||||
if action != "write":
|
||||
continue
|
||||
if old is not None:
|
||||
@@ -2062,22 +2102,56 @@ class Bundle:
|
||||
rep["signers_created"] = False
|
||||
if self.sshd_dropin_not_in_force:
|
||||
rep["sshd_dropin_not_in_force"] = True # R-982 A2: written and reloaded, but sshd -T shows root passwords allowed
|
||||
if self.sshd_step_undone:
|
||||
rep["sshd_step_undone"] = self.sshd_step_undone[:300] # X4: only the SSH step was put back
|
||||
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}}
|
||||
"files": {p[0]: self.record_entry(p) 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 record_entry(self, p):
|
||||
"""The record's word for one plan entry: the sha it installed, the sha it kept, or what it did instead."""
|
||||
dest, content, _, action, old, _ = p
|
||||
if dest == SSHD_DROPIN and self.sshd_step_undone:
|
||||
return sha256_hex(old) if old is not None else "absent" # X4: the SSH step was put back
|
||||
if action in ("write", "same"):
|
||||
return sha256_hex(content)
|
||||
if action == "kept":
|
||||
return sha256_hex(old)
|
||||
return "removed" if action == "remove" else "skipped"
|
||||
|
||||
def lan_addresses(self):
|
||||
"""X3 (R-1257): the host's own global IPv4 addresses — the ones a LAN client reaches sshd on."""
|
||||
rc, out, _ = self.r.host(["ip", "-4", "-o", "addr", "show", "scope", "global"], 30)
|
||||
if rc != 0:
|
||||
return []
|
||||
return sorted(set(re.findall(r"\binet (\d+\.\d+\.\d+\.\d+)/", out)))
|
||||
|
||||
def undo_sshd_step(self, plan, why):
|
||||
"""X4 (R-1257): `sshd -t` refused the host's whole config with our step in it — put back ONLY that file (the
|
||||
previous copy, or none) and keep the rest of the bundle. sshd was not reloaded, so it never ran the refused state."""
|
||||
for dest, _, _, action, old, old_mode in plan:
|
||||
if dest != SSHD_DROPIN or action not in ("write", "remove"):
|
||||
continue
|
||||
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)
|
||||
self.sshd_step_undone = why
|
||||
self.r.log(f"os-apply: BUNDLE WARNING sshd -t refused the host's config with the SSH step — ONLY that step is put "
|
||||
f"back, the rest of the bundle stays: {why[:200]} (R-1257)")
|
||||
|
||||
def after_install(self, plan):
|
||||
written = {p[0] for p in plan if p[3] == "write"}
|
||||
removed = {p[0] for p in plan if p[3] == "remove"}
|
||||
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")
|
||||
@@ -2098,16 +2172,29 @@ class Bundle:
|
||||
# felhom-sshd is not touched. Then the POSITIVE check: sshd's own effective view for root. A box whose
|
||||
# sshd_config has no Include of sshd_config.d, an earlier drop-in or a Match block can keep the password route
|
||||
# — that is said loudly in the log and the report, never silently recorded as done (A2).
|
||||
if SSHD_DROPIN in written:
|
||||
self.must(["/usr/sbin/sshd", "-t"], "sshd -t (the host's whole sshd config)")
|
||||
# R-1257 X4: a failing `sshd -t` puts back ONLY the SSH step (undo_sshd_step); it used to refuse and undo the
|
||||
# whole bundle. X3: the in-force check asks sshd about each of the host's own LAN addresses — a probe from a
|
||||
# documentation address (192.0.2.1) read a `Match Address <LAN>` block that re-allows passwords as in force.
|
||||
if SSHD_DROPIN in written or SSHD_DROPIN in removed:
|
||||
rc, out, err = self.r.host(["/usr/sbin/sshd", "-t"], 120)
|
||||
if rc != 0:
|
||||
self.undo_sshd_step(plan, f"sshd -t failed (rc={rc}): {(out + err).strip()[-200:]}")
|
||||
return
|
||||
self.must(["systemctl", "try-reload-or-restart", "ssh"], "systemctl try-reload-or-restart ssh")
|
||||
rc, out, err = self.r.host(["/usr/sbin/sshd", "-T", "-C", "user=root,host=felhom-check,addr=192.0.2.1"], 60)
|
||||
eff = re.findall(r"^permitrootlogin\s+(\S+)", out, re.M | re.I)
|
||||
if rc != 0 or eff != ["without-password"]:
|
||||
self.r.log(f"os-apply: BUNDLE WARNING the root-key-only drop-in is NOT in force: sshd -T says "
|
||||
f"permitrootlogin={eff or '?'} (rc={rc}) — check sshd_config's Include, an earlier drop-in "
|
||||
f"or a Match block (R-982)")
|
||||
self.sshd_dropin_not_in_force = True
|
||||
if SSHD_DROPIN in written:
|
||||
addrs = self.lan_addresses()
|
||||
if not addrs:
|
||||
self.r.log("os-apply: BUNDLE WARNING no global IPv4 address read for the sshd in-force check — "
|
||||
"probing 192.0.2.1 only (R-1257)")
|
||||
addrs = ["192.0.2.1"]
|
||||
for addr in addrs:
|
||||
rc, out, err = self.r.host(["/usr/sbin/sshd", "-T", "-C", f"user=root,host=felhom-check,addr={addr}"], 60)
|
||||
eff = re.findall(r"^permitrootlogin\s+(\S+)", out, re.M | re.I)
|
||||
if rc != 0 or eff != ["without-password"]:
|
||||
self.r.log(f"os-apply: BUNDLE WARNING the root-key-only drop-in is NOT in force for {addr}: sshd -T "
|
||||
f"says permitrootlogin={eff or '?'} (rc={rc}) — check sshd_config's Include, an earlier "
|
||||
f"drop-in or a Match block (R-982, R-1257)")
|
||||
self.sshd_dropin_not_in_force = True
|
||||
|
||||
def must(self, argv, what):
|
||||
rc, out, err = self.r.host(argv, 120)
|
||||
|
||||
@@ -45,8 +45,10 @@ class St:
|
||||
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):
|
||||
def __init__(self, bundle_bytes, signed, oob=False, signers=True, install_mode="appliance"):
|
||||
self.files, self.modes, self.uids = {}, {}, {}
|
||||
if install_mode is not None: # the installer's ROOT-OWNED record (D43 = A reads it for the sshd drop-in)
|
||||
self.put(osapply.INSTALL_STATE, json.dumps({"completed": [], "mode": install_mode}).encode(), 0o644)
|
||||
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:
|
||||
@@ -513,6 +515,116 @@ class RootKeyOnly(unittest.TestCase):
|
||||
reloads = [i for i, c in enumerate(box.calls) if c == ["systemctl", "try-reload-or-restart", "ssh"]]
|
||||
self.assertEqual(len(reloads), 2, box.calls) # after_install, then the undo
|
||||
|
||||
class ByoKeepsItsSSH(unittest.TestCase):
|
||||
"""D43 = A (operator 2026-10-24, R-1257): the root-key-only drop-in goes ONLY to a box whose root-owned install record
|
||||
says appliance; on a BYO host nothing is written and a copy WE wrote is removed; a box with no record is left as it
|
||||
is. X3: the in-force check asks sshd about each of the host's own LAN addresses. X4: a failing `sshd -t` puts back
|
||||
only the SSH step. RED-PROOF (2026-10-24, felhom-agent f5c3d77 + these tests): with the drop-in's policy back to
|
||||
`replace`, test_a_byo_host_gets_no_dropin and test_our_copy_comes_off_a_byo_host fail; with the probe back to
|
||||
192.0.2.1 only, test_a_match_block_for_the_lan_is_not_in_force fails; with `sshd -t` back to `must`,
|
||||
test_a_failing_sshd_t_undoes_only_the_ssh_step fails (rc 3, the whole bundle undone)."""
|
||||
DEST = "/etc/ssh/sshd_config.d/10-felhom-root-key-only.conf"
|
||||
OURS = (REPO / "configs" / "felhom-root-ssh-key-only.conf").read_bytes()
|
||||
|
||||
def test_a_byo_host_gets_no_dropin(self):
|
||||
data, sha = real_bundle()
|
||||
box = Box(data, signed_job(sha), install_mode="byo")
|
||||
rc, rep = run(box)
|
||||
self.assertEqual(rc, 0, rep)
|
||||
self.assertNotIn(self.DEST, box.files)
|
||||
self.assertNotIn(["/usr/sbin/sshd", "-t"], box.calls)
|
||||
self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
|
||||
self.assertIn(self.DEST, rep["bundle"]["skipped"])
|
||||
rec = json.loads(box.files[osapply.BUNDLE_RECORD])
|
||||
self.assertEqual(rec["files"][self.DEST], "skipped")
|
||||
|
||||
def test_our_copy_comes_off_a_byo_host(self):
|
||||
data, sha = real_bundle()
|
||||
box = Box(data, signed_job(sha), install_mode="byo")
|
||||
box.put(self.DEST, self.OURS, 0o644) # an earlier bundle wrote it (before D43)
|
||||
rc, rep = run(box)
|
||||
self.assertEqual(rc, 0, rep)
|
||||
self.assertNotIn(self.DEST, box.files)
|
||||
self.assertEqual(rep["bundle"]["removed"], [self.DEST])
|
||||
self.assertTrue(any(k.endswith(self.DEST) and k.startswith(osapply.BUNDLE_PREV_DIR) for k in box.files),
|
||||
"the previous copy must be kept for the undo")
|
||||
self.assertLess(box.calls.index(["/usr/sbin/sshd", "-t"]), box.calls.index(["systemctl", "try-reload-or-restart", "ssh"]))
|
||||
self.assertEqual(json.loads(box.files[osapply.BUNDLE_RECORD])["files"][self.DEST], "removed")
|
||||
|
||||
def test_the_owners_own_file_stays_on_a_byo_host(self):
|
||||
data, sha = real_bundle()
|
||||
box = Box(data, signed_job(sha), install_mode="byo")
|
||||
box.put(self.DEST, b"# my own\nPermitRootLogin yes\n", 0o644)
|
||||
rc, rep = run(box)
|
||||
self.assertEqual(rc, 0, rep)
|
||||
self.assertEqual(box.files[self.DEST], b"# my own\nPermitRootLogin yes\n")
|
||||
self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
|
||||
|
||||
def test_a_box_with_no_record_is_left_as_it_is(self):
|
||||
for setup in ("none", "agent-writable"):
|
||||
data, sha = real_bundle()
|
||||
box = Box(data, signed_job(sha), install_mode=None if setup == "none" else "byo")
|
||||
if setup == "agent-writable":
|
||||
box.modes[osapply.INSTALL_STATE] = 0o666 # proves nothing
|
||||
box.put(self.DEST, self.OURS, 0o644)
|
||||
rc, rep = run(box)
|
||||
self.assertEqual(rc, 0, rep)
|
||||
self.assertEqual(box.files.get(self.DEST), self.OURS, setup)
|
||||
self.assertIn(self.DEST, rep["bundle"]["skipped"], setup)
|
||||
|
||||
def _fake(self, box, addrs=("192.168.0.50",), match=None, sshd_t_rc=0):
|
||||
orig = box.host
|
||||
def host(argv, timeout=600, stdin=None):
|
||||
if argv[:3] == ["ip", "-4", "-o"]:
|
||||
box.calls.append(argv)
|
||||
return 0, "".join(f"2: vmbr0 inet {a}/24 brd x scope global vmbr0\\ valid_lft forever\n" for a in addrs), ""
|
||||
if argv[:2] == ["/usr/sbin/sshd", "-T"]:
|
||||
box.calls.append(argv)
|
||||
val = "yes" if match and f"addr={match}" in argv[-1] else "without-password"
|
||||
return 0, f"port 22\npermitrootlogin {val}\n", ""
|
||||
if argv == ["/usr/sbin/sshd", "-t"]:
|
||||
box.calls.append(argv)
|
||||
return (sshd_t_rc, "", "/etc/ssh/sshd_config line 9: Bad configuration option") if sshd_t_rc else (0, "", "")
|
||||
return orig(argv, timeout, stdin)
|
||||
box.host = host
|
||||
|
||||
def test_a_match_block_for_the_lan_is_not_in_force(self):
|
||||
data, sha = real_bundle()
|
||||
box = Box(data, signed_job(sha))
|
||||
self._fake(box, addrs=("192.168.0.50", "10.77.0.4"), match="192.168.0.50")
|
||||
rc, rep = run(box)
|
||||
self.assertEqual(rc, 0, rep)
|
||||
probed = [c[-1] for c in box.calls if c[:2] == ["/usr/sbin/sshd", "-T"]]
|
||||
self.assertTrue(any("addr=192.168.0.50" in p for p in probed) and any("addr=10.77.0.4" in p for p in probed), probed)
|
||||
self.assertFalse(any("addr=192.0.2.1" in p for p in probed), probed)
|
||||
self.assertTrue(rep["bundle"].get("sshd_dropin_not_in_force"), rep)
|
||||
self.assertTrue(any("NOT in force for 192.168.0.50" in l for l in box.logs), box.logs)
|
||||
|
||||
def test_the_lan_check_passes_when_every_address_is_key_only(self):
|
||||
data, sha = real_bundle()
|
||||
box = Box(data, signed_job(sha))
|
||||
self._fake(box, addrs=("192.168.0.50", "10.77.0.4"))
|
||||
rc, rep = run(box)
|
||||
self.assertEqual(rc, 0, rep)
|
||||
self.assertNotIn("sshd_dropin_not_in_force", rep["bundle"])
|
||||
|
||||
def test_a_failing_sshd_t_undoes_only_the_ssh_step(self):
|
||||
for before in (None, b"# MANAGED BY FELHOM (the agent's config bundle) \xe2\x80\x94 R-982 old\nPermitRootLogin prohibit-password\n"):
|
||||
data, sha = real_bundle()
|
||||
box = Box(data, signed_job(sha))
|
||||
if before is not None:
|
||||
box.put(self.DEST, before, 0o644)
|
||||
self._fake(box, sshd_t_rc=255)
|
||||
rc, rep = run(box)
|
||||
self.assertEqual(rc, 0, rep) # the bundle stays
|
||||
self.assertEqual(box.files.get(self.DEST), before, "only the SSH step is put back")
|
||||
self.assertIn("/usr/local/sbin/felhom-os-apply", box.files, "the rest of the bundle stays")
|
||||
self.assertIn("sshd -t failed", rep["bundle"].get("sshd_step_undone", ""), rep)
|
||||
self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls, "sshd never ran the refused state")
|
||||
want = "absent" if before is None else hashlib.sha256(before).hexdigest()
|
||||
self.assertEqual(json.loads(box.files[osapply.BUNDLE_RECORD])["files"][self.DEST], want)
|
||||
|
||||
|
||||
class TrustBootstrap(unittest.TestCase):
|
||||
def test_missing_signers_verifies_against_the_pinned_key_and_creates_it(self):
|
||||
data, sha = real_bundle()
|
||||
|
||||
Reference in New Issue
Block a user