feat(proxy): implement multi-proxy rotation

[maycon]

Summary

Implement support for multiple proxies with rotation strategies. This enables:

• Bypassing IP-based rate limiting
• Distributing traffic across proxy pool
• Resilience when individual proxies fail

Configuration

config:
  proxy:
    pool:
      - url: "http://proxy1.example.com:8080"
      - url: "http://proxy2.example.com:8080"
      - url: "socks5://proxy3.example.com:1080"
        auth:
          username: "user"
          password: "pass"
    
    rotation: round_robin  # round_robin | random | per_thread | per_request | failover
    
    # Optional: health check
    health_check:
      enabled: true
      interval: 30  # seconds
      timeout: 5
      test_url: "http://httpbin.org/ip"
    
    # Optional: retry on proxy failure
    retry:
      enabled: true
      max_attempts: 3
      remove_failed: true  # Remove proxy from pool after X failures
      failure_threshold: 3

Rotation Strategies

Strategy	Description
round_robin	Cycle through proxies sequentially
random	Random proxy selection per request
per_thread	Each race thread gets dedicated proxy
per_request	New proxy for each HTTP request
failover	Use first proxy, switch only on failure

Implementation

1. Update models/config.py

from enum import Enum
from typing import List, Optional
from dataclasses import dataclass, field

class ProxyRotation(Enum):
    ROUND_ROBIN = "round_robin"
    RANDOM = "random"
    PER_THREAD = "per_thread"
    PER_REQUEST = "per_request"
    FAILOVER = "failover"

@dataclass
class ProxyHealthCheck:
    enabled: bool = False
    interval: int = 30
    timeout: int = 5
    test_url: str = "http://httpbin.org/ip"

@dataclass
class ProxyRetry:
    enabled: bool = True
    max_attempts: int = 3
    remove_failed: bool = True
    failure_threshold: int = 3

@dataclass
class ProxyPoolConfig:
    pool: List[ProxyConfig] = field(default_factory=list)
    rotation: ProxyRotation = ProxyRotation.ROUND_ROBIN
    health_check: Optional[ProxyHealthCheck] = None
    retry: Optional[ProxyRetry] = None

2. Create proxy/manager.py

import threading
import random
from typing import Optional, Dict
from itertools import cycle

class ProxyManager:
    """Manages proxy pool and rotation."""
    
    def __init__(self, config: ProxyPoolConfig):
        self.config = config
        self.proxies = config.pool.copy()
        self.healthy_proxies = config.pool.copy()
        self._lock = threading.Lock()
        self._round_robin = cycle(range(len(self.proxies)))
        self._thread_assignments: Dict[int, ProxyConfig] = {}
        self._failure_counts: Dict[str, int] = {}
    
    def get_proxy(self, thread_id: Optional[int] = None) -> Optional[ProxyConfig]:
        """Get next proxy based on rotation strategy."""
        with self._lock:
            if not self.healthy_proxies:
                return None
            
            if self.config.rotation == ProxyRotation.ROUND_ROBIN:
                idx = next(self._round_robin) % len(self.healthy_proxies)
                return self.healthy_proxies[idx]
            
            elif self.config.rotation == ProxyRotation.RANDOM:
                return random.choice(self.healthy_proxies)
            
            elif self.config.rotation == ProxyRotation.PER_THREAD:
                if thread_id not in self._thread_assignments:
                    idx = thread_id % len(self.healthy_proxies)
                    self._thread_assignments[thread_id] = self.healthy_proxies[idx]
                return self._thread_assignments[thread_id]
            
            elif self.config.rotation == ProxyRotation.FAILOVER:
                return self.healthy_proxies[0]
            
            return self.healthy_proxies[0]
    
    def report_failure(self, proxy: ProxyConfig) -> None:
        """Report proxy failure, potentially removing from pool."""
        if not self.config.retry or not self.config.retry.remove_failed:
            return
            
        with self._lock:
            proxy_url = proxy.url or proxy.http
            self._failure_counts[proxy_url] = self._failure_counts.get(proxy_url, 0) + 1
            
            if self._failure_counts[proxy_url] >= self.config.retry.failure_threshold:
                self.healthy_proxies = [p for p in self.healthy_proxies if p != proxy]
    
    def report_success(self, proxy: ProxyConfig) -> None:
        """Reset failure count on success."""
        with self._lock:
            proxy_url = proxy.url or proxy.http
            self._failure_counts[proxy_url] = 0
    
    async def health_check_loop(self) -> None:
        """Background task to check proxy health."""
        if not self.config.health_check or not self.config.health_check.enabled:
            return
        
        while True:
            await asyncio.sleep(self.config.health_check.interval)
            await self._check_all_proxies()
    
    async def _check_all_proxies(self) -> None:
        """Check health of all proxies."""
        for proxy in self.proxies:
            is_healthy = await self._check_proxy(proxy)
            with self._lock:
                if is_healthy and proxy not in self.healthy_proxies:
                    self.healthy_proxies.append(proxy)
                elif not is_healthy and proxy in self.healthy_proxies:
                    self.healthy_proxies.remove(proxy)

