mirror of
https://github.com/ThePhaseless/Byparr.git
synced 2026-11-10 11:08:20 +00:00
ThePhaseless
406ad428c4
invisible-core 49.46+ looks up the egress IP and checks GitHub for a newer GeoIP database on every browser launch, so a network that blocks those hosts delayed every request by the lookup timeouts. Memoize both lookups (success for the process lifetime, failures retried after a backoff, proxied egress lookups uncached), warm them up in the FastAPI lifespan before the port opens, and gate get_browser() on the same warm-up. Drop the unused --init mode.
376 lines
13 KiB
Python
376 lines
13 KiB
Python
import base64
|
|
from http import HTTPStatus
|
|
from json import JSONDecodeError
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import httpx2
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from invisible_playwright.async_api import Error as PlaywrightError
|
|
from invisible_playwright.async_api import TimeoutError as PlaywrightTimeoutError
|
|
from starlette.testclient import TestClient
|
|
|
|
from main import app
|
|
from src.challenge import CF_INTERSTITIAL_INDICATORS_SELECTORS
|
|
from src.endpoints import read_item
|
|
from src.models import LinkRequest
|
|
from src.utils import BrowserDepClass, TimeoutTimer, remaining_ms
|
|
|
|
client = TestClient(app)
|
|
|
|
test_websites = [
|
|
"https://ext.to/",
|
|
# "https://www.ygg.re/",
|
|
"https://extratorrent.st/",
|
|
"https://speed.cd/login",
|
|
'https://www.yggtorrent.top/engine/search?do=search&order=desc&sort=publish_date&name="UNESCAPED"+"DOUBLEQUOTES"&category=2145',
|
|
"https://1337x.to/home/",
|
|
"http://1337x.to/home/",
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("website", test_websites)
|
|
def test_bypass(website: str):
|
|
"""
|
|
Tests if the service can bypass cloudflare/DDOS-GUARD on given websites.
|
|
|
|
This test is skipped if the website is not reachable or does not have cloudflare/DDOS-GUARD.
|
|
"""
|
|
test_request = httpx2.get(
|
|
website,
|
|
)
|
|
if (
|
|
test_request.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR
|
|
and "Just a moment..." not in test_request.text
|
|
):
|
|
try:
|
|
error_details = test_request.json()
|
|
except JSONDecodeError:
|
|
error_details = test_request.text
|
|
pytest.skip(
|
|
f"Skipping {website} - ({test_request.status_code}) {error_details}"
|
|
)
|
|
|
|
response = client.post(
|
|
"/v1",
|
|
json=LinkRequest.model_construct(
|
|
url=website, cmd="request.get", max_timeout=60
|
|
).model_dump(),
|
|
)
|
|
|
|
assert response.status_code == HTTPStatus.OK
|
|
solution = response.json()["solution"]
|
|
assert "_cf_chl_opt" not in solution["response"]
|
|
assert "__cf_chl" not in solution["url"]
|
|
|
|
|
|
def test_json_api():
|
|
"""
|
|
JSON APIs must return 200, not crash on the UA evaluate.
|
|
|
|
Firefox renders application/json in a built-in viewer whose CSP blocks
|
|
Playwright's eval-based evaluate() (issue #394). The browser must be
|
|
launched with the viewer disabled so /v1 works and returns the raw JSON.
|
|
"""
|
|
url = "https://api.ipify.org?format=json"
|
|
test_request = httpx2.get(url)
|
|
if test_request.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR:
|
|
pytest.skip(
|
|
f"Skipping JSON API test - upstream error ({test_request.status_code})"
|
|
)
|
|
|
|
response = client.post(
|
|
"/v1",
|
|
json=LinkRequest.model_construct(url=url, cmd="request.get").model_dump(),
|
|
)
|
|
|
|
if response.status_code == HTTPStatus.REQUEST_TIMEOUT:
|
|
pytest.skip("Skipping JSON API test - timed out (upstream issue)")
|
|
|
|
assert response.status_code == HTTPStatus.OK
|
|
solution = response.json()["solution"]
|
|
assert solution["userAgent"]
|
|
assert '"ip"' in solution["response"]
|
|
|
|
|
|
def test_lifespan_warms_up_before_serving():
|
|
"""Uvicorn opens the port only after the GeoIP warm-up has finished."""
|
|
with patch("main.warm_up", new=AsyncMock()) as warm, TestClient(app) as local:
|
|
warm.assert_awaited_once()
|
|
assert (
|
|
local.get("/", follow_redirects=False).status_code
|
|
== HTTPStatus.MOVED_PERMANENTLY
|
|
)
|
|
|
|
|
|
def test_health_check():
|
|
"""
|
|
Tests the health check endpoint.
|
|
|
|
This test ensures that the health check
|
|
endpoint returns HTTPStatus.OK.
|
|
"""
|
|
response = client.get("/health")
|
|
assert response.status_code == HTTPStatus.OK
|
|
|
|
|
|
def test_pdf_handling():
|
|
"""Tests that PDF URLs return the raw PDF bytes, not the Firefox viewer HTML."""
|
|
pdf_url = "https://mondaymandala.com/wp-content/uploads/Mickey-And-Minnie-Mouse-Holding-An-Easter-Egg-Basket-Coloring-Page-For-Kids.pdf"
|
|
response = client.post(
|
|
"/v1",
|
|
json=LinkRequest.model_construct(url=pdf_url, cmd="request.get").model_dump(),
|
|
)
|
|
if response.status_code == HTTPStatus.REQUEST_TIMEOUT:
|
|
pytest.skip("Skipping PDF test - timed out (upstream issue)")
|
|
assert response.status_code == HTTPStatus.OK
|
|
solution = response.json()["solution"]
|
|
if solution.get("contentType") != "application/pdf":
|
|
pytest.skip(
|
|
"Skipping PDF test - PDF bytes could not be fetched (upstream issue)"
|
|
)
|
|
assert solution["response"] # non-empty base64
|
|
|
|
decoded = base64.b64decode(solution["response"])
|
|
assert decoded[:5] == b"%PDF-"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("payload", "expected"),
|
|
[
|
|
({"max_timeout": 60}, 60), # native API: seconds
|
|
({"maxTimeout": 60}, 60), # FlareSolverr alias, seconds-range value
|
|
({"maxTimeout": 60000}, 60), # FlareSolverr alias: milliseconds
|
|
({"maxTimeout": 55000}, 55),
|
|
({"maxTimeout": 1000}, 1),
|
|
({}, 60), # default
|
|
],
|
|
)
|
|
def test_max_timeout_normalization(payload: dict, expected: int):
|
|
"""MaxTimeout must accept FlareSolverr's milliseconds while keeping seconds."""
|
|
request = LinkRequest(url="https://example.com", **payload)
|
|
assert request.max_timeout == expected
|
|
|
|
|
|
def fake_dep(
|
|
*,
|
|
fail_states: set[str] | None = None,
|
|
challenged: bool = False,
|
|
marker_counts: list[int] | None = None,
|
|
widget_box: dict[str, float] | None = None,
|
|
widget_error: Exception | None = None,
|
|
user_agent: str | None = "UnitTestBrowser/1.0",
|
|
) -> BrowserDepClass:
|
|
"""Build a browser dependency pair backed by mocks."""
|
|
page = AsyncMock()
|
|
page.url = "https://example.test/login"
|
|
page.goto.return_value = MagicMock(
|
|
status=HTTPStatus.OK,
|
|
headers={"content-type": "text/html"},
|
|
request=MagicMock(headers={"user-agent": user_agent} if user_agent else {}),
|
|
)
|
|
page.title.return_value = "Login"
|
|
page.content.return_value = "<html><title>Login</title></html>"
|
|
remaining = list(marker_counts or [])
|
|
|
|
def count_for(selector: str) -> int:
|
|
"""Answer the marker check from the script, else from `challenged`."""
|
|
if selector not in CF_INTERSTITIAL_INDICATORS_SELECTORS or not remaining:
|
|
return 1 if challenged else 0
|
|
return remaining.pop(0) if len(remaining) > 1 else remaining[0]
|
|
|
|
def locator(selector: str) -> MagicMock:
|
|
handle = MagicMock()
|
|
handle.count = AsyncMock(side_effect=lambda: count_for(selector))
|
|
handle.first.bounding_box = AsyncMock(
|
|
return_value=widget_box, side_effect=widget_error
|
|
)
|
|
handle.first.input_value = AsyncMock(return_value="")
|
|
return handle
|
|
|
|
page.locator = MagicMock(side_effect=locator)
|
|
|
|
def wait_for_load_state(state: str, **_kwargs: object) -> None:
|
|
"""Fail the wait when asked for a configured state."""
|
|
if state in (fail_states or set()):
|
|
message = "load state wait timed out"
|
|
raise PlaywrightTimeoutError(message)
|
|
|
|
page.wait_for_load_state.side_effect = wait_for_load_state
|
|
|
|
context = AsyncMock()
|
|
context.cookies.return_value = []
|
|
return BrowserDepClass(page=page, context=context)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_networkidle_timeout_after_domcontentloaded_returns_content():
|
|
"""Pages that never go idle after DOM load must still return their content."""
|
|
dep = fake_dep(fail_states={"networkidle"})
|
|
response = await read_item(
|
|
LinkRequest(url="https://example.test/login"),
|
|
dep,
|
|
)
|
|
|
|
assert response.status == "ok"
|
|
assert response.solution.response == "<html><title>Login</title></html>"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_html_is_read_after_a_js_redirect_lands():
|
|
"""A non-Cloudflare guard that redirects during the idle wait must not leak (#418)."""
|
|
dep = fake_dep()
|
|
dep.page.content.return_value = "<html><title>Checking</title></html>"
|
|
load_state = dep.page.wait_for_load_state.side_effect
|
|
|
|
def redirect_while_idling(state: str, **kwargs: object) -> None:
|
|
load_state(state, **kwargs)
|
|
if state == "networkidle":
|
|
dep.page.content.return_value = "<html><title>Article</title></html>"
|
|
|
|
dep.page.wait_for_load_state.side_effect = redirect_while_idling
|
|
|
|
response = await read_item(LinkRequest(url="https://example.test/login"), dep)
|
|
|
|
assert response.solution.response == "<html><title>Article</title></html>"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_domcontentloaded_timeout_returns_408():
|
|
"""Fatal timeouts during initial page load still return a controlled 408."""
|
|
with pytest.raises(HTTPException) as exc:
|
|
await read_item(
|
|
LinkRequest(url="https://example.test/login"),
|
|
fake_dep(fail_states={"domcontentloaded"}),
|
|
)
|
|
|
|
assert exc.value.status_code == HTTPStatus.REQUEST_TIMEOUT
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unreachable_host_is_a_502_not_a_500():
|
|
"""An upstream we cannot reach is a gateway failure, never a Byparr crash."""
|
|
dep = fake_dep()
|
|
dep.page.goto.side_effect = PlaywrightError("Page.goto: NS_ERROR_UNKNOWN_HOST")
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await read_item(LinkRequest(url="https://nope.invalid/"), dep)
|
|
|
|
assert exc.value.status_code == HTTPStatus.BAD_GATEWAY
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_is_always_ok_like_flaresolverr():
|
|
"""FlareSolverr hardcodes 200 because Selenium cannot report the real code."""
|
|
dep = fake_dep()
|
|
dep.page.goto.return_value = MagicMock(
|
|
status=HTTPStatus.FORBIDDEN,
|
|
headers={"content-type": "text/html"},
|
|
request=MagicMock(headers={"user-agent": "UnitTestBrowser/1.0"}),
|
|
)
|
|
|
|
response = await read_item(LinkRequest(url="https://example.test/login"), dep)
|
|
|
|
assert response.solution.status == HTTPStatus.OK
|
|
|
|
|
|
def test_exhausted_budget_never_disables_playwright_timeouts():
|
|
"""Playwright reads timeout=0 as no timeout at all, so the floor must hold."""
|
|
spent = TimeoutTimer(duration=0)
|
|
|
|
assert spent.remaining() == 0
|
|
assert remaining_ms(spent) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_user_agent_header_is_not_a_500():
|
|
"""A request without a user-agent header degrades to empty, never a 500 (#394)."""
|
|
response = await read_item(
|
|
LinkRequest(url="https://example.test/login"),
|
|
fake_dep(user_agent=None),
|
|
)
|
|
|
|
assert response.status == "ok"
|
|
assert response.solution.user_agent == ""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_checkbox_is_clicked_while_the_challenge_is_up():
|
|
"""A measurable widget gets a humanised press, not a raw synthetic click."""
|
|
dep = fake_dep(
|
|
challenged=True,
|
|
marker_counts=[1, 1, 0],
|
|
widget_box={"x": 100.0, "y": 200.0, "width": 300.0, "height": 60.0},
|
|
)
|
|
|
|
response = await read_item(
|
|
LinkRequest(url="https://example.test/login", max_timeout=5), dep
|
|
)
|
|
|
|
assert response.status == "ok"
|
|
dep.page.mouse.move.assert_awaited_once_with(125.0, 230.0)
|
|
dep.page.mouse.down.assert_awaited_once()
|
|
dep.page.mouse.up.assert_awaited_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_challenge_that_clears_on_its_own_is_never_clicked():
|
|
"""A challenge is over when its markup goes, and until then we keep our hands off."""
|
|
dep = fake_dep(
|
|
challenged=True,
|
|
marker_counts=[1, 0],
|
|
widget_box={"x": 100.0, "y": 200.0, "width": 300.0, "height": 60.0},
|
|
)
|
|
|
|
response = await read_item(
|
|
LinkRequest(url="https://example.test/login", max_timeout=5), dep
|
|
)
|
|
|
|
assert response.status == "ok"
|
|
dep.page.mouse.down.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slow_widget_read_does_not_abort_the_solve():
|
|
"""One widget read outlasting its short timeout is retried, never a 500 (#419)."""
|
|
dep = fake_dep(
|
|
challenged=True,
|
|
marker_counts=[1, 1, 1, 0],
|
|
widget_error=PlaywrightTimeoutError(
|
|
"Locator.bounding_box: Timeout 1000ms exceeded."
|
|
),
|
|
)
|
|
|
|
response = await read_item(
|
|
LinkRequest(url="https://example.test/login", max_timeout=5), dep
|
|
)
|
|
|
|
assert response.status == "ok"
|
|
dep.page.mouse.down.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_challenge_that_never_clears_returns_408():
|
|
"""A challenge still up when the budget runs out is a timeout, not a 500."""
|
|
dep = fake_dep(challenged=True, marker_counts=[1])
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await read_item(
|
|
LinkRequest(url="https://example.test/login", max_timeout=2), dep
|
|
)
|
|
|
|
assert exc.value.status_code == HTTPStatus.REQUEST_TIMEOUT
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_marker_vanishing_mid_navigation_is_not_a_solved_challenge():
|
|
"""The marker drops out between challenge rounds; one clear read proves nothing."""
|
|
dep = fake_dep(challenged=True, marker_counts=[1, 0, 1])
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await read_item(
|
|
LinkRequest(url="https://example.test/login", max_timeout=2), dep
|
|
)
|
|
|
|
assert exc.value.status_code == HTTPStatus.REQUEST_TIMEOUT
|