crucible-proxychecker 7.4.0

Fast, multi-source proxy scraper and validator — CLI + library.

🔥 crucible-proxy

Fast, multi-source proxy scraper and validator — CLI tool and Python library.

Scrapes proxies from 67 free sources, validates them concurrently, and reports anonymity level (Elite / Anonymous / Transparent) — all in one command or one import.

v7.1.0 new: source health statistics + multi-endpoint anonymity detection (no more httpbin.org single point of failure).

v7.2.0 new: proxy cache (skip re-checking already-validated proxies) + coverage reporting.

v7.3.0 new: Session — a one-object, fully-configurable API. No glue code, configure everything declaratively (or from JSON) and call .run().

v7.4.0 new: Session shortcuts (live_only, to_list, best, fastest), callbacks (on_result/on_alive/on_progress), async (run_async), export to Clash/v2ray/Shadowsocks, and city/ASN filters.

---

Contents

• Install
• CLI Usage
  • Basic examples
  • All flags
  • Environment variables
  • Config file
• Library Usage
  • Session API (easiest)
  • Quick start
  • Custom config
  • Async-friendly pattern
  • Rotating proxy pool
  • Requests integration
  • Playwright integration
  • Filter by country / anonymity
  • Save results manually
  • Debug report
  • Proxy cache
  • Source health statistics
  • Multiple anonymity check endpoints
• Output formats
• Anonymity levels
• Config reference
• Examples
• Contributing

---

Install

pip install crucible-proxychecker

For TOML config file support on Python 3.10:

pip install "crucible-proxychecker[toml]"

Requires Python ≥ 3.10. requests[socks] and colorama are installed automatically.

---

CLI Usage

Basic examples

# Interactive menu — pick proxy types interactively
crucible-proxy

# Check one type
crucible-proxy --type socks5
crucible-proxy --type http

# Check all four types at once
crucible-proxy --all

# Save results as JSON instead of TXT
crucible-proxy --type socks5 --output-format json

# Save results as CSV
crucible-proxy --all --output-format csv --output-dir ./proxies

# Faster: more workers, shorter timeout
crucible-proxy --type http --workers 400 --timeout 5

# Slower but more accurate: verify each proxy twice
crucible-proxy --type socks4 --verify-twice true

# Verbose logging to see what's happening under the hood
crucible-proxy --type https --log-level DEBUG

# Write logs to a file
crucible-proxy --all --log-level INFO --log-file crucible.log

# Write a full debug JSON report (includes dead proxies + error messages)
crucible-proxy --type socks5 --debug-report

# No colours (e.g. for piping / CI)
crucible-proxy --type http --no-color

# Dump a config template you can edit
crucible-proxy --generate-config > crucible_proxy.toml

All flags

Flag	Default	Description
--type TYPE	—	One of http, https, socks4, socks5
--all	—	Check all four types
--workers N	200	Concurrent validation threads
--timeout SEC	8	Per-request timeout in seconds
--retries N	2	Retries before marking a proxy dead
--verify-twice BOOL	true	Confirm live proxies with a second request
--output-format FMT	txt	txt | json | csv
--output-dir DIR	./output	Where to write result files
--no-color	false	Disable ANSI colours
--log-level LEVEL	WARNING	DEBUG INFO WARNING ERROR CRITICAL
--log-file PATH	—	Also write logs to this file
--debug-report	false	Write a full JSON debug report per run
--generate-config	—	Print TOML config template and exit

Environment variables

All settings can be set via CRUCIBLE_* environment variables. Useful for Docker / CI:

export CRUCIBLE_WORKERS=400
export CRUCIBLE_TIMEOUT=5
export CRUCIBLE_OUTPUT_FORMAT=json
export CRUCIBLE_OUTPUT_DIR=/data/proxies
export CRUCIBLE_LOG_LEVEL=INFO
export CRUCIBLE_VERIFY_TWICE=false
export CRUCIBLE_DEBUG_REPORT=true
export CRUCIBLE_NO_COLOR=true

