erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tools (backup): add backup/restore:

Browse source 
Add an offline backup/restore engine in `tools.backup`:

- format based on XEP-0227 (pie), extended with Libervia sections
- handle several data types (params, private, roster, pep , gpg, files and db)
- per profile export/import and global handlers
- a manifest is used to have metadata and checksums
- full database snapshot/restore with the `db` type
- sqla: add `get_db_url`
- tools.common.utils: add `is_safe_relpath` and `is_process_running`

rel 480
This commit is contained in:
Goffi 2026-09-15 11:34:58 +02:00
commit 86a91b5850
9 changed files with 2246 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

22
libervia/backend/memory/sqla_config.py
View file

@ -16,23 +16,41 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
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),
}

532
libervia/backend/tools/backup/__init__.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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

412
libervia/backend/tools/backup/core.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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

210
libervia/backend/tools/backup/gpg.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
import os
from pathlib import Path
import shutil
from libervia.backend.tools.common.gpg_provider import get_provider
from libervia.backend.tools.common.gpg_provider.exceptions import (
GPGProviderError,
InvalidPacket,
UnknownKey,
)
from .core import BackupError, GPG_DIR, GPG_FILE_TPL, GlobalHandler, Report, Staging
# errors the GPG provider can raise (they have no common base class)
GPG_ERRORS = (GPGProviderError, InvalidPacket, UnknownKey)
def export_gpg(home: Path, staging: Staging) -> None:
"""Copy a profile GPG home to the staging
@param home: GPG home to copy (the name is the profile name)
@param staging: staging where the GPG files are copied
"""
if not home.is_dir():
return
for path in sorted(home.rglob("*")):
if path.is_file():
staging.add_file(
GPG_FILE_TPL.format(
profile=home.name, path=path.relative_to(home).as_posix()
),
path,
0o600,
)
def gpg_list_keys(home: Path) -> set[str]:
"""List the keys of a GPG home
@param home: GPG home path
@return: key fingerprints
@raise BackupError: if the keys can't be listed
"""
if not home.is_dir():
return set()
try:
provider = get_provider(str(home))
return {key.fingerprint for key in provider.list_public_keys("")}
except GPG_ERRORS as e:
raise BackupError(f"can't list GPG keys: {e}") from e
def _chmod_gpg_home(home: Path) -> None:
"""Set the restrictive permissions required by GPG on a home"""
for dirpath, dirnames, filenames in os.walk(home):
for dirname in dirnames:
os.chmod(Path(dirpath) / dirname, 0o700)
for filename in filenames:
os.chmod(Path(dirpath) / filename, 0o600)
os.chmod(home, 0o700)
def check_gpg_restore(
target_home: Path, backup_home: Path
) -> tuple[list[str], list[str]]:
"""Check a GPG home restore is safe (union merge, nothing is deleted)
The only risk is restoring an old backup onto a diverged keyring.
@param target_home: existing GPG home (restore target)
@param backup_home: GPG home from the backup
@return: (added, diverged) fingerprints
"""
backup_keys = gpg_list_keys(backup_home)
target_keys = gpg_list_keys(target_home)
return sorted(backup_keys - target_keys), sorted(target_keys - backup_keys)
def import_gpg(target_home: Path, backup_home: Path) -> int:
"""Merge the keys of a backup GPG home into a target home
Union merge: nothing is ever deleted or overwritten; if the target doesn't exist, the
backup home is simply copied.
@param target_home: GPG home to restore into
@param backup_home: GPG home from the backup
@return: number of secret key files added to the target
@raise BackupError: if the import fails
"""
if not target_home.exists():
target_home.parent.mkdir(parents=True, exist_ok=True)
shutil.copytree(backup_home, target_home)
_chmod_gpg_home(target_home)
return 0
# union copy of the secret keys (never overwrites an existing one)
src_keys = backup_home / "private-keys-v1.d"
dst_keys = target_home / "private-keys-v1.d"
dst_keys.mkdir(parents=True, exist_ok=True)
added = 0
if src_keys.is_dir():
for key in sorted(src_keys.glob("*.key")):
dst = dst_keys / key.name
if not dst.exists():
shutil.copy2(key, dst)
added += 1
# the keybox is a binary database which can't be merged: the backup keys are
# exported and re-imported (the secret keys are already in the copied .key files)
try:
src_provider = get_provider(str(backup_home))
dst_provider = get_provider(str(target_home))
packets = [
src_provider.export_public_key(key)
for key in src_provider.list_public_keys("")
]
for packet in packets:
try:
dst_provider.import_public_key(packet)
except InvalidPacket:
# key already in the target keybox: nothing to import
continue
except GPG_ERRORS as e:
raise BackupError(f"GPG import failed: {e}") from e
_chmod_gpg_home(target_home)
return added
class GpgHandler(GlobalHandler):
name = "gpg"
def _staged_profiles(self, staging: Staging) -> list[str]:
return sorted(
{name.split("/")[1] for name in staging.names() if name.startswith("gpg/")}
)
def export(
self,
local_dir: Path,
staging: Staging,
report: Report,
profiles: list[str] | None = None,
) -> None:
"""Copy the GPG homes to the staging
@param profiles: GPG homes to copy
None to copy all of them
@param report: export report
"""
gpg_dir = local_dir / GPG_DIR
if not gpg_dir.is_dir():
return
wanted = set(profiles) if profiles is not None else None
for home in sorted(gpg_dir.iterdir()):
if home.is_dir() and (wanted is None or home.name in wanted):
export_gpg(home, staging)
def import_(
self,
local_dir: Path,
staging: Staging,
report: Report,
profiles: list[str] | None = None,
force: bool = False,
) -> None:
"""Merge the backup GPG homes into the local ones (union merge)
A keyring which diverged from the backup is skipped unless ``force`` is set (the
backup may be older than the current keys).
@param profiles: GPG homes to import (None means all of them)
"""
wanted = set(profiles) if profiles is not None else None
for profile_name in self._staged_profiles(staging):
if wanted is not None and profile_name not in wanted:
continue
backup_home = staging.path(f"gpg/{profile_name}")
if not backup_home.is_dir() or not any(backup_home.iterdir()):
continue
target_home = local_dir / GPG_DIR / profile_name
added, diverged = check_gpg_restore(target_home, backup_home)
if diverged and not force:
report.skipped.append(
f"gpg: {profile_name}: the current keyring has keys not in "
f"the backup ({', '.join(diverged)})"
)
continue
count = import_gpg(target_home, backup_home)
report.done.append(
f"gpg: keyring merged into {target_home} ({count} new secret key file(s))"
)
for fp in added:
report.done.append(f"gpg: added key {fp}")

166
libervia/backend/tools/backup/handlers.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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
)

48
libervia/backend/tools/backup/models.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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",
)

656
libervia/backend/tools/backup/pie.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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 <row> 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 <item> 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 <item> element, as in the cache.
@param item_id: item id
@param publisher: item publisher (attribute of the <item> 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")

169
libervia/backend/tools/backup/xep_0227.py Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
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 (``<user name=...>``)
@param sections: section elements to include under the ``<user>`` 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 <host> element in pie document")
user = host.find(f"{{{NS_PIE}}}user")
if user is None:
raise PieError("missing <user> 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 <item> 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("<payload>") : -len("</payload>")]
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

33
libervia/backend/tools/common/utils.py
View file

@ -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