Cómo escribir un verificador de proxies en Python

Un verificador de proxies en Python verifica la funcionalidad, velocidad, anonimato y ubicación geográfica de un servidor proxy realizando solicitudes HTTP a través de él a una URL de destino conocida y analizando la respuesta.

Fundamentos de la Verificación de Proxies

El núcleo de un verificador de proxies implica enviar una solicitud HTTP a través de un proxy especificado y evaluar el resultado. Este proceso típicamente utiliza la librería requests, que simplifica las interacciones HTTP en Python.

Verificación Básica de Proxy HTTP

Para verificar un proxy HTTP o HTTPS básico, configure el diccionario proxies en el método requests.get(). Una URL de destino confiable, como httpbin.org/ip o ipinfo.io/json, es esencial. Estos servicios devuelven la dirección IP pública desde la que se originó la solicitud, permitiendo la verificación.

import requests
import time

def check_http_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Checks an HTTP/HTTPS proxy for basic connectivity and response time.

    Args:
        proxy_address: The proxy string (e.g., "http://user:pass@ip:port").
        timeout: Maximum time in seconds to wait for a response.

    Returns:
        A dictionary with proxy status, response time, and error message if any.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/ip" # Use http for httpbin, or https for ipinfo
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2) # Latency in ms
        origin_ip = response.json().get('origin')

        return {
            "proxy": proxy_address,
            "status": "Alive",
            "latency_ms": latency,
            "origin_ip": origin_ip,
            "error": None
        }
    except requests.exceptions.Timeout:
        return {"proxy": proxy_address, "status": "Timeout", "latency_ms": None, "origin_ip": None, "error": "Request timed out"}
    except requests.exceptions.ConnectionError:
        return {"proxy": proxy_address, "status": "Dead", "latency_ms": None, "origin_ip": None, "error": "Connection failed"}
    except requests.exceptions.HTTPError as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"HTTP Error: {e}"}
    except Exception as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"An unexpected error occurred: {e}"}

# Example usage:
# print(check_http_proxy("http://203.0.113.45:8080"))
# print(check_http_proxy("https://198.51.100.22:443"))

Manejo de Diferentes Tipos de Proxy

La librería requests soporta nativamente proxies HTTP y HTTPS. Para proxies SOCKS (SOCKS4 y SOCKS5), se requiere una dependencia adicional requests[socks]. Instálela usando pip install requests[socks]. El esquema de la URL del proxy cambia entonces a socks5:// o socks4://.

