#!/usr/bin/env python3 # -*- coding: utf-8 -*- """test_upgrade_bench.py — the test bench's two night-of-2026-09-23 faults (R-653, R-656). No Docker. python3 scripts/test_upgrade_bench.py """ import importlib.util import os import shutil import tempfile import unittest HERE = os.path.dirname(os.path.abspath(__file__)) spec = importlib.util.spec_from_file_location("upgrade_test_mod", os.path.join(HERE, "upgrade-test.py")) ut = importlib.util.module_from_spec(spec) spec.loader.exec_module(ut) class LoadVerdict(unittest.TestCase): """R-653. RED-PROOF (REPORT.md): make load_verdict always return "reached" — the ghost case fails.""" def test_every_request_errored_is_inconclusive(self): # ghost's first watch, 2026-09-23 night: 11 797 requests, all `err` self.assertEqual(ut.load_verdict(11797, {"err": 11797}), "inconclusive") def test_answers_of_any_code_are_the_app_answering(self): self.assertEqual(ut.load_verdict(11429, {"200": 5712, "401": 5717}), "reached") self.assertEqual(ut.load_verdict(10, {"302": 5, "err": 5}), "reached") def test_under_half_is_inconclusive_and_none_is_inconclusive(self): self.assertEqual(ut.load_verdict(10, {"200": 4, "err": 6}), "inconclusive") self.assertEqual(ut.load_verdict(0, {}), "inconclusive") class ClearScratch(unittest.TestCase): """R-656. RED-PROOF (REPORT.md): make clear_scratch_folders return [] without removing — the first test fails with the last run's config still there.""" def setUp(self): self.root = tempfile.mkdtemp(prefix="bench-scratch-") self.hdd = os.path.join(self.root, "hdd") self.env = {"HDD_PATH": self.hdd, "USERDATA_PATH": self.hdd + "/userdata"} self.said = [] def tearDown(self): shutil.rmtree(self.root, ignore_errors=True) def plant(self, rel): p = os.path.join(self.hdd, rel, "config", "config.php") os.makedirs(os.path.dirname(p), exist_ok=True) open(p, "w").write("last run") return p def test_the_apps_own_folders_are_cleared_and_said(self): mine = self.plant("appdata/nextcloud") other = self.plant("appdata/immich") comp = "services:\n nextcloud:\n volumes:\n - ${HDD_PATH}/appdata/nextcloud:/var/www/html\n" got = ut.clear_scratch_folders(comp, self.env, self.said.append) self.assertFalse(os.path.exists(mine), "the last run's files are still there") self.assertTrue(os.path.exists(other), "another app's folder was touched") self.assertEqual(len(got), 1) self.assertTrue(any("cleared before FROM" in s for s in self.said)) def test_never_a_bare_root_never_outside(self): keep = self.plant("appdata/x") comp = (" volumes:\n - ${HDD_PATH}:/data\n - ${HDD_PATH}/:/d2\n" " - ${HDD_PATH}/../escape:/e\n") got = ut.clear_scratch_folders(comp, self.env, self.said.append) self.assertEqual(got, []) self.assertTrue(os.path.exists(keep)) class ChangedFiles(unittest.TestCase): """2026-09-30: a `files_may_change` mark names its files (R-734's method, now in the harness).""" def test_names_changed_added_removed_and_nothing_else(self): before = {"/d": {"a.txt": "1:aa", "b.txt": "1:bb", "gone": "1:cc"}} after = {"/d": {"a.txt": "1:aa", "b.txt": "2:bd", "new": "1:dd"}} self.assertEqual(ut.changed_files(before, after), ["/d/b.txt (changed)", "/d/gone (removed)", "/d/new (added)"]) self.assertEqual(ut.changed_files(before, before), []) class MarkerIgnore(unittest.TestCase): """R-734 (`09` §3 decision 145): app-owned marker files do not make the `files_may_change` mark; any other changed file still does. The data is the bench's own measurement (audits/more-night-apps-2026-09-30/D/evidence/ D-immich-0b82-768/files-{before,after}-detail.json). RED-PROOF (REPORT): make classify_changes ignore every changed file of an app with a list — test_an_unlisted_changed_file_still_marks fails.""" SRC = "/srv/felhom-gate/hdd/appdata/immich" BEFORE = {SRC: {"profile/.immich": "13:7cb1f8a19d944d7e", "upload/.immich": "13:67848ba791bd7ef6", "encoded-video/.immich": "13:68bdf71e87fabadc", "backups/.immich": "13:7cb1f8a19d944d7e", "thumbs/.immich": "13:7cb1f8a19d944d7e", "library/.immich": "13:67848ba791bd7ef6"}, "/srv/felhom-gate/hdd/userdata/media/photos": {}} AFTER = {SRC: {"profile/.immich": "13:0bacc6528e07d8cb", "upload/.immich": "13:a68db79a3fbe6725", "encoded-video/.immich": "13:2cb994a10bfe243c", "backups/.immich": "13:648fe4b9664fadff", "thumbs/.immich": "13:0bacc6528e07d8cb", "library/.immich": "13:a68db79a3fbe6725"}, "/srv/felhom-gate/hdd/userdata/media/photos": {}} def mark(self, app, before, after): counted, ignored = ut.classify_changes(app, before, after) tree = sorted({c.split(" (")[0].rsplit("/", 2)[0] for c in counted + [i["file"] for i in ignored]}) return ut.files_may_change(tree, counted, ignored), counted, ignored def test_the_six_measured_markers_do_not_mark_and_each_carries_a_reason(self): marked, counted, ignored = self.mark("immich", self.BEFORE, self.AFTER) self.assertFalse(marked) self.assertEqual(counted, []) self.assertEqual(len(ignored), 6) self.assertTrue(all(i["reason"] for i in ignored)) def test_an_unlisted_changed_file_still_marks(self): after = {k: dict(v) for k, v in self.AFTER.items()} after[self.SRC]["library/admin/2026/IMG_0001.jpg"] = "2048:abcd" # added: a household photo before = {k: dict(v) for k, v in self.BEFORE.items()} before[self.SRC]["upload/upload/x.jpg"] = "10:aa" after[self.SRC]["upload/upload/x.jpg"] = "10:bb" # changed, same size as a marker marked, counted, ignored = self.mark("immich", before, after) self.assertTrue(marked) self.assertIn(self.SRC + "/library/admin/2026/IMG_0001.jpg (added)", counted) self.assertIn(self.SRC + "/upload/upload/x.jpg (changed)", counted) self.assertEqual(len(ignored), 6) def test_a_listed_name_that_grew_or_vanished_still_marks(self): after = {k: dict(v) for k, v in self.AFTER.items()} after[self.SRC]["thumbs/.immich"] = "4096:ffff" # no longer a 13-byte marker del after[self.SRC]["upload/.immich"] # removed marked, counted, _ = self.mark("immich", self.BEFORE, after) self.assertTrue(marked) self.assertIn(self.SRC + "/thumbs/.immich (changed)", counted) self.assertIn(self.SRC + "/upload/.immich (removed)", counted) def test_the_list_belongs_to_its_app(self): # the same files under another app (and another bind) count: the list is per app marked, counted, ignored = self.mark("calibre-web", self.BEFORE, self.AFTER) self.assertTrue(marked) self.assertEqual((len(counted), ignored), (6, [])) def test_a_moved_tree_the_file_walk_cannot_name_keeps_the_mark(self): self.assertTrue(ut.files_may_change(["/d"], [], [])) self.assertFalse(ut.files_may_change([], [], [])) def test_every_entry_has_a_reason(self): for app, rules in ut.MARKER_IGNORE.items(): for suffix, path, max_size, reason in rules: self.assertTrue(suffix and path and max_size > 0 and len(reason) > 40, (app, path)) # --- R-624 (`09` §3 decision 146): the bench-only admin seed, and the run's secrets ----------------------------------- import sys # noqa: E402 from unittest import mock # noqa: E402 sys.path.insert(0, HERE) import upgrade_fixtures_box as fxbox # noqa: E402 ADMIN = "a" * 8 + "0123456789abcdef0123456789abcdef0123456789abcdef01234567" # stands in for a generated hex:32 class FakeVenue: """walk.py's interface without a network: records every call; answers like vaultwarden 1.36 did on 9202 (audits/evidence-p1fixes-2026-09-15/E1-vaultwarden-spike.txt): a stranger's register 400, POST /admin with the right token sets VW_ADMIN, /admin/invite 200 with that cookie, an invited address registers 200.""" def __init__(self, bench): if bench: self.VENUE = "bench" self.GENERATED = {"vaultwarden": {"ADMIN_TOKEN": ADMIN, "DOMAIN": "gate.invalid"}} self.calls, self.invited = [], set() def wait_app(self, *a, **k): return True def app_curl(self, sub, path, *extra, method=None, data=None, timeout=45): headers = [] for i, x in enumerate(extra): if x == "-H" and extra[i + 1].startswith("@"): headers.append(open(extra[i + 1][1:]).read().strip()) self.header_file = extra[i + 1][1:] self.calls.append({"path": path, "argv": list(extra), "data": data, "file_headers": headers}) if path == "/identity/accounts/register": email = json.loads(data)["email"] return 0, ("200" if email in self.invited else "400"), '{"message":"Registration not allowed"}' if path == "/admin": if data == "token=" + ADMIN: return 0, "200", "HTTP/1.1 200 OK\r\nset-cookie: VW_ADMIN=jwt.admin.session; Path=/admin\r\n\r\n" return 0, "401", "HTTP/1.1 401\r\n\r\nInvalid admin token" if path == "/admin/invite": if "Cookie: VW_ADMIN=jwt.admin.session" in headers: self.invited.add(json.loads(data)["email"]) return 0, "200", "{}" return 0, "401", "" return 0, "404", "" import json # noqa: E402 class BenchAdminSeedGuard(unittest.TestCase): """The admin seed runs on the bench ONLY. RED-PROOF (REPORT): make bench_admin_seed_allowed return (True, "") — test_the_box_walk_never_signs_in_as_admin and each single-condition refusal fail.""" def setUp(self): self.tmp = tempfile.mkdtemp(prefix="bench-guard-") self.nobox = os.path.join(self.tmp, "no-such-stacks") self.said = [] def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def allowed(self, venue, env, box_marker): with mock.patch.dict(os.environ, env, clear=False), mock.patch.object(fxbox, "BOX_MARKER", box_marker): if "FELHOM_BENCH_ADMIN_SEED" not in env: os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None) return fxbox.bench_admin_seed_allowed(venue) def test_all_three_conditions_allow(self): self.assertEqual(self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox), (True, "")) def test_each_condition_alone_refuses(self): ok, why = self.allowed(FakeVenue(False), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox) self.assertFalse(ok) self.assertIn("not the bench venue", why) ok, why = self.allowed(FakeVenue(True), {}, self.nobox) self.assertFalse(ok) self.assertIn("FELHOM_BENCH_ADMIN_SEED", why) ok, why = self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "yes"}, self.nobox) self.assertFalse(ok) ok, why = self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.tmp) # a box: stacks exists self.assertFalse(ok) self.assertIn("Felhom box", why) def test_the_bench_venue_names_itself(self): import upgrade_boxport self.assertEqual(upgrade_boxport.Venue.VENUE, "bench") def seed(self, venue, env, box_marker): with mock.patch.dict(os.environ, env, clear=False), mock.patch.object(fxbox, "BOX_MARKER", box_marker): if "FELHOM_BENCH_ADMIN_SEED" not in env: os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None) fx_ = fxbox.Vaultwarden() return fx_, fx_.seed(venue, "vault", self.said.append) def test_the_box_walk_never_signs_in_as_admin(self): v = FakeVenue(False) fx_, got = self.seed(v, {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox) self.assertIsNone(got) self.assertEqual([c["path"] for c in v.calls], ["/identity/accounts/register"]) self.assertIn("NOT tried", fx_.tried) def test_the_bench_seeds_through_the_admin_invite_and_never_shows_the_token(self): v = FakeVenue(True) _, got = self.seed(v, {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox) self.assertIsNotNone(got) self.assertEqual([c["path"] for c in v.calls], ["/identity/accounts/register", "/admin", "/admin/invite", "/identity/accounts/register"]) for c in v.calls: self.assertFalse(any(ADMIN in a or "VW_ADMIN" in a for a in c["argv"]), "a secret on the command line") self.assertEqual(v.calls[2]["file_headers"], ["Cookie: VW_ADMIN=jwt.admin.session"]) self.assertFalse(os.path.exists(v.header_file), "the cookie header file was not shredded") self.assertFalse(any(ADMIN in s or "jwt.admin.session" in s for s in self.said), "a secret was printed") class FakeBoxWalk(FakeVenue): """box_walk's interface on a test box: no VENUE, HP/GUEST, the run's installs, and guest() that answers like the in-box admin invite script would. Records every script sent into the box.""" def __init__(self, hp="demo-hp", guest="9202", deployed=("vaultwarden",), drill=True, invite="RESULT signin=200 session=yes invite=200"): FakeVenue.__init__(self, False) self.GENERATED = {} self.HP, self.GUEST, self.DEPLOYED_THIS_RUN = hp, guest, set(deployed) self.drill, self.invite, self.scripts = drill, invite, [] def guest(self, script, timeout=600): self.scripts.append(script) if "controller.yaml" in script: return ("git:\n repo_url: https://gitea.dooplex.hu/admin/%s.git\n" % ("app-catalog-drill" if self.drill else "app-catalog-felhom.eu")) if "/admin/invite" in script: m = __import__("re").search(r'"email":"([^"]+)"', script) if self.invite.endswith("invite=200"): self.invited.add(m.group(1)) return "some noise\n" + self.invite + "\n" return "" class BoxAdminSeedGuard(unittest.TestCase): """R-890 (`09` §3 decision 149): the admin seed on a TEST BOX — scratch 9202 on demo-hp, an app this run installed, the box on the drill catalog, the run opted in. RED-PROOF (REPORT): make box_admin_seed_allowed return (True, "") — test_each_condition_alone_refuses and test_a_household_shaped_box_is_never_seeded fail.""" ENV = {"FELHOM_BOX_ADMIN_SEED": "1"} def allowed(self, w, env): with mock.patch.dict(os.environ, env, clear=False): if "FELHOM_BOX_ADMIN_SEED" not in env: os.environ.pop("FELHOM_BOX_ADMIN_SEED", None) return fxbox.box_admin_seed_allowed(w, "vaultwarden") def test_all_conditions_allow(self): self.assertEqual(self.allowed(FakeBoxWalk(), self.ENV), (True, "")) def test_each_condition_alone_refuses(self): cases = [(FakeVenue(True), self.ENV, "bench venue"), (FakeBoxWalk(), {}, "FELHOM_BOX_ADMIN_SEED"), (FakeBoxWalk(), {"FELHOM_BOX_ADMIN_SEED": "yes"}, "FELHOM_BOX_ADMIN_SEED"), (FakeBoxWalk(guest="9201"), self.ENV, "not a test box"), (FakeBoxWalk(hp="demo-felhom"), self.ENV, "not a test box"), (FakeBoxWalk(deployed=()), self.ENV, "did not install"), (FakeBoxWalk(drill=False), self.ENV, "drill catalog")] for w, env, want in cases: ok, why = self.allowed(w, env) self.assertFalse(ok, want) self.assertIn(want, why) def seed(self, w, env): with mock.patch.dict(os.environ, env, clear=False): if "FELHOM_BOX_ADMIN_SEED" not in env: os.environ.pop("FELHOM_BOX_ADMIN_SEED", None) os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None) fx_ = fxbox.Vaultwarden() return fx_, fx_.seed(w, "vault", lambda *a: None) def test_the_test_box_seeds_through_the_invite_inside_the_box(self): w = FakeBoxWalk() _, got = self.seed(w, self.ENV) self.assertIsNotNone(got) self.assertEqual([c["path"] for c in w.calls], ["/identity/accounts/register", "/identity/accounts/register"]) inv = [x for x in w.scripts if "/admin/invite" in x] self.assertEqual(len(inv), 1) # the token is read inside the box and handed to curl on stdin; the cookie file is shredded self.assertIn("printenv ADMIN_TOKEN", inv[0]) self.assertIn("token@-", inv[0]) self.assertIn('shred -u "$H"', inv[0]) self.assertIn(got["email"], inv[0]) def test_a_household_shaped_box_is_never_seeded(self): for w in (FakeBoxWalk(guest="9201"), FakeBoxWalk(deployed=()), FakeBoxWalk(drill=False)): fx_, got = self.seed(w, self.ENV) self.assertIsNone(got) self.assertFalse([x for x in w.scripts if "/admin/invite" in x], "the admin route was tried") self.assertIn("NOT tried", fx_.tried) def test_a_refused_invite_is_inconclusive(self): w = FakeBoxWalk(invite="RESULT signin=401 session=no") fx_, got = self.seed(w, self.ENV) self.assertIsNone(got) self.assertIn("test-box admin invite", fx_.tried) def test_only_a_drill_address_is_invited(self): with self.assertRaises(ValueError): fxbox.box_admin_invite(FakeBoxWalk(), 'x"}; rm -rf /; {"@gate.invalid') class SecretHygiene(unittest.TestCase): """The run's secrets: .env 0600 and shredded, every evidence file redacted. RED-PROOF (REPORT): make redact_tree return [] without rewriting — test_evidence_files_are_redacted fails.""" FELHOM = ("deploy_fields:\n - env_var: DOMAIN\n type: domain\n - env_var: ADMIN_TOKEN\n type: secret\n" " generate: \"hex:32\"\n - env_var: SIGNUPS_ALLOWED\n type: text\n default: \"false\"\n") def setUp(self): self.tmp = tempfile.mkdtemp(prefix="bench-secrets-") def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def test_secret_values_are_the_generated_fields_and_the_seed_password(self): if ut.cvp is None: import importlib.util as iu sp = iu.spec_from_file_location("cvp", os.path.join(HERE, "check-volume-persistence.py")) m = iu.module_from_spec(sp) sp.loader.exec_module(m) ut.cvp = m env = {"DOMAIN": "gate.invalid", "ADMIN_TOKEN": ADMIN, "SIGNUPS_ALLOWED": "false"} got = ut.secret_values(self.FELHOM, env, {"email": "drill-1@gate.invalid", "pw": "Drill-0011223344"}) self.assertEqual(set(got), {ADMIN, "Drill-0011223344"}) def test_env_is_0600_and_shredded(self): p = ut.Path(self.tmp) / ".env" ut.write_secret_file(p, "ADMIN_TOKEN=%s\n" % ADMIN) self.assertEqual(os.stat(p).st_mode & 0o777, 0o600) ut.write_secret_file(p, "ADMIN_TOKEN=%s\n" % ADMIN) # a re-render replaces it, still 0600 self.assertEqual(os.stat(p).st_mode & 0o777, 0o600) ut.shred_file(p) self.assertFalse(p.exists()) ut.shred_file(p) # a missing file is fine def test_evidence_files_are_redacted(self): ev = ut.Path(self.tmp) / "evidence" / "MV-vaultwarden" (ev / "sub").mkdir(parents=True) (ev / "run.log").write_text("token=%s ok\n" % ADMIN) (ev / "sub" / "to-full.log").write_text("clean line\n") (ev / "verdict.json").write_text(json.dumps({"abort_detail": "pw Drill-0011223344"})) held = ut.redact_tree(ev, [ADMIN, "Drill-0011223344"]) self.assertEqual(len(held), 2) for p in ev.rglob("*"): if p.is_file(): t = p.read_text() self.assertNotIn(ADMIN, t) self.assertNotIn("Drill-0011223344", t) self.assertIn(ut.REDACTED, (ev / "run.log").read_text()) self.assertEqual((ev / "sub" / "to-full.log").read_text(), "clean line\n") def test_redact_longest_first(self): self.assertEqual(ut.redact("x abcdefgh123 y", ["abcdefgh123", "abcdefgh"]), "x y") if __name__ == "__main__": unittest.main(verbosity=2)