libervia-backend/libervia/backend/tools/backup/gpg.py

#!/usr/bin/env python3

# Libervia: an XMPP client
# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.


import os
from pathlib import Path
import shutil

from libervia.backend.tools.common.gpg_provider import get_provider
from libervia.backend.tools.common.gpg_provider.exceptions import (
    GPGProviderError,
    InvalidPacket,
    UnknownKey,
)

from .core import BackupError, GPG_DIR, GPG_FILE_TPL, GlobalHandler, Report, Staging

# errors the GPG provider can raise (they have no common base class)
GPG_ERRORS = (GPGProviderError, InvalidPacket, UnknownKey)


def export_gpg(home: Path, staging: Staging) -> None:
    """Copy a profile GPG home to the staging

    @param home: GPG home to copy (the name is the profile name)
    @param staging: staging where the GPG files are copied
    """
    if not home.is_dir():
        return
    for path in sorted(home.rglob("*")):
        if path.is_file():
            staging.add_file(
                GPG_FILE_TPL.format(
                    profile=home.name, path=path.relative_to(home).as_posix()
                ),
                path,
                0o600,
            )


def gpg_list_keys(home: Path) -> set[str]:
    """List the keys of a GPG home

    @param home: GPG home path
    @return: key fingerprints
    @raise BackupError: if the keys can't be listed
    """
    if not home.is_dir():
        return set()
    try:
        provider = get_provider(str(home))
        return {key.fingerprint for key in provider.list_public_keys("")}
    except GPG_ERRORS as e:
        raise BackupError(f"can't list GPG keys: {e}") from e


def _chmod_gpg_home(home: Path) -> None:
    """Set the restrictive permissions required by GPG on a home"""
    for dirpath, dirnames, filenames in os.walk(home):
        for dirname in dirnames:
            os.chmod(Path(dirpath) / dirname, 0o700)
        for filename in filenames:
            os.chmod(Path(dirpath) / filename, 0o600)
    os.chmod(home, 0o700)


def check_gpg_restore(
    target_home: Path, backup_home: Path
) -> tuple[list[str], list[str]]:
    """Check a GPG home restore is safe (union merge, nothing is deleted)

    The only risk is restoring an old backup onto a diverged keyring.

    @param target_home: existing GPG home (restore target)
    @param backup_home: GPG home from the backup
    @return: (added, diverged) fingerprints
    """
    backup_keys = gpg_list_keys(backup_home)
    target_keys = gpg_list_keys(target_home)
    return sorted(backup_keys - target_keys), sorted(target_keys - backup_keys)


def import_gpg(target_home: Path, backup_home: Path) -> int:
    """Merge the keys of a backup GPG home into a target home

    Union merge: nothing is ever deleted or overwritten; if the target doesn't exist, the
    backup home is simply copied.

    @param target_home: GPG home to restore into
    @param backup_home: GPG home from the backup
    @return: number of secret key files added to the target
    @raise BackupError: if the import fails
    """
    if not target_home.exists():
        target_home.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(backup_home, target_home)
        _chmod_gpg_home(target_home)
        return 0
    # union copy of the secret keys (never overwrites an existing one)
    src_keys = backup_home / "private-keys-v1.d"
    dst_keys = target_home / "private-keys-v1.d"
    dst_keys.mkdir(parents=True, exist_ok=True)
    added = 0
    if src_keys.is_dir():
        for key in sorted(src_keys.glob("*.key")):
            dst = dst_keys / key.name
            if not dst.exists():
                shutil.copy2(key, dst)
                added += 1
    # the keybox is a binary database which can't be merged: the backup keys are
    # exported and re-imported (the secret keys are already in the copied .key files)
    try:
        src_provider = get_provider(str(backup_home))
        dst_provider = get_provider(str(target_home))
        packets = [
            src_provider.export_public_key(key)
            for key in src_provider.list_public_keys("")
        ]
        for packet in packets:
            try:
                dst_provider.import_public_key(packet)
            except InvalidPacket:
                # key already in the target keybox: nothing to import
                continue
    except GPG_ERRORS as e:
        raise BackupError(f"GPG import failed: {e}") from e
    _chmod_gpg_home(target_home)
    return added


class GpgHandler(GlobalHandler):
    name = "gpg"

    def _staged_profiles(self, staging: Staging) -> list[str]:
        return sorted(
            {name.split("/")[1] for name in staging.names() if name.startswith("gpg/")}
        )

    def export(
        self,
        local_dir: Path,
        staging: Staging,
        report: Report,
        profiles: list[str] | None = None,
    ) -> None:
        """Copy the GPG homes to the staging

        @param profiles: GPG homes to copy
            None to copy all of them
        @param report: export report
        """
        gpg_dir = local_dir / GPG_DIR
        if not gpg_dir.is_dir():
            return
        wanted = set(profiles) if profiles is not None else None
        for home in sorted(gpg_dir.iterdir()):
            if home.is_dir() and (wanted is None or home.name in wanted):
                export_gpg(home, staging)

    def import_(
        self,
        local_dir: Path,
        staging: Staging,
        report: Report,
        profiles: list[str] | None = None,
        force: bool = False,
    ) -> None:
        """Merge the backup GPG homes into the local ones (union merge)

        A keyring which diverged from the backup is skipped unless ``force`` is set (the
        backup may be older than the current keys).

        @param profiles: GPG homes to import (None means all of them)
        """
        wanted = set(profiles) if profiles is not None else None
        for profile_name in self._staged_profiles(staging):
            if wanted is not None and profile_name not in wanted:
                continue
            backup_home = staging.path(f"gpg/{profile_name}")
            if not backup_home.is_dir() or not any(backup_home.iterdir()):
                continue
            target_home = local_dir / GPG_DIR / profile_name
            added, diverged = check_gpg_restore(target_home, backup_home)
            if diverged and not force:
                report.skipped.append(
                    f"gpg: {profile_name}: the current keyring has keys not in "
                    f"the backup ({', '.join(diverged)})"
                )
                continue
            count = import_gpg(target_home, backup_home)
            report.done.append(
                f"gpg: keyring merged into {target_home} ({count} new secret key file(s))"
            )
            for fp in added:
                report.done.append(f"gpg: added key {fp}")