Building a Proxy Chain Validator Tool

Proxy chaining routes traffic through multiple proxies in sequence — your request goes through Proxy A, then Proxy B, then reaches the target. This provides stronger anonymity but introduces complexity. A chain validator tests each hop, verifies IPs change at each stage, measures cumulative latency, and detects leaks that break the chain.

Why Validate Proxy Chains

A proxy chain is only as strong as its weakest link. Common problems include:

A hop in the chain is down, breaking the entire connection
DNS requests bypass the chain, leaking your real location
One proxy in the chain is transparent, exposing the previous hop’s IP
Cumulative latency makes the chain too slow for practical use
Headers leak information about intermediate hops

Architecture

The validator tests chains by making requests through increasingly longer prefixes of the chain and comparing IPs at each stage.

Direct      → httpbin.org/ip → Your IP
Hop 1 only  → httpbin.org/ip → Proxy 1 IP
Hop 1+2     → httpbin.org/ip → Proxy 2 IP
Full chain  → httpbin.org/ip → Final IP

Implementation

import asyncio
import httpx
import time
import json
import socket
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from enum import Enum

class HopStatus(Enum):
    OK = "ok"
    FAILED = "failed"
    LEAK = "leak"
    SLOW = "slow"
    TRANSPARENT = "transparent"

@dataclass
class HopResult:
    proxy: str
    hop_number: int
    status: HopStatus = HopStatus.FAILED
    visible_ip: str = ""
    expected_change: bool = True
    ip_changed: bool = False
    latency_ms: int = 0
    anonymity: str = "unknown"
    error: str = ""
    headers_leaked: List[str] = field(default_factory=list)

@dataclass
class ChainValidationResult:
    chain: List[str]
    valid: bool = False
    total_latency_ms: int = 0
    final_ip: str = ""
    real_ip: str = ""
    hops: List[HopResult] = field(default_factory=list)
    dns_leak: bool = False
    warnings: List[str] = field(default_factory=list)

