starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-09-24 18:40:29 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

fix(bypass): reuse external bypasser clearance instead of re-solving (#1223)

Browse Source 
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.
This commit is contained in:
CaliBrain
2026-08-16 12:08:45 -04:00
committed by GitHub
parent ebb833a82c
commit 651096ed7b
10 changed files with 674 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shelfmark/bypass/cookie_store.py
+260
View File
@@ -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()
shelfmark/bypass/external_bypasser.py
+37 -1
View File
@@ -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)",
shelfmark/bypass/internal_bypasser.py
+13 -197
View File
@@ -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.
shelfmark/download/http.py
+16 -19
View File
@@ -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,
shelfmark/release_sources/direct_download.py
+24 -5
View File
@@ -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(
tests/bypass/test_ddg_cookie_reuse.py
+7 -6
View File
@@ -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():
tests/bypass/test_external_bypasser.py
+161
View File
@@ -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": "<html>ok</html>",
"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": "<html>ok</html>",
"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": "<html>ok</html>",
"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 == "<html>ok</html>"
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": "<html>ok</html>", "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")
== "<html>ok</html>"
)
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": "<html>ok</html>"})
result = external_bypasser._fetch_via_bypasser("https://annas-archive.gl/search?q=dune")
assert result == "<html>ok</html>"
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
tests/bypass/test_internal_bypasser.py
+10 -4
View File
@@ -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",
tests/direct_download/test_search_parked_mirror.py
+9 -4
View File
@@ -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 = """<!doctype html><html><head><title>annas-archive.li</title></head>
@@ -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):
tests/download/test_http_bypasser_fallbacks.py
+137 -17
View File
@@ -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 = "<table>results</table>"
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 "<table>results</table>"
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 == "<table>results</table>"
assert second == "<table>results</table>"
# 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 "<table>results</table>",
)
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 == "<table>results</table>"
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: