Test a static proxy with curl and Python

"""Verify a static proxy without logging credentials.

Python 3.9+. Install: python -m pip install "requests[socks]==2.*"
Set PROXY_HOST, PROXY_PORT, PROXY_USER, and PROXY_PASSWORD.
PROXY_SCHEME: http (default), socks5, or socks5h.
Run: python examples/static_proxy_check.py

This script makes a direct request, three proxy requests separated by an
interval, and a final direct request. It does not test geolocation or uptime.
"""
from datetime import datetime, timezone
from ipaddress import ip_address
import math
import os
import platform
import sys
import time
from urllib.parse import quote, urlsplit

import requests


class CheckError(RuntimeError):
    """A safe diagnostic that contains no proxy credentials."""


def positive_seconds(env, name, default):
    try:
        value = float(env.get(name, str(default)))
    except (TypeError, ValueError):
        raise ValueError(f"{name} must be a number") from None
    if not math.isfinite(value) or value <= 0:
        raise ValueError(f"{name} must be finite and greater than zero")
    return value


def configuration(env):
    required = ("PROXY_HOST", "PROXY_PORT", "PROXY_USER", "PROXY_PASSWORD")
    missing = [name for name in required if not env.get(name)]
    if missing:
        raise ValueError("Missing environment variables: " + ", ".join(missing))
    scheme = env.get("PROXY_SCHEME", "http").strip().lower()
    if scheme not in {"http", "socks5", "socks5h"}:
        raise ValueError("PROXY_SCHEME must be http, socks5, or socks5h")
    host = env["PROXY_HOST"].strip()
    if host.startswith("[") and host.endswith("]"):
        host = host[1:-1]
    if not host or any(c.isspace() or c in "/?#@" for c in host):
        raise ValueError("PROXY_HOST must be a hostname or IP without a URL or port")
    if ":" in host:
        try:
            if ip_address(host).version != 6:
                raise ValueError
        except ValueError:
            raise ValueError("Use a valid IPv6 host; keep the port in PROXY_PORT") from None
        host = "[" + host + "]"
    try:
        port = int(env["PROXY_PORT"])
    except (TypeError, ValueError):
        raise ValueError("PROXY_PORT must be an integer") from None
    if not 1 <= port <= 65535:
        raise ValueError("PROXY_PORT must be between 1 and 65535")
    user = quote(env["PROXY_USER"], safe="")
    password = quote(env["PROXY_PASSWORD"], safe="")
    proxy_url = f"{scheme}://{user}:{password}@{host}:{port}"
    endpoint = env.get("IP_CHECK_URL", "https://api.ipify.org?format=json")
    parsed = urlsplit(endpoint)
    local_http = parsed.scheme == "http" and parsed.hostname in {"localhost", "127.0.0.1", "::1"}
    if not parsed.hostname or (parsed.scheme != "https" and not local_http):
        raise ValueError("IP_CHECK_URL must use HTTPS; HTTP loopback is allowed for local tests")
    if parsed.username or parsed.password or parsed.fragment:
        raise ValueError("IP_CHECK_URL must not contain credentials or a fragment")
    expected = env.get("EXPECTED_EXIT_IP", "").strip()
    if expected:
        try:
            expected = str(ip_address(expected))
        except ValueError:
            raise ValueError("EXPECTED_EXIT_IP must be a valid IP address") from None
    return {
        "scheme": scheme,
        "proxies": {"http": proxy_url, "https": proxy_url},
        "endpoint": endpoint,
        "expected": expected,
        "interval": positive_seconds(env, "VERIFY_INTERVAL_SECONDS", 30),
        "timeout": (
            positive_seconds(env, "CONNECT_TIMEOUT_SECONDS", 10),
            positive_seconds(env, "READ_TIMEOUT_SECONDS", 20),
        ),
    }


def get_ip(endpoint, proxies, timeout):
    try:
        with requests.Session() as session:
            # Ignore HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, NO_PROXY and .netrc.
            session.trust_env = False
            response = session.get(
                endpoint,
                proxies=proxies,
                timeout=timeout,
                allow_redirects=False,
            )
            response.raise_for_status()
            if response.status_code != 200:
                raise CheckError(f"unexpected_status={response.status_code}; use an IP JSON endpoint")
            try:
                payload = response.json()
                if not isinstance(payload, dict) or not isinstance(payload.get("ip"), str):
                    raise ValueError
                address = str(ip_address(payload["ip"].strip()))
            except (ValueError, TypeError):
                raise CheckError("invalid_response: expected JSON with a valid string 'ip' field") from None
            return response.status_code, address
    except requests.exceptions.ProxyError:
        raise CheckError("proxy_error: check protocol, host, port, authentication, and any IP allowlist") from None
    except requests.exceptions.SSLError:
        raise CheckError("tls_error: check the certificate chain; keep TLS verification enabled") from None
    except requests.exceptions.Timeout:
        raise CheckError("timeout: check reachability, firewall, DNS, and assigned endpoint") from None
    except requests.exceptions.ConnectionError:
        raise CheckError("connection_error: check DNS, network access, and the endpoint") from None
    except requests.exceptions.HTTPError as error:
        status = error.response.status_code if error.response is not None else "unknown"
        raise CheckError(f"http_error={status}: check endpoint response and authentication") from None
    except requests.exceptions.InvalidSchema:
        raise CheckError("unsupported_scheme: install requests[socks] and check PROXY_SCHEME") from None
    except requests.exceptions.RequestException:
        raise CheckError("request_error: check client configuration; share only a redacted diagnostic") from None


def run(config, sleep=time.sleep, clock=time.monotonic):
    print("verified_at_utc=" + datetime.now(timezone.utc).isoformat())
    print(f"operating_system={platform.system()} {platform.release()}")
    print(f"client=Python {sys.version.split()[0]}, requests {requests.__version__}")
    print("proxy_scheme=" + config["scheme"])
    print(f"wait_between_proxy_requests_seconds={config['interval']:g}")
    before_status, before = get_ip(config["endpoint"], None, config["timeout"])
    print(f"direct_before_status={before_status} direct_before_ip={before}")
    started = clock()
    exits = []
    for attempt in range(3):
        if attempt:
            sleep(config["interval"])
        status, address = get_ip(config["endpoint"], config["proxies"], config["timeout"])
        exits.append(address)
        elapsed = clock() - started
        print(f"proxy_check={attempt + 1} elapsed_seconds={elapsed:.3f} status={status} exit_ip={address}")
    after_status, after = get_ip(config["endpoint"], None, config["timeout"])
    print(f"direct_after_status={after_status} direct_after_ip={after}")
    same_exit = len(set(exits)) == 1
    baseline_stable = before == after
    differs = all(address != before and address != after for address in exits)
    expected_matches = not config["expected"] or all(address == config["expected"] for address in exits)
    print(f"same_exit={same_exit}")
    print(f"direct_baseline_stable={baseline_stable}")
    print(f"direct_differs_from_exit={differs}")
    print(f"expected_exit_matches={expected_matches if config['expected'] else 'not_checked'}")
    if not (same_exit and baseline_stable and differs and expected_matches):
        raise CheckError("verification_failed: inspect routing, baseline changes, and the current allocation")
    print("verification_passed=True")
    return 0


def main():
    try:
        return run(configuration(os.environ))
    except (ValueError, CheckError) as error:
        # Only our own safe messages are printed, never raw Requests exceptions.
        print(str(error), file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        print("verification_interrupted", file=sys.stderr)
        return 130


if __name__ == "__main__":
    sys.exit(main())