bittorrent_peer_ids.py

"""Peer ID constants and parser based on BEP 0020.

Source: docs/bep_0020.rst_post.html
"""

from __future__ import annotations

from dataclasses import dataclass
import re
from typing import Optional


PEER_ID_LENGTH = 20

# Azureus-style: -XXyyyy-
AZUREUS_STYLE_CLIENTS: dict[str, str] = {
    "AG": "Ares",
    "A~": "Ares",
    "AR": "Arctic",
    "AV": "Avicora",
    "AX": "BitPump",
    "AZ": "Azureus",
    "BB": "BitBuddy",
    "BC": "BitComet",
    "BF": "Bitflu",
    "BG": "BTG (Rasterbar libtorrent)",
    "BR": "BitRocket",
    "BS": "BTSlave",
    "BX": "~Bittorrent X",
    "CD": "Enhanced CTorrent",
    "CT": "CTorrent",
    "DE": "DelugeTorrent",
    "DP": "Propagate Data Client",
    "EB": "EBit",
    "ES": "electric sheep",
    "FT": "FoxTorrent",
    "FW": "FrostWire",
    "FX": "Freebox BitTorrent",
    "GS": "GSTorrent",
    "HL": "Halite",
    "HN": "Hydranode",
    "KG": "KGet",
    "KT": "KTorrent",
    "LH": "LH-ABC",
    "LP": "Lphant",
    "LT": "libtorrent",
    "lt": "libTorrent",
    "LW": "LimeWire",
    "MO": "MonoTorrent",
    "MP": "MooPolice",
    "MR": "Miro",
    "MT": "MoonlightTorrent",
    "NX": "Net Transport",
    "PD": "Pando",
    "qB": "qBittorrent",
    "QD": "QQDownload",
    "QT": "Qt 4 Torrent example",
    "RT": "Retriever",
    "S~": "Shareaza alpha/beta",
    "SB": "~Swiftbit",
    "SS": "SwarmScope",
    "ST": "SymTorrent",
    "st": "sharktorrent",
    "SZ": "Shareaza",
    "TN": "TorrentDotNET",
    "TR": "Transmission",
    "TS": "Torrentstorm",
    "TT": "TuoTu",
    "UL": "uLeecher!",
    "UT": "uTorrent",
    "UW": "uTorrent Web",
    "VG": "Vagaa",
    "WD": "WebTorrent Desktop",
    "WT": "BitLet",
    "WW": "WebTorrent",
    "WY": "FireTorrent",
    "XL": "Xunlei",
    "XT": "XanTorrent",
    "XX": "Xtorrent",
    "ZT": "ZipTorrent",
}

AZUREUS_STYLE_UNKNOWN_IDS: dict[str, str] = {
    "BD": "Unknown client (seen in wild)",
    "NP": "Unknown client (seen in wild)",
    "wF": "Unknown client (seen in wild)",
}

SHADOW_STYLE_CLIENTS: dict[str, str] = {
    "A": "ABC",
    "O": "Osprey Permaseed",
    "Q": "BTQueue",
    "R": "Tribler",
    "S": "Shadow's client",
    "T": "BitTornado",
    "U": "UPnP NAT Bit Torrent",
}

SHADOW_VERSION_CHARS = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz.-"


@dataclass(slots=True)
class ParsedPeerID:
    raw: bytes
    style: str
    client_code: str
    client_name: str
    version: Optional[str] = None
    extra: Optional[str] = None


def _safe_ascii(peer_id: bytes) -> str:
    return peer_id.decode("ascii", errors="replace")


def _format_azureus_version(v: str) -> str:
    # BEP 20: four ascii digits after client code.
    return ".".join(str(int(ch)) for ch in v) if v.isdigit() else v


