url

plantdb.client.url

Robust server‑availability checker with built‑in SSRF / safety guards.

ServerCheckResult dataclass

ServerCheckResult(url, status_code, ok, final_url, message, n_redirects=0)

Class containing information about the server's response.

is_server_available

is_server_available(url, timeout=5.0, max_redirects=2, max_retries=3, backoff_factor=0.3, allow_private_ip=False, validate_host=True, cert_path=None, verify_ssl=True)

Check if a server is available by making an HTTP request to the given URL.

Parameters:

Name	Type	Description	Default
url	str	The URL of the server to check.	required
timeout	float	The maximum number of seconds to wait for a response. Default is 5.0.	5.0
max_redirects	int	The maximum number of redirects to follow. Default is 2.	2
max_retries	int	The maximum number of retries in case of failure. Default is 3.	3
backoff_factor	float	A factor to multiply the delay between retry attempts. Default is 0.3.	0.3
allow_private_ip	bool	If True, the checker will accept URLs whose resolved IPs are private or non‑routable.	False
validate_host	bool	Whether to validate the hostname against a whitelist & blacklist. Default is True. Default is False.	True
cert_path	str or Path	Path to a CA bundle to use. Default is None.	None
verify_ssl	bool	Whether to verify SSL certificates. Default is True.	True

Returns:

Type	Description
ServerCheckResult	An object containing details about the server check result. url (str): The original URL that was checked. status_code (int): The HTTP status code received from the server. ok (bool): True if the server responded with a successful status code, False otherwise. final_url (str): The final URL after all redirects have been followed.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> from plantdb.client.url import is_server_available
>>> # EXAMPLE 1: Test with a valid online server
>>> result = is_server_available("https://example.com")
>>> print(result.ok)
True
>>> print(result.url)
https://example.com
>>> # EXAMPLE 2: Test with a valid local IP (allowed for test purposes) and a running server
>>> server = TestRestApiServer(test=True, ssl=False)  # initialize the server
>>> server.start()  # start the server
>>> result = is_server_available("http://127.0.0.1:5000", allow_private_ip=True)
>>> print(result.ok)
True
>>> print(result.message)
Valid URL.
>>> server.stop()  # stop the server

Notes

This function uses the requests library to perform HTTP requests. It handles retries and redirects according to the parameters provided.

See Also

requests.Session : The class used for sending HTTP requests. plantdb.client.url._parse_url: the url parsing method. plantdb.client.url._validate_hostname: the url validation method used if validate_host is True.

Source code in plantdb/client/url.py