# Example for SOCKS5 proxy
def check_socks_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Checks a SOCKS proxy. Requires 'requests[socks]' to be installed.
    """
    proxies = {
        "http": f"socks5://{proxy_address}",
        "https": f"socks5://{proxy_address}",
    }
    target_url = "http://httpbin.org/ip"
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2)
        origin_ip = response.json().get('origin')

        return {
            "proxy": proxy_address,
            "status": "Alive",
            "latency_ms": latency,
            "origin_ip": origin_ip,
            "error": None
        }
    except requests.exceptions.Timeout:
        return {"proxy": proxy_address, "status": "Timeout", "latency_ms": None, "origin_ip": None, "error": "Request timed out"}
    except requests.exceptions.ConnectionError:
        return {"proxy": proxy_address, "status": "Dead", "latency_ms": None, "origin_ip": None, "error": "Connection failed"}
    except Exception as e:
        return {"proxy": proxy_address, "status": "Error", "latency_ms": None, "origin_ip": None, "error": f"SOCKS error: {e}"}

# Example usage:
# print(check_socks_proxy("user:pass@192.0.2.10:1080"))

Evaluación de la Calidad del Proxy

Más allá de la conectividad básica, la calidad de un proxy se define por su velocidad, anonimato y ubicación geográfica.

Velocidad y Latencia

La latencia se mide por el tiempo transcurrido entre el envío de la solicitud y la recepción del primer byte de la respuesta. La función time.time() se puede usar para capturar los tiempos de inicio y fin, con la diferencia indicando la latencia. Típicamente se reporta en milisegundos.

# The latency calculation is already included in the check_http_proxy and check_socks_proxy functions.
# latency = round((end_time - start_time) * 1000, 2) # Latency in ms

Niveles de Anonimato

El anonimato del proxy se refiere a la eficacia con la que un proxy oculta la dirección IP original y la identidad del cliente. Esto se determina analizando encabezados HTTP específicos que el proxy podría añadir o modificar. Servicios como httpbin.org/headers o ipinfo.io/json pueden usarse para inspeccionar estos encabezados.

• Élite/Anonimato Alto: El proxy oculta la dirección IP del cliente y no envía ningún encabezado de identificación como X-Forwarded-For, Via o Proxy-Connection. El servidor de destino solo ve la IP del proxy.
• Anónimo: El proxy oculta la dirección IP del cliente pero puede enviar encabezados como Via o X-Forwarded-For con la IP del proxy, indicando el uso de un proxy. La IP original del cliente no se expone directamente.
• Transparente: El proxy reenvía la dirección IP del cliente en encabezados como X-Forwarded-For. El servidor de destino sabe que la solicitud provino de un proxy y puede identificar la IP del cliente original.

Para verificar el anonimato, envíe una solicitud a una URL que devuelva los encabezados de la solicitud, luego inspeccione la respuesta. También necesita conocer su propia dirección IP pública (sin un proxy) para comparar.

import requests

def get_my_ip():
    """Returns the client's public IP address without a proxy."""
    try:
        response = requests.get("http://httpbin.org/ip", timeout=5)
        response.raise_for_status()
        return response.json().get('origin')
    except requests.exceptions.RequestException:
        return None

