🧅 Tor Session Manager

Una librería Python liviana para gestionar sesiones Tor y rotar circuitos programáticamente.

🎯 Casos de Uso

Esta librería está diseñada para propósitos legítimos:

• Web Scraping Ético: Rotar IPs para respetar rate limits mientras recolectás datos que tenés autorización para acceder
• Investigación de Seguridad: Testear cómo tus aplicaciones manejan requests desde diferentes ubicaciones geográficas
• Testing de Privacidad: QA para aplicaciones enfocadas en privacidad y sistemas de detección de VPN/proxy
• Investigación Académica: Estudiar comportamiento de redes, patrones de censura o características de la red Tor
• Penetration Testing: Evaluaciones de seguridad autorizadas que requieren rotación de IP

⚠️ Aviso de Responsabilidad: Siempre respetá robots.txt, términos de servicio y rate limits. Esta herramienta es solo para uso legítimo. El autor no se responsabiliza por mal uso.

📦 Instalación

pip install tor-session-manager

Prerequisitos

Necesitás Tor corriendo localmente con el puerto de control habilitado:

macOS (Homebrew):

brew install tor
# Editá /opt/homebrew/etc/tor/torrc y agregá:
#   ControlPort 9051
#   CookieAuthentication 1
brew services start tor

Linux (Ubuntu/Debian):

sudo apt install tor
# Editá /etc/tor/torrc y descomentá/agregá:
#   ControlPort 9051
#   CookieAuthentication 1
sudo systemctl restart tor

Windows: Descargá desde torproject.org y configurá torrc.

🚀 Inicio Rápido

Uso Básico

from tor_session_manager import TorClient

with TorClient() as client:
    print(f"IP actual: {client.get_ip()}")
    
    client.rotate()  # Obtener nuevo circuito
    
    print(f"Nueva IP: {client.get_ip()}")

One-liner

from tor_session_manager import rotate_and_get_ip

nueva_ip = rotate_and_get_ip()
print(f"Nueva IP: {nueva_ip}")

Con Requests

import requests
from tor_session_manager import TorClient

client = TorClient()

# Usá la propiedad proxies con cualquier llamada de requests
response = requests.get(
    "https://httpbin.org/ip",
    proxies=client.proxies,
    timeout=30
)
print(response.json())

Scraping con Rotación

import time
import requests
from tor_session_manager import TorClient

def scrape_con_rotacion(urls: list[str], delay: float = 1.0):
    """Scrapear URLs con rotación automática de IP y rate limiting."""
    resultados = []
    
    with TorClient() as client:
        for i, url in enumerate(urls):
            # Rotar cada 10 requests
            if i > 0 and i % 10 == 0:
                client.rotate()
                print(f"Rotado a nueva IP: {client.get_ip()}")
            
            # Delay respetuoso entre requests
            time.sleep(delay)
            
            response = requests.get(url, proxies=client.proxies, timeout=30)
            resultados.append(response.text)
    
    return resultados

📖 Referencia de API

TorClient

TorClient(
    control_port: int = 9051,    # Puerto de control de Tor
    socks_port: int = 9050,      # Puerto proxy SOCKS de Tor  
    password: str = None,        # Password del puerto de control (si no usás cookie auth)
    rotate_delay: float = 2.0,   # Segundos de espera después de rotar
)

Métodos:

💡 Rotación verificada: rotate() a veces reutiliza el mismo nodo de salida, así que la IP puede no cambiar. wait_for_new_ip() rota repetidamente hasta observar una IP distinta (o agotar max_attempts), útil cuando necesitás garantizar un exit nuevo entre requests.

with TorClient() as client:
    ip_vieja = client.get_ip()
    ip_nueva = client.wait_for_new_ip(previous_ip=ip_vieja)
    print(f"{ip_vieja} -> {ip_nueva}")

🔁 IP checkers con fallback: get_ip() prueba varios servicios en orden (ipify, ifconfig.me, icanhazip, httpbin), de modo que la caída de uno solo no rompe la resolución de IP. Si todos fallan levanta AllIPCheckersFailedError.

Inspección de circuitos

Podés ver qué relays componen tus circuitos y por qué país salís:

