diff --git a/README.md b/README.md index 17b0c64..d142149 100644 --- a/README.md +++ b/README.md @@ -19,9 +19,12 @@ derives canonical resource identities and selection ranges, builds lazy content trees, and reports only safe Syncthing folders below the configured root. Renamed/noncanonical torrent paths remain visible but are marked noncanonical so later job preflight can reject them without hiding the resource. -The qBittorrent read adapter uses cookie authentication, bounded responses, -one reauthentication attempt on session expiry, hash-scoped file/metainfo -fetches, and never logs credentials, cookies, response bodies, or endpoints. +The qBittorrent adapter uses cookie authentication, bounded responses, one +reauthentication attempt on session expiry, and hash-scoped file/metainfo +fetches. Its mutation boundary adds targets stopped, applies only selected vs +skipped state, supports qBittorrent 4 pause and qBittorrent 5 stop semantics, +performs guarded full rechecks, and removes entries without requesting data +deletion. It never logs credentials, cookies, response bodies, or endpoints. 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 @@ -39,6 +42,9 @@ then sparse-aware copy fallback; every intent/result is journaled in SQLite. Receiver materialization never overwrites a path, can reuse a same-size regular file for later qBittorrent verification, and replays completed operations idempotently after a database reopen. +The Syncthing transfer observer requests scoped rescans and combines verified +local manifest payloads, folder completion, and outstanding job-prefix needs; +an advisory percentage alone is never treated as transfer completion. ```bash archive-client --config /etc/archive-control/client.toml --check-config diff --git a/src/archive_clients/qbittorrent.py b/src/archive_clients/qbittorrent.py index 0b0ef50..2c1f831 100644 --- a/src/archive_clients/qbittorrent.py +++ b/src/archive_clients/qbittorrent.py @@ -4,7 +4,11 @@ from __future__ import annotations import json import threading +import time +import uuid +from dataclasses import dataclass from http.cookiejar import CookieJar +from pathlib import PurePosixPath from typing import Any from urllib import error, parse, request @@ -16,6 +20,25 @@ class QBittorrentError(RuntimeError): pass +class QBittorrentDownloadAttempt(QBittorrentError): + pass + + +class QBittorrentHttpError(QBittorrentError): + def __init__(self, status: int): + super().__init__(f"qBittorrent HTTP request failed with status {status}") + self.status = status + + +@dataclass(frozen=True) +class RecheckResult: + torrent_hash: str + selected_file_indices: tuple[int, ...] + downloaded_bytes_before: int + downloaded_bytes_after: int + final_state: str + + class _NoRedirect(request.HTTPRedirectHandler): def redirect_request(self, *_: Any, **__: Any) -> None: return None @@ -68,6 +91,174 @@ class QBittorrentReader: raise QBittorrentError("qBittorrent lookup is ambiguous") return self._normalize(exact[0], torrent_hash) + def add_stopped( + self, + metainfo: bytes, + save_path: str, + ) -> None: + if not metainfo: + raise QBittorrentError("torrent metainfo is empty") + candidate = PurePosixPath(save_path) + try: + candidate.relative_to(self.config.api_root) + except ValueError as exc: + raise QBittorrentError( + "qBittorrent save path is outside its configured root" + ) from exc + if ( + not candidate.is_absolute() + or ".." in candidate.parts + or "." in candidate.parts + ): + raise QBittorrentError("qBittorrent save path is unsafe") + response = self._multipart( + "/api/v2/torrents/add", + { + "savepath": save_path, + "autoTMM": "false", + # qBittorrent 4.x calls this paused and 5.x calls it stopped. + "paused": "true", + "stopped": "true", + }, + "torrents", + "source.torrent", + metainfo, + ) + _require_mutation_success(response, "add torrent") + + def set_selection( + self, + torrent_hash: str, + selected_file_indices: list[int] | tuple[int, ...], + total_file_count: int, + ) -> None: + selected = sorted(set(selected_file_indices)) + if total_file_count < 1: + raise QBittorrentError("torrent file count must be positive") + if not selected or selected[0] < 0 or selected[-1] >= total_file_count: + raise QBittorrentError("selected torrent file index is invalid") + self._post_form( + "/api/v2/torrents/filePrio", + { + "hash": torrent_hash, + "id": f"0-{total_file_count - 1}", + "priority": "0", + }, + ) + self._post_form( + "/api/v2/torrents/filePrio", + { + "hash": torrent_hash, + "id": "|".join(str(index) for index in selected), + "priority": "1", + }, + ) + + def stop(self, torrent_hash: str) -> None: + try: + self._post_form( + "/api/v2/torrents/stop", {"hashes": torrent_hash} + ) + except QBittorrentHttpError as exc: + if exc.status != 404: + raise + self._post_form( + "/api/v2/torrents/pause", {"hashes": torrent_hash} + ) + + def delete_entry(self, torrent_hash: str) -> None: + self._post_form( + "/api/v2/torrents/delete", + {"hashes": torrent_hash, "deleteFiles": "false"}, + ) + + def recheck_and_wait( + self, + torrent_hash: str, + selected_file_indices: list[int] | tuple[int, ...], + *, + timeout: float, + poll_interval: float = 1, + ) -> RecheckResult: + selected = tuple(sorted(set(selected_file_indices))) + if not selected: + raise QBittorrentError("recheck selection cannot be empty") + if timeout <= 0 or poll_interval < 0: + raise QBittorrentError("recheck timing values are invalid") + self.stop(torrent_hash) + baseline = self._downloaded_bytes(torrent_hash) + self._post_form( + "/api/v2/torrents/recheck", {"hashes": torrent_hash} + ) + deadline = time.monotonic() + timeout + while True: + record = self._torrent_record(torrent_hash) + state = record.get("state") + if not isinstance(state, str): + raise QBittorrentError("qBittorrent torrent state is invalid") + downloaded = self._downloaded_bytes(torrent_hash) + if downloaded > baseline or _is_download_state(state): + try: + self.stop(torrent_hash) + finally: + raise QBittorrentDownloadAttempt( + "qBittorrent attempted to download during verification" + ) + files = self._json( + "/api/v2/torrents/files", {"hash": torrent_hash} + ) + if not isinstance(files, list) or not all( + isinstance(item, dict) for item in files + ): + raise QBittorrentError("qBittorrent file list is invalid") + progress = { + item.get("index"): item.get("progress") + for item in files + } + if all( + isinstance(progress.get(index), (int, float)) + and float(progress[index]) >= 1 + for index in selected + ) and state not in {"checkingDL", "checkingUP", "checkingResumeData"}: + return RecheckResult( + torrent_hash, + selected, + baseline, + downloaded, + state, + ) + if time.monotonic() >= deadline: + raise QBittorrentError("qBittorrent recheck timed out") + time.sleep(min(poll_interval, max(0, deadline - time.monotonic()))) + + def _torrent_record(self, torrent_hash: str) -> dict[str, Any]: + torrents = self._json( + "/api/v2/torrents/info", {"hashes": torrent_hash} + ) + if not isinstance(torrents, list): + raise QBittorrentError("qBittorrent lookup response is invalid") + exact = [ + item for item in torrents + if isinstance(item, dict) + and str(item.get("hash", "")).lower() == torrent_hash.lower() + ] + if len(exact) != 1: + raise QBittorrentError("qBittorrent target torrent is missing") + return exact[0] + + def _downloaded_bytes(self, torrent_hash: str) -> int: + properties = self._json( + "/api/v2/torrents/properties", {"hash": torrent_hash} + ) + if not isinstance(properties, dict): + raise QBittorrentError("qBittorrent properties response is invalid") + value = properties.get("total_downloaded") + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise QBittorrentError( + "qBittorrent downloaded byte counter is invalid" + ) + return value + def _normalize( self, torrent: dict[str, Any], torrent_hash: str ) -> NormalizedResource: @@ -115,6 +306,81 @@ class QBittorrentReader: "qBittorrent request failed after reauthentication" ) from retry_exc + def _post_form(self, path: str, parameters: dict[str, str]) -> None: + body = parse.urlencode(parameters).encode("utf-8") + call = request.Request( + self._url(path), + data=body, + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + response = self._authorized_request(call, maximum=1024 * 1024) + _require_mutation_success(response, "mutate torrent") + + def _multipart( + self, + path: str, + fields: dict[str, str], + file_field: str, + filename: str, + content: bytes, + ) -> bytes: + boundary = f"archive-control-{uuid.uuid4().hex}" + chunks: list[bytes] = [] + for name, value in fields.items(): + chunks.extend(( + f"--{boundary}\r\n".encode(), + ( + f'Content-Disposition: form-data; name="{name}"\r\n\r\n' + ).encode(), + value.encode(), + b"\r\n", + )) + chunks.extend(( + f"--{boundary}\r\n".encode(), + ( + "Content-Disposition: form-data; " + f'name="{file_field}"; filename="{filename}"\r\n' + ).encode(), + b"Content-Type: application/x-bittorrent\r\n\r\n", + content, + b"\r\n", + f"--{boundary}--\r\n".encode(), + )) + call = request.Request( + self._url(path), + data=b"".join(chunks), + method="POST", + headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, + ) + return self._authorized_request(call, maximum=1024 * 1024) + + def _authorized_request(self, call: request.Request, maximum: int) -> bytes: + with self._lock: + if not self._authenticated: + self._login() + try: + return self._read( + self._opener.open(call, timeout=self.timeout), maximum + ) + except error.HTTPError as exc: + if exc.code not in {401, 403}: + raise QBittorrentHttpError(exc.code) from exc + self._authenticated = False + self._login() + try: + return self._read( + self._opener.open(call, timeout=self.timeout), maximum + ) + except (error.HTTPError, error.URLError, OSError) as retry_exc: + raise QBittorrentError( + "qBittorrent request failed after reauthentication" + ) from retry_exc + except (error.URLError, OSError) as exc: + raise QBittorrentError( + "qBittorrent service is unavailable" + ) from exc + def _login(self) -> None: body = parse.urlencode({ "username": self.config.username, @@ -161,3 +427,15 @@ class QBittorrentReader: if len(data) > maximum: raise QBittorrentError("qBittorrent response exceeds limit") return data + + +def _is_download_state(state: str) -> bool: + return state not in {"checkingDL"} and ( + state.endswith("DL") + or state in {"downloading", "metaDL", "forcedMetaDL"} + ) + + +def _require_mutation_success(response: bytes, operation: str) -> None: + if response.strip() not in {b"", b"Ok."}: + raise QBittorrentError(f"qBittorrent refused to {operation}") diff --git a/src/archive_clients/syncthing.py b/src/archive_clients/syncthing.py index 90014cf..a749b30 100644 --- a/src/archive_clients/syncthing.py +++ b/src/archive_clients/syncthing.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import os +import stat import time import uuid from dataclasses import dataclass @@ -12,6 +13,11 @@ from typing import Any, Protocol from urllib import error, parse, request from archive_clients.config import ConfigError, ServiceConfig +from archive_clients.transfer import ( + TransferIntegrityError, + UnsafeTransferPath, + load_published_transfer, +) from archive_control.v1 import route_pb2 @@ -91,6 +97,98 @@ class ConfiguredRoute: created_folder: bool +@dataclass(frozen=True) +class SyncthingTransferStatus: + fraction_complete: float + bytes_complete: int + bytes_total: int + complete: bool + needed_items: int + + +class SyncthingTransferObserver: + """Observe one exact Archive Control job prefix within a route folder.""" + + def __init__( + self, + transport: SyncthingTransport, + folder_id: str, + job_relative_path: str, + local_job_directory: Path, + ): + relative = PurePosixPath(job_relative_path) + if ( + not job_relative_path + or relative.is_absolute() + or any(part in {"", ".", ".."} for part in relative.parts) + ): + raise UnsafeTransferPath("Syncthing job prefix is unsafe") + self.transport = transport + self.folder_id = folder_id + self.job_relative_path = relative.as_posix() + self.local_job_directory = local_job_directory + + def rescan(self) -> None: + self.transport.post( + "/rest/db/scan?" + + parse.urlencode({ + "folder": self.folder_id, + "sub": self.job_relative_path, + }) + ) + + def status(self) -> SyncthingTransferStatus: + published = load_published_transfer(self.local_job_directory) + total = 0 + for entry in published.manifest.files: + total += _verified_job_file_size( + self.local_job_directory, + entry.payload_relative_path, + entry.logical_bytes, + ) + for artifact in published.manifest.artifacts: + total += _verified_job_file_size( + self.local_job_directory, + artifact.payload_relative_path, + artifact.logical_bytes, + ) + + completion = self.transport.get_json( + "/rest/db/completion?" + + parse.urlencode({"folder": self.folder_id}) + ) + raw_fraction = completion.get("completion") + if ( + isinstance(raw_fraction, bool) + or not isinstance(raw_fraction, (int, float)) + or not 0 <= float(raw_fraction) <= 100 + ): + raise RouteSetupError("Syncthing completion response is invalid") + need = self.transport.get_json( + "/rest/db/need?" + + parse.urlencode({ + "folder": self.folder_id, + "page": 1, + "perpage": 1000, + }) + ) + needed_names = _needed_names(need) + prefix = self.job_relative_path.rstrip("/") + "/" + relevant = { + name for name in needed_names + if name == self.job_relative_path or name.startswith(prefix) + } + fraction = float(raw_fraction) / 100 + complete = fraction == 1 and not relevant + return SyncthingTransferStatus( + fraction, + total if complete else int(total * fraction), + total, + complete, + len(relevant), + ) + + class SyncthingRouteManager: def __init__( self, @@ -315,6 +413,47 @@ def _by_key( return matches[0] if matches else None +def _verified_job_file_size( + job_directory: Path, + relative_path: str, + expected_bytes: int, +) -> int: + relative = PurePosixPath(relative_path) + if ( + not relative_path + or relative.is_absolute() + or any(part in {"", ".", ".."} for part in relative.parts) + ): + raise TransferIntegrityError("manifest payload path is unsafe") + current = job_directory + for component in relative.parts[:-1]: + current = current / component + metadata = current.lstat() + if not stat.S_ISDIR(metadata.st_mode): + raise TransferIntegrityError( + "manifest payload parent is not a directory" + ) + metadata = (current / relative.name).lstat() + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size != expected_bytes: + raise TransferIntegrityError( + "manifest payload does not match its declared size" + ) + return metadata.st_size + + +def _needed_names(value: dict[str, Any]) -> set[str]: + result: set[str] = set() + for key in ("progress", "queued", "rest"): + items = value.get(key, []) + if not isinstance(items, list): + raise RouteSetupError("Syncthing need response is invalid") + for item in items: + if not isinstance(item, dict) or not isinstance(item.get("name"), str): + raise RouteSetupError("Syncthing need item is invalid") + result.add(item["name"]) + return result + + def _nonce_name(client_id: str) -> str: return f".archive-control-route-nonce.{client_id}" diff --git a/tests/test_qbittorrent.py b/tests/test_qbittorrent.py index c25e48d..f963d85 100644 --- a/tests/test_qbittorrent.py +++ b/tests/test_qbittorrent.py @@ -6,6 +6,7 @@ import tempfile import unittest from pathlib import Path, PurePosixPath from unittest.mock import patch +from urllib import error from archive_clients.bencode import encode from archive_clients.config import ServiceConfig @@ -35,6 +36,8 @@ class _Opener: def open(self, call, timeout): self.calls.append(call) response = next(self.values) + if isinstance(response, BaseException): + raise response if isinstance(response, tuple): return _Response(*response) return _Response(response) @@ -95,6 +98,90 @@ class QBittorrentReaderTests(unittest.TestCase): self.assertEqual(resources, []) self.assertEqual(len(opener.calls), 2) + def test_stopped_add_selection_recheck_and_entry_only_delete(self): + torrent_hash = "a" * 40 + responses = [ + b"Ok.", # login + b"Ok.", # multipart add + b"", # skip all files + b"", # select requested files + b"", # stop before recheck + b'{"total_downloaded":0}', + b"", # recheck + json.dumps([{ + "hash": torrent_hash, "state": "checkingUP", + }]).encode(), + b'{"total_downloaded":0}', + b'[{"index":0,"progress":1},{"index":1,"progress":0}]', + json.dumps([{ + "hash": torrent_hash, "state": "stoppedUP", + }]).encode(), + b'{"total_downloaded":0}', + b'[{"index":0,"progress":1},{"index":1,"progress":0}]', + b"", # entry-only delete + ] + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + password = root / "password" + password.write_text("secret", encoding="utf-8") + os.chmod(password, 0o600) + config = ServiceConfig( + "http://qb", PurePosixPath("/downloads"), root, + username="admin", password_file=password, + ) + opener = _Opener(responses) + with patch( + "archive_clients.qbittorrent.request.build_opener", + return_value=opener, + ): + adapter = QBittorrentReader(config) + adapter.add_stopped(b"torrent", "/downloads/archive") + adapter.set_selection(torrent_hash, [0], 2) + result = adapter.recheck_and_wait( + torrent_hash, [0], timeout=1, poll_interval=0 + ) + adapter.delete_entry(torrent_hash) + + self.assertEqual(result.final_state, "stoppedUP") + self.assertEqual(result.selected_file_indices, (0,)) + calls = [ + call for call in opener.calls + if not isinstance(call, str) + ] + add = calls[1] + self.assertIn(b'name="stopped"\r\n\r\ntrue', add.data) + form_bodies = [ + call.data.decode() + for call in calls[2:] + if getattr(call, "data", None) + ] + self.assertIn("priority=0", form_bodies[0]) + self.assertIn("priority=1", form_bodies[1]) + self.assertIn("deleteFiles=false", form_bodies[-1]) + + def test_stop_falls_back_to_qbittorrent_4_pause_endpoint(self): + missing = error.HTTPError( + "http://qb/api/v2/torrents/stop", 404, "not found", {}, None + ) + responses = [b"Ok.", missing, b""] + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + password = root / "password" + password.write_text("secret", encoding="utf-8") + os.chmod(password, 0o600) + config = ServiceConfig( + "http://qb", PurePosixPath("/downloads"), root, + username="admin", password_file=password, + ) + opener = _Opener(responses) + with patch( + "archive_clients.qbittorrent.request.build_opener", + return_value=opener, + ): + QBittorrentReader(config).stop("a" * 40) + + self.assertTrue(opener.calls[-1].full_url.endswith("/torrents/pause")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_syncthing.py b/tests/test_syncthing.py index e569ac7..69bf1c7 100644 --- a/tests/test_syncthing.py +++ b/tests/test_syncthing.py @@ -3,14 +3,18 @@ import tempfile import time import unittest from pathlib import Path, PurePosixPath +from uuid import uuid4 from archive_clients.config import ServiceConfig +from archive_clients.state import ClientStore from archive_clients.syncthing import ( RoutePathConflict, RouteSetupTimeout, SyncthingRouteManager, + SyncthingTransferObserver, ) -from archive_control.v1 import route_pb2 +from archive_clients.transfer import stage_transfer +from archive_control.v1 import route_pb2, transfer_pb2 class FakeTransport: @@ -40,6 +44,26 @@ class FakeTransport: self.posts.append(path) +class FakeTransferTransport: + def __init__(self, completion, need): + self.completion = completion + self.need = need + self.posts = [] + + def get_json(self, path): + if path.startswith("/rest/db/completion?"): + return self.completion + if path.startswith("/rest/db/need?"): + return self.need + raise AssertionError(path) + + def put_json(self, path, payload): + raise AssertionError((path, payload)) + + def post(self, path): + self.posts.append(path) + + class SyncthingRouteManagerTests(unittest.TestCase): def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() @@ -164,6 +188,78 @@ class SyncthingRouteManagerTests(unittest.TestCase): ) self.assertTrue(self.transport.posts) + def test_job_prefix_rescan_progress_and_completion(self): + source = Path(self.temp_dir.name) / "source" + source.mkdir() + (source / "payload.bin").write_bytes(b"x" * 8192) + metainfo = Path(self.temp_dir.name) / "source.torrent" + metainfo.write_bytes(b"torrent") + store = ClientStore(Path(self.temp_dir.name) / "state.db") + store.initialize() + manifest = transfer_pb2.TransferManifest( + manifest_version=1, + job_id=str(uuid4()), + source_client_id="cache-1", + target_client_id="archive-1", + route_id="route-1", + ) + manifest.resource_id.info_hash_v1_hex = "a" * 40 + manifest.created_at.seconds = 1_700_000_000 + manifest.files.add( + file_index=0, + payload_relative_path="payload/payload.bin", + target_canonical_path="payload.bin", + logical_bytes=8192, + ) + manifest.artifacts.add( + kind=transfer_pb2.ARTIFACT_KIND_TORRENT_FILE, + payload_relative_path="metainfo/source.torrent", + logical_bytes=metainfo.stat().st_size, + ) + published = stage_transfer( + manifest, + source_root=source, + sync_root=self.root, + store=store, + artifact_sources={"metainfo/source.torrent": metainfo}, + ) + relative = ( + f".archive-control/jobs/{manifest.job_id}" + ) + in_progress_transport = FakeTransferTransport( + {"completion": 50}, + { + "progress": [{"name": f"{relative}/payload/payload.bin"}], + "queued": [], + "rest": [], + }, + ) + observer = SyncthingTransferObserver( + in_progress_transport, + "route-1", + relative, + published.job_directory, + ) + observer.rescan() + in_progress = observer.status() + self.assertEqual(in_progress.fraction_complete, 0.5) + self.assertFalse(in_progress.complete) + self.assertEqual(in_progress.needed_items, 1) + self.assertIn("folder=route-1", in_progress_transport.posts[0]) + self.assertIn("sub=.archive-control%2Fjobs%2F", in_progress_transport.posts[0]) + + complete = SyncthingTransferObserver( + FakeTransferTransport( + {"completion": 100}, + {"progress": [], "queued": [], "rest": []}, + ), + "route-1", + relative, + published.job_directory, + ).status() + self.assertTrue(complete.complete) + self.assertEqual(complete.bytes_complete, complete.bytes_total) + if __name__ == "__main__": unittest.main()