R-180: refuse in pre-flight an archive storage the agent's token will not be granted on

The step-8 restore reads the golden as the agent's token; a storage outside PVE_STORAGES (and not
the backup target, which step 6 grants since R-185) 403'd at step 8/8 — after the token was minted
and root@pam rotated. Pre-flight now refuses it with the two remedies (move the golden, or add the
storage to --acl-storages). scripts/test_hostinstall.py: test_archive_storage_* (4).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
This commit is contained in:
2026-10-05 21:19:18 +02:00
parent efe76093ce
commit 7f3944eff9
2 changed files with 59 additions and 1 deletions
+41 -1
View File
@@ -11,7 +11,7 @@ functions act on are variables the test points into the temp directory. Nothing
Where behaviour cannot be isolated, a STATIC test checks the wiring (the function is CALLED, from the
right place) — a helper defined and never called is the seam-built-never-wired shape.
Rows: R-275, R-276, R-881 (uninstall residue); R-306 (--preflight-only writes no state); R-130 (lvm minimum wording).
Rows: R-275, R-276, R-881 (uninstall residue); R-306 (--preflight-only writes no state); R-130 (lvm minimum wording); R-180 (archive storage outside the ACL set).
Run: python3 scripts/test_hostinstall.py
"""
import os
@@ -381,6 +381,46 @@ def test_lvm_minimum_is_named_as_what_it_does():
assert "recommended" in m.group(0) and "continues" in m.group(0), m.group(0)
# ── R-180: the archive storage must be one the agent's token is granted on ─────────────────────
def _granted(storages, archive, target="felhom-backup"):
sb = Sandbox()
try:
rc, out = sb.run(func("_archive_storage_granted") +
'PVE_STORAGES=(%s); ARCHIVE_STORAGE="%s"; BACKUP_TARGET_ID="%s"\n'
'if _archive_storage_granted; then echo GRANTED; else echo REFUSED; fi\n'
% (storages, archive, target))
assert rc == 0, out
return out.strip()
finally:
sb.close()
def test_archive_storage_in_the_acl_set_is_granted():
assert _granted("local local-lvm felhom-pbs", "local") == "GRANTED", "archive storage case ('local local-lvm felhom-pbs', 'local') -> %s, want GRANTED" % _granted("local local-lvm felhom-pbs", "local")
assert _granted("local nvme-scratch", "nvme-scratch") == "GRANTED", "archive storage case ('local nvme-scratch', 'nvme-scratch') -> %s, want GRANTED" % _granted("local nvme-scratch", "nvme-scratch") # --acl-storages adds it
def test_archive_storage_outside_the_acl_set_is_refused():
# the demo-hp 2026-08-03 shape with a storage that is not the backup target
assert _granted("local local-lvm felhom-pbs", "nvme-scratch") == "REFUSED", "archive storage case ('local local-lvm felhom-pbs', 'nvme-scratch') -> %s, want REFUSED" % _granted("local local-lvm felhom-pbs", "nvme-scratch")
assert _granted("local local-lvm felhom-pbs", "local-lvm2") == "REFUSED", "archive storage case ('local local-lvm felhom-pbs', 'local-lvm2') -> %s, want REFUSED" % _granted("local local-lvm felhom-pbs", "local-lvm2") # no prefix match
assert _granted("local local-lvm felhom-pbs", "felhom-backup", target="") == "REFUSED", "archive storage case ('local local-lvm felhom-pbs', 'felhom-backup', target='') -> %s, want REFUSED" % _granted("local local-lvm felhom-pbs", "felhom-backup", target="")
def test_archive_storage_on_the_backup_target_is_granted_in_step_6():
assert _granted("local local-lvm felhom-pbs", "felhom-backup") == "GRANTED", "archive storage case ('local local-lvm felhom-pbs', 'felhom-backup') -> %s, want GRANTED" % _granted("local local-lvm felhom-pbs", "felhom-backup")
def test_archive_storage_check_is_wired_into_preflight():
body = func("step_preflight")
m = re.search(r"archive storage '\$ARCHIVE_STORAGE' present.*?_archive_storage_granted[^\n]*\n[^\n]*\|\| die", body, re.S)
assert m, "step_preflight does not refuse when _archive_storage_granted fails"
# and before anything is minted: the call sits in step_preflight, which runs before step_token.
pre = [m.start() for m in re.finditer(r"^\s*step_preflight\s*$", SRC, re.M)]
tok = re.search(r"^step_token\s*$", SRC, re.M)
assert pre and tok and max(pre) < tok.start(), "preflight no longer runs before the token step"
def main():
tests = [(n, f) for n, f in sorted(globals().items()) if n.startswith("test_") and callable(f)]
fails = 0