R-1285: after every bundle install the agent asks the wrapper now on disk for the appliance-only policy pass, so a BYO host upgraded from an older wrapper loses our SSH drop-in in one update. Red-proved. Commit only, not released.

configs/felhom-os-apply

@@ -491,7 +491,8 @@ class Apply:
 
     def check_plan(self, plan):
         mode = plan.get("mode", "apply")
-        if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle", "agent_update", "oom-check") \
+        if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle", "bundle-policy", "agent_update",
+                        "oom-check") \
                 + KERNEL_MODES:
             raise Refused("R11", f"unknown mode {mode!r}")
         if mode == "agent_update":
@@ -502,6 +503,10 @@ class Apply:
             if plan.get("layer") != "host":
                 raise Refused("R11", "bundle is a host-layer mode")
             return mode, "host", 0, "bundle"
+        if mode == "bundle-policy":
+            if plan.get("layer") != "host":
+                raise Refused("R11", "bundle-policy is a host-layer mode")
+            return mode, "host", 0, "bundle-policy"
         layer = plan.get("layer")
         if layer not in ("guest", "host", "docker", "pve", "kernel"):
             raise Refused("R12", f"layer {layer!r} is not guest, host, docker, pve or kernel")
@@ -1074,6 +1079,8 @@ class Apply:
             return self.facts()
         if self.mode == "bundle":
             return Bundle(self).from_plan(plan)
+        if self.mode == "bundle-policy":
+            return Bundle(self).policy_pass()
         if self.mode == "agent_update":
             return self.agent_update(plan)
         if self.layer in ("host", "pve", "kernel"):
@@ -2064,6 +2071,42 @@ class Bundle:
         ver, files = self.parse(data, sha)
         return self.install(ver, sha, files, "installer", False)
 
+    def policy_pass(self):
+        """R-1285 (D43 = A): the appliance-only files' policy, applied by THIS wrapper. A bundle that brings a new wrapper
+        is planned by the OLD one, which did not know the policy — so on a BYO host upgraded from 0.164.0 our SSH drop-in
+        stayed until the next bundle. The agent runs this pass right after every bundle install (the wrapper on disk is
+        then the new one). It only REMOVES a file WE wrote (our header) from a box whose root-owned install record says
+        byo, keeps the previous copy beside the bundle's, and reloads sshd; it writes nothing, so it needs no signature.
+        A failing `sshd -t` puts the file back (R-1257 X4's rule). Report key `bundle_policy`."""
+        log = self.r.log
+        install_mode = self.install_mode()
+        removed = []
+        if install_mode == "byo":
+            stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime(self.r.now()))
+            prev = os.path.join(BUNDLE_PREV_DIR, f"{stamp}-policy")
+            for dest, _src, _mode, _check, policy in BUNDLE_FILES:
+                if policy != "appliance":
+                    continue
+                old, old_mode, _ = self.live(dest)
+                if old is None or not old.startswith(SSHD_DROPIN_HEADER):
+                    continue
+                self.r.put_file(prev + dest, old, 0o600)
+                self.r.remove(dest)
+                removed.append((dest, old, old_mode))
+                log(f"os-apply: BUNDLE POLICY REMOVED {dest} (ours, on a BYO host — D43, R-1285)")
+            if any(d == SSHD_DROPIN for d, _, _ in removed):
+                rc, out, err = self.r.host(["/usr/sbin/sshd", "-t"], 120)
+                if rc != 0:
+                    for d, old, old_mode in removed:
+                        self.r.put_file(d, old, old_mode)
+                    log(f"os-apply: BUNDLE POLICY WARNING sshd -t failed without our drop-in (rc={rc}) — put back")
+                    removed = []
+                else:
+                    self.must(["systemctl", "try-reload-or-restart", "ssh"], "systemctl try-reload-or-restart ssh")
+        self.report["bundle_policy"] = {"install_mode": install_mode, "removed": [d for d, _, _ in removed]}
+        log(f"os-apply: BUNDLE POLICY DONE install_mode={install_mode} removed={len(removed)}")
+        return 0
+
     # ---------- install, self-check, undo ----------
     def install(self, ver, sha, files, authority, bootstrap):
         log = self.r.log

configs/test_felhom_config_bundle.py

@@ -588,6 +588,47 @@ class ByoKeepsItsSSH(unittest.TestCase):
             return orig(argv, timeout, stdin)
         box.host = host
 
+    # R-1285: the policy pass the agent runs right after every bundle install, with the wrapper NOW on disk. No
+    # signature (it only removes a file we wrote). RED-PROOF: make policy_pass return before the loop → our copy stays.
+    POLICY_PLAN = {"release_id": "bundle-policy-0.166.0", "layer": "host", "mode": "bundle-policy"}
+
+    def test_the_policy_pass_removes_our_copy_from_a_byo_host(self):
+        data, sha = real_bundle()
+        box = Box(data, None, install_mode="byo")
+        box.put(PLAN, json.dumps(self.POLICY_PLAN).encode(), 0o600, uid=999)
+        box.put(self.DEST, self.OURS, 0o644)   # the old wrapper's bundle run left it
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertNotIn(self.DEST, box.files)
+        self.assertEqual(rep["bundle_policy"], {"install_mode": "byo", "removed": [self.DEST]}, rep)
+        self.assertTrue(any(k.endswith(self.DEST) and k.startswith(osapply.BUNDLE_PREV_DIR) for k in box.files))
+        self.assertLess(box.calls.index(["/usr/sbin/sshd", "-t"]), box.calls.index(["systemctl", "try-reload-or-restart", "ssh"]))
+
+    def test_the_policy_pass_leaves_an_appliance_and_the_owners_file(self):
+        for mode, content in (("appliance", None), ("byo", b"# my own\nPermitRootLogin yes\n")):
+            data, sha = real_bundle()
+            box = Box(data, None, install_mode=mode)
+            box.put(PLAN, json.dumps(self.POLICY_PLAN).encode(), 0o600, uid=999)
+            body = content if content is not None else self.OURS
+            box.put(self.DEST, body, 0o644)
+            rc, rep = run(box)
+            self.assertEqual(rc, 0, rep)
+            self.assertEqual(box.files.get(self.DEST), body, mode)
+            self.assertEqual(rep["bundle_policy"]["removed"], [], mode)
+            self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls, mode)
+
+    def test_the_policy_pass_puts_our_copy_back_when_sshd_t_fails(self):
+        data, sha = real_bundle()
+        box = Box(data, None, install_mode="byo")
+        box.put(PLAN, json.dumps(self.POLICY_PLAN).encode(), 0o600, uid=999)
+        box.put(self.DEST, self.OURS, 0o644)
+        self._fake(box, sshd_t_rc=1)
+        rc, rep = run(box)
+        self.assertEqual(rc, 0, rep)
+        self.assertEqual(box.files.get(self.DEST), self.OURS)
+        self.assertEqual(rep["bundle_policy"]["removed"], [])
+        self.assertNotIn(["systemctl", "try-reload-or-restart", "ssh"], box.calls)
+
     def test_a_match_block_for_the_lan_is_not_in_force(self):
         data, sha = real_bundle()
         box = Box(data, signed_job(sha))