retoor/stealthii
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add PUT/PATCH/DELETE/HEAD, full aiohttp/httpx param parity, screenshots,

Browse Source 
async-with tab claiming, and full documentation

- get/post/put/patch/delete/head now accept the same request shape as
  aiohttp/httpx: data/json/files/cookies/auth/allow_redirects/
  follow_redirects/max_redirects, plus the existing session= for named
  cookie-persisting contexts.
- screenshot(url, path=None) captures a PNG, saved to disk or returned
  as base64.
- `async with stealth as page` claims a tab directly, task-safe for
  concurrent use on one shared instance (keyed off asyncio.current_task,
  not instance state).
- Strip all docstrings/comments from the source (author line excepted),
  per house style.
- README.md rewritten as the complete documentation: rationale, JA3/JA4
  comparison, dependencies, construction/laziness/lifetime, every method
  with examples, named sessions, configuration, error handling, testing.
- STEALTH_PATCHES.md: a literal, patch-by-patch account of every launch
  argument and JS patch applied, with rationale and verification method.
- 20 new tests: full HTTP-method coverage, cookies, auth, multipart,
  download, render, screenshot, named-session isolation, retry/error
  paths, and guaranteed page/context cleanup under exception.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0116i7dYLNZTXNR7Lqf439bV
This commit is contained in:
retoor
2026-09-09 13:56:28 +02:00
co-authored by Claude Sonnet 5
parent a3bb7639d2
commit 86c3a95cb8
7 changed files with 540 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+250 -78
View File
@@ -2,45 +2,68 @@
`retoor <retoor@molodetz.nl>`
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.
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.
## Why
## Why this exists
`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.
`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 present 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.
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. Measured directly
(`tests/test_stealth_detection.py::test_fast_path_uses_real_chromium_fingerprint`)
against a plain `aiohttp` client hitting the same TLS-fingerprint echo
endpoint:
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.
| Client | JA3 | JA4 |
|---|---|---|
| stealthii (`context.request`) | `944d1e1858cd278718f8a46b65d3212f` | `t13d5211_b262b3658495_8e6e362c5eac` |
| aiohttp | `304734bb1c086c3453b387400cf83f11` | `t13d1812h1_85036bcba153_d41ae481755e` |
## What it does not do
Distinct hashes, produced by the same physical request path our
`get`/`post`/etc. use — proof the fast path is not just claiming to be
Chrome, it is Chrome's own network stack.
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.
On top of that, `stealthii` removes the JS-level automation tells a
stock headless Chromium exposes — `navigator.webdriver`, a missing
`window.chrome`, an empty plugin list, a SwiftShader WebGL renderer, a
permissions-query mismatch. Every patch and its exact rationale is
documented in [`STEALTH_PATCHES.md`](./STEALTH_PATCHES.md); the result
is verified clean against `bot.sannysoft.com`'s full detection suite on
every test run.
## What this 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=`, below) for those targets; do not expect it alone
to be a universal bypass. See "Known limitations" in
[`STEALTH_PATCHES.md`](./STEALTH_PATCHES.md) for the specific gaps,
including that the fast request path negotiates HTTP/1.1 rather than
HTTP/2.
## Dependencies
Exactly one: [`playwright`](https://playwright.dev/python/) (`>=1.40.0`),
declared in `pyproject.toml`. No other third-party runtime dependency —
no `requests`, no `aiohttp`, no stealth-plugin package. The Chromium
binary itself is installed separately via `playwright install chromium`
(see Install, below); it is Playwright's own dependency, not
stealthii's.
## Install
@@ -49,90 +72,229 @@ pip install -e .
playwright install chromium
```
## Usage
## The `Stealthii` object: construction, laziness, lifetime
`Stealthii()` does no I/O. It only records configuration. The browser
process, its stealth-patched contexts, and every tab are created lazily,
on the first call that actually needs them — the first `get()`,
`post()`, `render()`, `page()`, or an explicit `start()`.
```python
from stealthii import Stealthii
stealth = Stealthii() # cheap, lazy — launches nothing yet
resp = await stealth.get("https://example.com/api", params={"q": "x"})
stealth = Stealthii() # cheap — nothing has launched yet
```
An instance is *not* a Python singleton in the enforced sense (nothing
stops you from constructing several) — but a `Stealthii` instance owns
one Chromium process and every context derived from it, so the intended
usage is exactly like an `aiohttp.ClientSession` or an `httpx.Client`:
construct one, hand it to whatever needs it, and share it for the
lifetime of your application. Constructing a second instance launches a
second, entirely separate Chromium process; there is no coordination
between instances.
Call `start()` once at application boot if you want the browser launch
latency paid up front rather than by whichever request happens to be
first:
```python
await stealth.start() # optional — every method below also does this lazily
```
Call `shutdown()` once when your application exits:
```python
await stealth.shutdown() # closes every context and the browser process
```
In an `aiohttp` web application, wire these into `on_startup`/`on_cleanup`;
in any other async app, wherever your own startup/shutdown hooks live.
## HTTP methods
`get`, `post`, `put`, `patch`, `delete`, `head` all accept the request
shape you already know from `aiohttp`/`httpx`:
```python
resp = await stealth.get(url, params={"q": "x"}, headers={...}, timeout=15)
resp.status
await resp.json()
await resp.text()
await stealth.shutdown() # call once at host process shutdown
await resp.read() # raw bytes
```
`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`.
- `params` — query string parameters (dict).
- `headers` — request headers (dict).
- `data` — a `dict` is sent form-urlencoded; `bytes`/`str` is sent as
the raw body. Matches both libraries' `data=` behaviour.
- `json` — JSON-encoded body; sets `Content-Type: application/json`.
- `files` — multipart file upload, `httpx`-style:
`files={"upload": ("name.txt", b"content")}`. Combine with `data` for
the plain form fields alongside the file(s).
- `cookies` — a per-call `dict`, layered on top of whatever the
context's own cookie jar already holds for that URL, without
mutating the jar. Matches `aiohttp`/`httpx`'s per-call `cookies=`.
- `auth` — an `(username, password)` tuple, sent as HTTP Basic auth.
- `allow_redirects` (the `aiohttp` spelling) and `follow_redirects`
(the `httpx` spelling) are both accepted; either disables following
redirects when passed `False`.
- `max_redirects` — cap when following redirects (default `20`).
- `timeout` — seconds; falls back to the instance's `request_timeout`.
- `session` — see "Named sessions", below.
```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"]
await stealth.post(url, json={"hello": "world"})
await stealth.post(url, data={"x": "1"}) # form-urlencoded
await stealth.post(url, data={"field": "value"},
files={"upload": ("name.txt", b"content")}) # multipart
await stealth.get(url, cookies={"flavor": "choc"})
await stealth.get(url, auth=("user", "pass"))
await stealth.get(url, allow_redirects=False)
```
### Named sessions (cookie persistence / warm-up)
None of these open a tab — they go through the shared context's
`context.request` fast path, described above.
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:
### `download()`
Raw bytes, for images, PDFs, or any other binary attachment — the same
fast, no-tab path as `get()`:
```python
await stealth.render("https://example.com/", session="acme") # picks up cookies
resp = await stealth.get("https://example.com/api", session="acme") # reuses them
png_bytes = await stealth.download("https://example.com/photo.png")
```
### Raw page access
## `render()` and `screenshot()`: full page execution
For anything the high-level API doesn't cover, `page()` yields a real
Playwright `Page` (already stealth-patched), guaranteed to close even on
exception:
For targets that need a JS challenge solved — an SPA that only renders
results client-side, a Cloudflare-style "verifying your browser"
interstitial — escalate to a real page load with `render()`:
```python
result = await stealth.render(url, collect_links=True)
result["status"] # HTTP status of the final navigation
result["text"] # document.body.innerText after JS has run and settled
result["links"] # only populated when collect_links=True
```
`screenshot()` renders the same way and captures a PNG:
```python
png_b64 = await stealth.screenshot(url) # base64 string
path = await stealth.screenshot(url, path="shot.png") # saved to disk, path returned
```
With `path` given, the file is written and `path` is returned. Without
it, the PNG bytes come back base64-encoded — convenient for embedding
directly in a data URI or a JSON API response.
## Raw page access: `page()` and `async with stealth as page`
For anything the high-level API does not cover, `page()` yields a real
Playwright `Page` — already stealth-patched — and guarantees it (and,
for an anonymous call, its context) closes again even if your code
raises:
```python
async with stealth.page() as pg:
await pg.goto("https://example.com")
await pg.goto(url)
await pg.click("#accept-cookies")
await pg.fill("#search", "query")
```
`Stealthii` also supports using the instance itself as the context
manager, claiming a tab directly:
```python
async with stealth as pg:
await pg.goto(url)
title = await pg.title()
```
This is exactly `page()` under the hood, with one guarantee on top:
it is task-safe. Two coroutines running `async with stealth as pg` at
the same time on the same shared `Stealthii` instance each get, and
independently release, their own tab — the claim is tracked per
`asyncio` task, not on the instance itself, so concurrent use never
cross-wires which `pg` belongs to which caller.
## Named sessions: cookie persistence and warm-up
Every call above is anonymous by default — it uses one shared context
that is transparently recycled (see below), with no cookie continuity
implied between calls. Pass `session="some-name"` to any of
`get`/`post`/`put`/`patch`/`delete`/`head`/`download`/`render`/
`screenshot`/`page` to use a dedicated, reused context under that name
instead. Cookies set under one name persist across every subsequent
call using that same name, in either direction — a `render()` call can
warm up a session that a later `get()` then reuses, or vice versa,
because both operate on the very same underlying browser context:
```python
await stealth.render("https://example.com/login", session="acme") # solves a JS challenge, sets cookies
resp = await stealth.get("https://example.com/api/data", session="acme") # reuses those cookies
```
Anonymous calls and named-session calls never share cookies with each
other — each name (including the implicit `None` for anonymous calls)
is its own isolated context.
## Configuration
All configuration is passed once, at construction — `Stealthii(...)` does
no I/O itself; the browser launches lazily on first real use:
Everything is set once, at construction:
```python
stealth = Stealthii(
max_tabs=4, # concurrent render()/page() tabs
max_tabs=4, # concurrent render()/page()/screenshot() 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
request_timeout=15.0, # default seconds for get/post/...
max_retries=3, # attempts before raising, for both request() and render()/goto
disk_cache_bytes=1, # Chromium's own on-disk cache size, kept near-zero
recycle_context_after=(500, 1800.0), # (max requests, max seconds) per context, named or not
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.
`fingerprint` selects which hash `fingerprint_info()` reports — it does
not change what is sent on the wire; both values describe the same
genuine Chromium handshake, just in a different published format:
```python
await stealth.fingerprint_info()
# {"mode": "ja4", "hash": "t13d5211_...", "http_version": "HTTP/1.1"}
```
### 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.
A browser that runs for the entire lifetime of a process otherwise
accumulates cookies and an on-disk HTTP cache without bound — a slow,
silent resource leak in anything long-running. `stealthii` closes and
recreates any context, anonymous or named, once it crosses
`recycle_context_after` (request count or age, whichever comes first),
and fully relaunches the browser process — discarding its whole profile
directory — every `recycle_browser_after_seconds`. A named session that
needs to stay warm longer than the default 30 minutes / 500 requests
should get a larger `recycle_context_after` passed at construction.
## Error handling
- `StealthUnavailableError` — the browser itself could not be started
(Playwright not installed, or the Chromium launch failed). Raised
from `start()` and anything that needs the browser.
- `StealthRequestError` — a `get`/`post`/`put`/`patch`/`delete`/`head`/
`download` call failed outright after `max_retries` attempts, each
separated by a short backoff. The retry is recursive and bounded, not
an unbounded loop.
- `render()`/`screenshot()`/`page()`'s navigation retries the same way,
bounded by `max_retries`, but returns a `status: None` result on
final failure rather than raising, since a page render already
produces a body of best-effort content even when the final navigation
attempt did not fully succeed.
- Both the tab (`page()`) and, for anonymous calls, its context are
guaranteed to close in a `finally`, even when your code inside
`async with stealth.page() as pg:` (or `async with stealth as pg:`)
raises.
## Testing
@@ -140,3 +302,13 @@ sessions meant to stay warm longer should get a larger
make test-unit # no network required
make test-integration # hits real sites; requires Chromium installed
```
`tests/test_client.py` covers construction, configuration, and the
pure request-shaping logic without any network access.
`tests/test_http_methods.py` and `tests/test_stealth_detection.py` are
integration tests (`@pytest.mark.online`) that exercise every method
above — including named-session persistence, per-call cookies, auth,
multipart upload, retries against an unreachable host, and the
sannysoft stealth regression suite — against real endpoints.
`scripts/smoke_live.py` is a plain manual script covering the same
ground end-to-end, useful for a quick sanity check outside pytest.
STEALTH_PATCHES.md
+132
View File
@@ -0,0 +1,132 @@
# Stealth Patches
`retoor <retoor@molodetz.nl>`
This document enumerates every modification stealthii applies to the
underlying Chromium/Playwright stack and the rationale for each. It is
maintained as living documentation: any change to `stealth_js.py` or the
launch arguments in `client.py` must be reflected here in the same commit.
## Launch-level patches
Applied in `Stealthii._ensure_browser_locked()` at browser launch.
| Flag | Effect |
|---|---|
| `--disable-blink-features=AutomationControlled` | Removes the `navigator.webdriver` flag and associated automation-only code paths at the engine level, before any JS patch runs. |
| `ignore_default_args=['--enable-automation']` | Playwright passes `--enable-automation` to Chromium by default, which enables the automation infobar and related detectable behaviour. Suppressed. |
| `--no-sandbox` | Required to run headless Chromium as a non-root, unprivileged process in this deployment environment. Not a stealth measure. |
| `--disable-dev-shm-usage` | Avoids `/dev/shm` exhaustion under containerized/limited-memory environments. Not a stealth measure. |
| `--disk-cache-size=<bytes>` / `--media-cache-size=<bytes>` | Bounds Chromium's own on-disk HTTP/media cache, configurable via `disk_cache_bytes` (default near-zero). Not a stealth measure — an operational safeguard against unbounded disk growth in a long-lived process. |
## JS-level patches
Applied via `context.add_init_script(STEALTH_INIT_JS)` on every context
before any page script runs, so the patched state is present from the very
first document, including cross-origin iframes.
### `navigator.webdriver`
Redefined to return `undefined`. This flag is `true` on any
CDP-automated Chromium and is the single most widely checked automation
signal.
### `window.chrome`
A stock headless Chromium launched via Playwright does not expose
`window.chrome` at all, unlike a real Chrome install. When absent, a
`window.chrome` object is constructed with `runtime`, `loadTimes`, `csi`,
and `app` members shaped like the real object's public surface.
### `navigator.permissions.query`
A stock headless Chromium reports `denied` for a `notifications`
permission query while `Notification.permission` itself still reports
`default` — a mismatch a real browser never produces. The query is
patched to always return the same value as `Notification.permission`
for that one permission name; all other permission names are passed
through to the original implementation unmodified.
### `navigator.plugins` / `navigator.mimeTypes`
A stock headless Chromium reports an empty plugin list and a
`navigator.plugins` value that is a plain array rather than a
`PluginArray`. Five entries matching Chrome's real built-in PDF-plugin
set (`PDF Viewer`, `Chrome PDF Viewer`, `Chromium PDF Viewer`,
`Microsoft Edge PDF Viewer`, `WebKit built-in PDF`) are constructed, each
individually reassigned the real `Plugin.prototype`, and the containing
array reassigned `PluginArray.prototype` — including
`Symbol.toStringTag`, so both a property count check and a
`Object.prototype.toString.call(...)` type check pass. `navigator.mimeTypes`
is patched the same way for the corresponding `application/pdf` MIME type.
### `navigator.deviceMemory`, `navigator.hardwareConcurrency`
Set to `8` each. Present mainly for consistency with the spoofed WebGL
GPU (below) and a realistic desktop profile; a stock headless Chromium
already reports plausible values here in most cases.
### `WebGLRenderingContext.prototype.getParameter` / `WebGL2RenderingContext.prototype.getParameter`
A stock headless Chromium renders WebGL through SwiftShader (a software
rasterizer), which reports `UNMASKED_VENDOR_WEBGL` (constant `37445`) as
`Google Inc. (Google)` and `UNMASKED_RENDERER_WEBGL` (constant `37446`)
containing the literal string `SwiftShader` — an unambiguous headless
signal checked by essentially every fingerprinting script. Both
constants are intercepted and rewritten to report a real discrete GPU
(`Google Inc. (NVIDIA)` / an ANGLE-wrapped `NVIDIA GeForce RTX 3060`
string); every other parameter is passed through unmodified.
### `HTMLIFrameElement.prototype.contentWindow`
A known secondary check reads `window.chrome` from inside a freshly
created, same-origin iframe rather than the top-level document — since
each document gets its own JS globals, an iframe's `window.chrome` can
be absent even when the top-level one was patched. The `contentWindow`
getter is wrapped so that, whenever the returned window is missing
`chrome`, it is assigned from the outer patched value before being
returned.
### `Function.prototype.toString` spoofing on every patch above
Every function or getter installed by the patches above is wrapped in a
`Proxy` whose own `toString` is overridden to return the exact string a
real native function produces (`function name() { [native code] }`).
Without this, a page can distinguish a patched native method from a real
one by calling `.toString()` on it and observing actual JavaScript source
instead of the native-code marker — a check used by several stealth-
detection scripts specifically to unmask other stealth patches.
## Verification
Patches are verified against
[bot.sannysoft.com](https://bot.sannysoft.com)'s full detection table
(`tests/test_stealth_detection.py::test_sannysoft_detection_suite_is_clean`),
run on every change. As of the current `STEALTH_INIT_JS`, every row on
that page passes, including `WebDriver (New)`, `Chrome (New)`,
`Plugins is of type PluginArray`, `WebGL Vendor`/`WebGL Renderer`, and
the `HEADCHR_*`/`PHANTOM_*`/`SELENIUM_DRIVER` probe blocks.
The `get`/`post`/`put`/`patch`/`delete`/`head`/`download` fast path (no
tab, `context.request`) is separately verified
(`test_fast_path_uses_real_chromium_fingerprint`) to carry a JA3/JA4
hash distinct from a plain Python `aiohttp` client, confirming it uses
the browser's own TLS stack rather than Python's.
## Known limitations
- The `context.request` fast path negotiates HTTP/1.1, not HTTP/2 — a
real Chrome page load always uses HTTP/2 against an HTTP/2-capable
server. A system that cross-checks the TLS fingerprint against the
negotiated protocol can detect this specific inconsistency. `render()`
(full page navigation) does not have this limitation.
- None of the above defeats IP-reputation-based rate limiting, nor the
behavioural layer of systems such as DataDome or a Cloudflare managed
challenge. Those require a clean source IP and, for the hardest
targets, human-like interaction; stealthii's patches remove
fingerprint-level tells, not behavioural ones.
- Verified against Chromium's stock headless build as shipped with
Playwright at the time of writing. A Chromium upgrade can change
default values the patches assume (plugin list shape, WebGL constants,
permission defaults); re-run the sannysoft regression test after any
Playwright/Chromium version bump.
stealthii/__init__.py
-5
View File
@@ -1,9 +1,4 @@
# retoor <retoor@molodetz.nl>
"""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
stealthii/client.py
+2 -128
View File
@@ -1,25 +1,4 @@
# retoor <retoor@molodetz.nl>
"""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
@@ -40,24 +19,17 @@ _LAUNCH_ARGS = [
]
_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."""
pass
class StealthRequestError(Exception):
"""A request through the stealth browser failed outright, after
exhausting retries."""
pass
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):
@@ -93,40 +65,6 @@ class _ContextSlot:
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,
@@ -163,16 +101,11 @@ class Stealthii:
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:
@@ -227,11 +160,6 @@ class Stealthii:
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()
@@ -257,13 +185,6 @@ class Stealthii:
@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()
@@ -288,12 +209,6 @@ class Stealthii:
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
@@ -310,10 +225,6 @@ class Stealthii:
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)
@@ -335,12 +246,6 @@ class Stealthii:
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)
@@ -361,10 +266,6 @@ class Stealthii:
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:
@@ -377,10 +278,6 @@ class Stealthii:
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():
@@ -419,21 +316,6 @@ class Stealthii:
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 {})
@@ -551,16 +433,10 @@ class Stealthii:
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 {}
@@ -568,7 +444,5 @@ class Stealthii:
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()
stealthii/stealth_js.py
-7
View File
@@ -1,11 +1,4 @@
# retoor <retoor@molodetz.nl>
"""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) => {
tests/test_http_methods.py
+156
View File
@@ -0,0 +1,156 @@
# retoor <retoor@molodetz.nl>
import pytest
from stealthii import Stealthii, StealthRequestError
@pytest.fixture
async def stealth():
s = Stealthii()
yield s
await s.shutdown()
@pytest.mark.online
@pytest.mark.asyncio
async def test_get_json(stealth):
resp = await stealth.get('https://httpbin.org/get', params={'a': '1'})
assert resp.status == 200
assert (await resp.json())['args'] == {'a': '1'}
@pytest.mark.online
@pytest.mark.asyncio
async def test_post_json_body(stealth):
resp = await stealth.post('https://httpbin.org/post', json={'hello': 'world'})
assert resp.status == 200
assert (await resp.json())['json'] == {'hello': 'world'}
@pytest.mark.online
@pytest.mark.asyncio
async def test_post_form_data(stealth):
resp = await stealth.post('https://httpbin.org/post', data={'x': '1', 'y': '2'})
assert resp.status == 200
assert (await resp.json())['form'] == {'x': '1', 'y': '2'}
@pytest.mark.online
@pytest.mark.asyncio
async def test_put(stealth):
resp = await stealth.put('https://httpbin.org/put', json={'k': 'v'})
assert resp.status == 200
assert (await resp.json())['json'] == {'k': 'v'}
@pytest.mark.online
@pytest.mark.asyncio
async def test_patch(stealth):
resp = await stealth.patch('https://httpbin.org/patch', json={'k': 'v'})
assert resp.status == 200
@pytest.mark.online
@pytest.mark.asyncio
async def test_delete(stealth):
resp = await stealth.delete('https://httpbin.org/delete')
assert resp.status == 200
@pytest.mark.online
@pytest.mark.asyncio
async def test_head(stealth):
resp = await stealth.head('https://httpbin.org/get')
assert resp.status == 200
@pytest.mark.online
@pytest.mark.asyncio
async def test_per_call_cookies_layer_on_context_jar(stealth):
resp = await stealth.get('https://httpbin.org/cookies', cookies={'flavor': 'choc'})
assert (await resp.json())['cookies'] == {'flavor': 'choc'}
@pytest.mark.online
@pytest.mark.asyncio
async def test_basic_auth(stealth):
ok = await stealth.get('https://httpbin.org/basic-auth/user/pass', auth=('user', 'pass'))
assert ok.status == 200
bad = await stealth.get('https://httpbin.org/basic-auth/user/pass', auth=('user', 'wrong'))
assert bad.status == 401
@pytest.mark.online
@pytest.mark.asyncio
async def test_download_binary(stealth):
data = await stealth.download('https://httpbin.org/image/png')
assert data[:8] == b'\x89PNG\r\n\x1a\n'
@pytest.mark.online
@pytest.mark.asyncio
async def test_multipart_file_upload(stealth):
resp = await stealth.post(
'https://httpbin.org/post',
data={'field': 'value'},
files={'upload': ('name.txt', b'file-content')},
)
body = await resp.json()
assert body['form'] == {'field': 'value'}
assert body['files']['upload'] == 'file-content'
@pytest.mark.online
@pytest.mark.asyncio
async def test_render_returns_text_and_links(stealth):
out = await stealth.render('https://example.com', collect_links=True)
assert out['status'] == 200
assert 'Example Domain' in out['text']
assert isinstance(out['links'], list)
@pytest.mark.online
@pytest.mark.asyncio
async def test_screenshot_base64_and_path(stealth, tmp_path):
b64 = await stealth.screenshot('https://example.com')
assert len(b64) > 1000
out_path = str(tmp_path / 'shot.png')
result_path = await stealth.screenshot('https://example.com', path=out_path)
assert result_path == out_path
with open(out_path, 'rb') as f:
assert f.read(8) == b'\x89PNG\r\n\x1a\n'
@pytest.mark.online
@pytest.mark.asyncio
async def test_named_session_persists_cookies_across_calls(stealth):
await stealth.get('https://httpbin.org/cookies/set/persisted/yes', session='warm')
resp = await stealth.get('https://httpbin.org/cookies', session='warm')
assert (await resp.json())['cookies'] == {'persisted': 'yes'}
@pytest.mark.online
@pytest.mark.asyncio
async def test_anonymous_calls_do_not_share_named_session_cookies(stealth):
await stealth.get('https://httpbin.org/cookies/set/persisted/yes', session='isolated')
resp = await stealth.get('https://httpbin.org/cookies')
assert (await resp.json())['cookies'] == {}
@pytest.mark.online
@pytest.mark.asyncio
async def test_request_raises_stealth_request_error_on_unreachable_host(stealth):
with pytest.raises(StealthRequestError):
await stealth.get('https://this-host-does-not-exist.invalid/', timeout=3)
@pytest.mark.online
@pytest.mark.asyncio
async def test_page_and_context_close_even_when_caller_raises(stealth):
pages = []
with pytest.raises(ValueError):
async with stealth.page() as pg:
pages.append(pg)
raise ValueError('boom')
assert pages[0].is_closed()
tests/test_stealth_detection.py
-7
View File
@@ -7,9 +7,6 @@ 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:
@@ -29,8 +26,6 @@ async def test_sannysoft_detection_suite_is_clean():
@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()
@@ -50,8 +45,6 @@ async def test_async_with_instance_claims_isolated_concurrent_tabs():
@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()