with TorClient() as client:
    info = client.get_circuit_info()  # circuito activo (BUILT/GENERAL más reciente)
    print(f"Circuito {info.id} · estado {info.status}")
    for hop in info.path:
        print(f"  {hop.nickname or hop.fingerprint} ({hop.country or '??'})")

    exit_relay = info.exit_relay
    print(f"Salida: {exit_relay.address} en {info.exit_country}")

    # También podés listar todos los circuitos o consultar solo el país de salida
    print(f"Circuitos activos: {len(client.list_circuits())}")
    print(f"País de salida: {client.get_exit_country()}")

CircuitInfo expone id, status, purpose, path (lista de RelayInfo), build_flags, created, y las propiedades exit_relay y exit_country. RelayInfo tiene fingerprint, nickname, address y country. La resolución de IP/país del exit es best-effort: si el GeoIP o la consulta al consenso fallan, esos campos quedan en None sin romper la llamada.

Medición de calidad del circuito

Podés medir qué tan bueno es el circuito actual (latencia y ancho de banda, a nivel aplicación a través de Tor) para decidir si conviene rotar:

with TorClient() as client:
    health = client.benchmark()
    print(f"Latencia: {health.latency_ms:.0f} ms")
    print(f"Throughput: {health.throughput_kbps:.0f} KB/s")

    # O medir por separado
    print(f"Latencia: {client.measure_latency(samples=5):.0f} ms")

    # Rotar si el circuito está lento
    if health.latency_ms and health.latency_ms > 2000:
        client.rotate()

benchmark() es best-effort: si una métrica falla (p. ej. cae el endpoint de throughput), ese campo queda en None y la otra se reporta igual. CircuitHealth expone latency_ms, throughput_kbps, samples, measured_at (epoch) y la propiedad ok (True si al menos una métrica se resolvió).

Elegir el nodo de salida (país / a demanda)

Podés forzar por qué país —o por qué relay— salís, útil para esquivar bloqueos del destino que prohíben ciertas IPs de salida:

with TorClient() as client:
    # Salir por un país específico y confirmar que el circuito funciona
    nueva_ip = client.new_circuit(exit_country="de")
    print(f"Saliendo por Alemania: {nueva_ip} ({client.get_exit_country()})")

    # Pinear el exit solo para un bloque; se limpia automáticamente al salir
    with client.pinned_exit(countries=["nl"]):
        requests.get(url, proxies=client.proxies)  # sale por Países Bajos

    # Limpiar cualquier restricción manualmente
    client.reset_exit_nodes()

⚠️ StrictNodes y países sin exits: con strict=True (por defecto) Tor usa solo exits que cumplan la restricción. Si el país no tiene exits usables, no puede armar circuito; new_circuit(verify=True) (default) lo detecta y levanta TorSessionError con un mensaje claro en vez de colgarse. La restricción persiste a nivel del proceso Tor hasta que la limpiás con reset_exit_nodes() (o usás pinned_exit, que lo hace por vos).

Pool de circuitos y "best circuit"

Para elegir el circuito más rápido, CircuitPool mantiene varios circuitos en paralelo, los benchmarkea y te deja usar el mejor:

with TorClient() as client:
    with client.circuit_pool(size=4) as pool:
        pool.build().benchmark()          # levanta 4 circuitos y los mide
        best = pool.pin_fastest()
        print(f"Mejor circuito: {best.exit_ip} · {best.health.latency_ms:.0f} ms")

        # Usar el circuito más rápido para tus requests
        requests.get(url, proxies=pool.pinned_proxies)

        # Ver el ranking, o descartar/pinear otro
        for c in pool.ranked():
            print(c.name, c.exit_ip, c.health.latency_ms)
        pool.prune(keep=2)                 # quedarse solo con los 2 mejores

Cada "lane" del pool es un circuito independiente: se logra dándole a cada requests.Session una credencial SOCKS distinta, y Tor (con IsolateSOCKSAuth, activo por defecto) rutea cada credencial por un circuito separado. PooledCircuit expone session, proxies, health (CircuitHealth) y exit_ip. drop(circuit) recicla un lane (nueva credencial = circuito nuevo), útil para descartar uno lento o bloqueado.

Detección de bloqueo del destino y auto-rotación

Muchos sitios bloquean IPs de salida Tor (Cloudflare, CAPTCHA, 403/429). request_with_retry() detecta esas respuestas y rota a un exit nuevo automáticamente hasta pasar:

