v0.144.0 code: R8 measures the real download (R-865); an OS pass's report survives a killed agent (R-868); the debug pass runs from the saved block when the hub is away (R-866)

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
2026-10-05 07:14:32 +02:00
parent c8d12f1f2a
commit ca78c17b29
10 changed files with 662 additions and 28 deletions
@@ -73,6 +73,7 @@ class Fake:
        self.sig_rc = 0
        self.nonces = {}
        self.clock = 1791115200.0  # 2026-10-04T12:00:00Z
+        self.saved_reports = []  # R-868: (plan path, report) the wrapper kept on disk
        self.files[osapply.TRUST_FILE] = json.dumps({"host_id": "demo-hp-bb76ea", "ring0_slow_lane": False})
        self.stats[osapply.TRUST_FILE] = St(mode=statmod.S_IFREG | 0o644, uid=0)
        self.files[osapply.TRUST_SIGNERS] = 'felhom-op-1 namespaces="felhom-op-v1" ssh-ed25519 AAAA\n'
@@ -113,6 +114,10 @@ class Fake:
    def log(self, line):
        self.logs.append(line)

+    def save_report(self, plan_path, report):
+        self.saved_reports.append((plan_path, json.loads(json.dumps(report))))
+        return plan_path.replace("/plan-", "/report-")
+
    def host(self, argv, timeout=600, stdin=None):
        self.calls.append(("host", argv))
        if argv[0] == "/usr/sbin/pct" and argv[1] == "status":
@@ -168,8 +173,8 @@ class Fake:
            if "-f" in a:
                self.dpkg_audit = ""
                return 0, "Setting up x (1) ...\n" if getattr(self, "repaired", False) else "", ""
-            if "-s" in a:
-                return self.sim(a)
+            if "--print-uris" in a or "-s" in a:
+                return self.sim(a)  # --print-uris prints and installs nothing, with or without -s (9202, 2026-10-05)
            if "install" in a:
                if self.install_rc:
                    return self.install_rc, "", "E: boom"
@@ -232,7 +237,14 @@ class Fake:

    def sim(self, a):
        if "--print-uris" in a:
-            return 0, "'http://x/libc6.deb' libc6.deb 4000000 SHA256:x\n", ""
+            if "-s" in a:
+                # real apt (measured 9202 2026-10-05): with -s it prints the SIMULATION, no URI list
+                return 0, "Inst libc6 [2.41-12+deb13u4] (2.41-12+deb13u4 Debian:13.7/stable [amd64])\n", ""
+            # real apt's line shape, verbatim from 9202 2026-10-05 (audits/night-fixes-2026-10-05/partC/)
+            return 0, ("Need to get 4347 kB of archives.\n"
+                       "'http://deb.debian.org/debian/pool/main/b/bash/bash_5.2.37-2%2bb10_amd64.deb' bash_5.2.37-2+b10_amd64.deb 1500792 MD5Sum:27b11721fea83d73b96e0f7023863771\n"
+                       "'http://deb.debian.org/debian/pool/main/g/glibc/libc6_2.41-12%2bdeb13u4_amd64.deb' libc6_2.41-12+deb13u4_amd64.deb 2846580 MD5Sum:5559581916477ef1f57ea9f82cecf22e\n"
+                       + getattr(self, "extra_uris", "")), ""
        if "dist-upgrade" in a:
            if getattr(self, "pending_sim", None) is not None and not getattr(self, "_pending_used", False):
                self._pending_used = True
@@ -273,7 +285,7 @@ class Happy(unittest.TestCase):
        self.assertEqual(rep["pending"][0]["name"], "bash")
        self.assertTrue(any(l.startswith("os-apply: REPAIR ") for l in f.logs), "the repair line must always print")
        self.assertTrue(any(l.startswith("os-apply: DONE rc=0") for l in f.logs))
-        inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2]]
+        inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2] and "--print-uris" not in c[2]]
        self.assertTrue(inst and "Dpkg::Options::=--force-confold" in inst[0][2], "must keep existing config files")

    def test_already_current_is_a_no_op(self):
@@ -343,7 +355,7 @@ class Refusals(unittest.TestCase):
        self.assertEqual(rc, 2, rep)
        self.assertEqual(rep["refused"]["code"], code, rep)
        self.assertTrue(any(l.startswith(f"os-apply: REFUSED: {code} ") for l in f.logs), f.logs)
-        inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2]]
+        inst = [c for c in f.calls if c[0] == "guest" and "install" in c[2] and "-s" not in c[2] and "-f" not in c[2] and "--print-uris" not in c[2]]
        self.assertEqual(inst, [], "a refusal must install nothing")
        return rep

@@ -454,6 +466,24 @@ class Refusals(unittest.TestCase):
        f.free = 100 * 1024 * 1024
        self.refused(f, "R8")

