From 417bd2c59b00615c24e5219c3369088cdd32068f Mon Sep 17 00:00:00 2001
From: Harold <8866033@gmail.com>
Date: Sun, 19 Apr 2026 20:11:58 +0800
Subject: [PATCH] feat: add cupfox firewall helpers
---
py/tests/test_茶杯狐.py | 44 +++++++++++++++++++++
py/茶杯狐.py | 85 ++++++++++++++++++++++++++++++++++++++++-
2 files changed, 128 insertions(+), 1 deletion(-)
diff --git a/py/tests/test_茶杯狐.py b/py/tests/test_茶杯狐.py
index b5521ec..b95e238 100644
--- a/py/tests/test_茶杯狐.py
+++ b/py/tests/test_茶杯狐.py
@@ -48,3 +48,47 @@ class TestCupfoxSpider(unittest.TestCase):
def test_firewall_encrypt_returns_base64_text(self, _mock_randint):
encoded = self.spider._cupfox_firewall_encrypt("PX")
self.assertEqual(base64.b64decode(encoded).decode("utf-8"), "PwXh7w")
+
+ def test_request_with_firewall_retries_after_robot_verification(self):
+ calls = []
+
+ def fake_request(url, method="GET", body=None, headers=None):
+ calls.append({"url": url, "method": method, "body": body, "headers": headers or {}})
+ if len(calls) == 1:
+ return {
+ "status_code": 200,
+ "text": '',
+ "headers": {"set-cookie": ["session=abc; Path=/"]},
+ }
+ if "robot.php" in url:
+ self.assertEqual(method, "POST")
+ self.assertIn("Cookie", headers)
+ self.assertIn("value=", body)
+ self.assertIn("token=", body)
+ return {
+ "status_code": 200,
+ "text": "ok",
+ "headers": {"set-cookie": ["shield=passed; Path=/"]},
+ }
+ return {
+ "status_code": 200,
+ "text": "ok",
+ "headers": {},
+ }
+
+ self.spider._request_text = fake_request
+ html = self.spider._request_with_firewall("https://www.cupfox.ai/search/test----------1---.html")
+
+ self.assertEqual(html, "ok")
+ self.assertEqual(len(calls), 3)
+ self.assertEqual(calls[2]["headers"]["Cookie"], "session=abc; shield=passed")
+
+ def test_extract_player_data_reads_embedded_json(self):
+ html = ''
+ data = self.spider._extract_player_data(html)
+ self.assertEqual(data["url"], "vid-1")
+ self.assertEqual(data["from"], "lineA")
+
+ def test_decode2_recovers_shifted_text(self):
+ encoded = "QXdCQ2tE"
+ self.assertEqual(self.spider._decode2(encoded), "P0")
diff --git a/py/茶杯狐.py b/py/茶杯狐.py
index beca930..da4831a 100644
--- a/py/茶杯狐.py
+++ b/py/茶杯狐.py
@@ -1,9 +1,10 @@
# coding=utf-8
import base64
+import json
import random
import re
import sys
-from urllib.parse import urljoin
+from urllib.parse import quote, urljoin
from base.spider import Spider as BaseSpider
@@ -80,3 +81,85 @@ class Spider(BaseSpider):
+ self.firewall_chars[random.randint(0, 61)]
)
return base64.b64encode(encoded.encode("utf-8")).decode("utf-8")
+
+ def _request_text(self, url, method="GET", body=None, headers=None):
+ request_headers = dict(self.headers)
+ if headers:
+ request_headers.update(headers)
+ if method == "POST":
+ response = self.post(url, data=body, headers=request_headers, timeout=15)
+ else:
+ response = self.fetch(url, headers=request_headers, timeout=15)
+ return {
+ "status_code": response.status_code,
+ "text": response.text or "",
+ "headers": dict(response.headers or {}),
+ }
+
+ def _request_with_firewall(self, url):
+ cookie_jar = {}
+ first = self._request_text(url)
+ self._merge_set_cookie(cookie_jar, first["headers"].get("set-cookie", []))
+ if not re.search(r"人机验证|verifyBox", first["text"] or ""):
+ if int(first["status_code"] or 0) != 200:
+ raise ValueError(f"HTTP {first['status_code']} @ {url}")
+ return first["text"]
+
+ token_raw = self._extract_firewall_token(first["text"])
+ if not token_raw:
+ return first["text"]
+
+ verify_body = (
+ "value="
+ + quote(self._cupfox_firewall_encrypt(url))
+ + "&token="
+ + quote(self._cupfox_firewall_encrypt(token_raw))
+ )
+ verify_headers = {
+ "Referer": url,
+ "Origin": self.host,
+ "Content-Type": "application/x-www-form-urlencoded",
+ }
+ cookie_text = self._cookie_header(cookie_jar)
+ if cookie_text:
+ verify_headers["Cookie"] = cookie_text
+ verify = self._request_text(
+ self.host + "/robot.php",
+ method="POST",
+ body=verify_body,
+ headers=verify_headers,
+ )
+ self._merge_set_cookie(cookie_jar, verify["headers"].get("set-cookie", []))
+
+ second_headers = {}
+ solved_cookie = self._cookie_header(cookie_jar)
+ if solved_cookie:
+ second_headers["Cookie"] = solved_cookie
+ second = self._request_text(url, headers=second_headers)
+ if int(second["status_code"] or 0) != 200:
+ raise ValueError(f"HTTP {second['status_code']} @ {url}")
+ return second["text"]
+
+ def _extract_player_data(self, html_text):
+ matched = re.search(r"player_aaaa\s*=\s*(\{[\s\S]*?\})\s*;?", str(html_text or ""))
+ if not matched:
+ return None
+ try:
+ return json.loads(matched.group(1))
+ except Exception:
+ return None
+
+ def _decode2(self, encoded):
+ if not encoded:
+ return ""
+ lookup = {}
+ for index, char in enumerate(self.firewall_chars):
+ lookup[char] = self.firewall_chars[(index + 59) % 62]
+ try:
+ raw = base64.b64decode(str(encoded).encode("utf-8")).decode("utf-8")
+ except Exception:
+ return ""
+ result = ""
+ for index in range(1, len(raw), 3):
+ result += lookup.get(raw[index], raw[index])
+ return result