2b (unreleased): the root step that keeps a copy of Proxmox's root CA (wrapper mode pve-ca-copy, R-1380)

configs/felhom-os-apply

@@ -161,6 +161,13 @@ KERNEL_MODES = ("kernel-status", "kernel-reboot", "kernel-boot", "kernel-good",
 KERNEL_ACTIVE = ("staged", "oneshot", "judging", "reverting")
 # The clean-up of old kernels (R-1368, `09` §3 decision 286, `11` §5.11.1). A step that ENDED in one of these phases
 # and is not "good" keeps every kernel (the box fell back or reverted); any other phase is "not ended" (P1).
+# Trust in Proxmox by a copy of the box's own root CA (R-1380, `09` §3 decision 303, felhom.eu `03` §12a). The agent's
+# user cannot read Proxmox's CA file (0640 root:www-data); this root step keeps a readable copy of that ONE public
+# certificate where the agent's `proxmox.tls.ca_file` points.
+PVE_CA_SOURCE = "/etc/pve/pve-root-ca.pem"
+PVE_CA_COPY = "/etc/felhom-agent/pve-root-ca.pem"      # 0644 root: a CA certificate is public; the agent only reads it
+PVE_CA_MAX_BYTES = 64 * 1024
+PEM_CERT_RE = re.compile(rb"-----BEGIN CERTIFICATE-----\s[A-Za-z0-9+/=\s]+-----END CERTIFICATE-----")
 # The security-profile refresh (R-1388, R-1301, `11` §5.12; measured on demo-hp 2026-11-14,
 # felhom.eu audits/profile-refresh-spike-2026-11-14/). AppArmor keeps one cache folder per kernel feature set; a
 # profile that changed while ANOTHER kernel ran leaves an old compiled copy in this kernel's folder, with a time that
@@ -514,7 +521,7 @@ class Apply:
     def check_plan(self, plan):
         mode = plan.get("mode", "apply")
         if mode not in ("apply", "inventory", "health", "facts", "live-restore-on", "bundle", "bundle-policy", "agent_update",
-                        "oom-check", "profile-refresh") \
+                        "oom-check", "profile-refresh", "pve-ca-copy") \
                 + KERNEL_MODES:
             raise Refused("R11", f"unknown mode {mode!r}")
         if mode == "agent_update":
@@ -529,6 +536,15 @@ class Apply:
             if plan.get("layer") != "host":
                 raise Refused("R11", "bundle-policy is a host-layer mode")
             return mode, "host", 0, "bundle-policy"
+        if mode == "pve-ca-copy":
+            if plan.get("layer") != "host":
+                raise Refused("R11", "pve-ca-copy is a host-layer mode")
+            # R-1380 (D80): ONE fixed source, ONE fixed place — the plan may name NOTHING (no path, no file, no content)
+            extra = sorted(k for k in plan if k not in ("mode", "layer", "release_id", "run_id", "trigger", "ring"))
+            if extra:
+                raise Refused("T1", f"a pve-ca-copy plan carries {', '.join(extra)} — it takes nothing from a plan: "
+                                    f"the source is {PVE_CA_SOURCE} and the copy is {PVE_CA_COPY}, both fixed")
+            return mode, "host", 0, "pve-ca-copy"
         if mode == "profile-refresh":
             if plan.get("layer") != "host":
                 raise Refused("R11", "profile-refresh is a host-layer mode")
@@ -1124,6 +1140,10 @@ class Apply:
             return Bundle(self).from_plan(plan)
         if self.mode == "bundle-policy":
             return Bundle(self).policy_pass()
+        if self.mode == "pve-ca-copy":
+            # NO appliance check, on purpose (like the bundle): the copy is the agent's OWN trust file in the agent's own
+            # config folder, and it changes nothing of the host's — a BYO host's agent talks to Proxmox the same way.
+            return self.pve_ca_copy()
         if self.mode == "profile-refresh":
             self.check_appliance()
             return self.profile_refresh(plan)
@@ -1192,6 +1212,56 @@ class Apply:
             self.report["oom_check"] = self.oom_check()
         return 0
 
