s0k0landClaude Sonnet 5
19dbc139a8
Скриншот капчи снимается в физических пикселях (CSS-размер × device_scale_factor). Тест сравнивал результат с жёстко заданным (120, 40) без учёта, что profile.py при каждой новой сессии берёт случайный правдоподобный DPR из датасета browserforge — тест падал, когда выпадало DPR != 1 (например, 180×60 при DPR=1.5). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
295 lines
12 KiB
Python
295 lines
12 KiB
Python
"""Сквозные тесты с настоящим Chrome и сетью: api и worker в одном процессе (tests/harness.py), без
|
||
RabbitMQ и Redis. Запускаются в Docker (нужны Chrome и Xvfb):
|
||
|
||
make test-browser
|
||
|
||
Весь стенд с очередью проверяет make smoke.
|
||
"""
|
||
import json
|
||
import re
|
||
import uuid
|
||
|
||
import pytest
|
||
|
||
from api import profile
|
||
from tests.harness import run_env
|
||
from worker import windows
|
||
|
||
pytestmark = pytest.mark.browser
|
||
|
||
CF_URL = "https://freelancehunt.com/projects"
|
||
CF_SELECTOR = 'a[href*="/project/"]'
|
||
|
||
# то, что обычно входит в хеш fingerprint-библиотек и видно странице
|
||
FP_JS = """async () => {
|
||
const c = document.createElement('canvas'); c.width = 240; c.height = 60;
|
||
const x = c.getContext('2d'); x.textBaseline = 'top'; x.font = '16px Arial';
|
||
x.fillStyle = '#f60'; x.fillRect(10, 10, 100, 30); x.fillStyle = '#069'; x.fillText('Robotex fp', 4, 20);
|
||
const buf = await crypto.subtle.digest('SHA-256', new TextEncoder().encode(c.toDataURL()));
|
||
const uad = await navigator.userAgentData.getHighEntropyValues(['fullVersionList']);
|
||
document.body.innerHTML = '<pre id="fp">' + JSON.stringify({
|
||
ua: navigator.userAgent,
|
||
ua_ch: uad.fullVersionList.map(b => b.brand + '/' + b.version.split('.')[0]).join(','),
|
||
screen: [screen.width, screen.height, devicePixelRatio],
|
||
canvas: Array.from(new Uint8Array(buf)).slice(0, 6).map(b => b.toString(16).padStart(2, '0')).join(''),
|
||
}) + '</pre>';
|
||
}"""
|
||
|
||
# пустой canvas не зашумлён, подменённые функции выглядят нативными
|
||
NOISE_JS = """() => {
|
||
const blank = document.createElement('canvas');
|
||
blank.width = 50; blank.height = 50;
|
||
const empty = document.createElement('canvas');
|
||
empty.width = 50; empty.height = 50;
|
||
const data = blank.getContext('2d').getImageData(0, 0, 50, 50).data;
|
||
return {
|
||
blank_same: blank.toDataURL() === empty.toDataURL(),
|
||
blank_zero: data.every(v => v === 0),
|
||
native: [HTMLCanvasElement.prototype.toDataURL, HTMLCanvasElement.prototype.toBlob,
|
||
CanvasRenderingContext2D.prototype.getImageData, Function.prototype.toString]
|
||
.every(f => /\\{ \\[native code\\] \\}$/.test(Function.prototype.toString.call(f)) && /native code/.test(String(f))),
|
||
lengths: [HTMLCanvasElement.prototype.toDataURL.length, CanvasRenderingContext2D.prototype.getImageData.length],
|
||
};
|
||
}"""
|
||
|
||
|
||
@pytest.fixture
|
||
def env(monkeypatch, tmp_path):
|
||
with run_env(monkeypatch, tmp_path) as harness:
|
||
yield harness
|
||
|
||
|
||
@pytest.fixture
|
||
def client(env):
|
||
return env.client
|
||
|
||
|
||
def solve(client, session_id=None, **body):
|
||
client.cookies.clear()
|
||
if session_id:
|
||
client.cookies.set("session_id", session_id)
|
||
resp = client.post("/v1/solve", json={"actions": [], **body})
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json() | {"session_id": resp.cookies["session_id"]}
|
||
|
||
|
||
def test_basic_page(client):
|
||
data = solve(client, url="https://example.com", screenshot=True, actions=[
|
||
{"type": "wait_for", "selector": "p", "timeout": 10},
|
||
{"type": "inner_html", "selector": "title"},
|
||
])
|
||
assert data["status"] == "ok", data
|
||
assert data["actions"][1]["result"] == "Example Domain"
|
||
assert len(data["screenshot"]) > 1000
|
||
|
||
|
||
def test_action_failed_on_real_page(client):
|
||
data = solve(client, url="https://example.com", actions=[{"type": "inner_html", "selector": "#missing"}])
|
||
assert data["code"] == "action_failed"
|
||
assert data["error"] == "inner_html: selector not found: #missing"
|
||
|
||
|
||
def test_cookies_replace_profile_cookies(client):
|
||
url = "https://httpbin.org/cookies"
|
||
read = [{"type": "inner_html", "selector": "pre"}]
|
||
cookie = lambda name: {"name": name, "value": "1", "url": "https://httpbin.org", "expires": 4102444800}
|
||
|
||
first = solve(client, url=url, cookies=[cookie("a")], actions=read)
|
||
sid = first["session_id"]
|
||
assert json.loads(first["actions"][0]["result"])["cookies"] == {"a": "1"}
|
||
|
||
# без cookies в запросе профильные сохраняются
|
||
second = solve(client, sid, url=url, actions=read)
|
||
assert json.loads(second["actions"][0]["result"])["cookies"] == {"a": "1"}
|
||
|
||
# переданные cookies заменяют профильные
|
||
third = solve(client, sid, url=url, cookies=[cookie("b")], actions=read)
|
||
assert json.loads(third["actions"][0]["result"])["cookies"] == {"b": "1"}
|
||
|
||
|
||
def test_fingerprint_applied_and_stable(client, monkeypatch):
|
||
async def page_fp(page):
|
||
# в основном мире страницы, как скрипты сайта (по умолчанию Patchright выполняет в изолированном)
|
||
await page.evaluate(FP_JS, isolated_context=False)
|
||
return json.loads(await page.inner_html("#fp"))
|
||
|
||
results = {}
|
||
original = windows.make_page_action
|
||
|
||
def capture(payload):
|
||
action = original(payload)
|
||
|
||
async def wrapped(page):
|
||
out = await action(page)
|
||
out["fp"] = await page_fp(page)
|
||
return out
|
||
return wrapped
|
||
|
||
monkeypatch.setattr(windows, "make_page_action", capture)
|
||
|
||
class Spy(windows.AsyncStealthySession):
|
||
async def fetch(self, url, page_action, **kwargs):
|
||
out = await super().fetch(url, page_action=page_action, **kwargs)
|
||
results.setdefault(self._user_data_dir, []).append(out["fp"])
|
||
return out
|
||
|
||
monkeypatch.setattr(windows, "AsyncStealthySession", Spy)
|
||
|
||
sids = [solve(client, url="https://example.com")["session_id"] for _ in range(4)]
|
||
solve(client, sids[0], url="https://example.com")
|
||
|
||
fps = {}
|
||
for d, measured in results.items():
|
||
saved = profile.load(windows.USER_DATA_ROOT / d.rsplit("/", 1)[-1])
|
||
fp = saved["fingerprint"]
|
||
for m in measured:
|
||
assert m["screen"] == [fp["screen"]["width"], fp["screen"]["height"], fp["device_scale_factor"]]
|
||
# UA родной: версия в строке UA совпадает с userAgentData
|
||
major = re.search(r"Google Chrome/(\d+)", m["ua_ch"]).group(1)
|
||
assert f"Chrome/{major}." in m["ua"]
|
||
assert all(m == measured[0] for m in measured), "fingerprint changed within a profile"
|
||
fps[d] = measured[0]
|
||
assert len(fps) == 4
|
||
# шум canvas (#89): у каждого профиля свой хеш canvas, значит и отпечаток целиком
|
||
assert len({f["canvas"] for f in fps.values()}) == 4
|
||
assert len({json.dumps(f, sort_keys=True) for f in fps.values()}) == 4
|
||
|
||
|
||
def test_canvas_noise_is_invisible(client, monkeypatch):
|
||
"""Шум не трогает пустой canvas, а подменённые функции неотличимы от нативных по toString и length."""
|
||
seen = {}
|
||
original = windows.make_page_action
|
||
|
||
def capture(payload):
|
||
action = original(payload)
|
||
|
||
async def wrapped(page):
|
||
seen.update(await page.evaluate(NOISE_JS, isolated_context=False))
|
||
return await action(page)
|
||
return wrapped
|
||
|
||
monkeypatch.setattr(windows, "make_page_action", capture)
|
||
assert solve(client, url="https://example.com")["status"] == "ok"
|
||
assert seen == {"blank_same": True, "blank_zero": True, "native": True, "lengths": [0, 4]}
|
||
|
||
|
||
def test_ua_in_response_is_native(client):
|
||
"""user_agent в ответе — родной UA браузера, тот, с которым получен cf_clearance."""
|
||
data = solve(client, url="https://example.com")
|
||
assert re.fullmatch(r"Mozilla/5\.0 \(X11; Linux x86_64\) .* Chrome/\d+\.0\.0\.0 Safari/537\.36", data["user_agent"])
|
||
assert "Headless" not in data["user_agent"]
|
||
|
||
|
||
def test_cloudflare(client):
|
||
data = solve(client, url=CF_URL, captcha_type="cloudflare", timeout=120, actions=[
|
||
{"type": "wait_for", "selector": CF_SELECTOR, "timeout": 60},
|
||
])
|
||
assert data["status"] == "ok", data.get("error")
|
||
assert "cf_clearance" in {c["name"] for c in data["cookies"]}
|
||
assert uuid.UUID(data["session_id"])
|
||
|
||
|
||
def test_window_held_between_tasks(env):
|
||
"""browser_idle_timeout: вторая задача без url выполняется в том же окне, на уже открытой странице."""
|
||
read = [{"type": "inner_html", "selector": "title"}]
|
||
client = env.client
|
||
|
||
first = client.post("/v2/tasks", json={"url": "https://example.com", "actions": read, "browser_idle_timeout": 1})
|
||
task = client.get(first.json()["url"]).json()
|
||
assert task["result"]["status"] == "ok", task
|
||
assert env.pool.used == 1
|
||
|
||
second = client.post("/v2/tasks", json={"actions": read, "browser_idle_timeout": 1}).json()
|
||
result = client.get(second["url"]).json()["result"]
|
||
assert result["status"] == "ok", result
|
||
assert result["actions"][0]["result"] == "Example Domain"
|
||
# окно не перезапускалось: страница не открывалась заново
|
||
assert result["duration_ms"] < task["result"]["duration_ms"]
|
||
|
||
assert client.delete("/v2/sessions/current").json() == {"closed": True}
|
||
assert env.pool.used == 0
|
||
|
||
|
||
def test_typing_like_a_human(client):
|
||
"""typing: клик по полю и посимвольный ввод, включая кириллицу; поле сначала очищается."""
|
||
data = solve(client, url="https://httpbin.org/forms/post", actions=[
|
||
{"type": "fill", "selector": "input[name=custname]", "value": "old value"},
|
||
{"type": "typing", "selector": "input[name=custname]", "value": "Иван Smith-2"},
|
||
{"type": "submit", "selector": "form button"},
|
||
{"type": "inner_text", "selector": "body"},
|
||
])
|
||
assert data["status"] == "ok", data
|
||
assert json.loads(data["actions"][-1]["result"])["form"]["custname"] == "Иван Smith-2"
|
||
assert data["actions"][1]["duration_ms"] > 12 * 50 # посимвольно, с паузами
|
||
|
||
|
||
|
||
# капча-картинка (#109): картинка рисуется на canvas при загрузке, поэтому каждый новый запрос src дал бы другой код
|
||
CAPTCHA_PAGE = b"""<!doctype html><body>
|
||
<form id="signup"><img class="captcha"><input name="captcha" autocomplete="off"><div id="sent"></div></form>
|
||
<script>
|
||
const c = document.createElement('canvas'); c.width = 120; c.height = 40;
|
||
const x = c.getContext('2d'); x.fillStyle = '#fff'; x.fillRect(0, 0, 120, 40);
|
||
x.fillStyle = '#000'; x.font = '24px monospace'; x.fillText('k' + Math.random().toString(36).slice(2, 6), 8, 28);
|
||
document.querySelector('img.captcha').src = c.toDataURL();
|
||
document.querySelector('input').addEventListener('input', e => document.getElementById('sent').textContent = e.target.value);
|
||
</script></body>"""
|
||
|
||
|
||
@pytest.fixture
|
||
def captcha_site():
|
||
"""Страница с капчей-картинкой на локальном HTTP-сервере."""
|
||
import threading
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
def do_GET(self):
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||
self.end_headers()
|
||
self.wfile.write(CAPTCHA_PAGE)
|
||
|
||
def log_message(self, *args):
|
||
pass
|
||
|
||
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||
yield f"http://127.0.0.1:{server.server_port}/signup"
|
||
server.shutdown()
|
||
|
||
|
||
def test_captcha_image(client, captcha_site, monkeypatch):
|
||
"""Снимок элемента (PNG размером с картинку) уходит в 2captcha, ответ набирается в поле."""
|
||
import base64
|
||
import struct
|
||
from twocaptcha import TwoCaptcha
|
||
|
||
# скриншот снимается в физических пикселях (CSS-размер × device_scale_factor); фиксируем DPR=1,
|
||
# иначе размер PNG зависит от случайного отпечатка browserforge (экран/DPR — #109)
|
||
monkeypatch.setattr(profile, "generate_fingerprint", lambda: {
|
||
"screen": {"width": profile.MIN_SCREEN[0], "height": profile.MIN_SCREEN[1]},
|
||
"device_scale_factor": 1,
|
||
"canvas_seed": 1,
|
||
})
|
||
|
||
sent = {}
|
||
|
||
def fake_solve(self, **kwargs):
|
||
sent.update(kwargs)
|
||
return {"captchaId": "1", "code": "k42z"}
|
||
|
||
monkeypatch.setattr(TwoCaptcha, "solve", fake_solve)
|
||
|
||
data = solve(client, url=captcha_site, captcha="KEY", actions=[
|
||
{"type": "captcha", "selector": "img.captcha", "value": "image", "target": "input[name=captcha]", "max_len": 6},
|
||
{"type": "inner_text", "selector": "#sent"},
|
||
])
|
||
assert data["status"] == "ok", data
|
||
assert [a.get("result") for a in data["actions"]] == ["k42z", "k42z"]
|
||
|
||
png = base64.b64decode(sent["body"])
|
||
assert png.startswith(b"\x89PNG")
|
||
width, height = struct.unpack(">II", png[16:24])
|
||
assert (width, height) == (120, 40)
|
||
assert sent["method"] == "base64" and sent["maxLen"] == 6
|