WIP: Upgrade script

securedrop/debian/config/usr/bin/securedrop-noble-migration.py

@@ -0,0 +1,75 @@
+#!/usr/bin/python3
+"""
+Migrate a SecureDrop server from focal to noble
+
+This script should never be run directly, only via the systemd service.
+"""
+
+import json
+import os
+import subprocess
+import sys
+import time
+import traceback
+from enum import Enum
+from pathlib import Path
+
+if os.getuid() != 0:
+    print("You need to run this as root")
+    sys.exit(1)
+
+
+class Stage(Enum):
+    PENDING_UPDATES = 1
+    MIGRATION_CHECK = 2
+    BACKUP = 3
+    SUSPEND_OSSEC = 4
+    DISABLE_UNATTENDED_UPDATES = 5
+    CHANGE_APT_SOURCES = 6
+    APT_GET_UPDATE = 7
+    APT_GET_UPGRADE_NO_NEW = 8
+    APT_GET_FULL_UPGRADE = 9
+    REENABLE_UNATTENDED_UPDATES = 10
+    RESUME_OSSEC = 11
+    REBOOT = 12
+    INTEGRITY_CHECK = 13
+    REMOVE_BACKUP = 14
+    DONE = 15
+
+
+def pending_updates() -> None:
+    subprocess.check_call(["apt-get", "update"])
+    subprocess.check_call(["unattended-upgrade"])
+
+
+def migration_check() -> None:
+    subprocess.check_call(["systemctl", "start", "securedrop-noble-migration-check"])
+    while True:
+        result = subprocess.check_output(
+            ["systemctl", "is-active", "securedrop-noble-migration-check"], text=True
+        )
+        if result.stdout.strip() != "active":
+            break
+        time.sleep(1)
+
+    check_data = json.loads(Path("/etc/securedrop-noble-migration-check.json").read_text())
+
+    if "error" in check_data:
+        raise RuntimeError(f"Migration check failed: {check_data['error']}")
+    if not all(check_data.values()):
+        raise RuntimeError("Migration check failed; run directly to see details.")
+
+
+def backup() -> None:
+    # Create a root-only directory to store the backup
+    os.mkdirs("/var/lib/backups/", mode=0o700, exist_ok=True)
+    subprocess.check_call(["/usr/bin/securedrop-app-backup.py", "--dest", "/var/lib/backups/"])
+
+
+def main() -> None:
+    try:
+        data = json.loads(Path("/etc/securedrop-upgrade.json").read_text())
+        state = Stage(data["stage"])
+    except Exception:
+        traceback.print_exc()
+        state = Stage.PENDING_UPDATES