siftlode/backend/app/downloads/urlguard.py

"""Refuse a user-supplied download URL that points back into our own network (SEC-12).

The Download Center deliberately accepts an arbitrary ``http(s)`` URL — that is the feature — and
hands it to yt-dlp running on the WORKER. Without a guard, any approved user can make the server
fetch anything the server can reach: other containers on the compose network, the LAN, a cloud
metadata endpoint. The failure text then comes back to them in ``job.error``, so "connection
refused" vs "HTTP Error 401" vs "timed out" turns the download box into an internal port scanner.

What this closes: a URL whose host resolves to a loopback / private / link-local / reserved address
is refused, at enqueue AND again in the worker before extraction. Both, because DNS can answer
differently between the two moments, and the worker is where the socket is actually opened.

What it does NOT close, stated plainly rather than implied, because both need the same fix — gating
the connection itself with a custom socket factory inside yt-dlp, a much larger change than this:

* DNS rebinding: a host that passes the worker's check and then answers yt-dlp's own resolution
  with an internal address.
* HTTP redirects: a public host that answers 3xx with an internal ``Location``. yt-dlp follows it,
  and only the URL the user TYPED was ever judged here. An attacker-controlled host or any public
  open redirect reaches it, so this is the cheaper of the two. It does NOT usually yield the
  response body — a non-media body fails the mp4 remux and stores nothing — but it does hand back
  the readout above: the redaction in ``errors.user_error_text`` fires only for the ``network``
  class, so a dead internal port (refused → ``network`` → redacted) is distinguishable from a live
  one (an HTTP answer that is not media → ``unsupported_host``/``format_unavailable``, raw text
  kept). Tracked as SEC-17.

The two checks here stop the straightforward case (typing ``http://127.0.0.1:6379`` or an internal
hostname into the box), which is the reachable one.

Not used for YouTube jobs: those resolve to a fixed public host from an 11-char id, so there is no
user-controlled destination to guard.
"""

import ipaddress
import socket
from typing import Literal
from urllib.parse import urlparse

# Reason codes travel to the client inside the structured 422 detail, so they are part of the API
# vocabulary — keep them stable. The UI names each one (`downloads.errors.urlBlocked.<reason>`, held by
# the frontend's backendIds contract, which reads this Literal).
UrlBlockedReason = Literal["scheme", "no_host", "malformed", "unresolvable", "private"]
REASON_SCHEME: UrlBlockedReason = "scheme"  # not http(s)
REASON_NO_HOST: UrlBlockedReason = "no_host"  # URL carries no hostname
REASON_MALFORMED: UrlBlockedReason = (
    "malformed"  # unparseable authority or port; urlparse raises late
)
REASON_UNRESOLVABLE: UrlBlockedReason = (
    "unresolvable"  # DNS said nothing (not the same as "blocked")
)
REASON_PRIVATE: UrlBlockedReason = (
    "private"  # resolves into a network the server must not fetch from
)


class BlockedUrl(Exception):
    """A user-supplied URL that must not be fetched server-side. ``reason`` is one of the
    REASON_* codes above."""

    def __init__(self, reason: UrlBlockedReason):
        super().__init__(reason)
        self.reason = reason


def _is_forbidden(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
    """Every address family we must not open a connection to. Checked explicitly rather than
    leaning on ``is_private`` alone, whose coverage differs between IPv4 and IPv6."""
    # ::ffff:127.0.0.1 is loopback wearing an IPv6 hat — unwrap before judging it.
    mapped = getattr(ip, "ipv4_mapped", None)
    if mapped is not None:
        ip = mapped
    return bool(
        ip.is_private
        or ip.is_loopback
        or ip.is_link_local
        or ip.is_multicast
        or ip.is_reserved
        or ip.is_unspecified
    )


def _resolve(host: str, port: int | None) -> list[str]:
    """Every address ``host`` resolves to. Raises BlockedUrl(REASON_UNRESOLVABLE) when DNS has no
    answer — reported as its own outcome, because "we could not look that up" and "we refuse to
    fetch that" are different things to tell a user."""
    try:
        infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
    except socket.gaierror as exc:
        raise BlockedUrl(REASON_UNRESOLVABLE) from exc
    return [info[4][0] for info in infos]


def check_url(raw: str) -> None:
    """Raise BlockedUrl unless ``raw`` is an http(s) URL whose host resolves only to addresses the
    server is allowed to fetch from. A host with even ONE forbidden address is refused: a
    round-robin answer must not become a coin flip."""
    # `urlparse` itself raises ValueError on a malformed authority ("http://[::1/x" — an unclosed
    # bracket), and `.port` raises on ":99999" / ":abc" when READ rather than at parse time. Both
    # escape the caller's `except BlockedUrl` as a 500 unless caught here, which is the opposite of
    # what this module exists to do: turn a bad URL into a structured, localizable refusal.
    try:
        u = urlparse((raw or "").strip())
        port = u.port
    except ValueError as exc:
        raise BlockedUrl(REASON_MALFORMED) from exc
    if u.scheme not in ("http", "https"):
        raise BlockedUrl(REASON_SCHEME)
    host = u.hostname
    if not host:
        raise BlockedUrl(REASON_NO_HOST)
    for addr in _resolve(host, port):
        try:
            ip = ipaddress.ip_address(addr)
        except ValueError:  # getaddrinfo returned something unparseable; treat as unusable
            raise BlockedUrl(REASON_PRIVATE) from None
        if _is_forbidden(ip):
            raise BlockedUrl(REASON_PRIVATE)