with TorClient() as client:
    # Rota de exit hasta obtener una respuesta no bloqueada (o levanta BlockedResponseError)
    response = client.request_with_retry("https://sitio-que-bloquea-tor.com", max_rotations=5)
    print(response.status_code)

    # ¿El exit actual está bloqueado por este sitio?
    motivo = client.is_exit_blocklisted("https://sitio-que-bloquea-tor.com")
    if motivo:
        print(f"Exit bloqueado: {motivo}")

La detección (detect_block) reconoce challenges de Cloudflare/CAPTCHA en el body y statuses de bloqueo (403/429/503). Es heurística —un 403 legítimo puede parecer un bloqueo—, así que los marcadores y statuses son ajustables:

response = client.request_with_retry(
    url,
    markers=("acceso denegado", "captcha"),   # marcadores de challenge propios
    statuses={403, 429, 503, 418},             # statuses considerados bloqueo
)

Cuando se agotan las rotaciones sin éxito, levanta BlockedResponseError (con .reason y .response para inspeccionar el último intento).

Async / scraping concurrente

Para scraping paralelo hay un cliente async, TorClientAsync (sobre aiohttp). Instalá el extra:

pip install tor-session-manager[async]

import asyncio
from tor_session_manager import TorClientAsync

async def main():
    async with TorClientAsync(max_concurrency=10) as client:
        print(await client.get_ip())

        # Descargar muchas URLs concurrentemente (acotado por max_concurrency)
        urls = ["https://httpbin.org/get"] * 50
        responses = await client.gather_requests(urls)
        ok = [r for r in responses if r is not None]
        print(f"{len(ok)}/{len(urls)} OK")

        # También async: rotate(), request_with_retry(), benchmark(),
        # set_exit_country(), new_circuit()
        await client.rotate()

asyncio.run(main())

El control-plane (rotate, selección de exit) usa stem por debajo vía asyncio.to_thread; el data-plane (requests, medición) va por aiohttp con aiohttp_socks (DNS ruteado por Tor). Cada URL que falla en gather_requests queda como None en su posición, sin cortar el lote.

Bridges y pluggable transports (bajo censura)

Cuando el ISP o el país bloquean Tor mismo, los relays normales son inalcanzables. Con bridges (relays no listados) y pluggable transports (obfs4, snowflake, meek, webtunnel) el tráfico se ofusca para no parecer Tor. La librería lanza un Tor gestionado con esa config:

from tor_session_manager import launch_bridged_tor, TorClient

# Las bridge lines las obtenés de https://bridges.torproject.org o Telegram @GetBridgesBot
proc = launch_bridged_tor(
    bridges=[
        "obfs4 192.0.2.1:443 0123...CDEF cert=... iat-mode=0",
        "obfs4 192.0.2.2:443 89AB...4567 cert=... iat-mode=0",
    ],
    socks_port=9050,
    control_port=9051,
)

# Ahora usás el TorClient normal contra esos puertos
with TorClient(control_port=9051, socks_port=9050) as client:
    print(client.get_ip())

proc.terminate()  # cerrar el Tor gestionado al terminar

⚠️ Binarios de los transports: obfs4proxy, snowflake-client, etc. no vienen con la librería — tienen que estar instalados y en el PATH (o pasás la ruta con transport_paths={"obfs4": "/ruta/obfs4proxy"}). Si falta el binario, se levanta BridgeConfigError con un mensaje claro.

UseBridges/Bridge/ClientTransportPlugin son opciones de arranque de Tor (no se pueden encender en un Tor ya corriendo), por eso el flujo soportado es lanzar un Tor propio. También podés usar parse_bridge_line() y build_bridge_config() por separado si querés armar tu propio torrc.

Anti-fingerprint: rotación de headers + pacing

Rotar la IP no alcanza si cada request lleva el mismo User-Agent: ese fingerprint te identifica igual. TorClient puede rotar perfiles de headers coherentes (de navegadores reales) y pacear el ritmo por host con backoff adaptativo ante 429:

client = TorClient(rotate_headers=True, rate_limit=1.0)  # 1s mínimo entre requests por host

with client:
    # request() inyecta un perfil de headers distinto por request y respeta el ritmo
    r1 = client.request("https://ejemplo.com/a")
    r2 = client.request("https://ejemplo.com/b")  # espera ~1s; si el host devuelve 429, el ritmo se ensancha

También podés usar los componentes sueltos: HeaderRotator (round-robin sobre BROWSER_PROFILES, o shuffle=True) y RateLimiter (pacing por host con acquire()/record()).

