diff --git a/README.md b/README.md index 02fab75..ba9566b 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,12 @@ Durable inventory commands now stream bounded atomic summary, lookup, and content-tree chunks. Page tokens are guarded by timestamp-independent snapshot revisions, stale trees fail explicitly, and slow scans run outside the socket reader so heartbeat acknowledgements remain responsive. +Durable `EnsureRoute` commands use scoped Syncthing device/folder REST updates, +safe API/local path mapping, exact pairwise configuration read-back, and +bidirectional nonce/acknowledgement files. Route attempt nonces, ownership, and +ordered updates survive restart; accepted attempts resume on reconnect without +concurrent duplicate execution. Conflicting existing folders and foreign data +are reported without being overwritten. ```bash archive-client --config /etc/archive-control/client.toml --check-config @@ -38,10 +44,10 @@ archive-client-backup --database /var/lib/archive-control/client.db \ Secrets must be regular files without group/world permissions. The daemon never stores them in SQLite or sends the shared token after registration. -This foundation currently executes heartbeat and state-snapshot commands. -Other mutation commands are durably rejected as unsupported until their -service and file-operation executors are added; they are never falsely -acknowledged as accepted. +This foundation currently executes heartbeat, inventory, state-snapshot, and +route-provisioning commands. Other mutation commands are durably rejected as +unsupported until their service and file-operation executors are added; they +are never falsely acknowledged as accepted. Run tests and build using containers: diff --git a/src/archive_clients/backup.py b/src/archive_clients/backup.py index 782c2d4..15253b1 100644 --- a/src/archive_clients/backup.py +++ b/src/archive_clients/backup.py @@ -15,6 +15,7 @@ from pathlib import Path from archive_clients.config import BackupConfig from archive_clients.locking import DatabaseLease +from archive_clients.state import SCHEMA_VERSION class BackupError(RuntimeError): @@ -183,7 +184,7 @@ class SQLiteBackupManager: 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: + if version != SCHEMA_VERSION: raise BackupError(f"unsupported backup schema version {version}") except sqlite3.Error as exc: raise BackupError("backup is not a readable SQLite database") from exc diff --git a/src/archive_clients/daemon.py b/src/archive_clients/daemon.py index d5958e1..727f6ea 100644 --- a/src/archive_clients/daemon.py +++ b/src/archive_clients/daemon.py @@ -26,8 +26,14 @@ from archive_clients.protocol import ( from archive_clients.qbittorrent import QBittorrentReader from archive_clients.services import ServiceProbe from archive_clients.state import ClientStore, CommandConflict +from archive_clients.syncthing import ( + RoutePathConflict, + RouteSetupError, + RouteSetupTimeout, + SyncthingRouteManager, +) from archive_control.v1 import ( - client_pb2, common_pb2, control_pb2, inventory_pb2, job_pb2, + client_pb2, common_pb2, control_pb2, inventory_pb2, job_pb2, route_pb2, ) @@ -41,6 +47,7 @@ class ArchiveClientDaemon: probes: list[FilesystemProbe], service_probes: list[ServiceProbe], resource_reader: QBittorrentReader | None = None, + route_manager: SyncthingRouteManager | None = None, ): if len(probes) != 2: raise ValueError( @@ -53,11 +60,25 @@ class ArchiveClientDaemon: InventoryService(resource_reader, config.client_id) if resource_reader is not None else None ) + healthy_syncthing = any( + probe.service == "syncthing" + and probe.state == common_pb2.HEALTH_STATE_HEALTHY + for probe in service_probes + ) + self.routes = route_manager or ( + SyncthingRouteManager( + config.syncthing, + sparse_supported=probes[1].sparse_files, + ) + if healthy_syncthing and probes[1].writable + else None + ) self.store = ClientStore(config.state_db) self.backups = SQLiteBackupManager( config.state_db, config.backup_dir, config.backup ) self._lease = DatabaseLease(config.state_db) + self._active_route_commands: set[str] = set() async def run(self) -> None: await asyncio.to_thread(self._lease.acquire) @@ -148,6 +169,7 @@ class ArchiveClientDaemon: writer = asyncio.create_task(self._writer(websocket, outbound)) command_tasks: set[asyncio.Task[None]] = set() try: + await self._resume_route_commands(outbound, command_tasks) async for frame in websocket: await self._handle(decode(frame), outbound, command_tasks) finally: @@ -214,6 +236,10 @@ class ArchiveClientDaemon: client_pb2.CLIENT_FEATURE_INVENTORY_CHUNKS, client_pb2.CLIENT_FEATURE_CONTENT_TREE, )) + if self.routes is not None: + request.capabilities.features.append( + client_pb2.CLIENT_FEATURE_ROUTE_PROVISIONING + ) for cursor in self.store.list_active_job_cursors(): active = request.active_jobs.add() active.job_id = str(cursor["job_id"]) @@ -344,6 +370,67 @@ class ArchiveClientDaemon: completed, command_tasks ) ) + elif ( + accepted is not None + and accepted_for_execution + and command.WhichOneof("payload") == "ensure_route" + and self.routes is not None + ): + self._schedule_route_command( + command, + envelope.message_id, + outbound, + command_tasks, + ) + + async def _resume_route_commands( + self, + outbound: asyncio.Queue[str], + command_tasks: set[asyncio.Task[None]], + ) -> None: + for row in await asyncio.to_thread(self.store.list_accepted_commands): + acknowledgement = decode_message( + str(row["acknowledgement_json"]), control_pb2.CommandAck() + ) + if acknowledgement.status != control_pb2.COMMAND_ACK_STATUS_ACCEPTED: + continue + command = decode_message( + str(row["command_json"]), control_pb2.Command() + ) + if command.WhichOneof("payload") == "ensure_route": + self._schedule_route_command( + command, "", outbound, command_tasks + ) + + def _schedule_route_command( + self, + command: control_pb2.Command, + correlation_id: str, + outbound: asyncio.Queue[str], + command_tasks: set[asyncio.Task[None]], + ) -> None: + if command.command_id in self._active_route_commands: + return + self._active_route_commands.add(command.command_id) + task = asyncio.create_task( + self._execute_route(command, correlation_id, outbound), + name=f"route-{command.ensure_route.route.route_id}", + ) + command_tasks.add(task) + task.add_done_callback( + lambda completed: self._route_command_finished( + command.command_id, completed, command_tasks + ) + ) + + def _route_command_finished( + self, + command_id: str, + task: asyncio.Task[None], + command_tasks: set[asyncio.Task[None]], + ) -> None: + self._active_route_commands.discard(command_id) + self._command_finished(task, command_tasks) async def _send_inventory( self, @@ -359,6 +446,207 @@ class ArchiveClientDaemon: response.inventory_chunk.CopyFrom(chunk) await outbound.put(encode(response)) + async def _execute_route( + self, + command: Any, + correlation_id: str, + outbound: asyncio.Queue[str], + ) -> None: + assert self.routes is not None + spec = command.ensure_route.route + attempt = await asyncio.to_thread( + self.store.begin_route_attempt, + command.command_id, + spec.route_id, + encode_message(spec), + str(uuid.uuid4()), + ) + for row in await asyncio.to_thread( + self.store.route_update_rows, command.command_id + ): + response = new_envelope() + response.correlation_id = correlation_id + response.route_update.CopyFrom( + decode_message(str(row["update_json"]), control_pb2.RouteUpdate()) + ) + await outbound.put(encode(response)) + if attempt["state"] in {"ready", "failed"}: + return + + deadline = time.monotonic() + spec.setup_timeout_seconds + configured = None + try: + attempt = await asyncio.to_thread( + self.store.get_route_attempt, command.command_id + ) + if int(attempt["last_sequence"]) == 0: + await self._emit_route_update( + command.command_id, + correlation_id, + outbound, + sequence=1, + state="provisioning", + local_route=self._local_route( + spec, route_pb2.ROUTE_STATE_PROVISIONING + ), + nonce=str(attempt["nonce"]), + detail="route command accepted; configuring Syncthing", + ) + configured = await asyncio.to_thread( + self.routes.configure, spec, deadline + ) + await asyncio.to_thread( + self.store.record_route_ownership, + command.command_id, + configured.created_device, + configured.created_folder, + ) + attempt = await asyncio.to_thread( + self.store.get_route_attempt, command.command_id + ) + configured.local_route.archive_control_created = bool( + attempt["created_folder"] + ) + if int(attempt["last_sequence"]) < 2: + configured.local_route.state = route_pb2.ROUTE_STATE_VERIFYING + configured.local_route.observed_at.GetCurrentTime() + await self._emit_route_update( + command.command_id, + correlation_id, + outbound, + sequence=2, + state="verifying", + local_route=configured.local_route, + nonce=str(attempt["nonce"]), + detail="Syncthing configuration verified; exchanging nonces", + ) + local_seen, peer_seen = await asyncio.to_thread( + self.routes.verify, + configured, + spec, + self.config.client_id, + str(attempt["nonce"]), + deadline, + ) + attempt = await asyncio.to_thread( + self.store.get_route_attempt, command.command_id + ) + if int(attempt["last_sequence"]) < 3: + configured.local_route.state = route_pb2.ROUTE_STATE_READY + configured.local_route.detail = "bidirectional nonce verification passed" + configured.local_route.observed_at.GetCurrentTime() + await self._emit_route_update( + command.command_id, + correlation_id, + outbound, + sequence=3, + state="ready", + local_route=configured.local_route, + nonce=str(attempt["nonce"]), + detail="bidirectional nonce verification passed", + local_nonce_seen_by_peer=local_seen, + peer_nonce_seen_locally=peer_seen, + ) + except Exception as exc: + attempt = await asyncio.to_thread( + self.store.get_route_attempt, command.command_id + ) + if attempt is None or attempt["state"] == "failed": + return + code, retryable = _route_error(exc) + detail = str(exc) or type(exc).__name__ + local_route = ( + configured.local_route + if configured is not None + else self._local_route(spec, route_pb2.ROUTE_STATE_UNHEALTHY) + ) + local_route.state = route_pb2.ROUTE_STATE_UNHEALTHY + local_route.detail = detail + local_route.observed_at.GetCurrentTime() + await self._emit_route_update( + command.command_id, + correlation_id, + outbound, + sequence=int(attempt["last_sequence"]) + 1, + state="failed", + local_route=local_route, + nonce=str(attempt["nonce"]), + detail=detail, + error_code=code, + retryable=retryable, + ) + + async def _emit_route_update( + self, + command_id: str, + correlation_id: str, + outbound: asyncio.Queue[str], + *, + sequence: int, + state: str, + local_route: route_pb2.LocalRoute, + nonce: str, + detail: str, + local_nonce_seen_by_peer: bool = False, + peer_nonce_seen_locally: bool = False, + error_code: int = common_pb2.ERROR_CODE_UNSPECIFIED, + retryable: bool = False, + ) -> None: + update = control_pb2.RouteUpdate( + command_id=command_id, + update_id=str(uuid.uuid4()), + sequence=sequence, + ) + update.route.CopyFrom(local_route) + update.verification.route_id = local_route.route_id + update.verification.nonce = nonce + update.verification.local_nonce_seen_by_peer = local_nonce_seen_by_peer + update.verification.peer_nonce_seen_locally = peer_nonce_seen_locally + update.verification.state = local_route.state + update.verification.detail = detail + update.verification.observed_at.GetCurrentTime() + if error_code != common_pb2.ERROR_CODE_UNSPECIFIED: + update.error.code = error_code + update.error.message = detail + update.error.retryable = retryable + encoded_update = encode_message(update) + await asyncio.to_thread( + self.store.record_route_update, + command_id, + sequence, + state, + encoded_update, + ) + response = new_envelope() + response.correlation_id = correlation_id + response.route_update.CopyFrom(update) + await outbound.put(encode(response)) + + def _local_route( + self, spec: route_pb2.EnsureRouteSpec, state: int + ) -> route_pb2.LocalRoute: + device_id = next( + ( + probe.device_id + for probe in self.service_probes + if probe.service == "syncthing" + ), + "", + ) + route = route_pb2.LocalRoute( + route_id=spec.route_id, + local_relative_path=spec.local_relative_path, + folder_type=route_pb2.SYNCTHING_FOLDER_TYPE_SEND_RECEIVE, + local_syncthing_device_id=device_id, + peer_syncthing_device_ids=[spec.peer_syncthing_device_id], + state=state, + writable=self.probes[1].writable, + sparse_supported=self.probes[1].sparse_files, + detail="route provisioning", + ) + route.observed_at.GetCurrentTime() + return route + @staticmethod def _command_finished( task: asyncio.Task[None], command_tasks: set[asyncio.Task[None]] @@ -411,6 +699,24 @@ class ArchiveClientDaemon: acknowledgement.error.message = "inventory scope is unsupported" else: acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_ACCEPTED + elif command.WhichOneof("payload") == "ensure_route": + spec = command.ensure_route.route + if self.routes is None: + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_UNAVAILABLE + acknowledgement.error.message = "Syncthing route adapter is unavailable" + elif ( + not spec.route_id + or not spec.peer_client_id + or not spec.peer_syncthing_device_id + or not spec.local_relative_path + or spec.setup_timeout_seconds <= 0 + ): + acknowledgement.status = control_pb2.COMMAND_ACK_STATUS_REJECTED + acknowledgement.error.code = common_pb2.ERROR_CODE_INVALID_ARGUMENT + acknowledgement.error.message = "ensure route specification is invalid" + 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 @@ -418,3 +724,15 @@ class ArchiveClientDaemon: "command is not supported by this client build" ) return acknowledgement + + +def _route_error(exc: Exception) -> tuple[int, bool]: + if isinstance(exc, RoutePathConflict): + return common_pb2.ERROR_CODE_PATH_CONFLICT, False + if isinstance(exc, RouteSetupTimeout): + return common_pb2.ERROR_CODE_TIMEOUT, True + if isinstance(exc, PermissionError): + return common_pb2.ERROR_CODE_PERMISSION_DENIED, False + if isinstance(exc, RouteSetupError): + return common_pb2.ERROR_CODE_UNAVAILABLE, True + return common_pb2.ERROR_CODE_INTERNAL, False diff --git a/src/archive_clients/state.py b/src/archive_clients/state.py index 09cccd3..898d3cd 100644 --- a/src/archive_clients/state.py +++ b/src/archive_clients/state.py @@ -11,6 +11,9 @@ from dataclasses import dataclass from pathlib import Path +SCHEMA_VERSION = 2 + + class CommandConflict(RuntimeError): pass @@ -40,12 +43,17 @@ class ClientStore: raise RuntimeError("client database file permissions are unsafe") with self._connect() as connection: version = connection.execute("PRAGMA user_version").fetchone()[0] - if version > 1: + if version > SCHEMA_VERSION: raise RuntimeError( f"client database schema {version} is newer than supported" ) - connection.executescript(_SCHEMA) - connection.execute("PRAGMA user_version = 1") + if version == 0: + connection.executescript(_SCHEMA_V1) + connection.execute("PRAGMA user_version = 1") + version = 1 + if version == 1: + connection.executescript(_SCHEMA_V2) + connection.execute("PRAGMA user_version = 2") if connection.execute("PRAGMA foreign_key_check").fetchall(): raise RuntimeError("client database foreign-key check failed") if not existed: @@ -93,6 +101,16 @@ class ClientStore: ).fetchall() return [dict(row) for row in rows] + def list_accepted_commands(self) -> list[dict[str, object]]: + with self._connect() as connection: + rows = connection.execute( + """ + SELECT command_id, command_json, acknowledgement_json + FROM commands ORDER BY rowid + """ + ).fetchall() + return [dict(row) for row in rows] + def job_snapshot_rows( self, job_ids: list[str] ) -> list[dict[str, object]]: @@ -165,6 +183,133 @@ class ClientStore: ), ) + def begin_route_attempt( + self, + command_id: str, + route_id: str, + spec_json: str, + nonce: str, + ) -> dict[str, object]: + spec = _canonical(json.loads(spec_json)) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + "SELECT * FROM route_attempts WHERE command_id = ?", + (command_id,), + ).fetchone() + if existing: + if ( + existing["route_id"] != route_id + or existing["spec_json"] != spec + ): + raise CommandConflict( + "route command ID was reused with different content" + ) + return dict(existing) + connection.execute( + """ + INSERT INTO route_attempts ( + command_id, route_id, spec_json, nonce, state, + last_sequence + ) VALUES (?, ?, ?, ?, 'accepted', 0) + """, + (command_id, route_id, spec, nonce), + ) + row = connection.execute( + "SELECT * FROM route_attempts WHERE command_id = ?", + (command_id,), + ).fetchone() + return dict(row) + + def record_route_update( + self, + command_id: str, + sequence: int, + state: str, + update_json: str, + ) -> None: + update = _canonical(json.loads(update_json)) + with self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + existing = connection.execute( + """ + SELECT update_json FROM route_attempt_updates + WHERE command_id = ? AND sequence = ? + """, + (command_id, sequence), + ).fetchone() + if existing: + if existing["update_json"] != update: + raise CommandConflict( + "route update sequence has different content" + ) + return + attempt = connection.execute( + """ + SELECT last_sequence FROM route_attempts + WHERE command_id = ? + """, + (command_id,), + ).fetchone() + if attempt is None: + raise CommandConflict("route attempt does not exist") + expected = attempt["last_sequence"] + 1 + if sequence != expected: + raise CommandConflict( + f"route update sequence must be {expected}, received {sequence}" + ) + connection.execute( + """ + INSERT INTO route_attempt_updates ( + command_id, sequence, update_json + ) VALUES (?, ?, ?) + """, + (command_id, sequence, update), + ) + connection.execute( + """ + UPDATE route_attempts + SET state = ?, last_sequence = ? + WHERE command_id = ? + """, + (state, sequence, command_id), + ) + + def record_route_ownership( + self, command_id: str, created_device: bool, created_folder: bool + ) -> None: + with self._connect() as connection: + cursor = connection.execute( + """ + UPDATE route_attempts + SET created_device = MAX(created_device, ?), + created_folder = MAX(created_folder, ?) + WHERE command_id = ? + """, + (int(created_device), int(created_folder), command_id), + ) + if cursor.rowcount != 1: + raise CommandConflict("route attempt does not exist") + + def get_route_attempt(self, command_id: str) -> dict[str, object] | None: + with self._connect() as connection: + row = connection.execute( + "SELECT * FROM route_attempts WHERE command_id = ?", + (command_id,), + ).fetchone() + return dict(row) if row else None + + def route_update_rows(self, command_id: str) -> list[dict[str, object]]: + with self._connect() as connection: + rows = connection.execute( + """ + SELECT sequence, update_json FROM route_attempt_updates + WHERE command_id = ? ORDER BY sequence + """, + (command_id,), + ).fetchall() + return [dict(row) for row in rows] + def _connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self.database, isolation_level=None, timeout=5) connection.row_factory = sqlite3.Row @@ -179,7 +324,7 @@ def _canonical(value: object) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) -_SCHEMA = """ +_SCHEMA_V1 = """ CREATE TABLE IF NOT EXISTS commands ( command_id TEXT PRIMARY KEY, payload_sha256 TEXT NOT NULL, @@ -210,3 +355,23 @@ CREATE TABLE IF NOT EXISTS file_journal ( state TEXT NOT NULL ); """ + + +_SCHEMA_V2 = """ +CREATE TABLE route_attempts ( + command_id TEXT PRIMARY KEY REFERENCES commands(command_id), + route_id TEXT NOT NULL, + spec_json TEXT NOT NULL, + nonce TEXT NOT NULL, + state TEXT NOT NULL, + last_sequence INTEGER NOT NULL DEFAULT 0, + created_device INTEGER NOT NULL DEFAULT 0 CHECK(created_device IN (0, 1)), + created_folder INTEGER NOT NULL DEFAULT 0 CHECK(created_folder IN (0, 1)) +); +CREATE TABLE route_attempt_updates ( + command_id TEXT NOT NULL REFERENCES route_attempts(command_id), + sequence INTEGER NOT NULL, + update_json TEXT NOT NULL, + PRIMARY KEY(command_id, sequence) +); +""" diff --git a/src/archive_clients/syncthing.py b/src/archive_clients/syncthing.py new file mode 100644 index 0000000..d922739 --- /dev/null +++ b/src/archive_clients/syncthing.py @@ -0,0 +1,381 @@ +"""Scoped Syncthing route provisioning and nonce verification.""" + +from __future__ import annotations + +import json +import os +import time +import uuid +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any, Protocol +from urllib import error, parse, request + +from archive_clients.config import ConfigError, ServiceConfig +from archive_control.v1 import route_pb2 + + +class RouteSetupError(RuntimeError): + pass + + +class RoutePathConflict(RouteSetupError): + pass + + +class RouteSetupTimeout(RouteSetupError): + pass + + +class SyncthingTransport(Protocol): + def get_json(self, path: str) -> dict[str, Any]: ... + + def put_json(self, path: str, payload: dict[str, Any]) -> None: ... + + def post(self, path: str) -> None: ... + + +class SyncthingHttp: + def __init__(self, config: ServiceConfig, timeout: float = 10): + self.endpoint = config.endpoint.rstrip("/") + self.api_key = config.read_api_key() or "" + self.timeout = timeout + self.opener = request.build_opener(_NoRedirect()) + + def get_json(self, path: str) -> dict[str, Any]: + response = self._open(path, "GET") + try: + value = json.loads(response.decode("utf-8")) + except (UnicodeError, json.JSONDecodeError) as exc: + raise RouteSetupError("Syncthing returned invalid JSON") from exc + if not isinstance(value, dict): + raise RouteSetupError("Syncthing returned a non-object response") + return value + + def put_json(self, path: str, payload: dict[str, Any]) -> None: + self._open( + path, + "PUT", + json.dumps(payload, separators=(",", ":")).encode("utf-8"), + ) + + def post(self, path: str) -> None: + self._open(path, "POST", b"") + + def _open(self, path: str, method: str, body: bytes | None = None) -> bytes: + headers = {"X-API-Key": self.api_key} + if body is not None: + headers["Content-Type"] = "application/json" + req = request.Request( + f"{self.endpoint}{path}", data=body, headers=headers, method=method + ) + try: + with self.opener.open(req, timeout=self.timeout) as response: + return response.read() + except error.HTTPError as exc: + if exc.code in {401, 403}: + raise PermissionError("Syncthing authentication failed") from exc + raise RouteSetupError( + f"Syncthing HTTP request failed with status {exc.code}" + ) from exc + except (error.URLError, OSError, UnicodeError, json.JSONDecodeError) as exc: + raise RouteSetupError("Syncthing API is unavailable") from exc + + +@dataclass(frozen=True) +class ConfiguredRoute: + local_device_id: str + local_path: Path + local_route: route_pb2.LocalRoute + created_device: bool + created_folder: bool + + +class SyncthingRouteManager: + def __init__( + self, + config: ServiceConfig, + sparse_supported: bool, + transport: SyncthingTransport | None = None, + poll_interval: float = 1, + ): + self.config = config + self.sparse_supported = sparse_supported + self.transport = transport or SyncthingHttp(config) + self.poll_interval = poll_interval + + def configure( + self, spec: route_pb2.EnsureRouteSpec, deadline: float + ) -> ConfiguredRoute: + local_path, api_path = self._route_paths(spec.local_relative_path) + status = self.transport.get_json("/rest/system/status") + local_device_id = status.get("myID") + if not isinstance(local_device_id, str) or not local_device_id: + raise RouteSetupError("Syncthing returned no local device ID") + configuration = self.transport.get_json("/rest/config") + devices = _object_list(configuration, "devices") + folders = _object_list(configuration, "folders") + + folder = _by_key(folders, "id", spec.route_id) + if folder is not None: + self._validate_folder( + folder, + api_path, + local_device_id, + spec.peer_syncthing_device_id, + ) + peer = _by_key(devices, "deviceID", spec.peer_syncthing_device_id) + created_device = peer is None + if peer is None: + addresses = list(spec.peer_addresses) or ["dynamic"] + self.transport.put_json( + "/rest/config/devices/" + + parse.quote(spec.peer_syncthing_device_id, safe=""), + { + "deviceID": spec.peer_syncthing_device_id, + "name": spec.peer_client_id, + "addresses": addresses, + }, + ) + + if folder is None: + self._prepare_new_directory(local_path) + self.transport.put_json( + "/rest/config/folders/" + parse.quote(spec.route_id, safe=""), + { + "id": spec.route_id, + "label": f"archive-control:{spec.route_id}", + "path": api_path, + "type": "sendreceive", + "devices": [ + {"deviceID": local_device_id}, + {"deviceID": spec.peer_syncthing_device_id}, + ], + }, + ) + else: + local_path.mkdir(parents=True, exist_ok=True) + + while True: + current = self.transport.get_json("/rest/config") + current_folder = _by_key( + _object_list(current, "folders"), "id", spec.route_id + ) + if current_folder is not None: + self._validate_folder( + current_folder, + api_path, + local_device_id, + spec.peer_syncthing_device_id, + ) + break + self._wait(deadline) + + route = route_pb2.LocalRoute( + route_id=spec.route_id, + local_relative_path=spec.local_relative_path, + folder_type=route_pb2.SYNCTHING_FOLDER_TYPE_SEND_RECEIVE, + local_syncthing_device_id=local_device_id, + peer_syncthing_device_ids=[spec.peer_syncthing_device_id], + state=route_pb2.ROUTE_STATE_PROVISIONING, + writable=True, + sparse_supported=self.sparse_supported, + archive_control_created=folder is None, + detail="Syncthing device and folder configuration verified", + ) + route.observed_at.GetCurrentTime() + return ConfiguredRoute( + local_device_id, + local_path, + route, + created_device, + folder is None, + ) + + def verify( + self, + configured: ConfiguredRoute, + spec: route_pb2.EnsureRouteSpec, + local_client_id: str, + nonce: str, + deadline: float, + ) -> tuple[bool, bool]: + local_nonce = configured.local_path / _nonce_name(local_client_id) + _atomic_json( + local_nonce, + { + "route_id": spec.route_id, + "client_id": local_client_id, + "nonce": nonce, + }, + ) + peer_seen_locally = False + local_seen_by_peer = False + while not (peer_seen_locally and local_seen_by_peer): + peer_nonce_path = configured.local_path / _nonce_name( + spec.peer_client_id + ) + peer_nonce = _read_nonce( + peer_nonce_path, spec.route_id, spec.peer_client_id + ) + if peer_nonce is not None: + peer_seen_locally = True + _atomic_json( + configured.local_path + / _ack_name(spec.peer_client_id, local_client_id), + { + "route_id": spec.route_id, + "nonce": peer_nonce, + }, + ) + acknowledgement = _read_ack( + configured.local_path + / _ack_name(local_client_id, spec.peer_client_id), + spec.route_id, + ) + local_seen_by_peer = acknowledgement == nonce + if peer_seen_locally and local_seen_by_peer: + break + try: + self.transport.post( + "/rest/db/scan?" + parse.urlencode({"folder": spec.route_id}) + ) + except RouteSetupError: + pass + self._wait(deadline) + return local_seen_by_peer, peer_seen_locally + + def _route_paths(self, relative_path: str) -> tuple[Path, str]: + relative = PurePosixPath(relative_path) + if ( + not relative_path + or relative.is_absolute() + or any(part in {"", ".", ".."} for part in relative.parts) + ): + raise RoutePathConflict("route path is unsafe") + api_path = (self.config.api_root / relative).as_posix() + try: + local_path = self.config.roots.api_to_local(api_path) + except ConfigError as exc: + raise RoutePathConflict("route path is outside the sync root") from exc + resolved_root = self.config.local_root.resolve(strict=False) + try: + local_path.resolve(strict=False).relative_to(resolved_root) + except ValueError as exc: + raise RoutePathConflict("route path escapes through a symlink") from exc + return local_path, api_path + + @staticmethod + def _prepare_new_directory(path: Path) -> None: + if path.exists() and any(path.iterdir()): + raise RoutePathConflict("new route path contains unrelated data") + path.mkdir(parents=True, exist_ok=True) + + @staticmethod + def _validate_folder( + folder: dict[str, Any], + api_path: str, + local_device_id: str, + peer_device_id: str, + ) -> None: + raw_devices = folder.get("devices") + device_ids = { + item.get("deviceID") + for item in raw_devices + if isinstance(item, dict) and isinstance(item.get("deviceID"), str) + } if isinstance(raw_devices, list) else set() + if folder.get("path") != api_path: + raise RoutePathConflict("existing route ID uses a different path") + if folder.get("type") != "sendreceive": + raise RouteSetupError("existing route folder is not sendreceive") + if device_ids != {local_device_id, peer_device_id}: + raise RouteSetupError("existing route folder has different devices") + if folder.get("paused") is True: + raise RouteSetupError("existing route folder is paused") + + def _wait(self, deadline: float) -> None: + if time.monotonic() >= deadline: + raise RouteSetupTimeout("route setup timed out") + time.sleep(min(self.poll_interval, max(0, deadline - time.monotonic()))) + + +def _object_list(value: dict[str, Any], key: str) -> list[dict[str, Any]]: + items = value.get(key) + if not isinstance(items, list) or any(not isinstance(item, dict) for item in items): + raise RouteSetupError(f"Syncthing configuration {key} is invalid") + return items + + +def _by_key( + items: list[dict[str, Any]], key: str, wanted: str +) -> dict[str, Any] | None: + matches = [item for item in items if item.get(key) == wanted] + if len(matches) > 1: + raise RouteSetupError(f"Syncthing configuration contains duplicate {key}") + return matches[0] if matches else None + + +def _nonce_name(client_id: str) -> str: + return f".archive-control-route-nonce.{client_id}" + + +def _ack_name(nonce_owner: str, observer: str) -> str: + return f".archive-control-route-ack.{nonce_owner}.{observer}" + + +def _atomic_json(path: Path, value: dict[str, Any]) -> None: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":")) + temporary = path.with_name( + f".{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp" + ) + with temporary.open("x", encoding="utf-8") as target: + target.write(encoded) + target.flush() + os.fsync(target.fileno()) + os.replace(temporary, path) + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + + +def _read_nonce(path: Path, route_id: str, client_id: str) -> str | None: + value = _read_json(path) + if value is None: + return None + if value.get("route_id") != route_id or value.get("client_id") != client_id: + raise RouteSetupError("peer route nonce identity is invalid") + nonce = value.get("nonce") + if not isinstance(nonce, str) or not nonce: + raise RouteSetupError("peer route nonce is invalid") + return nonce + + +def _read_ack(path: Path, route_id: str) -> str | None: + value = _read_json(path) + if value is None: + return None + if value.get("route_id") != route_id: + raise RouteSetupError("peer route acknowledgement identity is invalid") + nonce = value.get("nonce") + if not isinstance(nonce, str) or not nonce: + raise RouteSetupError("peer route acknowledgement is invalid") + return nonce + + +def _read_json(path: Path) -> dict[str, Any] | None: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + raise RouteSetupError("route verification file is unreadable") from exc + if not isinstance(value, dict): + raise RouteSetupError("route verification file is invalid") + return value + + +class _NoRedirect(request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None diff --git a/tests/test_daemon.py b/tests/test_daemon.py index 6a07e6f..39c9a23 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -14,13 +14,130 @@ from archive_clients.config import ClientConfig, ConnectionConfig, ServiceConfig from archive_clients.daemon import ArchiveClientDaemon from archive_clients.probes import FilesystemProbe from archive_clients.services import ServiceProbe +from archive_clients.syncthing import ConfiguredRoute from archive_clients.protocol import decode, encode, encode_message, new_envelope from archive_control.v1 import ( - client_pb2, common_pb2, control_pb2, inventory_pb2, job_pb2, + client_pb2, common_pb2, control_pb2, inventory_pb2, job_pb2, route_pb2, ) class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): + async def test_ensure_route_is_durable_and_duplicate_replays_updates(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + token = root / "token" + token.write_text("shared-secret", encoding="utf-8") + os.chmod(token, 0o600) + service = ServiceConfig( + "http://local", PurePosixPath("/sync"), root + ) + config = ClientConfig( + "cache-1", "Cache 1", "cache", "ws://control", token, + root / "state.db", root / "backups", service, service, + ) + local_route = route_pb2.LocalRoute( + route_id="route-1", + local_relative_path="routes/route-1", + folder_type=route_pb2.SYNCTHING_FOLDER_TYPE_SEND_RECEIVE, + local_syncthing_device_id="LOCAL", + peer_syncthing_device_ids=["PEER"], + state=route_pb2.ROUTE_STATE_PROVISIONING, + writable=True, + ) + local_route.observed_at.GetCurrentTime() + manager = Mock() + manager.configure.return_value = ConfiguredRoute( + "LOCAL", root / "routes/route-1", local_route, True, True + ) + manager.verify.return_value = (True, True) + probe = FilesystemProbe(root, True, True, True, True, True) + syncthing_probe = ServiceProbe( + "syncthing", + common_pb2.HEALTH_STATE_HEALTHY, + datetime.now(timezone.utc), + device_id="LOCAL", + ) + daemon = ArchiveClientDaemon( + config, + [probe, probe], + [syncthing_probe], + route_manager=manager, + ) + await asyncio.to_thread(daemon.store.initialize) + command = new_envelope() + command.command.command_id = str(uuid4()) + command.command.created_at.CopyFrom(command.sent_at) + spec = command.command.ensure_route.route + spec.route_id = "route-1" + spec.peer_client_id = "archive-1" + spec.peer_syncthing_device_id = "PEER" + spec.peer_addresses.append("dynamic") + spec.local_relative_path = "routes/route-1" + spec.setup_timeout_seconds = 1800 + outbound = asyncio.Queue() + tasks = set() + await daemon._handle(command, outbound, tasks) + self.assertEqual( + decode(await outbound.get()).command_ack.status, + control_pb2.COMMAND_ACK_STATUS_ACCEPTED, + ) + await next(iter(tasks)) + first_updates = [decode(await outbound.get()).route_update for _ in range(3)] + self.assertEqual( + [update.sequence for update in first_updates], [1, 2, 3] + ) + self.assertEqual( + first_updates[-1].verification.state, + route_pb2.ROUTE_STATE_READY, + ) + self.assertTrue(first_updates[-1].route.archive_control_created) + + duplicate = new_envelope() + duplicate.command.CopyFrom(command.command) + tasks = set() + await daemon._handle(duplicate, outbound, tasks) + self.assertEqual( + decode(await outbound.get()).command_ack.status, + control_pb2.COMMAND_ACK_STATUS_DUPLICATE, + ) + await next(iter(tasks)) + replay = [decode(await outbound.get()).route_update for _ in range(3)] + self.assertEqual( + [update.update_id for update in replay], + [update.update_id for update in first_updates], + ) + self.assertEqual(manager.configure.call_count, 1) + + accepted_before_crash = new_envelope() + accepted_before_crash.command.CopyFrom(command.command) + accepted_before_crash.command.command_id = str(uuid4()) + acknowledgement = daemon._initial_acknowledgement( + accepted_before_crash.command, set(), False + ) + await asyncio.to_thread( + daemon.store.accept_command, + accepted_before_crash.command.command_id, + encode_message(accepted_before_crash.command), + encode_message(acknowledgement), + ) + restarted = ArchiveClientDaemon( + config, + [probe, probe], + [syncthing_probe], + route_manager=manager, + ) + resumed_outbound = asyncio.Queue() + resumed_tasks = set() + await restarted._resume_route_commands( + resumed_outbound, resumed_tasks + ) + await next(iter(resumed_tasks)) + resumed = [ + decode(await resumed_outbound.get()).route_update + for _ in range(3) + ] + self.assertEqual([item.sequence for item in resumed], [1, 2, 3]) + async def test_slow_inventory_does_not_block_heartbeat(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) @@ -213,6 +330,10 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase): client_pb2.CLIENT_FEATURE_INVENTORY_CHUNKS, observed["features"], ) + self.assertIn( + client_pb2.CLIENT_FEATURE_ROUTE_PROVISIONING, + observed["features"], + ) 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) diff --git a/tests/test_state.py b/tests/test_state.py index 1e6d468..7bce711 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -51,6 +51,41 @@ class ClientStoreTests(unittest.TestCase): 2, 3, False, ) + def test_route_attempt_nonce_and_updates_survive_duplicate_commands(self): + with tempfile.TemporaryDirectory() as directory: + store = ClientStore(Path(directory) / "state.db") + store.initialize() + command_id = str(uuid4()) + store.accept_command(command_id, '{"route":"one"}', '{"ok":true}') + first = store.begin_route_attempt( + command_id, "route-1", '{"routeId":"route-1"}', "nonce-1" + ) + duplicate = store.begin_route_attempt( + command_id, "route-1", '{"routeId":"route-1"}', "new-nonce" + ) + self.assertEqual(first["nonce"], "nonce-1") + self.assertEqual(duplicate["nonce"], "nonce-1") + store.record_route_update( + command_id, 1, "provisioning", '{"sequence":1}' + ) + store.record_route_ownership(command_id, True, True) + store.record_route_update( + command_id, 1, "provisioning", '{"sequence":1}' + ) + store.record_route_update( + command_id, 2, "ready", '{"sequence":2}' + ) + self.assertEqual(store.get_route_attempt(command_id)["state"], "ready") + self.assertEqual(store.get_route_attempt(command_id)["created_folder"], 1) + self.assertEqual( + [row["sequence"] for row in store.route_update_rows(command_id)], + [1, 2], + ) + with self.assertRaises(CommandConflict): + store.record_route_update( + command_id, 2, "ready", '{"sequence":2,"changed":true}' + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_syncthing.py b/tests/test_syncthing.py new file mode 100644 index 0000000..db15619 --- /dev/null +++ b/tests/test_syncthing.py @@ -0,0 +1,151 @@ +import json +import tempfile +import time +import unittest +from pathlib import Path, PurePosixPath + +from archive_clients.config import ServiceConfig +from archive_clients.syncthing import ( + RoutePathConflict, + RouteSetupTimeout, + SyncthingRouteManager, +) +from archive_control.v1 import route_pb2 + + +class FakeTransport: + def __init__(self): + self.status = {"myID": "LOCAL"} + self.config = {"devices": [], "folders": []} + self.puts = [] + self.posts = [] + + def get_json(self, path): + if path == "/rest/system/status": + return self.status + if path == "/rest/config": + return self.config + raise AssertionError(path) + + def put_json(self, path, payload): + self.puts.append((path, payload)) + if path.startswith("/rest/config/devices/"): + self.config["devices"].append(payload) + elif path.startswith("/rest/config/folders/"): + self.config["folders"].append(payload) + else: + raise AssertionError(path) + + def post(self, path): + self.posts.append(path) + + +class SyncthingRouteManagerTests(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.root = Path(self.temp_dir.name) / "sync" + self.root.mkdir() + self.transport = FakeTransport() + self.manager = SyncthingRouteManager( + ServiceConfig( + "http://syncthing", + PurePosixPath("/sync"), + self.root, + ), + sparse_supported=True, + transport=self.transport, + poll_interval=0, + ) + self.spec = route_pb2.EnsureRouteSpec( + route_id="route-1", + peer_client_id="archive-1", + peer_syncthing_device_id="PEER", + peer_addresses=["tcp://archive:22000"], + local_relative_path="routes/route-1", + setup_timeout_seconds=1800, + ) + + def tearDown(self): + self.temp_dir.cleanup() + + def test_configure_adds_only_peer_and_pairwise_folder(self): + configured = self.manager.configure( + self.spec, time.monotonic() + 1 + ) + self.assertEqual(configured.local_device_id, "LOCAL") + self.assertTrue(configured.local_path.is_dir()) + self.assertEqual(len(self.transport.puts), 2) + folder = self.transport.config["folders"][0] + self.assertEqual(folder["path"], "/sync/routes/route-1") + self.assertEqual(folder["type"], "sendreceive") + self.assertEqual( + {item["deviceID"] for item in folder["devices"]}, + {"LOCAL", "PEER"}, + ) + self.assertTrue(configured.local_route.archive_control_created) + + repeated = self.manager.configure(self.spec, time.monotonic() + 1) + self.assertEqual(len(self.transport.puts), 2) + self.assertFalse(repeated.local_route.archive_control_created) + + def test_existing_folder_conflicts_are_never_overwritten(self): + self.transport.config["folders"].append( + { + "id": "route-1", + "path": "/somewhere-else", + "type": "sendreceive", + "devices": [{"deviceID": "LOCAL"}, {"deviceID": "PEER"}], + } + ) + with self.assertRaisesRegex(RoutePathConflict, "different path"): + self.manager.configure(self.spec, time.monotonic() + 1) + self.assertEqual(self.transport.puts, []) + + def test_bidirectional_nonce_and_ack_are_required(self): + configured = self.manager.configure(self.spec, time.monotonic() + 1) + peer_nonce = configured.local_path / ( + ".archive-control-route-nonce.archive-1" + ) + peer_nonce.write_text( + json.dumps({ + "route_id": "route-1", + "client_id": "archive-1", + "nonce": "peer-nonce", + }), + encoding="utf-8", + ) + acknowledgement = configured.local_path / ( + ".archive-control-route-ack.cache-1.archive-1" + ) + acknowledgement.write_text( + json.dumps({"route_id": "route-1", "nonce": "local-nonce"}), + encoding="utf-8", + ) + verified = self.manager.verify( + configured, + self.spec, + "cache-1", + "local-nonce", + time.monotonic() + 1, + ) + self.assertEqual(verified, (True, True)) + peer_ack = configured.local_path / ( + ".archive-control-route-ack.archive-1.cache-1" + ) + self.assertEqual(json.loads(peer_ack.read_text())["nonce"], "peer-nonce") + + def test_verification_times_out_without_peer_evidence(self): + configured = self.manager.configure(self.spec, time.monotonic() + 1) + with self.assertRaises(RouteSetupTimeout): + self.manager.verify( + configured, + self.spec, + "cache-1", + "local-nonce", + time.monotonic() + 0.01, + ) + self.assertTrue(self.transport.posts) + + +if __name__ == "__main__": + unittest.main()