feat(v2.4.0): JSVMP 双路径攻克体系 + jsdom 环境补丁知识库
基于实战经验重构 JSVMP 分析框架： - SKILL.md: 新增路径选择决策树（路径 A 算法追踪 vs 路径 B 环境伪装） - SKILL.md: 路径 B「环境伪装六步法」完整方法论（采集→对比→分级→补丁→验证→端到端） - SKILL.md: 新增场景 10「JSVMP + jsdom 环境伪装」、解法模式 E - SKILL.md: 标注三板斧适用边界与局限，经验法则扩充至 22 条 - references/jsdom-env-patches.md: 新增 jsdom 环境补丁知识库（Function.toString 三层防御 / plugins 完整模拟 / DOM 布局属性 / 30+ API 存根） - cases/: 新增经验库系统（模板 + 首个 JSVMP 实战案例含 58 项环境检测差异全表） - references/jsvmp-analysis.md: JSVMP 专项分析详细指南 - references/troubleshooting.md: 常见问题排查指南 - templates/python-request/: Python 项目模板 - 各 references 文档增强、Python 技术栈支持、MCP v1.1 适配（44→52 工具） Made-with: Cursor
This commit is contained in:
白夜
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
|
||||
"Referer": "https://example.com/"
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
{
|
||||
"sign_key": "your_sign_key_here",
|
||||
"encrypt_key": "1234567890abcdef",
|
||||
"iv": "abcdef1234567890"
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
"""
|
||||
主入口脚本 - 纯 Python 协议请求
|
||||
使用前请修改 CONFIG 中的配置和 utils/sign.py 中的签名逻辑
|
||||
"""
|
||||
import json
|
||||
import time
|
||||
from utils.sign import generate_sign, generate_m
|
||||
from utils.request import RequestClient
|
||||
|
||||
|
||||
CONFIG = {
|
||||
"base_url": "https://example.com",
|
||||
"api_path": "/api/data",
|
||||
"cookies": {
|
||||
"session_id": "your_session_id_here",
|
||||
},
|
||||
"total_pages": 5,
|
||||
"delay": 1.5,
|
||||
}
|
||||
|
||||
# 从配置文件加载 Headers 模板
|
||||
with open("config/headers.json", "r", encoding="utf-8") as f:
|
||||
HEADERS_TEMPLATE = json.load(f)
|
||||
|
||||
# 从配置文件加载密钥
|
||||
with open("config/keys.json", "r", encoding="utf-8") as f:
|
||||
KEYS = json.load(f)
|
||||
|
||||
|
||||
def fetch_page(client: RequestClient, page: int) -> dict:
|
||||
"""请求单页数据"""
|
||||
timestamp = str(int(time.time()))
|
||||
params = {
|
||||
"page": str(page),
|
||||
"ts": timestamp,
|
||||
}
|
||||
|
||||
# 计算签名（根据实际逻辑修改）
|
||||
sign = generate_sign(params, KEYS.get("sign_key", ""))
|
||||
params["sign"] = sign
|
||||
|
||||
# 如需加密参数 m
|
||||
# m = generate_m(params, KEYS.get("encrypt_key", ""), KEYS.get("iv", ""))
|
||||
# params["m"] = m
|
||||
|
||||
url = f"{CONFIG['base_url']}{CONFIG['api_path']}"
|
||||
response = client.get(url, params=params)
|
||||
return response.json()
|
||||
|
||||
|
||||
def extract_data(response_data: dict) -> list:
|
||||
"""从响应中提取目标数据（根据实际数据结构修改）"""
|
||||
return response_data.get("data", {}).get("list", [])
|
||||
|
||||
|
||||
def calculate_result(all_data: list):
|
||||
"""汇总计算结果（根据实际需求修改）"""
|
||||
print(f"\n采集完成，共 {len(all_data)} 条数据")
|
||||
for i, item in enumerate(all_data[:3]):
|
||||
print(f" [{i+1}] {item}")
|
||||
if len(all_data) > 3:
|
||||
print(f" ... 共 {len(all_data)} 条")
|
||||
|
||||
|
||||
def main():
|
||||
client = RequestClient(
|
||||
cookies=CONFIG["cookies"],
|
||||
headers=HEADERS_TEMPLATE,
|
||||
)
|
||||
|
||||
all_data = []
|
||||
|
||||
for page in range(1, CONFIG["total_pages"] + 1):
|
||||
try:
|
||||
print(f"正在请求第 {page} 页...")
|
||||
response_data = fetch_page(client, page)
|
||||
|
||||
# 打印关键中间值（调试用）
|
||||
print(f" 状态: {response_data.get('code', 'unknown')}")
|
||||
|
||||
page_data = extract_data(response_data)
|
||||
if not page_data:
|
||||
print(f" 第 {page} 页无数据，停止翻页")
|
||||
break
|
||||
|
||||
all_data.extend(page_data)
|
||||
print(f" 获取 {len(page_data)} 条数据")
|
||||
|
||||
if page < CONFIG["total_pages"]:
|
||||
time.sleep(CONFIG["delay"])
|
||||
|
||||
except Exception as e:
|
||||
print(f" 第 {page} 页请求失败: {e}")
|
||||
if hasattr(e, "response") and e.response is not None:
|
||||
print(f" 响应状态码: {e.response.status_code}")
|
||||
print(f" 响应内容: {e.response.text[:200]}")
|
||||
break
|
||||
|
||||
calculate_result(all_data)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,5 @@
|
||||
requests>=2.31.0
|
||||
pycryptodome>=3.20.0
|
||||
# execjs>=1.5.1
|
||||
# curl_cffi>=0.6.0
|
||||
# httpx[http2]>=0.27.0
|
||||
@@ -0,0 +1,121 @@
|
||||
"""
|
||||
HTTP 请求封装
|
||||
包含 Cookie 管理、重试机制、频率控制
|
||||
"""
|
||||
import time
|
||||
import random
|
||||
import requests
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class RequestClient:
|
||||
"""HTTP 客户端封装"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
cookies: Optional[dict] = None,
|
||||
headers: Optional[dict] = None,
|
||||
max_retries: int = 3,
|
||||
retry_delay: float = 2.0,
|
||||
):
|
||||
self.session = requests.Session()
|
||||
self.max_retries = max_retries
|
||||
self.retry_delay = retry_delay
|
||||
|
||||
default_headers = {
|
||||
"User-Agent": (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/120.0.0.0 Safari/537.36"
|
||||
),
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
|
||||
"Accept-Encoding": "gzip, deflate, br",
|
||||
"Connection": "keep-alive",
|
||||
}
|
||||
|
||||
if headers:
|
||||
default_headers.update(headers)
|
||||
self.session.headers.update(default_headers)
|
||||
|
||||
if cookies:
|
||||
for name, value in cookies.items():
|
||||
self.session.cookies.set(name, value)
|
||||
|
||||
def request(self, method: str, url: str, **kwargs) -> requests.Response:
|
||||
"""带重试的请求"""
|
||||
last_error = None
|
||||
|
||||
for attempt in range(1, self.max_retries + 1):
|
||||
try:
|
||||
response = self.session.request(method, url, **kwargs)
|
||||
|
||||
if response.status_code == 429:
|
||||
wait = self.retry_delay * (2 ** attempt) + random.uniform(0, 1)
|
||||
print(f" 触发频率限制，等待 {wait:.1f}s 后重试 ({attempt}/{self.max_retries})")
|
||||
time.sleep(wait)
|
||||
continue
|
||||
|
||||
if response.status_code in (403, 412):
|
||||
print(f" 请求被拒绝 ({response.status_code})，可能需要检查 Cookie 或签名")
|
||||
response.raise_for_status()
|
||||
|
||||
if response.status_code >= 500:
|
||||
wait = self.retry_delay * attempt + random.uniform(0, 1)
|
||||
print(f" 服务端错误 ({response.status_code})，等待 {wait:.1f}s 后重试 ({attempt}/{self.max_retries})")
|
||||
time.sleep(wait)
|
||||
continue
|
||||
|
||||
return response
|
||||
|
||||
except requests.RequestException as e:
|
||||
last_error = e
|
||||
if attempt < self.max_retries:
|
||||
wait = self.retry_delay * attempt
|
||||
print(f" 请求异常: {e}，等待 {wait:.1f}s 后重试 ({attempt}/{self.max_retries})")
|
||||
time.sleep(wait)
|
||||
|
||||
raise last_error or Exception("请求失败，已达最大重试次数")
|
||||
|
||||
def get(self, url: str, **kwargs) -> requests.Response:
|
||||
return self.request("GET", url, **kwargs)
|
||||
|
||||
def post(self, url: str, **kwargs) -> requests.Response:
|
||||
return self.request("POST", url, **kwargs)
|
||||
|
||||
def request_with_delay(
|
||||
self, method: str, url: str, delay: float = 1.0, **kwargs
|
||||
) -> requests.Response:
|
||||
"""带延迟的请求（避免触发频率限制）"""
|
||||
jitter = random.uniform(0, delay * 0.3)
|
||||
time.sleep(delay + jitter)
|
||||
return self.request(method, url, **kwargs)
|
||||
|
||||
def set_cookie(self, name: str, value: str, domain: str = ""):
|
||||
"""设置 Cookie"""
|
||||
self.session.cookies.set(name, value, domain=domain or None)
|
||||
|
||||
def update_cookies_from_response(self, response: requests.Response):
|
||||
"""从响应中更新 Cookie"""
|
||||
self.session.cookies.update(response.cookies)
|
||||
|
||||
def get_cookie_string(self) -> str:
|
||||
"""获取 Cookie 字符串"""
|
||||
return "; ".join(f"{c.name}={c.value}" for c in self.session.cookies)
|
||||
|
||||
|
||||
# ====== curl_cffi 版本（对抗 TLS 指纹检测时使用）======
|
||||
#
|
||||
# from curl_cffi import requests as curl_requests
|
||||
#
|
||||
# class CurlRequestClient:
|
||||
# """使用 curl_cffi 的 HTTP 客户端，支持浏览器 TLS 指纹模拟"""
|
||||
#
|
||||
# def __init__(self, impersonate: str = "chrome120"):
|
||||
# self.session = curl_requests.Session(impersonate=impersonate)
|
||||
#
|
||||
# def get(self, url: str, **kwargs):
|
||||
# return self.session.get(url, **kwargs)
|
||||
#
|
||||
# def post(self, url: str, **kwargs):
|
||||
# return self.session.post(url, **kwargs)
|
||||
@@ -0,0 +1,108 @@
|
||||
"""
|
||||
签名函数封装
|
||||
根据逆向分析结论修改具体的签名逻辑
|
||||
"""
|
||||
import hashlib
|
||||
import hmac
|
||||
import base64
|
||||
import json
|
||||
import time
|
||||
|
||||
# 如需执行提取的 JS 代码
|
||||
# import execjs
|
||||
|
||||
|
||||
def md5(text: str) -> str:
|
||||
"""MD5 哈希（返回 32 位小写 hex）"""
|
||||
return hashlib.md5(text.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def sha256(text: str) -> str:
|
||||
"""SHA-256 哈希（返回 64 位小写 hex）"""
|
||||
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def hmac_sha256(key: str, message: str) -> str:
|
||||
"""HMAC-SHA256 签名（返回 hex）"""
|
||||
return hmac.new(
|
||||
key.encode("utf-8"),
|
||||
message.encode("utf-8"),
|
||||
hashlib.sha256
|
||||
).hexdigest()
|
||||
|
||||
|
||||
def aes_encrypt(plaintext: str, key: str, iv: str) -> str:
|
||||
"""AES-CBC 加密（返回 Base64）"""
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Util.Padding import pad
|
||||
|
||||
cipher = AES.new(
|
||||
key.encode("utf-8"),
|
||||
AES.MODE_CBC,
|
||||
iv.encode("utf-8")
|
||||
)
|
||||
padded = pad(plaintext.encode("utf-8"), AES.block_size)
|
||||
encrypted = cipher.encrypt(padded)
|
||||
return base64.b64encode(encrypted).decode("utf-8")
|
||||
|
||||
|
||||
def aes_decrypt(ciphertext_b64: str, key: str, iv: str) -> str:
|
||||
"""AES-CBC 解密"""
|
||||
from Crypto.Cipher import AES
|
||||
from Crypto.Util.Padding import unpad
|
||||
|
||||
cipher = AES.new(
|
||||
key.encode("utf-8"),
|
||||
AES.MODE_CBC,
|
||||
iv.encode("utf-8")
|
||||
)
|
||||
decrypted = cipher.decrypt(base64.b64decode(ciphertext_b64))
|
||||
return unpad(decrypted, AES.block_size).decode("utf-8")
|
||||
|
||||
|
||||
def generate_sign(params: dict, secret_key: str) -> str:
|
||||
"""
|
||||
生成请求签名（根据实际逻辑修改）
|
||||
|
||||
示例签名逻辑：
|
||||
1. 参数按 key 字典序排序
|
||||
2. 拼接为 key1=value1&key2=value2 格式
|
||||
3. 末尾追加密钥
|
||||
4. 对拼接字符串做 MD5
|
||||
|
||||
Args:
|
||||
params: 请求参数字典
|
||||
secret_key: 签名密钥
|
||||
Returns:
|
||||
签名字符串
|
||||
"""
|
||||
sorted_params = sorted(params.items(), key=lambda x: x[0])
|
||||
param_str = "&".join(f"{k}={v}" for k, v in sorted_params)
|
||||
sign_str = f"{param_str}&key={secret_key}"
|
||||
|
||||
# 打印签名明文（调试用，正式使用时删除）
|
||||
print(f" 签名明文: {sign_str}")
|
||||
|
||||
result = md5(sign_str)
|
||||
print(f" 签名结果: {result}")
|
||||
return result
|
||||
|
||||
|
||||
def generate_m(params: dict, encrypt_key: str, iv: str) -> str:
|
||||
"""
|
||||
生成加密参数 m（根据实际逻辑修改）
|
||||
|
||||
示例：将参数 JSON 序列化后 AES 加密
|
||||
"""
|
||||
plaintext = json.dumps(params, separators=(",", ":"), sort_keys=True)
|
||||
return aes_encrypt(plaintext, encrypt_key, iv)
|
||||
|
||||
|
||||
# ====== execjs 方式（当 JS 逻辑过于复杂时使用）======
|
||||
#
|
||||
# def generate_sign_via_js(params: dict) -> str:
|
||||
# """使用 execjs 执行提取的 JS 代码生成签名"""
|
||||
# with open("config/sign_logic.js", "r", encoding="utf-8") as f:
|
||||
# js_code = f.read()
|
||||
# ctx = execjs.compile(js_code)
|
||||
# return ctx.call("generateSign", params)
|
||||
Reference in New Issue
Block a user