fix(updater): retain two verified backup generations

This commit is contained in:
2026-10-02 15:27:07 +02:00
parent 23d49f7186
commit b01cf2c207
5 changed files with 434 additions and 6 deletions

View File

@@ -17,6 +17,7 @@ from pathlib import Path
import re
import shutil
import signal
import stat
import subprocess
import sys
import time
@@ -34,6 +35,8 @@ LEGACY_MATRIX_SERVICES = {"matrix-db", "matrix-synapse", "matrix-redis", "matrix
"matrix-livekit", "matrix-rtc-jwt", "matrix-turn", "matrix-well-known"}
CORE_SERVICES = {"backend", "frontend", "db", "traefik", "node-exporter"}
IMAGE_ID = re.compile(r"sha256:[a-f0-9]{64}\Z")
UPDATE_JOB = re.compile(r"[0-9]{8}-[0-9]{6}-[a-f0-9]{8}\Z")
BACKUPS_TO_KEEP = 2
def compose_strings(value, *, escape=False):
@@ -882,6 +885,137 @@ class Updater:
raise
time.sleep(2)
def _backup_manifest(self, job, mounts, *, verify=False):
"""Only updater-owned regular backup files may be retained or removed."""
directory = job / "backup"
device = self.home.stat().st_dev
for path in (job, directory):
info = path.lstat()
if not stat.S_ISDIR(info.st_mode) or info.st_dev != device:
raise UpdateError("Backup-Verzeichnis ist kein reguläres lokales Verzeichnis.")
if any(path == directory or directory in path.parents or path in directory.parents for path in mounts):
raise UpdateError("Backup ist eingehängt oder wird von einem Container verwendet.")
files = set()
size = 0
for parent, directories, names in os.walk(directory, followlinks=False):
for name in directories + names:
path = Path(parent) / name
info = path.lstat()
if info.st_dev != device or not (stat.S_ISDIR(info.st_mode) or stat.S_ISREG(info.st_mode)):
raise UpdateError("Backup enthält Links, besondere Dateien oder fremde Dateisysteme.")
if stat.S_ISREG(info.st_mode):
files.add(str(path.relative_to(directory)))
size += info.st_blocks * 512
marker = job / "backup-pruned.json"
if marker.is_symlink() or (marker.exists() and not stat.S_ISREG(marker.lstat().st_mode)):
raise UpdateError("Backup-Löschprotokoll muss eine reguläre Datei sein.")
pending = read_json(marker, {})
if pending.get("status") == "deleting" and not verify:
# Resume an interrupted deletion using the manifest saved before removing any file.
manifest = pending["manifest"]
else:
manifest = read_json(directory / "complete.json")
checksums = manifest["sha256"]
required = {"database.dump", "globals.sql", "files.tar", "files.list", "containers.json",
"config/.env", "config/docker-compose.yml"}
if manifest.get("storageProvider") == "seaweedfs":
required |= {"storage-image.tar", "config/.updates/storage-provider.json"}
if (not isinstance(checksums, dict) or not required <= checksums.keys()
or manifest.get("database") != "database.dump" or manifest.get("files") != "files.tar"):
raise UpdateError("Unvollständiges Backup-Prüfmanifest.")
for name, digest in checksums.items():
path = Path(name)
if (path.is_absolute() or ".." in path.parts or str(path) != name or name == "complete.json"
or not re.fullmatch(r"[a-f0-9]{64}", str(digest))):
raise UpdateError("Ungültiger Pfad oder Prüfsumme im Backup-Manifest.")
if not files <= (checksums.keys() | {"complete.json"}):
raise UpdateError("Backup enthält unbekannte Dateien.")
if verify:
if files != checksums.keys() | {"complete.json"}:
raise UpdateError("Eine Datei der aufzubewahrenden Sicherung fehlt.")
for name, digest in checksums.items():
if file_sha(directory / name) != digest:
raise UpdateError("Prüfsumme einer aufzubewahrenden Sicherung stimmt nicht.")
return manifest, size
def _prune_backups(self, *, dry_run=False):
"""Caller holds the update lock. Never select migration or failed-job backups."""
jobs = []
for job in self.home.iterdir():
if not UPDATE_JOB.fullmatch(job.name) or job.is_symlink() or not job.is_dir():
continue
status_path = job / "status.json"
if status_path.is_symlink() or not status_path.is_file():
continue
state = read_json(job / "status.json", {})
if (state.get("command") == "apply" and state.get("status") == "succeeded"
and state.get("phase") == "complete" and state.get("recoveryRequired") is False
and state.get("maintenance") is False and state.get("job") == str(job)
and state.get("backup") == str(job / "backup")
and ((job / "backup").exists() or (job / "backup").is_symlink())):
started = datetime.datetime.fromisoformat(state["startedAt"])
if started.tzinfo is None:
raise UpdateError("Backup-Auftrag hat keine eindeutige Zeitangabe.")
jobs.append((started, job, state))
jobs.sort(key=lambda item: (item[0], item[1].name), reverse=True)
report = {"status": "preview" if dry_run else "complete", "keep": BACKUPS_TO_KEEP,
"checkedAt": now(), "retained": [], "removed": [], "planned": [], "skipped": []}
if not jobs:
return report
installed = read_json(self.home / "installed.json", {})
latest = jobs[0][2]
if (installed.get("manifest", {}).get("version") != latest.get("version")
or installed.get("manifest", {}).get("commit") != latest.get("revision")
or installed.get("composeSha256") != file_sha(self.root / "docker-compose.yml")):
raise UpdateError("Neueste erfolgreiche Sicherung gehört nicht zum installierten Stand; nichts gelöscht.")
# Include same-device bind mounts, which os.path.ismount alone cannot detect.
mounts = []
for line in Path("/proc/self/mountinfo").read_text().splitlines():
value = re.sub(r"\\([0-7]{3})", lambda m: chr(int(m[1], 8)), line.split()[4])
path = Path(value)
if path != Path("/") and (path == self.home or self.home in path.parents):
mounts.append(path)
for cid in self.runner.run(["docker", "ps", "--all", "--quiet"]).split():
for mount in self.inspect(cid).get("Mounts", []):
path = Path(mount["Source"]).resolve()
# A read-only host-root metrics mount does not consume backup data.
if path == Path("/") and not mount.get("RW", True):
continue
mounts.append(path)
# Check both retained generations before deleting any older generation.
for _, job, _ in jobs[:BACKUPS_TO_KEEP]:
self._backup_manifest(job, mounts, verify=True)
report["retained"].append(str(job / "backup"))
for _, job, _ in jobs[BACKUPS_TO_KEEP:]:
try:
manifest, size = self._backup_manifest(job, mounts)
item = {"backup": str(job / "backup"), "bytes": size}
if dry_run:
report["planned"].append(item)
continue
marker = {"status": "deleting", "startedAt": now(), "manifest": manifest,
"retained": report["retained"], **item}
write_json(job / "backup-pruned.json", marker)
shutil.rmtree(job / "backup")
write_json(job / "backup-pruned.json", {**marker, "status": "removed", "completedAt": now()})
report["removed"].append(item)
except (OSError, ValueError, KeyError, TypeError, UpdateError) as error:
report["skipped"].append({"job": str(job), "reason": str(error) if isinstance(error, UpdateError) else type(error).__name__})
if report["skipped"]:
report["status"] = "warning"
if not dry_run:
write_json(self.home / "backup-retention.json", report)
return report
def prune_backups(self, *, dry_run=False):
with self.lock():
state = read_json(self.home / "status.json", {})
if (state.get("status") in ("queued", "running") or state.get("recoveryRequired")
or state.get("maintenance") or state.get("phase") == "storage_migration"):
raise UpdateError("Aktiven oder unterbrochenen Update-/Migrationsauftrag zuerst abschließen.")
self.runner = self.runner_factory(self.home / "backup-retention.log")
return self._prune_backups(dry_run=dry_run)
def apply(self, accept_compose):
if not self.managed and not accept_compose:
raise UpdateError("Compose-Datei ist noch nicht verwaltet oder wurde bearbeitet. compose.diff prüfen, Anpassungen in die Override-Datei übertragen und --accept-compose-replacement verwenden.")
@@ -973,7 +1107,6 @@ class Updater:
self.save("checked", status="checked")
else:
self.apply(accept_compose)
return {**self.state, "plan": plan}
except BaseException as error:
if self.state.get("maintenance") and not self.state.get("mutationStarted"):
try:
@@ -989,6 +1122,25 @@ class Updater:
self.save(status="interrupted" if isinstance(error, (KeyboardInterrupt, SystemExit)) else "failed",
error=str(error) if isinstance(error, UpdateError) else type(error).__name__)
raise
# Retention is post-success maintenance, outside the update/recovery exception path.
# An I/O error or shutdown here must never stop the newly verified application.
if command == "apply":
try:
retention = self._prune_backups()
self.save(backupRetention=retention)
with self.runner.log.open("a") as log:
log.write(f"{now()} Backup-Aufbewahrung: {len(retention['retained'])} behalten, "
f"{len(retention['removed'])} entfernt, {len(retention['skipped'])} übersprungen.\n")
except Exception as error:
with contextlib.suppress(Exception):
warning = {"status": "warning", "checkedAt": now(), "keep": BACKUPS_TO_KEEP,
"error": str(error) if isinstance(error, UpdateError) else type(error).__name__}
write_json(self.home / "backup-retention.json", warning)
self.save(backupRetention=warning)
with self.runner.log.open("a") as log:
log.write(f"{now()} Backup-Bereinigung nicht abgeschlossen; Update erfolgreich. "
"Auf dem Host prune-backups --dry-run prüfen.\n")
return {**self.state, "plan": plan}
def resolve(self, version):
if not version or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9.+-]{0,100}", version):
@@ -1073,13 +1225,16 @@ def main(argv=None):
print("Python ab Version 3.10 wird benötigt.", file=sys.stderr)
return 1
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("command", choices=("check", "apply", "status", "resolve", "adopt-storage"))
parser.add_argument("command", choices=("check", "apply", "status", "resolve", "adopt-storage", "prune-backups"))
parser.add_argument("--directory", default=os.environ.get("FEDEO_DIR", "/opt/fedeo"))
parser.add_argument("--version", help="Zielversion, z. B. 3.1.0; check wählt sonst den neuesten stabilen Release")
parser.add_argument("--accept-compose-replacement", action="store_true", help="Zuvor geprüften Austausch einer nicht verwalteten Compose-Datei akzeptieren")
parser.add_argument("--dry-run", action="store_true", help="Nur für prune-backups: Sicherungen prüfen und Löschvorschau ausgeben")
args = parser.parse_args(argv)
if args.command == "apply" and not args.version:
parser.error("apply benötigt eine explizite --version; zuerst check ausführen")
if args.dry_run and args.command != "prune-backups":
parser.error("--dry-run ist nur für prune-backups verfügbar")
os.umask(0o077)
signal.signal(signal.SIGTERM, lambda *_: sys.exit(143))
try:
@@ -1090,6 +1245,8 @@ def main(argv=None):
result = updater.resolve(args.version)
elif args.command == "adopt-storage":
result = updater.adopt_storage()
elif args.command == "prune-backups":
result = updater.prune_backups(dry_run=args.dry_run)
else:
result = updater.execute(args.command, args.version, args.accept_compose_replacement)
print(json.dumps(result, ensure_ascii=False, indent=2))

