#!/usr/bin/env python3 """Filesystem and manifest checks for the privileged backup/restore helpers.""" import hashlib import fnmatch import os import re import shutil import sys import tarfile import time from pathlib import Path, PurePosixPath class BackupTarInfo(tarfile.TarInfo): @classmethod def frombuf(cls, buf, encoding, errors): member = super().frombuf(buf, encoding, errors) return cls._fix_gnu_name(member, buf, encoding, errors) @classmethod def _frombuf(cls, buf, encoding, errors, **kwargs): # Newer CPython security backports call the internal parser directly. member = super()._frombuf(buf, encoding, errors, **kwargs) return cls._fix_gnu_name(member, buf, encoding, errors) @staticmethod def _fix_gnu_name(member, buf, encoding, errors): if buf[257:265] == b"ustar \0": # GNU incremental headers store access/change timestamps where ustar # stores a path prefix. Python's generic parser can prepend that octal # timestamp to every filename. GNU long-name records are applied by # tarfile after this header has been decoded. member.name = buf[:100].split(b"\0", 1)[0].decode(encoding, errors) if member.isdir(): member.name = member.name.rstrip("/") return member def local_path(value: str) -> Path: path = Path(value) if not path.is_absolute() or ".." in path.parts: raise ValueError("Backup destination must be an absolute path without traversal") resolved = path.resolve() roots = (Path("/var/backups"), Path("/srv"), Path("/mnt"), Path("/media")) if not any(root in resolved.parents for root in roots): raise ValueError("Backup destination must be a subdirectory of /var/backups, /srv, /mnt or /media") return resolved def relative_path(value: str) -> str: path = PurePosixPath(value) if path.is_absolute() or ".." in path.parts or not path.parts: raise ValueError(f"Unsafe backup path: {value!r}") return str(path) def verify_manifest(source: Path) -> None: """Every restored payload must be listed; extra files are not authenticated.""" expected = {} for line in (source / "SHA256SUMS").read_text(encoding="utf-8").splitlines(): escaped = line.startswith("\\") if escaped: line = line[1:] match = re.fullmatch(r"([a-fA-F0-9]{64}) [ *](.+)", line) if not match: raise ValueError("Invalid SHA256SUMS entry") digest, name = match.groups() if escaped: name = re.sub(r"\\([\\nr])", lambda m: {"\\": "\\", "n": "\n", "r": "\r"}[m[1]], name) name = relative_path(name) if name in expected or name in {"SHA256SUMS", "SHA256SUMS.sig"}: raise ValueError("Duplicate or self-referencing SHA256SUMS entry") expected[name] = digest.lower() actual = set() for directory, directories, files in os.walk(source, followlinks=False): for name in directories + files: if (Path(directory) / name).is_symlink(): raise ValueError("Backup directory contains a symbolic link") if Path(directory) == source: directories[:] = [name for name in directories if name != ".encrypted-export"] for name in files: file_path = Path(directory) / name relative = file_path.relative_to(source).as_posix() if relative in {"SHA256SUMS", "SHA256SUMS.sig"}: continue if not file_path.is_file(): raise ValueError("Backup contains a non-regular file") actual.add(relative) if actual != set(expected): raise ValueError("Backup files do not match the signed manifest (missing or extra payload)") for name, expected_digest in expected.items(): digest = hashlib.sha256() with (source / name).open("rb") as payload: for block in iter(lambda: payload.read(1024 * 1024), b""): digest.update(block) if digest.hexdigest() != expected_digest: raise ValueError(f"Backup checksum mismatch: {name!r}") def verify_archive(archive_path: Path, target_root: Path, prefixes: list[str]) -> None: excludes = [prefix.removeprefix("--exclude=") for prefix in prefixes if prefix.startswith("--exclude=")] prefixes = [prefix for prefix in prefixes if not prefix.startswith("--exclude=")] roots = [(prefix.rstrip("/"), prefix.endswith("/")) for prefix in prefixes] target_root = target_root.resolve() def allowed(name: str, directory: bool = False) -> bool: return any( name == root or (is_dir and name.startswith(root + "/")) or (directory and root.startswith(name + "/")) for root, is_dir in roots ) with tarfile.open(archive_path, "r:*", tarinfo=BackupTarInfo) as archive: for member in archive: name = relative_path(member.name) ancestors = [name, *(str(parent) for parent in PurePosixPath(name).parents if str(parent) != ".")] if any(fnmatch.fnmatchcase(path, pattern) or ("/" not in pattern and fnmatch.fnmatchcase(PurePosixPath(path).name, pattern)) for pattern in excludes for path in ancestors): continue is_directory = member.isdir() or member.type == b"D" # GNU incremental dump directory if not allowed(name, is_directory): raise ValueError(f"Archive member outside restore scope: {name!r}") if not (member.isfile() or is_directory or member.issym() or member.islnk()): raise ValueError(f"Unsupported archive member: {name!r}") if member.mode & 0o6000: raise ValueError(f"Privileged mode bits in archive: {name!r}") # Existing symlinks at the restore destination are not trusted either. resolved_parent = (target_root / name).parent.resolve() try: parent_name = resolved_parent.relative_to(target_root).as_posix() except ValueError as exc: raise ValueError("Restore destination contains a symlink escaping the target") from exc if parent_name != "." and not allowed(parent_name, True): raise ValueError("Restore destination resolves outside allowed paths") if member.issym() or member.islnk(): link = PurePosixPath(member.linkname) if member.islnk(): linked_name = relative_path(member.linkname) else: origin = PurePosixPath("/") if link.is_absolute() else PurePosixPath("/") / PurePosixPath(name).parent linked_name = os.path.normpath(str(origin / link)).lstrip("/") if not allowed(linked_name): raise ValueError(f"Archive link escapes restore scope: {name!r}") if member.type == b"D": # GNU incremental directory records can trigger deletions. Refuse # traversal/rename records that cannot be confined to this directory. payload = archive.extractfile(member) if payload: for record in payload.read().split(b"\0"): if record and (record[:1] not in (b"Y", b"N", b"D") or record[1:] in (b".", b"..") or b"/" in record[1:]): raise ValueError("Unsafe incremental directory record") def retain_chains(directory: Path, days: int) -> None: if not 1 <= days <= 36500: raise ValueError("Retention must be between 1 and 36500 days") cutoff = time.time() - days * 86400 groups = {} for path in directory.iterdir(): if path.is_symlink() or not path.is_dir() or not re.fullmatch(r"\d{8}T\d{6}Z-(?:full|incremental)", path.name): continue if not (path / "SHA256SUMS").is_file() or not (path / "SHA256SUMS.sig").is_file(): continue metadata = dict(line.split("=", 1) for line in (path / "metadata.env").read_text().splitlines() if "=" in line) schedule = metadata.get("LIMRISTEM_MAIL_BACKUP_SCHEDULE_ID", "manual") groups.setdefault(schedule, []).append((path, metadata)) for records in groups.values(): records.sort(key=lambda record: record[0].name) # A physical incremental can depend on an older filesystem-full backup. # Keep that chain until an operator has consolidated it with mariabackup. if any(meta.get("LIMRISTEM_MAIL_BACKUP_DB_MODE") == "physical-incremental" for _, meta in records): continue anchors = [path for path, _ in records if path.name.endswith("-full") and path.stat().st_mtime < cutoff] if not anchors: continue anchor = anchors[-1] for path, _ in records: if path.name < anchor.name and path.stat().st_mtime < cutoff: shutil.rmtree(path) def main() -> None: command, *arguments = sys.argv[1:] if command == "local-path": local_path(arguments[0]) elif command == "manifest": verify_manifest(Path(arguments[0])) elif command == "archive": verify_archive(Path(arguments[0]), Path(arguments[1]), arguments[2:]) elif command == "retention": retain_chains(Path(arguments[0]), int(arguments[1])) else: raise ValueError("Unknown backup safety command") if __name__ == "__main__": try: main() except (ValueError, OSError, tarfile.TarError) as error: print(f"Backup safety check failed: {error}", file=sys.stderr) raise SystemExit(1)