diff --git a/README.md b/README.md index 93e4fdc..fe20dd0 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,46 @@ python python/run_tests.py requests httpx ### Other Python scripts * [requests-random-proxy.py](python/requests-random-proxy.py) - Random proxy rotation +* [requests-waterfall-proxy.py](python/requests-waterfall-proxy.py) - Ordered proxy waterfall (cheap tiers first) +* [requests-waterfall-cache-proxy.py](python/requests-waterfall-cache-proxy.py) - Waterfall with TTL decision cache per host +* [curl-cffi-waterfall-proxy.py](python/curl-cffi-waterfall-proxy.py) - Waterfall with free TLS-fingerprint tier (`curl_cffi`) + +These waterfall examples follow the [proxy waterfall](https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04) pattern: try direct (and optionally TLS impersonation) before datacenter / residential / unlocker proxies, escalate only when content validation fails, and optionally remember the winning tier. Shared helpers live in [waterfall_common.py](python/waterfall_common.py). They are not part of `run_tests.py`. + +```bash +# At least one proxy tier required +export PROXY_URL='http://user:pass@proxy.example.com:8080' +# Optional more expensive tiers +export PROXY_URL_RESIDENTIAL='http://user:pass@residential.example:8080' +export PROXY_URL_UNLOCKER='http://user:pass@unlocker.example:8080' + +# Optional content checks (defaults work with api.ipify.org) +export EXPECT_MUST_CONTAIN='ip' +export EXPECT_MIN_BYTES=10 + +python python/requests-waterfall-proxy.py + +# Decision cache (optional file persistence) +export WATERFALL_CACHE_PATH=/tmp/waterfall-cache.json +export WATERFALL_CACHE_TTL=86400 +python python/requests-waterfall-cache-proxy.py + +# TLS fingerprint tier (install optional dep first) +pip install 'curl_cffi>=0.6.0' +python python/curl-cffi-waterfall-proxy.py +``` + +| Variable | Purpose | +|----------|---------| +| `PROXY_URL` / `PROXY_URL_DATACENTER` | Datacenter tier (also `HTTPS_PROXY`) | +| `PROXY_URL_RESIDENTIAL` | Residential tier | +| `PROXY_URL_UNLOCKER` | Managed anti-bot / unlocker tier | +| `EXPECT_STATUS` | Required HTTP status (default `200`) | +| `EXPECT_MIN_BYTES` | Minimum body length (default `0`) | +| `EXPECT_MUST_CONTAIN` | Required body substring | +| `EXPECT_BLOCK_MARKERS` | Extra soft-block markers (comma-separated) | +| `WATERFALL_CACHE_TTL` | Cache TTL seconds (default `86400`) | +| `WATERFALL_CACHE_PATH` | Optional JSON cache file | > **Note:** Like the Ruby, JavaScript, and PHP examples here, these scripts use each library's normal proxy options only. Most of them do not send custom headers on the HTTPS `CONNECT` tunnel or surface proxy `CONNECT` response headers. For that, see [python-proxy-headers](https://github.com/proxymesh/python-proxy-headers) or [scrapy-proxy-headers](https://github.com/proxymesh/scrapy-proxy-headers). diff --git a/python/curl-cffi-waterfall-proxy.py b/python/curl-cffi-waterfall-proxy.py new file mode 100644 index 0000000..0e188d3 --- /dev/null +++ b/python/curl-cffi-waterfall-proxy.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +""" +Proxy waterfall with a free TLS-fingerprint tier (article Tier 0.5). + +Same ladder and content validation as requests-waterfall-proxy.py, but inserts +a direct request that impersonates Chrome via curl_cffi before spending on +proxy IPs. Requires: pip install curl_cffi + +Tiers (skip missing optional proxies): + 1. direct — curl_cffi, no impersonation, no proxy + 2. tls — curl_cffi impersonate=chrome, no proxy + 3. datacenter — PROXY_URL / PROXY_URL_DATACENTER / HTTPS_PROXY + 4. residential — PROXY_URL_RESIDENTIAL + 5. unlocker — PROXY_URL_UNLOCKER + +Example: + pip install 'curl_cffi>=0.6.0' + export PROXY_URL='http://user:pass@us-ca.proxymesh.com:31280' + python python/curl-cffi-waterfall-proxy.py + +Article: +https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04 +""" +import sys + +from waterfall_common import ( + build_tiers, + get_test_url, + load_expect, + print_config, + print_success, + require_proxy_tier, + run_waterfall, +) + +try: + from curl_cffi import requests as curl_requests +except ImportError: + print( + 'Error: curl_cffi is required for this example\n' + " pip install 'curl_cffi>=0.6.0'", + file=sys.stderr, + ) + sys.exit(1) + + +def fetch(tier, url): + kwargs = {'timeout': 30} + if tier.proxy_url: + kwargs['proxies'] = {'http': tier.proxy_url, 'https': tier.proxy_url} + if tier.impersonate: + kwargs['impersonate'] = tier.impersonate + response = curl_requests.get(url, **kwargs) + return response.status_code, response.text + + +def main() -> int: + tiers = build_tiers(include_tls=True) + require_proxy_tier(tiers) + test_url = get_test_url() + expect = load_expect() + print_config(tiers, test_url, expect) + + result = run_waterfall(tiers, test_url, expect, fetch) + if result.ok: + print_success(result) + return 0 + + print('\nAll tiers failed.') + return 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/python/requests-waterfall-cache-proxy.py b/python/requests-waterfall-cache-proxy.py new file mode 100644 index 0000000..ed434fb --- /dev/null +++ b/python/requests-waterfall-cache-proxy.py @@ -0,0 +1,165 @@ +#!/usr/bin/env python3 +""" +Proxy waterfall with a TTL decision cache. + +Same ladder and content validation as requests-waterfall-proxy.py, plus +remembering which tier worked for a host so later requests start there +instead of re-walking cheaper tiers every time. + +Cache: + WATERFALL_CACHE_TTL Seconds to remember a winning tier (default: 86400) + WATERFALL_CACHE_PATH Optional JSON file path for persistence across runs + +On a cache hit, start at the cached tier. If that tier fails, invalidate the +entry and continue down the remaining ladder. On success, refresh the cache. + +Example: + export PROXY_URL='http://user:pass@us-ca.proxymesh.com:31280' + export PROXY_URL_RESIDENTIAL='http://user:pass@residential.example:8080' + export WATERFALL_CACHE_PATH=/tmp/waterfall-cache.json + python python/requests-waterfall-cache-proxy.py + +Article: +https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04 +""" +from __future__ import annotations + +import json +import os +import sys +import time +from pathlib import Path +from typing import Any, Dict, Optional + +import requests + +from waterfall_common import ( + build_tiers, + get_test_url, + load_expect, + print_config, + print_success, + require_proxy_tier, + run_waterfall, + tier_index_by_name, + url_pattern, +) + + +def fetch(tier, url): + proxies = None + if tier.proxy_url: + proxies = {'http': tier.proxy_url, 'https': tier.proxy_url} + response = requests.get(url, proxies=proxies, timeout=30) + return response.status_code, response.text + + +def _cache_ttl() -> int: + return int(os.environ.get('WATERFALL_CACHE_TTL', '86400')) + + +def _cache_path() -> Optional[Path]: + raw = os.environ.get('WATERFALL_CACHE_PATH') + return Path(raw) if raw else None + + +def load_cache() -> Dict[str, Any]: + path = _cache_path() + if not path or not path.is_file(): + return {} + try: + with path.open() as f: + data = json.load(f) + return data if isinstance(data, dict) else {} + except (OSError, json.JSONDecodeError) as exc: + print(f'Warning: could not read cache {path}: {exc}', file=sys.stderr) + return {} + + +def save_cache(cache: Dict[str, Any]) -> None: + path = _cache_path() + if not path: + return + try: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open('w') as f: + json.dump(cache, f, indent=2, sort_keys=True) + f.write('\n') + except OSError as exc: + print(f'Warning: could not write cache {path}: {exc}', file=sys.stderr) + + +def get_cached_tier(cache: Dict[str, Any], key: str) -> Optional[str]: + entry = cache.get(key) + if not isinstance(entry, dict): + return None + expires_at = entry.get('expires_at') + tier_name = entry.get('tier') + if not isinstance(expires_at, (int, float)) or not isinstance(tier_name, str): + return None + if time.time() >= expires_at: + return None + return tier_name + + +def set_cached_tier(cache: Dict[str, Any], key: str, tier_name: str) -> None: + cache[key] = { + 'tier': tier_name, + 'expires_at': time.time() + _cache_ttl(), + } + + +def invalidate(cache: Dict[str, Any], key: str) -> None: + cache.pop(key, None) + + +def main() -> int: + tiers = build_tiers(include_tls=False) + require_proxy_tier(tiers) + test_url = get_test_url() + expect = load_expect() + print_config(tiers, test_url, expect) + + key = url_pattern(test_url) + cache = load_cache() + cached_name = get_cached_tier(cache, key) + start_index = 0 + + if cached_name: + idx = tier_index_by_name(tiers, cached_name) + if idx is not None: + start_index = idx + print(f'Cache HIT for {key!r}: start at tier {cached_name!r} (index {idx})') + else: + print( + f'Cache HIT for {key!r}: tier {cached_name!r} not in current ladder; ' + 'starting from cheapest' + ) + invalidate(cache, key) + else: + print(f'Cache MISS for {key!r}: start at cheapest tier') + print() + + result = run_waterfall( + tiers, test_url, expect, fetch, start_index=start_index + ) + + if result.ok: + set_cached_tier(cache, key, result.tier.name) + save_cache(cache) + print(f'Cached winner {result.tier.name!r} for {key!r} (ttl={_cache_ttl()}s)') + print_success(result) + return 0 + + # Cached start failed and remaining ladder failed — drop stale entry. + if cached_name: + invalidate(cache, key) + save_cache(cache) + print(f'Invalidated cache for {key!r}') + + print('\nAll tiers failed.') + return 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/python/requests-waterfall-proxy.py b/python/requests-waterfall-proxy.py new file mode 100644 index 0000000..ef36aa4 --- /dev/null +++ b/python/requests-waterfall-proxy.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +""" +Basic proxy waterfall: try cheap tiers first, escalate only on failure. + +Uses content-level validation (status, min body size, soft-block markers, +optional must_contain) so a soft block (HTTP 200 captcha page) escalates. + +Tiers (skip missing optional proxies): + 1. direct — no proxy + 2. datacenter — PROXY_URL / PROXY_URL_DATACENTER / HTTPS_PROXY + 3. residential — PROXY_URL_RESIDENTIAL + 4. unlocker — PROXY_URL_UNLOCKER + +Configuration via environment variables (see waterfall_common.py). + +Example: + export PROXY_URL='http://user:pass@us-ca.proxymesh.com:31280' + export PROXY_URL_RESIDENTIAL='http://user:pass@residential.example:8080' + python python/requests-waterfall-proxy.py + +Article: +https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04 +""" +import sys + +import requests + +from waterfall_common import ( + build_tiers, + get_test_url, + load_expect, + print_config, + print_success, + require_proxy_tier, + run_waterfall, +) + + +def fetch(tier, url): + proxies = None + if tier.proxy_url: + proxies = {'http': tier.proxy_url, 'https': tier.proxy_url} + response = requests.get(url, proxies=proxies, timeout=30) + return response.status_code, response.text + + +def main() -> int: + tiers = build_tiers(include_tls=False) + require_proxy_tier(tiers) + test_url = get_test_url() + expect = load_expect() + print_config(tiers, test_url, expect) + + result = run_waterfall(tiers, test_url, expect, fetch) + if result.ok: + print_success(result) + return 0 + + print('\nAll tiers failed.') + return 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/python/requirements.txt b/python/requirements.txt index b744ad5..62a51ca 100644 --- a/python/requirements.txt +++ b/python/requirements.txt @@ -11,3 +11,5 @@ autoscraper>=1.1.0 beautifulsoup4>=4.12.0,<4.13 lxml>=5.0.0 scrapy>=2.11.0 +# Optional: only needed for curl-cffi-waterfall-proxy.py (Tier 0.5 TLS fingerprint) +# curl_cffi>=0.6.0 diff --git a/python/waterfall_common.py b/python/waterfall_common.py new file mode 100644 index 0000000..5e92aa8 --- /dev/null +++ b/python/waterfall_common.py @@ -0,0 +1,246 @@ +""" +Shared helpers for proxy waterfall examples. + +Pattern from: +https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04 + +Environment: + TEST_URL Target URL (default: https://api.ipify.org?format=json) + PROXY_URL Datacenter / default proxy (also PROXY_URL_DATACENTER / HTTPS_PROXY) + PROXY_URL_RESIDENTIAL Optional residential proxy tier + PROXY_URL_UNLOCKER Optional managed anti-bot / unlocker proxy tier + EXPECT_STATUS Required HTTP status (default: 200) + EXPECT_MIN_BYTES Minimum response body length (default: 0) + EXPECT_MUST_CONTAIN Substring that must appear in the body (optional) + EXPECT_BLOCK_MARKERS Extra soft-block markers, comma-separated (optional) +""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Callable, List, Optional, Sequence, Tuple +from urllib.parse import urlparse + +DEFAULT_TEST_URL = 'https://api.ipify.org?format=json' + +# Soft-block / challenge substrings (content-level failure, not just status). +DEFAULT_BLOCK_MARKERS = ( + 'captcha', + 'cf-challenge', + 'challenge-platform', + 'access denied', + 'attention required', + 'verify you are human', + 'just a moment', +) + + +@dataclass(frozen=True) +class Tier: + name: str + proxy_url: Optional[str] # None = direct (no proxy) + impersonate: Optional[str] = None # e.g. "chrome" for curl_cffi Tier 0.5 + + +@dataclass +class Expect: + status: int = 200 + min_bytes: int = 0 + must_contain: Optional[str] = None + block_markers: Tuple[str, ...] = DEFAULT_BLOCK_MARKERS + + +@dataclass +class AttemptResult: + tier: Tier + ok: bool + reason: str + status_code: Optional[int] = None + body: Optional[str] = None + + +def mask_proxy_url(url: Optional[str]) -> str: + """Mask password in a proxy URL for logging.""" + if not url: + return '(direct)' + try: + parsed = urlparse(url) + if parsed.password: + return url.replace(f':{parsed.password}@', ':****@', 1) + return url + except Exception: + return url + + +def load_expect() -> Expect: + status = int(os.environ.get('EXPECT_STATUS', '200')) + min_bytes = int(os.environ.get('EXPECT_MIN_BYTES', '0')) + must_contain = os.environ.get('EXPECT_MUST_CONTAIN') or None + extra = os.environ.get('EXPECT_BLOCK_MARKERS', '') + markers = list(DEFAULT_BLOCK_MARKERS) + if extra.strip(): + markers.extend(m.strip() for m in extra.split(',') if m.strip()) + return Expect( + status=status, + min_bytes=min_bytes, + must_contain=must_contain, + block_markers=tuple(markers), + ) + + +def get_test_url() -> str: + return os.environ.get('TEST_URL', DEFAULT_TEST_URL) + + +def _datacenter_proxy() -> Optional[str]: + return ( + os.environ.get('PROXY_URL_DATACENTER') + or os.environ.get('PROXY_URL') + or os.environ.get('HTTPS_PROXY') + ) + + +def build_tiers(*, include_tls: bool = False) -> List[Tier]: + """ + Build the cheap-to-expensive ladder. Missing optional proxy tiers are skipped. + """ + tiers: List[Tier] = [Tier(name='direct', proxy_url=None)] + if include_tls: + tiers.append(Tier(name='tls', proxy_url=None, impersonate='chrome')) + + datacenter = _datacenter_proxy() + if datacenter: + tiers.append(Tier(name='datacenter', proxy_url=datacenter)) + + residential = os.environ.get('PROXY_URL_RESIDENTIAL') + if residential: + tiers.append(Tier(name='residential', proxy_url=residential)) + + unlocker = os.environ.get('PROXY_URL_UNLOCKER') + if unlocker: + tiers.append(Tier(name='unlocker', proxy_url=unlocker)) + + return tiers + + +def require_proxy_tier(tiers: Sequence[Tier]) -> None: + """Fail fast when no proxy tier is configured (keep these as proxy examples).""" + if any(t.proxy_url for t in tiers): + return + raise SystemExit( + 'Error: Set at least one proxy tier env var\n' + " PROXY_URL / PROXY_URL_DATACENTER (datacenter)\n" + ' PROXY_URL_RESIDENTIAL (optional)\n' + ' PROXY_URL_UNLOCKER (optional)\n' + "Example: export PROXY_URL='http://user:pass@proxy.example.com:8080'" + ) + + +def is_good(status_code: int, body: str, expect: Expect) -> Tuple[bool, str]: + """Content-level validation — soft blocks fail even on HTTP 200.""" + if status_code != expect.status: + return False, f'status {status_code} != {expect.status}' + if len(body) < expect.min_bytes: + return False, f'body {len(body)} bytes < min {expect.min_bytes}' + lower = body.lower() + for marker in expect.block_markers: + if marker.lower() in lower: + return False, f'soft-block marker {marker!r}' + if expect.must_contain is not None and expect.must_contain not in body: + return False, f'missing must_contain {expect.must_contain!r}' + return True, 'ok' + + +def url_pattern(url: str) -> str: + """Cache key: hostname (article uses URL pattern / domain).""" + parsed = urlparse(url) + return parsed.netloc.lower() or url + + +def describe_tier(tier: Tier) -> str: + parts = [tier.name] + if tier.impersonate: + parts.append(f'impersonate={tier.impersonate}') + parts.append(mask_proxy_url(tier.proxy_url)) + return ' | '.join(parts) + + +def print_config(tiers: Sequence[Tier], test_url: str, expect: Expect) -> None: + print(f'Test URL: {test_url}') + print(f'Expect: status={expect.status} min_bytes={expect.min_bytes}') + if expect.must_contain: + print(f' must_contain={expect.must_contain!r}') + print('Tiers:') + for t in tiers: + print(f' - {describe_tier(t)}') + print() + + +Fetcher = Callable[[Tier, str], Tuple[int, str]] + + +def run_waterfall( + tiers: Sequence[Tier], + url: str, + expect: Expect, + fetch: Fetcher, + *, + start_index: int = 0, +) -> AttemptResult: + """ + Walk tiers from start_index onward. Returns the first good AttemptResult, + or the last failure if all tiers fail. + """ + if start_index < 0 or start_index >= len(tiers): + start_index = 0 + + last: Optional[AttemptResult] = None + for tier in tiers[start_index:]: + print(f'Trying tier: {describe_tier(tier)}') + try: + status_code, body = fetch(tier, url) + except Exception as exc: + last = AttemptResult( + tier=tier, + ok=False, + reason=f'request error: {exc}', + ) + print(f' FAIL: {last.reason}') + continue + + ok, reason = is_good(status_code, body, expect) + last = AttemptResult( + tier=tier, + ok=ok, + reason=reason, + status_code=status_code, + body=body, + ) + if ok: + print(f' OK: {reason}') + return last + print(f' FAIL: {reason} (status={status_code}, bytes={len(body)})') + + if last is None: + return AttemptResult( + tier=tiers[0] if tiers else Tier(name='none', proxy_url=None), + ok=False, + reason='no tiers to try', + ) + return last + + +def print_success(result: AttemptResult) -> None: + body = result.body or '' + snippet = body if len(body) <= 500 else body[:500] + '...' + print() + print(f'Winner: {describe_tier(result.tier)}') + print(f'Status: {result.status_code}') + print(f'Body: {snippet}') + + +def tier_index_by_name(tiers: Sequence[Tier], name: str) -> Optional[int]: + for i, t in enumerate(tiers): + if t.name == name: + return i + return None