View File

@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Real migration -> adoption -> two updates, including a backup restore, on synthetic data.
"""Real migration -> adoption -> three updates, retention and restore on synthetic data.
Requires local storage test images and registry:2; never contacts a FEDEO instance.
Release HTTP/health responses are fixtures; Compose, registry, SDK, PostgreSQL,
@@ -101,7 +101,8 @@ def run():
docker('push', image['repository'] + ':fixture')
# Copy health, networking and SDK settings from the original fixture, but start the new image's migration script.
updater.releases = Releases(config, images)
for version in ('3.4.0', '3.5.0'):
backups = []
for version in ('3.4.0', '3.5.0', '3.6.0'):
http.version = version
checked = updater.execute('check', version)
assert checked['plan']['storageProvider'] == 'seaweedfs'
@@ -112,6 +113,15 @@ def run():
old_matrix = migration.inspect(retained)
assert not old_matrix['State']['Running'] and old_matrix['HostConfig']['RestartPolicy']['Name'] == 'no'
backup = Path(result['backup'])
backups.append(backup)
retention = result['backupRetention']
assert retention['status'] == 'complete', retention
assert len(retention['retained']) == min(len(backups), 2)
if len(backups) == 3:
assert not backups[0].exists()
assert u.read_json(backups[0].parent / 'backup-pruned.json')['status'] == 'removed'
assert backups[1].is_dir() and backups[2].is_dir()
assert (root / '.updates/storage-migration/backup/complete.json').is_file()
with tarfile.open(backup / 'files.tar') as archive:
assert archive.extractfile('matrix/postgres/retained.txt').read() == b'retired Matrix'
assert 'storage/config/s3.json' in archive.getnames()
@@ -136,7 +146,8 @@ def run():
migration.helper('restore-verify', endpoint='http://' + restore_name + ':9000')
finally:
docker('rm', '-f', restore_name)
print(json.dumps({'result': 'passed', 'restoredUpdaterBackup': str(backup), 'evidence': str(root)}), flush=True)
print(json.dumps({'result': 'passed', 'retention': 'two_generations_verified',
'restoredUpdaterBackup': str(backup), 'evidence': str(root)}), flush=True)
except BaseException:
print('Failure evidence: ' + str(root), flush=True)
for path in sorted((root / '.updates').glob('*/commands.log')):

View File

@@ -0,0 +1,190 @@
import json
from pathlib import Path
import shutil
import tempfile
import unittest
from unittest.mock import patch
from test_updater import u
class RetentionRunner:
def __init__(self, log, owner):
self.log, self.owner = log, owner
def run(self, command):
if command[:2] == ['docker', 'ps']:
return 'container' if self.owner.mounts else ''
if command[:2] == ['docker', 'inspect']:
return json.dumps([{'Mounts': self.owner.mounts}])
raise AssertionError(command)
class RetentionTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
(self.root / 'docker-compose.yml').write_text('unchanged compose')
self.mounts = []
self.updater = u.Updater(self.root, runner_factory=lambda log: RetentionRunner(log, self))
self.jobs = [self.backup(i) for i in range(1, 5)]
latest = u.read_json(self.jobs[-1] / 'status.json')
u.write_json(self.updater.home / 'installed.json', {
'manifest': {'version': latest['version'], 'commit': latest['revision']},
'composeSha256': u.file_sha(self.root / 'docker-compose.yml')})
u.write_json(self.updater.home / 'status.json', latest)
def backup(self, index, *, name=None, status='succeeded', command='apply'):
job = self.updater.home / (name or f'202601{index:02d}-120000-{index:08x}')
directory = job / 'backup'
directory.mkdir(parents=True)
files = ['database.dump', 'globals.sql', 'files.tar', 'files.list', 'containers.json',
'config/.env', 'config/docker-compose.yml']
for name in files:
path = directory / name
path.parent.mkdir(exist_ok=True)
path.write_text(f'{index}: {name}')
u.write_json(directory / 'complete.json', {
'database': 'database.dump', 'files': 'files.tar', 'storageProvider': 'minio',
'sha256': {name: u.file_sha(directory / name) for name in files}})
u.write_json(job / 'status.json', {
'command': command, 'status': status, 'phase': 'complete', 'job': str(job),
'backup': str(directory), 'maintenance': False, 'recoveryRequired': False,
'version': f'3.{index}.0', 'revision': str(index) * 40,
'startedAt': f'2026-01-{index:02d}T12:00:00+00:00'})
(job / 'commands.log').write_text('audit log stays')
return job
def test_keep_two_and_preserve_failed_migration_and_audit_records(self):
special = [self.backup(5, status='failed'), self.backup(6, name='storage-migration'),
self.backup(7, name='foreign'), self.backup(8, command='check')]
report = self.updater.prune_backups()
self.assertEqual(report['status'], 'complete')
self.assertEqual(len(report['removed']), 2)
self.assertEqual(len(report['retained']), 2)
for job in self.jobs[:2]:
self.assertFalse((job / 'backup').exists())
self.assertEqual(u.read_json(job / 'backup-pruned.json')['status'], 'removed')
self.assertTrue((job / 'commands.log').is_file())
for job in self.jobs[2:] + special:
self.assertTrue((job / 'backup/complete.json').is_file())
self.assertEqual(self.updater.prune_backups()['removed'], [])
def test_dry_run_checks_but_does_not_delete_or_write_markers(self):
result = self.updater.prune_backups(dry_run=True)
self.assertEqual(len(result['planned']), 2)
self.assertFalse((self.updater.home / 'backup-retention.json').exists())
for job in self.jobs:
self.assertTrue((job / 'backup').is_dir())
self.assertFalse((job / 'backup-pruned.json').exists())
def test_corrupt_either_retained_generation_prevents_all_deletions(self):
for job in self.jobs[2:]:
with self.subTest(job=job.name):
path = job / 'backup/database.dump'
original = path.read_bytes()
path.write_bytes(b'corrupt')
with self.assertRaisesRegex(u.UpdateError, 'Prüfsumme'):
self.updater.prune_backups()
self.assertTrue(all((p / 'backup').exists() for p in self.jobs))
path.write_bytes(original)
def test_incomplete_retained_backup_prevents_deletion(self):
(self.jobs[-1] / 'backup/globals.sql').unlink()
with self.assertRaisesRegex(u.UpdateError, 'fehlt'):
self.updater.prune_backups()
self.assertTrue((self.jobs[0] / 'backup').exists())
def test_seaweedfs_requires_image_and_provider_snapshot(self):
path = self.jobs[-1] / 'backup/complete.json'
manifest = u.read_json(path)
manifest['storageProvider'] = 'seaweedfs'
u.write_json(path, manifest)
with self.assertRaisesRegex(u.UpdateError, 'Unvollständiges'):
self.updater.prune_backups()
self.assertTrue((self.jobs[0] / 'backup').exists())
def test_unknown_files_and_links_are_preserved(self):
(self.jobs[0] / 'backup/foreign.txt').write_text('do not delete')
outside = self.root / 'outside'
outside.write_text('keep')
(self.jobs[1] / 'backup/files.tar').unlink()
(self.jobs[1] / 'backup/files.tar').symlink_to(outside)
result = self.updater.prune_backups()
self.assertEqual(len(result['skipped']), 2)
self.assertEqual(result['removed'], [])
self.assertEqual(outside.read_text(), 'keep')
def test_symlinked_backup_directory_is_not_followed(self):
backup = self.jobs[0] / 'backup'
outside = self.root / 'outside'
backup.rename(outside)
backup.symlink_to(outside, target_is_directory=True)
result = self.updater.prune_backups()
self.assertTrue(outside.is_dir())
self.assertTrue(backup.is_symlink())
self.assertEqual(len(result['skipped']), 1)
def test_manifest_path_escape_is_rejected(self):
path = self.jobs[-1] / 'backup/complete.json'
manifest = u.read_json(path)
manifest['sha256']['../outside'] = 'a' * 64
u.write_json(path, manifest)
with self.assertRaisesRegex(u.UpdateError, 'Ungültiger Pfad'):
self.updater.prune_backups()
self.assertTrue((self.jobs[0] / 'backup').exists())
def test_container_mount_protects_backup_but_metrics_root_mount_does_not(self):
self.mounts = [{'Source': '/', 'RW': False},
{'Source': str(self.jobs[0] / 'backup/files.tar'), 'RW': False}]
result = self.updater.prune_backups()
self.assertEqual(len(result['skipped']), 1)
self.assertTrue((self.jobs[0] / 'backup').exists())
self.assertFalse((self.jobs[1] / 'backup').exists())
def test_same_device_bind_mount_is_protected(self):
read_text = Path.read_text
mount = self.jobs[0] / 'backup/config'
def read(path, *args, **kwargs):
if path == Path('/proc/self/mountinfo'):
return f'1 2 3:4 / {mount} rw - ext4 /dev/test rw\n'
return read_text(path, *args, **kwargs)
with patch.object(Path, 'read_text', read):
result = self.updater.prune_backups()
self.assertEqual(len(result['skipped']), 1)
self.assertTrue((self.jobs[0] / 'backup').exists())
def test_partial_deletion_can_resume_without_touching_retained_backups(self):
original = shutil.rmtree
def fail(path, *args, **kwargs):
if path == self.jobs[0] / 'backup':
(path / 'complete.json').unlink()
(path / 'database.dump').unlink()
raise OSError('simulated disk error')
return original(path, *args, **kwargs)
with patch.object(u.shutil, 'rmtree', fail):
result = self.updater.prune_backups()
self.assertEqual(len(result['skipped']), 1)
self.assertEqual(u.read_json(self.jobs[0] / 'backup-pruned.json')['status'], 'deleting')
result = self.updater.prune_backups()
self.assertEqual(len(result['removed']), 1)
self.assertTrue(all((p / 'backup').exists() for p in self.jobs[2:]))
def test_recovery_running_and_lock_block_cleanup(self):
for state in ({'recoveryRequired': True}, {'maintenance': True}, {'status': 'running'},
{'phase': 'storage_migration'}):
with self.subTest(state=state):
u.write_json(self.updater.home / 'status.json', state)
with self.assertRaises(u.UpdateError):
self.updater.prune_backups()
u.write_json(self.updater.home / 'status.json', {'status': 'failed', 'recoveryRequired': False})
with self.updater.lock(), self.assertRaises(u.UpdateError):
self.updater.prune_backups()
self.assertEqual(len(self.updater.prune_backups()['removed']), 2)
def test_changed_installation_blocks_cleanup(self):
(self.root / 'docker-compose.yml').write_text('manually restored')
with self.assertRaisesRegex(u.UpdateError, 'installierten Stand'):
self.updater.prune_backups()
self.assertTrue(all((p / 'backup').exists() for p in self.jobs))

View File

@@ -646,6 +646,39 @@ class UpdateTests(unittest.TestCase):
self.assertFalse((self.root / '.updates/installed.json').exists())
self.assertEqual(self.runner.calls[-1][-3:], ['backend', 'frontend', 'traefik'])
def test_backup_retention_runs_only_after_successful_public_verification(self):
def prune():
self.assertEqual(self.updater.status()['status'], 'succeeded')
self.assertTrue(any(url.endswith('/api/health') for url in self.http.requests))
return {'retained': [], 'removed': [], 'skipped': []}
with patch.object(self.updater, '_prune_backups', side_effect=prune) as cleanup:
self.run_update()
cleanup.assert_called_once()
def test_failed_update_never_prunes_backups(self):
self.http.bad_health = True
with patch.object(self.updater, '_prune_backups') as cleanup:
with self.assertRaises(u.UpdateError):
self.run_update()
cleanup.assert_not_called()
def test_retention_error_does_not_fail_update_or_stop_the_new_application(self):
with patch.object(self.updater, '_prune_backups', side_effect=OSError('disk error')):
result = self.run_update()
self.assertEqual(result['status'], 'succeeded')
self.assertEqual(result['backupRetention']['status'], 'warning')
self.assertEqual(u.read_json(self.updater.home / 'backup-retention.json')['status'], 'warning')
self.assertFalse(result['recoveryRequired'])
self.assertEqual(sum('stop' in args for args in self.runner.calls), 1)
def test_interrupt_during_retention_preserves_completed_update(self):
with patch.object(self.updater, '_prune_backups', side_effect=KeyboardInterrupt):
with self.assertRaises(KeyboardInterrupt):
self.run_update()
self.assertEqual(self.updater.status()['status'], 'succeeded')
self.assertFalse(self.updater.status()['recoveryRequired'])
self.assertEqual(sum('stop' in args for args in self.runner.calls), 1)
def test_resolve_requires_a_working_manually_recovered_installation(self):
self.runner.fail = lambda args: 'migrate' in args
with self.assertRaises(u.UpdateError):