# Use your own httpbin-compatible anonymity-check server
export CRUCIBLE_ANONYMITY_CHECK_URL=http://my-server.local/headers

crucible-proxy --all

Config file

Run crucible-proxy --generate-config > crucible_proxy.toml to create a template.
The tool looks for a config file in these locations (first found wins):

1. ./crucible_proxy.toml (project root)
2. ~/.config/crucible_proxy/config.toml
3. ~/.crucible_proxy.toml

Priority order: CLI flags → env vars → config file → built-in defaults.

# crucible_proxy.toml

timeout      = 10
workers      = 300
verify_twice = false

output_dir    = "./results"
output_format = "json"

log_level    = "INFO"
log_file     = "crucible.log"
debug_report = true

anonymity_check_url = "http://my-httpbin.internal/headers"

---

Library Usage

Session API (easiest)

The Session class wraps fetching, validation, filtering, and saving into a single configurable object. Ideal if you don't want to wire functions together:

from crucible_proxy import Session

session = Session(
    types         = ["http", "socks5"],   # which protocols
    workers       = 300,                   # concurrency
    timeout       = 5,                     # per-request seconds
    verify_twice  = False,                 # second-pass confirmation
    source_timeout= 10,                    # seconds per source URL
    max_sources   = 5,                     # cap sources per type (None = all)

    # filters
    elite         = True,                  # ELITE only
    max_latency   = 1000,                  # drop slow proxies (ms)
    min_score     = 70,                    # quality score 0-100
    countries     = ["US", "DE", "TR"],    # allow-list
    exclude_ports = [8080],                # deny-list

    # output
    save          = True,
    output_format = "json",                # txt | json | csv
    output_dir    = "./proxies",
)

results = session.run()
# results -> {"http": [CheckResult, ...], "socks5": [...]}

for proto, proxies in results.items():
    print(f"{proto}: {len(proxies)} proxies passed the filters")

From a dict or JSON file

from crucible_proxy import Session

# From a dict
session = Session.from_dict({
    "types":       ["socks5"],
    "workers":     200,
    "elite":       True,
    "max_latency": 1500,
})

# From a JSON config file
session = Session.from_json("proxy_config.json")

# Save the current config back to JSON
session.save_config("proxy_config.json")

Example proxy_config.json:

{
  "types":         ["http", "socks5"],
  "workers":       300,
  "timeout":       5,
  "elite":         true,
  "max_latency":   1000,
  "countries":     ["US", "DE"],
  "output_format": "json",
  "output_dir":    "./proxies"
}

Detailed report

report = session.run_with_stats()
for proto, info in report.items():
    print(f"{proto}: {info['alive']}/{info['total']} alive, "
          f"{info['matched']} matched filters, {info['elapsed_s']}s")
    for s in info["source_stats"]:
        print(f"  {s.domain}: +{s.proxies_found}" if s.success else f"  {s.domain}: FAIL")

Shortcut methods

session = Session(types=["http", "socks5"], elite=True, max_latency=1000)

flat    = session.live_only()   # [CheckResult, ...] across all types
strings = session.to_list()     # ["1.2.3.4:8080", ...]
top10   = session.best(10)      # top 10 by quality score (0-100)
fast10  = session.fastest(10)   # 10 lowest-latency proxies

Callbacks (advanced)

Plug in your own functions — full control without writing the loop yourself:

def on_alive(r):
    print(f"LIVE {r.proxy}  {r.country}/{r.city}  {r.latency_ms}ms  {r.anonymity.value}")

def on_progress(checked, total):
    print(f"\r{checked}/{total}", end="")

session = Session(
    types       = ["http"],
    on_alive    = on_alive,
    on_progress = on_progress,
)
session.run()

Async

import asyncio
from crucible_proxy import Session

async def main():
    results = await Session(types=["http"]).run_async()
    print(sum(len(v) for v in results.values()), "proxies found")

asyncio.run(main())

Export to proxy-client formats

session = Session(types=["socks5"], elite=True)
session.run()