+    # ---------- trust in Proxmox by a copy of its root CA (R-1380, D80) ----------
+    def pve_ca_copy(self):
+        """Copy exactly PVE_CA_SOURCE to exactly PVE_CA_COPY — only if the source is ONE CA certificate that has not
+        ended — written whole or not at all (a root-owned temp file beside the target, fsync, rename). "same" when the
+        copy already holds the source's bytes. The bytes that are checked are the bytes that are written."""
+        r = self.r
+        res = {"outcome": None, "path": PVE_CA_COPY}
+        self.report["pve_ca_copy"] = res
+        try:
+            data = r.read_bytes(PVE_CA_SOURCE)
+        except OSError as e:
+            raise Refused("T2", f"{PVE_CA_SOURCE} cannot be read ({e.__class__.__name__}) — is this a Proxmox host?")
+        if not data or len(data) > PVE_CA_MAX_BYTES:
+            raise Refused("T3", f"{PVE_CA_SOURCE} is {len(data)} bytes — not one certificate")
+        certs = PEM_CERT_RE.findall(data)
+        if len(certs) != 1 or b"PRIVATE KEY" in data:
+            raise Refused("T3", f"{PVE_CA_SOURCE} holds {len(certs)} certificate(s)"
+                                + (" and a private key" if b"PRIVATE KEY" in data else "") + " — exactly one CA certificate is expected")
+        try:
+            text = data.decode("ascii")      # PEM is ASCII; the runner hands text to openssl's stdin
+        except UnicodeDecodeError:
+            raise Refused("T3", f"{PVE_CA_SOURCE} is not a PEM text file")
+        rc, out, err = r.host(["openssl", "x509", "-noout", "-ext", "basicConstraints", "-enddate"], 30, stdin=text)
+        if rc != 0:
+            raise Refused("T3", f"{PVE_CA_SOURCE} is not a readable certificate: {(out + err).strip()[-160:]}")
+        if not re.search(r"\bCA:TRUE\b", out):
+            raise Refused("T3", f"{PVE_CA_SOURCE} is not a CA certificate (no CA:TRUE) — a leaf or any other file is never copied")
+        m = re.search(r"notAfter=(.+)", out)
+        try:
+            import calendar
+            not_after = calendar.timegm(time.strptime(m.group(1).strip(), "%b %d %H:%M:%S %Y GMT"))   # UTC, no DST
+        except (AttributeError, ValueError):
+            raise Refused("T3", f"the end date of {PVE_CA_SOURCE} cannot be read")
+        days = int((not_after - r.now()) // 86400)
+        if days < 0:
+            raise Refused("T3", f"{PVE_CA_SOURCE} ended {-days} day(s) ago — an ended CA is never copied")
+        res.update(sha256=sha256_hex(data), not_after=_iso(not_after), days_left=days)
+        try:
+            have = r.read_bytes(PVE_CA_COPY)
+        except OSError:
+            have = None
+        if have == data:
+            res["outcome"] = "same"
+            return 0
+        r.put_file(PVE_CA_COPY, data, 0o644)
+        res["outcome"], res["replaced"] = "written", have is not None
+        r.log(f"os-apply: PVE CA COPY written {PVE_CA_COPY} sha256={res['sha256'][:16]} ends={res['not_after']} "
+              f"({'replaced an older copy' if have is not None else 'first copy'})")
+        return 0
+
     # ---------- the security-profile refresh (R-1388, R-1301, `11` §5.12) ----------
     def profile_cache(self):
         """{"<folder>/<file>": sha256} of every cached profile, every kernel's folder."""

configs/test_felhom_os_apply.py

@@ -2633,5 +2633,143 @@ class ProfileRefresh(unittest.TestCase):
         run(f)
         self.assertFalse([c for c in f.calls if c[1][0] in ("apt-get", "dpkg", "apt")], "no package command at all")
 
+
+# ---------- trust in Proxmox by a copy of its root CA (R-1380, D80, felhom.eu `03` §12a) ----------
+CA_PEM = b"-----BEGIN CERTIFICATE-----\nQUJDREVGR0hJSktMTU5PUFFSU1RVVldYWVo=\n-----END CERTIFICATE-----\n"
+CA_PEM_2 = b"-----BEGIN CERTIFICATE-----\nWllYV1ZVVFNSUVBPTk1MS0pJSEdGRURDQkE=\n-----END CERTIFICATE-----\n"
+
+
+class TFake(Fake):
+    """A Proxmox host: the root CA file, the copy (or none), and openssl's answer about the bytes it is given."""
+
+    def __init__(self, source=CA_PEM, copy=None, ca=True, ends="Aug 17 15:44:42 2036 GMT"):
+        super().__init__()
+        self.source, self.tree, self.ca, self.ends, self.openssl_rc = source, {}, ca, ends, 0
+        if copy is not None:
+            self.tree[osapply.PVE_CA_COPY] = copy
+        self.openssl_stdin, self.puts = [], []
+        self.plan = {"release_id": "pve-ca", "layer": "host", "mode": "pve-ca-copy"}
+
+    def read_bytes(self, p):
+        if p == osapply.PVE_CA_SOURCE:
+            if self.source is None:
+                raise FileNotFoundError(p)
+            return self.source
+        if p in self.tree:
+            return self.tree[p]
+        raise FileNotFoundError(p)
+
+    def put_file(self, path, data, mode):
+        self.puts.append((path, data, mode))
+        self.tree[path] = data
+
+    def host(self, argv, timeout=600, stdin=None):
+        if argv[0] == "openssl":
+            self.calls.append(("host", argv))
+            self.openssl_stdin.append(stdin)
+            if self.openssl_rc:
+                return self.openssl_rc, "", "Could not read certificate from <stdin>"
+            return 0, ("X509v3 Basic Constraints: critical\n    CA:%s\nnotAfter=%s\n" % ("TRUE" if self.ca else "FALSE", self.ends)), ""
+        return super().host(argv, timeout, stdin)
+
+
+class PVECACopy(unittest.TestCase):
+    """Red-proofs: felhom.eu audits/night-build-2026-11-14/A/redproof-wrapper.txt."""
+
+    def refused(self, f, code):
+        rc, rep = run(f)
+        self.assertEqual((rc, rep["refused"]["code"]), (2, code), rep)
+        self.assertEqual(f.puts, [], "nothing is written on a refusal")
+        return rep
+
+    def test_a_new_box_gets_the_copy(self):
+        f = TFake()
+        rc, rep = run(f)
+        self.assertEqual(rc, 0, rep)
+        self.assertEqual(f.puts, [(osapply.PVE_CA_COPY, CA_PEM, 0o644)], "ONE file, the fixed place, the source's bytes, 0644")
+        self.assertEqual(f.openssl_stdin, [CA_PEM.decode()], "the bytes that are checked are the bytes that are written")
+        c = rep["pve_ca_copy"]
+        self.assertEqual((c["outcome"], c["path"], c["replaced"]), ("written", "/etc/felhom-agent/pve-root-ca.pem", False))
+        self.assertEqual(c["not_after"], "2036-08-17T15:44:42Z")
+        self.assertGreater(c["days_left"], 3000)
+
+    def test_the_same_copy_is_not_written_again(self):
+        f = TFake(copy=CA_PEM)
+        rc, rep = run(f)
+        self.assertEqual((rc, rep["pve_ca_copy"]["outcome"], f.puts), (0, "same", []), rep)
+
+    def test_a_renewed_ca_refreshes_the_copy(self):
+        f = TFake(source=CA_PEM_2, copy=CA_PEM)
+        rc, rep = run(f)
+        self.assertEqual((rc, rep["pve_ca_copy"]["outcome"], rep["pve_ca_copy"]["replaced"]), (0, "written", True), rep)
+        self.assertEqual(f.tree[osapply.PVE_CA_COPY], CA_PEM_2)
+
+    def test_a_source_that_is_not_a_ca_certificate_is_refused(self):
+        self.refused(TFake(ca=False), "T3")                                  # a leaf certificate
+        self.refused(TFake(source=b"hello\n"), "T3")                         # not a certificate at all
+        self.refused(TFake(source=CA_PEM + CA_PEM_2), "T3")                  # two certificates (a chain, a bundle)
+        self.refused(TFake(source=CA_PEM + b"-----BEGIN PRIVATE KEY-----\nAAAA\n-----END PRIVATE KEY-----\n"), "T3")
+        self.refused(TFake(source=b""), "T3")
+        self.refused(TFake(source=CA_PEM + b"x" * (osapply.PVE_CA_MAX_BYTES + 1)), "T3")
+        f = TFake()
+        f.openssl_rc = 1                                                     # openssl cannot parse it
+        self.refused(f, "T3")
+        self.refused(TFake(ends="Aug 17 15:44:42 2020 GMT"), "T3")           # an ended CA
+        self.refused(TFake(ends="garbage"), "T3")
+
+    def test_no_source_is_refused(self):
+        self.refused(TFake(source=None), "T2")
+
+    def test_with_the_real_openssl_and_the_real_runner(self):
+        """NOT a fake: a CA and a leaf made by openssl in a temp folder, judged through Runner.host as on a box (the
+        fake cannot show a wrong argument or a bytes/text mix-up). Skipped where openssl is missing."""
+        import shutil, tempfile
+        if not shutil.which("openssl"):
+            self.skipTest("no openssl here")
+        d = tempfile.mkdtemp()
+        try:
+            def sh(*a):
+                subprocess.run(a, check=True, capture_output=True, cwd=d)
+            sh("openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "ca.key", "-out", "ca.pem", "-days", "30",
+               "-subj", "/CN=test CA", "-addext", "basicConstraints=critical,CA:TRUE")
+            sh("openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-keyout", "leaf.key", "-out", "leaf.pem", "-days", "30",
+               "-subj", "/CN=leaf", "-addext", "basicConstraints=CA:FALSE")
+            real = osapply.Runner()
+            for name, want in (("ca.pem", "written"), ("leaf.pem", "T3")):
+                data = open(os.path.join(d, name), "rb").read()
+                f = TFake(source=data)
+                f.host = lambda argv, timeout=600, stdin=None: real.host(argv, timeout, stdin) if argv[0] == "openssl" else Fake.host(f, argv, timeout, stdin)
+                f.clock = __import__("time").time()
+                rc, rep = run(f)
+                got = rep["pve_ca_copy"]["outcome"] if rc == 0 else rep["refused"]["code"]
+                self.assertEqual(got, want, f"{name}: {rep}")
+                if want == "written":
+                    self.assertIn(rep["pve_ca_copy"]["days_left"], (29, 30))
+        finally:
+            shutil.rmtree(d, ignore_errors=True)
+
+    def test_a_plan_that_names_a_path_or_a_file_is_refused(self):
+        for extra in ({"source": "/etc/shadow"}, {"path": "/etc/sudoers.d/x"}, {"dest": "/root/.ssh/authorized_keys"},
+                      {"content_b64": "QQ=="}, {"packages": []}, {"file": "x"}, {"vmid": 9201}, {"dry_run": True}):
+            f = TFake()
+            f.plan.update(extra)
+            rep = self.refused(f, "T1")
+            self.assertEqual(f.openssl_stdin, [], "refused before the source is even read")
+        g = TFake()
+        g.plan["layer"] = "kernel"
+        self.refused(g, "R11")
+
+    def test_a_refused_source_keeps_an_older_copy(self):
+        f = TFake(source=b"not a certificate", copy=CA_PEM)
+        self.refused(f, "T3")
+        self.assertEqual(f.tree[osapply.PVE_CA_COPY], CA_PEM, "written whole or not at all")
+
+    def test_it_runs_on_a_byo_host_too_and_changes_nothing_else(self):
+        f = TFake()
+        f.files[osapply.INSTALL_STATE] = json.dumps({"mode": "byo"})
+        rc, rep = run(f)
+        self.assertEqual((rc, rep["pve_ca_copy"]["outcome"]), (0, "written"), rep)
+        self.assertEqual([c[1][0] for c in f.calls], ["openssl"], "one openssl call; no package, no service, no other file")
+
 if __name__ == "__main__":
     unittest.main()