# retoor <retoor@molodetz.nl>
"""
Stealth HTTP client built on top of httpx that reproduces the network
behaviour of a modern Google Chrome desktop browser as faithfully as a
pure-Python stack permits.
WHY THIS IS THE BEST PRACTICAL STEALTH CLIENT
---------------------------------------------
Anti-bot platforms (Cloudflare, Akamai, DataDome, PerimeterX, Imperva and
the like) classify clients on three independent layers. A request only
looks "human" when all three agree with one another. Most scraping code
fails because it spoofs a single layer (usually just the User-Agent string)
while leaving the others screaming "automated tool". This client aligns
every layer that the standard library exposes:
1. TLS layer (JA3 / cipher fingerprint)
The SSL context is configured with a Chrome-aligned TLS 1.2 cipher list,
TLS 1.2 as the floor, TLS 1.3 negotiation, and an ALPN advertisement of
``h2`` before ``http/1.1`` exactly like Chrome. This moves the JA3 hash
away from the default Python/OpenSSL fingerprint that detection vendors
blocklist on sight. Note that OpenSSL does not let Python order the TLS 1.3
ciphersuites, nor rewrite the supported-groups / signature-algorithm lists,
so this is Chrome-*aligned* rather than byte-identical (see HONEST LIMITATIONS).
2. HTTP/2 layer (frame and header behaviour)
Real Chrome speaks HTTP/2 to virtually every modern host. This client
enables HTTP/2 by default, sends lowercase header names, and preserves a
Chrome-accurate header *order* - the order itself is a fingerprint that
naive clients get wrong even when the header values are correct.
3. Application layer (headers and client hints)
A complete, correctly ordered set of Chrome headers is emitted, including
the modern User-Agent Client Hints (``sec-ch-ua``, ``sec-ch-ua-mobile``,
``sec-ch-ua-platform``) and the request-context ``Sec-Fetch-*`` family.
The ``Sec-Fetch-*`` values are recomputed per request type so a document
navigation, a sub-resource fetch and a file download each carry the
metadata Chrome would actually attach in that situation.
WHAT YOU CAN EXPECT
-------------------
- Requests that pass the overwhelming majority of header- and TLS-based
bot heuristics, including Cloudflare's "I'm Under Attack" passive checks
for endpoints that do not mandate a JavaScript challenge.
- Transparent HTTP/2, gzip/deflate/br/zstd decompression, cookie
persistence across a session, and connection reuse.
- Single and bulk asynchronous file downloads with bounded concurrency,
streaming to disk (constant memory regardless of file size), filename
slugification and path-traversal protection.
HONEST LIMITATIONS
------------------
The Python standard ``ssl`` module cannot rewrite the TLS extension order,
the supported-groups list or the signature-algorithm list, so the JA3 hash
produced here is *Chrome-like* rather than *byte-identical* to Chrome. For
targets that fingerprint those low-level extension fields (a small minority,
but a growing one) a native TLS stack such as ``curl_cffi`` or BoringSSL is
required. This client is engineered to be the strongest stealth achievable
without leaving the httpx ecosystem, and it degrades gracefully: every layer
it can control is made indistinguishable from Chrome.
USE CASES
---------
- Resilient web scraping and data collection against header/TLS gating.
- Monitoring, price tracking and availability checks on protected sites.
- Mirroring and bulk asset downloading (images, documents, archives).
- Integration testing of CDN and WAF rules from a realistic client.
EXAMPLE
-------
import asyncio
from devplacepy.stealth import ChromeStealthClient
async def main() -> None:
async with ChromeStealthClient() as client:
response = await client.get("https://example.com")
print(response.status_code, len(response.text))
results = await client.download_files(
[
"https://example.com/a.jpg",
"https://example.com/b.pdf",
],
destination="downloads",
concurrency=4,
)
for result in results:
print(result.url, result.ok, result.path)
asyncio.run(main())
"""
from __future__ import annotations
import asyncio
import logging
import re
import ssl
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable, Mapping, Sequence
from urllib.parse import unquote, urlsplit
import httpx
logger = logging.getLogger("stealth")
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 = "149"
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"
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
# Do NOT pin a single ECDH curve: Chrome advertises several groups
# (X25519, P-256, P-384, plus an MLKEM hybrid), and pinning to one both
# narrows the supported-groups fingerprint and breaks the handshake with
# servers that lack that curve. Let OpenSSL advertise its default group set.
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 stealth_transport(
profile: ChromeProfile | None = None,
**kwargs: object,
) -> httpx.AsyncBaseTransport:
if CURL_AVAILABLE:
return CurlTransport()
return httpx.AsyncHTTPTransport(verify=chrome_ssl_context(), http2=True, **kwargs)
def stealth_async_client(
*,
profile: ChromeProfile | None = None,
headers: Mapping[str, str] | None = None,
transport: httpx.AsyncBaseTransport | 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)
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,
**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,
**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:
# Some CDNs emit a br/zstd stream the decoder rejects. Recover the
# content by retrying once with only the universally-safe encodings,
# rather than failing a site that actually serves a valid body.
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",
]
Reference in New Issue View Git Blame Copy Permalink