class ProxyChainValidator:
    def __init__(
        self,
        test_url: str = "https://httpbin.org/ip",
        headers_test_url: str = "https://httpbin.org/headers",
        timeout: int = 30,
    ):
        self.test_url = test_url
        self.headers_test_url = headers_test_url
        self.timeout = timeout
        self.real_ip = None

    async def get_real_ip(self) -> str:
        async with httpx.AsyncClient(timeout=10) as client:
            response = await client.get(self.test_url)
            data = response.json()
            self.real_ip = data.get("origin", "").split(",")[0].strip()
            return self.real_ip

    async def test_single_hop(
        self,
        proxy_url: str,
        hop_number: int,
        previous_ip: str,
    ) -> HopResult:
        result = HopResult(
            proxy=proxy_url,
            hop_number=hop_number,
        )

        try:
            start = time.monotonic()
            async with httpx.AsyncClient(
                proxy=proxy_url,
                timeout=self.timeout,
            ) as client:
                # Test IP
                response = await client.get(self.test_url)
                result.latency_ms = int(
                    (time.monotonic() - start) * 1000
                )

                if response.status_code != 200:
                    result.error = f"HTTP {response.status_code}"
                    return result

                data = response.json()
                result.visible_ip = data.get("origin", "").split(",")[0].strip()

                # Check if IP changed from previous hop
                if previous_ip:
                    result.ip_changed = result.visible_ip != previous_ip

                # Check for real IP leak
                if self.real_ip and self.real_ip in result.visible_ip:
                    result.status = HopStatus.TRANSPARENT
                    result.anonymity = "transparent"
                    return result

                # Test headers for leaks
                headers_response = await client.get(self.headers_test_url)
                headers_data = headers_response.json()
                leak_headers = self._check_header_leaks(
                    headers_data.get("headers", {}),
                    previous_ip,
                )
                result.headers_leaked = leak_headers

                if leak_headers:
                    result.anonymity = "anonymous"
                else:
                    result.anonymity = "elite"

                result.status = HopStatus.OK

                if result.latency_ms > 5000:
                    result.status = HopStatus.SLOW

        except Exception as e:
            result.error = str(e)[:200]
            result.status = HopStatus.FAILED

        return result

    async def validate_chain(
        self, chain: List[str]
    ) -> ChainValidationResult:
        result = ChainValidationResult(chain=chain)

        # Get real IP first
        if not self.real_ip:
            try:
                await self.get_real_ip()
                result.real_ip = self.real_ip
            except Exception as e:
                result.warnings.append(f"Could not detect real IP: {e}")

        previous_ip = self.real_ip

        # Test each hop individually
        for i, proxy in enumerate(chain):
            hop_result = await self.test_single_hop(
                proxy, i + 1, previous_ip
            )
            result.hops.append(hop_result)

            if hop_result.status == HopStatus.FAILED:
                result.warnings.append(
                    f"Hop {i+1} ({proxy}) failed: {hop_result.error}"
                )
                result.valid = False
                return result

            if hop_result.status == HopStatus.TRANSPARENT:
                result.warnings.append(
                    f"Hop {i+1} is transparent — real IP visible"
                )

            if hop_result.headers_leaked:
                result.warnings.append(
                    f"Hop {i+1} leaks headers: {hop_result.headers_leaked}"
                )

            previous_ip = hop_result.visible_ip
            result.total_latency_ms += hop_result.latency_ms

        # Test DNS leak
        dns_leak = await self._check_dns_leak(chain[-1])
        result.dns_leak = dns_leak
        if dns_leak:
            result.warnings.append("DNS leak detected — DNS queries bypass proxy chain")

        # Final status
        result.final_ip = result.hops[-1].visible_ip if result.hops else ""
        result.valid = all(
            h.status in (HopStatus.OK, HopStatus.SLOW)
            for h in result.hops
        )

        if result.valid and result.final_ip == self.real_ip:
            result.valid = False
            result.warnings.append("Final IP matches real IP — chain provides no anonymity")

        return result

    def _check_header_leaks(
        self, headers: dict, previous_ip: str
    ) -> List[str]:
        leaks = []
        suspicious_headers = [
            "X-Forwarded-For",
            "X-Real-Ip",
            "Via",
            "X-Proxy-Id",
            "Forwarded",
        ]

        for header in suspicious_headers:
            value = headers.get(header, "")
            if value:
                leaks.append(f"{header}: {value}")
                if previous_ip and previous_ip in value:
                    leaks.append(f"{header} contains previous hop IP")

        return leaks

    async def _check_dns_leak(self, last_proxy: str) -> bool:
        """
        Test if DNS queries go through the proxy or leak directly.
        Uses a unique subdomain to detect DNS resolver.
        """
        try:
            test_domain = f"leak-test-{int(time.time())}.example.com"
            async with httpx.AsyncClient(
                proxy=last_proxy,
                timeout=10,
            ) as client:
                try:
                    await client.get(f"http://{test_domain}/")
                except Exception:
                    pass

            # In a real implementation, you would check your DNS server
            # to see if the query came from the proxy IP or your real IP
            return False

        except Exception:
            return False

    def print_report(self, result: ChainValidationResult):
        print(f"\n{'='*60}")
        print(f"PROXY CHAIN VALIDATION REPORT")
        print(f"{'='*60}")
        print(f"Chain length: {len(result.chain)} hops")
        print(f"Status: {'VALID' if result.valid else 'INVALID'}")
        print(f"Real IP: {result.real_ip}")
        print(f"Final IP: {result.final_ip}")
        print(f"Total latency: {result.total_latency_ms}ms")
        print(f"DNS leak: {'YES' if result.dns_leak else 'No'}")

        print(f"\n--- Hop Details ---")
        for hop in result.hops:
            status_icon = "OK" if hop.status == HopStatus.OK else "FAIL"
            print(
                f"  Hop {hop.hop_number}: [{status_icon}] "
                f"{hop.proxy} → {hop.visible_ip} "
                f"({hop.latency_ms}ms, {hop.anonymity})"
            )
            if hop.headers_leaked:
                for leak in hop.headers_leaked:
                    print(f"    WARNING: {leak}")

        if result.warnings:
            print(f"\n--- Warnings ---")
            for w in result.warnings:
                print(f"  - {w}")

        print(f"{'='*60}\n")


