diff --git a/libervia/backend/memory/sqla_config.py b/libervia/backend/memory/sqla_config.py index ff2ea1ef..0ed79f2e 100644 --- a/libervia/backend/memory/sqla_config.py +++ b/libervia/backend/memory/sqla_config.py @@ -16,23 +16,41 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from pathlib import Path from urllib.parse import quote + from libervia.backend.core.constants import Const as C from libervia.backend.tools import config +def get_db_url(database_path: Path, readonly: bool = False) -> str: + """Get the SQLAlchemy URL of a sqlite database + + @param database_path: path to the sqlite DB + @param readonly: open in read only mode (sqlite URI mode) + @return: SQLAlchemy URL + """ + if readonly: + # file: URI, `uri=true` is needed for the dialect to use it as such + return ( + f"sqlite+aiosqlite:///file:{quote(str(database_path))}" + "?mode=ro&timeout=30&uri=true" + ) + return f"sqlite+aiosqlite:///{quote(str(database_path))}?timeout=30" + + def get_db_config() -> dict: """Get configuration for database @return: dict with following keys: - type: only "sqlite" for now - path: path to the sqlite DB + - url: SQLAlchemy URL """ main_conf = config.parse_main_conf() database_path = main_conf.common.local_dir / C.SAVEFILE_DATABASE - url = f"sqlite+aiosqlite:///{quote(str(database_path))}?timeout=30" return { "type": "sqlite", "path": database_path, - "url": url, + "url": get_db_url(database_path), } diff --git a/libervia/backend/tools/backup/__init__.py b/libervia/backend/tools/backup/__init__.py new file mode 100644 index 00000000..5f76c2b9 --- /dev/null +++ b/libervia/backend/tools/backup/__init__.py @@ -0,0 +1,532 @@ +#!/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 . + +from collections.abc import Callable +from datetime import datetime, timezone +import hashlib +import importlib.metadata +import lzma +import os +from pathlib import Path +import stat +import tarfile +import time +from typing import Any + +from alembic.runtime import migration as al_migration +from lxml import etree +from pydantic import ValidationError +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from libervia.backend.core.log import getLogger +from libervia.backend.memory import sqla_config +from libervia.backend.memory.sqla_mapping import ParamInd, Profile +from libervia.backend.tools.common.utils import is_safe_relpath + +from .core import ( + BackupContext, + BackupError, + CHUNK_SIZE, + FORMAT, + FORMAT_VERSION, + GlobalHandler, + Handler, + MANIFEST_FILE, + PIE_FILE_TPL, + ProfileData, + ProfileHandler, + Report, + Staging, + get_db_driver, + sqlite_db_config, +) +from .handlers import DATA_TYPES, DB_BACKED_TYPES, REGISTRY +from .models import Manifest, ProfileMeta +from .xep_0227 import ( + PieError, + build_pie, + parse_pie, + sections_by_ns, +) + +# pie section namespaces handled by this version of Libervia, derived from the handlers +# registry +KNOWN_SECTION_NS = frozenset(ns for handler in REGISTRY for ns in handler.known_ns()) + +# pie entry name check, derived from the pie file template +_PIE_NAME_PREFIX, _PIE_NAME_SUFFIX = PIE_FILE_TPL.split("{}") + +log = getLogger(__name__) + +__all__ = [ + "Backup", + "BackupContext", + "BackupError", + "DATA_TYPES", + "DB_BACKED_TYPES", + "GlobalHandler", + "Handler", + "Manifest", + "PIE_FILE_TPL", + "ProfileData", + "ProfileHandler", + "ProfileMeta", + "Report", + "Staging", + "get_db_driver", + "sqlite_db_config", +] + + +class _ProgressFile: + """File-like wrapper that reports write progress (for archive compression)""" + + def __init__( + self, stream, progress_cb: Callable[[int, int], None] | None, total: int + ): + self._stream = stream + self._progress_cb = progress_cb + self._total = total + self._pos = 0 + self._reported = 0 + + def write(self, data: bytes) -> int: + count = self._stream.write(data) + self._pos += count + if self._progress_cb is not None and ( + self._pos - self._reported >= 256 * 1024 or self._pos >= self._total + ): + self._reported = self._pos + self._progress_cb(min(self._pos, self._total), self._total) + return count + + def flush(self) -> None: + self._stream.flush() + + def tell(self) -> int: + return self._pos + + def close(self) -> None: + self._stream.close() + + +class Backup: + """Backup/restore engine""" + + @staticmethod + async def open_context( + db_config: dict, + param_file: Path, + local_dir: Path, + readonly: bool = False, + ) -> BackupContext: + """Open a backup context on a database + + @param db_config: database config (see "sqla_config.get_db_config") + @param param_file: path of the parameters file + @param local_dir: local data directory (GPG homes, ...) + @param readonly: open the database in read only mode (export) + @return: backup context + """ + path = Path(db_config["path"]) + log.debug(f"opening database {path} for backup (readonly={readonly})") + url = sqla_config.get_db_url(path, readonly) + engine = create_async_engine(url) + return BackupContext( + engine=engine, + session_factory=async_sessionmaker(engine, expire_on_commit=False), + db_url=url, + param_file=param_file, + local_dir=local_dir, + db=get_db_driver(db_config), + ) + + @staticmethod + async def get_profiles(ctx: BackupContext) -> list[ProfileData]: + """List profiles with their bare JID + + @param ctx: backup context + @return: profiles data + """ + async with ctx.session() as session: + profs = ( + (await session.execute(select(Profile).order_by(Profile.name))) + .scalars() + .all() + ) + jid_rows = ( + await session.execute( + select(ParamInd.profile_id, ParamInd.value).where( + ParamInd.category == "Connection", + ParamInd.name == "JabberID", + ) + ) + ).all() + jids = { + profile_id: (value or "").strip().split("/")[0] + for profile_id, value in jid_rows + } + return [ + ProfileData(name=p.name, id=p.id, jid=jids.get(p.id) or None) for p in profs + ] + + @staticmethod + def get_app_version() -> str: + """Libervia version, as reported by the package metadata""" + try: + return importlib.metadata.version("libervia-backend") + except importlib.metadata.PackageNotFoundError: + return "unknown" + + @staticmethod + async def get_alembic_rev(ctx: BackupContext) -> str | None: + """Get the alembic (schema) revision of the database""" + + def _get_heads(conn) -> tuple[str, ...]: + context = al_migration.MigrationContext.configure(conn) + return context.get_current_heads() + + async with ctx.engine.connect() as conn: + try: + heads = await conn.run_sync(_get_heads) + except Exception: + log.error( + f"no alembic revision: {ctx.db_url} has no alembic_version table" + ) + return None + return heads[0] if heads else None + + @staticmethod + def get_handlers(types: list[str]) -> list[Handler]: + """Get the handlers of data types, in export order + + @param types: data types (see DATA_TYPES) + @return: the matching handlers + """ + wanted = set(types) + return [handler for handler in REGISTRY if handler.name in wanted] + + @staticmethod + async def export_profile( + ctx: BackupContext, + staging: Staging, + profile: ProfileData, + handlers: list[ProfileHandler], + ) -> Report: + """Export a profile data to the staging + + @param ctx: backup context + @param staging: staging holding the archive entries + @param profile: profile to export + @param handlers: profile handlers to run (in export order) + @return: export report + """ + report = Report() + sections: list[etree._Element] = [] + for handler in handlers: + sections.extend(await handler.export(ctx, staging, profile, report)) + if sections: + staging.add_bytes( + PIE_FILE_TPL.format(profile.name), + build_pie(profile.host_jid, profile.user_name, sections), + 0o600, + ) + return report + + @staticmethod + async def import_profile( + ctx: BackupContext, + staging: Staging, + profile_name: str, + handlers: list[ProfileHandler], + report: Report, + force: bool = False, + ) -> None: + """Import a profile data into the database + + Unknown pie sections are reported and ignored. The roster section is not imported + yet. + + @param ctx: backup context + @param staging: staging holding the archive entries (pie document and parameters + file) + @param profile_name: profile to import for + @param handlers: profile handlers to run (in export order) + @param report: import report + @param force: the restore is dangerous, the user confirmed it + """ + pie_file = PIE_FILE_TPL.format(profile_name) + if pie_file not in staging: + report.skipped.append(f"{profile_name}: no data in archive") + return + try: + host_jid, user_name, sections = parse_pie(staging.path(pie_file).read_bytes()) + except PieError as e: + raise BackupError(f"{profile_name}: {e}") from e + jid = f"{user_name}@{host_jid}" if host_jid else user_name + by_ns = sections_by_ns(sections) + for ns in sorted(set(by_ns) - KNOWN_SECTION_NS): + report.skipped.append(f"unknown section {ns!r} ignored") + async with ctx.session() as session: + for handler in handlers: + await handler.import_( + ctx, session, staging, profile_name, jid, by_ns, report, force=force + ) + await session.commit() + + @staticmethod + def build_manifest( + app_version: str, + alembic_rev: str | None, + profiles: dict[str, ProfileMeta], + staging: Staging, + ) -> Manifest: + """Build the archive manifest and stage it + + @param app_version: Libervia version + @param alembic_rev: database schema revision + @param profiles: profile name: exported data (jid, types) + @param staging: staging holding the archive entries (the manifest is added to it) + @return: the manifest + """ + manifest = Manifest( + format=FORMAT, + format_version=FORMAT_VERSION, + app="libervia", + app_version=app_version, + alembic=alembic_rev, + date=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + profiles=profiles, + files={name: {"sha256": staging.get_hash(name)} for name in staging.names()}, + ) + staging.add_bytes( + MANIFEST_FILE, manifest.model_dump_json(indent=2).encode(), 0o644 + ) + return manifest + + @staticmethod + def _load_manifest(data: bytes) -> Manifest: + """Parse and validate the archive manifest + + @raise BackupError: if the manifest is invalid or not from this format + """ + try: + manifest = Manifest.model_validate_json(data) + except ValidationError as e: + raise BackupError(f"invalid manifest: {e}") from e + if manifest.format != FORMAT: + raise BackupError(f"unsupported format: {manifest.format!r}") + if manifest.format_version != FORMAT_VERSION: + raise BackupError(f"unsupported format version: {manifest.format_version}") + return manifest + + @staticmethod + def _manifest_of(tar: tarfile.TarFile) -> Manifest: + """Read and validate the manifest of an opened archive + + @param tar: opened backup archive + @return: archive manifest + @raise BackupError: if the manifest is missing or invalid + """ + manifest_file = tar.extractfile(MANIFEST_FILE) + if manifest_file is None: + raise BackupError(f"missing manifest: {MANIFEST_FILE!r}") + return Backup._load_manifest(manifest_file.read()) + + @staticmethod + def _data_members(tar: tarfile.TarFile) -> list[tarfile.TarInfo]: + """Archive entries holding data (the manifest is not included)""" + return [m for m in tar.getmembers() if m.isfile() and m.name != MANIFEST_FILE] + + @staticmethod + def make_archive( + staging: Staging, + manifest: Manifest, + dest_dir: Path, + progress_cb: Callable[[int, int], None] | None = None, + ) -> Path: + """Pack the staged backup entries in a tar.xz archive + + @param staging: staging holding the archive entries (including manifest) + @param manifest: archive manifest (used for the archive name) + @param dest_dir: directory where the archive is written + @param progress_cb: optional progress callback ``callable(written, total)`` + @return: path of the created archive + """ + dest_dir.mkdir(parents=True, exist_ok=True) + date = datetime.fromisoformat(manifest.date) + base_name = f"libervia-backup-{date.strftime('%Y-%m-%d_%H-%M-%S')}" + path = dest_dir / f"{base_name}.tar.xz" + counter = 1 + while path.exists(): + path = dest_dir / f"{base_name}-{counter}.tar.xz" + counter += 1 + mtime = int(time.time()) + # approximate size of the tar stream (for the progress bar): each entry is a + # 512-byte header plus its data padded to a 512-byte block, and the 20 KB + # margin covers the end-of-archive marker and the extra headers tar writes + # for long entry names + total = sum( + 512 + ((staging.size(name) + 511) // 512) * 512 for name in staging.names() + ) + total += 20 * 1024 + with lzma.open(path, "wb") as comp: + wrapper = _ProgressFile(comp, progress_cb, total) + # the tarfile stubs type fileobj with read/seek, which are not used + # in write mode + fileobj: Any = wrapper + with tarfile.open(fileobj=fileobj, mode="w") as tar: + for name in staging.names(): + src = staging.path(name) + src_stat = src.stat() + info = tarfile.TarInfo(name) + info.size = src_stat.st_size + info.mtime = mtime + # the entry mode is set by the one who staged it (private + # entries are owner-only) + info.mode = stat.S_IMODE(src_stat.st_mode) + with src.open("rb") as f: + tar.addfile(info, f) + wrapper.close() + if progress_cb is not None: + progress_cb(total, total) + os.chmod(path, 0o600) + return path + + @staticmethod + def read_archive( + path: Path, + staging: Staging, + progress_cb: Callable[[int, int], None] | None = None, + ) -> Manifest: + """Read and check a backup archive, extracting its entries to the staging + + @param path: path to the archive + @param staging: staging where the entries are extracted + @param progress_cb: optional progress callback ``callable(done, total)`` + @return: archive manifest + @raise BackupError: if the archive is invalid or tampered + """ + try: + with tarfile.open(path, "r:xz") as tar: + manifest = Backup._manifest_of(tar) + members = Backup._data_members(tar) + total = sum(m.size for m in members) + done = 0 + for member in members: + if not is_safe_relpath(member.name): + raise BackupError(f"invalid entry name {member.name!r}") + dst = staging.path(member.name) + dst.parent.mkdir(parents=True, exist_ok=True) + digest = hashlib.sha256() + src_f = tar.extractfile(member) + if src_f is None: + raise BackupError(f"can't extract {member.name!r}") + with dst.open("wb") as out: + while chunk := src_f.read(CHUNK_SIZE): + out.write(chunk) + digest.update(chunk) + done += len(chunk) + if progress_cb: + progress_cb(done, total) + # we don't want special bits in the permissions (to avoid setuid, + # for instance) + os.chmod(dst, member.mode & 0o777) + expected = manifest.files.get(member.name, {}).get("sha256") + if expected is None: + # every entry of a valid archive is listed in the + # manifest: an unlisted one means the archive was + # modified after the export + raise BackupError( + f"unlisted entry {member.name!r} (archive tampered?)" + ) + if digest.hexdigest() != expected: + raise BackupError( + f"{member.name}: checksum mismatch (archive tampered?)" + ) + except (tarfile.TarError, OSError) as e: + raise BackupError(f"can't read archive: {e}") from e + for name in manifest.files: + if name not in staging: + raise BackupError(f"{name}: missing from archive") + return manifest + + @staticmethod + def verify_archive( + path: Path, + progress_cb: Callable[[str, int, int], None] | None = None, + entry_cb: Callable[[str, dict], None] | None = None, + ) -> tuple[Manifest, dict[str, dict]]: + """Stream-check a backup archive (checksums and pie validity) + + The entries are not extracted, only hashed (and parsed, for pie documents). + + @param path: path to the archive + @param progress_cb: optional (entry name, bytes done, total bytes) callback, + called on each read chunk + @param entry_cb: optional (entry name, entry info) callback, called on each + entry completion + @return: (manifest, per-entry info: size, sha256, checksum status, pie error if + any) + @raise BackupError: if the archive is unreadable or the manifest is missing + """ + try: + with tarfile.open(path, "r:xz") as tar: + manifest = Backup._manifest_of(tar) + members = Backup._data_members(tar) + total = sum(m.size for m in members) + results: dict[str, dict] = {} + done = 0 + for member in members: + is_pie = member.name.startswith( + _PIE_NAME_PREFIX + ) and member.name.endswith(_PIE_NAME_SUFFIX) + digest = hashlib.sha256() + data = b"" + src_f = tar.extractfile(member) + if src_f is None: + raise BackupError(f"can't extract {member.name!r}") + while chunk := src_f.read(CHUNK_SIZE): + digest.update(chunk) + if is_pie: + data += chunk + done += len(chunk) + if progress_cb: + progress_cb(member.name, done, total) + pie_error = None + if is_pie: + try: + parse_pie(data) + except PieError as e: + pie_error = f"invalid pie: {e}" + expected = manifest.files.get(member.name, {}).get("sha256") + info = { + "size": member.size, + "sha256": digest.hexdigest(), + "checksum": ( + "ok" if expected == digest.hexdigest() else "mismatch" + ), + "pie": pie_error, + } + results[member.name] = info + if entry_cb: + entry_cb(member.name, info) + return manifest, results + except (tarfile.TarError, OSError) as e: + raise BackupError(f"can't read archive: {e}") from e diff --git a/libervia/backend/tools/backup/core.py b/libervia/backend/tools/backup/core.py new file mode 100644 index 00000000..6f3b5c79 --- /dev/null +++ b/libervia/backend/tools/backup/core.py @@ -0,0 +1,412 @@ +#!/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 . + + +from abc import ABC, abstractmethod +from contextlib import asynccontextmanager +from dataclasses import dataclass, field +from datetime import datetime +import hashlib +import os +from pathlib import Path +import shutil +import sqlite3 +import tempfile +from typing import ClassVar +from urllib.parse import quote + +from lxml import etree +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker +from twisted.words.protocols.jabber import jid + +from libervia.backend.core.constants import Const as C + +from .xep_0227 import NS_PIE, PieError + +# Pie Libervia custom extensions: +NS_PARAMS = NS_PIE + "#libervia-params" +NS_PRIVATE = NS_PIE + "#libervia-private" +NS_PRIVATE_BIN = NS_PIE + "#libervia-private-bin" + +NS_ROSTER_VER = "urn:xmpp:features:rosterver" +ROSTER_VER_KEY = "@version@" + +FORMAT = "libervia-backup" +FORMAT_VERSION = 1 +MANIFEST_FILE = "manifest.json" +PIE_FILE_TPL = "server-data-{}.xml" + +GPG_FILE_TPL = "gpg/{profile}/{path}" +FILES_FILE_TPL = "files/{path}" +PARAM_FILE = "param.xml" +DB_FILE = "db/libervia.db" + +GPG_DIR = "gnupg-home" + +# chunk size for streaming reads (hashing, extraction) +CHUNK_SIZE = 1024 * 1024 + + +class BackupError(PieError): + """Error while exporting or importing a backup""" + + +@dataclass +class ProfileData: + """A profile and the data needed to build its pie document""" + + name: str + id: int + # None if the JabberID parameter is not set yet + jid: str | None + + def __post_init__(self): + # the JID may not be set yet, or may not be well-formed: keep the parsed JID when + # usable, None otherwise + self._parsed_jid = None + if self.jid is not None: + try: + self._parsed_jid = jid.JID(self.jid) + except jid.InvalidFormat: + pass + + @property + def host_jid(self) -> str | None: + if self._parsed_jid is None: + return None + return self._parsed_jid.host or None + + @property + def user_name(self) -> str: + if self._parsed_jid is not None and self._parsed_jid.user: + return self._parsed_jid.user + return self.name + + +@dataclass +class Report: + """Result of an export or import pass""" + + done: list[str] = field(default_factory=list) + skipped: list[str] = field(default_factory=list) + + +class Staging: + """Temporary directory where backup entries are written (as files)""" + + def __init__(self, prefix: str = "libervia-backup-") -> None: + """Create a new staging in a fresh temporary directory + + @param prefix: prefix of the temporary directory + """ + self._dir = Path(tempfile.mkdtemp(prefix=prefix)) + self._hashes: dict[str, str] = {} + + def path(self, name: str) -> Path: + """Get the path of an entry in the staging + + @param name: name of the entry + @return: path of the entry + """ + return self._dir / name + + def add_bytes(self, name: str, data: bytes, mode: int = 0o600) -> None: + """Write an entry to the staging from bytes + + @param name: name of the entry + @param data: data to write + @param mode: file mode of the entry + """ + path = self.path(name) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + os.chmod(path, mode) + self._hashes[name] = hashlib.sha256(data).hexdigest() + + def add_file(self, name: str, src: Path, mode: int = 0o600) -> None: + """Write an entry to the staging from a file + + @param name: name of the entry + @param src: path of the file to copy + @param mode: file mode of the entry + """ + path = self.path(name) + path.parent.mkdir(parents=True, exist_ok=True) + digest = hashlib.sha256() + with src.open("rb") as s, path.open("wb") as d: + for chunk in iter(lambda: s.read(CHUNK_SIZE), b""): + d.write(chunk) + digest.update(chunk) + os.chmod(path, mode) + self._hashes[name] = digest.hexdigest() + + def names(self) -> list[str]: + """Get the names of all the staged entries + + @return: sorted list of entry names + """ + return sorted( + path.relative_to(self._dir).as_posix() + for path in self._dir.rglob("*") + if path.is_file() + ) + + def __contains__(self, name: str) -> bool: + """Check if an entry exists in the staging + + @param name: name of the entry + @return: True if the entry exists + """ + return self.path(name).is_file() + + def size(self, name: str) -> int: + """Get the size of an entry + + @param name: name of the entry + @return: size of the entry, in bytes + """ + return self.path(name).stat().st_size + + def get_hash(self, name: str) -> str: + """Get the SHA-256 of a staged entry + + @param name: name of the entry + @return: SHA-256 of the entry, in hexadecimal + """ + if name in self._hashes: + return self._hashes[name] + # entries written directly to the staging (e.g. the database snapshot) are hashed + # on the fly + digest = hashlib.sha256() + with self.path(name).open("rb") as f: + for chunk in iter(lambda: f.read(CHUNK_SIZE), b""): + digest.update(chunk) + self._hashes[name] = digest.hexdigest() + return self._hashes[name] + + def close(self) -> None: + """Remove the staging directory and all its entries""" + shutil.rmtree(self._dir, ignore_errors=True) + + +class DbDriver(ABC): + """Engine-specific database snapshot/restore""" + + @abstractmethod + def snapshot(self, staging: Staging) -> None: + """Write a snapshot of the database to the staging + + @param staging: staging where the snapshot is written + """ + raise NotImplementedError + + @abstractmethod + def restore(self, staging: Staging, force: bool) -> Report: + """Restore the database snapshot from the staging + + The application must not be running: the database is replaced in place. + + @param staging: staging holding the snapshot + @param force: the restore is destructive, it requires force + @return: import report + @raise BackupError: if force is not set + """ + raise NotImplementedError + + +class SqliteDbDriver(DbDriver): + """Database driver for SQLite (backup via "sqlite3.backup")""" + + def __init__(self, db_path: Path): + self.db_path = db_path + + def _backup_to(self, src_path: Path, dst_path: Path) -> None: + """Copy a database to another file with "sqlite3.backup" + + The destination is created with mode 0600. + + @param src_path: source database (opened in read only mode) + @param dst_path: destination file + """ + dst_path.parent.mkdir(parents=True, exist_ok=True) + src = sqlite3.connect(f"file:{quote(str(src_path))}?mode=ro", uri=True) + try: + dst = sqlite3.connect(dst_path) + try: + src.backup(dst) + finally: + dst.close() + finally: + src.close() + os.chmod(dst_path, 0o600) + + def snapshot(self, staging: Staging) -> None: + self._backup_to(self.db_path, staging.path(DB_FILE)) + + def restore(self, staging: Staging, force: bool) -> Report: + report = Report() + if DB_FILE not in staging: + return report + if not force: + raise BackupError( + "restoring the database backup replaces the current database: " + "confirmation is required" + ) + db_path = self.db_path + db_path.parent.mkdir(parents=True, exist_ok=True) + if db_path.exists(): + timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + backup_path = db_path.with_name(f"{db_path.name}.pre-restore-{timestamp}") + # we copy with "sqlite3.backup" to include the changes not written to the main + # file yet (WAL) + self._backup_to(db_path, backup_path) + report.done.append(f"db: current database kept at {backup_path}") + # stale journal files belong to the previous (replaced) database + for suffix in ("-wal", "-shm", "-journal"): + old = db_path.with_name(db_path.name + suffix) + if old.exists(): + old.unlink() + shutil.copyfile(staging.path(DB_FILE), db_path) + os.chmod(db_path, 0o600) + report.done.append("db: database") + return report + + +def get_db_driver(db_config: dict) -> DbDriver: + """Get the database driver matching a database config + + @param db_config: database config + @return: database driver + @raise BackupError: if the engine has no database driver + """ + if db_config.get("type") == "sqlite": + return SqliteDbDriver(Path(db_config["path"])) + raise BackupError( + "database backup is not supported by the " + f"{db_config.get('type')!r} database backend" + ) + + +@dataclass +class BackupContext: + """Data access context used by the export/import functions""" + + engine: AsyncEngine + session_factory: async_sessionmaker + db_url: str + param_file: Path + local_dir: Path + db: DbDriver + + @asynccontextmanager + async def session(self): + async with self.session_factory() as session: + yield session + + async def close(self) -> None: + await self.engine.dispose() + + +def sqlite_db_config(local_dir: Path) -> dict: + """Database config for the local sqlite database of a local directory""" + return {"type": "sqlite", "path": local_dir / C.SAVEFILE_DATABASE} + + +class Handler(ABC): + """Export/import handler of a backup data type""" + + name: ClassVar[str] + """Data type name""" + + uses_db: ClassVar[bool] = False + """True if the handler reads or writes the database""" + + def known_ns(self) -> tuple[str, ...]: + """Namespaces of the pie sections handled by the handler + + Used to detect sections from unknown (e.g. newer) versions on import. + """ + return () + + +class ProfileHandler(Handler): + """Handler for per profile data, exported in the profile pie document""" + + @abstractmethod + async def export( + self, + ctx: BackupContext, + staging: Staging, + profile: ProfileData, + report: Report, + ) -> list[etree._Element]: + """Export the data of a profile, as pie sections + + @return: the pie sections to add to the profile pie document + """ + raise NotImplementedError + + @abstractmethod + async def import_( + self, + ctx: BackupContext, + session: AsyncSession, + staging: Staging, + profile_name: str, + profile_jid: str, + sections: dict[str, list[etree._Element]], + report: Report, + force: bool = False, + ) -> None: + """Import the data of a profile from its pie sections + + @param profile_jid: JID of the profile, as stored in the pie document + @param sections: the profile pie sections, grouped by namespace + @param force: the restore is dangerous, the user confirmed it + """ + raise NotImplementedError + + +class GlobalHandler(Handler): + """Handler for data not tied to a profile (GPG homes, files, database) + + Some of this data is still organized per profile (GPG homes): the optional + ``profiles`` argument limits the handler to these profiles (None means all of them). + """ + + @abstractmethod + def export( + self, + local_dir: Path, + staging: Staging, + report: Report, + profiles: list[str] | None = None, + ) -> None: + raise NotImplementedError + + @abstractmethod + def import_( + self, + local_dir: Path, + staging: Staging, + report: Report, + profiles: list[str] | None = None, + force: bool = False, + ) -> None: + """Import the data, ``force`` is required for a dangerous restore""" + raise NotImplementedError diff --git a/libervia/backend/tools/backup/gpg.py b/libervia/backend/tools/backup/gpg.py new file mode 100644 index 00000000..1f689b3a --- /dev/null +++ b/libervia/backend/tools/backup/gpg.py @@ -0,0 +1,210 @@ +#!/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 . + + +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}") diff --git a/libervia/backend/tools/backup/handlers.py b/libervia/backend/tools/backup/handlers.py new file mode 100644 index 00000000..c1a8cb76 --- /dev/null +++ b/libervia/backend/tools/backup/handlers.py @@ -0,0 +1,166 @@ +#!/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 . + + +from pathlib import Path +import shutil + +from libervia.backend.core.constants import Const as C + +from .core import ( + DB_FILE, + DbDriver, + FILES_FILE_TPL, + GlobalHandler, + Handler, + ProfileHandler, + Report, + Staging, + get_db_driver, + sqlite_db_config, +) +from .gpg import GpgHandler +from .pie import ParamsHandler, PepHandler, PrivateHandler, RosterHandler + + +class FilesHandler(GlobalHandler): + name = "files" + + def export( + self, + local_dir: Path, + staging: Staging, + report: Report, + profiles: list[str] | None = None, + ) -> None: + """Copy the file sharing blobs to the staging + + @param local_dir: local data directory + @param staging: staging where the blobs are copied + @param report: export report + """ + files_dir = local_dir / C.FILES_DIR + if not files_dir.is_dir(): + return + count = 0 + for path in sorted(files_dir.rglob("*")): + if path.is_file(): + staging.add_file( + FILES_FILE_TPL.format(path=path.relative_to(files_dir).as_posix()), + path, + 0o600, + ) + count += 1 + if count: + report.done.append(f"files: {count} file(s)") + + def import_( + self, + local_dir: Path, + staging: Staging, + report: Report, + profiles: list[str] | None = None, + force: bool = False, + ) -> None: + """Copy the staged file sharing blobs to the local data directory + + The blobs are content-addressed (the name is the content hash), so overwriting an + existing file is safe and repairs a corrupted one. + + @param local_dir: local data directory + @param staging: staging holding the blobs + @param report: import report + """ + files_dir = local_dir / C.FILES_DIR + # entry name prefix, derived from the staging name template + prefix = FILES_FILE_TPL.split("{}")[0] + count = 0 + for name in staging.names(): + if not name.startswith(prefix): + continue + dst = files_dir / name[len(prefix) :] + dst.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(staging.path(name), dst) + count += 1 + if count: + report.done.append(f"files: {count} file(s)") + + +class DbHandler(GlobalHandler): + name = "db" + + def _driver(self, local_dir: Path) -> DbDriver: + return get_db_driver(sqlite_db_config(local_dir)) + + def export( + self, + local_dir: Path, + staging: Staging, + report: Report, + profiles: list[str] | None = None, + ) -> None: + """Write a snapshot of the database to the staging + + @param report: export report + """ + self._driver(local_dir).snapshot(staging) + + def import_( + self, + local_dir: Path, + staging: Staging, + report: Report, + profiles: list[str] | None = None, + force: bool = False, + ) -> None: + """Restore the database snapshot from the staging (destructive) + + @param report: import report + @param force: the restore is destructive, it requires force + """ + if DB_FILE not in staging: + return + res = self._driver(local_dir).restore(staging, force) + report.done.extend(res.done) + report.skipped.extend(res.skipped) + + +# one handler per data type, in export order (the order is semantic: the database restore +# comes last, it replaces the database) +REGISTRY: list[Handler] = [ + ParamsHandler(), + PrivateHandler(), + RosterHandler(), + PepHandler(), + GpgHandler(), + FilesHandler(), + DbHandler(), +] + +# "db" is a non-portable copy of the database (it will be migrated automatically on start +# if the Libervia version is newer): it covers all the data, while the portable types +# don't cover everything yet. +DATA_TYPES: tuple[str, ...] = tuple(handler.name for handler in REGISTRY) + +# data types stored in the database: a database restore brings them back, so +# there is no need to import them on top of it +DB_BACKED_TYPES: tuple[str, ...] = tuple( + handler.name + for handler in REGISTRY + if isinstance(handler, ProfileHandler) and handler.uses_db +) diff --git a/libervia/backend/tools/backup/models.py b/libervia/backend/tools/backup/models.py new file mode 100644 index 00000000..3e43c33e --- /dev/null +++ b/libervia/backend/tools/backup/models.py @@ -0,0 +1,48 @@ +#!/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 . + +from pydantic import BaseModel, Field + + +class ProfileMeta(BaseModel): + """Exported data of a profile.""" + + jid: str | None = Field(default=None, description="JabberID of the profile") + types: list[str] = Field( + default_factory=list, description="data types exported for this profile" + ) + + +class Manifest(BaseModel): + """Manifest of a backup archive (manifest.json).""" + + format: str = Field(description="backup format name") + format_version: int = Field(description="backup format version") + app: str = Field(description="application that made the backup") + app_version: str = Field(description="version of the application") + alembic: str | None = Field( + default=None, description="database schema revision (None if unknown)" + ) + date: str = Field(description="backup date (UTC, ISO 8601)") + profiles: dict[str, ProfileMeta] = Field( + default_factory=dict, description="exported profiles, by name" + ) + files: dict[str, dict[str, str]] = Field( + default_factory=dict, + description="checksums of the archive entries, by entry name", + ) diff --git a/libervia/backend/tools/backup/pie.py b/libervia/backend/tools/backup/pie.py new file mode 100644 index 00000000..b6bee165 --- /dev/null +++ b/libervia/backend/tools/backup/pie.py @@ -0,0 +1,656 @@ +#!/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 . + + +from datetime import datetime, timezone +import json +import os +import shutil +import time + +from lxml import etree +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.sql import cast +from sqlalchemy.types import Text + +from libervia.backend.memory.sqla_mapping import ( + ParamGen, + ParamInd, + PrivateGen, + PrivateGenBin, + PrivateInd, + PrivateIndBin, + Profile, + PubsubItem, + PubsubNode, +) + +from .core import ( + BackupContext, + BackupError, + NS_PARAMS, + NS_PRIVATE, + NS_PRIVATE_BIN, + NS_ROSTER_VER, + PARAM_FILE, + ProfileData, + ProfileHandler, + ROSTER_VER_KEY, + Report, + Staging, +) +from .xep_0227 import ( + NS_PUBSUB, + NS_ROSTER, + append_xml_payload, + build_pep_items, + build_roster, + item_payload, +) + + +async def _get_profile_id(session, profile_name: str) -> int | None: + """Get the database id of a profile + + @param session: database session + @param profile_name: profile to look for + @return: profile id, or None if the profile is unknown + """ + return ( + await session.execute(select(Profile.id).where(Profile.name == profile_name)) + ).scalar_one_or_none() + + +class ParamsHandler(ProfileHandler): + name = "params" + uses_db = True + + def known_ns(self) -> tuple[str, ...]: + return (NS_PARAMS,) + + async def export( + self, + ctx: BackupContext, + staging: Staging, + profile: ProfileData, + report: Report, + ) -> list[etree._Element]: + section = await self._section(ctx, profile) + if ctx.param_file.is_file() and PARAM_FILE not in staging: + staging.add_file(PARAM_FILE, ctx.param_file, 0o644) + return [section] + + async def import_( + self, + ctx: BackupContext, + session: AsyncSession, + staging: Staging, + profile_name: str, + profile_jid: str, + sections: dict[str, list[etree._Element]], + report: Report, + force: bool = False, + ) -> None: + for section in sections.get(NS_PARAMS, []): + await self._import(session, profile_name, section, report) + if PARAM_FILE in staging: + shutil.copyfile(staging.path(PARAM_FILE), ctx.param_file) + os.chmod(ctx.param_file, 0o600) + report.done.append("params: parameters file") + + async def _section(self, ctx: BackupContext, profile: ProfileData) -> etree._Element: + """Build the params section from the database + + @param ctx: backup context + @param profile: profile to export + @return: the params section element + """ + async with ctx.session() as session: + gen = ( + ( + await session.execute( + select(ParamGen).order_by(ParamGen.category, ParamGen.name) + ) + ) + .scalars() + .all() + ) + ind = ( + ( + await session.execute( + select(ParamInd) + .where(ParamInd.profile_id == profile.id) + .order_by(ParamInd.category, ParamInd.name) + ) + ) + .scalars() + .all() + ) + section = etree.Element(f"{{{NS_PARAMS}}}params", nsmap={None: NS_PARAMS}) + for scope, rows in (("general", gen), ("individual", ind)): + for row in rows: + # values are stored as-is in the database, XML only carries strings + attr = { + "category": str(row.category), + "name": str(row.name), + "scope": scope, + } + if row.value is not None: + attr["value"] = str(row.value) + etree.SubElement(section, f"{{{NS_PARAMS}}}param", **attr) + return section + + async def _import( + self, session, profile_name: str, section: etree._Element, report: Report + ) -> None: + """Import a params section into the database + + @param session: database session + @param profile_name: profile to import for + @param section: params section element + @param report: import report (individual rows of unknown profiles are skipped) + """ + profile_id = await _get_profile_id(session, profile_name) + gen_rows = { + (row.category, row.name): row + for row in (await session.execute(select(ParamGen))).scalars() + } + ind_rows = {} + if profile_id is not None: + ind_rows = { + (row.category, row.name): row + for row in ( + await session.execute( + select(ParamInd).where(ParamInd.profile_id == profile_id) + ) + ).scalars() + } + count = 0 + for elt in section.findall(f"{{{NS_PARAMS}}}param"): + category, name = elt.get("category", ""), elt.get("name", "") + value = elt.get("value") + if elt.get("scope", "general") == "general": + row = gen_rows.get((category, name)) + if row is None: + session.add(ParamGen(category=category, name=name, value=value)) + else: + row.value = value + count += 1 + elif profile_id is None: + report.skipped.append(f"params: unknown profile {profile_name!r}") + else: + row = ind_rows.get((category, name)) + if row is None: + session.add( + ParamInd( + category=category, + name=name, + profile_id=profile_id, + value=value, + ) + ) + else: + row.value = value + count += 1 + if count: + report.done.append(f"params: {count} value(s)") + + +class PrivateHandler(ProfileHandler): + name = "private" + uses_db = True + + def known_ns(self) -> tuple[str, ...]: + return (NS_PRIVATE, NS_PRIVATE_BIN) + + async def export( + self, + ctx: BackupContext, + staging: Staging, + profile: ProfileData, + report: Report, + ) -> list[etree._Element]: + sections: list[etree._Element] = [] + for bin_ in (False, True): + if (section := await self._section(ctx, profile, bin_)) is not None: + sections.append(section) + return sections + + async def import_( + self, + ctx: BackupContext, + session: AsyncSession, + staging: Staging, + profile_name: str, + profile_jid: str, + sections: dict[str, list[etree._Element]], + report: Report, + force: bool = False, + ) -> None: + for ns, bin_ in ((NS_PRIVATE, False), (NS_PRIVATE_BIN, True)): + for section in sections.get(ns, []): + await self._import( + session, profile_name, section, bin_=bin_, report=report + ) + + async def _section( + self, ctx: BackupContext, profile: ProfileData, bin_: bool = False + ) -> etree._Element | None: + """Build a private data section (key-value storage, or its binary variant) + + Keys are not necessarily valid XML element names, so each value is stored + in a element with namespace, key and scope attributes (Libervia + extension). + + @param ctx: backup context + @param profile: profile to export + @param bin_: True for the binary (JSON) values + @return: the private section element, or None if empty + """ + ns = NS_PRIVATE_BIN if bin_ else NS_PRIVATE + gen_model = PrivateGenBin if bin_ else PrivateGen + ind_model = PrivateIndBin if bin_ else PrivateInd + async with ctx.session() as session: + gen = ( + ( + await session.execute( + select(gen_model).order_by(gen_model.namespace, gen_model.key) + ) + ) + .scalars() + .all() + ) + ind = ( + ( + await session.execute( + select(ind_model) + .where(ind_model.profile_id == profile.id) + .order_by(ind_model.namespace, ind_model.key) + ) + ) + .scalars() + .all() + ) + rows = [("general", row) for row in gen] + [("individual", row) for row in ind] + if not rows: + return None + section = etree.Element(f"{{{ns}}}private", nsmap={None: ns}) + for scope, row in rows: + attr = { + "namespace": str(row.namespace or ""), + "key": str(row.key), + "scope": scope, + } + if row.value is None: + # XML can't distinguish an empty value from a missing one + attr["nil"] = "true" + elt = etree.SubElement(section, f"{{{ns}}}row", **attr) + if row.value is not None: + if bin_: + # JSON values: serialize back with the same serializer + # used by the Json column type (byte-identical) + elt.text = json.dumps(row.value, ensure_ascii=False) + else: + elt.text = str(row.value) + return section + + async def _import( + self, + session, + profile_name: str, + section: etree._Element, + bin_: bool, + report: Report, + ) -> None: + """Import a private data section into the database + + @param session: database session + @param profile_name: profile to import for + @param section: private section element + @param bin_: True for the binary (JSON) values + @param report: import report (individual rows of unknown profiles are skipped) + """ + suffix = "_bin" if bin_ else "" + gen_model = PrivateGenBin if bin_ else PrivateGen + ind_model = PrivateIndBin if bin_ else PrivateInd + profile_id = await _get_profile_id(session, profile_name) + gen_rows = { + (row.namespace, row.key): row + for row in (await session.execute(select(gen_model))).scalars() + } + ind_rows = {} + if profile_id is not None: + ind_rows = { + (row.namespace, row.key): row + for row in ( + await session.execute( + select(ind_model).where(ind_model.profile_id == profile_id) + ) + ).scalars() + } + count = 0 + for elt in section.findall(f"{{{NS_PRIVATE_BIN if bin_ else NS_PRIVATE}}}row"): + namespace, key = elt.get("namespace", ""), elt.get("key", "") + nil = elt.get("nil") == "true" + if bin_ and not nil: + try: + # the Json column type will serialize it back + value = json.loads(elt.text or "") + except json.JSONDecodeError as e: + raise BackupError( + f"private{suffix}: invalid JSON for {namespace}/{key!r}: {e}" + ) from e + else: + value = None if nil else (elt.text or "") + if elt.get("scope", "general") == "general": + row = gen_rows.get((namespace, key)) + if row is None: + session.add(gen_model(namespace=namespace, key=key, value=value)) + else: + row.value = value + count += 1 + elif profile_id is None: + report.skipped.append( + f"private: unknown profile {profile_name!r}, row ignored" + ) + else: + row = ind_rows.get((namespace, key)) + if row is None: + session.add( + ind_model( + namespace=namespace, + key=key, + value=value, + profile_id=profile_id, + ) + ) + else: + row.value = value + count += 1 + if count: + report.done.append(f"private{suffix}: {count} value(s)") + + +class RosterHandler(ProfileHandler): + name = "roster" + uses_db = True + + def known_ns(self) -> tuple[str, ...]: + return (NS_ROSTER,) + + async def export( + self, + ctx: BackupContext, + staging: Staging, + profile: ProfileData, + report: Report, + ) -> list[etree._Element]: + # the roster is exported from its persistent cache in the database, kept + # up to date on every connection and roster push (roster versioning) + items: list[etree._Element] = [] + version: str | None = None + async with ctx.session() as session: + rows = ( + ( + await session.execute( + select(PrivateInd) + .where( + PrivateInd.profile_id == profile.id, + PrivateInd.namespace == NS_ROSTER_VER, + ) + .order_by(PrivateInd.key) + ) + ) + .scalars() + .all() + ) + for row in rows: + if row.key == ROSTER_VER_KEY: + version = row.value + else: + items.append(etree.fromstring(row.value.encode())) + if not items: + return [] + return [build_roster(items, version)] + + async def import_( + self, + ctx: BackupContext, + session: AsyncSession, + staging: Staging, + profile_name: str, + profile_jid: str, + sections: dict[str, list[etree._Element]], + report: Report, + force: bool = False, + ) -> None: + # importing the roster needs a server connection (the roster has to be pushed to + # the server), it is not implemented yet + if sections.get(NS_ROSTER): + report.skipped.append( + f"{profile_name}: roster not imported (not implemented yet, " + "needs a server connection)" + ) + + +class PepHandler(ProfileHandler): + name = "pep" + uses_db = True + + def known_ns(self) -> tuple[str, ...]: + return (NS_PUBSUB,) + + async def export( + self, + ctx: BackupContext, + staging: Staging, + profile: ProfileData, + report: Report, + ) -> list[etree._Element]: + section = await self._section(ctx, profile) + if section is None: + report.skipped.append("pep: no PEP node in cache") + return [] + return [section] + + async def import_( + self, + ctx: BackupContext, + session: AsyncSession, + staging: Staging, + profile_name: str, + profile_jid: str, + sections: dict[str, list[etree._Element]], + report: Report, + force: bool = False, + ) -> None: + await self._import( + session, profile_name, profile_jid, sections.get(NS_PUBSUB, []), report + ) + + async def _section( + self, ctx: BackupContext, profile: ProfileData + ) -> etree._Element | None: + """Build the PEP items section from the pubsub cache + + Node configuration is not cached locally, so only items are exported. + + @param ctx: backup context + @param profile: profile to export + @return: the PEP section element, or None if the profile has no PEP nodes + """ + async with ctx.session() as session: + # the service and data columns carry app-specific types (JID, parsed + # XML), so they are cast to text to keep the backup byte-identical + node_rows = ( + await session.execute( + select( + PubsubNode.id, + PubsubNode.name, + cast(PubsubNode.service, Text), + ) + .where(PubsubNode.profile_id == profile.id, PubsubNode.pep.is_(True)) + .order_by(PubsubNode.name) + ) + ).all() + if not node_rows: + return None + item_rows = ( + await session.execute( + select( + PubsubItem.node_id, + PubsubItem.name, + cast(PubsubItem.data, Text), + ) + .where( + PubsubItem.node_id.in_([node_id for node_id, _, _ in node_rows]) + ) + .order_by(PubsubItem.name) + ) + ).all() + by_id = {node_id: (name, service) for node_id, name, service in node_rows} + nodes: dict[tuple[str, str | None], list[tuple[str, str, str | None]]] = {} + for node_id, item_name, data in item_rows: + payload, publisher = self._item_parts(data) + name, service = by_id[node_id] + nodes.setdefault((name, service), []).append( + (item_name or "", payload or "", publisher) + ) + return build_pep_items(nodes) + + async def _import( + self, + session, + profile_name: str, + jid: str, + sections: list[etree._Element], + report: Report, + ) -> None: + """Import the PEP items sections into the pubsub cache + + Rows are written with plain SQL (the columns carry app-specific types) + to keep the data byte-identical. + + @param session: database session + @param profile_name: profile to import for + @param jid: profile JID, used as the node service + @param sections: PEP items section elements + @param report: import report (unknown profiles are skipped) + """ + profile_id = await _get_profile_id(session, profile_name) + if profile_id is None: + report.skipped.append(f"pep: unknown profile {profile_name!r}") + return + for section in sections: + for items in section.findall(f"{{{NS_PUBSUB}}}items"): + node_name = items.get("node") + if not node_name: + continue + # the node owner is not necessarily the pie user (e.g. service profiles), + # so the service is stored in the items element (Libervia extension); + # pie documents from other servers don't have it, and there the node + # owner is the pie user + service = items.get("service") or jid + await session.execute( + text( + "INSERT INTO pubsub_nodes (profile_id, service, name, pep, " + "subscribed, sync_state_updated, max_items) " + "VALUES (:profile_id, :service, :name, 1, 0, :now, 0) " + "ON CONFLICT (profile_id, service, name, pep) " + "DO UPDATE SET pep = 1" + ), + { + "profile_id": profile_id, + "service": service, + "name": node_name, + "now": time.time(), + }, + ) + node_id = ( + await session.execute( + text( + "SELECT id FROM pubsub_nodes " + "WHERE profile_id=:profile_id AND service=:service " + "AND name=:name" + ), + { + "profile_id": profile_id, + "service": service, + "name": node_name, + }, + ) + ).scalar_one() + now = datetime.now(timezone.utc) + for item in items.findall(f"{{{NS_PUBSUB}}}item"): + data = self._item_data( + item.get("id", ""), item.get("publisher"), item_payload(item) + ) + await session.execute( + text( + "INSERT INTO pubsub_items (node_id, name, data, created, " + "updated) VALUES (:node_id, :name, :data, :now, :now) " + "ON CONFLICT (node_id, name) DO UPDATE SET " + "data = excluded.data, updated = excluded.updated" + ), + { + "node_id": node_id, + "name": item.get("id", ""), + "data": data, + "now": now, + }, + ) + report.done.append(f"pep: node {node_name!r}") + + def _item_parts(self, data: str | None) -> tuple[str | None, str | None]: + """Extract the payload and publisher of a stored pubsub item + + The data column stores the full element, but the pie document + carries the payload and the publisher instead. + + @param data: stored item data + @return: (payload, publisher), payload is an XML string or plain text + """ + if not data: + return data, None + try: + root = etree.fromstring(data.encode()) + except etree.XMLSyntaxError: + return data, None + if etree.QName(root).localname != "item": + return data, None + return item_payload(root), root.get("publisher") + + def _item_data(self, item_id: str, publisher: str | None, payload: str) -> str: + """Build the data to store in the pubsub cache for a pie item + + The payload is wrapped in a pubsub element, as in the cache. + + @param item_id: item id + @param publisher: item publisher (attribute of the element) + @param payload: item payload (XML string or plain text) + @return: data to store + """ + # the item element is in the pubsub namespace (XEP-0060), as in the cache + wrapper = etree.Element(f"{{{NS_PUBSUB}}}item", nsmap={None: NS_PUBSUB}) + if item_id: + wrapper.set("id", item_id) + if publisher: + wrapper.set("publisher", publisher) + append_xml_payload(wrapper, payload) + return etree.tostring(wrapper, encoding="unicode") diff --git a/libervia/backend/tools/backup/xep_0227.py b/libervia/backend/tools/backup/xep_0227.py new file mode 100644 index 00000000..41c42b9c --- /dev/null +++ b/libervia/backend/tools/backup/xep_0227.py @@ -0,0 +1,169 @@ +#!/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 . + +from lxml import etree + +NS_PIE = "urn:xmpp:pie:0" +NS_ROSTER = "jabber:iq:roster" +NS_PUBSUB = "http://jabber.org/protocol/pubsub" + + +class PieError(Exception): + """Error while building or parsing a pie (XEP-0227) document""" + + +## pie document ## + + +def build_pie( + host_jid: str | None, user_name: str, sections: list[etree._Element] +) -> bytes: + """Build a pie document for a user + + @param host_jid: virtual host JID; the XEP makes the attribute required, so an empty + value is written when it is unknown (profile JID not set yet) + @param user_name: node part of the user JID (````) + @param sections: section elements to include under the ```` element + @return: serialized pie document + """ + root = etree.Element(f"{{{NS_PIE}}}server-data", nsmap={None: NS_PIE}) + host = etree.SubElement(root, f"{{{NS_PIE}}}host", jid=host_jid or "") + user = etree.SubElement(host, f"{{{NS_PIE}}}user", name=user_name) + for section in sections: + user.append(section) + return etree.tostring(root, xml_declaration=True, encoding="utf-8", pretty_print=True) + + +def parse_pie(data: bytes) -> tuple[str, str, list[etree._Element]]: + """Parse a pie document + + Unknown sections are kept as-is: the XEP requires importers to ignore data they don't + understand. + + @param data: serialized pie document + @return: (host jid, user name, sections) + @raise PieError: if the document is not a valid pie document + """ + try: + root = etree.fromstring(data) + except etree.XMLSyntaxError as e: + raise PieError(f"invalid pie document: {e}") from e + if root.tag != f"{{{NS_PIE}}}server-data": + raise PieError(f"unexpected root element {root.tag!r}, expected pie") + host = root.find(f"{{{NS_PIE}}}host") + if host is None: + raise PieError("missing element in pie document") + user = host.find(f"{{{NS_PIE}}}user") + if user is None: + raise PieError("missing element in pie document") + return host.get("jid", ""), user.get("name", ""), list(user) + + +def sections_by_ns(sections: list[etree._Element]) -> dict[str, list[etree._Element]]: + """Group pie sections by their namespace""" + by_ns: dict[str, list[etree._Element]] = {} + for section in sections: + ns = etree.QName(section).namespace or "" + by_ns.setdefault(ns, []).append(section) + return by_ns + + +## sections: roster ## + + +def build_roster( + items: list[etree._Element], version: str | None = None +) -> etree._Element: + """Build the roster section + + @param items: roster item elements, as in the roster query + @param version: roster version, if known + @return: the roster section element + """ + section = etree.Element(f"{{{NS_ROSTER}}}query", nsmap={None: NS_ROSTER}) + if version: + section.set("ver", version) + for item in items: + section.append(item) + return section + + +## sections: pep items ## + + +def append_xml_payload(parent: etree._Element, data: str | None) -> None: + """Append item data (stored as XML text) to an element, with text fallback + + @param parent: element to append the payload to + @param data: payload data (XML string or plain text) + """ + if not data: + return + try: + parent.append(etree.fromstring(data.encode())) + except etree.XMLSyntaxError: + parent.text = data + + +def item_payload(item: etree._Element) -> str: + """Serialize the payload of a pubsub item (without the wrapper) + + @param item: item element + @return: payload as string + """ + children = list(item) + if not children: + return item.text or "" + # keep the payload in a temporary container (several elements supported) + tmp = etree.Element("payload") + if (item.text or "").strip(): + tmp.text = item.text + for child in children: + if child.tail and not child.tail.strip(): + child.tail = None + tmp.append(child) + # exclusive: only the namespaces actually used by the payload are kept + data = etree.tostring(tmp, encoding="unicode", exclusive=True) + return data[len("") : -len("")] + + +def build_pep_items( + nodes: dict[tuple[str, str | None], list[tuple[str, str, str | None]]], +) -> etree._Element | None: + """Build the PEP items section + + @param nodes: (node name, service) mapped to the items as (item id, payload, + publisher) lists, payloads are XML strings or plain text. The service (node owner + JID) is stored in a ``service`` attribute + @return: the PEP items section element, or None if there is no node + """ + if not nodes: + return None + section = etree.Element(f"{{{NS_PUBSUB}}}pubsub", nsmap={None: NS_PUBSUB}) + for (node_name, service), items in sorted( + nodes.items(), key=lambda kv: (kv[0][0], kv[0][1] or "") + ): + items_elt = etree.SubElement(section, f"{{{NS_PUBSUB}}}items", node=node_name) + if service: + items_elt.set("service", service) + for item_id, payload, publisher in items: + item = etree.SubElement(items_elt, f"{{{NS_PUBSUB}}}item", id=item_id) + if publisher: + item.set("publisher", publisher) + append_xml_payload(item, payload) + return section diff --git a/libervia/backend/tools/common/utils.py b/libervia/backend/tools/common/utils.py index aff7b35b..0a0a3ea4 100644 --- a/libervia/backend/tools/common/utils.py +++ b/libervia/backend/tools/common/utils.py @@ -19,6 +19,8 @@ """Misc utils for both backend and frontends""" import collections.abc +import os +from pathlib import Path size_units = { "b": 1, @@ -178,3 +180,34 @@ def get_human_size(size: int | str, suffix: str = "o", sep: str = " ") -> str: """Return data size in a human readable format.""" size, symbol = get_size_multiplier(size, suffix) return f"{size:.2f}{sep}{symbol}" + + +def is_safe_relpath(name: str) -> bool: + """Check if a path is safe to join to a directory (as in a tar archive) + + @param name: path to check + @return: True if the path is relative and has no parent reference + """ + return not name.startswith("/") and ".." not in name.split("/") + + +def is_process_running(pid_file: Path) -> bool: + """Check if the process in a PID file is running + + @param pid_file: path to the PID file + @return: True if a live process matches the PID + """ + if not pid_file.is_file(): + return False + try: + pid = int(pid_file.read_text().strip()) + except (OSError, ValueError): + return False + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + # the process exists but belongs to another user + pass + return True