diff --git a/libervia/cli/base.py b/libervia/cli/base.py index c6a27895..06d7b04a 100644 --- a/libervia/cli/base.py +++ b/libervia/cli/base.py @@ -128,6 +128,9 @@ class LiberviaCli: """ + # commands which can be run without a backend connection + BRIDGELESS_COMMANDS = ("backup",) + def __init__(self): """ @@ -830,33 +833,44 @@ class LiberviaCli: if self.args.max is None: self.args.max = C.NO_LIMIT + def _is_bridgeless_command(self, args): + """Check if the command on command line can be run without a backend""" + argv = args if args is not None else sys.argv[1:] + return bool(argv) and argv[0] in self.BRIDGELESS_COMMANDS + async def main(self, args, namespace): try: self.bridge, self.bridge_settings = await dynamic_import.load_and_init_bridge( self.settings, verbose=False ) except exceptions.BridgeExceptionNoService as e: - self.disp( - _( - "Can't connect to Libervia backend, are you sure that it's " - "launched ?" - ), - error=True, - ) - self.quit(C.EXIT_BACKEND_NOT_FOUND, raise_exc=False) + if not self._is_bridgeless_command(args): + self.disp( + _( + "Can't connect to Libervia backend, are you sure that it's " + "launched ?" + ), + error=True, + ) + self.quit(C.EXIT_BACKEND_NOT_FOUND, raise_exc=False) + return + self.bridge = None + self.version = "unknown (no backend connection)" except exceptions.BridgeInitError as e: self.disp(_("Can't init bridge"), error=True) self.quit(C.EXIT_BRIDGE_ERROR, raise_exc=False) + return except Exception as e: raise exceptions.InternalError( "Something went wrong while initialising bridge." ) from e - self.loop.install_hooks(self.bridge_settings.type) + if self.bridge is not None: + self.loop.install_hooks(self.bridge_settings.type) - # we wait on init_pre_script instead of ready_get, so the CLI frontend can be used - # in init script. - await self.bridge.init_pre_script() - self.version = await self.bridge.version_get() + # we wait on init_pre_script instead of ready_get, so the CLI frontend can be + # used in init script. + await self.bridge.init_pre_script() + self.version = await self.bridge.version_get() self._bridge_connected() self.import_plugins() try: @@ -905,7 +919,12 @@ class LiberviaCli: loop = asyncio.get_running_loop() stdin_fut = loop.create_future() loop.add_reader(sys.stdin, self._read_stdin, stdin_fut) - return await stdin_fut + try: + return await stdin_fut + finally: + # remove the reader once the input is done, so the callback does not fire + # again on a resolved future (e.g. stdin reaching EOF on exit) + loop.remove_reader(sys.stdin) async def confirm(self, message): """Request user to confirm action, return answer as boolean""" diff --git a/libervia/cli/cmd_backup.py b/libervia/cli/cmd_backup.py new file mode 100644 index 00000000..9af861e5 --- /dev/null +++ b/libervia/cli/cmd_backup.py @@ -0,0 +1,740 @@ +#!/usr/bin/env python3 + +# Libervia CLI +# 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 . + +from collections.abc import Callable, Iterable, Iterator +from contextlib import contextmanager +import os +from pathlib import Path +import time + +from rich.progress import Progress + +from libervia.backend.core.constants import Const as CBackend +from libervia.backend.core.i18n import _ +from libervia.backend.tools import backup +from libervia.backend.tools.common import utils +from libervia.backend.tools.common.gpg_provider import get_provider +from libervia.backend.tools.common.gpg_provider.exceptions import ( + DecryptionFailed, + GPGProviderError, +) + +from . import base +from .constants import Const as C + +__commands__ = ["Backup"] + +PARAM_FILE_NAME = "param" + +# errors the GPG provider can raise (they have no common base class) +GPG_ERRORS = (GPGProviderError, DecryptionFailed) + +# progress bar labels of the global (not per profile) export handlers +GLOBAL_EXPORT_TASKS = { + "db": "snapshotting database", + "files": "copying files", + "gpg": "copying GPG homes", +} + +# progress bar labels of the global (not per profile) import handlers +GLOBAL_IMPORT_TASKS = { + "db": "restoring database", + "files": "copying files", + "gpg": "merging GPG homes", +} + +# fixed width of the entry name in the verify progress bar, so the bar does +# not shift horizontally as names of different lengths are displayed +VERIFY_NAME_WIDTH = 30 + +# max number of profiles listed by verify in non-verbose mode, beyond that a +# simple count is shown +MAX_PROFILES_SHOWN = 3 + + +def _get_types(parser, args) -> list[str]: + """Compute the list of types to handle. + + Types are generated from --include/--exclude: + - no option = all types + - --include is a whitelist (if given, only these types are handled) + - --no-db is a named exclusion shortcut (export only: the database + is not imported by default on import). + """ + included = set(args.include) + excluded = set(args.exclude) + if getattr(args, "no_db", False): + excluded.add("db") + unknown = (included | excluded) - set(backup.DATA_TYPES) + if unknown: + parser.error(f"unknown type: {', '.join(sorted(unknown))}") + both = included & excluded + if both: + parser.error("type both included and excluded: " + ", ".join(sorted(both))) + types = [t for t in backup.DATA_TYPES if t not in excluded] + if included: + types = [t for t in types if t in included] + return types + + +def _print_types(command, types: list[str]) -> None: + """Print the selected types, so the user can check before it starts""" + command.print(_("types included: {types}").format(types=", ".join(types))) + excluded = [t for t in backup.DATA_TYPES if t not in types] + if excluded: + command.print(_("types excluded: {types}").format(types=", ".join(excluded))) + + +def _print_counted(command, label: str, messages: list[str], error: bool = False) -> None: + """Print messages, counting repetitions (e.g. ``2x imported: ...``)""" + counts: dict[str, int] = {} + for msg in messages: + counts[msg] = counts.get(msg, 0) + 1 + for msg, count in counts.items(): + prefix = f"{count}x " if count > 1 else "" + (command.disp if error else command.print)(f"{label}{prefix}{msg}") + + +def _split_handlers( + types: list[str], +) -> tuple[list[backup.ProfileHandler], list[backup.GlobalHandler]]: + """Split the handlers to run, by kind (per profile / global)""" + handlers = backup.Backup.get_handlers(types) + return ( + [h for h in handlers if isinstance(h, backup.ProfileHandler)], + [h for h in handlers if isinstance(h, backup.GlobalHandler)], + ) + + +def _selected_profiles( + parser, requested: list[str], available: Iterable[str], not_found: str +) -> set[str]: + """Check the requested profiles exist, and return the names to handle + + @param parser: argparse parser (used to report the error) + @param requested: profile names requested with -p + @param available: names of the existing profiles + @param not_found: error message, with a ``{names}`` placeholder + """ + unknown = set(requested) - set(available) + if unknown: + parser.error(not_found.format(names=", ".join(sorted(unknown)))) + return set(requested) + + +def _fixed_width(name: str, width: int) -> str: + """Truncate a name to a fixed width, left-justified + + Used for the progress bar descriptions, so that the line does not shift + while names of different lengths are displayed + """ + if len(name) > width: + name = name[: width - 1] + "…" + return f"{name:<{width}}" + + +@contextmanager +def _byte_task(progress: Progress, label: str) -> Iterator[Callable[[int, int], None]]: + """Progress task for byte-based operations (building/extracting an archive) + + Yields a ``callable(done, total)`` progress callback + """ + task = progress.add_task(_(label), total=0) + try: + yield lambda done, total: progress.update(task, total=total, completed=done) + finally: + progress.remove_task(task) + + +def _run_global_handlers( + progress: Progress, + handlers: list[backup.GlobalHandler], + labels: dict[str, str], + verb: str, + run: Callable[[backup.GlobalHandler], None], +) -> None: + """Run the global handlers, each with its own progress task + + @param labels: progress bar label by handler name (a default is derived + from ``verb`` for the other handlers) + @param run: called for each handler + """ + for handler in handlers: + label = labels.get(handler.name, f"{verb} {handler.name} data") + task = progress.add_task(_(label), total=None) + run(handler) + progress.remove_task(task) + + +def _local_paths(host) -> tuple[Path, Path, Path]: + """Paths of the local data + + @param host: LiberviaCli instance + @return: (local_dir, db_path, param_file) + """ + local_dir = Path(host.settings.common.local_dir) + return local_dir, local_dir / CBackend.SAVEFILE_DATABASE, local_dir / PARAM_FILE_NAME + + +def _gpg_symmetric(path: Path, passphrase: str) -> Path: + """Encrypt a file with GPG symmetric encryption, replacing it + + @param path: file to encrypt + @param passphrase: passphrase to use + @return: path of the encrypted file + @raise backup.BackupError: if the encryption fails + """ + provider = get_provider() + encrypted = path.with_name(path.name + ".gpg") + try: + # TODO: encryption is currently done in memory, need to move to streaming + # encryption (same for decryption in _gpg_decrypt) + data = provider.encrypt_symmetrically(path.read_bytes(), passphrase) + except GPG_ERRORS as e: + encrypted.unlink(missing_ok=True) + raise backup.BackupError(f"GPG encryption failed: {e}") from e + encrypted.write_bytes(data) + path.unlink() + os.chmod(encrypted, 0o600) + return encrypted + + +def _gpg_decrypt(in_path: Path, out_path: Path, passphrase: str) -> None: + """Decrypt a GPG symmetric encrypted file + + @param in_path: file to decrypt (name must end with .gpg) + @param out_path: path where the decrypted file is written + @param passphrase: passphrase to use + @raise backup.BackupError: if the decryption fails + """ + provider = get_provider() + try: + data = provider.decrypt_symmetrically(in_path.read_bytes(), passphrase) + except DecryptionFailed as e: + out_path.unlink(missing_ok=True) + raise backup.BackupError("GPG decryption failed, check the passphrase") from e + except GPGProviderError as e: + out_path.unlink(missing_ok=True) + raise backup.BackupError(f"GPG decryption failed: {e}") from e + out_path.write_bytes(data) + os.chmod(out_path, 0o600) + + +async def _resolve_archive(host, args, data: backup.Staging) -> Path: + """Check a backup archive exists and decrypt it to the staging if needed + + @param host: LiberviaCli instance + @param args: parsed arguments (``path`` and ``passphrase`` attributes) + @param data: staging used to hold the decrypted archive (owner-only dir, wiped with + it: no plaintext is kept next to the .gpg file) + @return: path of the (decrypted) archive + """ + path = Path(args.path) + if not path.is_file(): + host.disp(_("no such file: {path}").format(path=path), error=True) + host.quit(C.EXIT_ERROR) + if path.name.endswith(".gpg"): + passphrase = args.passphrase + if passphrase is None: + passphrase = await host.a_pwd_input(_("please enter the GPG passphrase: ")) + out = data.path("archive.tar.xz") + with Progress(console=host.console) as progress: + progress.add_task(_("decrypting archive"), total=None) + _gpg_decrypt(path, out, passphrase) + path = out + return path + + +class Export(base.CommandBase): + def __init__(self, host): + super(Export, self).__init__( + host, + "export", + use_profile=False, + help=_("export Libervia data to a backup archive"), + ) + + def add_parser_options(self): + self.parser.add_argument( + "-o", + "--output-dir", + default="backup", + help=_("directory where the archive is written (default: backup)"), + ) + self.parser.add_argument( + "-p", + "--profile", + action="append", + default=[], + help=_("profile to export (default: all profiles)"), + ) + self.parser.add_argument( + "-i", + "--include", + action="append", + default=[], + help=_( + "type to include (repeatable; if given, only these types are " + "exported, default: all)" + ), + ) + self.parser.add_argument( + "-e", + "--exclude", + action="append", + default=[], + help=_("type to exclude (repeatable)"), + ) + self.parser.add_argument( + "--no-db", + action="store_true", + help=_( + "don't export the non-portable full database backup " + "(it is exported by default, as it is the only part covering " + "all the data; without it, data such as message history is " + "not backed up)" + ), + ) + self.parser.add_argument( + "--encrypt", + nargs="?", + const=True, + default=None, + metavar="PASSPHRASE", + help=_( + "encrypt the archive with GPG symmetric encryption " + "(passphrase asked if not given)" + ), + ) + + async def start(self): + types = _get_types(self.parser, self.args) + local_dir, db_path, param_file = _local_paths(self.host) + if not db_path.is_file(): + self.disp(_("no database found at {path}").format(path=db_path), error=True) + self.host.quit(C.EXIT_ERROR) + _print_types(self, types) + passphrase = self.args.encrypt + if passphrase is True: + passphrase = await self.host.a_pwd_input(_("please enter a GPG passphrase: ")) + ctx = None + staging = backup.Staging() + start_time = time.monotonic() + try: + ctx = await backup.Backup.open_context( + backup.sqlite_db_config(local_dir), param_file, local_dir, readonly=True + ) + profiles = await backup.Backup.get_profiles(ctx) + if self.args.profile: + selected = _selected_profiles( + self.parser, + self.args.profile, + (p.name for p in profiles), + "unknown profile: {names}", + ) + profiles = [p for p in profiles if p.name in selected] + # fixed width for the profile name in the progress bar, so that the line + # doesn't shift with names of different lengths + name_width = min(max((len(p.name) for p in profiles), default=0), 20) + prof_handlers, glob_handlers = _split_handlers(types) + with Progress(console=self.console) as progress: + task = progress.add_task(_("exporting"), total=len(profiles)) + global_report = backup.Report() + profiles_meta: dict[str, backup.ProfileMeta] = {} + reports: list[backup.Report] = [] + for profile in profiles: + reports.append( + await backup.Backup.export_profile( + ctx, staging, profile, prof_handlers + ) + ) + profiles_meta[profile.name] = backup.ProfileMeta( + jid=profile.jid, types=types + ) + progress.advance(task) + progress.update( + task, + description=_("exporting: {name}").format( + name=_fixed_width(profile.name, name_width) + ), + ) + progress.remove_task(task) + profile_names = [p.name for p in profiles] + _run_global_handlers( + progress, + glob_handlers, + GLOBAL_EXPORT_TASKS, + "exporting", + lambda h: h.export( + local_dir, staging, global_report, profiles=profile_names + ), + ) + reports.append(global_report) + manifest = backup.Backup.build_manifest( + backup.Backup.get_app_version(), + await backup.Backup.get_alembic_rev(ctx), + profiles_meta, + staging, + ) + with _byte_task(progress, "building archive") as update: + path = backup.Backup.make_archive( + staging, + manifest, + Path(self.args.output_dir), + progress_cb=update, + ) + if passphrase: + task = progress.add_task(_("encrypting"), total=None) + path = _gpg_symmetric(path, passphrase) + progress.remove_task(task) + self.print( + _("backup written to {path} in {elapsed}s").format( + path=path, elapsed=f"{time.monotonic() - start_time:.1f}" + ) + ) + _print_counted( + self, + " - ", + [msg for report in reports for msg in report.done + report.skipped], + ) + finally: + staging.close() + if ctx is not None: + await ctx.close() + self.host.quit(0) + + +class Import(base.CommandBase): + def __init__(self, host): + super(Import, self).__init__( + host, + "import", + use_profile=False, + help=_("import Libervia data from a backup archive"), + ) + + def add_parser_options(self): + self.parser.add_argument( + "path", help=_("path to the backup archive (optionally GPG encrypted)") + ) + self.parser.add_argument( + "-p", + "--profile", + action="append", + default=[], + help=_("profile to import (default: all profiles found in the archive)"), + ) + self.parser.add_argument( + "-i", + "--include", + action="append", + default=[], + help=_( + "type to include (repeatable; if given, only these types are " + "imported, default: all)" + ), + ) + self.parser.add_argument( + "-e", + "--exclude", + action="append", + default=[], + help=_("type to exclude (repeatable)"), + ) + self.parser.add_argument( + "--use-db", + action="store_true", + help=_( + "restore the full database backup (it replaces the current " + "database), importing only the data not stored in it " + "(gpg keys, files)" + ), + ) + self.parser.add_argument( + "--passphrase", + default=None, + help=_("GPG passphrase, if the archive is encrypted (asked if needed)"), + ) + self.parser.add_argument( + "--force", + action="store_true", + help=_( + "do not ask for confirmation of dangerous operations " + "(database restore, diverged GPG keyring)" + ), + ) + + async def start(self): + types = _get_types(self.parser, self.args) + if self.args.use_db: + if "db" in set(self.args.exclude): + self.parser.error("type both included and excluded: db") + # a database restore replaces the database: the data stored in it + # comes from the restore, so import only the data not stored in + # the database on top of it + wanted = {t for t in types if t not in backup.DB_BACKED_TYPES} + wanted.add("db") + types = [t for t in backup.DATA_TYPES if t in wanted] + elif "db" not in self.args.include: + # a database restore is destructive (the database is replaced): + # it only happens when explicitly requested (--use-db or -i db) + types = [t for t in types if t != "db"] + if "db" in types: + # a database restore replaces the database file: it must + # be done while no backend has it open + if self.host.bridge is not None or utils.is_process_running( + Path(self.host.settings.common.pid_dir) / f"{CBackend.APP_NAME_FILE}.pid" + ): + self.disp( + _( + "the backend is running: a database restore must " + "be done while it is stopped, please stop it and re-run" + ), + error=True, + ) + self.host.quit(C.EXIT_ERROR) + _print_types(self, types) + # a database restore is destructive: it is confirmed right away (before the + # long archive extraction, and outside of the progress bar: the prompt and + # the bar don't play well together on the console) + db_confirmed = self.args.force + if "db" in types and not db_confirmed: + await self.host.confirm_or_quit( + _( + "restoring the database backup replaces the current " + "database, are you sure?" + ) + ) + db_confirmed = True + data = backup.Staging() + report = backup.Report() + manifest: backup.Manifest | None = None + try: + path = await _resolve_archive(self.host, self.args, data) + with Progress(console=self.console) as progress: + with _byte_task(progress, "extracting archive") as update: + manifest = backup.Backup.read_archive(path, data, progress_cb=update) + # read_archive succeeded (the command has quit otherwise) + assert manifest is not None + self.print( + _("backup of {app_version} from {date}").format( + app_version=manifest.app_version, date=manifest.date + ) + ) + profiles = manifest.profiles + if self.args.profile: + selected = _selected_profiles( + self.parser, + self.args.profile, + profiles, + "profile not in archive: {names}", + ) + profiles = {n: d for n, d in profiles.items() if n in selected} + local_dir, db_path, param_file = _local_paths(self.host) + prof_handlers, glob_handlers = _split_handlers(types) + if any(h.uses_db for h in prof_handlers): + # only the handlers that read/write the database need a + # database context (the gpg and files imports don't) + if not db_path.is_file(): + self.disp( + _( + "no database found at {path}, please install " + "Libervia first" + ).format(path=db_path), + error=True, + ) + self.host.quit(C.EXIT_ERROR) + ctx = await backup.Backup.open_context( + backup.sqlite_db_config(local_dir), param_file, local_dir + ) + try: + name_width = min(max((len(n) for n in profiles), default=0), 20) + task = progress.add_task(_("importing"), total=len(profiles)) + for name in sorted(profiles): + await backup.Backup.import_profile( + ctx, + data, + name, + prof_handlers, + report, + force=self.args.force, + ) + progress.advance(task) + progress.update( + task, + description=_("importing: {name}").format( + name=_fixed_width(name, name_width) + ), + ) + progress.remove_task(task) + finally: + await ctx.close() + profile_names = sorted(profiles) + _run_global_handlers( + progress, + glob_handlers, + GLOBAL_IMPORT_TASKS, + "importing", + # the db restore was already authorized (--force or confirmed + # above), --force is passed to the other handlers only + lambda h: h.import_( + local_dir, + data, + report, + profiles=profile_names, + force=self.args.force or (db_confirmed and h.name == "db"), + ), + ) + except backup.BackupError as e: + self.disp(str(e), error=True) + self.host.quit(C.EXIT_ERROR) + finally: + data.close() + _print_counted(self, "imported: ", report.done) + _print_counted(self, "skipped: ", report.skipped, error=True) + self.host.quit(0) + + +class Verify(base.CommandBase): + def __init__(self, host): + super(Verify, self).__init__( + host, + "verify", + use_profile=False, + use_verbose=True, + help=_("check a backup archive integrity and display its content"), + ) + + def add_parser_options(self): + self.parser.add_argument( + "path", help=_("path to the backup archive (optionally GPG encrypted)") + ) + self.parser.add_argument( + "--passphrase", + default=None, + help=_("GPG passphrase, if the archive is encrypted (asked if needed)"), + ) + + async def start(self): + data = backup.Staging() + manifest: backup.Manifest | None = None + results: dict[str, dict] = {} + verbose = self.args.verbose > 0 + try: + path = await _resolve_archive(self.host, self.args, data) + try: + self.disp(_("starting verification...")) + with Progress(console=self.console) as progress: + task = progress.add_task(_("verifying"), total=None) + + def progress_cb(name: str, done: int, total: int) -> None: + progress.update( + task, + completed=done, + total=total, + description=_fixed_width(name, VERIFY_NAME_WIDTH), + ) + + def entry_cb(name: str, info: dict) -> None: + # a failed check stays on its own line, ok lines are + # verbose only + if info["checksum"] != "ok" or info["pie"]: + status = [] + if info["checksum"] != "ok": + status.append(_("checksum mismatch (archive tampered?)")) + if info["pie"]: + status.append(info["pie"]) + progress.console.print( + f"[red]{name}: {', '.join(status)}[/red]" + ) + elif verbose: + progress.console.print( + _(" {name} ({size} bytes): ok").format( + name=name, size=info["size"] + ) + ) + + manifest, results = backup.Backup.verify_archive( + path, progress_cb=progress_cb, entry_cb=entry_cb + ) + except backup.BackupError as e: + self.disp(str(e), error=True) + self.host.quit(C.EXIT_ERROR) + finally: + data.close() + # verify_archive succeeded (the command has quit otherwise) + assert manifest is not None + self.print( + _("format: {fmt} v{version}").format( + fmt=manifest.format, version=manifest.format_version + ) + ) + self.print( + _("app: {app} {version}").format( + app=manifest.app, version=manifest.app_version + ) + ) + self.print(_("alembic: {alembic}").format(alembic=manifest.alembic)) + self.print(_("date: {date}").format(date=manifest.date)) + profiles = sorted(manifest.profiles.items()) + if verbose or len(profiles) <= MAX_PROFILES_SHOWN: + for name, prof in profiles: + self.print( + _("profile {name}: {jid} ({types})").format( + name=name, jid=prof.jid or "-", types=", ".join(prof.types) + ) + ) + else: + self.print(_("({count} profiles)").format(count=len(profiles))) + if not results: + # only the manifest is in the archive (the selected types did not + # produce any data), there is nothing to check + self.console.print( + _("[yellow]archive contains no data entries (only the manifest)[/yellow]") + ) + self.host.quit(0) + succeeded = sum( + 1 for i in results.values() if i["checksum"] == "ok" and not i["pie"] + ) + if succeeded == len(results): + self.console.print( + _( + "[green]{done}/{total} checks successful, archive is sane[/green]" + ).format(done=succeeded, total=len(results)) + ) + self.host.quit(0) + self.console.print( + _( + "[red]{done}/{total} checks successful, archive integrity FAILED[/red]" + ).format(done=succeeded, total=len(results)) + ) + self.host.quit(C.EXIT_ERROR) + + +class Backup(base.CommandBase): + """Backup/export/import of Libervia data""" + + subcommands = (Export, Import, Verify) + + def __init__(self, host): + super(Backup, self).__init__( + host, + "backup", + use_profile=False, + help=_("backup/restore Libervia data (XEP-0227 based format)"), + )