session.export("clash",       "proxies-clash.yaml")   # Clash YAML
session.export("v2ray",       "proxies-v2ray.json")   # v2ray outbounds JSON
session.export("shadowsocks", "proxies-ss.txt")       # ss:// list
session.export("txt",         "proxies.txt")          # plain host:port

# Or get the string directly (no path)
yaml_str = session.export("clash")

City / ASN filtering

session = Session(
    types        = ["http"],
    countries    = ["DE", "US"],
    cities       = ["Berlin", "Frankfurt"],   # substring match
    exclude_asn  = ["AS9999"],                # drop specific networks
)

Quick start

from crucible_proxy import fetch_proxies, check_all, ProxyType

# Fetch from all sources for one protocol
proxies = fetch_proxies(ProxyType.SOCKS5)
print(f"Found {len(proxies)} proxies")

# Validate concurrently; results stream as they complete
for result in check_all(proxies, max_workers=200, timeout=8):
    if result.alive:
        print(
            f"{result.proxy}  "
            f"{result.latency_ms}ms  "
            f"{result.country}  "
            f"{result.anonymity.value}"
        )

Custom config (programmatic)

from crucible_proxy import load_config, fetch_proxies, check_all, ProxyType

cfg = load_config({
    "workers":       300,
    "timeout":       6,
    "verify_twice":  False,
    "output_format": "json",
    "output_dir":    "./results",
    "log_level":     "INFO",
    "debug_report":  True,
})

proxies = fetch_proxies(ProxyType.HTTP)
results = list(check_all(proxies, max_workers=cfg.workers, timeout=cfg.timeout))

Async-friendly pattern

check_all is synchronous (uses ThreadPoolExecutor internally), but you can run it in an executor to keep an async event loop unblocked:

import asyncio
from concurrent.futures import ThreadPoolExecutor
from crucible_proxy import fetch_proxies, check_all, ProxyType

async def get_alive_proxies(proto: ProxyType, workers: int = 200):
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor(max_workers=4) as pool:
        proxies = await loop.run_in_executor(pool, fetch_proxies, proto)
        results = await loop.run_in_executor(
            pool,
            lambda: list(check_all(proxies, max_workers=workers))
        )
    return [r for r in results if r.alive]

alive = asyncio.run(get_alive_proxies(ProxyType.SOCKS5))

Rotating proxy pool

import itertools
from crucible_proxy import fetch_proxies, check_all, ProxyType

proxies  = fetch_proxies(ProxyType.HTTP)
alive    = [r.proxy for r in check_all(proxies) if r.alive]
pool     = itertools.cycle(alive)

def next_proxy() -> dict:
    """Returns a requests-compatible proxy dict."""
    return next(pool).as_requests_dict()

# Use it:
import requests
for url in my_urls:
    resp = requests.get(url, proxies=next_proxy(), timeout=10)

Requests integration

import requests
from crucible_proxy import fetch_proxies, check_all, ProxyType, Anonymity

# Get elite-only proxies
proxies = fetch_proxies(ProxyType.HTTPS)
elite = [
    r.proxy for r in check_all(proxies)
    if r.alive and r.anonymity == Anonymity.ELITE
]

proxy = elite[0]
resp  = requests.get("https://httpbin.org/ip", proxies=proxy.as_requests_dict(), timeout=10)
print(resp.json())

Playwright integration

import asyncio
from playwright.async_api import async_playwright
from crucible_proxy import fetch_proxies, check_all, ProxyType

async def scrape(url: str):
    proxies = fetch_proxies(ProxyType.SOCKS5)
    alive   = [r.proxy for r in check_all(proxies, max_workers=100) if r.alive]
    proxy   = alive[0]

    async with async_playwright() as pw:
        browser = await pw.chromium.launch(
            proxy={
                "server":   f"socks5://{proxy.host}:{proxy.port}",
            }
        )
        page = await browser.new_page()
        await page.goto(url)
        content = await page.content()
        await browser.close()
    return content

asyncio.run(scrape("https://example.com"))

Filter by country / anonymity

from crucible_proxy import fetch_proxies, check_all, ProxyType, Anonymity

results = list(check_all(fetch_proxies(ProxyType.SOCKS5)))

# Fast proxies only (< 1 second latency)
fast = [r for r in results if r.alive and r.latency_ms and r.latency_ms < 1000]

# Only US proxies
us = [r for r in results if r.alive and r.country == "US"]

# Elite only (no headers leaked)
elite = [r for r in results if r.alive and r.anonymity == Anonymity.ELITE]

# Elite US proxies sorted by speed
best = sorted(
    [r for r in results if r.alive and r.country == "US" and r.anonymity == Anonymity.ELITE],
    key=lambda r: r.latency_ms or float("inf"),
)
print(f"Best US elite proxy: {best[0].proxy} @ {best[0].latency_ms}ms")

Save results manually

from pathlib import Path
from crucible_proxy import fetch_proxies, check_all, save_results, ProxyType

results = list(check_all(fetch_proxies(ProxyType.SOCKS5)))

# TXT (default)
simple, detailed = save_results(results, ProxyType.SOCKS5)

# JSON
json_path, _ = save_results(results, ProxyType.SOCKS5, fmt="json", output_dir=Path("./out"))

# CSV
csv_path, _  = save_results(results, ProxyType.SOCKS5, fmt="csv",  output_dir=Path("./out"))

Debug report

import time
from crucible_proxy import fetch_proxies, check_all, write_debug_report, ProxyType
from pathlib import Path

proxies = fetch_proxies(ProxyType.HTTP)
t0      = time.perf_counter()
results = list(check_all(proxies))
elapsed = time.perf_counter() - t0

report_path = write_debug_report(results, ProxyType.HTTP, elapsed, Path("./debug"))
print(f"Debug report: {report_path}")
# JSON contains: meta stats, all alive proxies, all dead proxies + error messages

Proxy cache

Avoid re-checking the same proxy twice within a session using ProxyCache:

from crucible_proxy import ProxyCache, fetch_proxies, check_all, ProxyType

cache = ProxyCache()

# First run — checks all proxies over the network
proxies = fetch_proxies(ProxyType.HTTP)
results = list(check_all(proxies, cache=cache))
print(f"Checked {len(proxies)} proxies, cache now has {len(cache)} entries")

# Second run with the same list — cached proxies returned instantly
results2 = list(check_all(proxies, cache=cache))
# No network calls made for already-checked proxies

# Clear the cache when you want fresh results
cache.clear()

The cache is type-aware: the same IP address cached as HTTP will still be checked fresh as SOCKS5.

---

Source health statistics

from crucible_proxy import fetch_proxies_with_stats, ProxyType

proxies, stats = fetch_proxies_with_stats(ProxyType.HTTP)

for s in stats:
    status = f"+{s.proxies_found}" if s.success else f"FAIL: {s.error[:40]}"
    print(f"{s.domain:<45} {status}  ({s.elapsed_s}s)")

# Filter out dead sources for next run
good_sources = [s.url for s in stats if s.success and s.proxies_found > 0]

Multiple anonymity check endpoints

By default the library tries 5 httpbin-compatible endpoints in order and uses the first one that responds. This means httpbin.org downtime no longer causes all proxies to be classified as UNKNOWN.

You can override the primary endpoint via env var:

export CRUCIBLE_ANONYMITY_CHECK_URL=http://my-httpbin.internal/headers

Or inspect the full list:

from crucible_proxy.constants import ANONYMITY_CHECK_URLS
print(ANONYMITY_CHECK_URLS)
# ['http://httpbin.org/headers', 'https://httpbin.org/headers',
#  'http://httpbingo.org/headers', 'https://httpbingo.org/headers',
#  'http://eu.httpbin.org/headers']

---

Output formats

TXT (default)

Two files per run:

# simple — host:port only, sorted by latency
103.149.130.38:80
45.77.56.114:30000

# detailed — full info
103.149.130.38:80   |  212ms      |  SG    |  elite
45.77.56.114:30000  |  334ms      |  US    |  anonymous

JSON

