Coverage for src/ph/bittorrent.py: 96.0%
582 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Small standard-library BitTorrent v1 client for selective Minerva downloads."""
3import hashlib
4import ipaddress
5import logging
6import re
7import secrets
8import socket
9import ssl
10import struct
11import time
12import unicodedata
13from collections.abc import Callable, Iterable, Iterator
14from concurrent.futures import ThreadPoolExecutor, as_completed
15from contextlib import contextmanager
16from dataclasses import dataclass
17from difflib import SequenceMatcher
18from math import isfinite
19from pathlib import Path
20from threading import Event, Lock
21from urllib.error import HTTPError, URLError
22from urllib.parse import urlencode, urlparse
23from urllib.request import Request, urlopen
25from ph.store import USER_AGENT
27LOGGER = logging.getLogger(__name__)
29type BValue = bytes | int | list[BValue] | dict[bytes, BValue]
30type PeerAddress = tuple[str, int]
31type TorrentProgress = Callable[[int, int], None]
32type CancelCallback = Callable[[], bool]
34_PROTOCOL = b"BitTorrent protocol"
35_UDP_PROTOCOL_ID = 0x41727101980
36_BLOCK_SIZE = 16 * 1024
37_MAX_TORRENT_BYTES = 16 * 1024 * 1024
38_MAX_TRACKER_BYTES = 2 * 1024 * 1024
39_MAX_PEER_ATTEMPTS = 240
40_PEER_RACE_WORKERS = 8
41_MAX_PEER_TIMEOUT_SECONDS = 8.0
42_MAX_TRACKER_QUERIES = 16
43_MAX_DISCOVERED_PEERS = 240
46@dataclass(slots=True)
47class _PeerGate:
48 lock: Lock
49 users: int = 0
52_PEER_GATES: dict[PeerAddress, _PeerGate] = {}
53_PEER_GATES_LOCK = Lock()
56class BitTorrentError(RuntimeError):
57 """Torrent metadata, tracker discovery, or peer transfer failed."""
60class BitTorrentCancelled(BitTorrentError):
61 """The caller requested cancellation of a torrent transfer."""
64@dataclass(frozen=True, slots=True)
65class TorrentFileChoice:
66 """One torrent entry that a user can explicitly approve."""
68 index: int
69 path: tuple[str, ...]
70 length: int
71 match_score: float
73 @property
74 def filename(self) -> str:
75 return self.path[-1]
78class TorrentSelectionRequired(BitTorrentError):
79 """Minerva changed its torrent and needs an explicit safe file choice."""
81 def __init__(
82 self,
83 torrent_url: str,
84 expected_filename: str,
85 catalogue_index: int,
86 candidates: tuple[TorrentFileChoice, ...],
87 total_files: int,
88 ) -> None:
89 self.torrent_url = torrent_url
90 self.expected_filename = expected_filename
91 self.catalogue_index = catalogue_index
92 self.candidates = candidates
93 self.total_files = total_files
94 names = ", ".join(choice.filename for choice in candidates[:3])
95 super().__init__(
96 "Minerva's torrent no longer has one unambiguous match for "
97 f"{expected_filename!r} at catalogue position {catalogue_index}. "
98 f"Closest candidates: {names}. Use the TUI to choose a file or cancel."
99 )
102@dataclass(frozen=True, slots=True)
103class BitTorrentSettings:
104 """User-configurable limits for Minerva's native BitTorrent downloads."""
106 udp_protocol_id: int = _UDP_PROTOCOL_ID
107 block_size: int = _BLOCK_SIZE
108 max_torrent_bytes: int = _MAX_TORRENT_BYTES
109 max_tracker_bytes: int = _MAX_TRACKER_BYTES
110 max_peer_attempts: int = _MAX_PEER_ATTEMPTS
111 peer_race_workers: int = _PEER_RACE_WORKERS
112 max_peer_timeout_seconds: float = _MAX_PEER_TIMEOUT_SECONDS
113 max_tracker_queries: int = _MAX_TRACKER_QUERIES
114 max_discovered_peers: int = _MAX_DISCOVERED_PEERS
116 def __post_init__(self) -> None:
117 if not 0 < self.udp_protocol_id <= 0xFFFFFFFFFFFFFFFF:
118 raise ValueError("UDP protocol ID must be a positive 64-bit integer.")
119 positive_values = {
120 "Block size": self.block_size,
121 "Maximum torrent bytes": self.max_torrent_bytes,
122 "Maximum tracker bytes": self.max_tracker_bytes,
123 "Maximum peer attempts": self.max_peer_attempts,
124 "Peer race workers": self.peer_race_workers,
125 "Maximum tracker queries": self.max_tracker_queries,
126 "Maximum discovered peers": self.max_discovered_peers,
127 }
128 invalid = next((label for label, value in positive_values.items() if value <= 0), None)
129 if invalid is not None:
130 raise ValueError(f"{invalid} must be greater than zero.")
131 if not isfinite(self.max_peer_timeout_seconds) or self.max_peer_timeout_seconds <= 0:
132 raise ValueError("Maximum peer timeout must be a positive finite number.")
135class _BDecoder:
136 def __init__(self, data: bytes) -> None:
137 self.data = data
138 self.position = 0
140 def decode(self) -> BValue:
141 if self.position >= len(self.data):
142 raise BitTorrentError("Unexpected end of bencoded data.")
143 marker = self.data[self.position]
144 if marker == ord("i"):
145 return self._integer()
146 if marker == ord("l"):
147 return self._list()
148 if marker == ord("d"):
149 return self._dictionary()
150 if ord("0") <= marker <= ord("9"):
151 return self._bytes()
152 raise BitTorrentError("Invalid bencode marker.")
154 def _integer(self) -> int:
155 self.position += 1
156 end = self.data.find(b"e", self.position)
157 if end < 0:
158 raise BitTorrentError("Unterminated bencode integer.")
159 raw = self.data[self.position : end]
160 self.position = end + 1
161 try:
162 return int(raw)
163 except ValueError as error:
164 raise BitTorrentError("Invalid bencode integer.") from error
166 def _bytes(self) -> bytes:
167 separator = self.data.find(b":", self.position)
168 if separator < 0:
169 raise BitTorrentError("Invalid bencode byte string.")
170 try:
171 length = int(self.data[self.position : separator])
172 except ValueError as error:
173 raise BitTorrentError("Invalid bencode byte-string length.") from error
174 self.position = separator + 1
175 end = self.position + length
176 if length < 0 or end > len(self.data):
177 raise BitTorrentError("Truncated bencode byte string.")
178 value = self.data[self.position : end]
179 self.position = end
180 return value
182 def _list(self) -> list[BValue]:
183 self.position += 1
184 values: list[BValue] = []
185 while self.position < len(self.data) and self.data[self.position] != ord("e"):
186 values.append(self.decode())
187 if self.position >= len(self.data):
188 raise BitTorrentError("Unterminated bencode list.")
189 self.position += 1
190 return values
192 def _dictionary(self) -> dict[bytes, BValue]:
193 self.position += 1
194 values: dict[bytes, BValue] = {}
195 while self.position < len(self.data) and self.data[self.position] != ord("e"):
196 key = self._bytes()
197 values[key] = self.decode()
198 if self.position >= len(self.data):
199 raise BitTorrentError("Unterminated bencode dictionary.")
200 self.position += 1
201 return values
204def bdecode(data: bytes) -> BValue:
205 """Decode one complete bencoded value."""
207 decoder = _BDecoder(data)
208 value = decoder.decode()
209 if decoder.position != len(data):
210 raise BitTorrentError("Trailing data after bencoded value.")
211 return value
214def bencode(value: BValue) -> bytes:
215 """Encode a value canonically so an info dictionary keeps its v1 hash."""
217 if isinstance(value, bytes):
218 return str(len(value)).encode() + b":" + value
219 if isinstance(value, int):
220 return b"i" + str(value).encode() + b"e"
221 if isinstance(value, list):
222 return b"l" + b"".join(bencode(item) for item in value) + b"e"
223 if isinstance(value, dict):
224 return (
225 b"d"
226 + b"".join(bencode(key) + bencode(item) for key, item in sorted(value.items()))
227 + b"e"
228 )
229 raise BitTorrentError("Unsupported bencode value.")
232@dataclass(frozen=True, slots=True)
233class TorrentFile:
234 path: tuple[str, ...]
235 length: int
236 offset: int
239@dataclass(frozen=True, slots=True)
240class TorrentMetadata:
241 info_hash: bytes
242 piece_length: int
243 piece_hashes: tuple[bytes, ...]
244 files: tuple[TorrentFile, ...]
245 trackers: tuple[str, ...]
246 total_length: int
249def parse_torrent(data: bytes) -> TorrentMetadata:
250 """Parse and validate the v1 fields needed for a selective file download."""
252 root = _as_dictionary(bdecode(data), "torrent root")
253 info = _as_dictionary(root.get(b"info"), "torrent info")
254 piece_length = _as_positive_integer(info.get(b"piece length"), "piece length")
255 pieces = _as_bytes(info.get(b"pieces"), "piece hashes")
256 if not pieces or len(pieces) % 20:
257 raise BitTorrentError("Torrent piece hashes are malformed.")
259 files: list[TorrentFile] = []
260 offset = 0
261 raw_files = info.get(b"files")
262 if isinstance(raw_files, list):
263 for raw_file in raw_files:
264 file_info = _as_dictionary(raw_file, "torrent file")
265 length = _as_nonnegative_integer(file_info.get(b"length"), "file length")
266 raw_path = _as_list(file_info.get(b"path"), "file path")
267 path = tuple(_decode_path_part(part) for part in raw_path)
268 if not path or any(part in ("", ".", "..") for part in path):
269 raise BitTorrentError("Torrent contains an unsafe file path.")
270 files.append(TorrentFile(path, length, offset))
271 offset += length
272 else:
273 length = _as_nonnegative_integer(info.get(b"length"), "file length")
274 name = _decode_path_part(info.get(b"name"))
275 files.append(TorrentFile((name,), length, 0))
276 offset = length
277 if not files or offset <= 0:
278 raise BitTorrentError("Torrent does not contain downloadable files.")
280 expected_piece_count = (offset + piece_length - 1) // piece_length
281 if len(pieces) // 20 != expected_piece_count:
282 raise BitTorrentError("Torrent piece count does not match its files.")
284 trackers: list[str] = []
285 announce_list = root.get(b"announce-list")
286 if isinstance(announce_list, list):
287 for tier in announce_list:
288 values = tier if isinstance(tier, list) else [tier]
289 for value in values:
290 if isinstance(value, bytes): 290 ↛ 289line 290 didn't jump to line 289 because the condition on line 290 was always true
291 trackers.append(value.decode("utf-8", errors="replace"))
292 announce = root.get(b"announce")
293 if isinstance(announce, bytes):
294 trackers.append(announce.decode("utf-8", errors="replace"))
295 trackers = list(
296 dict.fromkeys(url for url in trackers if urlparse(url).scheme in ("http", "https", "udp"))
297 )
298 if not trackers:
299 raise BitTorrentError("Torrent does not advertise a supported tracker.")
301 return TorrentMetadata(
302 info_hash=hashlib.sha1(bencode(info), usedforsecurity=False).digest(),
303 piece_length=piece_length,
304 piece_hashes=tuple(pieces[index : index + 20] for index in range(0, len(pieces), 20)),
305 files=tuple(files),
306 trackers=tuple(trackers),
307 total_length=offset,
308 )
311def download_torrent_file(
312 torrent_url: str,
313 file_index: int,
314 expected_filename: str,
315 destination: Path,
316 referer: str,
317 timeout_seconds: float,
318 progress: TorrentProgress | None = None,
319 cancelled: CancelCallback | None = None,
320 settings: BitTorrentSettings | None = None,
321 selected_path: tuple[str, ...] | None = None,
322 *,
323 resume: bool = False,
324) -> None:
325 """Download one expected file, treating its catalogue position only as a hint."""
327 effective_settings = settings or BitTorrentSettings()
328 _raise_if_cancelled(cancelled)
329 torrent_data = _read_url(
330 torrent_url,
331 referer,
332 timeout_seconds,
333 effective_settings.max_torrent_bytes,
334 )
335 _raise_if_cancelled(cancelled)
336 metadata = parse_torrent(torrent_data)
337 selected = _select_torrent_file(
338 metadata,
339 file_index,
340 expected_filename,
341 torrent_url=torrent_url,
342 selected_path=selected_path,
343 )
344 if selected.length == 0:
345 destination.write_bytes(b"")
346 return
347 if progress is not None:
348 progress(0, selected.length)
350 peer_id = b"-DW1000-" + secrets.token_bytes(12)
351 peers = discover_peers(metadata, peer_id, timeout_seconds, cancelled, effective_settings)
352 start_piece = selected.offset // metadata.piece_length
353 end_piece = (selected.offset + selected.length - 1) // metadata.piece_length
354 piece_indices = tuple(range(start_piece, end_piece + 1))
355 selected_end = selected.offset + selected.length
357 def selected_piece_bounds(piece_index: int, piece_size: int) -> tuple[int, int]:
358 piece_start = piece_index * metadata.piece_length
359 piece_end = piece_start + piece_size
360 return (
361 max(selected.offset, piece_start) - piece_start,
362 min(selected_end, piece_end) - piece_start,
363 )
365 def fetch_piece(piece_index: int) -> tuple[int, bytes]:
366 _raise_if_cancelled(cancelled)
367 length = min(
368 metadata.piece_length,
369 metadata.total_length - piece_index * metadata.piece_length,
370 )
371 data = _download_piece_from_peers(
372 metadata,
373 peer_id,
374 peers,
375 piece_index,
376 length,
377 timeout_seconds,
378 cancelled,
379 effective_settings,
380 )
381 return piece_index, data
383 partial = destination.with_name(destination.name + ".part")
384 written = 0
385 remaining_indices = piece_indices
386 if resume and partial.is_file():
387 existing = partial.stat().st_size
388 complete = 0
389 skipped = 0
390 for piece_index in piece_indices: 390 ↛ 401line 390 didn't jump to line 401 because the loop on line 390 didn't complete
391 piece_size = min(
392 metadata.piece_length,
393 metadata.total_length - piece_index * metadata.piece_length,
394 )
395 copy_start, copy_end = selected_piece_bounds(piece_index, piece_size)
396 contribution = copy_end - copy_start
397 if complete + contribution > existing:
398 break
399 complete += contribution
400 skipped += 1
401 if complete != existing: 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true
402 with partial.open("r+b") as output:
403 output.truncate(complete)
404 written = complete
405 remaining_indices = piece_indices[skipped:]
406 if progress is not None and written: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 progress(written, selected.length)
408 try:
409 with (
410 partial.open("ab" if resume and written else "wb") as output,
411 ThreadPoolExecutor(
412 max_workers=min(4, len(remaining_indices)) or 1,
413 thread_name_prefix="torrent-piece",
414 ) as executor,
415 ):
416 for piece_index, data in executor.map(
417 fetch_piece,
418 remaining_indices,
419 buffersize=4,
420 ):
421 _raise_if_cancelled(cancelled)
422 copy_start, copy_end = selected_piece_bounds(piece_index, len(data))
423 block = data[copy_start:copy_end]
424 output.write(block)
425 written += len(block)
426 if progress is not None:
427 progress(written, selected.length)
428 if written != selected.length:
429 raise BitTorrentError("The selected torrent file was truncated.")
430 partial.replace(destination)
431 except BaseException:
432 if not resume: 432 ↛ 434line 432 didn't jump to line 434 because the condition on line 432 was always true
433 partial.unlink(missing_ok=True)
434 raise
437def _select_torrent_file(
438 metadata: TorrentMetadata,
439 catalogue_index: int,
440 expected_filename: str,
441 *,
442 torrent_url: str = "",
443 selected_path: tuple[str, ...] | None = None,
444) -> TorrentFile:
445 """Find one unambiguous filename even when Minerva reordered its torrent."""
447 if selected_path is not None:
448 approved = [item for item in metadata.files if item.path == selected_path]
449 if len(approved) == 1:
450 return approved[0]
451 raise TorrentSelectionRequired(
452 torrent_url=torrent_url,
453 expected_filename=expected_filename,
454 catalogue_index=catalogue_index,
455 candidates=_torrent_file_candidates(metadata, expected_filename, catalogue_index, []),
456 total_files=len(metadata.files),
457 )
459 normalized_expected = _normalized_filename(expected_filename)
460 matches = [
461 (index, item)
462 for index, item in enumerate(metadata.files, start=1)
463 if _normalized_filename(item.path[-1]) == normalized_expected
464 ]
465 if len(matches) != 1:
466 raise TorrentSelectionRequired(
467 torrent_url=torrent_url,
468 expected_filename=expected_filename,
469 catalogue_index=catalogue_index,
470 candidates=_torrent_file_candidates(
471 metadata,
472 expected_filename,
473 catalogue_index,
474 matches,
475 ),
476 total_files=len(metadata.files),
477 )
478 resolved_index, selected = matches[0]
479 if resolved_index != catalogue_index:
480 LOGGER.warning(
481 "Minerva torrent order changed for file=%r catalogue_index=%d torrent_index=%d",
482 expected_filename,
483 catalogue_index,
484 resolved_index,
485 )
486 return selected
489def _normalized_filename(value: str) -> str:
490 return unicodedata.normalize("NFC", value).casefold()
493def _torrent_file_candidates(
494 metadata: TorrentMetadata,
495 expected_filename: str,
496 catalogue_index: int,
497 filename_matches: list[tuple[int, TorrentFile]],
498) -> tuple[TorrentFileChoice, ...]:
499 expected_title = _normalized_title(expected_filename)
501 def choice(index: int, item: TorrentFile) -> TorrentFileChoice:
502 score = SequenceMatcher(None, expected_title, _normalized_title(item.path[-1])).ratio()
503 return TorrentFileChoice(index, item.path, item.length, score)
505 if filename_matches:
506 return tuple(choice(index, item) for index, item in filename_matches)
508 ranked = sorted(
509 (choice(index, item) for index, item in enumerate(metadata.files, start=1)),
510 key=lambda item: (
511 -item.match_score,
512 abs(item.index - catalogue_index),
513 "/".join(item.path).casefold(),
514 ),
515 )
516 closest = ranked[:10]
517 hinted = next((item for item in ranked if item.index == catalogue_index), None)
518 if hinted is not None and hinted not in closest: 518 ↛ 519line 518 didn't jump to line 519 because the condition on line 518 was never true
519 closest.append(hinted)
520 return tuple(closest)
523def _normalized_title(filename: str) -> str:
524 title = Path(filename).stem
525 normalized = unicodedata.normalize("NFKD", title).casefold()
526 return " ".join(re.sub(r"[^a-z0-9]+", " ", normalized).split())
529def discover_peers(
530 metadata: TorrentMetadata,
531 peer_id: bytes,
532 timeout_seconds: float,
533 cancelled: CancelCallback | None = None,
534 settings: BitTorrentSettings | None = None,
535) -> tuple[PeerAddress, ...]:
536 """Merge several tracker responses so one stale swarm view cannot block a download."""
538 effective_settings = settings or BitTorrentSettings()
539 per_tracker_timeout = max(3.0, min(10.0, timeout_seconds / 3))
540 trackers = metadata.trackers[: effective_settings.max_tracker_queries]
542 def announce(tracker: str) -> tuple[PeerAddress, ...]:
543 _raise_if_cancelled(cancelled)
544 try:
545 peers = (
546 _announce_udp(
547 tracker,
548 metadata,
549 peer_id,
550 per_tracker_timeout,
551 effective_settings,
552 )
553 if tracker.startswith("udp://")
554 else _announce_http(
555 tracker,
556 metadata,
557 peer_id,
558 per_tracker_timeout,
559 effective_settings,
560 )
561 )
562 _raise_if_cancelled(cancelled)
563 return peers
564 except BitTorrentCancelled:
565 raise
566 except BitTorrentError, OSError, TimeoutError, ValueError:
567 return ()
569 peers: list[PeerAddress] = []
570 with ThreadPoolExecutor(
571 max_workers=min(8, len(trackers)),
572 thread_name_prefix="torrent-tracker",
573 ) as executor:
574 for discovered in executor.map(announce, trackers):
575 _raise_if_cancelled(cancelled)
576 peers.extend(discovered)
577 peers = list(dict.fromkeys(peer for peer in peers if _is_usable_peer(peer)))
578 if not peers:
579 raise BitTorrentError("No peers were returned by Minerva's torrent trackers.")
580 peers.sort(
581 key=lambda peer: hashlib.sha256(
582 peer_id + f"{peer[0]}:{peer[1]}".encode(),
583 ).digest()
584 )
585 return tuple(peers[: effective_settings.max_discovered_peers])
588def _announce_http(
589 tracker: str,
590 metadata: TorrentMetadata,
591 peer_id: bytes,
592 timeout_seconds: float,
593 settings: BitTorrentSettings,
594) -> tuple[PeerAddress, ...]:
595 parameters = urlencode(
596 {
597 "info_hash": metadata.info_hash,
598 "peer_id": peer_id,
599 "port": 6881,
600 "uploaded": 0,
601 "downloaded": 0,
602 "left": metadata.total_length,
603 "compact": 1,
604 "numwant": 80,
605 }
606 )
607 separator = "&" if "?" in tracker else "?"
608 response = _read_url(
609 f"{tracker}{separator}{parameters}",
610 "",
611 timeout_seconds,
612 settings.max_tracker_bytes,
613 )
614 payload = _as_dictionary(bdecode(response), "tracker response")
615 failure = payload.get(b"failure reason")
616 if isinstance(failure, bytes):
617 raise BitTorrentError(f"Tracker rejected the announce: {failure.decode(errors='replace')}")
618 return _parse_tracker_peers(payload.get(b"peers"))
621def _announce_udp(
622 tracker: str,
623 metadata: TorrentMetadata,
624 peer_id: bytes,
625 timeout_seconds: float,
626 settings: BitTorrentSettings,
627) -> tuple[PeerAddress, ...]:
628 parsed = urlparse(tracker)
629 if not parsed.hostname or not parsed.port:
630 raise BitTorrentError("UDP tracker URL is incomplete.")
631 address = socket.getaddrinfo(
632 parsed.hostname,
633 parsed.port,
634 socket.AF_INET,
635 socket.SOCK_DGRAM,
636 )[0][4]
637 transaction = secrets.randbits(32)
638 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
639 client.settimeout(timeout_seconds)
640 client.connect(address)
641 client.send(struct.pack(">QII", settings.udp_protocol_id, 0, transaction))
642 response = client.recv(2048)
643 if len(response) < 16:
644 raise BitTorrentError("UDP tracker returned a short connect response.")
645 action, response_transaction, connection_id = struct.unpack(">IIQ", response[:16])
646 if action != 0 or response_transaction != transaction:
647 raise BitTorrentError("UDP tracker connect response did not match.")
649 transaction = secrets.randbits(32)
650 packet = struct.pack(
651 ">QII20s20sQQQIIIiH",
652 connection_id,
653 1,
654 transaction,
655 metadata.info_hash,
656 peer_id,
657 0,
658 metadata.total_length,
659 0,
660 0,
661 0,
662 secrets.randbits(32),
663 80,
664 6881,
665 )
666 client.send(packet)
667 response = client.recv(65535)
668 if len(response) < 20:
669 raise BitTorrentError("UDP tracker returned a short announce response.")
670 action, response_transaction = struct.unpack(">II", response[:8])
671 if action != 1 or response_transaction != transaction:
672 raise BitTorrentError("UDP tracker announce response did not match.")
673 return _parse_compact_peers(response[20:])
676def _parse_tracker_peers(value: BValue | None) -> tuple[PeerAddress, ...]:
677 if isinstance(value, bytes):
678 return _parse_compact_peers(value)
679 if not isinstance(value, list):
680 return ()
681 peers: list[PeerAddress] = []
682 for item in value:
683 if not isinstance(item, dict):
684 continue
685 address = item.get(b"ip")
686 port = item.get(b"port")
687 if isinstance(address, bytes) and isinstance(port, int) and 0 < port < 65536:
688 peers.append((address.decode(errors="replace"), port))
689 return tuple(peers)
692def _parse_compact_peers(data: bytes) -> tuple[PeerAddress, ...]:
693 if len(data) % 6:
694 return ()
695 return tuple(
696 (
697 socket.inet_ntoa(data[index : index + 4]),
698 struct.unpack(">H", data[index + 4 : index + 6])[0],
699 )
700 for index in range(0, len(data), 6)
701 )
704def _download_piece_from_peers(
705 metadata: TorrentMetadata,
706 peer_id: bytes,
707 peers: tuple[PeerAddress, ...],
708 piece_index: int,
709 piece_length: int,
710 timeout_seconds: float,
711 cancelled: CancelCallback | None = None,
712 settings: BitTorrentSettings | None = None,
713) -> bytes:
714 effective_settings = settings or BitTorrentSettings()
715 start = int.from_bytes(peer_id[-4:], "big") + piece_index
716 attempts = min(effective_settings.max_peer_attempts, len(peers))
717 candidates = tuple(peers[(start + attempt) % len(peers)] for attempt in range(attempts))
718 peer_timeout = max(0.1, min(effective_settings.max_peer_timeout_seconds, timeout_seconds))
719 completed = Event()
721 def fetch(peer: PeerAddress) -> bytes | None:
722 _raise_if_cancelled(cancelled)
723 if completed.is_set():
724 return None
725 try:
726 with _peer_connection_slot(peer, completed, cancelled) as acquired:
727 if not acquired: 727 ↛ 728line 727 didn't jump to line 728 because the condition on line 727 was never true
728 return None
729 data = _download_piece(
730 peer,
731 metadata.info_hash,
732 peer_id,
733 piece_index,
734 piece_length,
735 peer_timeout,
736 completed,
737 cancelled,
738 effective_settings,
739 )
740 except BitTorrentCancelled:
741 raise
742 except BitTorrentError, OSError, TimeoutError:
743 return None
744 digest = hashlib.sha1(data, usedforsecurity=False).digest()
745 if digest == metadata.piece_hashes[piece_index]:
746 completed.set()
747 return data
748 return None
750 executor = ThreadPoolExecutor(
751 max_workers=min(effective_settings.peer_race_workers, len(candidates)),
752 thread_name_prefix="torrent-peer",
753 )
754 futures = [executor.submit(fetch, peer) for peer in candidates]
755 try:
756 for future in as_completed(futures):
757 _raise_if_cancelled(cancelled)
758 data = future.result()
759 if data is None:
760 continue
761 for pending in futures:
762 pending.cancel()
763 return data
764 finally:
765 for pending in futures:
766 pending.cancel()
767 executor.shutdown(wait=False, cancel_futures=True)
768 raise BitTorrentError(f"Could not retrieve verified torrent piece {piece_index}.")
771@contextmanager
772def _peer_connection_slot(
773 peer: PeerAddress,
774 completed: Event,
775 cancelled: CancelCallback | None,
776) -> Iterator[bool]:
777 """Allow only one active torrent connection to a peer across all downloads."""
779 with _PEER_GATES_LOCK:
780 gate = _PEER_GATES.setdefault(peer, _PeerGate(Lock()))
781 gate.users += 1
783 acquired = False
784 try:
785 while not acquired:
786 _raise_if_cancelled(cancelled)
787 if completed.is_set(): 787 ↛ 788line 787 didn't jump to line 788 because the condition on line 787 was never true
788 yield False
789 return
790 acquired = gate.lock.acquire(timeout=0.1)
791 _raise_if_cancelled(cancelled)
792 if completed.is_set(): 792 ↛ 793line 792 didn't jump to line 793 because the condition on line 792 was never true
793 yield False
794 return
795 yield True
796 finally:
797 if acquired: 797 ↛ 799line 797 didn't jump to line 799 because the condition on line 797 was always true
798 gate.lock.release()
799 with _PEER_GATES_LOCK:
800 gate.users -= 1
801 if gate.users == 0 and _PEER_GATES.get(peer) is gate:
802 del _PEER_GATES[peer]
805def _download_piece(
806 peer: PeerAddress,
807 info_hash: bytes,
808 peer_id: bytes,
809 piece_index: int,
810 piece_length: int,
811 timeout_seconds: float,
812 completed: Event | None = None,
813 cancelled: CancelCallback | None = None,
814 settings: BitTorrentSettings | None = None,
815) -> bytes:
816 effective_settings = settings or BitTorrentSettings()
817 _raise_if_cancelled(cancelled)
818 with socket.create_connection(peer, timeout=timeout_seconds) as connection:
819 connection.settimeout(timeout_seconds)
820 connection.sendall(bytes((len(_PROTOCOL),)) + _PROTOCOL + b"\0" * 8 + info_hash + peer_id)
821 handshake = _read_exact(connection, 68)
822 if handshake[1:20] != _PROTOCOL or handshake[28:48] != info_hash:
823 raise BitTorrentError("Peer returned an invalid handshake.")
824 connection.sendall(struct.pack(">IB", 1, 2))
826 available: bytes | None = None
827 unchoked = False
828 deadline = time.monotonic() + timeout_seconds
829 while not unchoked and time.monotonic() < deadline:
830 _raise_if_cancelled(cancelled)
831 if completed is not None and completed.is_set():
832 raise BitTorrentError("Another peer completed the requested piece.")
833 message_id, payload = _read_peer_message(connection, effective_settings.block_size)
834 if message_id == 1:
835 unchoked = True
836 elif message_id == 5:
837 available = payload
838 elif message_id == 4 and len(payload) == 4: 838 ↛ 829line 838 didn't jump to line 829 because the condition on line 838 was always true
839 announced = struct.unpack(">I", payload)[0]
840 if announced == piece_index: 840 ↛ 829line 840 didn't jump to line 829 because the condition on line 840 was always true
841 available = None
842 if not unchoked: 842 ↛ 843line 842 didn't jump to line 843 because the condition on line 842 was never true
843 raise BitTorrentError("Peer did not unchoke the download.")
844 if available is not None:
845 byte_index, bit_index = divmod(piece_index, 8)
846 if byte_index >= len(available) or not available[byte_index] & (1 << (7 - bit_index)): 846 ↛ 849line 846 didn't jump to line 849 because the condition on line 846 was always true
847 raise BitTorrentError("Peer does not have the requested piece.")
849 requests = [
850 (begin, min(effective_settings.block_size, piece_length - begin))
851 for begin in range(0, piece_length, effective_settings.block_size)
852 ]
853 pending = iter(requests)
854 received: dict[int, bytes] = {}
855 in_flight = 0
856 for _ in range(min(16, len(requests))):
857 begin, length = next(pending)
858 _send_piece_request(connection, piece_index, begin, length)
859 in_flight += 1
860 while in_flight:
861 _raise_if_cancelled(cancelled)
862 if completed is not None and completed.is_set(): 862 ↛ 863line 862 didn't jump to line 863 because the condition on line 862 was never true
863 raise BitTorrentError("Another peer completed the requested piece.")
864 message_id, payload = _read_peer_message(connection, effective_settings.block_size)
865 if message_id == 0:
866 raise BitTorrentError("Peer choked the download.")
867 if message_id != 7 or len(payload) < 8:
868 continue
869 response_piece, begin = struct.unpack(">II", payload[:8])
870 block = payload[8:]
871 if response_piece != piece_index or begin in received:
872 continue
873 expected = dict(requests).get(begin)
874 if expected != len(block):
875 raise BitTorrentError("Peer returned an invalid piece block.")
876 received[begin] = block
877 in_flight -= 1
878 try:
879 next_begin, next_length = next(pending)
880 except StopIteration:
881 continue
882 _send_piece_request(connection, piece_index, next_begin, next_length)
883 in_flight += 1
884 return b"".join(received[begin] for begin, _length in requests)
887def _send_piece_request(
888 connection: socket.socket,
889 piece_index: int,
890 begin: int,
891 length: int,
892) -> None:
893 connection.sendall(struct.pack(">IBIII", 13, 6, piece_index, begin, length))
896def _read_peer_message(
897 connection: socket.socket,
898 block_size: int = _BLOCK_SIZE,
899) -> tuple[int | None, bytes]:
900 length = struct.unpack(">I", _read_exact(connection, 4))[0]
901 if length == 0:
902 return None, b""
903 if length > block_size + 64:
904 raise BitTorrentError("Peer message is unexpectedly large.")
905 message = _read_exact(connection, length)
906 return message[0], message[1:]
909def _read_exact(connection: socket.socket, length: int) -> bytes:
910 chunks: list[bytes] = []
911 remaining = length
912 while remaining:
913 chunk = connection.recv(remaining)
914 if not chunk:
915 raise BitTorrentError("Peer closed the connection early.")
916 chunks.append(chunk)
917 remaining -= len(chunk)
918 return b"".join(chunks)
921def _read_url(url: str, referer: str, timeout_seconds: float, limit: int) -> bytes:
922 headers = {"User-Agent": USER_AGENT}
923 if referer:
924 headers["Referer"] = referer
925 request = Request(url, headers=headers)
926 try:
927 with urlopen(
928 request,
929 timeout=timeout_seconds,
930 context=ssl.create_default_context(),
931 ) as response:
932 data = response.read(limit + 1)
933 except HTTPError as error:
934 raise BitTorrentError("Torrent service returned HTTP %d." % error.code) from error
935 except (URLError, TimeoutError, OSError) as error:
936 reason = getattr(error, "reason", error)
937 raise BitTorrentError(f"Could not reach the torrent service: {reason}") from error
938 if len(data) > limit:
939 raise BitTorrentError("Torrent service response exceeded the safe size limit.")
940 return data
943def _as_dictionary(value: BValue | None, label: str) -> dict[bytes, BValue]:
944 if not isinstance(value, dict):
945 raise BitTorrentError(f"Invalid {label}.")
946 return value
949def _as_list(value: BValue | None, label: str) -> list[BValue]:
950 if not isinstance(value, list):
951 raise BitTorrentError(f"Invalid {label}.")
952 return value
955def _as_bytes(value: BValue | None, label: str) -> bytes:
956 if not isinstance(value, bytes):
957 raise BitTorrentError(f"Invalid {label}.")
958 return value
961def _as_positive_integer(value: BValue | None, label: str) -> int:
962 if not isinstance(value, int) or value <= 0:
963 raise BitTorrentError(f"Invalid {label}.")
964 return value
967def _as_nonnegative_integer(value: BValue | None, label: str) -> int:
968 if not isinstance(value, int) or value < 0:
969 raise BitTorrentError(f"Invalid {label}.")
970 return value
973def _decode_path_part(value: BValue | None) -> str:
974 raw = _as_bytes(value, "file path")
975 decoded = raw.decode("utf-8", errors="replace")
976 if "/" in decoded or "\\" in decoded:
977 raise BitTorrentError("Torrent contains an unsafe file path.")
978 return decoded
981def compact_peers(peers: Iterable[PeerAddress]) -> bytes:
982 """Encode IPv4 peers for deterministic tracker tests."""
984 return b"".join(socket.inet_aton(address) + struct.pack(">H", port) for address, port in peers)
987def _is_usable_peer(peer: PeerAddress) -> bool:
988 try:
989 address = ipaddress.ip_address(peer[0])
990 except ValueError:
991 return bool(peer[0])
992 return address.is_global
995def _raise_if_cancelled(cancelled: CancelCallback | None) -> None:
996 if cancelled is not None and cancelled():
997 raise BitTorrentCancelled("Torrent download cancelled.")