Como escrever um proxy checker em Python

Um proxy checker em Python verifica o funcionamento, a velocidade, o anonimato e a localização geográfica de um servidor proxy, fazendo requisições HTTP através dele para uma URL de destino conhecida e analisando a resposta.

Fundamentos da verificação de proxies

O núcleo de um proxy checker consiste em enviar uma requisição HTTP através de um proxy definido e avaliar o resultado. Esse processo normalmente usa a biblioteca requests, que simplifica as interações HTTP em Python.

Verificação básica de proxy HTTP

Para verificar um proxy HTTP ou HTTPS básico, configure o dicionário proxies no método requests.get(). Uma URL de destino confiável, como httpbin.org/ip ou ipinfo.io/json, é essencial. Esses serviços retornam o endereço IP público de onde a requisição partiu, permitindo a verificação.

import requests
import time

def check_http_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Verifica um proxy HTTP/HTTPS quanto à conectividade básica e ao tempo de resposta.

    Args:
        proxy_address: A string do proxy (ex.: "http://user:pass@ip:port").
        timeout: Tempo máximo em segundos de espera pela resposta.

    Returns:
        Um dicionário com o status do proxy, o tempo de resposta e a mensagem de erro, se houver.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/ip" # Use http para o httpbin, ou https para o ipinfo
    start_time = time.time()
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status() # Levanta HTTPError para respostas ruins (4xx ou 5xx)
        end_time = time.time()
        latency = round((end_time - start_time) * 1000, 2) # Latência em 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}"}

# Exemplo de uso:
# print(check_http_proxy("http://203.0.113.45:8080"))
# print(check_http_proxy("https://198.51.100.22:443"))

Lidando com diferentes tipos de proxy

A biblioteca requests suporta nativamente proxies HTTP e HTTPS. Para proxies SOCKS (SOCKS4 e SOCKS5), é necessária a dependência adicional requests[socks]. Instale-a com pip install requests[socks]. O esquema da URL do proxy então muda para socks5:// ou socks4://.