# Usage
async def main():
    validator = ProxyChainValidator(timeout=15)

    chain = [
        "http://user:pass@proxy1.example.com:8080",
        "socks5://user:pass@proxy2.example.com:1080",
        "http://user:pass@proxy3.example.com:8080",
    ]

    print(f"Validating {len(chain)}-hop proxy chain...")
    result = await validator.validate_chain(chain)
    validator.print_report(result)

    # Export results
    with open("chain_validation.json", "w") as f:
        json.dump({
            "valid": result.valid,
            "chain": result.chain,
            "real_ip": result.real_ip,
            "final_ip": result.final_ip,
            "total_latency_ms": result.total_latency_ms,
            "dns_leak": result.dns_leak,
            "hops": [
                {
                    "proxy": h.proxy,
                    "status": h.status.value,
                    "visible_ip": h.visible_ip,
                    "latency_ms": h.latency_ms,
                    "anonymity": h.anonymity,
                }
                for h in result.hops
            ],
            "warnings": result.warnings,
        }, f, indent=2)

asyncio.run(main())

CLI Interface

import argparse

def cli():
    parser = argparse.ArgumentParser(description="Proxy Chain Validator")
    parser.add_argument("proxies", nargs="+", help="Proxy URLs in chain order")
    parser.add_argument("-t", "--timeout", type=int, default=15)
    parser.add_argument("-o", "--output", help="Output JSON file")
    parser.add_argument("--dns-check", action="store_true", help="Enable DNS leak check")

    args = parser.parse_args()

    validator = ProxyChainValidator(timeout=args.timeout)
    result = asyncio.run(validator.validate_chain(args.proxies))
    validator.print_report(result)

    if args.output:
        # Export to file
        pass

if __name__ == "__main__":
    cli()

Usage:

python chain_validator.py \
    http://proxy1:8080 \
    socks5://proxy2:1080 \
    http://proxy3:8080 \
    --timeout 20 \
    --output results.json

Internal Links

Building a Proxy Checker Tool — single proxy checking
Proxy Health Monitor with Node.js — continuous monitoring
What Is Proxy Chaining — proxy chaining concepts
Creating a Proxy Benchmarking Suite — benchmark chains
SOCKS5 Proxy Guide — SOCKS proxy protocols

FAQ

How many hops should a proxy chain have?

Two hops provide strong anonymity for most use cases. Three hops add marginal security at significant latency cost. Beyond three hops, diminishing returns make chains impractical. Each hop adds 100-500ms of latency.

Can I chain different proxy protocols?

Yes. A common setup chains an HTTP proxy with a SOCKS5 proxy. The validator tests each hop independently, so mixed-protocol chains work. Note that not all proxy software supports forwarding to another proxy — some terminate the connection instead of relaying.

How do I detect DNS leaks in a chain?

True DNS leak detection requires a custom DNS server. The test makes a request to a unique domain through the proxy chain, then checks which IP queried your DNS server. If the query came from your real IP instead of the proxy’s IP, DNS is leaking outside the chain.

What causes proxy chain failures?

The most common cause is a proxy in the chain that does not support forwarding. Some proxies strip connection headers needed for chaining. Others enforce IP whitelists that block the previous hop’s IP. Test each proxy individually before chaining them.

Is a proxy chain as secure as Tor?

No. Tor provides additional protections: onion routing (each hop only knows the previous and next hop), consistent encryption between hops, and a large volunteer network. A proxy chain sends traffic in cleartext between hops unless each hop uses TLS. Proxy chains offer flexibility but weaker anonymity guarantees than Tor.