reflex_copy/reflex/utils/net.py

"""Helpers for downloading files from the network."""

import functools
import time
from collections.abc import Callable
from typing import ParamSpec, TypeVar

import httpx

from reflex.utils.decorator import once

from . import console


def _httpx_verify_kwarg() -> bool:
    """Get the value of the HTTPX verify keyword argument.

    Returns:
        True if SSL verification is enabled, False otherwise
    """
    from ..config import environment

    return not environment.SSL_NO_VERIFY.get()


_P = ParamSpec("_P")
_T = TypeVar("_T")


def _wrap_https_func(
    func: Callable[_P, _T],
) -> Callable[_P, _T]:
    """Wrap an HTTPS function with logging.

    Args:
        func: The function to wrap.

    Returns:
        The wrapped function.
    """

    @functools.wraps(func)
    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
        url = args[0]
        console.debug(f"Sending HTTPS request to {args[0]}")
        initial_time = time.time()
        try:
            response = func(*args, **kwargs)
        except httpx.ConnectError as err:
            if "CERTIFICATE_VERIFY_FAILED" in str(err):
                # If the error is a certificate verification error, recommend mitigating steps.
                console.error(
                    f"Certificate verification failed for {url}. Set environment variable SSL_CERT_FILE to the "
                    "path of the certificate file or SSL_NO_VERIFY=1 to disable verification."
                )
            raise
        else:
            console.debug(
                f"Received response from {url} in {time.time() - initial_time:.3f} seconds"
            )
            return response

    return wrapper


def _is_ipv4_supported() -> bool:
    """Determine if the system supports IPv4.

    Returns:
        True if the system supports IPv4, False otherwise.
    """
    try:
        httpx.head("http://1.1.1.1", timeout=3)
    except httpx.RequestError:
        return False
    else:
        return True


def _is_ipv6_supported() -> bool:
    """Determine if the system supports IPv6.

    Returns:
        True if the system supports IPv6, False otherwise.
    """
    try:
        httpx.head("http://[2606:4700:4700::1111]", timeout=3)
    except httpx.RequestError:
        return False
    else:
        return True


def _should_use_ipv6() -> bool:
    """Determine if the system supports IPv6.

    Returns:
        True if the system supports IPv6, False otherwise.
    """
    return not _is_ipv4_supported() and _is_ipv6_supported()


def _httpx_local_address_kwarg() -> str:
    """Get the value of the HTTPX local_address keyword argument.

    Returns:
        The local address to bind to
    """
    from ..config import environment

    return environment.REFLEX_HTTP_CLIENT_BIND_ADDRESS.get() or (
        "::" if _should_use_ipv6() else "0.0.0.0"
    )


@once
def _httpx_client() -> httpx.Client:
    """Get an HTTPX client.

    Returns:
        An HTTPX client.
    """
    return httpx.Client(
        transport=httpx.HTTPTransport(
            local_address=_httpx_local_address_kwarg(),
            verify=_httpx_verify_kwarg(),
        )
    )


get = _wrap_https_func(_httpx_client().get)