jira/apps/api/plane/utils/url_security.py

# Copyright (c) 2023-present Plane Software, Inc. and contributors
# SPDX-License-Identifier: AGPL-3.0-only
# See the LICENSE file for details.

"""
SSRF-safe outbound HTTP client.

The validators in :mod:`plane.utils.ip_address` resolve a hostname and confirm
that none of its addresses point at internal infrastructure. On their own they
are vulnerable to DNS rebinding (TOCTOU): the validator resolves the name, but
``requests`` resolves it a *second* time when it actually connects, and an
attacker who controls DNS can return a public IP to the validator and an
internal IP to the connection.

``pinned_fetch`` closes that window by resolving + validating once and then
connecting to the *validated IP literal* — urllib3 performs no second DNS
lookup, so the address that was checked is exactly the address that is reached.
The original hostname is still used for the ``Host`` header, TLS SNI and
certificate verification, so virtual-hosting and HTTPS continue to work.

Redirects are never auto-followed (``requests`` would re-resolve each hop and
reopen the rebinding window, and a ``Location`` can point at a new internal
host). ``pinned_fetch_following_redirects`` follows them manually, re-resolving,
re-validating and re-pinning every hop.
"""

# Python imports
import ipaddress
from urllib.parse import unquote, urljoin, urlsplit

# Third party imports
import requests
from requests.adapters import HTTPAdapter

# Module imports
from plane.utils.ip_address import resolve_and_validate

# 3xx status codes that carry a Location we may follow.
_REDIRECT_STATUSES = {301, 302, 303, 307, 308}

# Never route through an ambient proxy — a CONNECT to a proxy would tunnel to
# the original hostname and bypass the IP pinning entirely.
_NO_PROXIES = {"http": None, "https": None}


class PinnedIPAdapter(HTTPAdapter):
    """
    A ``requests`` transport adapter that connects to whatever IP literal is in
    the request URL while presenting ``server_hostname`` for TLS SNI and
    certificate verification.

    The IP literal in the URL means urllib3 opens the socket to that exact IP
    with no DNS resolution. Injecting ``server_hostname`` (and leaving
    ``assert_hostname`` at its ``None`` default so ``SSLContext.check_hostname``
    stays ``True``) makes the standard library verify the presented certificate
    against the real hostname rather than the IP.

    Instances hold no global state — one is mounted on a throwaway
    :class:`requests.Session` per request, so this is safe under any Celery pool
    (prefork / threads / gevent).
    """

    def __init__(self, server_hostname, *args, **kwargs):
        self._server_hostname = server_hostname
        super().__init__(*args, **kwargs)

    def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
        # requests >= 2.32 calls this (it replaced get_connection() as part of
        # the CVE-2024-35195 fix). requests is pinned to 2.33 in base.txt.
        host_params, pool_kwargs = self.build_connection_pool_key_attributes(
            request, verify, cert
        )
        # server_hostname is a recognised urllib3 SSL pool-key field, so pools
        # for different hostnames don't collide.
        pool_kwargs["server_hostname"] = self._server_hostname
        return self.poolmanager.connection_from_host(**host_params, pool_kwargs=pool_kwargs)


def _split_target(url):
    """Parse a URL into the pieces needed to build a pinned request.

    Returns ``(scheme, hostname, port, path, auth)`` where ``auth`` carries any
    URL-embedded credentials (``user:pass@host``) as a ``(user, pass)`` tuple so
    HTTP Basic Auth still works once the URL is rewritten to an IP literal.
    """
    parts = urlsplit(url)
    scheme = parts.scheme
    if scheme not in ("http", "https"):
        raise ValueError("Invalid URL scheme. Only HTTP and HTTPS are allowed")
    hostname = parts.hostname
    if not hostname:
        raise ValueError("Invalid URL: No hostname found")
    port = parts.port or (443 if scheme == "https" else 80)
    path = parts.path or "/"
    if parts.query:
        path = f"{path}?{parts.query}"
    auth = None
    if parts.username is not None or parts.password is not None:
        auth = (unquote(parts.username or ""), unquote(parts.password or ""))
    return scheme, hostname, port, path, auth