{
  "generated": "2025-06-06T10:23:41+00:00",
  "count": 42,
  "proxies": [
    {
      "proxy": "103.149.130.38:80",
      "host": "103.149.130.38",
      "port": 80,
      "type": "http",
      "latency_ms": 212.4,
      "country": "SG",
      "anonymity": "elite",
      "check_url": "http://httpbin.org/ip"
    }
  ]
}

CSV

proxy,host,port,type,latency_ms,country,anonymity,check_url
103.149.130.38:80,103.149.130.38,80,http,212.4,SG,elite,http://httpbin.org/ip

Debug report (--debug-report)

Written in addition to normal output, includes dead proxies and error messages:

{
  "meta": {
    "generated": "2025-06-06T10:23:41+00:00",
    "proxy_type": "socks5",
    "elapsed_s": 47.2,
    "total": 1842,
    "alive": 134,
    "dead": 1708,
    "success_pct": 7.3
  },
  "alive": [ ... ],
  "dead": [
    { "proxy": "1.2.3.4:1080", "error": "ConnectTimeout" },
    ...
  ]
}

---

Anonymity levels

Level	Meaning
ELITE	No forwarding headers sent — destination cannot detect a proxy
ANONYMOUS	Proxy headers present but your real IP is hidden
TRANSPARENT	Your real IP is visible in X-Forwarded-For or similar headers
UNKNOWN	Anonymity check endpoint was unreachable

---

Config reference

Key	Env var	CLI flag	Default	Description
timeout	CRUCIBLE_TIMEOUT	--timeout	8	Per-request timeout (seconds)
workers	CRUCIBLE_WORKERS	--workers	200	Concurrent threads
retries	CRUCIBLE_RETRIES	--retries	2	Check retries before marking dead
retry_delay	CRUCIBLE_RETRY_DELAY	—	0.6	Seconds between retries
verify_twice	CRUCIBLE_VERIFY_TWICE	--verify-twice	true	Second-pass confirmation
anonymity_check_url	CRUCIBLE_ANONYMITY_CHECK_URL	—	httpbin	Headers-echo endpoint
output_dir	CRUCIBLE_OUTPUT_DIR	--output-dir	./output	Result file directory
output_format	CRUCIBLE_OUTPUT_FORMAT	--output-format	txt	txt / json / csv
no_color	CRUCIBLE_NO_COLOR	--no-color	false	Disable ANSI colours
log_level	CRUCIBLE_LOG_LEVEL	--log-level	WARNING	Log verbosity
log_file	CRUCIBLE_LOG_FILE	--log-file	—	Log file path
debug_report	CRUCIBLE_DEBUG_REPORT	--debug-report	false	Write JSON debug report

---

Examples

The examples/ directory contains ready-to-run scripts:

proxy_suite.py — Ultimate Proxy Suite

A feature-rich CLI application built on top of this library:

pip install crucible-proxychecker
python examples/proxy_suite.py

Features:

• Interactive menu + full CLI mode (--all, --types, --elite-only …)
• Configurable source count per type (--max-sources 3)
• Per-source timeout (--src-timeout 10)
• Advanced filtering: country, latency, anonymity, score, port, regex
• Proxy quality scoring (0–100, S/A/B/C/D/F grades)
• Live progress bar + colored histograms
• Re-check mode (--recheck saved.txt)
• Single proxy verify (--verify 1.2.3.4:8080)
• Bulk verify from file (--bulk list.txt)
• Rotation simulator (--rotation)
• Session archive (JSON) + debug report
• Multi-type duplicate analysis

# Examples
python examples/proxy_suite.py --all --elite-only --max-lat 1000
python examples/proxy_suite.py --types http socks5 --format json
python examples/proxy_suite.py --all --max-sources 3 --src-timeout 10
python examples/proxy_suite.py --verify 1.2.3.4:8080 --type http
python examples/proxy_suite.py --list-sources

---

Contributing

git clone https://github.com/cruciblelab/Crucible-proxychecker
cd crucible-proxy
pip install -e ".[dev,toml]"
pytest tests/ -v

See CONTRIBUTING.md for guidelines.

---

License

MIT