diff --git a/tests/test_prune_storage_share_dumps.py b/tests/test_prune_storage_share_dumps.py
new file mode 100644
index 0000000..3397deb
--- /dev/null
+++ b/tests/test_prune_storage_share_dumps.py
@@ -0,0 +1,195 @@
+"""Synthetic tests for the list-first Storage Share dump pruning (no network, no credentials)."""
+
+import hashlib
+import importlib.util
+import io
+import json
+import urllib.error
+import urllib.parse
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location(
+ "prune", ROOT / "tools/bootstrap-recovery/prune-storage-share-dumps.py"
+)
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+FOLDER = "r01-net-kingdom-pg"
+
+
+def names(days, prefix="net-kingdom-pg-privacyidea_db"):
+ return {f"{prefix}-202610{d:02d}T023000Z.dump.age": 1000 + d for d in days}
+
+
+class Response(io.BytesIO):
+ def __init__(self, status, url, body=b""):
+ super().__init__(body)
+ self.status, self.url = status, url
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *a):
+ return False
+
+
+class Share:
+ def __init__(self, files, extra=()):
+ self.files, self.extra, self.deleted = dict(files), list(extra), []
+
+ def open(self, request, timeout):
+ url, method = request.full_url, request.get_method()
+ assert request.headers["Authorization"].startswith("Basic ")
+ if method == "PROPFIND":
+ rows = "".join(
+ f"{m.BASE}{FOLDER}/{urllib.parse.quote(n)}"
+ f"{s}"
+ f"HTTP/1.1 200 OK"
+ for n, s in [*self.files.items(), *self.extra]
+ )
+ body = (
+ f'{m.BASE}{FOLDER}/'
+ f"{rows}"
+ ).encode()
+ return Response(207, url, body)
+ name = urllib.parse.unquote(url.rsplit("/", 1)[1])
+ if method == "DELETE":
+ del self.files[name]
+ self.deleted.append(name)
+ return Response(204, url)
+ if method == "HEAD":
+ if name in self.files:
+ return Response(200, url)
+ raise urllib.error.HTTPError(url, 404, "Not Found", None, None)
+ raise AssertionError(method)
+
+
+def make_plan(tmp_path, share, *extra):
+ path = tmp_path / "plan.json"
+ assert (
+ m.main(
+ ["plan", "--folder", FOLDER, "--plan", str(path), *extra],
+ client=share,
+ secret="pw",
+ )
+ == 0
+ )
+ return path, hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def test_plan_keeps_newest_daily_and_weekly_per_database(tmp_path, capsys):
+ share = Share(
+ {**names(range(1, 29)), **names(range(1, 29), "net-kingdom-pg-other_db")}
+ )
+ path, _ = make_plan(tmp_path, share)
+ plan = json.loads(path.read_text())
+ for prefix in ("net-kingdom-pg-privacyidea_db", "net-kingdom-pg-other_db"):
+ kept = sorted(n for n in plan["keep"] if n.startswith(prefix + "-"))
+ assert (
+ f"{prefix}-20261028T023000Z.dump.age" in kept
+ and f"{prefix}-20261022T023000Z.dump.age" in kept
+ )
+ assert 7 <= len(kept) <= 11
+ assert not share.deleted and set(plan["keep"]) | set(plan["delete"]) == set(
+ share.files
+ )
+ assert "plan sha256" in capsys.readouterr().out
+
+
+def test_delete_needs_the_plan_hash_and_removes_only_planned_files(tmp_path):
+ share = Share(names(range(1, 29)))
+ path, digest = make_plan(tmp_path, share)
+ plan = json.loads(path.read_text())
+ assert (
+ m.main(
+ [
+ "delete",
+ "--plan",
+ str(path),
+ "--confirm",
+ "0" * 64,
+ "--receipt",
+ str(tmp_path / "r0.json"),
+ ],
+ client=share,
+ secret="pw",
+ )
+ == 1
+ )
+ assert not share.deleted
+ assert (
+ m.main(
+ [
+ "delete",
+ "--plan",
+ str(path),
+ "--confirm",
+ digest,
+ "--receipt",
+ str(tmp_path / "r.json"),
+ ],
+ client=share,
+ secret="pw",
+ )
+ == 0
+ )
+ assert sorted(share.deleted) == sorted(plan["delete"]) and set(share.files) == set(
+ plan["keep"]
+ )
+ receipt = json.loads((tmp_path / "r.json").read_text())
+ assert (
+ receipt["kept_all_present"]
+ and receipt["deleted"] == len(plan["delete"])
+ and "pw" not in json.dumps(receipt)
+ )
+
+
+def test_delete_refuses_when_the_folder_changed_since_the_plan(tmp_path):
+ share = Share(names(range(1, 29)))
+ path, digest = make_plan(tmp_path, share)
+ victim = min(json.loads(path.read_text())["delete"])
+ share.files[victim] += 1
+ assert (
+ m.main(
+ [
+ "delete",
+ "--plan",
+ str(path),
+ "--confirm",
+ digest,
+ "--receipt",
+ str(tmp_path / "r.json"),
+ ],
+ client=share,
+ secret="pw",
+ )
+ == 1
+ )
+ assert not share.deleted
+
+
+@pytest.mark.parametrize("folder", ["net-kingdom-identity", "openbao", "r01-x/..", ""])
+def test_only_r01_dump_folders_are_accepted(tmp_path, folder):
+ assert (
+ m.main(
+ ["plan", "--folder", folder, "--plan", str(tmp_path / "p.json")],
+ client=Share({}),
+ secret="pw",
+ )
+ == 1
+ )
+
+
+def test_unexpected_entries_stop_the_plan(tmp_path):
+ share = Share(names(range(1, 9)), extra=[("notes.txt", 5)])
+ assert (
+ m.main(
+ ["plan", "--folder", FOLDER, "--plan", str(tmp_path / "p.json")],
+ client=share,
+ secret="pw",
+ )
+ == 1
+ )
+ assert not (tmp_path / "p.json").exists()
diff --git a/tools/bootstrap-recovery/prune-storage-share-dumps.py b/tools/bootstrap-recovery/prune-storage-share-dumps.py
new file mode 100644
index 0000000..fdf2f13
--- /dev/null
+++ b/tools/bootstrap-recovery/prune-storage-share-dumps.py
@@ -0,0 +1,298 @@
+#!/usr/bin/env python3
+"""List-first pruning of one CNPG dump folder in the Storage Share Backup account.
+
+Founder-run emergency relief for a full Backup account (NK-WP-0055-T07 finding, 2026-10-29): the
+CNPG Option A dumps (`railiance-platform/tools/cmd/cnpg-option-a-backup`) prune only their local copy,
+so `railiance-backups/r01-*` folders grow without bound and the identity backup refuses on quota.
+Permanent remote retention belongs to railiance-platform; this tool only frees space once.
+
+ plan --folder r01-net-kingdom-pg --plan ~/prune-plan.json [--keep-daily 7] [--keep-weekly 4]
+ read-only: lists the folder, keeps per database the newest N dumps, the newest dump of each of
+ the last W ISO weeks and always the newest one, writes the plan, prints its sha256.
+ delete --plan ~/prune-plan.json --confirm --receipt ~/prune-receipt.json
+ re-lists, refuses unless every planned file is still there with its size and every kept file
+ still exists, deletes only the planned files and checks each is gone.
+
+The Backup app password is read at a hidden prompt. Only `r01-*` dump folders are accepted; the
+identity bundle and OpenBao folders are refused. No credential, URL or provider response is printed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import getpass
+import hashlib
+import json
+import os
+import re
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+import warnings
+import xml.etree.ElementTree as ET
+from datetime import UTC, datetime
+from pathlib import Path
+
+HOST = "https://nx4069.your-storageshare.de"
+BASE = "/remote.php/dav/files/Backup/railiance-backups/"
+FOLDER = re.compile(r"r01-[a-z0-9-]{1,60}\Z")
+DUMP = re.compile(r"(?P[a-z0-9_-]{1,120})-(?P\d{8}T\d{6}Z)\.dump\.age\Z")
+LIMIT = 16 * 1024 * 1024
+
+
+class Refused(Exception):
+ pass
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+ def redirect_request(self, *args, **kwargs):
+ raise Refused("redirect refused")
+
+
+def opener():
+ return urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
+
+
+def prompt():
+ if not sys.stdin.isatty() or not sys.stderr.isatty():
+ raise Refused("run in your own terminal")
+ with warnings.catch_warnings():
+ warnings.simplefilter("error", getpass.GetPassWarning)
+ secret = getpass.getpass("Storage Share Backup app password: ")
+ if not secret or any(c in secret for c in "\r\n\x00"):
+ raise Refused("empty or invalid password")
+ return secret
+
+
+def auth(secret):
+ return {
+ "Authorization": "Basic "
+ + base64.b64encode(("Backup:" + secret).encode()).decode(),
+ "Accept-Encoding": "identity",
+ }
+
+
+def folder_url(folder):
+ if not FOLDER.fullmatch(folder):
+ raise Refused("only r01-* dump folders are accepted")
+ return HOST + BASE + folder + "/"
+
+
+def listing(client, headers, folder):
+ url = folder_url(folder)
+ request = urllib.request.Request(
+ url,
+ method="PROPFIND",
+ headers=dict(headers, Depth="1", **{"Content-Type": "application/xml"}),
+ data=b'',
+ )
+ with client.open(request, timeout=60) as response:
+ if response.status != 207 or response.url != url:
+ raise Refused("listing refused")
+ raw = response.read(LIMIT + 1)
+ if len(raw) > LIMIT:
+ raise Refused("listing too large")
+ files = {}
+ for item in ET.fromstring(raw).findall("{DAV:}response"):
+ path = urllib.parse.unquote(
+ urllib.parse.urlsplit(item.findtext("{DAV:}href") or "").path
+ )
+ if not path.startswith(BASE + folder + "/"):
+ raise Refused("unexpected path in listing")
+ name = path[len(BASE + folder + "/") :].rstrip("/")
+ if not name:
+ continue
+ size = None
+ for row in item.findall("{DAV:}propstat"):
+ if " 200 " in (row.findtext("{DAV:}status") or ""):
+ if (
+ row.find("{DAV:}prop/{DAV:}resourcetype/{DAV:}collection")
+ is not None
+ ):
+ raise Refused("unexpected subfolder")
+ size = row.findtext("{DAV:}prop/{DAV:}getcontentlength")
+ if not DUMP.fullmatch(name) or size is None or not size.isdigit():
+ raise Refused("unexpected entry in folder; nothing planned")
+ files[name] = int(size)
+ return files
+
+
+def select(files, keep_daily, keep_weekly):
+ keep, groups = set(), {}
+ for name in files:
+ match = DUMP.fullmatch(name)
+ groups.setdefault(match["prefix"], []).append((match["stamp"], name))
+ for entries in groups.values():
+ entries.sort(reverse=True)
+ keep.update(name for _, name in entries[: max(1, keep_daily)])
+ weeks = []
+ for stamp, name in entries:
+ week = datetime.strptime(stamp, "%Y%m%dT%H%M%SZ").isocalendar()[:2]
+ if week not in weeks:
+ weeks.append(week)
+ if len(weeks) <= keep_weekly:
+ keep.add(name)
+ return sorted(keep), sorted(set(files) - keep)
+
+
+def summary(files, keep, delete):
+ groups = {}
+ for name in files:
+ prefix = DUMP.fullmatch(name)["prefix"]
+ g = groups.setdefault(
+ prefix,
+ {
+ "files": 0,
+ "bytes": 0,
+ "keep": 0,
+ "delete": 0,
+ "delete_bytes": 0,
+ "newest": None,
+ "newest_bytes": 0,
+ "oldest": None,
+ },
+ )
+ g["files"] += 1
+ g["bytes"] += files[name]
+ g["keep" if name in keep else "delete"] += 1
+ if name in delete:
+ g["delete_bytes"] += files[name]
+ stamp = DUMP.fullmatch(name)["stamp"]
+ if g["newest"] is None or stamp > g["newest"]:
+ g["newest"], g["newest_bytes"] = stamp, files[name]
+ if g["oldest"] is None or stamp < g["oldest"]:
+ g["oldest"] = stamp
+ return groups
+
+
+def gib(n):
+ return f"{n / 1024**3:.1f} GiB"
+
+
+def plan(args, client, secret):
+ files = listing(client, auth(secret), args.folder)
+ keep, delete = select(files, args.keep_daily, args.keep_weekly)
+ record = {
+ "schema": "netkingdom.storage-share-dump-prune-plan.v1",
+ "folder": args.folder,
+ "created_at": datetime.now(UTC).isoformat(),
+ "keep_daily": args.keep_daily,
+ "keep_weekly": args.keep_weekly,
+ "keep": {n: files[n] for n in keep},
+ "delete": {n: files[n] for n in delete},
+ }
+ raw = (json.dumps(record, indent=2, sort_keys=True) + "\n").encode()
+ fd = os.open(
+ Path(args.plan).expanduser(),
+ os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+ 0o600,
+ )
+ with os.fdopen(fd, "wb") as out:
+ out.write(raw)
+ for prefix, g in sorted(summary(files, set(keep), set(delete)).items()):
+ print(
+ f"{prefix}: {g['files']} files, {gib(g['bytes'])}, {g['oldest']} .. {g['newest']} "
+ f"(newest {g['newest_bytes'] / 1024**2:.0f} MiB); keep {g['keep']}, delete {g['delete']} "
+ f"freeing {gib(g['delete_bytes'])}"
+ )
+ print(
+ f"total: keep {len(keep)}, delete {len(delete)} freeing {gib(sum(record['delete'].values()))}"
+ )
+ print(f"plan sha256: {hashlib.sha256(raw).hexdigest()}")
+ return 0
+
+
+def delete(args, client, secret):
+ raw = Path(args.plan).expanduser().read_bytes()
+ if hashlib.sha256(raw).hexdigest() != args.confirm:
+ raise Refused("plan sha256 does not match --confirm")
+ record = json.loads(raw)
+ if (
+ record.get("schema") != "netkingdom.storage-share-dump-prune-plan.v1"
+ or not record["keep"]
+ ):
+ raise Refused("not a prune plan, or it keeps nothing")
+ headers = auth(secret)
+ files = listing(client, headers, record["folder"])
+ if any(files.get(n) != size for n, size in record["delete"].items()):
+ raise Refused("a planned file changed or vanished; make a new plan")
+ if any(n not in files for n in record["keep"]):
+ raise Refused("a kept file is missing; make a new plan")
+ base = folder_url(record["folder"])
+ deleted, freed = [], 0
+ for name in sorted(record["delete"]):
+ url = base + urllib.parse.quote(name)
+ with client.open(
+ urllib.request.Request(url, method="DELETE", headers=headers), timeout=60
+ ) as response:
+ if response.status not in (200, 204):
+ raise Refused("delete refused")
+ try:
+ client.open(
+ urllib.request.Request(url, method="HEAD", headers=headers), timeout=30
+ )
+ raise Refused("file still present after delete")
+ except urllib.error.HTTPError as error:
+ if error.code != 404:
+ raise Refused("unexpected status after delete") from None
+ deleted.append(name)
+ freed += record["delete"][name]
+ remaining = listing(client, headers, record["folder"])
+ receipt = {
+ "schema": "netkingdom.storage-share-dump-prune.v1",
+ "folder": record["folder"],
+ "plan_sha256": args.confirm,
+ "finished_at": datetime.now(UTC).isoformat(),
+ "deleted": len(deleted),
+ "bytes_freed": freed,
+ "kept_all_present": all(n in remaining for n in record["keep"]),
+ "remaining_files": len(remaining),
+ "credential_recorded": False,
+ }
+ fd = os.open(
+ Path(args.receipt).expanduser(),
+ os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,
+ 0o600,
+ )
+ with os.fdopen(fd, "w") as out:
+ out.write(json.dumps(receipt, indent=2, sort_keys=True) + "\n")
+ print(
+ f"deleted {len(deleted)} files, freed {gib(freed)}; kept files present: {receipt['kept_all_present']}"
+ )
+ return 0 if receipt["kept_all_present"] else 1
+
+
+def main(argv=None, client=None, secret=None):
+ parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+ sub = parser.add_subparsers(dest="command", required=True)
+ p = sub.add_parser("plan")
+ p.add_argument("--folder", required=True)
+ p.add_argument("--plan", required=True)
+ p.add_argument("--keep-daily", type=int, default=7)
+ p.add_argument("--keep-weekly", type=int, default=4)
+ d = sub.add_parser("delete")
+ d.add_argument("--plan", required=True)
+ d.add_argument("--confirm", required=True)
+ d.add_argument("--receipt", required=True)
+ args = parser.parse_args(argv)
+ try:
+ if args.command == "plan":
+ folder_url(args.folder)
+ if args.keep_daily < 1 or args.keep_weekly < 0:
+ raise Refused("keep at least one daily dump")
+ return (plan if args.command == "plan" else delete)(
+ args, client or opener(), secret or prompt()
+ )
+ except Refused as error:
+ print(f"refused: {error}", file=sys.stderr)
+ except urllib.error.HTTPError as error:
+ print(f"refused: provider answered HTTP {error.code}", file=sys.stderr)
+ except (urllib.error.URLError, OSError, ET.ParseError):
+ print("refused: provider or local I/O failure", file=sys.stderr)
+ return 1
+
+
+if __name__ == "__main__":
+ sys.exit(main())