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