"""Global outbound proxy and Cloudflare clearance helpers.""" from __future__ import annotations from dataclasses import dataclass, field, replace import json import re import threading import time from typing import Callable, Mapping from urllib import request as urllib_request from urllib.parse import quote, urlparse from curl_cffi.requests import Session from services.config import config FlareSolverrRequestMethod = Callable[[str, bytes, dict[str, str], float], bytes] def normalize_proxy_url(url: str) -> str: """Normalize proxy URLs for curl_cffi. SOCKS proxies should use remote-DNS resolution by default, so generic ``socks://`` and ``socks5://`` inputs are upgraded to ``socks5h://``. HTTP/HTTPS/socks5h inputs are otherwise left untouched except trimming. """ candidate = str(url or "").strip() if candidate and "://" not in candidate: candidate = _colon_proxy_to_url(candidate) lowered = candidate.lower() if lowered.startswith("socks://"): return "socks5h://" + candidate[len("socks://") :] if lowered.startswith("socks5://"): return "socks5h://" + candidate[len("socks5://") :] return candidate @dataclass(frozen=True) class ProxyRuntimeProfile: proxy_url: str = "" proxy_source: str = "direct" resource: bool = False runtime_enabled: bool = False egress_mode: str = "direct" skip_ssl_verify: bool = False reset_session_status_codes: tuple[int, ...] = field(default_factory=lambda: (403,)) clearance: dict[str, object] = field(default_factory=dict, repr=False) @property def clearance_enabled(self) -> bool: return ( self.runtime_enabled and bool(self.clearance.get("enabled")) and self.clearance_mode in {"manual", "flaresolverr"} ) @property def clearance_mode(self) -> str: return str(self.clearance.get("mode") or "none").strip().lower() @property def refresh_interval(self) -> int: try: return max(0, int(self.clearance.get("refresh_interval") or 0)) except (OverflowError, TypeError, ValueError): return 0 @property def timeout_sec(self) -> int: try: return max(1, int(self.clearance.get("timeout_sec") or 60)) except (OverflowError, TypeError, ValueError): return 60 @dataclass(frozen=True) class ClearanceBundle: target_host: str proxy_url: str = "" cookies: dict[str, str] = field(default_factory=dict, repr=False) user_agent: str = "" created_at: float = field(default_factory=time.time) expires_at: float | None = None def is_valid_for(self, target_host: str, proxy_url: str, *, now: float | None = None) -> bool: host = _normalize_host(target_host) if self.target_host and host and _normalize_host(self.target_host) != host: return False if normalize_proxy_url(self.proxy_url) != normalize_proxy_url(proxy_url): return False if self.expires_at is not None and (time.time() if now is None else now) >= self.expires_at: return False return bool(self.cookies or self.user_agent) def cookie_header(self) -> str: return _cookies_to_header(self.cookies) class FlareSolverrClearanceProvider: def __init__(self, flaresolverr_url: str, request_method: FlareSolverrRequestMethod | None = None) -> None: self.flaresolverr_url = str(flaresolverr_url or "").strip().rstrip("/") self._request_method = request_method or self._urllib_post def get_clearance(self, target_url: str, proxy_url: str = "", timeout_sec: int = 60) -> ClearanceBundle | None: if not self.flaresolverr_url: return None timeout = _coerce_timeout(timeout_sec) payload: dict[str, object] = { "cmd": "request.get", "url": str(target_url or ""), "maxTimeout": int(timeout * 1000), } proxy_url = normalize_proxy_url(proxy_url) if proxy_url: payload["proxy"] = {"url": proxy_url} endpoint = f"{self.flaresolverr_url}/v1" try: body = json.dumps(payload).encode("utf-8") raw_response = self._request_method( endpoint, body, {"Content-Type": "application/json"}, timeout, ) data = json.loads(raw_response.decode("utf-8") if isinstance(raw_response, bytes) else raw_response) except Exception: return None if not isinstance(data, dict) or str(data.get("status") or "").lower() != "ok": return None solution = data.get("solution") if not isinstance(solution, dict): return None target_host = _host_from_url(target_url) cookies = _filter_flaresolverr_cookies(solution.get("cookies"), target_host) user_agent = str(solution.get("userAgent") or "").strip() if not cookies and not user_agent: return None return ClearanceBundle( target_host=target_host, proxy_url=proxy_url, cookies=cookies, user_agent=user_agent, ) @staticmethod def _urllib_post(endpoint: str, body: bytes, headers: dict[str, str], timeout: float) -> bytes: req = urllib_request.Request(endpoint, data=body, headers=headers, method="POST") with urllib_request.urlopen(req, timeout=timeout) as response: return response.read() class ProxySettingsStore: def __init__( self, config_store=None, clearance_provider_factory: Callable[[str], FlareSolverrClearanceProvider] | None = None, ) -> None: self._config = config_store or config self._clearance_provider_factory = clearance_provider_factory or FlareSolverrClearanceProvider self._clearance_cache: dict[tuple[str, str], ClearanceBundle] = {} self._provider_cache: dict[str, FlareSolverrClearanceProvider] = {} self._flight_locks: dict[tuple[str, str], threading.Lock] = {} self._lock = threading.RLock() def get_profile( self, account: dict | None = None, proxy: str = "", resource: bool = False, upstream: bool = False, ) -> ProxyRuntimeProfile: runtime = self._get_runtime_settings() clearance = dict(runtime.get("clearance") if isinstance(runtime.get("clearance"), dict) else {}) runtime_enabled = bool(runtime.get("enabled")) egress_mode = str(runtime.get("egress_mode") or "direct").strip().lower() account_proxy = _clean((account or {}).get("proxy") if isinstance(account, dict) else "") explicit_proxy = _clean(proxy) legacy_proxy = _clean(self._config.get_proxy_settings()) runtime_proxy = "" runtime_proxy_source = "runtime" if upstream and runtime_enabled and egress_mode == "single_proxy": resource_proxy = _clean(runtime.get("resource_proxy_url")) if resource else "" runtime_proxy = resource_proxy or _clean(runtime.get("proxy_url")) runtime_proxy_source = "runtime_resource" if resource_proxy else "runtime" selected_proxy = "" source = "direct" if account_proxy: selected_proxy = account_proxy source = "account" elif runtime_proxy: selected_proxy = runtime_proxy source = runtime_proxy_source elif explicit_proxy: selected_proxy = explicit_proxy source = "explicit" elif legacy_proxy: selected_proxy = legacy_proxy source = "global" return ProxyRuntimeProfile( proxy_url=normalize_proxy_url(selected_proxy), proxy_source=source, resource=bool(resource), runtime_enabled=runtime_enabled, egress_mode=egress_mode, skip_ssl_verify=bool(runtime.get("skip_ssl_verify")) if runtime_enabled else False, reset_session_status_codes=_status_codes_tuple(runtime.get("reset_session_status_codes")), clearance=clearance, ) def build_session_kwargs( self, account: dict | None = None, proxy: str = "", resource: bool = False, upstream: bool = False, **session_kwargs, ) -> dict[str, object]: profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream) if profile.proxy_url: session_kwargs["proxy"] = profile.proxy_url if profile.runtime_enabled and profile.skip_ssl_verify: session_kwargs["verify"] = False return session_kwargs def build_headers( self, headers: Mapping[str, object] | None = None, target_url: str = "https://chatgpt.com", account: dict | None = None, proxy: str = "", resource: bool = False, upstream: bool = True, ) -> dict[str, object]: merged_headers: dict[str, object] = dict(headers or {}) profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream) if not profile.clearance_enabled: return merged_headers target_host = _host_from_url(target_url) bundle = self._bundle_for_headers(profile, target_host) if bundle is None or not bundle.is_valid_for(target_host, profile.proxy_url): return merged_headers if bundle.user_agent and _find_header_key(merged_headers, "user-agent") is None: merged_headers["User-Agent"] = bundle.user_agent if bundle.cookies: cookie_key = _find_header_key(merged_headers, "cookie") or "Cookie" existing_cookie = str(merged_headers.get(cookie_key) or "") cookie_header = _merge_cookie_header(existing_cookie, bundle.cookies) if cookie_header: merged_headers[cookie_key] = cookie_header return merged_headers def refresh_clearance( self, target_url: str = "https://chatgpt.com", account: dict | None = None, proxy: str = "", resource: bool = False, force: bool = False, upstream: bool = True, ) -> ClearanceBundle | None: profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream) if not profile.clearance_enabled: return None target_host = _host_from_url(target_url) key = self._cache_key(profile.proxy_url, target_host) if profile.clearance_mode == "manual": bundle = self._build_manual_bundle(profile, target_host) if bundle is not None: self._set_cached_bundle(key, bundle) return bundle if profile.clearance_mode != "flaresolverr": return None cached_before = self._get_cached_bundle(key) if cached_before is not None and not force and cached_before.is_valid_for(target_host, profile.proxy_url): return cached_before lock = self._get_flight_lock(key) if not lock.acquire(blocking=False): with lock: pass return self._get_cached_bundle(key) or cached_before try: cached_now = self._get_cached_bundle(key) if cached_now is not None and not force and cached_now.is_valid_for(target_host, profile.proxy_url): return cached_now flaresolverr_url = str(profile.clearance.get("flaresolverr_url") or "").strip() provider = self._get_provider(flaresolverr_url) new_bundle = provider.get_clearance(target_url, proxy_url=profile.proxy_url, timeout_sec=profile.timeout_sec) if new_bundle is not None: expires_at = time.time() + profile.refresh_interval if profile.refresh_interval else None if ( not new_bundle.target_host or normalize_proxy_url(new_bundle.proxy_url) != normalize_proxy_url(profile.proxy_url) or new_bundle.expires_at != expires_at ): new_bundle = replace( new_bundle, target_host=new_bundle.target_host or target_host, proxy_url=profile.proxy_url, expires_at=expires_at, ) self._set_cached_bundle(key, new_bundle) return new_bundle return cached_now or cached_before finally: lock.release() def invalidate_clearance( self, target_url: str = "https://chatgpt.com", account: dict | None = None, proxy: str = "", resource: bool = False, upstream: bool = True, ) -> None: profile = self.get_profile(account=account, proxy=proxy, resource=resource, upstream=upstream) target_host = _host_from_url(target_url) key = self._cache_key(profile.proxy_url, target_host) with self._lock: self._clearance_cache.pop(key, None) def get_runtime_status(self) -> dict[str, object]: profile = self.get_profile(upstream=True) with self._lock: cached_hosts = [host for _proxy, host in self._clearance_cache] cached_count = len(self._clearance_cache) return { "enabled": profile.runtime_enabled, "egress_mode": profile.egress_mode, "proxy_source": profile.proxy_source, "has_proxy": bool(profile.proxy_url), "clearance_enabled": profile.clearance_enabled, "clearance_mode": profile.clearance_mode, "has_clearance_bundle": cached_count > 0, "cached_clearance_hosts": sorted(set(cached_hosts)), } def _get_runtime_settings(self) -> dict[str, object]: try: runtime = self._config.get_proxy_runtime_settings() except AttributeError: runtime = {} return runtime if isinstance(runtime, dict) else {} def _bundle_for_headers(self, profile: ProxyRuntimeProfile, target_host: str) -> ClearanceBundle | None: key = self._cache_key(profile.proxy_url, target_host) if profile.clearance_mode == "manual": bundle = self._build_manual_bundle(profile, target_host) if bundle is not None: self._set_cached_bundle(key, bundle) return bundle if profile.clearance_mode == "flaresolverr": return self._get_cached_bundle(key) return None def _build_manual_bundle(self, profile: ProxyRuntimeProfile, target_host: str) -> ClearanceBundle | None: cookies = _parse_cookie_header(str(profile.clearance.get("cf_cookies") or "")) cf_clearance = str(profile.clearance.get("cf_clearance") or "").strip() if cf_clearance and "cf_clearance" not in cookies: cookies["cf_clearance"] = cf_clearance user_agent = str(profile.clearance.get("user_agent") or "").strip() if not cookies and not user_agent: return None now = time.time() expires_at = now + profile.refresh_interval if profile.refresh_interval else None return ClearanceBundle( target_host=target_host, proxy_url=profile.proxy_url, cookies=cookies, user_agent=user_agent, created_at=now, expires_at=expires_at, ) def _get_provider(self, flaresolverr_url: str) -> FlareSolverrClearanceProvider: url = str(flaresolverr_url or "").strip().rstrip("/") with self._lock: provider = self._provider_cache.get(url) if provider is None: provider = self._clearance_provider_factory(url) self._provider_cache[url] = provider return provider def _get_flight_lock(self, key: tuple[str, str]) -> threading.Lock: with self._lock: lock = self._flight_locks.get(key) if lock is None: lock = threading.Lock() self._flight_locks[key] = lock return lock def _get_cached_bundle(self, key: tuple[str, str]) -> ClearanceBundle | None: with self._lock: return self._clearance_cache.get(key) def _set_cached_bundle(self, key: tuple[str, str], bundle: ClearanceBundle) -> None: with self._lock: self._clearance_cache[key] = bundle @staticmethod def _cache_key(proxy_url: str, target_host: str) -> tuple[str, str]: return (normalize_proxy_url(proxy_url), _normalize_host(target_host)) def _clean(value: object) -> str: return str(value or "").strip() def _colon_proxy_to_url(url: str) -> str: parts = url.split(":", 3) if len(parts) == 4 and parts[1].isdigit(): host, port, username, password = parts return f"http://{quote(username, safe='')}:{quote(password, safe='')}@{host}:{port}" if len(parts) == 2 and parts[1].isdigit(): return f"http://{url}" return url def _normalize_host(host: str) -> str: return str(host or "").strip().strip(".").lower() def _host_from_url(url: str) -> str: candidate = str(url or "").strip() parsed = urlparse(candidate) if not parsed.hostname and candidate and "://" not in candidate: parsed = urlparse(f"https://{candidate}") return _normalize_host(parsed.hostname or "") def _status_codes_tuple(value: object) -> tuple[int, ...]: source = value if isinstance(value, list) else [403] codes: list[int] = [] for item in source: if isinstance(item, bool): continue try: code = int(item) except (OverflowError, TypeError, ValueError): continue if 100 <= code <= 599 and code not in codes: codes.append(code) return tuple(codes or [403]) def _coerce_timeout(value: object) -> float: try: timeout = float(value) except (OverflowError, TypeError, ValueError): timeout = 60.0 return max(1.0, timeout) def _is_valid_proxy_url(url: str) -> bool: parsed = urlparse(normalize_proxy_url(url)) return parsed.scheme in {"http", "https", "socks5", "socks5h"} and bool(parsed.netloc) def _domain_matches(host: str, domain: str) -> bool: normalized_host = _normalize_host(host) normalized_domain = _normalize_host(domain.lstrip(".")) if not normalized_domain: return True return normalized_host == normalized_domain or normalized_host.endswith(f".{normalized_domain}") def _filter_flaresolverr_cookies(raw_cookies: object, target_host: str) -> dict[str, str]: if not isinstance(raw_cookies, list): return {} filtered_cookies: dict[str, str] = {} for item in raw_cookies: if not isinstance(item, dict): continue name = str(item.get("name") or "").strip() if not name: continue value = str(item.get("value") or "") domain = str(item.get("domain") or "").strip() if not domain or _domain_matches(target_host, domain): filtered_cookies[name] = value return filtered_cookies def _parse_cookie_header(header: str) -> dict[str, str]: cookies: dict[str, str] = {} for part in str(header or "").split(";"): name, sep, value = part.strip().partition("=") if sep and name: cookies[name.strip()] = value.strip() return cookies def _cookies_to_header(cookies: Mapping[str, str]) -> str: return "; ".join(f"{name}={value}" for name, value in cookies.items() if name) def _merge_cookie_header(existing_header: str, cookies: Mapping[str, str]) -> str: existing = str(existing_header or "").strip() existing_names = set(_parse_cookie_header(existing).keys()) additions = [f"{name}={value}" for name, value in cookies.items() if name and name not in existing_names] if existing and additions: return existing.rstrip("; ") + "; " + "; ".join(additions) if existing: return existing return "; ".join(additions) def _find_header_key(headers: Mapping[str, object], name: str) -> str | None: target = name.lower() for key in headers: if str(key).lower() == target: return str(key) return None def _redact_url_credentials(text: str) -> str: return re.sub( r"((?:https?|socks5h?|socks)://)([^\s/@:]+):([^\s/@]+)@", r"\1[REDACTED]@", str(text or ""), flags=re.IGNORECASE, ) def test_proxy(url: str = "", *, timeout: float = 15.0) -> dict: candidate = normalize_proxy_url(_clean(url)) proxy_source = "input" if not candidate: profile = proxy_settings.get_profile(upstream=True) candidate = profile.proxy_url proxy_source = profile.proxy_source result_base = {"proxy_source": proxy_source, "has_proxy": bool(candidate)} if not candidate: return { "ok": False, "status": 0, "latency_ms": 0, "error": "no active proxy configured", **result_base, } if not _is_valid_proxy_url(candidate): return { "ok": False, "status": 0, "latency_ms": 0, "error": "invalid proxy url", **result_base, } session = Session(impersonate="edge101", verify=True, proxy=candidate) started = time.perf_counter() try: response = session.get( "https://chatgpt.com/api/auth/csrf", headers={"user-agent": "Mozilla/5.0 (chatgpt2api proxy test)"}, timeout=timeout, ) latency_ms = int((time.perf_counter() - started) * 1000) return { "ok": response.status_code < 500, "status": int(response.status_code), "latency_ms": latency_ms, "error": None if response.status_code < 500 else f"HTTP {response.status_code}", **result_base, } except Exception as exc: latency_ms = int((time.perf_counter() - started) * 1000) return { "ok": False, "status": 0, "latency_ms": latency_ms, "error": _redact_url_credentials(str(exc) or exc.__class__.__name__), **result_base, } finally: session.close() def test_clearance(target_url: str = "https://chatgpt.com") -> dict: target_url = str(target_url or "https://chatgpt.com").strip() or "https://chatgpt.com" started = time.perf_counter() status = proxy_settings.get_runtime_status() if not status.get("clearance_enabled"): return { "ok": False, "status": "disabled", "latency_ms": 0, "has_cookies": False, "user_agent": "", "error": "clearance is disabled", "runtime": status, } try: bundle = proxy_settings.refresh_clearance(target_url=target_url, force=True, upstream=True) except Exception as exc: latency_ms = int((time.perf_counter() - started) * 1000) return { "ok": False, "status": "error", "latency_ms": latency_ms, "has_cookies": False, "user_agent": "", "error": _redact_url_credentials(str(exc) or exc.__class__.__name__), "runtime": proxy_settings.get_runtime_status(), } latency_ms = int((time.perf_counter() - started) * 1000) runtime = proxy_settings.get_runtime_status() if bundle is None: return { "ok": False, "status": "failed", "latency_ms": latency_ms, "has_cookies": False, "user_agent": "", "error": "clearance refresh returned no bundle", "runtime": runtime, } return { "ok": True, "status": "ok", "latency_ms": latency_ms, "has_cookies": bool(bundle.cookies), "user_agent": bundle.user_agent or "", "error": None, "runtime": runtime, } proxy_settings = ProxySettingsStore()