⚠️ Sobre TLS/JA3: requests usa el ssl de la stdlib, así que el fingerprint JA3/TLS del handshake es fijo y no se puede cambiar desde esta librería. La rotación de headers es la palanca a nivel aplicación; para spoofear JA3 haría falta algo como curl_cffi (fuera de alcance).

🔒 Política de enrutamiento

Las sesiones internas de TorClient y CircuitPool usan socks5h://127.0.0.1:<puerto> (DNS a través del proxy) y trust_env=False. HTTP_PROXY, HTTPS_PROXY, ALL_PROXY, NO_PROXY y sus variantes en minúscula no reemplazan ni evitan ese proxy, tampoco en redirecciones. Si el SOCKS no está disponible, la solicitud falla; no se reintenta por una conexión directa. Los checkers alternativos de IP conservan la misma ruta.

client.request(...) y request_with_retry(...) rechazan el argumento proxies, incluso None o un diccionario vacío, con ValueError. TorClientAsync usa un conector SOCKS con DNS remoto y trust_env=False, y rechaza proxy y proxy_auth por solicitud. Elegí el puerto mediante socks_port al crear el cliente.

En Requests, desactivar trust_env también desactiva credenciales automáticas de .netrc y REQUESTS_CA_BUNDLE/CURL_CA_BUNDLE. Si necesitás una CA personalizada, pasá verify="/ruta/ca.pem" explícitamente a client.request; no desactives la verificación TLS.

Esta política cubre las sesiones administradas por la biblioteca. El diccionario client.proxies no endurece una sesión externa ni impide que el llamador la modifique. Para una sesión propia, configurá explícitamente session.trust_env = False y session.proxies = client.proxies, sin overrides por solicitud. Cambiar sesiones, adaptadores o conectores queda bajo responsabilidad del llamador. El proxy configurado debe ser un Tor confiable: este enrutamiento no garantiza anonimato ni demuestra por sí solo que el servidor SOCKS pertenece a Tor.

🖥️ CLI

Instalar el paquete deja disponible el comando tor-session:

tor-session ip                 # IP de salida actual
tor-session rotate             # rota y muestra la nueva IP
tor-session circuit            # relays del circuito activo y país de salida
tor-session country            # país del nodo de salida
tor-session benchmark          # latencia + throughput del circuito
tor-session status --json      # snapshot completo en JSON (para monitoreo)

Flags comunes: --control-port, --socks-port, --password y --json (salida machine-readable). Se aceptan antes o después del subcomando: tor-session --json status y tor-session status --json son equivalentes. Los puertos deben estar entre 1 y 65535. Preferí autenticación por cookie: un password pasado como argumento puede ser visible en el historial del shell o la lista de procesos.

status conserva los campos version, ready, ip, exit_country, num_circuits y health, y agrega checks y errors. Cada check usa ok, failed o not_checked:

• controller: conexión al puerto de control.
• authentication: autenticación con Stem; falla de credenciales distinguida de conexión.
• bootstrap: progreso informado por Tor; debe alcanzar 100%.
• socks: el listener local responde al saludo SOCKS5 sin autenticación.
• connectivity: un checker público devuelve una IP válida a través del SOCKS configurado.

ready requiere los cinco checks exitosos. Si falla SOCKS o el bootstrap está incompleto, la conectividad pública queda sin comprobar. La ruta SOCKS se diagnostica incluso si el controlador no está disponible. El saludo SOCKS y una IP pública válida no certifican que ese proxy sea Tor ni garantizan anonimato.

Exit codes: 0 si está listo, 1 si ready=false (o el comando falla), 2 para argumentos inválidos. Los errores de campos auxiliares se informan en errors y no cambian por sí solos la disponibilidad. En modo --json, los errores de ejecución se emiten como JSON; los errores de sintaxis de argumentos siguen el formato de argparse. Las sondas locales de diagnóstico usan un timeout de 3 segundos por operación; no es un límite global de status: los checkers y benchmarks conservan sus timeouts habituales. status hace consultas externas para conectividad y métricas a través del proxy.

Ejemplo de observabilidad (con pipefail para conservar el error de status):

set -o pipefail
tor-session status --json | jq '{ready, checks, ip, errors}'

🧪 Fixture de pytest

El paquete registra una fixture tor_client (vía entry point de pytest) que provee un TorClient listo, o skipea el test si Tor no está corriendo:

