diff --git a/src/api/admin.py b/src/api/admin.py index c8b9b80..5f94155 100644 --- a/src/api/admin.py +++ b/src/api/admin.py @@ -1632,304 +1632,11 @@ async def get_captcha_config(token: str = Depends(verify_admin_token)): @router.post("/api/captcha/score-test") async def test_captcha_score( - request: Optional[CaptchaScoreTestRequest] = None, - token: str = Depends(verify_admin_token) + _request: Optional[CaptchaScoreTestRequest] = None, + _token: str = Depends(verify_admin_token) ): - """使用当前打码方式获取 token,并提交到 antcpt 校验分数。""" - req = request or CaptchaScoreTestRequest() - website_url = (req.website_url or "https://antcpt.com/score_detector/").strip() - website_key = (req.website_key or "6LcR_okUAAAAAPYrPe-HK_0RULO1aZM15ENyM-Mf").strip() - action = (req.action or "homepage").strip() - verify_url = (req.verify_url or "https://antcpt.com/score_detector/verify.php").strip() - enterprise = bool(req.enterprise) - - started_at = time.time() - captcha_config = await db.get_captcha_config() - captcha_method = (captcha_config.captcha_method or config.captcha_method or "").strip().lower() - browser_proxy_enabled = bool(captcha_config.browser_proxy_enabled) - browser_proxy_url = captcha_config.browser_proxy_url or "" - - token_value: Optional[str] = None - fingerprint: Optional[Dict[str, Any]] = None - token_elapsed_ms = 0 - verify_elapsed_ms = 0 - verify_http_status = None - verify_result: Dict[str, Any] = {} - verify_headers: Dict[str, str] = {} - verify_proxy_used = False - verify_proxy_source = "none" - verify_proxy_url = "" - verify_impersonate = "chrome120" - page_verify_only = captcha_method in {"browser", "personal", "remote_browser"} - verify_mode = "browser_page" if page_verify_only else "server_post" - - try: - token_start = time.time() - if captcha_method == "browser": - from ..services.browser_captcha import BrowserCaptchaService - service = await BrowserCaptchaService.get_instance(db) - score_payload, browser_id = await service.get_custom_score( - website_url=website_url, - website_key=website_key, - verify_url=verify_url, - action=action, - enterprise=enterprise - ) - if isinstance(score_payload, dict): - token_value = score_payload.get("token") - verify_elapsed_ms = int(score_payload.get("verify_elapsed_ms") or 0) - verify_http_status = score_payload.get("verify_http_status") - verify_result = score_payload.get("verify_result") if isinstance(score_payload.get("verify_result"), dict) else {} - verify_mode = score_payload.get("verify_mode") or "browser_page" - score_token_elapsed = score_payload.get("token_elapsed_ms") - if isinstance(score_token_elapsed, (int, float)): - token_elapsed_ms = int(score_token_elapsed) - if token_value: - fingerprint = await service.get_fingerprint(browser_id) - verify_proxy_used = bool(browser_proxy_enabled and browser_proxy_url) - verify_proxy_source = "captcha_browser_proxy" if verify_proxy_used else "browser_direct" - verify_proxy_url = browser_proxy_url if verify_proxy_used else "" - elif captcha_method == "personal": - from ..services.browser_captcha_personal import BrowserCaptchaService - service = await BrowserCaptchaService.get_instance(db) - score_payload = await service.get_custom_score( - website_url=website_url, - website_key=website_key, - verify_url=verify_url, - action=action, - enterprise=enterprise - ) - if isinstance(score_payload, dict): - token_value = score_payload.get("token") - verify_elapsed_ms = int(score_payload.get("verify_elapsed_ms") or 0) - verify_http_status = score_payload.get("verify_http_status") - verify_result = score_payload.get("verify_result") if isinstance(score_payload.get("verify_result"), dict) else {} - verify_mode = score_payload.get("verify_mode") or "browser_page" - score_token_elapsed = score_payload.get("token_elapsed_ms") - if isinstance(score_token_elapsed, (int, float)): - token_elapsed_ms = int(score_token_elapsed) - if token_value: - fingerprint = service.get_last_fingerprint() - verify_proxy_used = bool(browser_proxy_enabled and browser_proxy_url) - verify_proxy_source = "captcha_browser_proxy" if verify_proxy_used else "browser_direct" - verify_proxy_url = browser_proxy_url if verify_proxy_used else "" - elif captcha_method == "remote_browser": - score_payload = await _score_test_with_remote_browser_service( - website_url=website_url, - website_key=website_key, - verify_url=verify_url, - action=action, - enterprise=enterprise, - ) - if isinstance(score_payload, dict): - if score_payload.get("success") is False: - raise RuntimeError(score_payload.get("message") or "远程打码分数测试失败") - token_value = score_payload.get("token") - verify_elapsed_ms = int(score_payload.get("verify_elapsed_ms") or 0) - verify_http_status = score_payload.get("verify_http_status") - verify_result = score_payload.get("verify_result") if isinstance(score_payload.get("verify_result"), dict) else {} - verify_mode = score_payload.get("verify_mode") or "remote_browser_page" - score_token_elapsed = score_payload.get("token_elapsed_ms") - if isinstance(score_token_elapsed, (int, float)): - token_elapsed_ms = int(score_token_elapsed) - fingerprint = score_payload.get("fingerprint") if isinstance(score_payload.get("fingerprint"), dict) else None - elif captcha_method in SUPPORTED_API_CAPTCHA_METHODS: - token_value = await _solve_recaptcha_with_api_service( - method=captcha_method, - website_url=website_url, - website_key=website_key, - action=action, - enterprise=enterprise - ) - else: - return { - "success": False, - "message": f"当前打码方式不支持分数测试: {captcha_method}", - "captcha_method": captcha_method, - "website_url": website_url, - "website_key": website_key, - "action": action, - "verify_url": verify_url, - "enterprise": enterprise, - "token_acquired": False, - "elapsed_ms": int((time.time() - started_at) * 1000) - } - if token_elapsed_ms <= 0: - token_elapsed_ms = int((time.time() - token_start) * 1000) - - # 远程有头打码的 custom-score 可能由页面内直接完成校验, - # 在部分实现里不会显式回传 token,本地按 verify_result 兜底判定。 - if captcha_method == "remote_browser" and not token_value and isinstance(verify_result, dict): - if verify_result.get("success") is True: - token_value = verify_result.get("token") or verify_result.get("gRecaptchaResponse") or "__verified_by_remote__" - - if not token_value: - return { - "success": False, - "message": "未获取到 reCAPTCHA token", - "captcha_method": captcha_method, - "website_url": website_url, - "website_key": website_key, - "action": action, - "verify_url": verify_url, - "enterprise": enterprise, - "token_acquired": False, - "token_elapsed_ms": token_elapsed_ms, - "browser_proxy_enabled": browser_proxy_enabled, - "browser_proxy_url": browser_proxy_url if browser_proxy_enabled else "", - "fingerprint": fingerprint, - "elapsed_ms": int((time.time() - started_at) * 1000) - } - - if verify_mode == "server_post" and not page_verify_only: - verify_start = time.time() - verify_headers = { - "accept": "application/json, text/javascript, */*; q=0.01", - "content-type": "application/json", - "origin": "https://antcpt.com", - "referer": website_url, - "x-requested-with": "XMLHttpRequest", - } - if isinstance(fingerprint, dict): - ua = (fingerprint.get("user_agent") or "").strip() - lang = (fingerprint.get("accept_language") or "").strip() - sec_ch_ua = (fingerprint.get("sec_ch_ua") or "").strip() - sec_ch_ua_mobile = (fingerprint.get("sec_ch_ua_mobile") or "").strip() - sec_ch_ua_platform = (fingerprint.get("sec_ch_ua_platform") or "").strip() - - if ua: - verify_headers["user-agent"] = ua - if lang: - verify_headers["accept-language"] = lang if "," in lang else f"{lang},zh;q=0.9" - if sec_ch_ua: - verify_headers["sec-ch-ua"] = sec_ch_ua - if sec_ch_ua_mobile: - verify_headers["sec-ch-ua-mobile"] = sec_ch_ua_mobile - if sec_ch_ua_platform: - verify_headers["sec-ch-ua-platform"] = sec_ch_ua_platform - - if verify_headers.get("user-agent"): - for header_name, header_value in _guess_client_hints_from_user_agent( - verify_headers.get("user-agent", "") - ).items(): - if header_value and not verify_headers.get(header_name): - verify_headers[header_name] = header_value - verify_impersonate = _guess_impersonate_from_user_agent(verify_headers.get("user-agent", "")) - - verify_proxies, verify_proxy_used, verify_proxy_source, verify_proxy_url = ( - await _resolve_score_test_verify_proxy( - captcha_method=captcha_method, - browser_proxy_enabled=browser_proxy_enabled, - browser_proxy_url=browser_proxy_url - ) - ) - - async with AsyncSession() as session: - verify_resp = await session.post( - verify_url, - json={"g-recaptcha-response": token_value}, - headers=verify_headers, - proxies=verify_proxies, - impersonate=verify_impersonate, - timeout=30 - ) - verify_elapsed_ms = int((time.time() - verify_start) * 1000) - verify_http_status = verify_resp.status_code - - try: - verify_result = verify_resp.json() - except Exception: - verify_result = {"raw": verify_resp.text} - else: - verify_headers = { - "origin": "https://antcpt.com", - "referer": website_url, - "x-requested-with": "XMLHttpRequest", - } - if isinstance(fingerprint, dict): - verify_headers.update({ - "user-agent": fingerprint.get("user_agent", ""), - "accept-language": fingerprint.get("accept_language", ""), - "sec-ch-ua": fingerprint.get("sec_ch_ua", ""), - "sec-ch-ua-mobile": fingerprint.get("sec_ch_ua_mobile", ""), - "sec-ch-ua-platform": fingerprint.get("sec_ch_ua_platform", ""), - }) - - verify_success = bool(verify_result.get("success")) if isinstance(verify_result, dict) else False - score_value = verify_result.get("score") if isinstance(verify_result, dict) else None - - return { - "success": verify_success, - "message": "分数校验成功" if verify_success else "分数校验未通过", - "captcha_method": captcha_method, - "website_url": website_url, - "website_key": website_key, - "action": action, - "verify_url": verify_url, - "enterprise": enterprise, - "token_acquired": True, - "token_preview": _mask_token(token_value), - "token_elapsed_ms": token_elapsed_ms, - "verify_elapsed_ms": verify_elapsed_ms, - "verify_http_status": verify_http_status, - "score": score_value, - "verify_result": verify_result, - "verify_request_meta": { - "mode": verify_mode, - "proxy_used": verify_proxy_used, - "user_agent": verify_headers.get("user-agent", ""), - "accept_language": verify_headers.get("accept-language", ""), - "sec_ch_ua": verify_headers.get("sec-ch-ua", ""), - "sec_ch_ua_mobile": verify_headers.get("sec-ch-ua-mobile", ""), - "sec_ch_ua_platform": verify_headers.get("sec-ch-ua-platform", ""), - "origin": verify_headers.get("origin", ""), - "referer": verify_headers.get("referer", ""), - "x_requested_with": verify_headers.get("x-requested-with", ""), - "proxy_source": verify_proxy_source, - "proxy_url": verify_proxy_url, - "impersonate": verify_impersonate, - }, - "browser_proxy_enabled": browser_proxy_enabled, - "browser_proxy_url": browser_proxy_url if browser_proxy_enabled else "", - "fingerprint": fingerprint, - "elapsed_ms": int((time.time() - started_at) * 1000) - } - except Exception as e: - return { - "success": False, - "message": f"分数测试失败: {str(e)}", - "captcha_method": captcha_method, - "website_url": website_url, - "website_key": website_key, - "action": action, - "verify_url": verify_url, - "enterprise": enterprise, - "token_acquired": bool(token_value), - "token_preview": _mask_token(token_value), - "token_elapsed_ms": token_elapsed_ms, - "verify_elapsed_ms": verify_elapsed_ms, - "verify_http_status": verify_http_status, - "verify_result": verify_result, - "verify_request_meta": { - "mode": verify_mode, - "proxy_used": verify_proxy_used, - "user_agent": verify_headers.get("user-agent", ""), - "accept_language": verify_headers.get("accept-language", ""), - "sec_ch_ua": verify_headers.get("sec-ch-ua", ""), - "sec_ch_ua_mobile": verify_headers.get("sec-ch-ua-mobile", ""), - "sec_ch_ua_platform": verify_headers.get("sec-ch-ua-platform", ""), - "origin": verify_headers.get("origin", ""), - "referer": verify_headers.get("referer", ""), - "x_requested_with": verify_headers.get("x-requested-with", ""), - "proxy_source": verify_proxy_source, - "proxy_url": verify_proxy_url, - "impersonate": verify_impersonate, - }, - "browser_proxy_enabled": browser_proxy_enabled, - "browser_proxy_url": browser_proxy_url if browser_proxy_enabled else "", - "fingerprint": fingerprint, - "elapsed_ms": int((time.time() - started_at) * 1000) - } + """分数测试已禁用。""" + raise HTTPException(status_code=403, detail="已禁用分数测试") # ========== Plugin Configuration Endpoints ========== diff --git a/static/manage.html b/static/manage.html index cfcf66d..35086e9 100644 --- a/static/manage.html +++ b/static/manage.html @@ -526,11 +526,9 @@ -
测试目标:https://antcpt.com/score_detector/