From 651096ed7b205d5a84c5b6a212a2a7d4637d43db Mon Sep 17 00:00:00 2001 From: CaliBrain Date: Sun, 16 Aug 2026 12:08:45 -0400 Subject: [PATCH] fix(bypass): reuse external bypasser clearance instead of re-solving (#1223) Direct download was unusable behind an external bypasser (FlareSolverr / Byparr): every request paid a 403 plus a full solve, and a search that never ran was reported to the user as "No books found". Clearance was discarded on the external path. get_cf_cookies_for_domain and get_cf_user_agent_for_domain returned {} / None whenever USING_EXTERNAL_BYPASSER was set, and _fetch_via_bypasser read only solution.response - dropping solution.cookies and solution.userAgent, which FlareSolverr-compatible services do return. A solve therefore cleared the one request that paid for it and nothing else, and a file download - which the solver cannot proxy, being binary - presented no clearance at all. Diagnosed from a v1.3.9 debug bundle: ~35s in the bypasser per search, on every search. - Move the cookie jar out of internal_bypasser into bypass/cookie_store. internal_bypasser imports seleniumbase at module scope, which is the dependency an external-bypasser deployment is entitled not to have, so it cannot host a store the external path depends on. - Harvest solution.cookies and solution.userAgent after a successful solve. The existing filtering applies unchanged, so the per-check __ddg8_/__ddg9_/__ddg10_ trio is still dropped and the external path cannot reintroduce the ?check=1 loop fixed in ebb833a. The UA matters as much as the cookies: Cloudflare ties cf_clearance to the UA that solved the challenge. - Read cookie fields from either shape - CDP objects or JSON mappings. Both use the same field names, expires included. - Point http.py's getters and _purge_clearance at the shared store, so either bypasser fills and drains the same jar. - Give the Docker helper-subprocess handoff explicit export_store / import_store rather than reaching into module globals. An unsolved challenge was also indistinguishable from an empty result. _looks_like_aa_page() counted the challenge markers as "recognisably AA", so _fetch_search_table handed a DDoS-Guard interstitial back as a legitimate no-table response and the user was told their query found nothing when the search never ran. Split challenge detection out and raise SearchUnavailableError with the reason instead. The mirror is still not quarantined - every mirror shares the same protection, so it is not the mirror's fault. Verified: 2531 unit tests pass; ruff, basedpyright and vulture clean; e2e bypasser-external profile passes (5). Its mock FlareSolverr already returned cookies and userAgent from /v1 - the contract was there, shelfmark was not reading it. Refs #1220. Deliberately not "Fixes": this removes the re-solve and makes a failed solve legible, but if Byparr genuinely cannot clear AA's current DDoS-Guard, the reporter now gets that as an error rather than a silent "no books found". The download path may swallow interstitials the same way; not audited here. --- shelfmark/bypass/cookie_store.py | 260 ++++++++++++++++++ shelfmark/bypass/external_bypasser.py | 38 ++- shelfmark/bypass/internal_bypasser.py | 210 +------------- shelfmark/download/http.py | 35 ++- shelfmark/release_sources/direct_download.py | 29 +- tests/bypass/test_ddg_cookie_reuse.py | 13 +- tests/bypass/test_external_bypasser.py | 161 +++++++++++ tests/bypass/test_internal_bypasser.py | 14 +- .../test_search_parked_mirror.py | 13 +- .../download/test_http_bypasser_fallbacks.py | 154 +++++++++-- 10 files changed, 674 insertions(+), 253 deletions(-) create mode 100644 shelfmark/bypass/cookie_store.py diff --git a/shelfmark/bypass/cookie_store.py b/shelfmark/bypass/cookie_store.py new file mode 100644 index 0000000..4ea2692 --- /dev/null +++ b/shelfmark/bypass/cookie_store.py @@ -0,0 +1,260 @@ +"""Clearance cookies won by a bypass, shared by every bypasser implementation. + +Kept in its own module rather than inside a bypasser because both of them feed it and +both read from it. The internal bypasser cannot host it: it imports seleniumbase at +module scope, which is exactly the dependency an external-bypasser deployment is +entitled not to have installed. +""" + +import threading +import time +from collections.abc import Mapping +from typing import Any +from urllib.parse import urlparse + +from shelfmark.core.logger import setup_logger + +logger = setup_logger(__name__) + +# Cookie storage - shared with requests library for Cloudflare bypass +# Nested mapping of domain to cookie name to cookie metadata. +_cf_cookies: dict[str, dict] = {} +_cf_cookies_lock = threading.Lock() + +# User-Agent storage - Cloudflare ties cf_clearance to the UA that solved the challenge +_cf_user_agents: dict[str, str] = {} + +# Protection cookie names we care about (Cloudflare and DDoS-Guard) +CF_COOKIE_NAMES = {"cf_clearance", "__cf_bm", "cf_chl_2", "cf_chl_prog"} +DDG_COOKIE_NAMES = { + "__ddg1_", + "__ddg2_", + "__ddg5_", + "__ddg8_", + "__ddg9_", + "__ddg10_", + "__ddgid_", + "__ddgmark_", + "ddg_last_challenge", +} + +# DDoS-Guard cookies that describe *one* check rather than granting clearance, and so +# must never be replayed on a later request. Observed live on Anna's Archive: +# +# __ddg9_ the client IP address +# __ddg10_ the unix timestamp the check was issued +# __ddg8_ an opaque token issued with them, same ~40 minute expiry +# +# Clearance itself lives in __ddg1_/__ddg2_/__ddgid_ (roughly a year) and __ddg5_. +# Replaying the trio is actively harmful: once the timestamp ages out - or the egress +# IP changes, which happens routinely behind a VPN - the values no longer describe the +# caller, DDoS-Guard re-arms its check and answers every request with a ?check=1 +# redirect. That is the redirect loop, and it is self-inflicted. Dropping them simply +# lets DDoS-Guard issue a fresh set, exactly as it does for a browser. +DDG_EPHEMERAL_COOKIE_NAMES = { + "__ddg8_", + "__ddg9_", + "__ddg10_", + "ddg_last_challenge", +} + + +def _get_base_domain(domain: str) -> str: + """Extract base domain from hostname (e.g., 'www.example.com' -> 'example.com').""" + return ".".join(domain.split(".")[-2:]) if "." in domain else domain + + +def _get_full_cookie_domains() -> set[str]: + """Return mirror domains that need full-session cookie extraction.""" + from shelfmark.core.mirrors import get_zlib_cookie_domains + + return {_get_base_domain(domain) for domain in get_zlib_cookie_domains()} + + +def _should_extract_cookie(name: str, *, extract_all: bool) -> bool: + """Determine if a cookie should be extracted based on its name.""" + # Checked before extract_all: a per-check token is wrong to replay for every + # domain, including the full-session ones. + if name in DDG_EPHEMERAL_COOKIE_NAMES: + return False + if extract_all: + return True + is_cf = name in CF_COOKIE_NAMES or name.startswith("cf_") + is_ddg = name in DDG_COOKIE_NAMES or name.startswith("__ddg") + return is_cf or is_ddg + + +def _cookie_field(cookie: Any, name: str) -> Any: + """Read one field from a cookie in either shape we are handed. + + The internal bypasser extracts CDP cookie objects; an external bypasser returns + the same fields as JSON objects, so the difference is attribute versus key access. + """ + if isinstance(cookie, Mapping): + return cookie.get(name) + return getattr(cookie, name, None) + + +def _cookie_expiry(cookie: Any) -> float | None: + """A cookie's absolute expiry, or None when it is a session cookie. + + The two spellings are not interchangeable and both reach this store. CDP and + Playwright cookies carry `expires`; the WebDriver cookie object - what a + Selenium-based solver such as FlareSolverr returns - carries `expiry`. Reading + only one silently turns every cookie from the other into a never-expiring one, + which is exactly how dead clearance ends up replayed forever (see + get_cf_cookies_for_domain). + + The value is coerced rather than trusted: it arrives as JSON from a service we + do not control, and a string here used to raise straight out of the store. + """ + for field in ("expires", "expiry"): + raw = _cookie_field(cookie, field) + if raw is None: + continue + try: + expiry = float(raw) + except TypeError, ValueError: + logger.debug("Unreadable cookie expiry %r; treating as a session cookie", raw) + return None + # <= 0 is how both shapes spell "session cookie", not "expired in 1970". + return expiry if expiry > 0 else None + return None + + +def store_extracted_cookies( + *, + url: str, + cookies: list[Any], + user_agent: str | None = None, +) -> None: + """Store filtered bypass cookies (and optional UA) for a URL domain.""" + parsed = urlparse(url) + domain = parsed.hostname or "" + if not domain: + return + + base_domain = _get_base_domain(domain) + extract_all = base_domain in _get_full_cookie_domains() + + cookies_found: dict[str, dict[str, Any]] = {} + for cookie in cookies: + name = _cookie_field(cookie, "name") or "" + if not _should_extract_cookie(name, extract_all=extract_all): + continue + secure = _cookie_field(cookie, "secure") + cookies_found[name] = { + "value": _cookie_field(cookie, "value") or "", + "domain": _cookie_field(cookie, "domain") or domain, + "path": _cookie_field(cookie, "path") or "/", + "expiry": _cookie_expiry(cookie), + "secure": True if secure is None else bool(secure), + "httpOnly": True, + } + + if not cookies_found: + return + + with _cf_cookies_lock: + _cf_cookies[base_domain] = cookies_found + if user_agent: + _cf_user_agents[base_domain] = user_agent + logger.debug("Stored UA for %s: %s...", base_domain, str(user_agent)[:60]) + else: + logger.debug("No UA captured for %s", base_domain) + + cookie_type = "all" if extract_all else "protection" + logger.debug("Extracted %s %s cookies for %s", len(cookies_found), cookie_type, base_domain) + + +def _is_cookie_expired(cookie: dict[str, Any]) -> bool: + """Whether a stored cookie's expiry has passed. Session cookies never expire here.""" + expiry = cookie.get("expiry") + if expiry is None: + expiry = cookie.get("expires") + if not expiry or expiry <= 0: + return False + return time.time() > expiry + + +def get_cf_cookies_for_domain(domain: str) -> dict[str, str]: + """Get stored cookies for a domain. Returns empty dict if none available.""" + if not domain: + return {} + + base_domain = _get_base_domain(domain) + + with _cf_cookies_lock: + cookies = _cf_cookies.get(base_domain, {}) + if not cookies: + return {} + + cf_clearance = cookies.get("cf_clearance", {}) + if cf_clearance and _is_cookie_expired(cf_clearance): + logger.debug("CF cookies expired for %s", base_domain) + _cf_cookies.pop(base_domain, None) + return {} + + # Expiry applies to every cookie, not just Cloudflare's. DDoS-Guard domains + # have no cf_clearance, so the check above never fired for them and dead + # cookies were replayed indefinitely - the server answers those with a + # challenge, which is indistinguishable from having sent nothing at all. + live = {name: c for name, c in cookies.items() if not _is_cookie_expired(c)} + if len(live) != len(cookies): + expired = sorted(set(cookies) - set(live)) + logger.debug("Dropping expired cookies for %s: %s", base_domain, expired) + if live: + _cf_cookies[base_domain] = live + else: + _cf_cookies.pop(base_domain, None) + + return {name: c["value"] for name, c in live.items()} + + +def has_valid_cf_cookies(domain: str) -> bool: + """Check if we have valid Cloudflare cookies for a domain.""" + return bool(get_cf_cookies_for_domain(domain)) + + +def get_cf_user_agent_for_domain(domain: str) -> str | None: + """Get the User-Agent that was used during bypass for a domain.""" + if not domain: + return None + with _cf_cookies_lock: + return _cf_user_agents.get(_get_base_domain(domain)) + + +def export_store() -> tuple[dict[str, dict], dict[str, str]]: + """Snapshot the whole store, for handing to another process. + + The internal bypasser's Docker helper solves in a subprocess, so the clearance it + wins has to be serialized back to the parent or the solve is lost with the child. + """ + with _cf_cookies_lock: + return ( + {domain: dict(cookies) for domain, cookies in _cf_cookies.items()}, + dict(_cf_user_agents), + ) + + +def import_store(cookies: object, user_agents: object) -> None: + """Merge a snapshot produced by :func:`export_store` into this process's store.""" + with _cf_cookies_lock: + if isinstance(cookies, dict): + _cf_cookies.update(cookies) + if isinstance(user_agents, dict): + _cf_user_agents.update( + {str(domain): str(agent) for domain, agent in user_agents.items()} + ) + + +def clear_cf_cookies(domain: str | None = None) -> None: + """Clear stored Cloudflare cookies and User-Agent. If domain is None, clear all.""" + with _cf_cookies_lock: + if domain: + base_domain = _get_base_domain(domain) + _cf_cookies.pop(base_domain, None) + _cf_user_agents.pop(base_domain, None) + else: + _cf_cookies.clear() + _cf_user_agents.clear() diff --git a/shelfmark/bypass/external_bypasser.py b/shelfmark/bypass/external_bypasser.py index d3db4f9..3901874 100644 --- a/shelfmark/bypass/external_bypasser.py +++ b/shelfmark/bypass/external_bypasser.py @@ -2,17 +2,19 @@ import random import time -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any import requests from shelfmark.bypass import BypassCancelledError +from shelfmark.bypass.cookie_store import store_extracted_cookies from shelfmark.core.config import config from shelfmark.core.logger import setup_logger from shelfmark.core.utils import normalize_http_url from shelfmark.download.network import get_ssl_verify if TYPE_CHECKING: + from collections.abc import Mapping from threading import Event from shelfmark.download import network @@ -63,6 +65,31 @@ def max_duration_seconds() -> float: return MAX_RETRY * read_timeout + backoff_total +def _store_solution_clearance(target_url: str, solution: Mapping[str, Any]) -> None: + """Keep the clearance the solver won, so later requests do not re-solve. + + A solve is the expensive part of an external bypass - tens of seconds of real + browser - and FlareSolverr-compatible services hand back the cookies and the + User-Agent that earned it. Dropping them meant every single request paid a 403 + plus a full solve, and a file download (which the solver cannot proxy, being + binary) never presented clearance at all. + + The UA matters as much as the cookies: Cloudflare ties cf_clearance to the UA + that solved the challenge, so replaying the cookie under our own UA is rejected. + """ + cookies = solution.get("cookies") or [] + if not isinstance(cookies, list): + logger.debug("External bypasser returned no usable cookie list for '%s'", target_url) + return + + user_agent = solution.get("userAgent") + store_extracted_cookies( + url=target_url, + cookies=cookies, + user_agent=user_agent if isinstance(user_agent, str) else None, + ) + + def _fetch_via_bypasser(target_url: str) -> str | None: """Make a single request to the external bypasser service. Returns HTML or None.""" raw_bypasser_url = _coerce_config_str( @@ -116,6 +143,15 @@ def _fetch_via_bypasser(target_url: str) -> str | None: logger.warning("External bypasser returned empty response for '%s'", target_url) return None + try: + _store_solution_clearance(target_url, solution) + except AttributeError, KeyError, TypeError, ValueError: + # Storing clearance is an optimisation; the page is the product. The + # solution JSON comes from a service we do not control, so a surprise in + # its cookie shape must not discard HTML that already cost a ~30s solve + # and send the caller round for up to MAX_RETRY more of them. + logger.debug("Could not store bypass clearance for '%s'", target_url, exc_info=True) + except requests.exceptions.Timeout: logger.warning( "External bypasser timed out for '%s' (connect: %ss, read: %.0fs)", diff --git a/shelfmark/bypass/internal_bypasser.py b/shelfmark/bypass/internal_bypasser.py index 419602a..ed7f499 100644 --- a/shelfmark/bypass/internal_bypasser.py +++ b/shelfmark/bypass/internal_bypasser.py @@ -28,6 +28,14 @@ from seleniumbase import cdp_driver from seleniumbase.undetected.cdp_driver.connection import ProtocolException from shelfmark.bypass import BypassCancelledError +from shelfmark.bypass.cookie_store import ( + clear_cf_cookies, + export_store, + get_cf_cookies_for_domain, + get_cf_user_agent_for_domain, + import_store, + store_extracted_cookies, +) from shelfmark.bypass.fingerprint import get_screen_size from shelfmark.config import env from shelfmark.config.env import LOG_DIR @@ -231,120 +239,6 @@ class _CdpWorker: _CDP_WORKER = _CdpWorker() -# Cookie storage - shared with requests library for Cloudflare bypass -# Nested mapping of domain to cookie name to cookie metadata. -_cf_cookies: dict[str, dict] = {} -_cf_cookies_lock = threading.Lock() - -# User-Agent storage - Cloudflare ties cf_clearance to the UA that solved the challenge -_cf_user_agents: dict[str, str] = {} - -# Protection cookie names we care about (Cloudflare and DDoS-Guard) -CF_COOKIE_NAMES = {"cf_clearance", "__cf_bm", "cf_chl_2", "cf_chl_prog"} -DDG_COOKIE_NAMES = { - "__ddg1_", - "__ddg2_", - "__ddg5_", - "__ddg8_", - "__ddg9_", - "__ddg10_", - "__ddgid_", - "__ddgmark_", - "ddg_last_challenge", -} - -# DDoS-Guard cookies that describe *one* check rather than granting clearance, and so -# must never be replayed on a later request. Observed live on Anna's Archive: -# -# __ddg9_ the client IP address -# __ddg10_ the unix timestamp the check was issued -# __ddg8_ an opaque token issued with them, same ~40 minute expiry -# -# Clearance itself lives in __ddg1_/__ddg2_/__ddgid_ (roughly a year) and __ddg5_. -# Replaying the trio is actively harmful: once the timestamp ages out - or the egress -# IP changes, which happens routinely behind a VPN - the values no longer describe the -# caller, DDoS-Guard re-arms its check and answers every request with a ?check=1 -# redirect. That is the redirect loop, and it is self-inflicted. Dropping them simply -# lets DDoS-Guard issue a fresh set, exactly as it does for a browser. -DDG_EPHEMERAL_COOKIE_NAMES = { - "__ddg8_", - "__ddg9_", - "__ddg10_", - "ddg_last_challenge", -} - - -def _get_base_domain(domain: str) -> str: - """Extract base domain from hostname (e.g., 'www.example.com' -> 'example.com').""" - return ".".join(domain.split(".")[-2:]) if "." in domain else domain - - -def _get_full_cookie_domains() -> set[str]: - """Return mirror domains that need full-session cookie extraction.""" - from shelfmark.core.mirrors import get_zlib_cookie_domains - - return {_get_base_domain(domain) for domain in get_zlib_cookie_domains()} - - -def _should_extract_cookie(name: str, *, extract_all: bool) -> bool: - """Determine if a cookie should be extracted based on its name.""" - # Checked before extract_all: a per-check token is wrong to replay for every - # domain, including the full-session ones. - if name in DDG_EPHEMERAL_COOKIE_NAMES: - return False - if extract_all: - return True - is_cf = name in CF_COOKIE_NAMES or name.startswith("cf_") - is_ddg = name in DDG_COOKIE_NAMES or name.startswith("__ddg") - return is_cf or is_ddg - - -def _store_extracted_cookies( - *, - url: str, - cookies: list[Any], - user_agent: str | None = None, -) -> None: - """Store filtered bypass cookies (and optional UA) for a URL domain.""" - parsed = urlparse(url) - domain = parsed.hostname or "" - if not domain: - return - - base_domain = _get_base_domain(domain) - extract_all = base_domain in _get_full_cookie_domains() - - cookies_found: dict[str, dict[str, Any]] = {} - for cookie in cookies: - name = getattr(cookie, "name", "") or "" - if not _should_extract_cookie(name, extract_all=extract_all): - continue - expires = getattr(cookie, "expires", None) - if expires is not None and expires <= 0: - expires = None - cookies_found[name] = { - "value": getattr(cookie, "value", ""), - "domain": getattr(cookie, "domain", None) or domain, - "path": getattr(cookie, "path", None) or "/", - "expiry": expires, - "secure": bool(getattr(cookie, "secure", True)), - "httpOnly": True, - } - - if not cookies_found: - return - - with _cf_cookies_lock: - _cf_cookies[base_domain] = cookies_found - if user_agent: - _cf_user_agents[base_domain] = user_agent - logger.debug("Stored UA for %s: %s...", base_domain, str(user_agent)[:60]) - else: - logger.debug("No UA captured for %s", base_domain) - - cookie_type = "all" if extract_all else "protection" - logger.debug("Extracted %s %s cookies for %s", len(cookies_found), cookie_type, base_domain) - async def _extract_cookies_from_cdp(driver: Any, page: Any, url: str) -> None: """Extract cookies from a CDP browser after successful bypass.""" @@ -360,81 +254,12 @@ async def _extract_cookies_from_cdp(driver: Any, page: Any, url: str) -> None: except _CDP_OPERATION_ERRORS: user_agent = None - _store_extracted_cookies(url=url, cookies=all_cookies, user_agent=user_agent) + store_extracted_cookies(url=url, cookies=all_cookies, user_agent=user_agent) except _CDP_OPERATION_ERRORS as e: logger.debug("Failed to extract cookies: %s", e) -def _is_cookie_expired(cookie: dict[str, Any]) -> bool: - """Whether a stored cookie's expiry has passed. Session cookies never expire here.""" - expiry = cookie.get("expiry") - if expiry is None: - expiry = cookie.get("expires") - if not expiry or expiry <= 0: - return False - return time.time() > expiry - - -def get_cf_cookies_for_domain(domain: str) -> dict[str, str]: - """Get stored cookies for a domain. Returns empty dict if none available.""" - if not domain: - return {} - - base_domain = _get_base_domain(domain) - - with _cf_cookies_lock: - cookies = _cf_cookies.get(base_domain, {}) - if not cookies: - return {} - - cf_clearance = cookies.get("cf_clearance", {}) - if cf_clearance and _is_cookie_expired(cf_clearance): - logger.debug("CF cookies expired for %s", base_domain) - _cf_cookies.pop(base_domain, None) - return {} - - # Expiry applies to every cookie, not just Cloudflare's. DDoS-Guard domains - # have no cf_clearance, so the check above never fired for them and dead - # cookies were replayed indefinitely - the server answers those with a - # challenge, which is indistinguishable from having sent nothing at all. - live = {name: c for name, c in cookies.items() if not _is_cookie_expired(c)} - if len(live) != len(cookies): - expired = sorted(set(cookies) - set(live)) - logger.debug("Dropping expired cookies for %s: %s", base_domain, expired) - if live: - _cf_cookies[base_domain] = live - else: - _cf_cookies.pop(base_domain, None) - - return {name: c["value"] for name, c in live.items()} - - -def has_valid_cf_cookies(domain: str) -> bool: - """Check if we have valid Cloudflare cookies for a domain.""" - return bool(get_cf_cookies_for_domain(domain)) - - -def get_cf_user_agent_for_domain(domain: str) -> str | None: - """Get the User-Agent that was used during bypass for a domain.""" - if not domain: - return None - with _cf_cookies_lock: - return _cf_user_agents.get(_get_base_domain(domain)) - - -def clear_cf_cookies(domain: str | None = None) -> None: - """Clear stored Cloudflare cookies and User-Agent. If domain is None, clear all.""" - with _cf_cookies_lock: - if domain: - base_domain = _get_base_domain(domain) - _cf_cookies.pop(base_domain, None) - _cf_user_agents.pop(base_domain, None) - else: - _cf_cookies.clear() - _cf_user_agents.clear() - - def _cleanup_orphan_processes() -> int: """Kill orphan Chrome/Xvfb/ffmpeg processes. Only runs in Docker mode.""" if not env.DOCKERMODE: @@ -956,17 +781,7 @@ def _run_bypass_in_current_process(url: str, retry: int, cancel_flag: Event | No def _store_child_bypass_state(payload: dict[str, Any]) -> None: - cookies = payload.get("cookies") - if isinstance(cookies, dict): - with _cf_cookies_lock: - _cf_cookies.update(cookies) - - user_agents = payload.get("user_agents") - if isinstance(user_agents, dict): - with _cf_cookies_lock: - _cf_user_agents.update( - {str(domain): str(agent) for domain, agent in user_agents.items()} - ) + import_store(payload.get("cookies"), payload.get("user_agents")) def _prepare_child_browser_env(env_vars: dict[str, str]) -> dict[str, str]: @@ -1407,11 +1222,12 @@ def _run_child_process() -> int: try: html = get(url, retry=retry) + cookies, user_agents = export_store() payload = { "ok": True, "html": html, - "cookies": _cf_cookies, - "user_agents": _cf_user_agents, + "cookies": cookies, + "user_agents": user_agents, } result_path.write_text(json.dumps(payload), encoding="utf-8") except Exception as exc: # noqa: BLE001 - helper boundary must serialize failures. diff --git a/shelfmark/download/http.py b/shelfmark/download/http.py index 973aa60..e010e35 100644 --- a/shelfmark/download/http.py +++ b/shelfmark/download/http.py @@ -10,7 +10,7 @@ from urllib.parse import urljoin, urlparse import requests from tqdm import tqdm -from shelfmark.bypass import BypassCancelledError +from shelfmark.bypass import BypassCancelledError, cookie_store from shelfmark.core.config import config as app_config from shelfmark.core.logger import setup_logger from shelfmark.core.request_helpers import coerce_bool, normalize_positive_int @@ -145,19 +145,13 @@ def get_bypassed_page( def get_cf_cookies_for_domain(domain: str) -> dict[str, str]: - """Get CF cookies - only available with internal bypasser.""" - if _is_using_external_bypasser(): - logger.debug("External bypasser in use, CF cookies not available for %s", domain) - return {} - return _get_internal_bypasser().get_cf_cookies_for_domain(domain) + """Get the clearance cookies won by whichever bypasser solved this domain.""" + return cookie_store.get_cf_cookies_for_domain(domain) def get_cf_user_agent_for_domain(domain: str) -> str | None: - """Get CF user agent - only available with internal bypasser.""" - if _is_using_external_bypasser(): - logger.debug("External bypasser in use, CF user agent not available for %s", domain) - return None - return _get_internal_bypasser().get_cf_user_agent_for_domain(domain) + """Get the User-Agent that solved this domain's challenge, if one is stored.""" + return cookie_store.get_cf_user_agent_for_domain(domain) def _apply_cf_bypass(url: str, headers: dict) -> dict: @@ -374,15 +368,14 @@ def html_get_page( Called whenever the protection answered a request that *carried* cookies: being challenged while presenting them proves they no longer work, so keeping - them only guarantees the same rejection on every later request. Purging is - internal-bypasser only; with an external one get_cf_cookies_for_domain() - already returns {}. + them only guarantees the same rejection on every later request. Applies to + either bypasser, since both fill the same store. """ hostname = urlparse(target_url).hostname or "" - # An empty domain means "clear every host" to the bypasser, so skip the purge + # An empty domain means "clear every host" to the store, so skip the purge # rather than wipe clearance for sites that are working fine. - if hostname and not _is_using_external_bypasser(): - _get_internal_bypasser().clear_cf_cookies(hostname) + if hostname: + cookie_store.clear_cf_cookies(hostname) def _redirect_loop_handoff(bypass_url: str) -> str | tuple[str, str]: """Drop the host's stale clearance cookies, then bypass `bypass_url`. @@ -576,8 +569,12 @@ def html_get_page( # (another concurrent download may have completed bypass and extracted cookies) parsed = urlparse(current_url) fresh_cookies = get_cf_cookies_for_domain(parsed.hostname or "") - if fresh_cookies and not cookies: - # Cookies are now available - retry with cookies before using bypasser + if fresh_cookies and not cookies and attempt < retry_limit: + # Cookies are now available - retry with cookies before using bypasser. + # Guarded on there being a next attempt: `continue` on the last one + # ends the retry loop and abandons the request without ever offering + # the URL to the bypasser, and MAX_RETRY=1 is the supported setting. + # Same reasoning as the bypasser invocation below. logger.debug( "403 but cookies now available - retrying with cookies: %s", current_url, diff --git a/shelfmark/release_sources/direct_download.py b/shelfmark/release_sources/direct_download.py index 281586c..cd0537b 100644 --- a/shelfmark/release_sources/direct_download.py +++ b/shelfmark/release_sources/direct_download.py @@ -565,9 +565,15 @@ _CHALLENGE_MARKERS = ( def _looks_like_aa_page(html: str) -> bool: - """Whether ``html`` is recognisably Anna's Archive, or a challenge in front of it.""" + """Whether ``html`` is recognisably Anna's Archive itself.""" lowered = html.lower() - return any(marker in lowered for marker in (*_AA_PAGE_MARKERS, *_CHALLENGE_MARKERS)) + return any(marker in lowered for marker in _AA_PAGE_MARKERS) + + +def _looks_like_challenge_page(html: str) -> bool: + """Whether ``html`` is a protection interstitial rather than the site behind it.""" + lowered = html.lower() + return any(marker in lowered for marker in _CHALLENGE_MARKERS) def _fetch_search_table(url: str, selector: network.AAMirrorSelector) -> tuple[str, Tag | None]: @@ -596,9 +602,22 @@ def _fetch_search_table(url: str, selector: network.AAMirrorSelector) -> tuple[s if table is not None: msg = f"Expected results table tag, got {type(table).__name__}" raise TypeError(msg) - if "No files found." in html or _looks_like_aa_page(html): - # A real AA response - either genuinely empty, or a shape the caller - # should report as drift. Not the mirror's fault. + if "No files found." in html: + # A real, genuinely empty answer from a healthy mirror. + return html, None + if _looks_like_challenge_page(html): + # The bypass did not actually clear the protection - the interstitial is + # what came back. Rotating is pointless (every mirror shares the same + # protection) and reporting it as an empty result is worse: the user is + # told their query found nothing when the search never ran. + msg = ( + "Anna's Archive answered with an unsolved protection challenge. " + "Check that the bypasser is reachable and working." + ) + raise SearchUnavailableError(msg) + if _looks_like_aa_page(html): + # A real AA response in a shape the caller should report as drift. + # Not the mirror's fault. return html, None new_base, action = selector.next_mirror_or_rotate_dns( diff --git a/tests/bypass/test_ddg_cookie_reuse.py b/tests/bypass/test_ddg_cookie_reuse.py index 4b981ae..88cfee4 100644 --- a/tests/bypass/test_ddg_cookie_reuse.py +++ b/tests/bypass/test_ddg_cookie_reuse.py @@ -17,13 +17,14 @@ import time import pytest +import shelfmark.bypass.cookie_store as cs import shelfmark.bypass.internal_bypasser as ib @pytest.fixture(autouse=True) def _clean_cookie_store(monkeypatch): - monkeypatch.setattr(ib, "_cf_cookies", {}) - monkeypatch.setattr(ib, "_cf_user_agents", {}) + monkeypatch.setattr(cs, "_cf_cookies", {}) + monkeypatch.setattr(cs, "_cf_user_agents", {}) class _Cookie: @@ -39,7 +40,7 @@ class _Cookie: def _store(cookies, url="https://annas-archive.gl/search"): - ib._store_extracted_cookies(url=url, cookies=cookies, user_agent="UA/1.0") + cs.store_extracted_cookies(url=url, cookies=cookies, user_agent="UA/1.0") # --------------------------------------------------------------------------- # @@ -83,7 +84,7 @@ def test_cloudflare_cookies_are_unaffected(): def test_per_check_cookies_are_excluded_even_for_full_session_domains(monkeypatch): """extract_all exists for Z-Library sessions; it must not resurrect the trio.""" - monkeypatch.setattr(ib, "_get_full_cookie_domains", lambda: {"annas-archive.gl"}) + monkeypatch.setattr(cs, "_get_full_cookie_domains", lambda: {"annas-archive.gl"}) _store([_Cookie("sessionid", "s"), _Cookie("__ddg9_", "203.0.113.7")]) stored = ib.get_cf_cookies_for_domain("annas-archive.gl") @@ -111,7 +112,7 @@ def test_all_cookies_expired_returns_empty_so_caller_re_solves(): _store([_Cookie("__ddg1_", "dead", expires=past)]) assert ib.get_cf_cookies_for_domain("annas-archive.gl") == {} - assert ib.has_valid_cf_cookies("annas-archive.gl") is False + assert cs.has_valid_cf_cookies("annas-archive.gl") is False def test_unexpired_cookies_are_kept(): @@ -143,7 +144,7 @@ def test_expired_cookies_are_pruned_from_the_store(): ib.get_cf_cookies_for_domain("annas-archive.gl") - assert set(ib._cf_cookies["annas-archive.gl"]) == {"__ddg1_"} + assert set(cs._cf_cookies["annas-archive.gl"]) == {"__ddg1_"} def test_solve_that_yields_only_per_check_cookies_stores_nothing(): diff --git a/tests/bypass/test_external_bypasser.py b/tests/bypass/test_external_bypasser.py index 42e0c26..9b05a9c 100644 --- a/tests/bypass/test_external_bypasser.py +++ b/tests/bypass/test_external_bypasser.py @@ -1,5 +1,7 @@ """Tests for the external bypasser flow.""" +import pytest + class _FakeResponse: def __init__(self, payload: dict) -> None: @@ -55,6 +57,165 @@ def test_fetch_via_bypasser_posts_expected_payload_and_uses_ssl_verify(monkeypat ] +def _stub_solution(monkeypatch, external_bypasser, solution: dict) -> None: + """Answer one bypass with `solution`, with config and SSL stubbed out.""" + + def fake_get(key, default=""): + values = { + "EXT_BYPASSER_URL": "https://bypass.example", + "EXT_BYPASSER_PATH": "/v1", + "EXT_BYPASSER_TIMEOUT": 60000, + } + return values.get(key, default) + + monkeypatch.setattr(external_bypasser.config, "get", fake_get) + monkeypatch.setattr( + external_bypasser.requests, + "post", + lambda *_a, **_k: _FakeResponse({"status": "ok", "solution": solution}), + ) + monkeypatch.setattr(external_bypasser, "get_ssl_verify", lambda _url: False) + + +def test_solved_clearance_is_stored_for_reuse(monkeypatch): + """A solve costs tens of seconds of real browser; its clearance must be kept. + + Without this every request paid a 403 plus a full solve, and the file download - + which the solver cannot proxy - presented no clearance at all. + """ + import shelfmark.bypass.cookie_store as cookie_store + import shelfmark.bypass.external_bypasser as external_bypasser + + monkeypatch.setattr(cookie_store, "_cf_cookies", {}) + monkeypatch.setattr(cookie_store, "_cf_user_agents", {}) + _stub_solution( + monkeypatch, + external_bypasser, + { + "response": "ok", + "userAgent": "Mozilla/5.0 (solver)", + "cookies": [ + {"name": "__ddg1_", "value": "clearance", "domain": ".annas-archive.gl"}, + {"name": "__ddg2_", "value": "c2", "domain": ".annas-archive.gl"}, + # Per-check cookies: kept out of the store, same as the internal path. + {"name": "__ddg9_", "value": "203.0.113.7", "domain": ".annas-archive.gl"}, + ], + }, + ) + + external_bypasser._fetch_via_bypasser("https://annas-archive.gl/search?q=dune") + + assert cookie_store.get_cf_cookies_for_domain("annas-archive.gl") == { + "__ddg1_": "clearance", + "__ddg2_": "c2", + } + # Cloudflare ties clearance to the solving UA, so replaying one without the other fails. + assert cookie_store.get_cf_user_agent_for_domain("annas-archive.gl") == "Mozilla/5.0 (solver)" + + +@pytest.mark.parametrize( + ("field", "shape"), + [ + # Byparr drives Playwright/camoufox, whose cookies spell it "expires". + ("expires", "playwright"), + # FlareSolverr assigns driver.get_cookies() - the WebDriver cookie object, + # which spells it "expiry". Reading only "expires" made every FlareSolverr + # cookie immortal, so dead clearance was replayed forever. + ("expiry", "webdriver"), + ], +) +def test_expired_solution_cookie_is_not_replayed(monkeypatch, field, shape): + import time + + import shelfmark.bypass.cookie_store as cookie_store + import shelfmark.bypass.external_bypasser as external_bypasser + + monkeypatch.setattr(cookie_store, "_cf_cookies", {}) + monkeypatch.setattr(cookie_store, "_cf_user_agents", {}) + _stub_solution( + monkeypatch, + external_bypasser, + { + "response": "ok", + "cookies": [{"name": "__ddg1_", "value": "dead", field: int(time.time()) - 60}], + }, + ) + + external_bypasser._fetch_via_bypasser("https://annas-archive.gl/search?q=dune") + + assert cookie_store.get_cf_cookies_for_domain("annas-archive.gl") == {}, ( + f"a dead {shape} cookie was kept for replay" + ) + + +def test_solution_cookie_expiry_is_coerced_not_trusted(monkeypatch): + """The solver is not ours; a stringified expiry must be read, not raised on.""" + import time + + import shelfmark.bypass.cookie_store as cookie_store + import shelfmark.bypass.external_bypasser as external_bypasser + + monkeypatch.setattr(cookie_store, "_cf_cookies", {}) + monkeypatch.setattr(cookie_store, "_cf_user_agents", {}) + _stub_solution( + monkeypatch, + external_bypasser, + { + "response": "ok", + "cookies": [ + {"name": "__ddg1_", "value": "live", "expires": str(int(time.time()) + 3600)}, + {"name": "__ddg2_", "value": "dead", "expires": str(int(time.time()) - 60)}, + ], + }, + ) + + result = external_bypasser._fetch_via_bypasser("https://annas-archive.gl/search?q=dune") + + assert result == "ok" + assert cookie_store.get_cf_cookies_for_domain("annas-archive.gl") == {"__ddg1_": "live"} + + +def test_storing_clearance_can_never_discard_the_solved_page(monkeypatch): + """A solve costs ~30s; a surprise in the cookie shape must not throw it away. + + The store call sits inside the request try/except, whose handler returns None - + so without its own guard a raising store turned a good page into a failed fetch + and sent the caller round for up to MAX_RETRY more solves. + """ + import shelfmark.bypass.external_bypasser as external_bypasser + + _stub_solution( + monkeypatch, + external_bypasser, + {"response": "ok", "cookies": [{"name": "__ddg1_", "value": "v"}]}, + ) + + def boom(*_args, **_kwargs): + raise TypeError("unexpected cookie shape") + + monkeypatch.setattr(external_bypasser, "store_extracted_cookies", boom) + + assert ( + external_bypasser._fetch_via_bypasser("https://annas-archive.gl/search?q=dune") + == "ok" + ) + + +def test_solution_without_cookies_is_still_returned(monkeypatch): + """A solver that returns no cookie list must not break the page fetch.""" + import shelfmark.bypass.cookie_store as cookie_store + import shelfmark.bypass.external_bypasser as external_bypasser + + monkeypatch.setattr(cookie_store, "_cf_cookies", {}) + monkeypatch.setattr(cookie_store, "_cf_user_agents", {}) + _stub_solution(monkeypatch, external_bypasser, {"response": "ok"}) + + result = external_bypasser._fetch_via_bypasser("https://annas-archive.gl/search?q=dune") + + assert result == "ok" + assert cookie_store.get_cf_cookies_for_domain("annas-archive.gl") == {} + + def test_get_bypassed_page_retries_and_rotates_selector_between_attempts(monkeypatch): import shelfmark.bypass.external_bypasser as external_bypasser diff --git a/tests/bypass/test_internal_bypasser.py b/tests/bypass/test_internal_bypasser.py index 66b690a..1ec4c6b 100644 --- a/tests/bypass/test_internal_bypasser.py +++ b/tests/bypass/test_internal_bypasser.py @@ -113,7 +113,9 @@ def test_extract_cookies_from_cdp_keeps_full_session_cookies_for_configured_zlib async def evaluate(self, _expr): return "TestUA/1.0" - monkeypatch.setattr(internal_bypasser, "_get_full_cookie_domains", lambda: {"z-lib.fm"}) + from shelfmark.bypass import cookie_store + + monkeypatch.setattr(cookie_store, "_get_full_cookie_domains", lambda: {"z-lib.fm"}) internal_bypasser.clear_cf_cookies() asyncio.run( @@ -165,12 +167,14 @@ def test_extract_cookies_from_cdp_normalizes_session_expiry(): ) ) - stored = internal_bypasser._cf_cookies.get("example.com", {}) + from shelfmark.bypass import cookie_store + + stored = cookie_store._cf_cookies.get("example.com", {}) assert stored["cf_clearance"]["expiry"] is None assert internal_bypasser.get_cf_cookies_for_domain("example.com") == {"cf_clearance": "abc"} # Verify fallback to "expires" key for expiry checks - internal_bypasser._cf_cookies["example.com"]["cf_clearance"]["expires"] = int(time.time()) - 10 + cookie_store._cf_cookies["example.com"]["cf_clearance"]["expires"] = int(time.time()) - 10 assert internal_bypasser.get_cf_cookies_for_domain("example.com") == {} @@ -374,7 +378,9 @@ def test_try_with_cached_cookies_returns_none_on_request_exception(monkeypatch): import shelfmark.bypass.internal_bypasser as internal_bypasser internal_bypasser.clear_cf_cookies() - internal_bypasser._cf_cookies["example.com"] = { + from shelfmark.bypass import cookie_store + + cookie_store._cf_cookies["example.com"] = { "cf_clearance": { "value": "abc", "domain": "example.com", diff --git a/tests/direct_download/test_search_parked_mirror.py b/tests/direct_download/test_search_parked_mirror.py index d245644..674376b 100644 --- a/tests/direct_download/test_search_parked_mirror.py +++ b/tests/direct_download/test_search_parked_mirror.py @@ -5,6 +5,7 @@ parking page is indistinguishable from a broken search, so the mirror stays in rotation and every later search pays for it again. """ +import pytest from bs4 import Tag PARKED_PAGE = """annas-archive.li @@ -85,15 +86,19 @@ def test_genuinely_empty_aa_result_does_not_quarantine(monkeypatch): assert "No files found." in html -def test_challenge_page_does_not_quarantine(monkeypatch): - """A DDoS-Guard interstitial means the mirror is alive and holds our clearance.""" +def test_challenge_page_is_reported_not_passed_off_as_an_empty_result(monkeypatch): + """An unsolved interstitial means the search never ran. + + The mirror is alive and holds our clearance, so it must not be quarantined - but + returning it as "no table" made the caller tell the user their query found nothing. + """ dd, _calls = _patch_pages(monkeypatch, [DDOS_GUARD_PAGE]) selector = _Selector(["https://real.test", "https://other.test"]) - _html, table = dd._fetch_search_table("https://real.test/search?q=dune", selector) + with pytest.raises(dd.SearchUnavailableError, match="protection challenge"): + dd._fetch_search_table("https://real.test/search?q=dune", selector) assert selector.quarantined == [] - assert table is None def test_unreachable_mirror_raises_search_unavailable(monkeypatch): diff --git a/tests/download/test_http_bypasser_fallbacks.py b/tests/download/test_http_bypasser_fallbacks.py index a7d1640..ea00fd3 100644 --- a/tests/download/test_http_bypasser_fallbacks.py +++ b/tests/download/test_http_bypasser_fallbacks.py @@ -11,6 +11,134 @@ class _FakeResponse: self.url = url +def test_external_bypasser_clearance_is_presented_on_the_next_request(monkeypatch): + """Clearance is read from the shared store whichever bypasser filled it. + + Guards the regression where the external path returned {} unconditionally: every + request re-paid a 403 plus a full solve, and a download - which the solver cannot + proxy - presented no clearance at all. + """ + import shelfmark.bypass.cookie_store as cookie_store + import shelfmark.download.http as http + + monkeypatch.setattr(cookie_store, "_cf_cookies", {}) + monkeypatch.setattr(cookie_store, "_cf_user_agents", {}) + monkeypatch.setattr(http, "_is_cf_bypass_enabled", lambda: True) + monkeypatch.setattr(http, "_is_using_external_bypasser", lambda: True) + + cookie_store.store_extracted_cookies( + url="https://annas-archive.gl/search", + cookies=[{"name": "__ddg1_", "value": "clearance"}], + user_agent="Mozilla/5.0 (solver)", + ) + + headers: dict[str, str] = {} + cookies = http._apply_cf_bypass("https://annas-archive.gl/md5/abc", headers) + + assert cookies == {"__ddg1_": "clearance"} + assert headers["User-Agent"] == "Mozilla/5.0 (solver)" + + +def test_external_bypasser_solve_is_reused_instead_of_re_solved(monkeypatch): + """One solve should clear the following requests, not just the one that paid for it. + + A solve is tens of seconds of real browser, so re-running it per request is what + made direct download unusable behind an external bypasser. + """ + import shelfmark.bypass.cookie_store as cookie_store + import shelfmark.download.http as http + + monkeypatch.setattr(cookie_store, "_cf_cookies", {}) + monkeypatch.setattr(cookie_store, "_cf_user_agents", {}) + monkeypatch.setattr(http, "_is_cf_bypass_enabled", lambda: True) + monkeypatch.setattr(http, "_is_using_external_bypasser", lambda: True) + monkeypatch.setattr(http, "_bypass_grace_seconds", lambda: 100.0) + monkeypatch.setattr(http, "get_proxies", lambda _url: {}) + monkeypatch.setattr(http, "get_ssl_verify", lambda _url: True) + monkeypatch.setattr(http.network, "should_rotate_dns_for_url", lambda _url: False) + monkeypatch.setattr(http.time, "sleep", lambda _seconds: None) + + class _Cleared: + is_redirect = False + status_code = 200 + cookies: dict[str, str] = {} + text = "results
" + url = "https://annas-archive.gl/search?q=dune" + + def raise_for_status(self) -> None: + return None + + def gated_get(url: str, **kwargs): + if kwargs.get("cookies", {}).get("cf_clearance") != "token": + error = requests.exceptions.HTTPError("forbidden") + error.response = _FakeResponse(403, url=url) + raise error + return _Cleared() + + solves: list[str] = [] + + def fake_solve(url: str, *_args, **_kwargs): + solves.append(url) + cookie_store.store_extracted_cookies( + url=url, + cookies=[{"name": "cf_clearance", "value": "token"}], + user_agent="Mozilla/5.0 (solver)", + ) + return "results
" + + monkeypatch.setattr(http.requests, "get", gated_get) + monkeypatch.setattr(http, "get_bypassed_page", fake_solve) + + url = "https://annas-archive.gl/search?q=dune" + first = http.html_get_page(url, retry=2, allow_bypasser_fallback=True, success_delay=0) + second = http.html_get_page(url, retry=2, allow_bypasser_fallback=True, success_delay=0) + + assert first == "results
" + assert second == "results
" + # The second request rode the stored clearance instead of paying for another solve. + assert solves == [url] + + +def test_403_with_a_concurrently_won_clearance_still_reaches_the_bypasser(monkeypatch): + """The last attempt must hand off, not `continue` into the end of the loop. + + Another worker's solve can land between our request and its 403, which used to + send this branch back round the retry loop - but on the final attempt (and + MAX_RETRY=1 is the supported setting) `continue` just ends it, abandoning the + request without ever offering the URL to the bypasser. + """ + import shelfmark.download.http as http + + monkeypatch.setattr(http, "_is_cf_bypass_enabled", lambda: True) + monkeypatch.setattr(http, "_bypass_grace_seconds", lambda: 100.0) + # A concurrent solve has filled the store, but this request went out before it did. + monkeypatch.setattr(http, "get_cf_cookies_for_domain", lambda _hostname: {"__ddg1_": "fresh"}) + monkeypatch.setattr(http, "_apply_cf_bypass", lambda _url, _headers: {}) + monkeypatch.setattr(http, "get_proxies", lambda _url: {}) + monkeypatch.setattr(http, "get_ssl_verify", lambda _url: True) + monkeypatch.setattr(http.network, "should_rotate_dns_for_url", lambda _url: False) + monkeypatch.setattr(http.time, "sleep", lambda _seconds: None) + + def gated(url: str, **_kwargs): + error = requests.exceptions.HTTPError("forbidden") + error.response = _FakeResponse(403, url=url) + raise error + + bypassed: list[str] = [] + monkeypatch.setattr(http.requests, "get", gated) + monkeypatch.setattr( + http, + "get_bypassed_page", + lambda url, *_a, **_k: bypassed.append(url) or "results
", + ) + + url = "https://annas-archive.gl/search?q=dune" + html = http.html_get_page(url, retry=1, allow_bypasser_fallback=True, success_delay=0) + + assert html == "results
" + assert bypassed == [url] + + def test_html_get_page_ignores_status_callback_failure(monkeypatch): """A raising status_callback must not break the bypass it was reporting on.""" import shelfmark.download.http as http @@ -190,15 +318,13 @@ def test_redirect_loop_purges_stale_cookies_and_switches_to_bypasser(monkeypatch stale = {"__ddg8_": "stale"} cleared: list[str] = [] - class _FakeInternalBypasser: - @staticmethod - def clear_cf_cookies(domain: str) -> None: - cleared.append(domain) - stale.clear() + def fake_clear(domain: str) -> None: + cleared.append(domain) + stale.clear() monkeypatch.setattr(http, "_is_cf_bypass_enabled", lambda: True) monkeypatch.setattr(http, "_is_using_external_bypasser", lambda: False) - monkeypatch.setattr(http, "_get_internal_bypasser", lambda: _FakeInternalBypasser) + monkeypatch.setattr(http.cookie_store, "clear_cf_cookies", fake_clear) monkeypatch.setattr(http, "_bypass_grace_seconds", lambda: 100.0) monkeypatch.setattr(http, "_apply_cf_bypass", lambda _url, _headers: dict(stale)) monkeypatch.setattr(http, "get_proxies", lambda _url: {}) @@ -349,17 +475,11 @@ def test_html_get_page_redirect_loop_purges_cookies_and_bypasses(monkeypatch): cleared: list[str] = [] - class FakeInternalBypasser: - def clear_cf_cookies(self, domain: str) -> None: - cleared.append(domain) - - def get_cf_cookies_for_domain(self, _domain: str) -> dict[str, str]: - return {"__ddg2_": "stale"} - - def get_cf_user_agent_for_domain(self, _domain: str) -> str | None: - return None - - monkeypatch.setattr(http, "_get_internal_bypasser", lambda: FakeInternalBypasser()) + monkeypatch.setattr(http.cookie_store, "clear_cf_cookies", cleared.append) + monkeypatch.setattr( + http.cookie_store, "get_cf_cookies_for_domain", lambda _domain: {"__ddg2_": "stale"} + ) + monkeypatch.setattr(http.cookie_store, "get_cf_user_agent_for_domain", lambda _domain: None) monkeypatch.setattr(http, "get_bypassed_page", lambda *_args, **_kwargs: "SOLVED") class _FakeRedirect: