proxyprobe.py

#!/usr/bin/env python3
"""proxyprobe - check a list of proxies for liveness, latency, exit IP and anonymity.

Single file, one dependency (httpx). Run it directly or install it:

    python proxyprobe.py proxies.txt
    proxyprobe proxies.txt --json

Proxy list format, one per line (blank lines and # comments are ignored):

    http://user:pass@host:8080
    socks5://host:1080
    host:8080                  # scheme defaults to http

MIT licensed. https://github.com/roamproxy/proxyprobe
"""
from __future__ import annotations

import argparse
import asyncio
import json
import re
import sys
import time
from dataclasses import dataclass, asdict, field
from typing import Iterable, Sequence

try:
    import httpx
except ImportError:  # pragma: no cover - only hit when dependency is missing
    sys.exit("proxyprobe needs httpx. Install it with:  pip install 'httpx[socks]'")

__version__ = "0.1.0"

# Endpoint that echoes the caller's IP as plain text. Kept deliberately neutral and
# overridable (--echo-url) so nobody has to route their proxy list through a host
# they don't trust.
DEFAULT_ECHO_URL = "https://api.ipify.org"
# Echoes the request headers back as JSON, used to grade anonymity.
DEFAULT_HEADERS_URL = "https://httpbin.org/headers"
# Free, key-less geo lookup. Rate limited, so it is opt-in via --geo.
DEFAULT_GEO_URL = "http://ip-api.com/json/{ip}?fields=status,country,countryCode,city"

SCHEMES = ("http://", "https://", "socks5://", "socks5h://", "socks4://")

# Headers that leak the fact a proxy is in play, or the client behind it.
PROXY_HEADERS = (
    "via",
    "x-forwarded-for",
    "x-real-ip",
    "forwarded",
    "proxy-connection",
    "x-proxy-id",
)


@dataclass
class Result:
    proxy: str                      # password-masked, safe to print or log
    ok: bool = False
    latency_ms: float | None = None
    exit_ip: str | None = None
    anonymity: str | None = None    # elite | anonymous | transparent
    country: str | None = None
    city: str | None = None
    error: str | None = None
    leaked_headers: list[str] = field(default_factory=list)


def mask(proxy: str) -> str:
    """Hide the password so results can be pasted into an issue or a CI log.

    The password is matched greedily up to the *last* '@': passwords routinely
    contain '@' and ':', and a lazy match would leave the tail of the secret in
    the output.
    """
    return re.sub(r"://([^:/@]+):(.*)@", r"://\1:***@", proxy)


def normalize(line: str) -> str | None:
    """Turn one input line into a proxy URL, or None if the line is not one."""
    line = line.strip()
    if not line or line.startswith("#"):
        return None
    if not line.startswith(SCHEMES):
        line = "http://" + line
    return line


def parse_proxies(lines: Iterable[str]) -> list[str]:
    """Parse a proxy list, dropping blanks/comments and de-duplicating in order."""
    seen: dict[str, None] = {}
    for line in lines:
        p = normalize(line)
        if p is not None:
            seen.setdefault(p, None)
    return list(seen)


def grade_anonymity(headers: dict, direct_ip: str | None, exit_ip: str | None) -> tuple[str, list[str]]:
    """Classify how much the proxy gives away.

    transparent - our real IP is visible through the proxy
    anonymous   - real IP hidden, but proxy headers announce a proxy is in use
    elite       - looks like an ordinary direct request
    """
    lowered = {k.lower(): str(v) for k, v in headers.items()}
    leaked = [h for h in PROXY_HEADERS if h in lowered]

    if direct_ip and any(direct_ip in v for v in lowered.values()):
        return "transparent", leaked
    if leaked:
        return "anonymous", leaked
    return "elite", leaked


async def probe_one(
    proxy: str,
    *,
    client_kwargs: dict,
    echo_url: str,
    headers_url: str,
    geo_url: str | None,
    direct_ip: str | None,
) -> Result:
    res = Result(proxy=mask(proxy))
    started = time.perf_counter()
    try:
        async with httpx.AsyncClient(proxy=proxy, **client_kwargs) as client:
            r = await client.get(echo_url)
            r.raise_for_status()
            res.latency_ms = round((time.perf_counter() - started) * 1000, 1)
            res.exit_ip = r.text.strip()
            res.ok = True

            # Anonymity needs a second call; a failure here must not sink a
            # proxy that already proved it works.
            try:
                hr = await client.get(headers_url)
                payload = hr.json().get("headers", {})
                res.anonymity, res.leaked_headers = grade_anonymity(payload, direct_ip, res.exit_ip)
            except Exception:
                pass

            if geo_url and res.exit_ip:
                try:
                    gr = await client.get(geo_url.format(ip=res.exit_ip))
                    g = gr.json()
                    if g.get("status") == "success":
                        res.country = g.get("country")
                        res.city = g.get("city")
                except Exception:
                    pass
    except Exception as exc:
        res.error = f"{type(exc).__name__}: {exc}"[:120] or "unknown error"
    return res