+    # R-865: the download is the real one. 2 GB of URIs, 5 GB free: 5 GB < 3 x 2 GB -> R8, with the size in the line.
+    # COMPANION RED-PROOF: put "-s" back into download_bytes -> 0 B -> no refusal -> this test fails.
+    def test_R8_counts_the_real_download(self):
+        f = Fake()
+        f.free = 5 * 1024 ** 3
+        f.extra_uris = "'http://deb.debian.org/debian/pool/main/b/big/big_1_amd64.deb' big_1_amd64.deb 2000000000 MD5Sum:x\n"
+        rep = self.refused(f, "R8")
+        self.assertIn("download 2004347372 B", str(rep))
+
+    def test_download_bytes_never_simulates(self):
+        f = Fake()
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        calls = [c[2] for c in f.calls if c[0] == "guest" and "--print-uris" in c[2]]
+        self.assertTrue(calls, "download_bytes was never called")
+        for c in calls:
+            self.assertNotIn("-s", c, f"--print-uris with -s prints no URIs: {c}")
+
    def test_R9_guest_locked_by_a_backup(self):
        f = Fake()
        f.files["/etc/pve/lxc/9201.conf"] = CONF_OK + "lock: backup\n"
@@ -962,3 +992,88 @@ class RealSignatureCheck(unittest.TestCase):

 if __name__ == "__main__":
    unittest.main()
+
+
+class UnsentReport(unittest.TestCase):
+    """R-868 (v0.144.0): an apply pass keeps its report on disk until the agent has sent it.
+    COMPANION RED-PROOF: drop the r.save_report call in main() -> test_apply_keeps_a_copy fails."""
+
+    def test_apply_keeps_a_copy_with_the_agents_ids(self):
+        f = Fake()
+        f.plan.update(run_id="20261005T0257-ab12", trigger="debug", ring=0)
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        self.assertEqual(len(f.saved_reports), 1, "an apply pass must keep its report on disk")
+        path, saved = f.saved_reports[0]
+        self.assertEqual(path, PLAN)
+        self.assertEqual(saved, rep, "the copy is the report the agent would have read")
+        self.assertEqual((saved["run_id"], saved["trigger"], saved["ring"]), ("20261005T0257-ab12", "debug", 0))
+
+    def test_a_refusal_is_kept_too(self):
+        f = Fake()
+        f.free = 1
+        rc, rep = run(f)
+        self.assertEqual(rc, 2)
+        self.assertEqual(f.saved_reports[0][1]["refused"]["code"], "R8")
+
+    def test_other_modes_keep_nothing(self):
+        f = Fake()
+        f.plan["mode"] = "health"
+        run(f)
+        self.assertEqual(f.saved_reports, [])
+
+    def test_odd_ids_are_dropped_not_trusted(self):
+        f = Fake()
+        f.plan.update(run_id="../../etc/x", trigger="Night; rm", ring=True)
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        for k in ("run_id", "trigger", "ring"):
+            self.assertNotIn(k, rep)
+
+
+class SaveReportOnDisk(unittest.TestCase):
+    """The real Runner.save_report on a temp dir: root writes into the AGENT's directory, so a symlink must never
+    be followed — neither for the directory nor for the file name."""
+
+    def setUp(self):
+        import tempfile
+        self.tmp = tempfile.mkdtemp()
+        self.dir = os.path.join(self.tmp, "os")
+        os.mkdir(self.dir)
+        self.prev = osapply.PLAN_DIR
+        osapply.PLAN_DIR = self.dir
+        self.r = osapply.Runner()
+        self.r.agent_uid = lambda: os.getuid()
+
+    def tearDown(self):
+        import shutil
+        osapply.PLAN_DIR = self.prev
+        shutil.rmtree(self.tmp)
+
+    def test_writes_0600_next_to_the_plan(self):
+        p = self.r.save_report(os.path.join(self.dir, "plan-r1-guest-apply.json"), {"mode": "apply"})
+        self.assertEqual(p, os.path.join(self.dir, "report-r1-guest-apply.json"))
+        st = os.stat(p)
+        self.assertEqual(statmod.S_IMODE(st.st_mode), 0o600)
+        with open(p) as fh:
+            self.assertEqual(json.load(fh), {"mode": "apply"})
+
+    def test_a_symlink_at_the_name_is_replaced_not_followed(self):
+        victim = os.path.join(self.tmp, "victim")
+        with open(victim, "w") as fh:
+            fh.write("untouched")
+        os.symlink(victim, os.path.join(self.dir, "report-r2-guest-apply.json"))
+        self.r.save_report(os.path.join(self.dir, "plan-r2-guest-apply.json"), {"mode": "apply"})
+        with open(victim) as fh:
+            self.assertEqual(fh.read(), "untouched")
+        self.assertFalse(os.path.islink(os.path.join(self.dir, "report-r2-guest-apply.json")))
+
+    def test_a_symlinked_directory_is_refused(self):
+        real = os.path.join(self.tmp, "elsewhere")
+        os.mkdir(real)
+        link = os.path.join(self.tmp, "linked")
+        os.symlink(real, link)
+        osapply.PLAN_DIR = link
+        with self.assertRaises(OSError):
+            self.r.save_report(os.path.join(link, "plan-r3-guest-apply.json"), {"mode": "apply"})
+        self.assertEqual(os.listdir(real), [])