3. Update connection strategies

# connection/preconnect.py

class PreconnectStrategy(ConnectionStrategy):
    def __init__(
        self, 
        sync: Optional[SyncMechanism] = None,
        proxy_manager: Optional[ProxyManager] = None,
    ):
        super().__init__(sync)
        self._proxy_manager = proxy_manager
    
    def _connect(self, thread_id: int) -> None:
        proxies = None
        if self._proxy_manager:
            proxy = self._proxy_manager.get_proxy(thread_id)
            if proxy:
                proxies = proxy.to_httpx_proxies()
        
        client = httpx.Client(
            proxies=proxies,
            # ...
        )

Acceptance Criteria

• Multiple proxies in pool configuration
• Round-robin rotation strategy
• Random rotation strategy
• Per-thread proxy assignment
• Per-request proxy rotation
• Failover strategy
• Proxy health checking (optional)
• Automatic removal of failed proxies
• Thread-safe proxy selection
• CLI support for multiple proxies (--proxy url1 --proxy url2)
• Integration with all connection strategies
• Unit tests for rotation strategies
• Integration tests with mock proxy servers
• Documentation with examples

Example Usage

Rate Limit Bypass

config:
  host: "target.com"
  proxy:
    pool:
      - url: "http://proxy1:8080"
      - url: "http://proxy2:8080"
      - url: "http://proxy3:8080"
    rotation: per_request

states:
  race_attack:
    race:
      threads: 30  # 10 requests per proxy

Resilient Testing

config:
  proxy:
    pool:
      - url: "http://primary-proxy:8080"
      - url: "http://backup-proxy:8080"
    rotation: failover
    health_check:
      enabled: true
      interval: 60
    retry:
      enabled: true
      remove_failed: true

CLI

# Multiple proxies via CLI
treco attack.yaml \
  --proxy "http://proxy1:8080" \
  --proxy "http://proxy2:8080" \
  --proxy-rotation random

Testing

# tests/test_proxy_rotation.py

def test_round_robin_rotation():
    config = ProxyPoolConfig(
        pool=[proxy1, proxy2, proxy3],
        rotation=ProxyRotation.ROUND_ROBIN
    )
    manager = ProxyManager(config)
    
    assert manager.get_proxy() == proxy1
    assert manager.get_proxy() == proxy2
    assert manager.get_proxy() == proxy3
    assert manager.get_proxy() == proxy1  # Cycles back

def test_per_thread_assignment():
    manager = ProxyManager(config)
    
    # Same thread always gets same proxy
    assert manager.get_proxy(thread_id=0) == manager.get_proxy(thread_id=0)
    
    # Different threads may get different proxies
    p0 = manager.get_proxy(thread_id=0)
    p1 = manager.get_proxy(thread_id=1)
    # Distribution depends on pool size

def test_failed_proxy_removal():
    manager = ProxyManager(config)
    
    for _ in range(3):
        manager.report_failure(proxy1)
    
    # proxy1 should be removed from healthy pool
    assert proxy1 not in manager.healthy_proxies

References

• httpx proxy docs: https://www.python-httpx.org/advanced/#http-proxying
• Proxy rotation patterns: https://scrapfly.io/blog/how-to-rotate-proxies-in-web-scraping/