async def probe_all(
    proxies: Sequence[str],
    *,
    concurrency: int,
    timeout: float,
    echo_url: str,
    headers_url: str,
    geo_url: str | None,
    direct_ip: str | None,
    progress: bool = False,
) -> list[Result]:
    sem = asyncio.Semaphore(max(1, concurrency))
    client_kwargs = {"timeout": timeout, "follow_redirects": True}
    done = 0

    async def worker(p: str) -> Result:
        nonlocal done
        async with sem:
            r = await probe_one(
                p,
                client_kwargs=client_kwargs,
                echo_url=echo_url,
                headers_url=headers_url,
                geo_url=geo_url,
                direct_ip=direct_ip,
            )
        done += 1
        if progress:
            print(f"\r  checked {done}/{len(proxies)}", end="", file=sys.stderr, flush=True)
        return r

    results = await asyncio.gather(*(worker(p) for p in proxies))
    if progress:
        print("\r" + " " * 32 + "\r", end="", file=sys.stderr, flush=True)
    return list(results)


async def get_direct_ip(echo_url: str, timeout: float) -> str | None:
    """Our IP without a proxy - the baseline that makes 'transparent' detectable."""
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            r = await client.get(echo_url)
            return r.text.strip()
    except Exception:
        return None


def render_table(results: Sequence[Result], show_geo: bool) -> str:
    cols = ["PROXY", "OK", "MS", "EXIT IP", "ANONYMITY"]
    if show_geo:
        cols += ["LOCATION"]
    cols += ["ERROR"]

    rows = []
    for r in results:
        location = " ".join(x for x in (r.city, r.country) if x)
        row = [
            r.proxy,
            "yes" if r.ok else "no",
            f"{r.latency_ms:.0f}" if r.latency_ms is not None else "-",
            r.exit_ip or "-",
            r.anonymity or "-",
        ]
        if show_geo:
            row.append(location or "-")
        row.append(r.error or "")
        rows.append(row)

    widths = [len(c) for c in cols]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(cell))
    # Keep the error column from pushing the table off screen.
    widths[-1] = min(widths[-1], 46)

    def line(cells: Sequence[str]) -> str:
        out = []
        for i, cell in enumerate(cells):
            cell = cell if len(cell) <= widths[i] else cell[: widths[i] - 1] + "…"
            out.append(cell.ljust(widths[i]))
        return "  ".join(out).rstrip()

    parts = [line(cols), line(["-" * w for w in widths])]
    parts += [line(r) for r in rows]
    return "\n".join(parts)


def summarize(results: Sequence[Result]) -> str:
    ok = [r for r in results if r.ok]
    if not results:
        return "no proxies checked"
    lat = sorted(r.latency_ms for r in ok if r.latency_ms is not None)
    median = f"{lat[len(lat) // 2]:.0f}ms median" if lat else "no timings"
    grades = {}
    for r in ok:
        if r.anonymity:
            grades[r.anonymity] = grades.get(r.anonymity, 0) + 1
    grade_str = ", ".join(f"{v} {k}" for k, v in sorted(grades.items())) or "anonymity not graded"
    return f"{len(ok)}/{len(results)} working  |  {median}  |  {grade_str}"


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="proxyprobe",
        description="Check proxies for liveness, latency, exit IP and anonymity.",
        epilog="Proxy list: one per line, '#' comments allowed. Use '-' to read stdin.",
    )
    p.add_argument("file", help="file with one proxy per line, or '-' for stdin")
    p.add_argument("-c", "--concurrency", type=int, default=20, help="parallel checks (default: 20)")
    p.add_argument("-t", "--timeout", type=float, default=10.0, help="per-request timeout in seconds (default: 10)")
    p.add_argument("--json", action="store_true", help="emit JSON instead of a table")
    p.add_argument("--geo", action="store_true", help="also look up country/city (rate limited)")
    p.add_argument("--working-only", action="store_true", help="only output proxies that responded")
    p.add_argument("--echo-url", default=DEFAULT_ECHO_URL, help=f"IP echo endpoint (default: {DEFAULT_ECHO_URL})")
    p.add_argument("--headers-url", default=DEFAULT_HEADERS_URL, help="header echo endpoint used for anonymity grading")
    p.add_argument("--geo-url", default=DEFAULT_GEO_URL, help="geo lookup URL template, must contain {ip}")
    p.add_argument("-q", "--quiet", action="store_true", help="suppress the progress counter")
    p.add_argument("-V", "--version", action="version", version=f"proxyprobe {__version__}")
    return p


def main(argv: Sequence[str] | None = None) -> int:
    args = build_parser().parse_args(argv)

    if args.file == "-":
        lines = sys.stdin.read().splitlines()
    else:
        try:
            with open(args.file, encoding="utf-8") as fh:
                lines = fh.read().splitlines()
        except OSError as exc:
            print(f"proxyprobe: cannot read {args.file}: {exc}", file=sys.stderr)
            return 2

    proxies = parse_proxies(lines)
    if not proxies:
        print("proxyprobe: no proxies found in input", file=sys.stderr)
        return 2

    show_progress = not args.quiet and not args.json and sys.stderr.isatty()

    direct_ip = asyncio.run(get_direct_ip(args.echo_url, args.timeout))
    results = asyncio.run(
        probe_all(
            proxies,
            concurrency=args.concurrency,
            timeout=args.timeout,
            echo_url=args.echo_url,
            headers_url=args.headers_url,
            geo_url=args.geo_url if args.geo else None,
            direct_ip=direct_ip,
            progress=show_progress,
        )
    )

    if args.working_only:
        results = [r for r in results if r.ok]

    if args.json:
        print(json.dumps([asdict(r) for r in results], indent=2))
    else:
        if results:
            print(render_table(results, show_geo=args.geo))
            print()
        print(summarize(results))

    # Exit 1 when nothing worked, so this is usable in CI and shell pipelines.
    return 0 if any(r.ok for r in results) else 1


if __name__ == "__main__":
    raise SystemExit(main())