mirror of
https://github.com/TencentCloud/Octop.git
synced 2026-10-30 07:53:04 +08:00
Adds 极验行为验 v4 to the login captcha layer, following the provider plug-in seam (one dataclass + register call backend-side, one widget adapter frontend-side; the settings UI is data-driven so only locale keys were needed): - providers.py: _GeetestV4Provider — login token carries the frontend getValidate() result as JSON (lot_number / captcha_output / pass_token / gen_time); siteverify posts form-urlencoded to gcaptcha4.geetest.com/validate?captcha_id=<id> with sign_token = HMAC-SHA256(captcha_key, lot_number); only result == "success" passes (fail / status:error both reject). Aliases: geetest, geetest4, gt4. Fail-closed on transport errors, consistent with the other providers. - CaptchaField: geetest-v4 popup via gt4.js bind mode — initGeetest4, showCaptcha on ready, getValidate() serialized as the captcha_token; error/close resolve undefined so login re-prompts. - Settings UI: captcha_id / captcha_key fields via locale keys. Tests: provider unit tests (URL + form fields + HMAC against an independent computation, malformed/missing-field rejection, interpret matrix), ensure_captcha flow through a local siteverify double (form body + signature asserted; fail result rejects), popup adapter tests (validate payload JSON, error path), builtin-order + integration list assertions updated. Locale parity green.
246 lines
7.2 KiB
Python
246 lines
7.2 KiB
Python
"""CaptchaProvider registry — slugs, aliases, plug-in seam."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from octop.infra.auth.captcha import get_provider, list_providers, parse_slug, register
|
|
from octop.infra.errors import ErrorCode, OctopError
|
|
|
|
|
|
def test_parse_slug_rejects_unknown() -> None:
|
|
with pytest.raises(ValueError):
|
|
parse_slug("not-a-vendor")
|
|
|
|
|
|
def test_parse_slug_rejects_none() -> None:
|
|
with pytest.raises(ValueError):
|
|
parse_slug("none")
|
|
|
|
|
|
def test_parse_slug_accepts_slider() -> None:
|
|
assert parse_slug("slider") == "slider"
|
|
|
|
|
|
def test_parse_slug_trims_and_lowercases() -> None:
|
|
assert parse_slug(" Turnstile ") == "turnstile"
|
|
|
|
|
|
def test_parse_slug_aliases_recaptcha_v3() -> None:
|
|
assert parse_slug("recaptcha_v3") == "recaptcha-v3"
|
|
|
|
|
|
def test_parse_slug_aliases_tencent() -> None:
|
|
assert parse_slug("tcaptcha") == "tencent"
|
|
|
|
|
|
def test_parse_slug_aliases_geetest_v4() -> None:
|
|
assert parse_slug("geetest") == "geetest-v4"
|
|
assert parse_slug("geetest4") == "geetest-v4"
|
|
assert parse_slug("gt4") == "geetest-v4"
|
|
|
|
|
|
def test_parse_slug_recaptcha_stays_v2() -> None:
|
|
assert parse_slug("recaptcha") == "recaptcha"
|
|
|
|
|
|
def test_get_provider_slider_does_not_require_token() -> None:
|
|
provider = get_provider("slider")
|
|
assert provider is not None
|
|
assert provider.requires_token is False
|
|
|
|
|
|
def test_list_providers_is_builtin_registration_order() -> None:
|
|
assert list_providers() == [
|
|
"slider",
|
|
"tencent",
|
|
"turnstile",
|
|
"hcaptcha",
|
|
"recaptcha-v3",
|
|
"geetest-v4",
|
|
]
|
|
|
|
|
|
def test_unlisted_recaptcha_v2_still_resolvable() -> None:
|
|
assert "recaptcha" not in list_providers()
|
|
assert get_provider("recaptcha") is not None
|
|
assert parse_slug("recaptcha") == "recaptcha"
|
|
|
|
|
|
class _FakeStrong:
|
|
slug = "fake-strong"
|
|
requires_token = True
|
|
siteverify_url = "http://127.0.0.1/siteverify"
|
|
requires_score = False
|
|
aliases: tuple[str, ...] = ()
|
|
|
|
def interpret(self, body: dict[str, Any], *, min_score: float) -> None:
|
|
del body, min_score
|
|
|
|
|
|
def test_register_makes_test_double_reachable(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from octop.infra.auth.captcha import providers as captcha_providers
|
|
|
|
monkeypatch.setattr(captcha_providers, "_REGISTRY", dict(captcha_providers._REGISTRY))
|
|
register(_FakeStrong())
|
|
assert get_provider("fake-strong") is not None
|
|
assert parse_slug("fake-strong") == "fake-strong"
|
|
|
|
|
|
def _tencent() -> Any:
|
|
provider = get_provider("tencent")
|
|
assert provider is not None
|
|
return provider
|
|
|
|
|
|
def test_tencent_verify_call_builds_tc3_signed_post() -> None:
|
|
call = _tencent().verify_call(
|
|
site_key="195642000",
|
|
secret="app-secret",
|
|
token="tr03ticket:@rand",
|
|
client_ip="203.0.113.7",
|
|
cam_id="AKIDcam",
|
|
cam_key="camkey",
|
|
)
|
|
assert call.method == "POST"
|
|
assert call.url == "https://captcha.tencentcloudapi.com/"
|
|
assert call.headers is not None
|
|
assert call.headers["Authorization"].startswith("TC3-HMAC-SHA256 Credential=AKIDcam/")
|
|
assert call.headers["X-TC-Action"] == "DescribeCaptchaResult"
|
|
assert call.headers["X-TC-Version"] == "2019-07-22"
|
|
assert "Host" not in call.headers # httpx sets it from the URL
|
|
assert call.json_body == {
|
|
"CaptchaType": 9,
|
|
"Ticket": "tr03ticket",
|
|
"Randstr": "@rand",
|
|
"UserIp": "203.0.113.7",
|
|
"CaptchaAppId": 195642000,
|
|
"AppSecretKey": "app-secret",
|
|
}
|
|
|
|
|
|
def test_tencent_verify_call_requires_cam_keys() -> None:
|
|
with pytest.raises(OctopError) as exc:
|
|
_tencent().verify_call(
|
|
site_key="195642000",
|
|
secret="app-secret",
|
|
token="tr03ticket:@rand",
|
|
client_ip="127.0.0.1",
|
|
)
|
|
assert exc.value.code == ErrorCode.CAPTCHA_FAILED
|
|
|
|
|
|
def test_tencent_verify_call_rejects_non_integer_appid() -> None:
|
|
with pytest.raises(OctopError):
|
|
_tencent().verify_call(
|
|
site_key="not-an-int",
|
|
secret="app-secret",
|
|
token="t:@r",
|
|
client_ip="127.0.0.1",
|
|
cam_id="AKIDcam",
|
|
cam_key="camkey",
|
|
)
|
|
|
|
|
|
def test_tencent_interpret_accepts_code_one() -> None:
|
|
_tencent().interpret({"Response": {"CaptchaCode": 1, "CaptchaMsg": "OK"}}, min_score=0.5)
|
|
|
|
|
|
def test_tencent_interpret_rejects_malicious_evil_level() -> None:
|
|
with pytest.raises(OctopError):
|
|
_tencent().interpret(
|
|
{"Response": {"CaptchaCode": 1, "EvilLevel": 100}},
|
|
min_score=0.5,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("code", [7, 8, 9, 15, 16, 21, 100])
|
|
def test_tencent_interpret_rejects_failure_codes(code: int) -> None:
|
|
with pytest.raises(OctopError):
|
|
_tencent().interpret(
|
|
{"Response": {"CaptchaCode": code, "CaptchaMsg": "nope"}},
|
|
min_score=0.5,
|
|
)
|
|
|
|
|
|
def test_tencent_interpret_rejects_api_error_envelope() -> None:
|
|
with pytest.raises(OctopError):
|
|
_tencent().interpret(
|
|
{"Response": {"Error": {"Code": "UnauthorizedOperation", "Message": "nope"}}},
|
|
min_score=0.5,
|
|
)
|
|
|
|
|
|
def _geetest() -> Any:
|
|
provider = get_provider("geetest-v4")
|
|
assert provider is not None
|
|
return provider
|
|
|
|
|
|
def _gt_token(**overrides: str) -> str:
|
|
payload = {
|
|
"lot_number": "4dc3cfc2cdff448cad8d13107198d473",
|
|
"captcha_output": "validate-output",
|
|
"pass_token": "pass-token",
|
|
"gen_time": "2026-09-20T12:00:00",
|
|
}
|
|
payload.update(overrides)
|
|
import json
|
|
|
|
return json.dumps(payload)
|
|
|
|
|
|
def test_geetest_verify_call_signs_lot_number() -> None:
|
|
import hashlib
|
|
import hmac
|
|
|
|
call = _geetest().verify_call(
|
|
site_key="captcha-id-1",
|
|
secret="captcha-key-1",
|
|
token=_gt_token(),
|
|
client_ip="127.0.0.1",
|
|
)
|
|
assert call.method == "POST"
|
|
assert call.url == "https://gcaptcha4.geetest.com/validate?captcha_id=captcha-id-1"
|
|
expected = hmac.new(
|
|
b"captcha-key-1", b"4dc3cfc2cdff448cad8d13107198d473", hashlib.sha256
|
|
).hexdigest()
|
|
assert call.data == {
|
|
"lot_number": "4dc3cfc2cdff448cad8d13107198d473",
|
|
"captcha_output": "validate-output",
|
|
"pass_token": "pass-token",
|
|
"gen_time": "2026-09-20T12:00:00",
|
|
"sign_token": expected,
|
|
}
|
|
assert call.json_body is None # form-urlencoded per vendor contract
|
|
|
|
|
|
def test_geetest_verify_call_rejects_non_json_token() -> None:
|
|
with pytest.raises(OctopError) as exc:
|
|
_geetest().verify_call(site_key="id", secret="key", token="not-json", client_ip="")
|
|
assert exc.value.code == ErrorCode.CAPTCHA_FAILED
|
|
|
|
|
|
def test_geetest_verify_call_rejects_missing_fields() -> None:
|
|
with pytest.raises(OctopError):
|
|
_geetest().verify_call(
|
|
site_key="id",
|
|
secret="key",
|
|
token=_gt_token(pass_token=""),
|
|
client_ip="",
|
|
)
|
|
|
|
|
|
def test_geetest_interpret_result_success_only() -> None:
|
|
provider = _geetest()
|
|
provider.interpret({"result": "success", "reason": ""}, min_score=0.5)
|
|
for body in (
|
|
{"result": "fail", "reason": "pass_token expire"},
|
|
{"status": "error", "code": "-50005", "msg": "illegal gen_time"},
|
|
{},
|
|
):
|
|
with pytest.raises(OctopError):
|
|
provider.interpret(body, min_score=0.5)
|