# Exemplo para proxy SOCKS5
def check_socks_proxy(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Verifica um proxy SOCKS. Requer 'requests[socks]' instalado.
    """
    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}"}

# Exemplo de uso:
# print(check_socks_proxy("user:[email protected]:1080"))

Avaliando a qualidade do proxy

Além da conectividade básica, a qualidade de um proxy é definida pela sua velocidade, anonimato e localização geográfica.

Velocidade e latência

A latência é medida pelo tempo decorrido entre o envio da requisição e o recebimento do primeiro byte da resposta. A função time.time() pode ser usada para capturar os instantes inicial e final, sendo a diferença a latência. Normalmente ela é reportada em milissegundos.

# O cálculo da latência já está incluído nas funções check_http_proxy e check_socks_proxy.
# latency = round((end_time - start_time) * 1000, 2) # Latência em ms

Níveis de anonimato

O anonimato do proxy indica com que eficácia ele oculta o endereço IP original e a identidade do cliente. Isso é determinado analisando cabeçalhos HTTP específicos que o proxy pode adicionar ou modificar. Serviços como httpbin.org/headers ou ipinfo.io/json podem ser usados para inspecionar esses cabeçalhos.

• Elite/Alto anonimato: o proxy oculta o endereço IP do cliente e não envia nenhum cabeçalho identificador como X-Forwarded-For, Via ou Proxy-Connection. O servidor de destino vê apenas o IP do proxy.
• Anônimo: o proxy oculta o endereço IP do cliente, mas pode enviar cabeçalhos como Via ou X-Forwarded-For com o IP do proxy, indicando o uso de um proxy. O IP original do cliente não é exposto diretamente.
• Transparente: o proxy repassa o endereço IP do cliente em cabeçalhos como X-Forwarded-For. O servidor de destino sabe que a requisição veio de um proxy e consegue identificar o IP original do cliente.

Para verificar o anonimato, envie uma requisição para uma URL que devolva os cabeçalhos da requisição e inspecione a resposta. Você também precisa saber seu próprio endereço IP público (sem proxy) para comparação.

import requests

def get_my_ip():
    """Retorna o endereço IP público do cliente sem 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:
    """
    Verifica o nível de anonimato de um proxy.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "http://httpbin.org/headers" # Retorna todos os cabeçalhos da requisição
    try:
        response = requests.get(target_url, proxies=proxies, timeout=timeout)
        response.raise_for_status()
        headers = response.json().get('headers', {})

        # Verifica os cabeçalhos X-Forwarded-For, Via, Proxy-Connection
        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"

        # Se o IP de origem (de httpbin.org/ip) for diferente de my_ip e não houver cabeçalhos identificadores.
        # Isso exige uma verificação separada a httpbin.org/ip através do proxy.
        # Para uma checagem mais precisa, compare o IP 'origin' de httpbin.org/ip através do proxy
        # com my_ip e depois verifique os cabeçalhos.

        # Verificação simplificada: sem cabeçalhos identificadores comuns, assume Elite (exige validação cuidadosa)
        return "Elite"

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

# Exemplo de uso:
# 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))

Geolocalização

A geolocalização identifica a localização física (país, cidade, região) do servidor proxy. Serviços como ipinfo.io/json fornecem essa informação em uma resposta JSON com base no IP de origem da requisição.

import requests

def get_proxy_geolocation(proxy_address: str, timeout: float = 10.0) -> dict:
    """
    Obtém os dados de geolocalização de um proxy.
    """
    proxies = {
        "http": proxy_address,
        "https": proxy_address,
    }
    target_url = "https://ipinfo.io/json" # ipinfo.io é bom para geolocalização
    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}

# Exemplo de uso:
# geo_data = get_proxy_geolocation("http://203.0.113.45:8080")
# print(geo_data)

Construindo um checker concorrente

Verificar proxies sequencialmente é ineficiente para listas grandes. A concorrência, usando multithreading, permite verificar vários proxies simultaneamente. O concurrent.futures.ThreadPoolExecutor do Python simplifica isso.

import concurrent.futures
import csv
import time

# Assume que check_http_proxy, check_anonymity e get_proxy_geolocation foram definidas acima

def comprehensive_proxy_check(proxy_address: str, my_ip: str, timeout: float = 10.0) -> dict:
    """
    Executa uma verificação completa de um único proxy.
    """
    result = check_http_proxy(proxy_address, timeout) # Conectividade básica e latência
    if result["status"] == "Alive":
        # Só prossegue com anonimato e geolocalização se o proxy estiver vivo
        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):
    """
    Lê proxies de um arquivo, verifica-os de forma concorrente e grava os resultados em 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)})

    # Grava os resultados em 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.")

# Exemplo de uso:
# Crie um arquivo chamado 'proxies.txt' com um proxy por linha, ex.:
# http://user:[email protected]:8080
# socks5://user:[email protected]:1080
# http://203.0.113.3:80
# run_concurrent_checker('proxies.txt', 'proxy_results.csv', max_workers=20, timeout=15)

Tratamento robusto de erros e boas práticas

• Exceções específicas: capture exceções específicas de requests.exceptions (ex.: Timeout, ConnectionError, HTTPError) para dar um retorno granular. Capturar Exception genérica deve ser o último recurso.
• User-Agent: inclua um cabeçalho User-Agent nas requisições para imitar um navegador e evitar bloqueios pelos serviços 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'}
• Escolha da URL de destino: escolha serviços confiáveis, rápidos e de acesso público que forneçam as informações necessárias (IP, cabeçalhos, geolocalização). Evite martelar um único serviço para não tomar banimento de IP.
• Entrada/saída: leia listas de proxies de arquivos de texto (um proxy por linha) e grave os resultados em formatos estruturados como CSV ou JSON, para facilitar a análise e a integração.
• Autenticação de proxy: garanta que as URLs de proxy estejam corretamente formatadas para autenticação (ex.: http://user:pass@ip:port).
• Retries: para checkers de nível de produção, considere implementar lógica de retry para erros transitórios. A biblioteca requests pode ser estendida com requests.adapters.HTTPAdapter e urllib3.util.retry.Retry para esse fim.