""" Real-world tracker probe for torrent-tracker. Given a .torrent file, this harness parses the metainfo, computes the tracker info_hash value, and speaks HTTP(S) and UDP tracker announce/scrape protocols directly. It is intentionally stdlib-only so it can run anywhere the C library builds, without requiring libtorrent. """ from __future__ import annotations import argparse import hashlib import os import random import socket import ssl import struct import sys import time from dataclasses import dataclass from typing import Any from urllib.parse import quote_from_bytes, urlsplit, urlunsplit from urllib.request import Request, urlopen UDP_PROTOCOL_ID = 0x0000041727101980 UDP_CONNECT = 0 UDP_ANNOUNCE = 1 UDP_SCRAPE = 2 UDP_ERROR = 3 class BencodeError(ValueError): pass class BDecoder: def __init__(self, data: bytes): self.data = data self.info_span: tuple[int, int] | None = None def parse(self) -> Any: value, pos = self._value(0, top=True) if pos != len(self.data): raise BencodeError(f"trailing data at byte {pos}") return value def _value(self, pos: int, *, top: bool = False) -> tuple[Any, int]: if pos >= len(self.data): raise BencodeError("unexpected end of bencode") c = self.data[pos] if c == ord("i"): return self._int(pos) if c == ord("l"): return self._list(pos) if c == ord("d"): return self._dict(pos, top=top) if ord("0") <= c <= ord("9"): return self._bytes(pos) raise BencodeError(f"invalid bencode byte {c!r} at {pos}") def _int(self, pos: int) -> tuple[int, int]: end = self.data.find(b"e", pos) if end < 0: raise BencodeError("unterminated integer") raw = self.data[pos + 1:end] if not raw: raise BencodeError("empty integer") return int(raw), end + 1 def _bytes(self, pos: int) -> tuple[bytes, int]: colon = self.data.find(b":", pos) if colon < 0: raise BencodeError("unterminated byte string length") n = int(self.data[pos:colon]) start = colon + 1 end = start + n if end > len(self.data): raise BencodeError("byte string exceeds input") return self.data[start:end], end def _list(self, pos: int) -> tuple[list[Any], int]: out: list[Any] = [] pos += 1 while pos < len(self.data) and self.data[pos] != ord("e"): value, pos = self._value(pos) out.append(value) if pos >= len(self.data): raise BencodeError("unterminated list") return out, pos + 1 def _dict(self, pos: int, *, top: bool = False) -> tuple[dict[bytes, Any], int]: out: dict[bytes, Any] = {} pos += 1 while pos < len(self.data) and self.data[pos] != ord("e"): key, pos = self._bytes(pos) value_start = pos value, pos = self._value(pos) if top and key == b"info": self.info_span = (value_start, pos) out[key] = value if pos >= len(self.data): raise BencodeError("unterminated dict") return out, pos + 1 @dataclass class TorrentMeta: path: str name: str info_hash: bytes info_hash_kind: str total_size: int trackers: list[str] @dataclass class AnnounceResult: tracker: str ok: bool protocol: str interval: int | None = None min_interval: int | None = None complete: int | None = None incomplete: int | None = None downloaded: int | None = None peers: list[tuple[str, int]] | None = None warning: str | None = None error: str | None = None elapsed_ms: float = 0.0 def _text(value: Any, default: str = "") -> str: if isinstance(value, bytes): return value.decode("utf-8", "replace") return default def _file_tree_size(node: Any) -> int: if not isinstance(node, dict): return 0 total = 0 file_marker = node.get(b"") if isinstance(file_marker, dict): total += int(file_marker.get(b"length", 0)) for key, child in node.items(): if key != b"": total += _file_tree_size(child) return total def _total_size(info: dict[bytes, Any]) -> int: if b"length" in info: return int(info[b"length"]) if b"files" in info: return sum(int(f.get(b"length", 0)) for f in info[b"files"]) if b"file tree" in info: return _file_tree_size(info[b"file tree"]) return 0 def _trackers(meta: dict[bytes, Any]) -> list[str]: urls: list[str] = [] announce = meta.get(b"announce") if isinstance(announce, bytes): urls.append(_text(announce)) tiers = meta.get(b"announce-list") if isinstance(tiers, list): for tier in tiers: if not isinstance(tier, list): continue for item in tier: if isinstance(item, bytes): urls.append(_text(item)) seen: set[str] = set() out: list[str] = [] for url in urls: if url and url not in seen: seen.add(url) out.append(url) return out def load_torrent(path: str) -> TorrentMeta: raw = open(path, "rb").read() dec = BDecoder(raw) meta = dec.parse() if not isinstance(meta, dict) or dec.info_span is None: raise BencodeError("metainfo does not contain a top-level info dict") info = meta[b"info"] info_raw = raw[dec.info_span[0]:dec.info_span[1]] if b"pieces" in info: info_hash = hashlib.sha1(info_raw).digest() kind = "v1 sha1" elif info.get(b"meta version") == 2: info_hash = hashlib.sha256(info_raw).digest()[:20] kind = "v2 sha256-truncated" else: raise BencodeError("unsupported torrent: no v1 pieces or v2 meta version") return TorrentMeta( path=path, name=_text(info.get(b"name"), os.path.basename(path)), info_hash=info_hash, info_hash_kind=kind, total_size=_total_size(info), trackers=_trackers(meta), ) def make_peer_id() -> bytes: return b"-TG0001-" + os.urandom(12) def _http_announce_url(url: str, meta: TorrentMeta, peer_id: bytes, port: int, key: int, numwant: int, event: str) -> str: parts = urlsplit(url) query = parts.query extra = [ ("info_hash", quote_from_bytes(meta.info_hash, safe="")), ("peer_id", quote_from_bytes(peer_id, safe="")), ("port", str(port)), ("uploaded", "0"), ("downloaded", "0"), ("left", str(meta.total_size)), ("compact", "1"), ("numwant", str(numwant)), ("key", str(key)), ] if event: extra.append(("event", event)) suffix = "&".join(f"{k}={v}" for k, v in extra) query = f"{query}&{suffix}" if query else suffix return urlunsplit((parts.scheme, parts.netloc, parts.path, query, parts.fragment)) def _decode_compact(peers: bytes, family: int) -> list[tuple[str, int]]: stride = 6 if family == socket.AF_INET else 18 addr_len = 4 if family == socket.AF_INET else 16 if len(peers) % stride != 0: raise ValueError(f"compact peer string length {len(peers)} is not a multiple of {stride}") out = [] for off in range(0, len(peers), stride): addr = socket.inet_ntop(family, peers[off:off + addr_len]) port = struct.unpack("!H", peers[off + addr_len:off + stride])[0] out.append((addr, port)) return out def _decode_peer_list(value: Any) -> list[tuple[str, int]]: peers: list[tuple[str, int]] = [] if isinstance(value, bytes): peers.extend(_decode_compact(value, socket.AF_INET)) elif isinstance(value, list): for item in value: if not isinstance(item, dict): continue ip = _text(item.get(b"ip")) port = item.get(b"port") if ip and isinstance(port, int): peers.append((ip, port)) return peers def _parse_tracker_dict(raw: bytes, tracker: str, protocol: str, elapsed_ms: float) -> AnnounceResult: data = BDecoder(raw).parse() if not isinstance(data, dict): return AnnounceResult(tracker, False, protocol, error="response is not a dict", elapsed_ms=elapsed_ms) failure = data.get(b"failure reason") if isinstance(failure, bytes): return AnnounceResult(tracker, False, protocol, error=_text(failure), elapsed_ms=elapsed_ms) peers = _decode_peer_list(data.get(b"peers", b"")) peers6 = data.get(b"peers6") if isinstance(peers6, bytes): peers.extend(_decode_compact(peers6, socket.AF_INET6)) return AnnounceResult( tracker=tracker, ok=True, protocol=protocol, interval=data.get(b"interval") if isinstance(data.get(b"interval"), int) else None, min_interval=data.get(b"min interval") if isinstance(data.get(b"min interval"), int) else None, complete=data.get(b"complete") if isinstance(data.get(b"complete"), int) else None, incomplete=data.get(b"incomplete") if isinstance(data.get(b"incomplete"), int) else None, peers=peers, warning=_text(data.get(b"warning message")) if b"warning message" in data else None, elapsed_ms=elapsed_ms, ) def announce_http(url: str, meta: TorrentMeta, peer_id: bytes, port: int, key: int, numwant: int, event: str, timeout: float) -> AnnounceResult: start = time.monotonic() announce_url = _http_announce_url(url, meta, peer_id, port, key, numwant, event) try: req = Request(announce_url, headers={"User-Agent": "torrent-tracker-harness/0.1"}) ctx = ssl.create_default_context() with urlopen(req, timeout=timeout, context=ctx) as resp: raw = resp.read(2 * 1024 * 1024) elapsed = (time.monotonic() - start) * 1000.0 return _parse_tracker_dict(raw, url, "http", elapsed) except Exception as exc: elapsed = (time.monotonic() - start) * 1000.0 return AnnounceResult(url, False, "http", error=str(exc), elapsed_ms=elapsed) def _udp_url_data(url: str) -> bytes: parts = urlsplit(url) data = parts.path or b"" if isinstance(data, str): data = data.encode("utf-8") if parts.query: data += b"?" + parts.query.encode("utf-8") return data def _udp_options(url: str) -> bytes: data = _udp_url_data(url) if not data: return b"" out = bytearray() for off in range(0, len(data), 255): chunk = data[off:off + 255] out += bytes([0x02, len(chunk)]) + chunk return bytes(out) def _udp_roundtrip(sock: socket.socket, packet: bytes, txid: int, timeout: float) -> bytes: deadline = time.monotonic() + timeout delay = min(timeout, 1.0) while True: sock.send(packet) remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("UDP tracker timed out") sock.settimeout(min(delay, remaining)) try: raw = sock.recv(65535) except socket.timeout: delay = min(delay * 2.0, 8.0) continue if len(raw) >= 8 and struct.unpack_from("!I", raw, 4)[0] == txid: return raw def _parse_udp_announce(raw: bytes, tracker: str, family: int, elapsed_ms: float) -> AnnounceResult: if len(raw) < 8: return AnnounceResult(tracker, False, "udp", error="short UDP response", elapsed_ms=elapsed_ms) action = struct.unpack_from("!I", raw, 0)[0] if action == UDP_ERROR: return AnnounceResult(tracker, False, "udp", error=raw[8:].decode("utf-8", "replace"), elapsed_ms=elapsed_ms) if action != UDP_ANNOUNCE or len(raw) < 20: return AnnounceResult(tracker, False, "udp", error=f"unexpected UDP action {action}", elapsed_ms=elapsed_ms) interval, incomplete, complete = struct.unpack_from("!III", raw, 8) peers = _decode_compact(raw[20:], family) return AnnounceResult(tracker, True, "udp", interval=interval, complete=complete, incomplete=incomplete, peers=peers, elapsed_ms=elapsed_ms) def announce_udp(url: str, meta: TorrentMeta, peer_id: bytes, port: int, key: int, numwant: int, event: str, timeout: float) -> AnnounceResult: parts = urlsplit(url) host = parts.hostname if not host: return AnnounceResult(url, False, "udp", error="missing UDP tracker host") tracker_port = parts.port or 80 event_id = {"": 0, "completed": 1, "started": 2, "stopped": 3}.get(event, 0) start = time.monotonic() try: infos = socket.getaddrinfo(host, tracker_port, 0, socket.SOCK_DGRAM) last_error: Exception | None = None for family, socktype, proto, _canon, sockaddr in infos: if family not in (socket.AF_INET, socket.AF_INET6): continue try: with socket.socket(family, socktype, proto) as sock: sock.connect(sockaddr) txid = random.getrandbits(32) connect = struct.pack("!QII", UDP_PROTOCOL_ID, UDP_CONNECT, txid) raw = _udp_roundtrip(sock, connect, txid, timeout) action, got_txid, conn_id = struct.unpack("!IIQ", raw[:16]) if action != UDP_CONNECT or got_txid != txid: raise OSError("invalid UDP connect response") txid = random.getrandbits(32) announce = struct.pack( "!QII20s20sQQQIIIiH", conn_id, UDP_ANNOUNCE, txid, meta.info_hash, peer_id, 0, meta.total_size, 0, event_id, 0, key, numwant, port, ) + _udp_options(url) raw = _udp_roundtrip(sock, announce, txid, timeout) elapsed = (time.monotonic() - start) * 1000.0 return _parse_udp_announce(raw, url, family, elapsed) except Exception as exc: last_error = exc continue elapsed = (time.monotonic() - start) * 1000.0 return AnnounceResult(url, False, "udp", error=str(last_error or "no usable address"), elapsed_ms=elapsed) except Exception as exc: elapsed = (time.monotonic() - start) * 1000.0 return AnnounceResult(url, False, "udp", error=str(exc), elapsed_ms=elapsed) def _scrape_url(url: str) -> str | None: parts = urlsplit(url) idx = parts.path.rfind("announce") if idx < 0: return None path = parts.path[:idx] + "scrape" + parts.path[idx + len("announce"):] return urlunsplit((parts.scheme, parts.netloc, path, parts.query, parts.fragment)) def scrape_http(url: str, meta: TorrentMeta, timeout: float) -> AnnounceResult: scrape = _scrape_url(url) if not scrape: return AnnounceResult(url, False, "http-scrape", error="no scrape URL") parts = urlsplit(scrape) q = parts.query suffix = "info_hash=" + quote_from_bytes(meta.info_hash, safe="") q = f"{q}&{suffix}" if q else suffix scrape = urlunsplit((parts.scheme, parts.netloc, parts.path, q, parts.fragment)) start = time.monotonic() try: req = Request(scrape, headers={"User-Agent": "torrent-tracker-harness/0.1"}) ctx = ssl.create_default_context() with urlopen(req, timeout=timeout, context=ctx) as resp: raw = resp.read(2 * 1024 * 1024) elapsed = (time.monotonic() - start) * 1000.0 data = BDecoder(raw).parse() if not isinstance(data, dict): raise ValueError("scrape response is not a dict") failure = data.get(b"failure reason") if isinstance(failure, bytes): return AnnounceResult(url, False, "http-scrape", error=_text(failure), elapsed_ms=elapsed) files = data.get(b"files", {}) entry = files.get(meta.info_hash) if isinstance(files, dict) else None if not isinstance(entry, dict): return AnnounceResult(url, False, "http-scrape", error="info_hash missing from scrape response", elapsed_ms=elapsed) return AnnounceResult(url, True, "http-scrape", complete=entry.get(b"complete"), incomplete=entry.get(b"incomplete"), downloaded=entry.get(b"downloaded"), elapsed_ms=elapsed) except Exception as exc: elapsed = (time.monotonic() - start) * 1000.0 return AnnounceResult(url, False, "http-scrape", error=str(exc), elapsed_ms=elapsed) def announce(url: str, meta: TorrentMeta, peer_id: bytes, port: int, key: int, numwant: int, event: str, timeout: float) -> AnnounceResult: scheme = urlsplit(url).scheme.lower() if scheme in ("http", "https"): return announce_http(url, meta, peer_id, port, key, numwant, event, timeout) if scheme == "udp": return announce_udp(url, meta, peer_id, port, key, numwant, event, timeout) return AnnounceResult(url, False, scheme or "unknown", error=f"unsupported tracker scheme {scheme!r}") def print_result(result: AnnounceResult, max_peers: int) -> None: status = "ok" if result.ok else "fail" print(f"[{status}] {result.protocol} {result.tracker} ({result.elapsed_ms:.0f} ms)") if result.error: print(f" error: {result.error}") if result.warning: print(f" warning: {result.warning}") if result.ok: stats = [] if result.interval is not None: stats.append(f"interval={result.interval}") if result.min_interval is not None: stats.append(f"min_interval={result.min_interval}") if result.complete is not None: stats.append(f"seeders={result.complete}") if result.incomplete is not None: stats.append(f"leechers={result.incomplete}") if result.downloaded is not None: stats.append(f"downloaded={result.downloaded}") if stats: print(" " + " ".join(stats)) peers = result.peers or [] if peers: shown = ", ".join(f"{host}:{port}" for host, port in peers[:max_peers]) suffix = "" if len(peers) <= max_peers else f" ... +{len(peers) - max_peers}" print(f" peers[{len(peers)}]: {shown}{suffix}") def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description="Probe real BitTorrent trackers from a .torrent file.") ap.add_argument("torrent", help="path to .torrent file") ap.add_argument("--tracker", action="append", help="tracker URL to probe instead of URLs from the torrent; repeatable") ap.add_argument("--max-trackers", type=int, default=8, help="maximum trackers to probe from the torrent") ap.add_argument("--timeout", type=float, default=8.0, help="per-tracker timeout in seconds") ap.add_argument("--numwant", type=int, default=50, help="numwant value in announce requests") ap.add_argument("--port", type=int, default=6881, help="port value to announce") ap.add_argument("--event", choices=["", "started", "completed", "stopped"], default="started") ap.add_argument("--scrape", action="store_true", help="also try HTTP scrape endpoints derived from announce URLs") ap.add_argument("--show-peers", type=int, default=10, help="number of returned peers to print per tracker") args = ap.parse_args(argv) meta = load_torrent(args.torrent) trackers = args.tracker or meta.trackers if args.max_trackers > 0: trackers = trackers[:args.max_trackers] print(f"torrent: {meta.name}") print(f"size: {meta.total_size} bytes") print(f"info_hash: {meta.info_hash.hex()} ({meta.info_hash_kind})") print(f"trackers: {len(trackers)}") if not trackers: print("no trackers found") return 1 peer_id = make_peer_id() key = random.getrandbits(32) successes = 0 for url in trackers: result = announce(url, meta, peer_id, args.port, key, args.numwant, args.event, args.timeout) if result.ok: successes += 1 print_result(result, args.show_peers) if args.scrape and urlsplit(url).scheme.lower() in ("http", "https"): print_result(scrape_http(url, meta, args.timeout), args.show_peers) return 0 if successes else 2 if __name__ == "__main__": raise SystemExit(main())