agent (unreleased): fixes after an independent review — one lock for the writers of agent.json, the trust run serialised and bounded, the CA copy keeps its previous root, the profile stanza from the wrapper's answer, a stale give-up ends, both boot tokens from one read
gates / gates (push) Successful in 48s

Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
This commit is contained in:
admin committed 2026-11-14 19:09:21 +01:00
1 parent 61feb95c06
commit ec7e699fd7
18 files changed
+333 -59

No files matched your search

+15 -5
View File
@@ -1216,7 +1216,8 @@ class Apply:
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."""
copy already starts with the source's bytes. The bytes that are checked are the bytes that are written first;
behind them the copy keeps its own previous certificate (see below), so the copy holds one or two."""
r = self.r
res = {"outcome": None, "path": PVE_CA_COPY}
self.report["pve_ca_copy"] = res
@@ -1253,11 +1254,20 @@ class Apply:
have = r.read_bytes(PVE_CA_COPY)
except OSError:
have = None
if have == data:
res["outcome"] = "same"
if have is not None and have.startswith(data):
res["outcome"] = "same" # the source's certificate is already first in the copy
return 0
r.put_file(PVE_CA_COPY, data, 0o644)
res["outcome"], res["replaced"] = "written", have is not None
# The copy the agent trusts by is replaced BEFORE anything can be proven with the new one (Proxmox serves its
# old certificate until pveproxy is reloaded). So the copy's previous FIRST certificate — written by this same
# step, so already checked as a CA — is kept behind the new one: the agent then trusts both roots across the
# change, also after a restart. At most two certificates; an ended one verifies nothing. (Review, 2026-11-14.)
want, kept = data, False
if have:
prev = PEM_CERT_RE.findall(have)
if prev and b"PRIVATE KEY" not in have and prev[0].strip() != certs[0].strip():
want, kept = data + prev[0] + b"\n", True
r.put_file(PVE_CA_COPY, want, 0o644)
res["outcome"], res["replaced"], res["kept_previous"] = "written", have is not None, kept
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
+22 -1
View File
@@ -2702,7 +2702,28 @@ class PVECACopy(unittest.TestCase):
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)
# the new certificate FIRST, and the copy's previous one kept behind it: Proxmox serves its old certificate until
# pveproxy is reloaded, and an agent that restarts in that window must still have a root that verifies it
self.assertTrue(f.tree[osapply.PVE_CA_COPY].startswith(CA_PEM_2))
self.assertEqual(osapply.PEM_CERT_RE.findall(f.tree[osapply.PVE_CA_COPY]),
osapply.PEM_CERT_RE.findall(CA_PEM_2) + osapply.PEM_CERT_RE.findall(CA_PEM))
self.assertTrue(rep["pve_ca_copy"]["kept_previous"])
self.assertEqual(f.openssl_stdin, [CA_PEM_2.decode()], "only the SOURCE is checked by openssl")
# the same source again: nothing is written (the copy already starts with it)
f2 = TFake(source=CA_PEM_2, copy=f.tree[osapply.PVE_CA_COPY])
rc, rep = run(f2)
self.assertEqual((rc, rep["pve_ca_copy"]["outcome"], f2.puts), (0, "same", []), rep)
# a third change: never more than two certificates — the new one and the one that was first
f3 = TFake(source=CA_PEM, copy=f.tree[osapply.PVE_CA_COPY])
rc, rep = run(f3)
self.assertEqual(rc, 0, rep)
self.assertEqual(osapply.PEM_CERT_RE.findall(f3.tree[osapply.PVE_CA_COPY]),
osapply.PEM_CERT_RE.findall(CA_PEM) + osapply.PEM_CERT_RE.findall(CA_PEM_2))
# a first copy keeps nothing behind it
f4 = TFake()
rc, rep = run(f4)
self.assertFalse(rep["pve_ca_copy"]["kept_previous"])
self.assertEqual(f4.tree[osapply.PVE_CA_COPY], CA_PEM)
def test_a_source_that_is_not_a_ca_certificate_is_refused(self):
self.refused(TFake(ca=False), "T3") # a leaf certificate