def is_server_available(
        url: str,
        timeout: float = 5.0,
        max_redirects: int = 2,
        max_retries: int = 3,
        backoff_factor: float = 0.3,
        allow_private_ip: bool = False,
        validate_host: bool = True,
        cert_path: Optional[str] = None,
        verify_ssl: bool = True,
) -> ServerCheckResult:
    """
    Check if a server is available by making an HTTP request to the given URL.

    Parameters
    ----------
    url : str
        The URL of the server to check.
    timeout : float, optional
        The maximum number of seconds to wait for a response. Default is 5.0.
    max_redirects : int, optional
        The maximum number of redirects to follow. Default is 2.
    max_retries : int, optional
        The maximum number of retries in case of failure. Default is 3.
    backoff_factor : float, optional
        A factor to multiply the delay between retry attempts. Default is 0.3.
    allow_private_ip : bool, optional
        If True, the checker will accept URLs whose resolved IPs are private or non‑routable.
    validate_host : bool, optional
        Whether to validate the hostname against a whitelist & blacklist. Default is True.
        Default is False.
    cert_path : str or pathlib.Path, optional
        Path to a CA bundle to use. Default is None.
    verify_ssl : bool, optional
        Whether to verify SSL certificates. Default is True.

    Returns
    -------
    ServerCheckResult
        An object containing details about the server check result.

        - url (str): The original URL that was checked.
        - status_code (int): The HTTP status code received from the server.
        - ok (bool): True if the server responded with a successful status code,
          False otherwise.
        - final_url (str): The final URL after all redirects have been followed.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> from plantdb.client.url import is_server_available
    >>> # EXAMPLE 1: Test with a valid online server
    >>> result = is_server_available("https://example.com")
    >>> print(result.ok)
    True
    >>> print(result.url)
    https://example.com
    >>> # EXAMPLE 2: Test with a valid local IP (allowed for test purposes) and a running server
    >>> server = TestRestApiServer(test=True, ssl=False)  # initialize the server
    >>> server.start()  # start the server
    >>> result = is_server_available("http://127.0.0.1:5000", allow_private_ip=True)
    >>> print(result.ok)
    True
    >>> print(result.message)
    Valid URL.
    >>> server.stop()  # stop the server

    Notes
    -----
    This function uses the `requests` library to perform HTTP requests. It handles
    retries and redirects according to the parameters provided.

    See Also
    --------
    requests.Session : The class used for sending HTTP requests.
    plantdb.client.url._parse_url: the url parsing method.
    plantdb.client.url._validate_hostname: the url validation method used if `validate_host` is True.
    """
    parsed = _parse_url(url, allow_private_ip=allow_private_ip, validate_host=validate_host)
    if not parsed:
        message = f"URL '{url}' is not a valid URL."
        logger.warning(message)
        return ServerCheckResult(
            url=url,
            status_code=0,
            ok=False,
            final_url=url,
            message=message,
        )

    # Configure retry strategy
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=backoff_factor,
        status_forcelist=[500, 502, 503, 504],
        allowed_methods={"GET", "HEAD"},
        redirect=max_redirects
    )

    # Create a custom SSL context to handle self-signed certificates
    ssl_context = None
    ca_certs = None

    if cert_path is not None:
        ssl_context = create_urllib3_context()

        # If verify_ssl is False, disable certificate verification
        if not verify_ssl:
            from ssl import CERT_NONE
            ssl_context.check_hostname = False
            ssl_context.verify_mode = CERT_NONE

        # If a custom cert path is provided, load the CA certificates
        if cert_path is not None:
            if not isinstance(cert_path, Path):
                cert_path = Path(cert_path)

            # Load custom CA bundle if it exists
            if cert_path.is_file():
                ssl_context.load_verify_locations(cafile=str(cert_path))
                ca_certs = str(cert_path)
            elif cert_path.is_dir():
                # Try to find cert.pem in the directory
                cert_file = cert_path / 'cert.pem'
                if cert_file.exists():
                    ssl_context.load_verify_locations(cafile=str(cert_file))
                    ca_certs = str(cert_file)
                else:
                    ssl_context.load_verify_locations(capath=str(cert_path))
                    ca_certs = str(cert_path)

    # Determine cert_reqs and ca_certs for PoolManager
    if not verify_ssl:
        cert_reqs = 'CERT_NONE'
        ca_certs = None
    elif ca_certs is not None:
        cert_reqs = 'CERT_REQUIRED'
    else:
        cert_reqs = 'CERT_REQUIRED'
        ca_certs = None

    # Create a PoolManager with the retry strategy
    http = urllib3.PoolManager(
        retries=retry_strategy,
        timeout=urllib3.Timeout(total=timeout),
        cert_reqs=cert_reqs,
        ca_certs=ca_certs,
        ssl_context=ssl_context,
    )

    current_url = URL(url).href
    try:
        # Perform the request
        resp = http.request(
            'GET',
            current_url,
            redirect=False,  # Manually handle redirects
            preload_content=False  # Similar to stream=True in requests
        )

        # Handle redirects manually
        redirect_count = 0
        while 300 <= resp.status < 400 and redirect_count < max_redirects:
            # Get redirect location
            location = resp.headers.get('Location')
            if not location:
                message = f"Request to {current_url} (redirect) failed to determine next location."
                return ServerCheckResult(
                    url=URL(url).href,
                    status_code=resp.status,
                    ok=False,
                    final_url=current_url,
                    message=message,
                )

            # Resolve relative URLs
            next_url = join_url(current_url, location)

            # Safety check on the next hop
            if not _parse_url(next_url, allow_private_ip=allow_private_ip):
                message = f"Request to {current_url} (redirect) is not a valid URL."
                return ServerCheckResult(
                    url=URL(url).href,
                    status_code=resp.status,
                    ok=False,
                    final_url=next_url,
                    message=message,
                )

            current_url = next_url
            redirect_count += 1

            # Perform redirect request
            resp = http.request(
                'GET',
                current_url,
                redirect=False,
                preload_content=False
            )

        # Return the final result
        return ServerCheckResult(
            url=URL(url).href,
            status_code=resp.status,
            ok=200 <= resp.status < 400,
            final_url=current_url,
            message="Valid URL.",
            n_redirects=redirect_count
        )

    except (MaxRetryError, NewConnectionError, TimeoutError) as err:
        message = f"Request to {current_url} failed with: {err}"
        return ServerCheckResult(
            url=URL(url).href,
            status_code=0,
            ok=False,
            final_url=current_url,
            message=message,
        )
    finally:
        # Ensure connection is released
        resp.release_conn() if 'resp' in locals() else None