FluxIP/deploy/backup.py

#!/usr/bin/env python3
"""Create a consistent SQLite backup, excluding the master key by default."""

from __future__ import annotations

import argparse
import base64
import hashlib
import os
import re
import shutil
import sqlite3
import tarfile
import tempfile
from contextlib import closing, suppress
from datetime import UTC, datetime
from pathlib import Path

BACKUP_ARTIFACT_PATTERN = re.compile(
    r"fluxip-(?P<kind>database|complete)-backup-"
    r"(?P<timestamp>\d{8}T\d{6}Z)\.tar\.gz(?:\.sha256)?"
)


def non_negative_int(value: str) -> int:
    parsed = int(value)
    if parsed < 0:
        raise argparse.ArgumentTypeError("必须是非负整数")
    return parsed


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="备份 FluxIP SQLite 数据库")
    parser.add_argument("--database", type=Path, default=Path("data/fluxip.db"))
    parser.add_argument("--output-dir", type=Path, default=Path("backups"))
    parser.add_argument(
        "--keep-last",
        type=non_negative_int,
        default=0,
        metavar="N",
        help="按备份类型只保留最近 N 套；0 表示不自动清理",
    )
    parser.add_argument(
        "--master-key",
        type=Path,
        help="要随数据库打包的主密钥；默认不打包",
    )
    parser.add_argument(
        "--allow-plaintext-key-archive",
        action="store_true",
        help="明确允许把主密钥写入普通 tar.gz；应立即转存到加密介质",
    )
    return parser.parse_args()


def validate_master_key(path: Path) -> None:
    try:
        encoded = path.read_bytes().strip()
        decoded = base64.urlsafe_b64decode(encoded)
    except (OSError, ValueError) as exc:
        raise RuntimeError(f"主密钥文件无法读取或格式错误：{path}") from exc
    if len(decoded) != 32:
        raise RuntimeError("主密钥不是 32 字节 URL-safe Base64，已停止备份")


def backup_database(source_path: Path, destination_path: Path) -> None:
    with (
        closing(sqlite3.connect(source_path, timeout=30)) as source,
        closing(sqlite3.connect(destination_path)) as destination,
    ):
        source.backup(destination)
        result = destination.execute("PRAGMA quick_check").fetchone()
    if not result or result[0] != "ok":
        raise RuntimeError(f"SQLite 备份完整性检查失败：{result}")


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def create_private_file(path: Path) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    os.close(descriptor)


def prune_backups(output_dir: Path, kind: str, keep_last: int) -> int:
    if keep_last == 0:
        return 0

    artifacts: dict[str, list[Path]] = {}
    for candidate in output_dir.iterdir():
        if candidate.is_symlink() or not candidate.is_file():
            continue
        match = BACKUP_ARTIFACT_PATTERN.fullmatch(candidate.name)
        if match is None or match.group("kind") != kind:
            continue
        artifacts.setdefault(match.group("timestamp"), []).append(candidate)

    expired_timestamps = sorted(artifacts, reverse=True)[keep_last:]
    for timestamp in expired_timestamps:
        for artifact in artifacts[timestamp]:
            artifact.unlink()
    return len(expired_timestamps)


def main() -> int:
    args = parse_args()
    database = args.database.resolve()
    master_key = args.master_key.resolve() if args.master_key else None
    output_dir = args.output_dir.resolve()

    if not database.is_file():
        raise RuntimeError(f"数据库不存在：{database}")
    if args.allow_plaintext_key_archive and master_key is None:
        raise RuntimeError("使用 --allow-plaintext-key-archive 时必须同时提供 --master-key")
    if master_key is not None and not args.allow_plaintext_key_archive:
        raise RuntimeError(
            "拒绝把主密钥写入未加密备份；如已准备加密存储，请显式增加 "
            "--allow-plaintext-key-archive"
        )
    if master_key is not None:
        if not master_key.is_file():
            raise RuntimeError(f"主密钥文件不存在：{master_key}")
        validate_master_key(master_key)

    output_dir.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    kind = "complete" if master_key is not None else "database"
    archive = output_dir / f"fluxip-{kind}-backup-{timestamp}.tar.gz"
    temporary_archive = archive.with_suffix(archive.suffix + ".tmp")

    try:
        create_private_file(temporary_archive)
        with tempfile.TemporaryDirectory(prefix=".fluxip-", dir=output_dir) as temporary:
            staging = Path(temporary)
            backup_database(database, staging / "fluxip.db")
            with tarfile.open(temporary_archive, "w:gz") as package:
                package.add(staging / "fluxip.db", arcname="fluxip.db")
                if master_key is not None:
                    shutil.copy2(master_key, staging / "master.key")
                    package.add(staging / "master.key", arcname="master.key")

        os.replace(temporary_archive, archive)
    finally:
        with suppress(FileNotFoundError):
            temporary_archive.unlink()
    with suppress(OSError):
        archive.chmod(0o600)

    checksum = archive.with_name(archive.name + ".sha256")
    checksum.write_text(f"{sha256(archive)}  {archive.name}\n", encoding="ascii")
    with suppress(OSError):
        checksum.chmod(0o600)

    removed_sets = prune_backups(output_dir, kind, args.keep_last)

    print(f"备份完成：{archive}")
    print(f"校验文件：{checksum}")
    if args.keep_last:
        print(f"保留策略：{kind} 类型最近 {args.keep_last} 套，已清理 {removed_sets} 套")
    if master_key is None:
        print("此备份不含主密钥；请确认主密钥已在独立的加密位置托管")
    else:
        print("警告：备份包含主密钥，请立即转存到加密介质并删除非必要副本")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())