69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import os
|
||
|
|
from typing import Any
|
||
|
|
from urllib.parse import urlsplit
|
||
|
|
|
||
|
|
|
||
|
|
def _is_truthy(value: str | None) -> bool:
|
||
|
|
return (value or "").strip().lower() in {"1", "true", "yes", "on"}
|
||
|
|
|
||
|
|
|
||
|
|
def _normalize_proxy_scheme(scheme: str) -> str:
|
||
|
|
normalized = (scheme or "").strip().lower()
|
||
|
|
if normalized == "socks4":
|
||
|
|
return "socks4a"
|
||
|
|
if normalized == "socks5":
|
||
|
|
return "socks5h"
|
||
|
|
return normalized
|
||
|
|
|
||
|
|
|
||
|
|
def build_proxy_url_from_proxychains_env(*, require_enabled: bool = True) -> str | None:
|
||
|
|
if require_enabled and not _is_truthy(os.environ.get("ENABLE_PROXYCHAINS")):
|
||
|
|
return None
|
||
|
|
|
||
|
|
scheme = _normalize_proxy_scheme(os.environ.get("PROXYCHAINS_PROXY_TYPE", "socks4"))
|
||
|
|
host = (os.environ.get("PROXYCHAINS_PROXY_HOST") or "").strip()
|
||
|
|
port = (os.environ.get("PROXYCHAINS_PROXY_PORT") or "").strip()
|
||
|
|
if not host or not port:
|
||
|
|
return None
|
||
|
|
return f"{scheme}://{host}:{port}"
|
||
|
|
|
||
|
|
|
||
|
|
def _same_proxy_endpoint(left: str, right: str) -> bool:
|
||
|
|
left_parts = urlsplit(left)
|
||
|
|
right_parts = urlsplit(right)
|
||
|
|
return (
|
||
|
|
left_parts.hostname == right_parts.hostname
|
||
|
|
and left_parts.port == right_parts.port
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def resolve_requests_proxy_url(proxy_url: str | None) -> str:
|
||
|
|
value = (proxy_url or "").strip()
|
||
|
|
if not value:
|
||
|
|
return ""
|
||
|
|
|
||
|
|
proxychains_url = build_proxy_url_from_proxychains_env()
|
||
|
|
if proxychains_url and _same_proxy_endpoint(value, proxychains_url):
|
||
|
|
# Avoid double-proxying requests when the whole process is already wrapped by proxychains.
|
||
|
|
return ""
|
||
|
|
|
||
|
|
return value
|
||
|
|
|
||
|
|
|
||
|
|
def build_requests_proxies(proxy_url: str | None) -> dict[str, str] | None:
|
||
|
|
value = resolve_requests_proxy_url(proxy_url)
|
||
|
|
if not value:
|
||
|
|
return None
|
||
|
|
return {"http": value, "https": value}
|
||
|
|
|
||
|
|
|
||
|
|
def apply_requests_proxy(session: Any, proxy_url: str | None) -> Any:
|
||
|
|
proxies = build_requests_proxies(proxy_url)
|
||
|
|
if not proxies:
|
||
|
|
return session
|
||
|
|
session.proxies.update(proxies)
|
||
|
|
session.trust_env = False
|
||
|
|
return session
|