def _request_to_ip(method, scheme, hostname, ip, port, path, *, headers, timeout, auth=None, **kwargs):
    """Issue a single request whose socket is pinned to ``ip``.

    With ``stream=True`` the session is kept open until the caller closes the
    response (closing the response also closes the session), so a streamed body
    can be read with a real size cap; otherwise the session is closed eagerly.
    """
    ip_obj = ipaddress.ip_address(ip)
    host_for_url = f"[{ip}]" if ip_obj.version == 6 else ip
    url = f"{scheme}://{host_for_url}:{port}{path}"

    request_headers = dict(headers or {})
    default_port = 443 if scheme == "https" else 80
    # Host header (and TLS) carry the ORIGINAL hostname, not the IP literal.
    # An IPv6-literal hostname must be bracketed in the Host header.
    host_label = f"[{hostname}]" if ":" in hostname else hostname
    request_headers["Host"] = host_label if port == default_port else f"{host_label}:{port}"

    session = requests.Session()
    session.trust_env = False  # ignore ambient proxy / netrc / env (see _NO_PROXIES)
    if scheme == "https":
        session.mount("https://", PinnedIPAdapter(server_hostname=hostname))

    try:
        response = session.request(
            method,
            url,
            headers=request_headers,
            timeout=timeout,
            allow_redirects=False,
            verify=True,
            proxies=_NO_PROXIES,
            auth=auth,
            **kwargs,
        )
    except BaseException:
        session.close()
        raise

    if kwargs.get("stream"):
        # Defer closing the session until the response is closed, so the
        # streamed body remains readable. response.close() now also closes
        # the session.
        _orig_close = response.close

        def _close_all(_orig=_orig_close, _sess=session):
            try:
                _orig()
            finally:
                _sess.close()

        response.close = _close_all
    else:
        session.close()
    return response


def _fetch_validated_hop(method, url, *, allowed_ips, allowed_hosts, headers, timeout, **kwargs):
    """
    Resolve ``url``'s host, validate it, then issue a single (non-redirecting)
    request pinned to a resolved IP. Returns ``(response, normalized_host)``.

    Hosts in ``allowed_hosts`` are operator-trusted (e.g. internal service DNS
    whose IPs are dynamic): they skip the private-IP *block* check, but the
    connection is STILL pinned to the resolved IP so a trusted hostname cannot
    be rebound to a different internal target between validation and connect.
    """
    scheme, hostname, port, path, auth = _split_target(url)

    normalized_host = hostname.rstrip(".").lower()
    trusted = bool(allowed_hosts) and normalized_host in {
        (h or "").rstrip(".").lower() for h in allowed_hosts if h
    }

    # Resolve once (and validate unless the host is operator-trusted), then pin
    # the connection to a resolved IP literal — urllib3 performs no second DNS
    # lookup, so the address validated here is exactly the one reached.
    ips = resolve_and_validate(hostname, allowed_ips=allowed_ips, require_safe=not trusted)

    last_exc = None
    for ip in ips:
        try:
            response = _request_to_ip(
                method, scheme, hostname, ip, port, path,
                headers=headers, timeout=timeout, auth=auth, **kwargs,
            )
            return response, normalized_host
        except requests.RequestException as exc:
            # Try the next resolved address (dual-stack / round-robin hosts).
            last_exc = exc
    if last_exc is not None:
        raise last_exc
    raise requests.ConnectionError(f"No reachable address for host: {hostname}")


def pinned_fetch(
    method,
    url,
    *,
    allowed_ips=None,
    allowed_hosts=None,
    headers=None,
    timeout=30,
    **kwargs,
):
    """
    SSRF-safe single request. Resolves + validates the target host and pins the
    connection to a validated IP (defeating DNS rebinding). Does NOT follow
    redirects.

    Raises:
        ValueError: if the URL is invalid or resolves to a blocked address.
        requests.RequestException: on network/transport errors.
    """
    response, _ = _fetch_validated_hop(
        method, url,
        allowed_ips=allowed_ips, allowed_hosts=allowed_hosts,
        headers=headers, timeout=timeout, **kwargs,
    )
    return response


def pinned_fetch_following_redirects(
    method,
    url,
    *,
    allowed_ips=None,
    allowed_hosts=None,
    headers=None,
    timeout=30,
    max_redirects=5,
    **kwargs,
):
    """
    SSRF-safe request that follows redirects manually, re-resolving,
    re-validating and re-pinning every hop. Returns ``(response, final_url)``.

    Raises:
        ValueError: if any URL in the chain is invalid or resolves to a blocked
            address.
        requests.TooManyRedirects: if the hop limit is exceeded.
        requests.RequestException: on network/transport errors.
    """
    current_url = url
    redirects = 0
    while True:
        response, _ = _fetch_validated_hop(
            method, current_url,
            allowed_ips=allowed_ips, allowed_hosts=allowed_hosts,
            headers=headers, timeout=timeout, **kwargs,
        )

        if response.status_code not in _REDIRECT_STATUSES:
            return response, current_url

        location = response.headers.get("Location")
        if not location:
            return response, current_url

        if redirects >= max_redirects:
            response.close()
            raise requests.TooManyRedirects(
                f"Exceeded {max_redirects} redirects for URL: {url}"
            )
        redirects += 1
        # Release the intermediate hop's connection/session before following.
        response.close()
        # Resolve the redirect target against the current URL; the next loop
        # iteration re-validates and re-pins it.
        current_url = urljoin(current_url, location)