rethink-public/aioweb_tls
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add package: pyproject + src

Browse Source 
TLSSession over aioweb's backend seam by composition: one session class
delegates the four seams to an injected backend. ships CurlCffi (curl_cffi
impersonate) and Noble (noble_tls Client) backends plus the TLSBackend
protocol for custom clients. tls clients are optional extras
([curl]/[noble]/[all]) with guarded imports; all aioweb features (domain/
header/ephemeral/proxy/retry/preview) inherited unchanged. src/ multi-module
layout, hatchling.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
disqualifier 2026-06-24 18:49:51 -04:00
parent aa378887d8
commit befa4cd196
5 changed files with 412 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
pyproject.toml Normal file
View File

@ -0,0 +1,23 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "aioweb_tls"
version = "0.1.0"
description = "TLS-fingerprinting backends for aioweb — curl_cffi / noble_tls ExtendedSession subclasses, config-free, installable."
requires-python = ">=3.10"
dependencies = [
"aioweb @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioweb.git@v0.1.0",
]
[project.optional-dependencies]
curl = ["curl_cffi>=0.15"]
noble = ["noble-tls>=0.1.5"]
all = ["curl_cffi>=0.15", "noble-tls>=0.1.5"]
[tool.hatch.build.targets.wheel]
packages = ["src/aioweb_tls"]
[tool.hatch.metadata]
allow-direct-references = true

28
src/aioweb_tls/__init__.py Normal file
View File

@ -0,0 +1,28 @@
"""
tls-fingerprinting backends for aioweb
one session class, TLSSession, takes an injected backend that swaps the HTTP client
(and thus the TLS/HTTP fingerprint) while inheriting every aioweb feature — header
overwrites, domain rewriting, ephemeral headers, proxies, retries, previews —
unchanged.
from aioweb_tls import TLSSession, CurlCffi, Noble
async with TLSSession(backend=CurlCffi(impersonate="chrome")) as s: # [curl] extra
resp = await s.request_with_retries("GET", url)
async with TLSSession(backend=Noble(client="chrome_133")) as s: # [noble] extra
await s.setup() # fetch Go lib once
resp = await s.request_with_retries("GET", url)
the tls clients are optional extras, not base deps. importing this package never
fails because an extra is missing; the matching RuntimeError is raised only when you
construct a backend whose client isn't installed. custom backends implement the
TLSBackend protocol and inject the same way.
"""
from .session import TLSSession
from .backends import CurlCffi, Noble
from .protocol import TLSBackend
__all__ = ["TLSSession", "CurlCffi", "Noble", "TLSBackend"]

207
src/aioweb_tls/backends.py Normal file
View File

