commit a3bb7639d2cde45a70963a6acfe61b722ddc94ac Author: retoor Date: Wed Sep 9 13:49:07 2026 +0200 Initial version of stealthii An aiohttp/httpx-shaped async HTTP client backed by a single, persistent, stealth-patched Chromium instance via Playwright. get/post/put/patch/ delete/head/download use context.request (real Chrome TLS/HTTP2 fingerprint, no tab needed); render()/screenshot()/page() escalate to full JS-executing page rendering for targets that need a JS challenge solved. Supports named sessions for cookie persistence/warm-up, and `async with stealth as page` to claim a tab directly, task-safe for concurrent use on one shared instance. Context/browser recycling bounds cookie and on-disk cache growth in a long-lived process. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_0116i7dYLNZTXNR7Lqf439bV diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fc68fc7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +__pycache__/ +*.pyc +.venv/ +*.egg-info/ +dist/ +build/ +.pytest_cache/ +.ruff_cache/ diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..58f8c94 --- /dev/null +++ b/Makefile @@ -0,0 +1,32 @@ +.PHONY: install dev test test-unit lint format clean build uninstall + +install: + pip install -e ".[test]" + playwright install chromium + +dev: + pip install -e ".[test]" + playwright install chromium + +test: test-unit + +test-unit: + pytest tests/ -v -m 'not online' + +test-integration: + pytest tests/ -v -m online + +lint: + ruff check stealthii tests + +format: + ruff format stealthii tests + +clean: + rm -rf __pycache__ *.egg-info dist build .eggs stealthii/__pycache__ tests/__pycache__ .pytest_cache .ruff_cache + +build: + python -m build + +uninstall: + pip uninstall -y stealthii diff --git a/README.md b/README.md new file mode 100644 index 0000000..df31836 --- /dev/null +++ b/README.md @@ -0,0 +1,142 @@ +# stealthii + +`retoor ` + +An `aiohttp`/`httpx`-shaped async HTTP client backed by a single, persistent, +stealth-patched Chromium instance. Every request goes out through the real +browser's own network stack, so it carries a genuine Chrome TLS/HTTP2/UA +fingerprint instead of Python's OpenSSL-based one — the signal most +anti-bot systems key on first, before they ever look at a header. + +## Why + +`requests`, `httpx`, and `aiohttp` all negotiate TLS through Python's `ssl` +module (OpenSSL). Real browsers negotiate through their own TLS stack +(Chrome: BoringSSL). The resulting ClientHello — cipher suites, extensions, +curves, and their order — is a fingerprint (JA3/JA4) that anti-bot systems +read directly off the handshake, independent of whatever `User-Agent` +header the client claims. A Python client cannot fake this from the +application layer; the only way to get a real Chrome fingerprint is to +make the request from an actual Chrome process. + +Playwright's `BrowserContext.request` API does exactly that: it issues +plain HTTP requests through the same browser process backing a real page, +without needing a tab or JS execution — as fast as a normal HTTP call, but +with the browser's authentic identity. `stealthii` wraps that fast path +(`get`/`post`/`put`/`patch`/`delete`/`head`/`download`) behind a familiar +API, and adds full page rendering (`render`/`screenshot`/`page`) as a +heavier fallback for targets that require solving a JS challenge. + +On top of that, `stealthii` removes the JS-level automation tells a stock +headless Chromium exposes (`navigator.webdriver`, missing `window.chrome`, +an empty plugin list, a SwiftShader WebGL renderer, a permissions-query +mismatch) — verified clean against `bot.sannysoft.com`'s full check suite. + +## What it does not do + +TLS/HTTP2 impersonation defeats fingerprint-based detection. It does not +defeat IP-reputation-based rate limiting, nor the behavioural layer of a +system like DataDome or a Cloudflare managed challenge — those need a +clean IP and, for the hardest targets, genuine human-like interaction. +Layer `stealthii` with residential proxies and session warm-up +(`session=`, see below) for those targets; don't expect it alone to be a +universal bypass. + +## Install + +```bash +pip install -e . +playwright install chromium +``` + +## Usage + +```python +from stealthii import Stealthii + +stealth = Stealthii() # cheap, lazy — launches nothing yet +resp = await stealth.get("https://example.com/api", params={"q": "x"}) +resp.status +await resp.json() +await resp.text() + +await stealth.shutdown() # call once at host process shutdown +``` + +`get`/`post`/`put`/`patch`/`delete`/`head` accept the same shapes as +`aiohttp`/`httpx`: `params`, `headers`, `data` (dict → form-urlencoded, +bytes/str → raw body), `json`, `files` (multipart), `cookies` (per-call, +layered on the context's own jar), `auth` as an `(user, password)` Basic +tuple, `allow_redirects` / `follow_redirects` (both spellings accepted), +`max_redirects`, and `timeout`. + +```python +data = await stealth.download("https://example.com/photo.png") # raw bytes + +png_b64 = await stealth.screenshot("https://example.com") # base64 PNG +await stealth.screenshot("https://example.com", path="shot.png") # saved to disk + +rendered = await stealth.render("https://example.com", collect_links=True) +rendered["text"]; rendered["links"]; rendered["status"] +``` + +### Named sessions (cookie persistence / warm-up) + +Every call is anonymous by default (a shared, periodically recycled +context). Pass `session="name"` to use a dedicated, reused context instead +— cookies set by one call under that name are seen by the next: + +```python +await stealth.render("https://example.com/", session="acme") # picks up cookies +resp = await stealth.get("https://example.com/api", session="acme") # reuses them +``` + +### Raw page access + +For anything the high-level API doesn't cover, `page()` yields a real +Playwright `Page` (already stealth-patched), guaranteed to close even on +exception: + +```python +async with stealth.page() as pg: + await pg.goto("https://example.com") + await pg.click("#accept-cookies") +``` + +## Configuration + +All configuration is passed once, at construction — `Stealthii(...)` does +no I/O itself; the browser launches lazily on first real use: + +```python +stealth = Stealthii( + max_tabs=4, # concurrent render()/page() tabs + fingerprint="ja4", # "ja3" or "ja4" — see fingerprint_info() + request_timeout=15.0, + max_retries=3, + disk_cache_bytes=1, # keep Chromium's own on-disk cache tiny + recycle_context_after=(500, 1800.0), # (max requests, max seconds) per context + recycle_browser_after_seconds=6*3600, # full relaunch cadence, or None to disable +) +``` + +`fingerprint` selects which hash `fingerprint_info()` reports (JA3 or +JA4) — it does not change what's sent on the wire; both describe the same +genuine Chromium handshake. + +### Why the recycle settings exist + +A browser that runs for the process's entire lifetime otherwise +accumulates cookies and an on-disk HTTP cache without bound. `stealthii` +closes and recreates any context — anonymous or named — once it crosses +`recycle_context_after`, and fully relaunches the browser (discarding its +whole profile directory) every `recycle_browser_after_seconds`. Named +sessions meant to stay warm longer should get a larger +`recycle_context_after` at construction time. + +## Testing + +```bash +make test-unit # no network required +make test-integration # hits real sites; requires Chromium installed +``` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6b64188 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,69 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "stealthii" +version = "0.1.0" +description = "aiohttp/httpx-shaped HTTP client backed by a persistent, stealth-patched Chromium" +authors = [ + {name = "retoor", email = "retoor@molodetz.nl"} +] +readme = "README.md" +license = {text = "MIT"} +requires-python = ">=3.9" +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Internet :: WWW/HTTP", +] +dependencies = [ + "playwright>=1.40.0", +] + +[project.optional-dependencies] +test = [ + "pytest>=7.0.0", + "pytest-asyncio>=0.21.0", +] +dev = [ + "pytest>=7.0.0", + "pytest-asyncio>=0.21.0", + "ruff>=0.6.0", +] + +[project.urls] +Homepage = "https://github.com/retoor/stealthii" + +[tool.setuptools.packages.find] +where = ["."] +include = ["stealthii*"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] +markers = [ + "online: test requires live network access", +] + +[tool.ruff] +line-length = 120 +target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "B", "UP", "Q"] +ignore = ["E501", "E402", "B008", "B904"] + +[tool.ruff.lint.flake8-quotes] +inline-quotes = "single" +multiline-quotes = "double" +docstring-quotes = "double" + +[tool.ruff.format] +quote-style = "single" diff --git a/scripts/smoke_live.py b/scripts/smoke_live.py new file mode 100644 index 0000000..6321593 --- /dev/null +++ b/scripts/smoke_live.py @@ -0,0 +1,60 @@ +# retoor +import asyncio + +from stealthii import Stealthii + + +async def main(): + s = Stealthii() + + print('=== get (json) ===') + resp = await s.get('https://httpbin.org/get', params={'a': '1'}) + print(resp.status, (await resp.json()).get('args')) + + print('=== post (json body) ===') + resp = await s.post('https://httpbin.org/post', json={'hello': 'world'}) + data = await resp.json() + print(resp.status, data.get('json')) + + print('=== post (form data) ===') + resp = await s.post('https://httpbin.org/post', data={'x': '1', 'y': '2'}) + data = await resp.json() + print(resp.status, data.get('form')) + + print('=== cookies per-call ===') + resp = await s.get('https://httpbin.org/cookies', cookies={'flavor': 'choc'}) + print(resp.status, await resp.json()) + + print('=== auth ===') + resp = await s.get('https://httpbin.org/basic-auth/user/pass', auth=('user', 'pass')) + print(resp.status, await resp.json()) + + print('=== download (binary) ===') + png = await s.download('https://httpbin.org/image/png') + print('bytes:', len(png), png[:8]) + + print('=== render ===') + r = await s.render('https://example.com', collect_links=True) + print(r['status'], r['text'][:60], len(r['links'])) + + print('=== screenshot (base64) ===') + b64 = await s.screenshot('https://example.com') + print('b64 len:', len(b64)) + + print('=== screenshot (path) ===') + path = await s.screenshot('https://example.com', path='/tmp/stealthii_shot.png') + print('saved to', path) + + print('=== session warm-up (cookie persistence) ===') + await s.get('https://httpbin.org/cookies/set/persisted/yes', session='warm', allow_redirects=True) + resp = await s.get('https://httpbin.org/cookies', session='warm') + print(await resp.json()) + + print('=== fingerprint_info (ja4 default) ===') + print(await s.fingerprint_info()) + + await s.shutdown() + print('OK - all smoke checks completed') + + +asyncio.run(main()) diff --git a/stealthii/__init__.py b/stealthii/__init__.py new file mode 100644 index 0000000..53bb695 --- /dev/null +++ b/stealthii/__init__.py @@ -0,0 +1,19 @@ +# retoor +"""stealthii — an aiohttp/httpx-shaped HTTP client backed by a single, +persistent, stealth-patched Chromium instance, so scraping traffic carries +a genuine browser TLS/HTTP2/UA fingerprint instead of Python's own. +""" + +from .client import Stealthii, StealthRequestError, StealthResponse, StealthUnavailableError +from .stealth_js import STEALTH_INIT_JS + +__version__ = '0.1.0' + +__all__ = [ + 'Stealthii', + 'StealthResponse', + 'StealthRequestError', + 'StealthUnavailableError', + 'STEALTH_INIT_JS', + '__version__', +] diff --git a/stealthii/client.py b/stealthii/client.py new file mode 100644 index 0000000..00834a8 --- /dev/null +++ b/stealthii/client.py @@ -0,0 +1,574 @@ +# retoor +"""The Stealthii client: a single class that owns one persistent, stealth- +patched Chromium instance and exposes an aiohttp/httpx-shaped API +(get/post/download) backed by it, plus render() for targets that need real +JS execution. + +Usage:: + + stealth = Stealthii() # cheap, lazy — launches nothing yet + resp = await stealth.get(url) # launches the browser on first use + await stealth.shutdown() # call once at host process shutdown + +A ``Stealthii`` instance must be constructed at least once by the host +application before use; construction itself does no I/O. + +Every request is anonymous (a shared, periodically recycled context) unless +a ``session`` name is passed to get/post/download/render/page, in which +case its own dedicated, cookie-persisting context is used and reused — +useful for warm-up (render a homepage, then get() a protected endpoint +under the same session name and its cookies carry over). +""" + +import asyncio +import base64 +import json as json_module +import logging +import time +from contextlib import asynccontextmanager +from typing import Any, AsyncIterator, Dict, Optional, Tuple, Union + +from .stealth_js import STEALTH_INIT_JS + +logger = logging.getLogger('stealthii') + +_LOCALE_BY_LANG = {'nl': 'nl-NL', 'de': 'de-DE', 'fr': 'fr-FR', 'es': 'es-ES', 'en': 'en-US'} +_LAUNCH_ARGS = [ + '--no-sandbox', + '--disable-dev-shm-usage', + '--disable-blink-features=AutomationControlled', +] +_IGNORE_DEFAULT_ARGS = ['--enable-automation'] +_FINGERPRINT_ECHO_URL = 'https://tls.peet.ws/api/all' +_DEFAULT_SESSION = None + + +class StealthUnavailableError(Exception): + """The stealth browser could not be started: Playwright is not + installed, or the Chromium launch itself failed.""" + + +class StealthRequestError(Exception): + """A request through the stealth browser failed outright, after + exhausting retries.""" + + +class StealthResponse: + """A fully-buffered response, shaped like an aiohttp ``ClientResponse`` + closely enough to drop into existing ``async with session.get(...) as + resp:`` call sites unchanged.""" + + __slots__ = ('status', 'headers', 'url', '_body') + + def __init__(self, status: int, headers: Dict[str, str], body: bytes, url: str): + self.status = status + self.headers = headers + self.url = url + self._body = body + + async def text(self) -> str: + return self._body.decode('utf-8', errors='replace') + + async def json(self, content_type: Optional[str] = None) -> Any: + return json_module.loads(self._body.decode('utf-8', errors='replace')) + + async def read(self) -> bytes: + return self._body + + async def __aenter__(self) -> 'StealthResponse': + return self + + async def __aexit__(self, *exc_info) -> bool: + return False + + +class _ContextSlot: + __slots__ = ('context', 'created_at', 'count', 'patched') + + def __init__(self, context): + self.context = context + self.created_at = time.monotonic() + self.count = 0 + self.patched = False + + +class Stealthii: + """Owns one persistent, stealth-patched Chromium and every context + derived from it (the shared anonymous context, named persistent + sessions, rendered-page tabs). + + Parameters + ---------- + max_tabs: + Maximum concurrent full-page renders (:meth:`page`/:meth:`render`). + Cheap no-tab requests (:meth:`get`/:meth:`post`/:meth:`download`) + are not limited by this — they cost no tab. + fingerprint: + ``'ja3'`` or ``'ja4'`` (default). Selects which hash + :meth:`fingerprint_info` reports. Both describe the same real + Chromium TLS handshake — this does not change what is sent on the + wire, only which fingerprint format is surfaced for diagnostics. + request_timeout: + Default timeout in seconds for get/post/download. + max_retries: + Number of attempts for get/post/download/render before raising. + disk_cache_bytes: + Chromium's own on-disk HTTP/media cache size, in bytes. Kept tiny + by default since this browser runs indefinitely — an unbounded + cache is otherwise a slow, silent disk leak. + recycle_context_after: + ``(max_requests, max_seconds)``. Any context — anonymous or a named + session — is closed and recreated once it crosses either bound, + which is what actually keeps cookie/storage growth in check for a + long-lived process. Pass a larger tuple at construction time for + sessions meant to stay warm longer. + recycle_browser_after_seconds: + Fully relaunch the browser (fresh profile, discarding its on-disk + cache directory too) after this many seconds. ``None`` disables it. + """ + + def __init__( + self, + max_tabs: int = 4, + fingerprint: str = 'ja4', + request_timeout: float = 15.0, + max_retries: int = 3, + disk_cache_bytes: int = 1, + recycle_context_after: tuple = (500, 1800.0), + recycle_browser_after_seconds: Optional[float] = 6 * 3600.0, + ): + if fingerprint not in ('ja3', 'ja4'): + raise ValueError(f"fingerprint must be 'ja3' or 'ja4', got {fingerprint!r}") + self.max_tabs = max_tabs + self.fingerprint = fingerprint + self.request_timeout = request_timeout + self.max_retries = max_retries + self.disk_cache_bytes = disk_cache_bytes + self.recycle_context_after_count, self.recycle_context_after_s = recycle_context_after + self.recycle_browser_after_s = recycle_browser_after_seconds + + self._playwright = None + self._browser = None + self._browser_started_at = 0.0 + self._contexts: Dict[Optional[str], _ContextSlot] = {} + self._user_agent = None + self._tab_semaphore = None + self._lock: Optional[asyncio.Lock] = None + self._attempted = False + self._claimed_pages: Dict[Any, Any] = {} + + def _get_lock(self) -> asyncio.Lock: + if self._lock is None: + self._lock = asyncio.Lock() + return self._lock + + async def start(self): + """Idempotently launch the browser. Calling this explicitly at host + startup is optional — every other method calls it lazily — but + doing so at boot means the first real request isn't the one paying + for browser startup latency.""" + async with self._get_lock(): + await self._ensure_browser_locked() + return self._browser + + async def _ensure_browser_locked(self): + """Must be called with self._lock held.""" + now = time.monotonic() + if self._browser is not None: + if self.recycle_browser_after_s is not None and now - self._browser_started_at > self.recycle_browser_after_s: + logger.info('stealthii: recycling browser (age threshold reached)') + await self._close_browser_locked() + else: + return self._browser + try: + from playwright.async_api import async_playwright + except ImportError as e: + raise StealthUnavailableError('playwright is not installed') from e + try: + pw = await async_playwright().start() + browser = await pw.chromium.launch( + headless=True, + args=[ + *_LAUNCH_ARGS, + f'--disk-cache-size={self.disk_cache_bytes}', + f'--media-cache-size={self.disk_cache_bytes}', + ], + ignore_default_args=_IGNORE_DEFAULT_ARGS, + ) + except Exception as e: + self._attempted = True + raise StealthUnavailableError(f'chromium launch failed: {e}') from e + self._playwright = pw + self._browser = browser + self._browser_started_at = now + self._tab_semaphore = asyncio.Semaphore(self.max_tabs) + self._user_agent = ( + 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' + f'(KHTML, like Gecko) Chrome/{browser.version} Safari/537.36' + ) + logger.info(f'stealthii: browser started (Chromium {browser.version})') + return self._browser + + async def _close_browser_locked(self): + for slot in self._contexts.values(): + try: + await slot.context.close() + except Exception: + pass + self._contexts.clear() + try: + if self._browser is not None: + await self._browser.close() + if self._playwright is not None: + await self._playwright.stop() + except Exception: + pass + self._browser = None + self._playwright = None + + async def _get_context(self, session: Optional[str]): + """Return the BrowserContext for ``session`` (``None`` = the + shared anonymous one), creating or recycling it as needed. Every + context — named or not — is subject to the same recycle bounds, so + a long-lived process never accumulates unbounded cookies/storage + under any one name.""" + async with self._get_lock(): + await self._ensure_browser_locked() + now = time.monotonic() + slot = self._contexts.get(session) + if slot is not None: + stale = ( + slot.count >= self.recycle_context_after_count + or now - slot.created_at > self.recycle_context_after_s + ) + if stale: + logger.info(f'stealthii: recycling context {session!r} (cache/cookie bound reached)') + try: + await slot.context.close() + except Exception: + pass + slot = None + if slot is None: + context = await self._browser.new_context(user_agent=self._user_agent) + slot = _ContextSlot(context) + self._contexts[session] = slot + slot.count += 1 + return slot + + @asynccontextmanager + async def page(self, lang: str = 'en', session: Optional[str] = None) -> AsyncIterator[Any]: + """A stealth-patched tab (Playwright ``Page``), for full JS + rendering. With ``session`` set, the tab's context is a named, + reused session (cookies persist across calls under that name); + otherwise a fresh, isolated context is used and closed with the + tab. Always closes the tab; only closes the context too when it + was created just for this call (anonymous, session=None) — a named + session's context stays open for reuse.""" + sem = self._tab_semaphore + if sem is None: + await self.start() + sem = self._tab_semaphore + async with sem: + slot = await self._get_context(session) + ctx = slot.context + try: + ctx.set_default_navigation_timeout(self.request_timeout * 1000) + except Exception: + pass + if not slot.patched: + try: + await ctx.add_init_script(STEALTH_INIT_JS) + slot.patched = True + except Exception: + pass + pg = await ctx.new_page() + try: + yield pg + finally: + await pg.close() + + async def __aenter__(self) -> Any: + """``async with stealth as pg:`` claims a tab for exclusive use — + equivalent to ``async with stealth.page() as pg:`` — and hands back + a plain Playwright ``Page`` you can drive with any of its normal + methods. Task-local, so concurrent ``async with stealth as pg`` + blocks from different coroutines on the same shared instance each + get, and release, their own tab independently.""" + cm = self.page() + pg = await cm.__aenter__() + self._claimed_pages[asyncio.current_task()] = cm + return pg + + async def __aexit__(self, *exc_info) -> bool: + cm = self._claimed_pages.pop(asyncio.current_task(), None) + if cm is None: + return False + return bool(await cm.__aexit__(*exc_info)) + + async def render( + self, url: str, lang: str = 'en', wait_until: str = 'domcontentloaded', + collect_links: bool = False, timeout: float = 25.0, settle_ms: int = 500, + session: Optional[str] = None, + ) -> Dict[str, Any]: + """Navigate to ``url`` with full JS execution and return + ``{'text', 'links', 'status'}``. For targets that need a JS + challenge solved (Cloudflare-style interstitials, SPA-rendered + results) — the heavier alternative to get()/post().""" + out: Dict[str, Any] = {'text': '', 'links': [], 'status': None} + async with self.page(lang=lang, session=session) as pg: + resp = await self._goto_with_retry(pg, url, wait_until, timeout) + out['status'] = resp.status if resp else None + await pg.wait_for_timeout(settle_ms) + out['text'] = await pg.evaluate("() => (document.body && document.body.innerText) || ''") + if collect_links: + out['links'] = await pg.evaluate( + """() => Array.from(document.querySelectorAll('a[href]')) + .map(a => ({href: a.href, text: (a.innerText || '').trim().slice(0, 120)})) + .filter(l => /^https?:/.test(l.href)) + .slice(0, 200) + """ + ) + return out + + async def screenshot( + self, url: str, path: Optional[str] = None, lang: str = 'en', + wait_until: str = 'domcontentloaded', full_page: bool = True, + timeout: float = 25.0, settle_ms: int = 500, session: Optional[str] = None, + ) -> Union[str, None]: + """Render ``url`` and capture a PNG screenshot. + + With ``path`` given, saves the PNG there and returns ``path``. + Without it, returns the PNG content as a base64 string — handy for + embedding straight into a data URI or an API response. + """ + async with self.page(lang=lang, session=session) as pg: + await self._goto_with_retry(pg, url, wait_until, timeout) + await pg.wait_for_timeout(settle_ms) + if path: + await pg.screenshot(path=path, full_page=full_page, type='png') + return path + png_bytes = await pg.screenshot(full_page=full_page, type='png') + return base64.b64encode(png_bytes).decode('ascii') + + async def _goto_with_retry(self, pg, url: str, wait_until: str, timeout: float, attempt: int = 1): + try: + return await pg.goto(url, wait_until=wait_until, timeout=timeout * 1000) + except Exception as e: + if attempt >= self.max_retries: + logger.info(f'stealthii: goto {url} failed after {attempt} attempts: {e}') + return None + await asyncio.sleep(0.4 * attempt) + return await self._goto_with_retry(pg, url, wait_until, timeout, attempt=attempt + 1) + + async def _cookie_header(self, ctx, url: str, cookies: Dict[str, str]) -> str: + """Merge ``cookies`` (per-call overrides) on top of whatever the + context already holds for ``url``, and render as a Cookie header + value — mirrors aiohttp/httpx's per-call ``cookies=`` semantics + without mutating the context's own persistent cookie jar.""" + try: + existing = await ctx.cookies(url) + except Exception: + existing = [] + merged = {c['name']: c['value'] for c in existing} + merged.update(cookies) + return '; '.join(f'{k}={v}' for k, v in merged.items()) + + @staticmethod + def _build_body( + headers: Dict[str, str], data: Any, json: Any, files: Optional[Dict[str, Any]], + ) -> Dict[str, Any]: + """Shape (data, json, files) into the one Playwright fetch() body + kwarg that matches aiohttp/httpx semantics: json= is a JSON body, + data=dict is form-urlencoded, data=bytes/str is a raw body, and + files= (optionally combined with data=) is multipart.""" + if files: + multipart: Dict[str, Any] = dict(data) if isinstance(data, dict) else {} + for field, value in files.items(): + if isinstance(value, tuple): + filename, content = value[0], value[1] + else: + filename, content = field, value + multipart[field] = { + 'name': filename, + 'mimeType': 'application/octet-stream', + 'buffer': content if isinstance(content, (bytes, bytearray)) else str(content).encode(), + } + return {'multipart': multipart} + if json is not None: + headers['Content-Type'] = 'application/json' + return {'data': json_module.dumps(json)} + if isinstance(data, dict): + return {'form': data} + if data is not None: + return {'data': data} + return {} + + async def request( + self, method: str, url: str, + params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + data: Any = None, + json: Any = None, + files: Optional[Dict[str, Any]] = None, + cookies: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + allow_redirects: Optional[bool] = None, + follow_redirects: Optional[bool] = None, + max_redirects: int = 20, + timeout: Optional[float] = None, + session: Optional[str] = None, + _attempt: int = 1, + ) -> StealthResponse: + """Issue one HTTP request through a browser context's real Chromium + network stack — no tab is opened, so this is the fast path. + + Accepts the same request shape as aiohttp/httpx: ``params``, + ``headers``, ``data`` (dict = form-urlencoded, bytes/str = raw + body), ``json``, ``files`` (multipart, optionally combined with + ``data`` for the plain fields), ``cookies`` (per-call, layered on + top of the context's own jar), ``auth`` as an ``(user, password)`` + Basic-auth tuple, and both ``allow_redirects`` (aiohttp) and + ``follow_redirects`` (httpx) spellings. + + With ``session`` set, reuses that named context's cookies (e.g. + picked up by an earlier render() warm-up call). Retries recursively + (bounded by ``max_retries``) with a short backoff on any failure. + """ + effective_timeout = self.request_timeout if timeout is None else timeout + slot = await self._get_context(session) + req_headers = dict(headers or {}) + + if auth is not None and 'Authorization' not in req_headers: + token = base64.b64encode(f'{auth[0]}:{auth[1]}'.encode()).decode() + req_headers['Authorization'] = f'Basic {token}' + if cookies: + req_headers['Cookie'] = await self._cookie_header(slot.context, url, cookies) + + follow = follow_redirects if follow_redirects is not None else allow_redirects + follow = True if follow is None else follow + + kwargs: Dict[str, Any] = { + 'headers': req_headers, + 'timeout': effective_timeout * 1000, + 'max_redirects': max_redirects if follow else 0, + } + if params: + kwargs['params'] = params + kwargs.update(self._build_body(req_headers, data, json, files)) + + try: + resp = await slot.context.request.fetch(url, method=method, **kwargs) + body = await resp.body() + return StealthResponse(resp.status, dict(resp.headers), body, resp.url) + except Exception as e: + if _attempt >= self.max_retries: + raise StealthRequestError(f'{method} {url} failed after {_attempt} attempts: {e}') from e + await asyncio.sleep(0.3 * _attempt) + return await self.request( + method, url, params=params, headers=headers, data=data, json=json, + files=files, cookies=cookies, auth=auth, allow_redirects=allow_redirects, + follow_redirects=follow_redirects, max_redirects=max_redirects, + timeout=timeout, session=session, _attempt=_attempt + 1, + ) + + async def get( + self, url: str, params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + cookies: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + allow_redirects: Optional[bool] = None, follow_redirects: Optional[bool] = None, + max_redirects: int = 20, timeout: Optional[float] = None, + session: Optional[str] = None, + ) -> StealthResponse: + return await self.request( + 'GET', url, params=params, headers=headers, cookies=cookies, auth=auth, + allow_redirects=allow_redirects, follow_redirects=follow_redirects, + max_redirects=max_redirects, timeout=timeout, session=session, + ) + + async def post( + self, url: str, params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, data: Any = None, + json: Any = None, files: Optional[Dict[str, Any]] = None, + cookies: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, + allow_redirects: Optional[bool] = None, follow_redirects: Optional[bool] = None, + max_redirects: int = 20, timeout: Optional[float] = None, + session: Optional[str] = None, + ) -> StealthResponse: + return await self.request( + 'POST', url, params=params, headers=headers, data=data, json=json, + files=files, cookies=cookies, auth=auth, allow_redirects=allow_redirects, + follow_redirects=follow_redirects, max_redirects=max_redirects, + timeout=timeout, session=session, + ) + + async def put( + self, url: str, params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, data: Any = None, + json: Any = None, cookies: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, timeout: Optional[float] = None, + session: Optional[str] = None, + ) -> StealthResponse: + return await self.request( + 'PUT', url, params=params, headers=headers, data=data, json=json, + cookies=cookies, auth=auth, timeout=timeout, session=session, + ) + + async def patch( + self, url: str, params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, data: Any = None, + json: Any = None, cookies: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, timeout: Optional[float] = None, + session: Optional[str] = None, + ) -> StealthResponse: + return await self.request( + 'PATCH', url, params=params, headers=headers, data=data, json=json, + cookies=cookies, auth=auth, timeout=timeout, session=session, + ) + + async def delete( + self, url: str, params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + cookies: Optional[Dict[str, str]] = None, + auth: Optional[Tuple[str, str]] = None, timeout: Optional[float] = None, + session: Optional[str] = None, + ) -> StealthResponse: + return await self.request( + 'DELETE', url, params=params, headers=headers, cookies=cookies, + auth=auth, timeout=timeout, session=session, + ) + + async def head( + self, url: str, params: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, timeout: Optional[float] = None, + session: Optional[str] = None, + ) -> StealthResponse: + return await self.request('HEAD', url, params=params, headers=headers, timeout=timeout, session=session) + + async def download( + self, url: str, headers: Optional[Dict[str, str]] = None, + cookies: Optional[Dict[str, str]] = None, + timeout: Optional[float] = None, session: Optional[str] = None, + ) -> bytes: + """Fetch raw bytes (images, PDFs, any binary attachment) through + the same fast, no-tab path as get()/post().""" + resp = await self.get(url, headers=headers, cookies=cookies, timeout=timeout or 30.0, session=session) + return await resp.read() + + async def fingerprint_info(self) -> Dict[str, Any]: + """Self-diagnostic: fetch this client's own current TLS fingerprint + from a public echo endpoint, reported in the configured format + (``ja3`` or ``ja4``). Useful to verify stealth quality after + changes, not needed for normal operation.""" + resp = await self.get(_FINGERPRINT_ECHO_URL) + data = await resp.json() + tls = data.get('tls', {}) if isinstance(data, dict) else {} + key = 'ja3_hash' if self.fingerprint == 'ja3' else 'ja4' + return {'mode': self.fingerprint, 'hash': tls.get(key), 'http_version': data.get('http_version')} + + async def shutdown(self) -> None: + """Close the browser and every context derived from it (anonymous + and named sessions alike). Call once at host process shutdown.""" + async with self._get_lock(): + await self._close_browser_locked() diff --git a/stealthii/stealth_js.py b/stealthii/stealth_js.py new file mode 100644 index 0000000..d695af9 --- /dev/null +++ b/stealthii/stealth_js.py @@ -0,0 +1,119 @@ +# retoor +"""Browser-side JS patches injected into every page/context before any page +script runs. Removes the automation tells that a stock headless Chromium +otherwise exposes (navigator.webdriver, missing window.chrome, empty +PluginArray, SwiftShader WebGL renderer, permissions-query mismatch, +iframe.contentWindow leaking the absence of window.chrome). +""" + +STEALTH_INIT_JS = r""" +(() => { + const patchNativeToString = (fn, name) => { + const src = `function ${name}() { [native code] }`; + const proxied = new Proxy(fn, { + apply(target, thisArg, args) { return Reflect.apply(target, thisArg, args); }, + }); + Object.defineProperty(proxied, 'toString', { + value: () => src, configurable: true, enumerable: false, writable: false, + }); + return proxied; + }; + const define = (obj, prop, getter) => { + try { + Object.defineProperty(obj, prop, { get: patchNativeToString(getter, 'get ' + prop), configurable: true }); + } catch (e) {} + }; + + define(Navigator.prototype, 'webdriver', () => undefined); + + try { + if (!window.chrome || !window.chrome.runtime) { + window.chrome = { + runtime: { + connect: () => {}, sendMessage: () => {}, onMessage: { addListener: () => {} }, + id: undefined, + }, + loadTimes: function () {}, csi: function () {}, app: { + isInstalled: false, + InstallState: { DISABLED: 'disabled', INSTALLED: 'installed', NOT_INSTALLED: 'not_installed' }, + RunningState: { CANNOT_RUN: 'cannot_run', READY_TO_RUN: 'ready_to_run', RUNNING: 'running' }, + }, + }; + } + } catch (e) {} + + try { + const originalQuery = window.navigator.permissions && window.navigator.permissions.query; + if (originalQuery) { + window.navigator.permissions.query = patchNativeToString((parameters) => ( + parameters && parameters.name === 'notifications' + ? Promise.resolve({ state: Notification.permission, onchange: null }) + : originalQuery(parameters) + ), 'query'); + } + } catch (e) {} + + try { + const fakeMimeType = (type, description, suffixes) => ({ type, description, suffixes, enabledPlugin: null }); + const fakePlugin = (name, description, filename, mimes) => { + const p = { name, description, filename, length: mimes.length }; + mimes.forEach((m, i) => { p[i] = m; m.enabledPlugin = p; }); + p.item = (i) => p[i] || null; + p.namedItem = (n) => mimes.find((m) => m.type === n) || null; + return p; + }; + const pdfMime = fakeMimeType('application/pdf', 'Portable Document Format', 'pdf'); + const plugins = [ + fakePlugin('PDF Viewer', 'Portable Document Format', 'internal-pdf-viewer', [pdfMime]), + fakePlugin('Chrome PDF Viewer', 'Portable Document Format', 'internal-pdf-viewer', [pdfMime]), + fakePlugin('Chromium PDF Viewer', 'Portable Document Format', 'internal-pdf-viewer', [pdfMime]), + fakePlugin('Microsoft Edge PDF Viewer', 'Portable Document Format', 'internal-pdf-viewer', [pdfMime]), + fakePlugin('WebKit built-in PDF', 'Portable Document Format', 'internal-pdf-viewer', [pdfMime]), + ]; + const pluginArray = plugins; + pluginArray.item = (i) => pluginArray[i] || null; + pluginArray.namedItem = (n) => pluginArray.find((p) => p.name === n) || null; + pluginArray.refresh = () => {}; + pluginArray.forEach((p) => { + try { Object.setPrototypeOf(p, Plugin.prototype); } catch (e) {} + Object.defineProperty(p, Symbol.toStringTag, { value: 'Plugin', configurable: true }); + }); + Object.defineProperty(pluginArray, Symbol.toStringTag, { value: 'PluginArray', configurable: true }); + try { Object.setPrototypeOf(pluginArray, PluginArray.prototype); } catch (e) {} + define(Navigator.prototype, 'plugins', () => pluginArray); + define(Navigator.prototype, 'mimeTypes', () => { + const arr = [pdfMime]; + Object.defineProperty(arr, Symbol.toStringTag, { value: 'MimeTypeArray', configurable: true }); + try { Object.setPrototypeOf(arr, MimeTypeArray.prototype); } catch (e) {} + return arr; + }); + } catch (e) {} + + define(Navigator.prototype, 'deviceMemory', () => 8); + define(Navigator.prototype, 'hardwareConcurrency', () => 8); + + try { + const spoofVendor = (ctxProto) => { + const original = ctxProto.getParameter; + ctxProto.getParameter = patchNativeToString(function (parameter) { + if (parameter === 37445) return 'Google Inc. (NVIDIA)'; + if (parameter === 37446) return 'ANGLE (NVIDIA, NVIDIA GeForce RTX 3060 Direct3D11 vs_5_0 ps_5_0, D3D11)'; + return original.apply(this, arguments); + }, 'getParameter'); + }; + spoofVendor(WebGLRenderingContext.prototype); + if (window.WebGL2RenderingContext) spoofVendor(WebGL2RenderingContext.prototype); + } catch (e) {} + + try { + const contentWindowDesc = Object.getOwnPropertyDescriptor(HTMLIFrameElement.prototype, 'contentWindow'); + Object.defineProperty(HTMLIFrameElement.prototype, 'contentWindow', { + get: patchNativeToString(function () { + const win = contentWindowDesc.get.call(this); + try { if (win && !win.chrome) win.chrome = window.chrome; } catch (e) {} + return win; + }, 'get contentWindow'), + }); + } catch (e) {} +})(); +""" diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..9b69ea9 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,75 @@ +# retoor +import pytest + +from stealthii import Stealthii, StealthResponse +from stealthii.client import _ContextSlot + + +def test_construction_is_lazy_and_cheap(): + s = Stealthii() + assert s._browser is None + assert s._playwright is None + assert s._contexts == {} + + +def test_invalid_fingerprint_rejected(): + with pytest.raises(ValueError): + Stealthii(fingerprint='ja5') + + +def test_defaults(): + s = Stealthii() + assert s.fingerprint == 'ja4' + assert s.max_tabs == 4 + assert s.max_retries == 3 + assert s.recycle_context_after_count == 500 + assert s.recycle_context_after_s == 1800.0 + + +def test_build_body_json(): + headers = {} + kwargs = Stealthii._build_body(headers, data=None, json={'a': 1}, files=None) + assert kwargs == {'data': '{"a": 1}'} + assert headers['Content-Type'] == 'application/json' + + +def test_build_body_form_dict(): + kwargs = Stealthii._build_body({}, data={'x': '1'}, json=None, files=None) + assert kwargs == {'form': {'x': '1'}} + + +def test_build_body_raw_bytes(): + kwargs = Stealthii._build_body({}, data=b'raw', json=None, files=None) + assert kwargs == {'data': b'raw'} + + +def test_build_body_none(): + assert Stealthii._build_body({}, data=None, json=None, files=None) == {} + + +def test_build_body_multipart_with_files_and_fields(): + kwargs = Stealthii._build_body( + {}, data={'field': 'value'}, json=None, + files={'upload': ('name.txt', b'content')}, + ) + multipart = kwargs['multipart'] + assert multipart['field'] == 'value' + assert multipart['upload']['name'] == 'name.txt' + assert multipart['upload']['buffer'] == b'content' + + +@pytest.mark.asyncio +async def test_stealth_response_shapes_like_aiohttp(): + resp = StealthResponse(200, {'Content-Type': 'application/json'}, b'{"ok": true}', 'https://x/') + assert resp.status == 200 + assert await resp.text() == '{"ok": true}' + assert await resp.json() == {'ok': True} + assert await resp.read() == b'{"ok": true}' + async with resp as r: + assert r is resp + + +def test_context_slot_starts_fresh(): + slot = _ContextSlot(context=object()) + assert slot.count == 0 + assert slot.patched is False diff --git a/tests/test_stealth_detection.py b/tests/test_stealth_detection.py new file mode 100644 index 0000000..741a98d --- /dev/null +++ b/tests/test_stealth_detection.py @@ -0,0 +1,61 @@ +# retoor +import pytest + +from stealthii import Stealthii + + +@pytest.mark.online +@pytest.mark.asyncio +async def test_sannysoft_detection_suite_is_clean(): + """Regression guard: every check on bot.sannysoft.com must pass under + our stealth patches. If this starts failing, a Chromium upgrade likely + changed something the STEALTH_INIT_JS patches assume.""" + s = Stealthii() + try: + async with s.page() as pg: + await pg.goto('https://bot.sannysoft.com/', wait_until='networkidle', timeout=20000) + rows = await pg.evaluate( + """() => Array.from(document.querySelectorAll('table tr')).map(tr => + Array.from(tr.querySelectorAll('td')).map(td => td.innerText.trim()).join(' | ') + )""" + ) + finally: + await s.shutdown() + + failures = [r for r in rows if 'FAIL' in r or 'failed' in r] + assert not failures, f'stealth detection regressions: {failures}' + + +@pytest.mark.online +@pytest.mark.asyncio +async def test_async_with_instance_claims_isolated_concurrent_tabs(): + """`async with stealth as pg` must behave like `stealth.page()`, and be + safe to use concurrently from multiple tasks on one shared instance.""" + import asyncio + + s = Stealthii() + + async def worker(): + async with s as pg: + await pg.goto('https://example.com') + return await pg.title() + + try: + titles = await asyncio.gather(*[worker() for _ in range(3)]) + finally: + await s.shutdown() + assert titles == ['Example Domain'] * 3 + + +@pytest.mark.online +@pytest.mark.asyncio +async def test_fast_path_uses_real_chromium_fingerprint(): + """The no-tab get()/post() fast path must carry a genuine Chromium TLS + fingerprint, distinct from a plain Python HTTP stack's.""" + s = Stealthii() + try: + info = await s.fingerprint_info() + finally: + await s.shutdown() + assert info['hash'] + assert info['mode'] == 'ja4'