563 lines
21 KiB
Python
563 lines
21 KiB
Python
"""
|
|
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())
|