app/probes/http.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 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
from __future__ import annotations
import os
import time
import httpx
from app.probes.bencode import BencodeError, bdecode, validate_peers_then_discard
from app.probes.types import ProbeOutcome
from app.services.ssrf import assert_safe_destination, is_public_ip
async def probe_http(
url: str,
*,
client: httpx.AsyncClient,
timeout: float = 10.0,
max_bytes: int = 1_048_576,
peer_id_prefix: bytes = b"-RASTRO-",
info_hash: bytes | None = None,
resolved_ips: list[str] | None = None,
) -> ProbeOutcome:
info_hash = info_hash or os.urandom(20)
peer_id = (peer_id_prefix[:8] + os.urandom(12))[:20]
from urllib.parse import quote_from_bytes
query = (
f"info_hash={quote_from_bytes(info_hash)}"
f"&peer_id={quote_from_bytes(peer_id)}"
f"&port=0&uploaded=0&downloaded=0&left=0&compact=1&numwant=0&event=started"
)
announce_url = f"{url}?{query}" if "?" not in url else f"{url}&{query}"
t0 = time.perf_counter()
try:
if resolved_ips is not None:
host = httpx.URL(url).host or ""
assert_safe_destination(host, resolved_ips)
async with client.stream(
"GET",
announce_url,
timeout=timeout,
follow_redirects=False,
) as response:
# Re-validate resolved addresses before consuming the body (anti-rebinding).
if resolved_ips:
for ip in resolved_ips:
if not is_public_ip(ip):
return ProbeOutcome(
status="down",
response_valid=False,
error_kind="blocked",
error_detail=f"blocked address: {ip}",
latency_ms=(time.perf_counter() - t0) * 1000,
)
chunks: list[bytes] = []
total = 0
async for chunk in response.aiter_bytes():
total += len(chunk)
if total > max_bytes:
return ProbeOutcome(
status="down",
response_valid=False,
error_kind="too_large",
error_detail="response exceeded MAX_RESPONSE_BYTES",
latency_ms=(time.perf_counter() - t0) * 1000,
)
chunks.append(chunk)
body = b"".join(chunks)
latency = (time.perf_counter() - t0) * 1000
if response.status_code != 200:
return ProbeOutcome(
status="down",
response_valid=False,
latency_ms=latency,
error_kind="http_status",
error_detail=f"HTTP {response.status_code}",
)
if not body:
return ProbeOutcome(
status="down",
response_valid=False,
latency_ms=latency,
error_kind="empty",
error_detail="empty HTTP response",
)
try:
decoded = bdecode(body)
except BencodeError as exc:
return ProbeOutcome(
status="down",
response_valid=False,
latency_ms=latency,
error_kind="malformed",
error_detail=str(exc)[:512],
)
if "failure reason" in decoded:
return ProbeOutcome(
status="degraded",
response_valid=False,
latency_ms=latency,
error_kind="tracker_error",
error_detail=str(decoded["failure reason"])[:512],
)
# Require peers field OR interval for a minimal healthy announce
has_peers = validate_peers_then_discard(decoded)
interval = decoded.get("interval")
if not has_peers and interval is None:
return ProbeOutcome(
status="degraded",
response_valid=False,
latency_ms=latency,
error_kind="incomplete",
error_detail="missing peers and interval",
)
seeders = decoded.get("complete")
leechers = decoded.get("incomplete")
return ProbeOutcome(
status="up",
response_valid=True,
latency_ms=latency,
ip_address=resolved_ips[0] if resolved_ips else None,
ip_family="4"
if resolved_ips and ":" not in resolved_ips[0]
else ("6" if resolved_ips else None),
tracker_interval_seconds=int(interval) if isinstance(interval, int) else None,
seeders=int(seeders) if isinstance(seeders, int) else None,
leechers=int(leechers) if isinstance(leechers, int) else None,
)
except httpx.TimeoutException:
return ProbeOutcome(
status="down",
response_valid=False,
error_kind="timeout",
error_detail="HTTP timeout",
latency_ms=(time.perf_counter() - t0) * 1000,
)
except httpx.ConnectError as exc:
detail = str(exc)
kind = "tls" if "SSL" in detail.upper() or "TLS" in detail.upper() else "connection"
return ProbeOutcome(
status="down",
response_valid=False,
error_kind=kind,
error_detail=detail[:512],
latency_ms=(time.perf_counter() - t0) * 1000,
)
except ValueError as exc:
return ProbeOutcome(
status="down",
response_valid=False,
error_kind="blocked",
error_detail=str(exc)[:512],
latency_ms=(time.perf_counter() - t0) * 1000,
)
except httpx.HTTPError as exc:
return ProbeOutcome(
status="down",
response_valid=False,
error_kind="http",
error_detail=str(exc)[:512],
latency_ms=(time.perf_counter() - t0) * 1000,
)
|