def check_anonymity(proxy_address: str, my_ip: str, timeout: float = 10.0) -> str:
    """
    Checks the anonymity level of a proxy.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/headers" # Returns all request headers
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        headers = response.json().get('headers', {})

        # Check for X-Forwarded-For, Via, Proxy-Connection headers
        if 'X-Forwarded-For' in headers and headers['X-Forwarded-For'] == my_ip:
            return "Transparent"
        if 'Via' in headers or 'X-Forwarded-For' in headers:
            return "Anonymous"

        # If the origin IP (from httpbin.org/ip) is different from my_ip and no identifying headers are present.
        # This requires a separate check to httpbin.org/ip through the proxy.
        # For a more precise check, one would compare the 'origin' IP from httpbin.org/ip through proxy
        # with my_ip and then check headers.

        # Simplified check: if no common identifying headers, assume Elite (requires careful validation)
        return "Elite"

    except requests.exceptions.RequestException:
        return "Unknown (Proxy Dead or Error)"

# Example usage:
# my_public_ip = get_my_ip()
# if my_public_ip:
#     print(f"My IP: {my_public_ip}")
#     print(check_anonymity("http://203.0.113.45:8080", my_public_ip))

Geolocalización

La geolocalización identifica la ubicación física (país, ciudad, región) del servidor proxy. Servicios como ipinfo.io/json proporcionan esta información en una respuesta JSON basada en la IP de origen de la solicitud.

import requests

def get_proxy_geolocation(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Retrieves geolocation data for a proxy.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "https://ipinfo.io/json" # ipinfo.io is good for geolocation
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        data = response.json()
        return {
            "country": data.get('country'),
            "region": data.get('region'),
            "city": data.get('city'),
            "org": data.get('org'),
            "timezone": data.get('timezone')
        }
    except requests.exceptions.RequestException:
        return {"country": None, "region": None, "city": None, "org": None, "timezone": None}

# Example usage:
# geo_data = get_proxy_geolocation("http://203.0.113.45:8080")
# print(geo_data)

Construyendo un Verificador Concurrente

Verificar proxies secuencialmente es ineficiente para listas grandes. La concurrencia, utilizando multithreading, permite verificar múltiples proxies simultáneamente. concurrent.futures.ThreadPoolExecutor de Python simplifica esto.

import concurrent.futures
import csv
import time

# Assume check_http_proxy, check_anonymity, get_proxy_geolocation are defined above

def comprehensive_proxy_check(proxy_address: str, my_ip: str, timeout: float = 10.0) -> dict:
    """
    Performs a comprehensive check for a single proxy.
    """
    result = check_http_proxy(proxy_address, timeout) # Basic connectivity and latency
    if result["status"] == "Alive":
        # Only proceed with anonymity and geolocation if proxy is alive
        anonymity = check_anonymity(proxy_address, my_ip, timeout)
        geolocation = get_proxy_geolocation(proxy_address, timeout)
        result.update({"anonymity": anonymity})
        result.update(geolocation)
    else:
        result.update({"anonymity": None, "country": None, "region": None, "city": None, "org": None, "timezone": None})
    return result

def run_concurrent_checker(proxy_list_file: str, output_file: str, max_workers: int = 10, timeout: float = 10.0):
    """
    Reads proxies from a file, checks them concurrently, and writes results to CSV.
    """
    proxies_to_check = []
    try:
        with open(proxy_list_file, 'r') as f:
            for line in f:
                proxy = line.strip()
                if proxy:
                    proxies_to_check.append(proxy)
    except FileNotFoundError:
        print(f"Error: Proxy list file '{proxy_list_file}' not found.")
        return

    my_ip = get_my_ip()
    if not my_ip:
        print("Could not determine client's public IP. Anonymity checks might be inaccurate.")

    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_to_proxy = {executor.submit(comprehensive_proxy_check, proxy, my_ip, timeout): proxy for proxy in proxies_to_check}
        for future in concurrent.futures.as_completed(future_to_proxy):
            proxy_address = future_to_proxy[future]
            try:
                result = future.result()
                results.append(result)
                print(f"Checked {proxy_address}: Status={result['status']}, Latency={result['latency_ms']}ms, Anonymity={result['anonymity']}")
            except Exception as exc:
                print(f"Proxy {proxy_address} generated an exception: {exc}")
                results.append({"proxy": proxy_address, "status": "Error", "error": str(exc)})

    # Write results to CSV
    if results:
        fieldnames = list(results[0].keys())
        with open(output_file, 'w', newline='') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(results)
        print(f"Results written to {output_file}")
    else:
        print("No proxies checked or no results to write.")

# Example usage:
# Create a file named 'proxies.txt' with one proxy per line, e.g.:
# http://user:pass@192.0.2.1:8080
# socks5://user:pass@198.51.100.2:1080
# http://203.0.113.3:80
# run_concurrent_checker('proxies.txt', 'proxy_results.csv', max_workers=20, timeout=15)

Manejo Robusto de Errores y Mejores Prácticas

• Excepciones Específicas: Capture excepciones requests.exceptions específicas (por ejemplo, Timeout, ConnectionError, HTTPError) para proporcionar retroalimentación granular. Una captura general de Exception debe ser el último recurso.
• User-Agent: Incluya un encabezado User-Agent en las solicitudes para imitar un navegador y evitar ser bloqueado por los servicios de destino.
  headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
• Selección de URL de Destino: Elija servicios confiables, rápidos y disponibles públicamente que proporcionen la información necesaria (IP, encabezados, geolocalización). Evite saturar un solo servicio para prevenir bloqueos de IP.
• Entrada/Salida: Lea listas de proxies de archivos de texto (un proxy por línea) y escriba los resultados en formatos estructurados como CSV o JSON para facilitar el análisis y la integración.
• Autenticación de Proxy: Asegúrese de que las URL de los proxies estén formateadas correctamente para la autenticación (por ejemplo, http://user:pass@ip:port).
• Reintentos: Para verificadores de grado de producción, considere implementar lógica de reintentos para errores transitorios. La librería requests se puede extender con requests.adapters.HTTPAdapter y urllib3.util.retry.Retry para este propósito.