def test_mi_scraper(tor_client):
    assert tor_client.get_ip()  # se saltea automáticamente si no hay Tor

Context Managers:

# Uso estándar - verifica que Tor esté listo
with TorClient() as client:
    ...

# Rotar antes de una operación específica
with client.rotated_session():
    # Circuito fresco para este bloque
    ...

Excepciones

⚙️ Cómo Funciona

Esta librería actúa como un puente entre tu código Python y la red Tor:

Tu código  →  TorClient  →  Tor Proxy (SOCKS5)  →  Internet
                   ↕
            Controlador Tor
           (rotación de circuitos)

Componentes

1. Proxy SOCKS5 (puerto 9050): Tu tráfico HTTP/HTTPS pasa por acá para salir a través de Tor
2. Puerto de Control (puerto 9051): Permite enviar comandos a Tor (como rotar circuitos)
3. TorClient: Maneja la autenticación y envía señales al controlador

Flujo de Rotación

1. Se envía señal NEWNYM al puerto de control de Tor
2. Tor construye un nuevo circuito con diferentes nodos relay
3. Se espera rotate_delay segundos (por defecto 2s) para que el circuito se establezca
4. Las próximas requests usarán el nuevo circuito → nueva IP de salida

Limitaciones

• Rate limits de Tor: No podés rotar circuitos instantáneamente. Tor tiene límites (~10 rotaciones/min)
• Velocidad: Tor es más lento que una conexión directa (3+ saltos en la red)
• Misma IP posible: Ocasionalmente podés obtener la misma IP después de rotar (aunque es poco probable)

🔧 Configuración

Puertos Personalizados

client = TorClient(
    control_port=9151,  # Puerto de control no estándar
    socks_port=9150,    # Puerto SOCKS no estándar
)

Autenticación por Password

Si usás password auth en lugar de cookie auth:

client = TorClient(password="tu_password_de_control")

Logging

Habilitá debug logging para ver qué está pasando:

import logging
logging.basicConfig(level=logging.DEBUG)

🧪 Testeando tu Setup

from tor_session_manager import TorClient

client = TorClient()

if client.is_ready():
    print("✅ Tor está corriendo y listo")
    with client:
        print(f"📍 Tu IP de Tor: {client.get_ip()}")
else:
    print("❌ Tor no está listo - verificá la instalación")

🔍 Troubleshooting

"TorConnectionError: Failed to connect to Tor controller"

Causas comunes:

• Tor no está corriendo → brew services start tor (macOS) o sudo systemctl start tor (Linux)
• Puerto de control no habilitado → agregá ControlPort 9051 en torrc
• Firewall bloqueando conexión local

Verificar:

# Verificar si Tor está corriendo
ps aux | grep tor

# En Linux, verificar status
sudo systemctl status tor

"TorNotReadyError: Tor is not fully bootstrapped"

Tor puede tardar unos segundos en conectarse a la red. Esperá ~10-15 segundos después de iniciar Tor antes de usar la librería.

Verificar status:

from tor_session_manager import TorClient

client = TorClient()
if client.is_ready():
    print("✅ Listo")
else:
    print("❌ Esperá un momento y volvé a intentar")

"IPFetchError: Failed to fetch IP address"

Causas:

• Tor no está ruteando el tráfico correctamente
• Problema de conectividad general
• Sitio de verificación de IP bloqueado

Solución:

1. Verificá que Tor esté corriendo
2. Probá manualmente: curl --proxy socks5h://127.0.0.1:9050 https://api.ipify.org
3. Si falla, verificá la configuración de Tor

Puertos personalizados no funcionan

Si cambiaste los puertos en torrc, asegurate de reiniciar Tor:

# macOS
brew services restart tor

# Linux
sudo systemctl restart tor

La rotación no cambia la IP

Esto puede pasar ocasionalmente. Tor tiene un pool finito de nodos de salida y puede asignarte el mismo. Intentá rotar nuevamente o esperá unos segundos.

Logging para debugging

Habilitá logs detallados para ver qué está pasando:

import logging
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

🤝 Contribuciones

¡Las contribuciones son bienvenidas! Sentite libre de abrir issues y pull requests.

📄 Licencia

MIT License - ver LICENSE para detalles.

👤 Autor

Pablo Alaniz - @PabloAlaniz

Construido para la comunidad de investigación de seguridad e ingeniería de datos 🔐