# retoor <retoor@molodetz.nl>
from __future__ import annotations
import asyncio
import logging
import re
import ssl
from dataclasses import dataclass, field
from os import environ
from pathlib import Path
from typing import Iterable, Mapping, Sequence
from urllib.parse import unquote, urlsplit
import httpx
logger = logging.getLogger("stealth")
OUTBOUND_PROXY_URL: str | None = environ.get("DEVPLACE_OUTBOUND_PROXY_URL", "").strip() or None
def _supported_accept_encoding() -> str:
encodings = ["gzip", "deflate"]
try:
import brotli # noqa: F401
encodings.append("br")
except ImportError:
logger.debug("brotli not installed; not advertising 'br'")
try:
import zstandard # noqa: F401
encodings.append("zstd")
except ImportError:
logger.debug("zstandard not installed; not advertising 'zstd'")
return ", ".join(encodings)
ACCEPT_ENCODING: str = _supported_accept_encoding()
CHROME_CIPHER_SUITES: str = ":".join(
[
"TLS_AES_128_GCM_SHA256",
"TLS_AES_256_GCM_SHA384",
"TLS_CHACHA20_POLY1305_SHA256",
"ECDHE-ECDSA-AES128-GCM-SHA256",
"ECDHE-RSA-AES128-GCM-SHA256",
"ECDHE-ECDSA-AES256-GCM-SHA384",
"ECDHE-RSA-AES256-GCM-SHA384",
"ECDHE-ECDSA-CHACHA20-POLY1305",
"ECDHE-RSA-CHACHA20-POLY1305",
"ECDHE-RSA-AES128-SHA",
"ECDHE-RSA-AES256-SHA",
"AES128-GCM-SHA256",
"AES256-GCM-SHA384",
"AES128-SHA",
"AES256-SHA",
]
)
ALPN_PROTOCOLS: tuple[str, ...] = ("h2", "http/1.1")
DEFAULT_CHROME_VERSION: str = "146"
DEFAULT_CHUNK_SIZE: int = 65536
IN_MEMORY_DOWNLOAD_CAP: int = 100 * 1024 * 1024
IMAGE_EXTENSIONS: tuple[str, ...] = (
".jpg",
".jpeg",
".png",
".gif",
".webp",
".avif",
".bmp",
".svg",
".ico",
".tiff",
)
MAX_FILENAME_LENGTH: int = 200
MAX_CONCURRENT_DOWNLOADS: int = 64
@dataclass(frozen=True)
class ChromeProfile:
version: str = DEFAULT_CHROME_VERSION
platform: str = "Windows"
platform_token: str = "Windows NT 10.0; Win64; x64"
accept_language: str = "en-US,en;q=0.9"
@property
def user_agent(self) -> str:
return (
f"Mozilla/5.0 ({self.platform_token}) AppleWebKit/537.36 "
f"(KHTML, like Gecko) Chrome/{self.version}.0.0.0 Safari/537.36"
)
@property
def sec_ch_ua(self) -> str:
return (
f'"Google Chrome";v="{self.version}", '
f'"Chromium";v="{self.version}", '
f'"Not)A;Brand";v="24"'
)
@classmethod
def windows(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
return cls(version=version, platform="Windows", platform_token="Windows NT 10.0; Win64; x64")
@classmethod
def macos(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
return cls(
version=version,
platform="macOS",
platform_token="Macintosh; Intel Mac OS X 10_15_7",
)
@classmethod
def linux(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
return cls(version=version, platform="Linux", platform_token="X11; Linux x86_64")
@dataclass
class DownloadResult:
url: str
path: Path | None
status_code: int | None
bytes_written: int
ok: bool
error: str | None = None
@dataclass
class BytesResult:
url: str
final_url: str
status_code: int
content_type: str
data: bytes
truncated: bool
error: str | None = None
def as_dict(self) -> dict[str, object]:
return {
"url": self.url,
"final_url": self.final_url,
"status_code": self.status_code,
"content_type": self.content_type,
"data": self.data,
"truncated": self.truncated,
"error": self.error,
}
def resource_kind_for_url(url: str) -> str:
path = urlsplit(url).path.lower()
return "image" if path.endswith(IMAGE_EXTENSIONS) else "empty"
CONSENT_GATE_MARKERS: tuple[str, ...] = ("privacy gate", "privacy-gate", "cmpproperties", "consent.js")
CONSENT_GATE_CALLBACK_RE = re.compile(r"decodeURIComponent\(\s*['\"]([^'\"]+)['\"]\s*\)")
def detect_consent_gate(html: str) -> str | None:
lowered = html.lower()
if not any(marker in lowered for marker in CONSENT_GATE_MARKERS):
return None
match = CONSENT_GATE_CALLBACK_RE.search(html)
if not match:
return None
candidate = unquote(match.group(1))
if not candidate.startswith(("http://", "https://")):
return None
return candidate
def build_chrome_ssl_context() -> ssl.SSLContext:
context = ssl.create_default_context()
context.check_hostname = True
context.verify_mode = ssl.CERT_REQUIRED
context.minimum_version = ssl.TLSVersion.TLSv1_2
context.maximum_version = ssl.TLSVersion.TLSv1_3
context.set_ciphers(CHROME_CIPHER_SUITES)
context.set_alpn_protocols(list(ALPN_PROTOCOLS))
context.options |= ssl.OP_NO_COMPRESSION
logger.debug("Built Chrome-aligned SSL context with %d ciphers", CHROME_CIPHER_SUITES.count(":") + 1)
return context
_SHARED_SSL_CONTEXT: ssl.SSLContext | None = None
DEFAULT_PROFILE: ChromeProfile = ChromeProfile.windows()
try:
from devplacepy.curl_transport import CurlTransport, IMPERSONATE_TARGET
CURL_AVAILABLE: bool = True
logger.info("Stealth transport: curl_cffi impersonation active (%s)", IMPERSONATE_TARGET)
except ImportError as error:
CurlTransport = None
IMPERSONATE_TARGET = ""
CURL_AVAILABLE = False
logger.warning("curl_cffi unavailable (%s); falling back to httpx stealth transport", error)
def curl_active() -> bool:
return CURL_AVAILABLE
def chrome_ssl_context() -> ssl.SSLContext:
global _SHARED_SSL_CONTEXT
if _SHARED_SSL_CONTEXT is None:
_SHARED_SSL_CONTEXT = build_chrome_ssl_context()
return _SHARED_SSL_CONTEXT
def chrome_base_headers(profile: ChromeProfile | None = None) -> dict[str, str]:
active = profile or DEFAULT_PROFILE
return {
"sec-ch-ua": active.sec_ch_ua,
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": f'"{active.platform}"',
"user-agent": active.user_agent,
"accept-encoding": ACCEPT_ENCODING,
"accept-language": active.accept_language,
}
def configured_proxy_url() -> str | None:
try:
from devplacepy.database import get_setting
configured = get_setting("outbound_proxy_url", "").strip()
except ImportError:
configured = ""
return configured or OUTBOUND_PROXY_URL
def stealth_transport(
profile: ChromeProfile | None = None,
*,
proxy: str | None = None,
**kwargs: object,
) -> httpx.AsyncBaseTransport:
effective_proxy = proxy or configured_proxy_url()
if CURL_AVAILABLE:
return CurlTransport(proxy=effective_proxy)
return httpx.AsyncHTTPTransport(verify=chrome_ssl_context(), http2=True, proxy=effective_proxy, **kwargs)
def stealth_async_client(
*,
profile: ChromeProfile | None = None,
headers: Mapping[str, str] | None = None,
transport: httpx.AsyncBaseTransport | None = None,
proxy: str | None = None,
**kwargs: object,
) -> httpx.AsyncClient:
merged_headers: dict[str, str] = {} if CURL_AVAILABLE else chrome_base_headers(profile)
if headers:
merged_headers.update({key.lower(): value for key, value in headers.items()})
if transport is None:
transport = stealth_transport(profile, proxy=proxy)
return httpx.AsyncClient(
transport=transport,
trust_env=True,
headers=merged_headers,
**kwargs,
)
def stealth_sync_client(
*,
profile: ChromeProfile | None = None,
headers: Mapping[str, str] | None = None,
proxy: str | None = None,
**kwargs: object,
) -> httpx.Client:
merged_headers = chrome_base_headers(profile)
if headers:
merged_headers.update({key.lower(): value for key, value in headers.items()})
return httpx.Client(
http2=True,
verify=chrome_ssl_context(),
trust_env=True,
headers=merged_headers,
proxy=proxy or configured_proxy_url(),
**kwargs,
)
def slugify_filename(name: str) -> str:
name = unquote(name).strip()
name = name.replace("\\", "/").split("/")[-1]
name = name.split("?")[0].split("#")[0]
name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-._")
if not name:
name = "download"
if len(name) > MAX_FILENAME_LENGTH:
stem, _, suffix = name.rpartition(".")
if stem and len(suffix) <= 16:
keep = MAX_FILENAME_LENGTH - len(suffix) - 1
name = f"{stem[:keep]}.{suffix}"
else:
name = name[:MAX_FILENAME_LENGTH]
return name
def resolve_destination(directory: Path, url: str, override_name: str | None) -> Path:
directory = directory.resolve()
raw_name = override_name if override_name else Path(urlsplit(url).path).name
safe_name = slugify_filename(raw_name)
candidate = (directory / safe_name).resolve()
if directory != candidate.parent:
raise ValueError(f"Refusing path traversal for {url!r} -> {candidate}")
return candidate
class ChromeStealthClient:
def __init__(
self,
profile: ChromeProfile | None = None,
*,
http2: bool = True,
timeout: float = 30.0,
follow_redirects: bool = True,
proxy: str | None = None,
trust_env: bool = True,
max_connections: int = 100,
max_keepalive_connections: int = 20,
) -> None:
self.profile = profile or ChromeProfile.windows()
self._ssl_context = build_chrome_ssl_context()
limits = httpx.Limits(
max_connections=max_connections,
max_keepalive_connections=max_keepalive_connections,
)
self._client = httpx.AsyncClient(
http2=http2,
verify=self._ssl_context,
timeout=httpx.Timeout(timeout),
follow_redirects=follow_redirects,
limits=limits,
proxy=proxy,
trust_env=trust_env,
headers=self._base_headers(),
)
logger.info(
"ChromeStealthClient ready: Chrome %s on %s, http2=%s, proxy=%s",
self.profile.version,
self.profile.platform,
http2,
bool(proxy),
)
def _base_headers(self) -> dict[str, str]:
return {
"sec-ch-ua": self.profile.sec_ch_ua,
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": f'"{self.profile.platform}"',
"user-agent": self.profile.user_agent,
"accept-encoding": ACCEPT_ENCODING,
"accept-language": self.profile.accept_language,
}
def _navigation_headers(self, referer: str | None) -> dict[str, str]:
headers = {
"upgrade-insecure-requests": "1",
"accept": (
"text/html,application/xhtml+xml,application/xml;q=0.9,"
"image/avif,image/webp,image/apng,*/*;q=0.8,"
"application/signed-exchange;v=b3;q=0.7"
),
"sec-fetch-site": "same-origin" if referer else "none",
"sec-fetch-mode": "navigate",
"sec-fetch-user": "?1",
"sec-fetch-dest": "document",
}
if referer:
headers["referer"] = referer
return headers
def _resource_headers(self, dest: str, referer: str | None) -> dict[str, str]:
accept = {
"image": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",
"empty": "*/*",
}.get(dest, "*/*")
headers = {
"accept": accept,
"sec-fetch-site": "same-origin" if referer else "cross-site",
"sec-fetch-mode": "no-cors" if dest == "image" else "cors",
"sec-fetch-dest": dest,
}
if referer:
headers["referer"] = referer
return headers
def _api_headers(self, referer: str | None) -> dict[str, str]:
headers = {
"accept": "application/json, text/plain, */*",
"sec-fetch-site": "same-origin" if referer else "cross-site",
"sec-fetch-mode": "cors",
"sec-fetch-dest": "empty",
}
if referer:
headers["referer"] = referer
return headers
@staticmethod
def _json_or_error(response: httpx.Response) -> dict[str, object]:
if response.status_code >= 400:
return {
"error": f"HTTP {response.status_code}",
"status_code": response.status_code,
"body": response.text[:5000],
}
try:
return response.json()
except ValueError:
return {
"error": "Invalid JSON response",
"status_code": response.status_code,
"body": response.text[:5000],
}
async def __aenter__(self) -> "ChromeStealthClient":
return self
async def __aexit__(self, *exc: object) -> None:
await self.aclose()
async def aclose(self) -> None:
await self._client.aclose()
logger.debug("ChromeStealthClient closed")
async def _send(self, method: str, url: str, headers: dict[str, str], **kwargs: object) -> httpx.Response:
try:
return await self._client.request(method, url, headers=headers, **kwargs)
except httpx.DecodingError:
logger.warning("content decode failed for %s; retrying without br/zstd", url)
retry_headers = dict(headers)
retry_headers["accept-encoding"] = "gzip, deflate"
return await self._client.request(method, url, headers=retry_headers, **kwargs)
async def request(
self,
method: str,
url: str,
*,
referer: str | None = None,
headers: Mapping[str, str] | None = None,
**kwargs: object,
) -> httpx.Response:
merged = self._navigation_headers(referer)
if headers:
merged.update({key.lower(): value for key, value in headers.items()})
logger.info("%s %s", method.upper(), url)
response = await self._send(method, url, merged, **kwargs)
logger.debug("%s %s -> %s (%s)", method.upper(), url, response.status_code, response.http_version)
return response
async def get(self, url: str, *, referer: str | None = None, **kwargs: object) -> httpx.Response:
return await self.request("GET", url, referer=referer, **kwargs)
async def post(self, url: str, *, referer: str | None = None, **kwargs: object) -> httpx.Response:
return await self.request("POST", url, referer=referer, **kwargs)
def stream(
self,
method: str,
url: str,
*,
referer: str | None = None,
dest: str = "empty",
headers: Mapping[str, str] | None = None,
**kwargs: object,
):
merged = self._resource_headers(dest, referer)
if headers:
merged.update({key.lower(): value for key, value in headers.items()})
return self._client.stream(method, url, headers=merged, **kwargs)
async def download_file(
self,
url: str,
destination: str | Path,
*,
filename: str | None = None,
referer: str | None = None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
dest_type: str = "empty",
) -> DownloadResult:
directory = Path(destination)
directory.mkdir(parents=True, exist_ok=True)
try:
target = resolve_destination(directory, url, filename)
except ValueError as error:
logger.error("Destination rejected for %s: %s", url, error)
return DownloadResult(url=url, path=None, status_code=None, bytes_written=0, ok=False, error=str(error))
written = 0
try:
async with self.stream("GET", url, referer=referer, dest=dest_type) as response:
response.raise_for_status()
with target.open("wb") as handle:
async for chunk in response.aiter_bytes(chunk_size):
handle.write(chunk)
written += len(chunk)
logger.info("Downloaded %s -> %s (%d bytes)", url, target, written)
return DownloadResult(
url=url,
path=target,
status_code=response.status_code,
bytes_written=written,
ok=True,
)
except httpx.HTTPStatusError as error:
logger.error("HTTP %s downloading %s", error.response.status_code, url)
return DownloadResult(
url=url,
path=None,
status_code=error.response.status_code,
bytes_written=written,
ok=False,
error=f"HTTP {error.response.status_code}",
)
except httpx.HTTPError as error:
logger.error("Transport error downloading %s: %s", url, error)
return DownloadResult(url=url, path=None, status_code=None, bytes_written=written, ok=False, error=str(error))
async def download_files(
self,
urls: Iterable[str] | Mapping[str, str],
destination: str | Path,
*,
concurrency: int = 8,
referer: str | None = None,
chunk_size: int = DEFAULT_CHUNK_SIZE,
dest_type: str = "empty",
) -> list[DownloadResult]:
if isinstance(urls, Mapping):
jobs: Sequence[tuple[str, str | None]] = [(url, name) for url, name in urls.items()]
else:
jobs = [(url, None) for url in urls]
bounded = max(1, min(concurrency, MAX_CONCURRENT_DOWNLOADS))
semaphore = asyncio.Semaphore(bounded)
logger.info("Starting bulk download of %d files with concurrency %d", len(jobs), bounded)
async def worker(url: str, name: str | None) -> DownloadResult:
async with semaphore:
return await self.download_file(
url,
destination,
filename=name,
referer=referer,
chunk_size=chunk_size,
dest_type=dest_type,
)
results = await asyncio.gather(*(worker(url, name) for url, name in jobs))
succeeded = sum(1 for result in results if result.ok)
logger.info("Bulk download finished: %d succeeded, %d failed", succeeded, len(results) - succeeded)
return list(results)
async def download_bytes(
self,
url: str,
*,
referer: str | None = None,
max_bytes: int = IN_MEMORY_DOWNLOAD_CAP,
chunk_size: int = DEFAULT_CHUNK_SIZE,
dest_type: str | None = None,
) -> BytesResult:
kind = dest_type if dest_type else resource_kind_for_url(url)
buffer = bytearray()
try:
async with self.stream("GET", url, referer=referer, dest=kind) as response:
content_type = response.headers.get("content-type", "")
async for chunk in response.aiter_bytes(chunk_size):
buffer.extend(chunk)
if len(buffer) > max_bytes:
break
final_url = str(response.url)
truncated = len(buffer) > max_bytes
data = bytes(buffer[:max_bytes])
logger.debug("download_bytes %s -> %d bytes (status %s)", url, len(data), response.status_code)
return BytesResult(
url=url,
final_url=final_url,
status_code=response.status_code,
content_type=content_type,
data=data,
truncated=truncated,
)
except httpx.HTTPError as error:
logger.error("download_bytes failed for %s: %s", url, error)
return BytesResult(
url=url,
final_url=url,
status_code=0,
content_type="",
data=b"",
truncated=False,
error=str(error),
)
async def post_json(
self,
url: str,
payload: object,
*,
headers: Mapping[str, str] | None = None,
referer: str | None = None,
timeout: float | None = None,
) -> dict[str, object]:
request_headers = self._api_headers(referer)
if headers:
request_headers.update({key.lower(): value for key, value in headers.items()})
kwargs: dict[str, object] = {"json": payload}
if timeout is not None:
kwargs["timeout"] = httpx.Timeout(timeout)
logger.info("POST(json) %s", url)
try:
response = await self._send("POST", url, request_headers, **kwargs)
except httpx.HTTPError as error:
logger.error("post_json transport error for %s: %s", url, error)
return {"error": f"{type(error).__name__}: {error}"}
return self._json_or_error(response)
async def get_json(
self,
url: str,
*,
referer: str | None = None,
timeout: float | None = None,
) -> dict[str, object]:
kwargs: dict[str, object] = {}
if timeout is not None:
kwargs["timeout"] = httpx.Timeout(timeout)
logger.info("GET(json) %s", url)
try:
response = await self._send("GET", url, self._api_headers(referer), **kwargs)
except httpx.HTTPError as error:
logger.error("get_json transport error for %s: %s", url, error)
return {"error": f"{type(error).__name__}: {error}"}
return self._json_or_error(response)
__all__ = [
"ChromeStealthClient",
"ChromeProfile",
"DownloadResult",
"BytesResult",
"build_chrome_ssl_context",
"chrome_ssl_context",
"chrome_base_headers",
"stealth_transport",
"stealth_async_client",
"stealth_sync_client",
"curl_active",
"resource_kind_for_url",
"slugify_filename",
"detect_consent_gate",
"configured_proxy_url",
]
Reference in New Issue View Git Blame Copy Permalink