diff --git a/README.md b/README.md index f41c2e0..42556cb 100644 --- a/README.md +++ b/README.md @@ -7,11 +7,18 @@ The current foundation provides strict TOML configuration, file-backed secrets, API/local root mapping, fail-fast permission and sparse-file probes, a durable SQLite command inbox, bounded one-writer WebSocket output, registration-first authentication, heartbeat handling, duplicate-command acknowledgements, and -indefinite capped exponential reconnect with optional jitter. +indefinite capped exponential reconnect with optional jitter. The state DB has +an exclusive process lease plus checksummed, integrity-verified online backups +with recent/daily/weekly retention. +Startup probes report qBittorrent/Web API/libtorrent versions, Syncthing +version/device identity, service health, and per-root hardlink/reflink/sparse +support without exposing local paths to control. ```bash archive-client --config /etc/archive-control/client.toml --check-config archive-client --config /etc/archive-control/client.toml --mode archive +archive-client-backup --database /var/lib/archive-control/client.db \ + --backup-dir /var/backups/archive-control list ``` `--mode` accepts only `archive` or `cache` and overrides the configured role. diff --git a/examples/client.toml b/examples/client.toml index c7a5093..d27382c 100644 --- a/examples/client.toml +++ b/examples/client.toml @@ -13,6 +13,12 @@ reconnect_max = "60s" reconnect_reset_after = "60s" reconnect_jitter = true +[backup] +interval = "6h" +recent = 12 +daily = 14 +weekly = 8 + [qbittorrent] endpoint = "http://qbittorrent:8080" username = "admin" @@ -26,4 +32,3 @@ api_key_file = "/run/secrets/syncthing_api_key" api_root = "/sync" local_root = "/data/sync" advertised_addresses = ["dynamic"] - diff --git a/pyproject.toml b/pyproject.toml index 3e84cfc..2ee37ee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ dependencies = ["protobuf==7.35.1", "websockets==16.0"] [project.scripts] archive-client = "archive_clients.cli:main" +archive-client-backup = "archive_clients.backup_cli:main" [tool.setuptools.packages.find] where = ["src"] - diff --git a/src/archive_clients/backup.py b/src/archive_clients/backup.py new file mode 100644 index 0000000..782c2d4 --- /dev/null +++ b/src/archive_clients/backup.py @@ -0,0 +1,227 @@ +"""Verified SQLite online backups and bounded tiered retention.""" + +from __future__ import annotations + +import hashlib +import hmac +import os +import shutil +import sqlite3 +import uuid +from contextlib import closing +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path + +from archive_clients.config import BackupConfig +from archive_clients.locking import DatabaseLease + + +class BackupError(RuntimeError): + pass + + +@dataclass(frozen=True) +class BackupRecord: + database: Path + checksum: Path + sha256: str + + +class SQLiteBackupManager: + def __init__( + self, + database: Path, + backup_dir: Path, + retention: BackupConfig, + busy_timeout_ms: int = 5000, + ): + self.database = database + self.backup_dir = backup_dir + self.retention = retention + self.busy_timeout_ms = busy_timeout_ms + + def create(self, label: str = "scheduled") -> BackupRecord: + if not self.database.is_file(): + raise BackupError("source database does not exist") + if not label or not all( + character.isalnum() or character in "-_" for character in label + ): + raise BackupError("backup label contains unsafe characters") + self.backup_dir.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ") + final = self.backup_dir / f"archive-client-{timestamp}-{label}.db" + checksum = final.with_suffix(final.suffix + ".sha256") + temporary = self.backup_dir / f".{final.name}.{uuid.uuid4().hex}.tmp" + checksum_temporary = self.backup_dir / ( + f".{checksum.name}.{uuid.uuid4().hex}.tmp" + ) + if final.exists() or checksum.exists(): + raise BackupError("backup destination already exists") + try: + with closing(sqlite3.connect(self.database)) as source, closing( + sqlite3.connect(temporary) + ) as target: + source.execute(f"PRAGMA busy_timeout = {self.busy_timeout_ms}") + source.backup(target) + os.chmod(temporary, 0o600) + self._verify_sqlite(temporary) + _fsync_file(temporary) + digest = _sha256(temporary) + os.replace(temporary, final) + with checksum_temporary.open("x", encoding="ascii") as sidecar: + sidecar.write(f"{digest} {final.name}\n") + sidecar.flush() + os.fsync(sidecar.fileno()) + os.chmod(checksum_temporary, 0o600) + os.replace(checksum_temporary, checksum) + _fsync_directory(self.backup_dir) + record = BackupRecord(final, checksum, digest) + self.verify(final) + self.prune() + return record + except Exception: + temporary.unlink(missing_ok=True) + checksum_temporary.unlink(missing_ok=True) + final.unlink(missing_ok=True) + checksum.unlink(missing_ok=True) + raise + + def verify(self, backup: Path) -> BackupRecord: + backup = Path(backup) + checksum = backup.with_suffix(backup.suffix + ".sha256") + if not backup.is_file() or not checksum.is_file(): + raise BackupError("backup database or checksum sidecar is missing") + parts = checksum.read_text(encoding="ascii").strip().split() + if len(parts) != 2 or parts[1] != backup.name: + raise BackupError("backup checksum sidecar is malformed") + actual = _sha256(backup) + if not hmac.compare_digest(parts[0], actual): + raise BackupError("backup checksum does not match") + self._verify_sqlite(backup) + return BackupRecord(backup, checksum, actual) + + def list(self) -> list[BackupRecord]: + if not self.backup_dir.exists(): + return [] + return [ + self.verify(path) + for path in sorted( + self.backup_dir.glob("archive-client-*.db"), reverse=True + ) + ] + + def restore(self, backup: Path) -> Path | None: + verified = self.verify(Path(backup)) + self.database.parent.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ") + temporary = self.database.parent / ( + f".{self.database.name}.restore-{uuid.uuid4().hex}.tmp" + ) + preserved = self.database.with_name( + f"{self.database.name}.suspect-{timestamp}" + ) + moved: list[tuple[Path, Path]] = [] + installed = False + with DatabaseLease(self.database): + try: + shutil.copyfile(verified.database, temporary) + os.chmod(temporary, 0o600) + _fsync_file(temporary) + self._verify_sqlite(temporary) + for source, destination in ( + (self.database, preserved), + (Path(f"{self.database}-wal"), Path(f"{preserved}-wal")), + (Path(f"{self.database}-shm"), Path(f"{preserved}-shm")), + ): + if source.exists(): + os.replace(source, destination) + moved.append((source, destination)) + os.replace(temporary, self.database) + installed = True + _fsync_directory(self.database.parent) + self._verify_sqlite(self.database) + return preserved if moved else None + except Exception: + temporary.unlink(missing_ok=True) + if installed: + self.database.unlink(missing_ok=True) + for source, destination in reversed(moved): + os.replace(destination, source) + _fsync_directory(self.database.parent) + raise + + def prune(self) -> list[Path]: + candidates = sorted( + self.backup_dir.glob("archive-client-*.db"), + key=lambda path: (path.stat().st_mtime_ns, path.name), + reverse=True, + ) + keep = set(candidates[: self.retention.recent]) + keep.update(_newest_per_period(candidates, self.retention.daily, "day")) + keep.update(_newest_per_period(candidates, self.retention.weekly, "week")) + removed = [] + for backup in candidates: + if backup in keep: + continue + backup.unlink(missing_ok=True) + backup.with_suffix(backup.suffix + ".sha256").unlink(missing_ok=True) + removed.append(backup) + if removed: + _fsync_directory(self.backup_dir) + return removed + + @staticmethod + def _verify_sqlite(database: Path) -> None: + uri = f"{database.resolve().as_uri()}?mode=ro" + try: + with closing(sqlite3.connect(uri, uri=True)) as connection: + if connection.execute("PRAGMA integrity_check").fetchall() != [ + ("ok",) + ]: + raise BackupError("SQLite integrity check failed") + if connection.execute("PRAGMA foreign_key_check").fetchall(): + raise BackupError("SQLite foreign-key check failed") + version = connection.execute("PRAGMA user_version").fetchone()[0] + if version != 1: + raise BackupError(f"unsupported backup schema version {version}") + except sqlite3.Error as exc: + raise BackupError("backup is not a readable SQLite database") from exc + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _fsync_file(path: Path) -> None: + with path.open("rb") as source: + os.fsync(source.fileno()) + + +def _fsync_directory(path: Path) -> None: + descriptor = os.open(path, os.O_RDONLY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + + +def _newest_per_period( + candidates: list[Path], limit: int, period: str +) -> list[Path]: + selected = [] + seen: set[object] = set() + for candidate in candidates: + instant = datetime.fromtimestamp(candidate.stat().st_mtime, tz=timezone.utc) + key: object = instant.date() if period == "day" else instant.isocalendar()[:2] + if key in seen: + continue + seen.add(key) + selected.append(candidate) + if len(selected) == limit: + break + return selected diff --git a/src/archive_clients/backup_cli.py b/src/archive_clients/backup_cli.py new file mode 100644 index 0000000..8228885 --- /dev/null +++ b/src/archive_clients/backup_cli.py @@ -0,0 +1,47 @@ +"""Offline list, verify, and restore commands for client backups.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Sequence + +from archive_clients.backup import SQLiteBackupManager +from archive_clients.config import BackupConfig + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="archive-client-backup") + parser.add_argument("--database", type=Path, required=True) + parser.add_argument("--backup-dir", type=Path, required=True) + commands = parser.add_subparsers(dest="command", required=True) + commands.add_parser("list") + verify = commands.add_parser("verify") + verify.add_argument("backup", type=Path) + restore = commands.add_parser("restore") + restore.add_argument("backup", type=Path) + arguments = parser.parse_args(argv) + manager = SQLiteBackupManager( + arguments.database, arguments.backup_dir, BackupConfig() + ) + if arguments.command == "list": + result: object = [ + {"database": str(item.database), "sha256": item.sha256} + for item in manager.list() + ] + elif arguments.command == "verify": + item = manager.verify(arguments.backup) + result = {"database": str(item.database), "sha256": item.sha256} + else: + preserved = manager.restore(arguments.backup) + result = { + "database": str(arguments.database), + "preserved": str(preserved) if preserved else None, + } + print(json.dumps(result, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/archive_clients/cli.py b/src/archive_clients/cli.py index 868afaa..d1bf3d6 100644 --- a/src/archive_clients/cli.py +++ b/src/archive_clients/cli.py @@ -11,7 +11,12 @@ from typing import Sequence from archive_clients.config import ClientConfig from archive_clients.daemon import ArchiveClientDaemon -from archive_clients.probes import probe_root +from archive_clients.logging_config import configure_logging +from archive_clients.probes import probe_root, probe_writable_directory +from archive_clients.services import probe_qbittorrent, probe_syncthing + + +logger = logging.getLogger(__name__) def main(argv: Sequence[str] | None = None) -> int: @@ -20,15 +25,16 @@ def main(argv: Sequence[str] | None = None) -> int: parser.add_argument("--mode", choices=("archive", "cache")) parser.add_argument("--check-config", action="store_true") arguments = parser.parse_args(argv) - logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") config = ClientConfig.load(arguments.config, arguments.mode) probes = [ probe_root(config.qbittorrent.local_root), probe_root(config.syncthing.local_root), ] - config.read_shared_token() - config.qbittorrent.read_password() - config.syncthing.read_api_key() + probe_writable_directory(config.state_db.parent) + probe_writable_directory(config.backup_dir) + shared_token = config.read_shared_token() + qb_password = config.qbittorrent.read_password() + syncthing_key = config.syncthing.read_api_key() if arguments.check_config: print(json.dumps({ "client_id": config.client_id, @@ -36,5 +42,32 @@ def main(argv: Sequence[str] | None = None) -> int: "filesystems": [probe.__dict__ | {"root": str(probe.root)} for probe in probes], }, sort_keys=True)) return 0 - asyncio.run(ArchiveClientDaemon(config, probes).run()) + configure_logging( + secret for secret in (shared_token, qb_password, syncthing_key) if secret + ) + logger.info( + "client_starting", + extra={"client_id": config.client_id, "role": config.role}, + ) + try: + service_probes = [ + probe_qbittorrent(config.qbittorrent), + probe_syncthing(config.syncthing), + ] + for service in service_probes: + logger.info( + "service_probe_completed", + extra={ + "client_id": config.client_id, + "role": config.role, + "service": service.service, + "health": service.state, + }, + ) + asyncio.run(ArchiveClientDaemon(config, probes, service_probes).run()) + except KeyboardInterrupt: + logger.info( + "client_stopped", + extra={"client_id": config.client_id, "role": config.role}, + ) return 0 diff --git a/src/archive_clients/config.py b/src/archive_clients/config.py index 540b0f2..bdb39b3 100644 --- a/src/archive_clients/config.py +++ b/src/archive_clients/config.py @@ -54,10 +54,16 @@ class ServiceConfig: return RootMapping(self.api_root, self.local_root) def read_password(self) -> str | None: - return _secret(self.password_file, "service password") if self.password_file else None + return ( + _secret(self.password_file, "service password") + if self.password_file else None + ) def read_api_key(self) -> str | None: - return _secret(self.api_key_file, "service API key") if self.api_key_file else None + return ( + _secret(self.api_key_file, "service API key") + if self.api_key_file else None + ) @dataclass(frozen=True) @@ -76,6 +82,14 @@ class JobsConfig: stall_after: float = 30 * 60 +@dataclass(frozen=True) +class BackupConfig: + interval: float = 6 * 60 * 60 + recent: int = 12 + daily: int = 14 + weekly: int = 8 + + @dataclass(frozen=True) class ClientConfig: client_id: str @@ -89,6 +103,7 @@ class ClientConfig: syncthing: ServiceConfig connection: ConnectionConfig = ConnectionConfig() jobs: JobsConfig = JobsConfig() + backup: BackupConfig = BackupConfig() @classmethod def load(cls, path: Path, mode_override: str | None = None) -> "ClientConfig": @@ -100,7 +115,7 @@ class ClientConfig: _keys(raw, { "client_id", "display_name", "role", "control_endpoint", "shared_token_file", "state_db", "backup_dir", "connection", - "jobs", "qbittorrent", "syncthing", + "jobs", "backup", "qbittorrent", "syncthing", }, "root") role = mode_override or raw.get("role") if role not in {"archive", "cache"}: @@ -113,14 +128,24 @@ class ClientConfig: raise ConfigError("display_name is invalid") connection = _connection(raw.get("connection", {})) jobs = _jobs(raw.get("jobs", {})) + backup = _backup(raw.get("backup", {})) + state_db = _absolute_path(raw, "state_db") + backup_dir = _absolute_path(raw, "backup_dir") + qbittorrent = _service(raw.get("qbittorrent"), "qbittorrent") + syncthing = _service(raw.get("syncthing"), "syncthing") + for protected in (qbittorrent.local_root, syncthing.local_root): + if _is_within(state_db, protected) or _is_within( + backup_dir, protected + ): + raise ConfigError( + "state_db and backup_dir must be outside data roots" + ) return cls( client_id, display_name, role, _endpoint(raw, "control_endpoint", {"ws", "wss"}), _absolute_path(raw, "shared_token_file"), - _absolute_path(raw, "state_db"), - _absolute_path(raw, "backup_dir"), - _service(raw.get("qbittorrent"), "qbittorrent"), - _service(raw.get("syncthing"), "syncthing"), connection, jobs, + state_db, backup_dir, qbittorrent, syncthing, connection, jobs, + backup, ) def read_shared_token(self) -> str: @@ -146,6 +171,12 @@ def _service(value: Any, name: str) -> ServiceConfig: not isinstance(address, str) or not address for address in addresses ): raise ConfigError(f"{name}.advertised_addresses must be a string array") + if name == "qbittorrent" and ( + username is None or "password_file" not in value + ): + raise ConfigError("qbittorrent username and password_file are required") + if name == "syncthing" and "api_key_file" not in value: + raise ConfigError("syncthing api_key_file is required") return ServiceConfig( _endpoint(value, "endpoint", {"http", "https"}), api_root, _absolute_path(value, "local_root"), username, @@ -160,7 +191,11 @@ def _service(value: Any, name: str) -> ServiceConfig: def _connection(value: Any) -> ConnectionConfig: if not isinstance(value, dict): raise ConfigError("connection must be a table") - _keys(value, {"registration_timeout", "heartbeat_interval", "offline_timeout", "reconnect_initial", "reconnect_max", "reconnect_reset_after", "reconnect_jitter"}, "connection") + _keys(value, { + "registration_timeout", "heartbeat_interval", "offline_timeout", + "reconnect_initial", "reconnect_max", "reconnect_reset_after", + "reconnect_jitter", + }, "connection") result = ConnectionConfig( registration_timeout=_duration(value.get("registration_timeout", "10s")), heartbeat_interval=_duration(value.get("heartbeat_interval", "15s")), @@ -186,12 +221,30 @@ def _jobs(value: Any) -> JobsConfig: return JobsConfig(stall_after=_duration(value.get("stall_after", "30m"))) +def _backup(value: Any) -> BackupConfig: + if not isinstance(value, dict): + raise ConfigError("backup must be a table") + _keys(value, {"interval", "recent", "daily", "weekly"}, "backup") + return BackupConfig( + interval=_duration(value.get("interval", "6h")), + recent=_positive_int(value.get("recent", 12), "backup.recent"), + daily=_positive_int(value.get("daily", 14), "backup.daily"), + weekly=_positive_int(value.get("weekly", 8), "backup.weekly"), + ) + + def _duration(value: Any) -> float: if not isinstance(value, str) or not (match := _DURATION.fullmatch(value)): raise ConfigError("duration must look like 15s or 30m") return int(match.group(1)) * _FACTORS[match.group(2)] +def _positive_int(value: Any, name: str) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < 1: + raise ConfigError(f"{name} must be a positive integer") + return value + + def _string(value: dict[str, Any], key: str) -> str: item = value.get(key) if not isinstance(item, str) or not item: @@ -205,11 +258,19 @@ def _path(value: dict[str, Any], key: str) -> Path: def _absolute_path(value: dict[str, Any], key: str) -> Path: path = _path(value, key) - if not path.is_absolute(): - raise ConfigError(f"{key} must be an absolute path") + if not path.is_absolute() or ".." in path.parts: + raise ConfigError(f"{key} must be an absolute normalized path") return path +def _is_within(path: Path, root: Path) -> bool: + try: + path.resolve(strict=False).relative_to(root.resolve(strict=False)) + return True + except ValueError: + return False + + def _endpoint( value: dict[str, Any], key: str, allowed_schemes: set[str] ) -> str: @@ -220,6 +281,7 @@ def _endpoint( or not parsed.hostname or parsed.username is not None or parsed.password is not None + or parsed.query or parsed.fragment ): schemes = "/".join(sorted(allowed_schemes)) diff --git a/src/archive_clients/daemon.py b/src/archive_clients/daemon.py index 445e529..94ef32c 100644 --- a/src/archive_clients/daemon.py +++ b/src/archive_clients/daemon.py @@ -11,7 +11,9 @@ from typing import Any from websockets.asyncio.client import connect +from archive_clients.backup import SQLiteBackupManager from archive_clients.config import ClientConfig +from archive_clients.locking import DatabaseLease from archive_clients.probes import FilesystemProbe from archive_clients.protocol import ( decode, @@ -20,6 +22,7 @@ from archive_clients.protocol import ( encode_message, new_envelope, ) +from archive_clients.services import ServiceProbe from archive_clients.state import ClientStore, CommandConflict from archive_control.v1 import client_pb2, common_pb2, control_pb2, job_pb2 @@ -28,13 +31,49 @@ logger = logging.getLogger(__name__) class ArchiveClientDaemon: - def __init__(self, config: ClientConfig, probes: list[FilesystemProbe]): + def __init__( + self, + config: ClientConfig, + probes: list[FilesystemProbe], + service_probes: list[ServiceProbe], + ): + if len(probes) != 2: + raise ValueError( + "qBittorrent and Syncthing filesystem probes are required" + ) self.config = config self.probes = probes + self.service_probes = service_probes self.store = ClientStore(config.state_db) + self.backups = SQLiteBackupManager( + config.state_db, config.backup_dir, config.backup + ) + self._lease = DatabaseLease(config.state_db) async def run(self) -> None: - await asyncio.to_thread(self.store.initialize) + await asyncio.to_thread(self._lease.acquire) + backup_task: asyncio.Task[None] | None = None + try: + await asyncio.to_thread(self.store.initialize) + await asyncio.to_thread(self._warn_if_backup_shares_filesystem) + backup_task = asyncio.create_task( + self._backup_loop(), name="archive-client-backups" + ) + await self._connection_loop() + finally: + if backup_task is not None: + backup_task.cancel() + await asyncio.gather(backup_task, return_exceptions=True) + await asyncio.to_thread(self._lease.release) + + def _warn_if_backup_shares_filesystem(self) -> None: + if ( + self.config.state_db.parent.stat().st_dev + == self.config.backup_dir.stat().st_dev + ): + logger.warning("database_and_backup_share_filesystem") + + async def _connection_loop(self) -> None: delay = self.config.connection.reconnect_initial while True: started = time.monotonic() @@ -43,13 +82,39 @@ class ArchiveClientDaemon: except asyncio.CancelledError: raise except Exception as exc: - logger.warning("control connection ended: %s", type(exc).__name__) - if time.monotonic() - started >= self.config.connection.reconnect_reset_after: + logger.warning( + "control_connection_ended", + extra={"error_type": type(exc).__name__}, + ) + if ( + time.monotonic() - started + >= self.config.connection.reconnect_reset_after + ): delay = self.config.connection.reconnect_initial - wait = random.uniform(0, delay) if self.config.connection.reconnect_jitter else delay + wait = ( + random.uniform(0, delay) + if self.config.connection.reconnect_jitter else delay + ) await asyncio.sleep(wait) delay = min(delay * 2, self.config.connection.reconnect_max) + async def _backup_loop(self) -> None: + while True: + await asyncio.sleep(self.config.backup.interval) + try: + record = await asyncio.to_thread(self.backups.create, "scheduled") + logger.info( + "database_backup_created", + extra={"backup_name": record.database.name}, + ) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.error( + "database_backup_failed", + extra={"error_type": type(exc).__name__}, + ) + async def _connection(self) -> None: async with connect( self.config.control_endpoint, ping_interval=None, compression=None, @@ -69,6 +134,7 @@ class ArchiveClientDaemon: raise RuntimeError("control rejected registration") if response.register_response.negotiated_version.major != 1: raise RuntimeError("control negotiated an unsupported protocol version") + logger.info("control_connection_registered") outbound: asyncio.Queue[str] = asyncio.Queue(maxsize=100) writer = asyncio.create_task(self._writer(websocket, outbound)) try: @@ -94,6 +160,23 @@ class ArchiveClientDaemon: request.capabilities.syncthing_advertised_addresses.extend( self.config.syncthing.advertised_addresses ) + for probe in self.service_probes: + health = request.capabilities.services.add() + health.service = probe.service + health.state = probe.state + health.version = probe.version + health.api_version = probe.api_version + health.detail = probe.detail + health.checked_at.FromDatetime(probe.checked_at) + if probe.service == "qbittorrent": + request.capabilities.qbittorrent_version = probe.version + request.capabilities.qbittorrent_web_api_version = ( + probe.api_version + ) + request.capabilities.libtorrent_version = probe.libtorrent_version + elif probe.service == "syncthing": + request.capabilities.syncthing_version = probe.version + request.capabilities.syncthing_device_id = probe.device_id for root_name, probe in zip( ("qbittorrent", "syncthing"), self.probes, strict=True ): @@ -102,9 +185,15 @@ class ArchiveClientDaemon: filesystem.readable = probe.readable filesystem.writable = probe.writable filesystem.hard_link = probe.hard_link + filesystem.reflink = probe.reflink filesystem.sparse_files = probe.sparse_files + if all(probe.hard_link for probe in self.probes): + request.capabilities.features.append(client_pb2.CLIENT_FEATURE_HARD_LINK) + if all(probe.reflink for probe in self.probes): + request.capabilities.features.append(client_pb2.CLIENT_FEATURE_REFLINK) if all(probe.sparse_files for probe in self.probes): request.capabilities.features.append(client_pb2.CLIENT_FEATURE_SPARSE_FILES) + request.capabilities.features.append(client_pb2.CLIENT_FEATURE_DB_BACKUP) for cursor in self.store.list_active_job_cursors(): active = request.active_jobs.add() active.job_id = str(cursor["job_id"]) @@ -132,14 +221,32 @@ class ArchiveClientDaemon: elif payload == "command": await self._accept_command(envelope, outbound) elif payload == "protocol_error": - logger.warning("control reported protocol error code=%s", envelope.protocol_error.error.code) + logger.warning( + "control_reported_protocol_error", + extra={"error_code": envelope.protocol_error.error.code}, + ) async def _accept_command( self, envelope: Any, outbound: asyncio.Queue[str] ) -> None: command = envelope.command - acknowledgement = self._initial_acknowledgement(command) + snapshot_rows: list[dict[str, object]] = [] + if command.WhichOneof("payload") == "request_job_snapshot": + requested = list(command.request_job_snapshot.job_ids) + snapshot_rows = await asyncio.to_thread( + self.store.job_snapshot_rows, requested + ) + missing = set(requested) - { + str(row["job_id"]) for row in snapshot_rows + } + else: + requested = [] + missing = set() + acknowledgement = self._initial_acknowledgement( + command, missing, bool(requested) + ) accepted = None + accepted_for_execution = False try: accepted = await asyncio.to_thread( self.store.accept_command, @@ -155,9 +262,15 @@ class ArchiveClientDaemon: acknowledgement.status == control_pb2.COMMAND_ACK_STATUS_ACCEPTED ): + accepted_for_execution = True acknowledgement.status = ( control_pb2.COMMAND_ACK_STATUS_DUPLICATE ) + else: + accepted_for_execution = ( + acknowledgement.status + == control_pb2.COMMAND_ACK_STATUS_ACCEPTED + ) except CommandConflict: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED acknowledgement.error.code = common_pb2.ERROR_CODE_CONFLICT @@ -169,26 +282,47 @@ class ArchiveClientDaemon: if ( accepted is not None - and not accepted.duplicate - and acknowledgement.status - == control_pb2.COMMAND_ACK_STATUS_ACCEPTED + and accepted_for_execution and command.WhichOneof("payload") == "request_job_snapshot" ): - snapshot = new_envelope() - snapshot.correlation_id = envelope.message_id - snapshot.client_state_snapshot.snapshot_id = str(uuid.uuid4()) - snapshot.client_state_snapshot.observed_at.CopyFrom(snapshot.sent_at) - await outbound.put(encode(snapshot)) + for row in snapshot_rows: + snapshot = new_envelope() + snapshot.correlation_id = envelope.message_id + job_snapshot = snapshot.job_snapshot + job_snapshot.job.definition.CopyFrom(decode_message( + str(row["definition_json"]), job_pb2.JobDefinition() + )) + job_snapshot.job.state = job_pb2.JobState.Value(str(row["state"])) + job_snapshot.job.revision = int(row["revision"]) + job_snapshot.job.committed = bool(row["committed"]) + job_snapshot.job.updated_at.CopyFrom(snapshot.sent_at) + job_snapshot.last_event_sequence = int( + row["last_event_sequence"] + ) + await outbound.put(encode(snapshot)) @staticmethod - def _initial_acknowledgement(command: Any) -> control_pb2.CommandAck: + def _initial_acknowledgement( + command: Any, + missing_snapshot_jobs: set[str], + has_snapshot_jobs: bool, + ) -> control_pb2.CommandAck: acknowledgement = control_pb2.CommandAck(command_id=command.command_id) if not command.command_id or command.WhichOneof("payload") is None: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT acknowledgement.error.message = "command ID and payload are required" elif command.WhichOneof("payload") == "request_job_snapshot": - acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED + if not has_snapshot_jobs: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT + acknowledgement.error.message = "snapshot job IDs are required" + elif missing_snapshot_jobs: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_NOT_FOUND + acknowledgement.error.message = "requested client job is not found" + else: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED else: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED acknowledgement.error.code = common_pb2.ERROR_CODE_UNSUPPORTED diff --git a/src/archive_clients/locking.py b/src/archive_clients/locking.py new file mode 100644 index 0000000..4a1a03f --- /dev/null +++ b/src/archive_clients/locking.py @@ -0,0 +1,49 @@ +"""Exclusive process ownership for the client state database.""" + +from __future__ import annotations + +import fcntl +import os +from pathlib import Path +from typing import IO + + +class DatabaseLockedError(RuntimeError): + pass + + +class DatabaseLease: + def __init__(self, database: Path): + self.path = Path(f"{database}.lock") + self._file: IO[str] | None = None + + def acquire(self) -> None: + if self._file is not None: + return + self.path.parent.mkdir(parents=True, exist_ok=True) + lock_file = self.path.open("a+", encoding="utf-8") + os.chmod(self.path, 0o600) + try: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + lock_file.close() + raise DatabaseLockedError( + "client database is owned by a running process" + ) from exc + self._file = lock_file + + def release(self) -> None: + if self._file is None: + return + try: + fcntl.flock(self._file.fileno(), fcntl.LOCK_UN) + finally: + self._file.close() + self._file = None + + def __enter__(self) -> "DatabaseLease": + self.acquire() + return self + + def __exit__(self, *_: object) -> None: + self.release() diff --git a/src/archive_clients/logging_config.py b/src/archive_clients/logging_config.py new file mode 100644 index 0000000..558e0f7 --- /dev/null +++ b/src/archive_clients/logging_config.py @@ -0,0 +1,50 @@ +"""Small JSON logger that redacts configured secret values.""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timezone +from typing import Iterable + + +class RedactingJsonFormatter(logging.Formatter): + def __init__(self, secrets: Iterable[str]): + super().__init__() + self._secrets = tuple( + sorted((secret for secret in secrets if secret), key=len, reverse=True) + ) + + def format(self, record: logging.LogRecord) -> str: + payload: dict[str, object] = { + "timestamp": datetime.fromtimestamp( + record.created, timezone.utc + ).isoformat(), + "level": record.levelname.lower(), + "logger": record.name, + "event": self._redact(record.getMessage()), + } + for name in ( + "client_id", "role", "error_type", "backup_name", "attempt", + "service", "health", "error_code", + ): + if hasattr(record, name): + payload[name] = self._redact(str(getattr(record, name))) + if record.exc_info: + payload["exception_type"] = record.exc_info[0].__name__ + return json.dumps(payload, sort_keys=True, separators=(",", ":")) + + def _redact(self, value: str) -> str: + for secret in self._secrets: + value = value.replace(secret, "[REDACTED]") + return value + + +def configure_logging(secrets: Iterable[str]) -> None: + handler = logging.StreamHandler() + handler.setFormatter(RedactingJsonFormatter(secrets)) + root = logging.getLogger() + root.handlers.clear() + root.addHandler(handler) + root.setLevel(logging.INFO) + logging.getLogger("websockets").setLevel(logging.WARNING) diff --git a/src/archive_clients/probes.py b/src/archive_clients/probes.py index 2cbdadc..c95a795 100644 --- a/src/archive_clients/probes.py +++ b/src/archive_clients/probes.py @@ -4,6 +4,7 @@ from __future__ import annotations import os import tempfile +import fcntl from dataclasses import dataclass from pathlib import Path @@ -18,9 +19,23 @@ class FilesystemProbe: readable: bool writable: bool hard_link: bool + reflink: bool sparse_files: bool +def probe_writable_directory(path: Path) -> None: + if not path.is_absolute() or not path.is_dir(): + raise ProbeError(f"configured directory is not existing and absolute: {path}") + try: + descriptor, raw_path = tempfile.mkstemp( + prefix=".archive-control-write-probe-", dir=path + ) + os.close(descriptor) + Path(raw_path).unlink() + except OSError as exc: + raise ProbeError(f"configured directory is not writable: {path}") from exc + + def probe_root(root: Path) -> FilesystemProbe: if not root.is_absolute() or not root.is_dir(): raise ProbeError(f"configured root is not an existing absolute directory: {root}") @@ -28,6 +43,7 @@ def probe_root(root: Path) -> FilesystemProbe: raise ProbeError(f"configured root permissions are insufficient: {root}") source: Path | None = None linked: Path | None = None + cloned: Path | None = None try: descriptor, raw_path = tempfile.mkstemp(prefix=".archive-control-probe-", dir=root) source = Path(raw_path) @@ -44,11 +60,20 @@ def probe_root(root: Path) -> FilesystemProbe: hard_link = linked.stat().st_ino == source.stat().st_ino except OSError: hard_link = False - return FilesystemProbe(root, True, True, hard_link, sparse) + cloned = source.with_name(f"{source.name}.clone") + try: + with source.open("rb") as source_file, cloned.open("xb") as clone_file: + fcntl.ioctl(clone_file.fileno(), 0x40049409, source_file.fileno()) + reflink = cloned.stat().st_size == source.stat().st_size + except OSError: + reflink = False + return FilesystemProbe(root, True, True, hard_link, reflink, sparse) except OSError as exc: raise ProbeError(f"filesystem capability probe failed for {root}") from exc finally: if linked is not None: linked.unlink(missing_ok=True) + if cloned is not None: + cloned.unlink(missing_ok=True) if source is not None: source.unlink(missing_ok=True) diff --git a/src/archive_clients/services.py b/src/archive_clients/services.py new file mode 100644 index 0000000..599bd7f --- /dev/null +++ b/src/archive_clients/services.py @@ -0,0 +1,171 @@ +"""Bounded, read-only startup probes for local service capabilities.""" + +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from datetime import datetime, timezone +from http.cookiejar import CookieJar +from typing import Any +from urllib import error, parse, request + +from archive_clients.config import ServiceConfig +from archive_control.v1 import common_pb2 + + +_MAX_RESPONSE_BYTES = 1024 * 1024 +_QB_VERSION = re.compile(r"^v?(\d+)\.(\d+)(?:\.|$)") + + +@dataclass(frozen=True) +class ServiceProbe: + service: str + state: int + checked_at: datetime + version: str = "" + api_version: str = "" + detail: str = "" + device_id: str = "" + libtorrent_version: str = "" + + +class _NoRedirect(request.HTTPRedirectHandler): + def redirect_request(self, *_: Any, **__: Any) -> None: + return None + + +def probe_qbittorrent( + config: ServiceConfig, timeout: float = 10 +) -> ServiceProbe: + checked_at = datetime.now(timezone.utc) + try: + opener = request.build_opener( + request.HTTPCookieProcessor(CookieJar()), _NoRedirect() + ) + credentials = parse.urlencode({ + "username": config.username, + "password": config.read_password(), + }).encode("utf-8") + login = request.Request( + _url(config.endpoint, "/api/v2/auth/login"), + data=credentials, + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + if _read(opener.open(login, timeout=timeout)).strip() != "Ok.": + raise PermissionError("qBittorrent authentication failed") + version = _text_get(opener, config.endpoint, "/api/v2/app/version", timeout) + api_version = _text_get( + opener, config.endpoint, "/api/v2/app/webapiVersion", timeout + ) + build = _json_get( + opener, config.endpoint, "/api/v2/app/buildInfo", timeout + ) + libtorrent = build.get("libtorrent", "") + if not all(isinstance(value, str) and value for value in ( + version, api_version, libtorrent, + )): + raise ValueError("qBittorrent returned incomplete version data") + if not _supported_qb_version(version): + return ServiceProbe( + "qbittorrent", common_pb2.HEALTH_STATE_UNHEALTHY, + checked_at, version=version, api_version=api_version, + detail="unsupported qBittorrent version", + libtorrent_version=libtorrent, + ) + return ServiceProbe( + "qbittorrent", common_pb2.HEALTH_STATE_HEALTHY, checked_at, + version=version, api_version=api_version, + libtorrent_version=libtorrent, + ) + except (PermissionError, error.HTTPError) as exc: + detail = "authentication failed" if isinstance( + exc, PermissionError + ) or getattr(exc, "code", 0) in {401, 403} else "HTTP probe failed" + return ServiceProbe( + "qbittorrent", common_pb2.HEALTH_STATE_UNHEALTHY, + checked_at, detail=detail, + ) + except (error.URLError, OSError, UnicodeError, ValueError, json.JSONDecodeError): + return ServiceProbe( + "qbittorrent", common_pb2.HEALTH_STATE_DEGRADED, + checked_at, detail="service unavailable or incompatible", + ) + + +def probe_syncthing( + config: ServiceConfig, timeout: float = 10 +) -> ServiceProbe: + checked_at = datetime.now(timezone.utc) + try: + opener = request.build_opener(_NoRedirect()) + headers = {"X-API-Key": config.read_api_key() or ""} + version = _json_get( + opener, config.endpoint, "/rest/system/version", timeout, headers + ) + status = _json_get( + opener, config.endpoint, "/rest/system/status", timeout, headers + ) + service_version = version.get("longVersion") or version.get("version") + device_id = status.get("myID") + if not isinstance(service_version, str) or not service_version: + raise ValueError("Syncthing returned no version") + if not isinstance(device_id, str) or not device_id: + raise ValueError("Syncthing returned no device ID") + return ServiceProbe( + "syncthing", common_pb2.HEALTH_STATE_HEALTHY, checked_at, + version=service_version, device_id=device_id, + ) + except error.HTTPError as exc: + detail = ( + "authentication failed" + if exc.code in {401, 403} else "HTTP probe failed" + ) + return ServiceProbe( + "syncthing", common_pb2.HEALTH_STATE_UNHEALTHY, + checked_at, detail=detail, + ) + except (error.URLError, OSError, UnicodeError, ValueError, json.JSONDecodeError): + return ServiceProbe( + "syncthing", common_pb2.HEALTH_STATE_DEGRADED, + checked_at, detail="service unavailable or incompatible", + ) + + +def _url(endpoint: str, path: str) -> str: + return f"{endpoint.rstrip('/')}{path}" + + +def _supported_qb_version(version: str) -> bool: + match = _QB_VERSION.match(version) + if match is None: + return False + major, minor = int(match.group(1)), int(match.group(2)) + return major == 5 or (major == 4 and minor in {5, 6}) + + +def _text_get( + opener: Any, endpoint: str, path: str, timeout: float, + headers: dict[str, str] | None = None, +) -> str: + call = request.Request(_url(endpoint, path), headers=headers or {}) + return _read(opener.open(call, timeout=timeout)).strip() + + +def _json_get( + opener: Any, endpoint: str, path: str, timeout: float, + headers: dict[str, str] | None = None, +) -> dict[str, Any]: + value = json.loads(_text_get(opener, endpoint, path, timeout, headers)) + if not isinstance(value, dict): + raise ValueError("service response is not an object") + return value + + +def _read(response: Any) -> str: + with response: + data = response.read(_MAX_RESPONSE_BYTES + 1) + if len(data) > _MAX_RESPONSE_BYTES: + raise ValueError("service response exceeds limit") + return data.decode("utf-8") diff --git a/src/archive_clients/state.py b/src/archive_clients/state.py index 81011e4..09cccd3 100644 --- a/src/archive_clients/state.py +++ b/src/archive_clients/state.py @@ -4,7 +4,9 @@ from __future__ import annotations import hashlib import json +import os import sqlite3 +import stat from dataclasses import dataclass from pathlib import Path @@ -13,6 +15,10 @@ class CommandConflict(RuntimeError): pass +class JobConflict(RuntimeError): + pass + + @dataclass(frozen=True) class CommandAcceptance: duplicate: bool @@ -25,6 +31,13 @@ class ClientStore: def initialize(self) -> None: self.database.parent.mkdir(parents=True, exist_ok=True) + existed = self.database.exists() + if existed: + metadata = self.database.stat() + if not stat.S_ISREG(metadata.st_mode): + raise RuntimeError("client database is not a regular file") + if metadata.st_mode & (stat.S_IRWXG | stat.S_IRWXO): + raise RuntimeError("client database file permissions are unsafe") with self._connect() as connection: version = connection.execute("PRAGMA user_version").fetchone()[0] if version > 1: @@ -35,6 +48,8 @@ class ClientStore: connection.execute("PRAGMA user_version = 1") if connection.execute("PRAGMA foreign_key_check").fetchall(): raise RuntimeError("client database foreign-key check failed") + if not existed: + os.chmod(self.database, 0o600) def accept_command( self, command_id: str, command_json: str, acknowledgement_json: str @@ -45,7 +60,10 @@ class ClientStore: with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") existing = connection.execute( - "SELECT payload_sha256, command_json, acknowledgement_json FROM commands WHERE command_id = ?", + """ + SELECT payload_sha256, command_json, acknowledgement_json + FROM commands WHERE command_id = ? + """, (command_id,), ).fetchone() if existing: @@ -75,6 +93,78 @@ class ClientStore: ).fetchall() return [dict(row) for row in rows] + def job_snapshot_rows( + self, job_ids: list[str] + ) -> list[dict[str, object]]: + if not job_ids: + return [] + placeholders = ",".join("?" for _ in job_ids) + with self._connect() as connection: + rows = connection.execute( + f""" + SELECT job_id, definition_json, state, revision, + last_event_sequence, committed + FROM jobs WHERE job_id IN ({placeholders}) ORDER BY job_id + """, + job_ids, + ).fetchall() + return [dict(row) for row in rows] + + def save_job( + self, + job_id: str, + definition_json: str, + state: str, + revision: int, + last_event_sequence: int, + committed: bool, + ) -> None: + definition = _canonical(json.loads(definition_json)) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + """ + SELECT definition_json, state, revision, + last_event_sequence, committed + FROM jobs WHERE job_id = ? + """, + (job_id,), + ).fetchone() + if existing and existing["definition_json"] != definition: + raise JobConflict("job definition is immutable") + if existing and ( + revision < existing["revision"] + or last_event_sequence < existing["last_event_sequence"] + or (existing["committed"] and not committed) + ): + raise JobConflict("job cursor cannot move backwards") + if existing and ( + revision == existing["revision"] + and last_event_sequence == existing["last_event_sequence"] + and ( + state != existing["state"] + or int(committed) != existing["committed"] + ) + ): + raise JobConflict("equal job cursor has conflicting state") + connection.execute( + """ + INSERT INTO jobs ( + job_id, definition_json, state, revision, + last_event_sequence, committed + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(job_id) DO UPDATE SET + state = excluded.state, + revision = excluded.revision, + last_event_sequence = excluded.last_event_sequence, + committed = excluded.committed + """, + ( + job_id, definition, state, revision, + last_event_sequence, int(committed), + ), + ) + def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self.database, isolation_level=None, timeout=5) connection.row_factory = sqlite3.Row diff --git a/tests/test_backup.py b/tests/test_backup.py new file mode 100644 index 0000000..b2ea464 --- /dev/null +++ b/tests/test_backup.py @@ -0,0 +1,57 @@ +import os +import tempfile +import unittest +from pathlib import Path + +from archive_clients.backup import BackupError, SQLiteBackupManager +from archive_clients.config import BackupConfig +from archive_clients.locking import DatabaseLease, DatabaseLockedError +from archive_clients.state import ClientStore + + +class BackupTests(unittest.TestCase): + def test_create_verify_detect_corruption_and_restore(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + database = root / "state" / "client.db" + backups = root / "backups" + store = ClientStore(database) + store.initialize() + manager = SQLiteBackupManager(database, backups, BackupConfig()) + record = manager.create("test") + self.assertEqual(manager.verify(record.database).sha256, record.sha256) + + database.write_bytes(b"broken") + preserved = manager.restore(record.database) + self.assertIsNotNone(preserved) + self.assertEqual(ClientStore(database).list_active_job_cursors(), []) + + with record.database.open("ab") as target: + target.write(b"corruption") + with self.assertRaisesRegex(BackupError, "checksum"): + manager.verify(record.database) + + def test_restore_refuses_a_live_database_lease(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + database = root / "client.db" + ClientStore(database).initialize() + manager = SQLiteBackupManager(database, root / "backups", BackupConfig()) + record = manager.create("test") + with DatabaseLease(database): + with self.assertRaises(DatabaseLockedError): + manager.restore(record.database) + + def test_backup_database_has_private_permissions(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + database = root / "client.db" + ClientStore(database).initialize() + record = SQLiteBackupManager( + database, root / "backups", BackupConfig() + ).create("test") + self.assertEqual(os.stat(record.database).st_mode & 0o777, 0o600) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py index 7e94240..dbb273c 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -73,6 +73,18 @@ class ConfigTests(unittest.TestCase): with self.assertRaisesRegex(ConfigError, "unknown jobs"): ClientConfig.load(path) + def test_state_and_backups_cannot_live_under_data_roots(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + path = root / "client.toml" + nested = _config(root).replace( + f'state_db = "{root / "state.db"}"', + f'state_db = "{root / "qb/state.db"}"', + ) + path.write_text(nested, encoding="utf-8") + with self.assertRaisesRegex(ConfigError, "outside data roots"): + ClientConfig.load(path) + def _config(root: Path, role: str = "cache") -> str: return f'''client_id = "cache-1" diff --git a/tests/test_daemon.py b/tests/test_daemon.py index e1efc66..f50d875 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -2,6 +2,7 @@ import asyncio import os import tempfile import unittest +from datetime import datetime, timezone from pathlib import Path, PurePosixPath from uuid import uuid4 @@ -10,13 +11,15 @@ from websockets.asyncio.server import serve from archive_clients.config import ClientConfig, ConnectionConfig, ServiceConfig from archive_clients.daemon import ArchiveClientDaemon from archive_clients.probes import FilesystemProbe -from archive_clients.protocol import decode, encode, new_envelope -from archive_control.v1 import client_pb2, common_pb2, control_pb2 +from archive_clients.services import ServiceProbe +from archive_clients.protocol import decode, encode, encode_message, new_envelope +from archive_control.v1 import client_pb2, common_pb2, control_pb2, job_pb2 class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): async def test_registration_heartbeat_and_duplicate_command(self): observed = {} + job_id = str(uuid4()) async def control(websocket): registration = decode(await websocket.recv()) @@ -29,6 +32,13 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): registration.register_request.capabilities .syncthing_advertised_addresses ) + observed["device_id"] = ( + registration.register_request.capabilities.syncthing_device_id + ) + observed["services"] = [ + item.service + for item in registration.register_request.capabilities.services + ] response = new_envelope() response.correlation_id = registration.message_id response.register_response.status = client_pb2.REGISTRATION_STATUS_ACCEPTED @@ -42,16 +52,21 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): command = new_envelope() command.command.command_id = command_id command.command.created_at.CopyFrom(command.sent_at) - command.command.request_job_snapshot.job_ids.append(str(uuid4())) + command.command.request_job_snapshot.job_ids.append(job_id) await websocket.send(encode(command)) observed["first"] = decode(await websocket.recv()).command_ack.status - observed["snapshot"] = ( - decode(await websocket.recv()).WhichOneof("payload") + snapshot = decode(await websocket.recv()) + observed["snapshot"] = snapshot.WhichOneof("payload") + observed["snapshot_job_id"] = ( + snapshot.job_snapshot.job.definition.job_id ) duplicate = new_envelope() duplicate.command.CopyFrom(command.command) await websocket.send(encode(duplicate)) observed["second"] = decode(await websocket.recv()).command_ack.status + observed["duplicate_snapshot"] = ( + decode(await websocket.recv()).WhichOneof("payload") + ) unsupported = new_envelope() unsupported.command.command_id = str(uuid4()) unsupported.command.created_at.CopyFrom(unsupported.sent_at) @@ -83,18 +98,42 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): root / "state.db", root / "backups", service, service, ConnectionConfig(registration_timeout=2), ) - probe = FilesystemProbe(root, True, True, True, True) - daemon = ArchiveClientDaemon(config, [probe, probe]) + probe = FilesystemProbe(root, True, True, True, True, True) + service_probe = ServiceProbe( + "syncthing", common_pb2.HEALTH_STATE_HEALTHY, + datetime.now(timezone.utc), version="v2", device_id="DEVICE", + ) + daemon = ArchiveClientDaemon( + config, [probe, probe], [service_probe] + ) await asyncio.to_thread(daemon.store.initialize) + definition = job_pb2.JobDefinition( + job_id=job_id, + operation=job_pb2.JOB_OPERATION_ARCHIVE, + ) + definition.created_at.GetCurrentTime() + await asyncio.to_thread( + daemon.store.save_job, + job_id, + encode_message(definition), + "JOB_STATE_WAITING", + 2, + 3, + False, + ) await daemon._connection() self.assertEqual(observed["token"], "shared-secret") self.assertEqual(observed["root_names"], ["qbittorrent", "syncthing"]) self.assertEqual(observed["addresses"], ["dynamic"]) + self.assertEqual(observed["device_id"], "DEVICE") + self.assertEqual(observed["services"], ["syncthing"]) self.assertEqual(observed["heartbeat"], 7) self.assertEqual(observed["first"], control_pb2.COMMAND_ACK_STATUS_ACCEPTED) self.assertEqual(observed["second"], control_pb2.COMMAND_ACK_STATUS_DUPLICATE) - self.assertEqual(observed["snapshot"], "client_state_snapshot") + self.assertEqual(observed["snapshot"], "job_snapshot") + self.assertEqual(observed["snapshot_job_id"], job_id) + self.assertEqual(observed["duplicate_snapshot"], "job_snapshot") self.assertEqual( observed["rejected"], ( diff --git a/tests/test_logging.py b/tests/test_logging.py new file mode 100644 index 0000000..1f09647 --- /dev/null +++ b/tests/test_logging.py @@ -0,0 +1,22 @@ +import json +import logging +import unittest + +from archive_clients.logging_config import RedactingJsonFormatter + + +class LoggingTests(unittest.TestCase): + def test_secret_is_redacted_from_message_and_fields(self): + formatter = RedactingJsonFormatter(["shared-secret"]) + record = logging.LogRecord( + "test", logging.INFO, __file__, 1, + "token=%s", ("shared-secret",), None, + ) + record.error_type = "shared-secret" + rendered = formatter.format(record) + self.assertNotIn("shared-secret", rendered) + self.assertEqual(json.loads(rendered)["error_type"], "[REDACTED]") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_protocol_and_probes.py b/tests/test_protocol_and_probes.py index d726a46..37e5d67 100644 --- a/tests/test_protocol_and_probes.py +++ b/tests/test_protocol_and_probes.py @@ -3,7 +3,7 @@ import tempfile import unittest from pathlib import Path -from archive_clients.probes import probe_root +from archive_clients.probes import probe_root, probe_writable_directory from archive_clients.protocol import ProtocolError, decode, encode, new_envelope @@ -26,6 +26,9 @@ class ProtocolAndProbeTests(unittest.TestCase): result = probe_root(root) self.assertTrue(result.readable) self.assertTrue(result.writable) + self.assertIsInstance(result.reflink, bool) + self.assertEqual(list(root.iterdir()), []) + probe_writable_directory(root) self.assertEqual(list(root.iterdir()), []) diff --git a/tests/test_services.py b/tests/test_services.py new file mode 100644 index 0000000..07dea84 --- /dev/null +++ b/tests/test_services.py @@ -0,0 +1,92 @@ +import io +import os +import tempfile +import unittest +from pathlib import Path, PurePosixPath +from unittest.mock import patch + +from archive_clients.config import ServiceConfig +from archive_clients.services import probe_qbittorrent, probe_syncthing +from archive_control.v1 import common_pb2 + + +class _Response: + def __init__(self, value: str): + self._source = io.BytesIO(value.encode("utf-8")) + + def read(self, size: int = -1) -> bytes: + return self._source.read(size) + + def __enter__(self): + return self + + def __exit__(self, *_): + return None + + +class _Opener: + def __init__(self, responses: list[str]): + self.responses = iter(responses) + self.requests = [] + + def open(self, call, timeout): + self.requests.append((call, timeout)) + return _Response(next(self.responses)) + + +class ServiceProbeTests(unittest.TestCase): + def test_qbittorrent_versions_are_normalized(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + password = self._secret(root, "password", "private") + config = ServiceConfig( + "http://qb:8080", PurePosixPath("/downloads"), root, + username="admin", password_file=password, + ) + opener = _Opener([ + "Ok.", "v5.0.4", "2.11.4", '{"libtorrent":"2.0.11"}', + ]) + with patch( + "archive_clients.services.request.build_opener", + return_value=opener, + ): + result = probe_qbittorrent(config) + self.assertEqual(result.state, common_pb2.HEALTH_STATE_HEALTHY) + self.assertEqual(result.version, "v5.0.4") + self.assertEqual(result.api_version, "2.11.4") + self.assertEqual(result.libtorrent_version, "2.0.11") + self.assertIn(b"password=private", opener.requests[0][0].data) + + def test_syncthing_identity_is_normalized(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + api_key = self._secret(root, "api-key", "private") + config = ServiceConfig( + "http://syncthing:8384", PurePosixPath("/sync"), root, + api_key_file=api_key, + ) + opener = _Opener([ + '{"version":"v2.0.1","longVersion":"syncthing v2.0.1"}', + '{"myID":"DEVICE-ID"}', + ]) + with patch( + "archive_clients.services.request.build_opener", + return_value=opener, + ): + result = probe_syncthing(config) + self.assertEqual(result.state, common_pb2.HEALTH_STATE_HEALTHY) + self.assertEqual(result.device_id, "DEVICE-ID") + self.assertEqual( + opener.requests[0][0].headers["X-api-key"], "private" + ) + + @staticmethod + def _secret(root: Path, name: str, value: str) -> Path: + path = root / name + path.write_text(value, encoding="utf-8") + os.chmod(path, 0o600) + return path + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_state.py b/tests/test_state.py index 2b7ee52..1e6d468 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -1,9 +1,10 @@ import tempfile import unittest +import os from pathlib import Path from uuid import uuid4 -from archive_clients.state import ClientStore, CommandConflict +from archive_clients.state import ClientStore, CommandConflict, JobConflict class ClientStoreTests(unittest.TestCase): @@ -12,6 +13,7 @@ class ClientStoreTests(unittest.TestCase): database = Path(directory) / "state.db" store = ClientStore(database) store.initialize() + self.assertEqual(os.stat(database).st_mode & 0o777, 0o600) command_id = str(uuid4()) first = store.accept_command(command_id, '{"b":2,"a":1}', '{"ok":true}') duplicate = ClientStore(database).accept_command( @@ -23,6 +25,32 @@ class ClientStoreTests(unittest.TestCase): with self.assertRaises(CommandConflict): store.accept_command(command_id, '{"a":2}', '{"ok":true}') + def test_job_definition_is_immutable_while_cursor_advances(self): + with tempfile.TemporaryDirectory() as directory: + store = ClientStore(Path(directory) / "state.db") + store.initialize() + store.save_job( + "job-1", '{"jobId":"job-1"}', "JOB_STATE_WAITING", + 1, 2, False, + ) + store.save_job( + "job-1", '{"jobId":"job-1"}', "JOB_STATE_RUNNING", + 2, 3, False, + ) + self.assertEqual( + store.job_snapshot_rows(["job-1"])[0]["revision"], 2 + ) + with self.assertRaisesRegex(JobConflict, "backwards"): + store.save_job( + "job-1", '{"jobId":"job-1"}', "JOB_STATE_WAITING", + 1, 2, False, + ) + with self.assertRaises(JobConflict): + store.save_job( + "job-1", '{"jobId":"other"}', "JOB_STATE_RUNNING", + 2, 3, False, + ) + if __name__ == "__main__": unittest.main()