@ -0,0 +1,207 @@
"""
tls backends for TLSSession
each backend is a stateless config+behavior object implementing the TLSBackend
protocol (see protocol.py). it owns its own config vocabulary — CurlCffi takes
impersonate=, Noble takes client= — so there is no shared kwarg-soup. TLSSession
owns the live session object and passes it into every method here.
the underlying tls clients are optional extras: constructing a backend whose client
is not installed raises a clear RuntimeError naming the extra to install. importing
this module never fails because an extra is missing.
"""
import logging
from aioweb import Response
log = logging.getLogger(__name__)
try:
from curl_cffi import AsyncSession as _CurlAsyncSession
_CURL_ERROR = None
except ImportError as error:
_CurlAsyncSession = None
_CURL_ERROR = error
try:
import noble_tls
from noble_tls import Client as _NobleClient
_NOBLE_ERROR = None
except ImportError as error:
noble_tls = None
_NobleClient = None
_NOBLE_ERROR = error
def _coerce_timeout(value):
"""turn aioweb's aiohttp ClientTimeout (or a number) into a plain number
aioweb.request() wraps a numeric timeout in an aiohttp.ClientTimeout before the
seam sees it; the tls clients want a number, so unwrap .total when present.
"""
total = getattr(value, "total", value)
return total if isinstance(total, (int, float)) else None
def _jar_to_dict(session):
"""best-effort map of a requests-style cookie jar on session to a plain dict"""
jar = getattr(session, "cookies", None)
if not jar:
return {}
try:
return {k: v for k, v in jar.items()}
except Exception:
return {}
class CurlCffi:
"""curl_cffi backend: forges JA3/JA4 + HTTP/2 via the curl-impersonate binary
config:
impersonate: browser profile to forge (default "chrome"); override per call
by passing impersonate= to any request method.
requires the [curl] extra (pip install "aioweb_tls[curl]").
"""
def __init__(self, impersonate: str = "chrome"):
if _CurlAsyncSession is None:
raise RuntimeError(
"curl_cffi is not installed; install the extra with "
"pip install \"aioweb_tls[curl]\""
) from _CURL_ERROR
self.impersonate = impersonate
def create_session(self, headers, timeout, **kwargs):
"""build the curl_cffi AsyncSession"""
return _CurlAsyncSession(headers=headers, timeout=timeout, **kwargs)
async def raw_request(self, session, method, url, **kwargs) -> Response:
"""send via curl_cffi and adapt the result into an aioweb.Response"""
impersonate = kwargs.pop("impersonate", self.impersonate)
timeout = _coerce_timeout(kwargs.pop("timeout", None))
if timeout is not None:
kwargs["timeout"] = timeout
proxy = kwargs.pop("proxy", None)
if proxy:
kwargs["proxy"] = proxy
response = await session.request(method, url, impersonate=impersonate, **kwargs)
content = response.content
if content is None:
content = response.text.encode() if response.text else b""
return Response(
status_code=response.status_code,
headers=dict(response.headers),
content=content,
url=str(response.url),
reason=getattr(response, "reason", None),
cookies=getattr(response, "cookies", None),
)
def is_closed(self, session) -> bool:
"""whether the curl_cffi session is closed"""
return bool(getattr(session, "closed", False))
def cookies_for_url(self, session, url) -> dict:
"""cookies curl_cffi would send for url (best-effort)"""
return _jar_to_dict(session)
async def close(self, session) -> None:
"""close the curl_cffi session"""
await session.close()
class Noble:
"""noble_tls backend: forges a browser TLS fingerprint via the tls-client Go lib
config:
client: noble_tls Client profile (enum or string, default "chrome_133").
noble_tls downloads a Go shared library on first use; setup() fetches it once
(run via TLSSession.setup() or lazily before the first request).
requires the [noble] extra (pip install "aioweb_tls[noble]").
"""
def __init__(self, client="chrome_133"):
if noble_tls is None:
raise RuntimeError(
"noble_tls is not installed; install the extra with "
"pip install \"aioweb_tls[noble]\""
) from _NOBLE_ERROR
self.client = self._resolve_client(client)
self._updated = False
@staticmethod
def _resolve_client(client):
"""turn a string or Client enum into a noble_tls Client value"""
if isinstance(client, str):
return getattr(_NobleClient, client.upper())
return client
async def setup(self) -> None:
"""fetch the noble_tls Go shared library once; idempotent
download_if_necessary handles the first-time fetch (no lib present);
update_if_necessary refreshes an existing one. try download first so a
clean environment works, falling back to update.
"""
if self._updated:
return
download = getattr(noble_tls, "download_if_necessary", None)
if download is not None:
await download()
else:
await noble_tls.update_if_necessary()
self._updated = True
def create_session(self, headers, timeout, **kwargs):
"""build the noble_tls Session"""
return noble_tls.Session(client=self.client, **kwargs)
async def raw_request(self, session, method, url, **kwargs) -> Response:
"""send via noble_tls and adapt the result into an aioweb.Response"""
await self.setup()
timeout = _coerce_timeout(kwargs.pop("timeout", None))
if timeout is not None:
kwargs["timeout_seconds"] = int(timeout)
proxy = kwargs.pop("proxy", None)
if proxy:
kwargs["proxy"] = proxy
response = await session.execute_request(method=method.upper(), url=url, **kwargs)
content = getattr(response, "content", None)
if content is None:
text = getattr(response, "text", "") or ""
content = text.encode()
return Response(
status_code=response.status_code,
headers=dict(getattr(response, "headers", {}) or {}),
content=content,
url=str(getattr(response, "url", url)),
reason=getattr(response, "reason", None),
cookies=getattr(response, "cookies", None),
)
def is_closed(self, session) -> bool:
"""noble_tls has no closed flag; closed state is tracked by TLSSession"""
return bool(getattr(session, "closed", False))
def cookies_for_url(self, session, url) -> dict:
"""cookies noble_tls would send for url (best-effort)"""
return _jar_to_dict(session)
async def close(self, session) -> None:
"""close the noble_tls session if it exposes a close"""
close = getattr(session, "close", None)
if close is None:
return
result = close()
if hasattr(result, "__await__"):
await result

60
src/aioweb_tls/protocol.py Normal file
View File

