import json import os from pathlib import Path import re import subprocess from urllib.parse import urlsplit class OperationsError(RuntimeError): pass def snapshot_id(value): if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value): raise OperationsError("A full snapshot ID is required") return value def json_lines(output): return [json.loads(line) for line in output.splitlines() if line.strip()] class ComposeStack: def __init__(self, root, project, env_file, workspace, files=(), local_repository=None, legacy=False, operations=True): if not re.fullmatch(r"[a-z][a-z0-9_-]{0,62}", project): raise OperationsError("Invalid Compose project name") self.root = Path(root).resolve() self.project = project self.operations = operations self.workspace = Path(workspace).resolve() if Path(env_file).is_symlink(): raise OperationsError("The Compose environment file must not be a symbolic link") self.env_file = Path(env_file).resolve() if not self.env_file.is_file(): raise OperationsError("An explicit Compose environment file is required") self.files = [self.root / "compose.yaml", *[Path(file).resolve() for file in files]] if operations: self.files.append(self.root / "compose.backup.yaml") if legacy: self.files.append(self.root / "compose.backup.legacy.yaml") self.environment = {"PATH": os.environ["PATH"], "DOCKER_HOST": "unix:///var/run/docker.sock", "DOCKER_CONFIG": str(self.workspace / "docker-config"), "COMPOSE_DISABLE_ENV_FILE": "1", "POST76_BACKUP_ENV_FILE": str(self.env_file)} if local_repository: repository = Path(local_repository).resolve() repository.mkdir(parents=True, exist_ok=True, mode=0o700) self.environment.update({"POST76_BACKUP_LOCAL_REPOSITORY_DIR": str(repository), "POST76_BACKUP_REPOSITORY": "/repository"}) self.files.append(self.root / "compose.backup.local.yaml") def mounts(self, capture, restore): self.environment["POST76_BACKUP_CAPTURE_DIR"] = str(capture) self.environment["POST76_BACKUP_RESTORE_DIR"] = str(restore) def command(self, arguments, step, output=None): prefix = ["docker", "compose"] if self.operations: prefix += ["--profile", "operations"] prefix += ["--env-file", str(self.env_file), "--project-name", self.project] for file in self.files: prefix += ["-f", str(file)] result = subprocess.run([*prefix, *arguments], cwd=self.root, env=self.environment, stdout=output if output is not None else subprocess.PIPE, stderr=subprocess.PIPE) if result.returncode: raise OperationsError(f"{step} failed (exit {result.returncode}); command output is withheld") return result.stdout if output is None else b"" def model(self): model = json.loads(self.command(["config", "--format", "json"], "Compose configuration")) if not self.operations: return model repository = model["services"]["backup"]["environment"]["RESTIC_REPOSITORY"] if repository != "/repository": url = urlsplit(repository[3:]) if repository.startswith("s3:") else None if not url or url.scheme != "https" or not url.hostname or url.username or url.password or url.query or url.fragment: raise OperationsError("Use a credential-free HTTPS S3 repository or the explicit local repository option") return model def state(self): output = self.command(["ps", "--all", "--format", "json"], "Service status").decode() rows = json.loads(output) if output.lstrip().startswith("[") else json_lines(output) result = {} for row in rows: service = row.get("Service") if service in result: raise OperationsError("Backup requires a single instance of each service") result[service] = row return result def stop(self, services): if services: self.command(["stop", "--timeout", "900", *services], "Pause application services") def resume(self, services): if services: self.command(["up", "--no-build", "--no-deps", "--detach", "--wait", "--wait-timeout", "180", *services], "Resume application services") def dump(self, destination): with Path(destination).open("xb") as output: self.command(["exec", "-T", "mysql", "sh", "/usr/local/bin/post76-mysql-dump"], "Database dump", output) Path(destination).chmod(0o600) if Path(destination).stat().st_size == 0: raise OperationsError("Database dump is empty") def snapshot_redis(self): output = self.command(["exec", "-T", "redis", "sh", "/usr/local/bin/post76-redis", "snapshot"], "Queue snapshot") if output.strip() != b"OK": raise OperationsError("Queue snapshot did not complete") class DockerRepository: def __init__(self, stack, legacy=False): self.stack = stack self.roots = {"capture": "/capture", "uploads": "/uploads", "keys": "/keys", "configuration": "/configuration", "redis": "/redis/dump.rdb"} if legacy: self.roots.update({"legacy_attachments": "/legacy/attachment", "legacy_avatars": "/legacy/avatar"}) def call(self, *arguments): return self.stack.command(["run", "--rm", "--no-deps", "-T", "backup", *arguments], "Encrypted repository operation") def snapshots(self, tags): values = json.loads(self.call("snapshots", "--tag", ",".join(tags))) or [] if not isinstance(values, list): raise OperationsError("Invalid snapshot listing") return values def capture(self, capture, tags): output = self.call("backup", "--read-concurrency", "1", *[item for tag in tags for item in ["--tag", tag]], *self.roots.values(), "--exclude", self.roots["keys"] + "/backup_password", "--exclude", self.roots["keys"] + "/backup_s3_credentials") summaries = [row for row in json_lines(output) if row.get("message_type") == "summary"] if len(summaries) != 1: raise OperationsError("Backup did not produce a complete summary") return snapshot_id(summaries[0].get("snapshot_id")) def verify_tag(self, identity): output = self.call("tag", "--add", "verified", snapshot_id(identity)) changed = [row for row in json_lines(output) if row.get("message_type") == "changed"] if len(changed) != 1: raise OperationsError("Backup verification marker was not committed") return snapshot_id(changed[0].get("new_snapshot_id")) def manifest(self, identity): return json.loads(self.call("dump", snapshot_id(identity), self.roots["capture"] + "/manifest.json")) def nodes(self, identity): return json_lines(self.call("ls", snapshot_id(identity))) def restore(self, identity, destination): self.call("restore", snapshot_id(identity), "--target", "/restore", "--verify", "--overwrite", "never")