def parse_peer_id(peer_id: bytes) -> ParsedPeerID:
    """Parse known BitTorrent peer ID conventions from BEP 0020."""
    if len(peer_id) != PEER_ID_LENGTH:
        raise ValueError(f"peer_id must be {PEER_ID_LENGTH} bytes")
    text = _safe_ascii(peer_id)

    # Mainline style: Mx-y-z-- (remaining bytes random)
    if text.startswith("M"):
        m = re.match(r"^M([0-9]+)-([0-9]+)-([0-9]+)-", text[:8])
        if m:
            version = ".".join(m.groups())
            return ParsedPeerID(peer_id, "mainline", "M", "Mainline", version)
        return ParsedPeerID(peer_id, "mainline", "M", "Mainline")

    # Azureus style: -XXyyyy-
    m = re.match(r"^-([ -~]{2})([0-9]{4})-", text[:8])
    if m:
        code = m.group(1)
        version_digits = m.group(2)
        client_name = AZUREUS_STYLE_CLIENTS.get(code) or AZUREUS_STYLE_UNKNOWN_IDS.get(code) or "Unknown"
        return ParsedPeerID(
            raw=peer_id,
            style="azureus",
            client_code=code,
            client_name=client_name,
            version=_format_azureus_version(version_digits),
        )

    # Shadow style: [A-Z][up to 5 chars]---
    if len(text) >= 8 and text[6:8] == "--":
        code = text[0]
        if code in SHADOW_STYLE_CLIENTS:
            version_token = text[1:6].rstrip("-")
            if all(ch in SHADOW_VERSION_CHARS for ch in version_token):
                return ParsedPeerID(
                    raw=peer_id,
                    style="shadow",
                    client_code=code,
                    client_name=SHADOW_STYLE_CLIENTS[code],
                    version=version_token or None,
                )

    # BitComet old style: exbc + two version bytes.
    if peer_id.startswith(b"exbc") or peer_id.startswith(b"FUTB"):
        major = peer_id[4]
        minor = peer_id[5]
        name = "BitComet (FUTB patch)" if peer_id.startswith(b"FUTB") else "BitComet"
        if b"LORD" in peer_id[6:12]:
            name = "BitLord"
        return ParsedPeerID(peer_id, "bitcomet", _safe_ascii(peer_id[:4]), name, f"{major}.{minor:02d}")

    # XBT style: XBTddd[d|-]-
    m = re.match(r"^XBT([0-9]{3})([d-])-", text[:8])
    if m:
        digits = m.group(1)
        version = f"{digits[0]}.{digits[1]}.{digits[2]}"
        extra = "debug" if m.group(2) == "d" else "release"
        return ParsedPeerID(peer_id, "xbt", "XBT", "XBT Client", version, extra=extra)

    # Opera style: OP + 4 build digits.
    m = re.match(r"^OP([0-9]{4})", text[:6])
    if m:
        return ParsedPeerID(peer_id, "opera", "OP", "Opera", version=m.group(1))

    # MLdonkey: -ML<dotted-version>-
    m = re.match(r"^-ML([0-9]+(?:\.[0-9]+)*)-", text)
    if m:
        return ParsedPeerID(peer_id, "mldonkey", "ML", "MLdonkey", version=m.group(1))

    # Bits on Wheels: -BOWxxx-
    m = re.match(r"^-BOW([A-Za-z0-9]{3})-", text[:8])
    if m:
        return ParsedPeerID(peer_id, "bow", "BOW", "Bits on Wheels", version=m.group(1))

    # Queen Bee (mainline-like but Q prefix)
    if text.startswith("Q"):
        m = re.match(r"^Q([0-9]+)-([0-9]+)-([0-9]+)-", text[:8])
        if m:
            return ParsedPeerID(peer_id, "queenbee", "Q", "Queen Bee", version=".".join(m.groups()))

    # BitTyrant special case.
    if text.startswith("AZ") and text[6:8] == "BT":
        return ParsedPeerID(peer_id, "bittyrant", "AZ", "BitTyrant")

    # TorrenTopia style noted in BEP 20.
    if text.startswith("346------"):
        return ParsedPeerID(peer_id, "torrentopia", "346", "TorrenTopia", version="1.90")

    return ParsedPeerID(peer_id, "unknown", text[:2], "Unknown")