"""HTTP transport with Tidal's request policy.
One :class:`Transport` per client instance — no shared sessions, no
globals. The policy implemented here (and *only* here) is what Tidal's
APIs actually demand, learned from the reference clients:
- catalog/auth calls send a **bare** ``Authorization: Bearer`` header
(extra device fingerprint headers are treated as suspect);
- HTTP 429 honours ``Retry-After`` (delta-seconds *and* HTTP-date) and
never consumes the transient-retry budget; the total *time* spent
waiting is bounded per logical request (short waits by
:data:`Config.rate_limit_max_wait`, long server-requested pauses by
:data:`Config.rate_limit_max_pause`), then
:class:`~tidalpy.errors.RateLimitError`. A 429 is an account-level
signal, so it also sets a **client-wide pause**: every catalog
request on this transport waits out the server-requested cooldown
instead of firing straight into its own 429 (reactive — a healthy
run never pauses);
- HTTP 401 with subStatus 4005 means "asset not ready" (bounded
backoff); any other 401 triggers exactly one token refresh;
- HTTP 404 subStatus 4022 means the issuing OAuth client was retired
(catalog/auth endpoints) — the saved session is cleared and the
host must log in again; on *playback* endpoints the same subStatus
is a playback-client rejection (this client/account/quality may be
unavailable for playback) — the session is kept and the error
stays actionable;
- HTTP 403 with a "not available in your country" message is a region
lock (distinct, non-retryable error);
- 5xx responses (gateway 502/503 during Tidal deploys) are retried
within the transient-retry budget, with half-to-full jitter.
Media URLs handed to us by remote manifests are additionally validated
(:func:`validate_media_url`) against SSRF: no credentials, no local
hosts, no private address space.
"""
from __future__ import annotations
import ipaddress
import logging
import random
import socket
import threading
import time
from collections.abc import Mapping
from datetime import timezone
from email.utils import parsedate_to_datetime
from typing import Any, Callable, Protocol
from urllib.parse import urljoin, urlsplit
import requests
from requests.adapters import HTTPAdapter
from .config import Config
from .errors import (
AuthError,
CancelledError,
DownloadError,
NotFoundError,
RateLimitError,
RegionLockedError,
TidalApiError,
)
logger = logging.getLogger("tidalpy.transport")
API_BASE = "https://api.tidal.com/v1/"
API_BASE_LEGACY = "https://api.tidalhifi.com/v1/"
OPENAPI_BASE = "https://openapi.tidal.com/v2/"
AUTH_BASE = "https://auth.tidal.com/v1/oauth2"
# subStatus: asset exists but is not ready for playback yet (fresh upload).
ASSET_NOT_READY_SUB_STATUS = 4005
# subStatus: the OAuth client that issued the session no longer exists.
STALE_CLIENT_SUB_STATUS = 4022
_DEFAULT_RATE_WAIT_SECONDS = 20.0
_DNS_CACHE_SECONDS = 60
# Manual redirect-following cap for small binary fetches (cover art).
_MAX_REDIRECT_HOPS = 5
_dns_cache: dict[str, tuple[float, tuple[str, ...]]] = {}
_dns_cache_lock = threading.Lock()
REGION_LOCK_MARKERS = ("not available in your country", "region", "geo")
[docs]
class AuthProvider(Protocol):
"""What the transport needs from the OAuth layer (implemented by
:class:`tidalpy.oauth.OAuth`)."""
[docs]
def access_token(self) -> str | None: ...
[docs]
def user_agent(self) -> str: ...
[docs]
def country_code(self) -> str | None: ...
[docs]
def handle_401(self) -> bool:
"""Attempt a one-shot token refresh; True when the token changed."""
[docs]
def clear_saved_session(self) -> None: ...
[docs]
def retry_after_seconds(response: requests.Response | None) -> float | None:
"""Wait requested by a ``Retry-After`` header, in seconds (or None).
Supports both the delta-seconds and HTTP-date forms.
"""
if response is None:
return None
raw = response.headers.get("Retry-After")
if not raw:
return None
raw = str(raw).strip()
try:
value = float(raw)
except ValueError:
try:
when = parsedate_to_datetime(raw)
if when.tzinfo is None:
when = when.replace(tzinfo=timezone.utc)
value = max(0.0, when.timestamp() - time.time())
except (TypeError, ValueError, OverflowError, OSError):
return None
if not _is_finite(value) or value < 0:
return None
return value
def _is_finite(value: float) -> bool:
return value == value and value not in (float("inf"), float("-inf"))
class _RateLimitBudget:
"""Cumulative 429 back-off allowance for one logical request.
Rate-limited responses are retried without consuming the normal
attempt budget; instead the total *time* slept is capped, so a
persistently throttled client eventually fails instead of hanging
forever (reference-fork parity: a count cap would give up while
an account-level limit still had a minute to clear).
Short waits draw from ``max_wait``. A *long* server-requested
pause (``Retry-After`` beyond ``max_wait``) draws from ``max_pause``
separately — honouring it is right, since Tidal answers every
request 429 until the pause lapses and failing one request would
fail every following one too.
"""
def __init__(self, max_wait: float, max_pause: float) -> None:
self.max_wait = max_wait
self.max_pause = max_pause
self.waited = 0.0
self.paused = 0.0
self.waits = 0
def allows_wait(self, seconds: float) -> bool:
return (
_is_finite(seconds)
and 0 < seconds <= self.max_wait
and self.waited + seconds <= self.max_wait
)
def allows_pause(self, seconds: float) -> bool:
return (
_is_finite(seconds)
and seconds > 0
and self.paused + seconds <= self.max_pause
)
def record_wait(self, seconds: float) -> None:
self.waited += seconds
self.waits += 1
def record_pause(self, seconds: float) -> None:
self.paused += seconds
[docs]
def response_facts(body: Any) -> tuple[Any, Any, str | None]:
"""``(status, subStatus, userMessage)`` from a Tidal error envelope."""
if not isinstance(body, dict):
return None, None, None
status = body.get("status")
sub = body.get("subStatus", body.get("sub_status"))
message = body.get("userMessage") or body.get("error")
if isinstance(message, str) and message:
return status, sub, message
return status, sub, None
[docs]
def response_error_codes(body: Any) -> list[str]:
"""Machine codes a Tidal error body carries (or ``[]``).
JSON:API-style ``errors[].code`` entries plus a top-level ``code``.
Callers use these to tell an *account-capability* rejection (e.g.
``CLIENT_NOT_ENTITLED`` — true for every track) from a
track-specific block (must not taint other tracks).
"""
codes: list[str] = []
if not isinstance(body, dict):
return codes
errors = body.get("errors")
if isinstance(errors, list):
for item in errors:
if isinstance(item, dict) and item.get("code"):
codes.append(str(item["code"]))
if body.get("code"):
codes.append(str(body["code"]))
return codes
def _looks_region_locked(message: str | None) -> bool:
if not message:
return False
lowered = message.lower()
return any(marker in lowered for marker in REGION_LOCK_MARKERS)
[docs]
class Transport:
"""A private HTTP session with the Tidal request policy applied."""
[docs]
def __init__(
self,
config: Config,
stop_event: threading.Event | None = None,
clock: Callable[[], float] | None = None,
sleeper: Callable[[float], None] | None = None,
media_resolver: Callable[[str], tuple[str, ...]] | None = None,
) -> None:
self._config = config
self._stop = stop_event or threading.Event()
self._clock = clock or time.monotonic
self._sleep = sleeper or self._sleep_interruptible
# SSRF guard for redirect hops (a test hook may inject a fake
# resolver to avoid DNS; None = the caching default).
self._resolver = media_resolver
self._auth: AuthProvider | None = None
# Client-wide rate pause (monotonic deadline, set by a 429 on
# any catalog request): sibling requests of a concurrent run
# wait it out instead of each earning their own 429.
self._rate_lock = threading.Lock()
self._rate_pause_until = 0.0
self._session = requests.Session()
# The pool must cover every concurrent media worker plus a few
# catalog calls; too small and workers serialize on sockets.
pool = max(10, config.download_workers)
adapter = HTTPAdapter(
pool_connections=pool, pool_maxsize=max(pool, config.download_workers * 2)
)
self._session.mount("https://", adapter)
self._session.mount("http://", adapter)
# No implicit environment state: requests' default (trust_env)
# would merge ~/.netrc credentials, HTTP(S)_PROXY environment
# variables, and REQUESTS_CA_BUNDLE into every request —
# global state this library's contract forbids. All of that
# policy arrives explicitly via Config (proxies, verify_tls),
# so a media fetch can never pick up credentials meant for
# some other service on the host.
self._session.trust_env = False
proxies = config.proxies_dict()
if proxies:
self._session.proxies.update(proxies)
# TLS verification override (corporate MITM proxies ship their own
# CA bundle); None = requests' default (verify against certifi).
verify = config.verify_tls
if verify is not None:
self._session.verify = verify
# requests only accepts a (connect, read) tuple.
self._timeouts = (config.timeout_connect, config.timeout_read)
# Base URLs (overridable via Config for proxies/test servers).
self._api_base = config.api_base or API_BASE
self._legacy_api_base = config.legacy_api_base or API_BASE_LEGACY
self._openapi_base = config.openapi_base or OPENAPI_BASE
self._auth_base = config.auth_base or AUTH_BASE
@property
def api_base(self) -> str:
return self._api_base
@property
def legacy_api_base(self) -> str:
return self._legacy_api_base
@property
def openapi_base(self) -> str:
return self._openapi_base
@property
def auth_base(self) -> str:
return self._auth_base
# -- lifecycle ---------------------------------------------------------
[docs]
def bind_auth(self, auth: AuthProvider) -> None:
self._auth = auth
[docs]
def set_stop_event(self, stop_event: threading.Event) -> None:
"""Bind the cancellation event for the in-flight operation.
One operation at a time per transport (i.e. per client); the
client rebinds this at the start of every download/stream.
"""
self._stop = stop_event
@property
def stop_event(self) -> threading.Event:
return self._stop
[docs]
def close(self) -> None:
self._session.close()
def __enter__(self) -> "Transport":
return self
def __exit__(self, *exc_info) -> None:
self.close()
# -- auth hooks ---------------------------------------------------------
def _check_stop(self) -> None:
if self._stop.is_set():
raise CancelledError("operation cancelled")
def _sleep_interruptible(self, seconds: float) -> None:
deadline = self._clock() + max(0.0, seconds)
while True:
self._check_stop()
remaining = deadline - self._clock()
if remaining <= 0:
return
time.sleep(min(1.0, remaining))
@staticmethod
def _with_jitter(backoff: float) -> float:
"""Half-to-full jitter: spreads retry storms off the same phase."""
return random.uniform(backoff / 2, backoff)
# -- catalog / openapi GET -------------------------------------------------
[docs]
def get_json(
self,
path: str,
params: dict | list[tuple[str, str]] | None = None,
base: str | None = None,
accept: str | None = None,
) -> dict:
"""GET a JSON API endpoint with the full Tidal policy applied.
``base`` defaults to the v1 API base (Config-overridable).
"""
if base is None:
base = self._api_base
if self._auth is None:
token = ""
user_agent = "tidalpy"
country = None
else:
token = self._auth.access_token() or ""
user_agent = self._auth.user_agent()
country = self._auth.country_code()
headers = {
"Authorization": f"Bearer {token}",
"User-Agent": user_agent,
"Accept": accept or "application/json",
}
# params may be a dict or a list of tuples (repeated keys, e.g.
# OpenAPI ?formats=FLAC_HIRES&formats=FLAC).
if isinstance(params, dict):
merged_params: dict | list = dict(params)
else:
merged_params = list(params or [])
if country:
if isinstance(merged_params, dict):
merged_params["countryCode"] = country
else:
merged_params.append(("countryCode", country))
url = base + path
max_retries = max(0, self._config.max_retries)
attempts = 0
refreshed = False
rate_budget = _RateLimitBudget(
self._config.rate_limit_max_wait, self._config.rate_limit_max_pause
)
asset_waits = 0
while True:
self._check_stop()
self._wait_rate_pause()
try:
response = self._session.get(
url,
headers=headers,
params=merged_params or None,
timeout=self._timeouts,
allow_redirects=False,
)
except requests.RequestException as exc:
attempts += 1
if attempts > max_retries:
raise DownloadError(f"request to Tidal failed: {exc}") from exc
backoff = self._with_jitter(
self._config.retry_backoff * (2 ** (attempts - 1))
)
logger.debug(
"transient request error (%s); retrying in %.1fs", exc, backoff
)
self._sleep(backoff)
continue
with response:
if response.status_code == 429:
requested = retry_after_seconds(response)
if (
requested is not None
and requested > self._config.rate_limit_max_wait
):
# A long, server-requested pause: honour it —
# failing instead would fail every following
# request too, since Tidal answers each one
# 429 until the pause lapses.
if not rate_budget.allows_pause(requested):
raise RateLimitError(
"Tidal rate limit (HTTP 429) asked for a "
f"pause beyond the "
f"{self._config.rate_limit_max_pause:.0f}s budget; "
"try again later"
)
rate_budget.record_pause(requested)
delay = requested
else:
delay = requested or _DEFAULT_RATE_WAIT_SECONDS
if not rate_budget.allows_wait(delay):
raise RateLimitError(
"Tidal rate limit (HTTP 429) persisted "
f"after {rate_budget.waited:.0f}s of waiting "
f"(budget {self._config.rate_limit_max_wait:.0f}s); "
"try again later"
)
rate_budget.record_wait(delay)
# Pause the whole client, not just this request: a
# concurrent run would fire its siblings straight into
# more 429s otherwise (reference-fork parity: one 429
# holds every request on the client back).
self._extend_rate_pause(delay)
if delay > 30:
logger.info(
"rate limited by Tidal; pausing %.0fs (client-wide)", delay
)
self._sleep(delay)
continue
if response.status_code == 401:
_, sub, message = response_facts(self._body(response))
if sub == ASSET_NOT_READY_SUB_STATUS:
asset_waits += 1
if asset_waits > self._config.asset_not_ready_max_waits:
raise TidalApiError(
"asset is not ready for playback yet "
f"(HTTP 401, subStatus {sub}): "
f"{message or 'still processing'}",
401,
sub,
)
delay = min(5 * asset_waits, 30)
logger.debug(
"asset not ready; retrying in %ss (%s/%s)",
delay,
asset_waits,
self._config.asset_not_ready_max_waits,
)
self._sleep(delay)
continue
if (
self._auth is not None
and not refreshed
and self._auth.handle_401()
):
refreshed = True
if self._auth.access_token():
headers["Authorization"] = (
f"Bearer {self._auth.access_token()}"
)
continue
hint = message or "the saved login is no longer valid; log in again"
raise AuthError(
f"authentication rejected (HTTP 401, subStatus {sub}): {hint}",
status_code=401,
sub_status=sub,
)
if response.status_code == 404:
_, sub, message = response_facts(self._body(response))
if sub == STALE_CLIENT_SUB_STATUS:
if "playbackinfo" in path:
# A 4022 on a *playback* endpoint is a
# playback-client rejection, not proof the
# login went stale: this client, account, or
# quality may simply be unavailable for
# playback. The session must survive —
# clearing it would log the host out of a
# perfectly good account (reference-fork
# live case).
quality = str(
dict(params or {}).get("audioquality") or "unknown"
)
country = (
self._auth.country_code()
if self._auth is not None
else None
)
raise TidalApiError(
"playback client rejected (HTTP 404, "
"subStatus 4022; "
f"country={country or 'unknown'}; "
f"quality={quality}): the saved login was "
"kept — this client, account, or quality "
"may be unavailable for playback; try a "
"lower quality",
status_code=404,
sub_status=sub,
)
if self._auth is not None:
# Before clearing, try one refresh with the
# *active* client (reference-fork parity,
# tidekeeper 4022 ladder: refresh once, retry
# once, only then clear): the token may be
# stale for reasons a fresh token fixes, and
# a second 4022 after a successful refresh is
# what actually proves the client is dead.
if not refreshed and self._auth.handle_401():
refreshed = True
if self._auth.access_token():
headers["Authorization"] = (
f"Bearer {self._auth.access_token()}"
)
continue
self._auth.clear_saved_session()
raise AuthError(
"the saved login references an OAuth client that no "
"longer exists (HTTP 404, subStatus 4022); the session "
"was cleared, please log in again",
status_code=404,
sub_status=sub,
)
raise NotFoundError(
f"resource not found (HTTP 404): {message or 'unknown id'}",
status_code=404,
sub_status=sub,
)
if response.status_code == 403:
body = self._body(response)
_, sub, message = response_facts(body)
if _looks_region_locked(message):
raise RegionLockedError(
"content not available in your country: "
f"{message or 'region locked'}",
status_code=403,
sub_status=sub,
)
raise TidalApiError(
f"request blocked (HTTP 403): {message or 'forbidden'}",
403,
sub,
codes=response_error_codes(body),
)
if 500 <= response.status_code < 600:
# Server-side blips (gateway 502/503 during deploys, 500
# on hot keys) usually clear within seconds: retry within
# the same budget as transient request errors.
attempts += 1
if attempts > max_retries:
_, sub, message = response_facts(self._body(response))
raise TidalApiError(
f"request failed (HTTP {response.status_code}) after "
f"{attempts} attempts: "
f"{message or (response.text or 'empty body')[:200]}",
response.status_code,
sub,
)
backoff = self._with_jitter(
self._config.retry_backoff * (2 ** (attempts - 1))
)
logger.debug(
"server error (HTTP %s); retrying in %.1fs",
response.status_code,
backoff,
)
self._sleep(backoff)
continue
if response.status_code != 200:
_, sub, message = response_facts(self._body(response))
raise TidalApiError(
f"request failed (HTTP {response.status_code}): "
f"{message or (response.text or 'empty body')[:200]}",
response.status_code,
sub,
)
body = self._body(response)
if body is None:
attempts += 1
if attempts > max_retries:
raise DownloadError(
f"request returned no JSON (HTTP {response.status_code})"
)
continue
if not isinstance(body, dict):
return body # type: ignore[no-any-return]
# v1 error envelope: a present *numeric* non-200 status is
# an application error. String statuses are domain data, not
# errors — the subscription endpoint answers 200 with
# ``{"status": "ACTIVE", ...}`` and must not be misread.
status = body.get("status")
if isinstance(status, int) and status != 200:
_, sub, message = response_facts(body)
raise TidalApiError(
f"Tidal API error (status {status}): "
f"{message or 'unknown error'}",
status,
sub,
)
return body
def _body(self, response: requests.Response) -> Any:
text = (response.text or "").strip()
if not text:
return None
try:
return response.json()
except ValueError:
return None
def _extend_rate_pause(self, seconds: float) -> None:
"""Hold every catalog request on this client back until the
server-requested cooldown ends (a 429 is an account-level
signal, not a per-request one)."""
deadline = self._clock() + max(0.0, seconds)
with self._rate_lock:
if deadline > self._rate_pause_until:
self._rate_pause_until = deadline
def _wait_rate_pause(self) -> None:
"""Wait out any client-wide rate pause before firing a request.
Reactive by design: the pause only exists after a 429, so a
healthy run never pays for it (and it lapses of its own)."""
with self._rate_lock:
remaining = self._rate_pause_until - self._clock()
if remaining <= 0:
return
logger.debug("waiting out client-wide rate pause (%.1fs)", remaining)
self._sleep(remaining)
# -- auth POST -------------------------------------------------------------
[docs]
def post_json(self, path: str, data: dict, base: str | None = None) -> dict:
"""POST form data to the OAuth endpoint (bounded attempts)."""
if base is None:
base = self._auth_base
user_agent = self._auth.user_agent() if self._auth else "tidalpy"
headers = {
"Accept": "application/json",
"User-Agent": user_agent,
"Content-Type": "application/x-www-form-urlencoded",
}
last_error = "auth request failed"
for attempt in range(3):
self._check_stop()
try:
response = self._session.post(
base + path,
data=data,
headers=headers,
timeout=(self._config.timeout_connect, self._config.timeout_read),
allow_redirects=False,
)
except requests.RequestException as exc:
last_error = str(exc)
if attempt == 2:
raise DownloadError(f"auth request failed: {exc}") from exc
continue
with response:
text = (response.text or "").strip()
if not text:
last_error = f"empty auth response (HTTP {response.status_code})"
continue
try:
body = response.json()
except ValueError:
last_error = (
f"invalid auth JSON (HTTP {response.status_code}): {text[:200]}"
)
continue
if isinstance(body, dict):
return body
raise DownloadError(last_error)
[docs]
def revoke_session(self, access_token: str) -> bool:
"""Best-effort remote session revocation (Tidal's ``/v1/logout``).
Never raises — a failed revocation reports ``False`` (and logs);
the caller still forgets the token locally. ``401`` counts as
success: the token no longer works, which is the goal.
"""
try:
with self._session.post(
self._api_base + "logout",
headers={"Authorization": f"Bearer {access_token}"},
timeout=self._timeouts,
allow_redirects=False,
) as response:
if response.status_code not in (200, 204, 401):
logger.warning(
"remote session revocation returned HTTP %s",
response.status_code,
)
return False
return True
except requests.RequestException as exc:
logger.warning(
"unable to revoke the Tidal session remotely: %s", type(exc).__name__
)
return False
# -- media (unauthenticated CDN) ---------------------------------------------
@staticmethod
def _encoded(headers: Mapping[str, Any]) -> bool:
"""True when the body is transfer-encoded (gzip etc.).
An encoded response advertises the *compressed* size — smaller
than the object — and trusting that length would make a healthy
local file look truncated (spurious re-download). ``identity``
(or an absent header) is safe.
"""
encoding = (headers.get("Content-Encoding") or "").strip().lower()
return bool(encoding) and encoding != "identity"
[docs]
def probe_size(self, url: str) -> int:
"""Remote size of a media URL, or ``-1`` when it cannot be determined.
A failed probe must never be interpreted as "file changed" —
callers fall back to trusting the local file (legacy bug fix:
one dead HEAD used to re-download whole playlists). The probe
requests ``identity`` encoding and distrusts encoded lengths
(reference-fork parity: a compressed length would mis-size the
file).
"""
try:
with self._session.head(
url,
headers={"Accept-Encoding": "identity"},
timeout=(self._config.timeout_connect, 30),
) as response:
if (
response.status_code == 200
and "Content-Length" in response.headers
and not self._encoded(response.headers)
):
return int(response.headers["Content-Length"])
except (requests.RequestException, ValueError):
pass
try:
with self._session.get(
url,
headers={"Range": "bytes=0-0", "Accept-Encoding": "identity"},
timeout=(self._config.timeout_connect, 30),
) as response:
if response.status_code == 206 and not self._encoded(response.headers):
content_range = response.headers.get("Content-Range", "")
if "/" in content_range:
total = content_range.split("/")[-1]
if total.isdigit():
return int(total)
if (
response.status_code == 200
and "Content-Length" in response.headers
and not self._encoded(response.headers)
):
return int(response.headers["Content-Length"])
except (requests.RequestException, ValueError):
pass
return -1
[docs]
def fetch(self, url: str, headers: dict | None = None) -> requests.Response:
"""Open a media URL (already SSRF-validated by the caller).
Connect-phase failures (DNS blip, dropped idle connection, CDN
reset before the first byte) are retried with backoff — an
exception from ``get(stream=True)`` means no body was consumed,
so a retry cannot corrupt a partial write. Mid-stream failures
surface later in ``iter_content`` and are the caller's policy.
"""
max_attempts = max(1, min(3, self._config.max_retries + 1))
last_error: Exception | None = None
for attempt in range(max_attempts):
self._check_stop()
try:
return self._session.get(
url,
headers=headers,
timeout=(
self._config.timeout_connect,
self._config.timeout_read,
),
stream=True,
)
except requests.RequestException as exc:
last_error = exc
if attempt == max_attempts - 1:
break
backoff = self._with_jitter(self._config.retry_backoff * (2**attempt))
logger.debug(
"media connect error (%s); retry %d in %.1fs",
exc,
attempt + 1,
backoff,
)
self._sleep(backoff)
raise DownloadError(
f"media fetch failed after {max_attempts} attempts: " f"{last_error}"
) from last_error
[docs]
def get_bytes(self, url: str, headers: dict | None = None) -> bytes:
"""Download a small binary (cover art) in full.
Redirects are followed *manually* (requests runs with
``allow_redirects=False``) and every hop is re-validated: the
``Location`` header is remote-supplied input, and following it
without the SSRF guard would let a public image URL redirect to
an internal host (reference-fork pin: validated redirects).
"""
current = url
for _hop in range(_MAX_REDIRECT_HOPS):
response = self.fetch(current, headers=headers)
try:
status = response.status_code
if status in (301, 302, 303, 307, 308):
location = response.headers.get("Location")
if not location:
return b""
if urlsplit(location).netloc == "":
# Relative Location: resolve against the current URL.
location = urljoin(current, location)
if self._config.validate_media_urls:
try:
validate_media_url(location, resolver=self._resolver)
except DownloadError as exc:
# A hop the SSRF guard refuses (an internal
# redirect target) stops the follow: the
# cover is garnish, the refusal is correct.
logger.debug("cover redirect hop refused: %s", exc)
return b""
current = location
continue
if status != 200:
return b""
return response.content
finally:
response.close()
logger.debug("cover fetch: %d redirects exceeded the cap", _MAX_REDIRECT_HOPS)
return b""
# ---------------------------------------------------------------------------
# Media URL validation (SSRF guard)
# ---------------------------------------------------------------------------
def _default_resolver(hostname: str) -> tuple[str, ...]:
now = time.monotonic()
with _dns_cache_lock:
cached = _dns_cache.get(hostname)
if cached is not None and now - cached[0] <= _DNS_CACHE_SECONDS:
return cached[1]
records = socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
addresses = tuple(
dict.fromkeys(str(record[4][0]).split("%", 1)[0] for record in records)
)
if not addresses:
raise DownloadError(f"media host {hostname!r} did not resolve to an address")
for address in addresses:
try:
ip = ipaddress.ip_address(address)
except ValueError:
raise DownloadError(f"media host {hostname!r} resolved to a bad address")
if not ip.is_global:
raise DownloadError(
f"media host {hostname!r} resolves to a private/local address "
f"({address}); refusing to fetch"
)
with _dns_cache_lock:
_dns_cache[hostname] = (now, addresses)
return addresses