@ -0,0 +1,60 @@
"""
the tls backend protocol
a backend is a stateless config+behavior object that teaches TLSSession how to talk
to one HTTP client. TLSSession owns the live session object (built by create_session)
and passes it into every backend call, so backends hold no per-request state.
implement this protocol to add a custom backend (e.g. a local Go TLS server); inject
it via TLSSession(backend=MyBackend(...)) and it inherits all of aioweb's domain /
header / ephemeral / proxy / retry / preview logic unchanged — those operate on plain
dicts and never touch the backend.
required:
create_session(headers: dict, timeout, **kwargs) -> session
build and return the live client session object. TLSSession stores it as
self.session and hands it back to every other method below.
async raw_request(session, method, url, **kwargs) -> aioweb.Response
send one request with `session` and adapt the client's response into an
aioweb.Response built from primitives (status_code, headers, content bytes,
url, reason). kwargs arrive aioweb-shaped — the base has already resolved the
proxy into kwargs["proxy"] and merged headers into kwargs["headers"], and a
numeric timeout is wrapped in an aiohttp.ClientTimeout (unwrap .total).
is_closed(session) -> bool
whether `session` is closed.
optional:
cookies_for_url(session, url) -> dict
cookies the client would send for url, for preview(). default {} (used when
the backend has no introspectable jar).
async setup() -> None
one-time async preparation (e.g. fetch a native lib). called once via
TLSSession.setup() and lazily before the first request; make it idempotent.
async close(session) -> None
close `session`. default awaits session.close() if present.
"""
from typing import Any, Protocol, runtime_checkable
from aioweb import Response
@runtime_checkable
class TLSBackend(Protocol):
"""the contract a TLSSession backend implements (see module docstring)"""
def create_session(self, headers: dict, timeout: Any, **kwargs) -> Any:
"""build and return the live client session object"""
...
async def raw_request(self, session: Any, method: str, url: str, **kwargs) -> Response:
"""send one request and adapt the result into an aioweb.Response"""
...
def is_closed(self, session: Any) -> bool:
"""whether the session is closed"""
...

94
src/aioweb_tls/session.py Normal file
View File

@ -0,0 +1,94 @@
"""
TLSSession — one aioweb session, any tls backend
TLSSession subclasses aioweb.ExtendedSession and delegates only the four backend
seams to an injected backend object (see protocol.py). everything else — header
overwrites, domain rewriting, ephemeral headers, proxies, retries, previews — is
inherited from aioweb unchanged, because that logic operates on plain dicts and
never touches the backend.
from aioweb_tls import TLSSession, CurlCffi, Noble
async with TLSSession(backend=CurlCffi(impersonate="chrome")) as s:
resp = await s.request_with_retries("GET", "https://tls.peet.ws/api/all")
if resp:
print(resp.json()["tls"]["ja3"])
s = TLSSession(backend=Noble(client="chrome_133"))
await s.setup() # fetch noble's Go lib once
...
a custom backend (e.g. a local Go TLS server) injects the same way — implement the
TLSBackend protocol and pass it as backend=. one TLSSession is the only session
class; the backend swaps the wire, not the session.
"""
import logging
from aioweb import ExtendedSession
log = logging.getLogger(__name__)
class TLSSession(ExtendedSession):
"""aioweb session whose HTTP backend is an injected, swappable tls client"""
def __init__(self, backend, *args, **kwargs):
"""build a session over a tls backend
args:
backend: a TLSBackend (CurlCffi, Noble, or a custom one); owns its own
config vocabulary and the four seam behaviors
*args, **kwargs: forwarded to aioweb.ExtendedSession
"""
self.backend = backend
self._closed = False
super().__init__(*args, **kwargs)
async def setup(self) -> None:
"""run the backend's one-time setup if it has one (idempotent)
delegates to backend.setup() when defined (e.g. noble fetching its Go lib);
a no-op for backends without it. also invoked lazily before the first request
by backends that guard their own setup, so calling this is optional but lets
callers pre-warm at startup.
"""
setup = getattr(self.backend, "setup", None)
if setup is not None:
await setup()
# -------------------------------------------------------------------------
# the four backend seams, delegated to self.backend
def _create_session(self, headers, timeout, **kwargs):
"""delegate session creation to the backend"""
return self.backend.create_session(headers, timeout, **kwargs)
async def _raw_request(self, method, url, **kwargs):
"""delegate the wire request to the backend, returning its aioweb.Response"""
return await self.backend.raw_request(self.session, method, url, **kwargs)
def _cookies_for_url(self, url) -> dict:
"""delegate preview cookie lookup to the backend (default {})"""
cookies_for_url = getattr(self.backend, "cookies_for_url", None)
if cookies_for_url is None:
return {}
return cookies_for_url(self.session, url)
def _is_closed(self) -> bool:
"""closed if explicitly closed here or the backend reports it"""
if self._closed:
return True
return self.backend.is_closed(self.session)
# -------------------------------------------------------------------------
# lifecycle — backends close differently, so route through the backend
async def close(self) -> None:
"""close via the backend's close (falls back to session.close)"""
self._closed = True
close = getattr(self.backend, "close", None)
if close is not None:
await close(self.session)
else:
await self.session.close()