all repos — rastro @ 303f75c7f98bcf857e2ef49ec4c4489ce9eb333b

BitTorrent tracker!

app/services/normalize.py (view raw)

 1
 2
 3
 4
 5
 6
 7
 8
 9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
 100
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
from __future__ import annotations

from dataclasses import dataclass
from urllib.parse import urlparse, urlunparse

SUPPORTED_SCHEMES = frozenset({"udp", "http", "https"})
UNSUPPORTED_SCHEMES = frozenset({"ws", "wss", "i2p", "ygg"})


class NormalizationError(ValueError):
    """Raised when a URL cannot be normalized into a supported tracker endpoint."""


@dataclass(frozen=True, slots=True)
class NormalizedTracker:
    canonical_url: str
    scheme: str
    hostname: str
    port: int
    path: str
    unsupported: bool = False


def _default_port(scheme: str) -> int:
    if scheme == "https":
        return 443
    if scheme == "http":
        return 80
    return 80  # udp commonly requires explicit port; fallback unused when port missing


def _idna_hostname(hostname: str) -> str:
    hostname = hostname.strip().rstrip(".").lower()
    if not hostname:
        raise NormalizationError("missing hostname")
    try:
        return hostname.encode("idna").decode("ascii")
    except UnicodeError as exc:
        raise NormalizationError("invalid IDN hostname") from exc


def normalize_tracker_url(raw: str) -> NormalizedTracker:
    """Normalize a tracker URL into a canonical endpoint identity."""
    if raw is None:
        raise NormalizationError("empty input")
    text = raw.strip()
    if not text or text.startswith("#"):
        raise NormalizationError("empty input")

    # Reject credentials early even before parse edge-cases
    if "@" in text.split("://", 1)[-1].split("/", 1)[0]:
        raise NormalizationError("credentials not allowed")

    parsed = urlparse(text)
    scheme = (parsed.scheme or "").lower()
    if not scheme:
        raise NormalizationError("missing scheme")

    if scheme in UNSUPPORTED_SCHEMES or scheme.endswith(".i2p") or "ygg" in scheme:
        host = _idna_hostname(parsed.hostname or "unsupported")
        try:
            port = parsed.port or 0
        except ValueError as exc:
            raise NormalizationError("invalid port") from exc
        return NormalizedTracker(
            canonical_url=text,
            scheme=scheme,
            hostname=host,
            port=port,
            path=parsed.path or "/",
            unsupported=True,
        )

    if scheme not in SUPPORTED_SCHEMES:
        raise NormalizationError(f"unsupported scheme: {scheme}")

    if parsed.username or parsed.password:
        raise NormalizationError("credentials not allowed")

    if not parsed.hostname:
        raise NormalizationError("missing hostname")

    hostname = _idna_hostname(parsed.hostname)

    try:
        port = parsed.port
    except ValueError as exc:
        raise NormalizationError("invalid port") from exc
    if port is None:
        # Prefer explicit ports when provided by source; otherwise defaults.
        port = _default_port(scheme)
    if not (1 <= port <= 65535):
        raise NormalizationError("invalid port")

    path = parsed.path or "/announce"
    if not path.startswith("/"):
        path = "/" + path
    # Drop fragment; query is uncommon for announce base URLs — strip for identity.
    # Preserve path as given by source (including /announce variants).

    netloc = f"{hostname}:{port}"
    canonical = urlunparse((scheme, netloc, path, "", "", ""))
    return NormalizedTracker(
        canonical_url=canonical,
        scheme=scheme,
        hostname=hostname,
        port=port,
        path=path,
        unsupported=False,
    )


def parse_tracker_list_text(body: str) -> list[str]:
    """Split a blank-line-tolerant tracker list into raw URL strings (deduped)."""
    seen: set[str] = set()
    out: list[str] = []
    for line in body.splitlines():
        item = line.strip()
        if not item or item.startswith("#"):
            continue
        if item in seen:
            continue
        seen.add(item)
        out.append(item)
    return out