"""Resolve outbound proxy for CPA mint HTTP + browser. Priority (highest first): 1. explicit argument 2. thread-local runtime pin (set_runtime_proxy) 3. environment https_proxy / HTTPS_PROXY / http_proxy / HTTP_PROXY Thread-local pin avoids cross-talk when multiple mint workers run with different proxies in the same process. """ from __future__ import annotations import os import threading from urllib.parse import urlparse _thread = threading.local() def set_runtime_proxy(proxy: str | None) -> None: """Pin proxy for the *current thread*. Empty clears pin.""" p = (proxy or "").strip() _thread.proxy = p or None def get_runtime_proxy() -> str | None: return getattr(_thread, "proxy", None) def resolve_proxy(explicit: str | None = None) -> str: for cand in ( (explicit or "").strip(), (get_runtime_proxy() or "").strip(), (os.environ.get("https_proxy") or "").strip(), (os.environ.get("HTTPS_PROXY") or "").strip(), (os.environ.get("http_proxy") or "").strip(), (os.environ.get("HTTP_PROXY") or "").strip(), ): if cand: return cand return "" def proxy_for_chromium(proxy: str) -> str: """Chromium --proxy-server cannot embed user:pass; host:port only.""" p = (proxy or "").strip() if not p: return "" u = urlparse(p if "://" in p else f"http://{p}") host = u.hostname or "" if not host: return "" port = u.port or (443 if (u.scheme or "http") == "https" else 80) scheme = u.scheme or "http" return f"{scheme}://{host}:{port}" def proxy_log_label(proxy: str) -> str: """Redact userinfo for logs.""" p = (proxy or "").strip() if not p: return "" try: u = urlparse(p if "://" in p else f"http://{p}") host = u.hostname or "?" port = u.port or "" auth = "user:***@" if u.username else "" return f"{u.scheme or 'http'}://{auth}{host}{(':' + str(port)) if port else ''}" except Exception: return "(proxy)"