1#!/usr/bin/env python3
2import argparse
3from contextlib import closing, contextmanager
4from datetime import datetime, timezone
5import hashlib
6import json
7import os
8from pathlib import Path
9import re
10import secrets
11import shutil
12import sqlite3
13import subprocess
14import time
15
16
17ROOT = Path("/opt/studio")
18STATE = Path("/var/lib/studio")
19BACKUPS = STATE / "backups"
20BACKUP_ID = re.compile(r"\d{8}T\d{6}Z-[0-9a-f]{6}\Z")
21SERVICE_ID = re.compile(r"[a-z][a-z0-9-]*\Z")
22RELEASE_ID = re.compile(r"[0-9a-f]{16}\Z")
23PODMAN = ["podman", "--url", "unix:///run/podman/podman.sock"]
24
25
26def run(*args, capture=False, **kwargs):
27 return subprocess.run(args, check=True, capture_output=capture, **kwargs)
28
29
30def current_release():
31 current = ROOT / "current"
32 if not current.is_symlink():
33 raise ValueError("No active home server release")
34 release = current.resolve()
35 if release.parent != ROOT / "releases" or not RELEASE_ID.fullmatch(release.name):
36 raise ValueError("Active home server release is invalid")
37 return release
38
39
40def postgres_container():
41 allocations = json.loads(run("nomad", "job", "allocs", "-json", "postgres", capture=True, text=True).stdout)
42 running = [a["ID"] for a in allocations if a["ClientStatus"] == "running" and a["DesiredStatus"] == "run"]
43 if len(running) != 1:
44 raise ValueError("Postgres needs one running allocation")
45 containers = run(*PODMAN, "ps", "--format", "{{.ID}} {{.Names}}", capture=True, text=True).stdout.splitlines()
46 matches = [parts[0] for line in containers if len(parts := line.split()) == 2 and parts[1].endswith(running[0])]
47 if len(matches) != 1:
48 raise ValueError("Postgres container is unavailable")
49 return matches[0]
50
51
52def database_inputs(service):
53 prefix = f"nomad/jobs/{service}/inputs/"
54 listed = json.loads(run("nomad", "var", "list", "-out=json", prefix, capture=True, text=True).stdout)
55 databases = {}
56 for item in listed:
57 path = item["Path"]
58 alias = path.removeprefix(prefix)
59 if not path.startswith(prefix) or not SERVICE_ID.fullmatch(alias):
60 raise ValueError(f"Invalid input path for {service}: {path}")
61 values = json.loads(run("nomad", "var", "get", "-out=json", path, capture=True, text=True).stdout)["Items"]
62 if "name" not in values or "username" not in values:
63 continue
64 name, owner = values["name"], values["username"]
65 if not owner.startswith("svc_"):
66 continue
67 if not re.fullmatch(r"[a-z][a-z0-9_]*", name) or owner != "svc_" + name:
68 raise ValueError(f"Invalid database allocation for {service}.{alias}")
69 if any(database == name for database, _ in databases.values()):
70 raise ValueError(f"Database allocated twice for {service}: {name}")
71 databases[alias] = (name, owner)
72 return databases
73
74
75def database_dump(container, database, destination, extensions):
76 with destination.open("wb") as output:
77 run(*PODMAN, "exec", container, "pg_dump", "-U", "postgres", "-Fc", "--no-owner", "--no-acl",
78 *(f"--exclude-extension={extension}" for extension in extensions), database, stdout=output)
79 if not destination.stat().st_size:
80 raise ValueError(f"Empty database backup: {database}")
81
82
83def checksum(path):
84 digest = hashlib.sha256()
85 with path.open("rb") as file:
86 for chunk in iter(lambda: file.read(1024 * 1024), b""):
87 digest.update(chunk)
88 return digest.hexdigest()
89
90
91def dataset_for(service):
92 root = Path("/srv/prod") / service
93 result = subprocess.run(["findmnt", "-n", "-o", "SOURCE,FSTYPE", "--mountpoint", str(root)], capture_output=True, text=True)
94 if result.returncode:
95 return None
96 source, fstype = result.stdout.split()
97 if fstype != "zfs" or not source.endswith("/prod/" + service):
98 raise ValueError(f"Unexpected dataset at {root}: {source}")
99 return source
100
101
102@contextmanager
103def cloned_postgres(snapshot, clone, mountpoint):
104 live_container = postgres_container()
105 inspect = json.loads(run(*PODMAN, "inspect", live_container, capture=True, text=True).stdout)[0]
106 mounts = [mount for mount in inspect["Mounts"] if mount["Destination"] == "/var/lib/postgresql"]
107 if len(mounts) != 1 or mounts[0]["Type"] != "bind":
108 raise ValueError("Postgres data mount is unavailable")
109 data_path = Path(mounts[0]["Source"])
110 if not data_path.is_relative_to("/srv/prod/postgres"):
111 raise ValueError("Postgres data is outside its dataset")
112 user = inspect["Config"]["User"]
113 if not re.fullmatch(r"[0-9]+:[0-9]+", user):
114 raise ValueError("Postgres container user is invalid")
115 pgdata = [env for env in inspect["Config"]["Env"] if env.startswith("PGDATA=")]
116 if len(pgdata) != 1 or not Path(pgdata[0][7:]).is_relative_to("/var/lib/postgresql"):
117 raise ValueError("Postgres data directory is invalid")
118 image = inspect["Image"]
119 if not re.fullmatch(r"[0-9a-f]{64}", image):
120 raise ValueError("Postgres image ID is invalid")
121 probe = "studio-backup-" + clone.rsplit("/", 1)[1]
122 created = False
123 try:
124 run("zfs", "clone", "-o", f"mountpoint={mountpoint}", snapshot, clone)
125 created = True
126 run(*PODMAN, "run", "-d", "--name", probe, "--network", "none", "--user", user,
127 "-e", pgdata[0], "-e", "POSTGRES_PASSWORD=backup-probe",
128 "-v", f"{Path(mountpoint) / data_path.relative_to('/srv/prod/postgres')}:/var/lib/postgresql", image,
129 stdout=subprocess.DEVNULL)
130 for _ in range(60):
131 ready = subprocess.run([*PODMAN, "exec", probe, "pg_isready", "-U", "postgres"], capture_output=True)
132 if ready.returncode == 0:
133 break
134 time.sleep(1)
135 else:
136 raise ValueError("Snapshot PostgreSQL did not become ready")
137 yield probe
138 finally:
139 subprocess.run([*PODMAN, "rm", "-f", probe], check=False, stdout=subprocess.DEVNULL)
140 if created:
141 run("zfs", "destroy", clone)
142
143
144def backup_dashboard(directory):
145 source = STATE / "dashboard"
146 directory.mkdir(mode=0o700)
147 for path in sorted(source.glob("*.sqlite")):
148 target = directory / path.name
149 with closing(sqlite3.connect(path.as_uri() + "?mode=ro", uri=True)) as live, closing(sqlite3.connect(target)) as copy:
150 live.backup(copy)
151 if copy.execute("PRAGMA quick_check").fetchone() != ("ok",):
152 raise ValueError("Dashboard database backup failed its integrity check")
153 if copy.execute("PRAGMA journal_mode=DELETE").fetchone() != ("delete",):
154 raise ValueError("Dashboard backup could not leave WAL mode")
155 target.chmod(0o600)
156 for suffix in ["-wal", "-shm"]:
157 target.with_name(target.name + suffix).unlink(missing_ok=True)
158 if (source / "pictures").exists():
159 shutil.copytree(source / "pictures", directory / "pictures", symlinks=True)
160 paths = sorted(path for path in directory.rglob("*") if path.is_file())
161 if any(path.is_symlink() for path in directory.rglob("*")):
162 raise ValueError("Dashboard backup contains a symlink")
163 return {str(path.relative_to(directory)): checksum(path) for path in paths}
164
165
166def backup(from_release, to_release):
167 if not RELEASE_ID.fullmatch(from_release) or not RELEASE_ID.fullmatch(to_release):
168 raise ValueError("Invalid release ID")
169 if current_release().name != from_release:
170 raise ValueError("Active release changed before backup")
171 names = json.loads((STATE / "managed-jobs.json").read_text())
172 if not names or any(not SERVICE_ID.fullmatch(name) for name in names):
173 raise ValueError("Managed jobs list is invalid")
174 backup_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + secrets.token_hex(3)
175 directory = BACKUPS / backup_id
176 manifest = {"id": backup_id, "fromRelease": from_release, "toRelease": to_release,
177 "time": int(time.time()), "services": {}}
178 snapshot = "studio-" + backup_id
179 datasets = set()
180 allocations = {}
181 for service in sorted(names):
182 entry = {}
183 dataset = dataset_for(service)
184 if dataset:
185 datasets.add(dataset)
186 entry["dataset"] = dataset
187 entry["snapshot"] = snapshot
188 allocated = database_inputs(service)
189 if allocated:
190 entry["databases"] = {}
191 allocations[service] = allocated
192 if entry:
193 manifest["services"][service] = entry
194 if allocations:
195 postgres_dataset = dataset_for("postgres")
196 if not postgres_dataset:
197 raise ValueError("Postgres needs a ZFS dataset for consistent backups")
198 datasets.add(postgres_dataset)
199 directory.mkdir(parents=True, mode=0o700)
200 snapshots = [f"{dataset}@{snapshot}" for dataset in sorted(datasets)]
201 created = False
202 try:
203 if (STATE / "dashboard").is_dir():
204 manifest["services"]["dashboard"] = {"files": backup_dashboard(directory / "dashboard")}
205 if snapshots:
206 run("zfs", "snapshot", *snapshots)
207 created = True
208 nomad_snapshot = directory / "nomad.snap"
209 run("nomad", "operator", "snapshot", "save", str(nomad_snapshot))
210 manifest["nomadSnapshot"] = {"file": nomad_snapshot.name, "sha256": checksum(nomad_snapshot)}
211 if allocations:
212 clone = postgres_dataset.rsplit("/prod/", 1)[0] + "/staging/" + backup_id + "-postgres"
213 mountpoint = Path("/srv/staging") / (backup_id + "-postgres")
214 with cloned_postgres(f"{postgres_dataset}@{snapshot}", clone, mountpoint) as probe:
215 for service, allocated in allocations.items():
216 entry = manifest["services"][service]
217 for alias, (database, owner) in sorted(allocated.items()):
218 result = run(*PODMAN, "exec", probe, "psql", "-U", "postgres", "-d", database,
219 "-At", "-c", "SELECT extname FROM pg_extension WHERE extname <> 'plpgsql' ORDER BY extname",
220 capture=True, text=True)
221 extensions = result.stdout.splitlines()
222 if any(not re.fullmatch(r"[a-z][a-z0-9_]*", extension) for extension in extensions):
223 raise ValueError(f"Invalid extension in {database}")
224 dump = directory / f"{service}-{alias}.dump"
225 database_dump(probe, database, dump, extensions)
226 entry["databases"][alias] = {"name": database, "owner": owner, "extensions": extensions,
227 "dump": dump.name, "sha256": checksum(dump)}
228 pending = directory / "manifest.pending"
229 pending.write_text(json.dumps(manifest, indent=2) + "\n")
230 pending.replace(directory / "manifest.json")
231 except Exception:
232 if created:
233 for target in snapshots:
234 subprocess.run(["zfs", "destroy", target], check=False)
235 shutil.rmtree(directory)
236 raise
237 print(f"backup={backup_id} services={len(manifest['services'])}")
238 return backup_id
239
240
241def list_backups():
242 for path in sorted(BACKUPS.glob("*/manifest.json")):
243 manifest = json.loads(path.read_text())
244 print(f"{manifest['id']} {manifest['fromRelease']} -> {manifest['toRelease']} {', '.join(sorted(manifest['services']))}")
245
246
247def restore(backup_id, service):
248 if not BACKUP_ID.fullmatch(backup_id) or not SERVICE_ID.fullmatch(service):
249 raise ValueError("Invalid backup or service ID")
250 directory = BACKUPS / backup_id
251 manifest = json.loads((directory / "manifest.json").read_text())
252 if manifest["id"] != backup_id or service not in manifest["services"]:
253 raise ValueError("Backup does not contain that service")
254 if current_release().name != manifest["fromRelease"]:
255 raise ValueError(f"Roll back code to {manifest['fromRelease']} before restoring data")
256 if service == "postgres":
257 raise ValueError("Postgres serves multiple services; restore a specific database owner")
258 entry = manifest["services"][service]
259 if service == "dashboard":
260 source = directory / "dashboard"
261 for name, digest in entry["files"].items():
262 path = source / name
263 if not path.resolve().is_relative_to(source.resolve()) or not path.is_file() or checksum(path) != digest:
264 raise ValueError("Dashboard backup is missing or changed")
265 run("systemctl", "stop", "studio-dashboard")
266 safety = BACKUPS / ("before-restore-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + secrets.token_hex(3))
267 safety.mkdir(mode=0o700)
268 files = backup_dashboard(safety / "dashboard")
269 (safety / "manifest.json").write_text(json.dumps({"files": files}) + "\n")
270 root = STATE / "dashboard"
271 for path in root.glob("*.sqlite*"):
272 path.unlink()
273 if (root / "pictures").exists():
274 shutil.rmtree(root / "pictures")
275 shutil.copytree(source, root, dirs_exist_ok=True)
276 run("systemctl", "start", "studio-dashboard")
277 run("systemctl", "is-active", "--quiet", "studio-dashboard")
278 print(f"restored=dashboard backup={backup_id} safety={safety.name}")
279 return
280 dataset = entry.get("dataset")
281 if dataset and dataset_for(service) != dataset:
282 raise ValueError("Service dataset changed since backup")
283 root = Path("/srv/prod") / service
284 source = root / ".zfs/snapshot" / entry["snapshot"] if dataset else None
285 if source and not source.is_dir():
286 raise ValueError("ZFS backup snapshot is unavailable")
287 databases = entry.get("databases", {"database": entry["database"]} if "database" in entry else {})
288 allocated = database_inputs(service)
289 for alias, database in databases.items():
290 if allocated.get(alias) != (database["name"], database["owner"]):
291 raise ValueError(f"Service database changed since backup: {alias}")
292 dump = directory / database["dump"]
293 if not dump.is_file() or checksum(dump) != database["sha256"]:
294 raise ValueError("Database dump is missing or changed")
295 run("nomad", "job", "stop", "-yes", service)
296 for _ in range(30):
297 allocations = json.loads(run("nomad", "job", "allocs", "-json", service, capture=True, text=True).stdout)
298 if not any(a["ClientStatus"] == "running" for a in allocations):
299 break
300 time.sleep(2)
301 else:
302 raise ValueError(f"{service} did not stop; data remains unchanged")
303 safety = "before-restore-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + secrets.token_hex(3)
304 if dataset:
305 run("zfs", "snapshot", f"{dataset}@{safety}")
306 container = postgres_container() if databases else None
307 for alias, database in databases.items():
308 database_dump(container, database["name"], directory / f"{service}-{alias}-{safety}.dump", database["extensions"])
309 if dataset:
310 run("rsync", "-aHAX", "--numeric-ids", "--one-file-system", "--delete", "--exclude=/.zfs/",
311 str(source) + "/", str(root) + "/")
312 for database in databases.values():
313 name, owner = database["name"], database["owner"]
314 run(*PODMAN, "exec", "-i", container, "psql", "-U", "postgres", "-d", "postgres", "-v", "ON_ERROR_STOP=1",
315 input=f"DROP DATABASE {name} WITH (FORCE);\nCREATE DATABASE {name} OWNER {owner};\n", text=True)
316 for extension in database["extensions"]:
317 run(*PODMAN, "exec", "-i", container, "psql", "-U", "postgres", "-d", name, "-v", "ON_ERROR_STOP=1",
318 input=f"CREATE EXTENSION {extension};\n", text=True)
319 with (directory / database["dump"]).open("rb") as input_file:
320 run(*PODMAN, "exec", "-i", container, "pg_restore", "-U", "postgres", "-d", name,
321 "--no-owner", "--no-acl", "--role=" + owner, stdin=input_file)
322 script = current_release() / "tools/studio.py"
323 run("python3", str(script), "deploy", service)
324 run("python3", str(script), "check", service)
325 print(f"restored={service} backup={backup_id} safety={safety}")
326
327
328def main():
329 parser = argparse.ArgumentParser(description="Back up or restore service data on the home server")
330 parser.add_argument("mode", choices=["backup", "list", "restore"])
331 parser.add_argument("first", nargs="?")
332 parser.add_argument("second", nargs="?")
333 parser.add_argument("--discard-writes", action="store_true")
334 args = parser.parse_args()
335 os.umask(0o077)
336 os.environ["NOMAD_TOKEN"] = (STATE / "nomad.token").read_text().strip()
337 if args.mode == "backup" and args.first and args.second and not args.discard_writes:
338 backup(args.first, args.second)
339 elif args.mode == "list" and not args.first and not args.second and not args.discard_writes:
340 list_backups()
341 elif args.mode == "restore" and args.first and args.second and args.discard_writes:
342 restore(args.first, args.second)
343 else:
344 parser.error("Expected backup FROM TO, list, or restore BACKUP SERVICE --discard-writes")
345
346
347if __name__ == "__main__":
348 main()