"""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())
