chaos
6f7c949681
Overwrite previous grok-keygen contents with the local registration/CPA protocol codebase (secrets and runtime data excluded).
72 lines
2.0 KiB
Python
72 lines
2.0 KiB
Python
"""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)"
|