逆向沃尔玛反爬虫：绕过PerimeterX人机验证与10秒等待机制

1. 项目概述当爬虫遇上沃尔玛的“智能门卫”做爬虫的朋友尤其是搞电商数据采集的这两年估计没少被沃尔玛折腾。以前那种简单的requests加User-Agent伪装就能畅通无阻的日子早就一去不复返了。现在你打开沃尔玛的商品页面稍微频繁一点或者换个IP大概率会看到一个旋转的小圈圈然后弹出一个验证码或者干脆告诉你“访问被限制”。这背后就是沃尔玛升级了它的反爬虫体系其中最关键、也最让人头疼的一环就是基于_pxvid令牌的人机验证机制。这个_pxvid你可以把它理解成沃尔玛发给每个“访客”的临时身份证。没有这个身份证或者身份证是伪造的、过期的你就别想进门。更“狡猾”的是这个身份证的签发过程不是简单的服务器响应而是通过一段运行在你浏览器里的JavaScript代码来完成的。这段代码会收集你浏览器环境的大量“指纹”信息——比如你的Canvas画布渲染结果、WebGL支持情况、字体列表、屏幕分辨率、时区、语言等等——然后把这些信息加密、混淆最终生成一个长长的、看起来像乱码的字符串这就是_pxvid。服务器拿到这个令牌会进行解密和校验判断你是一个真实的人类用户还是一个程序模拟的爬虫。所以我们传统的爬虫直接用Python的requests库去发请求是拿不到这个有效令牌的因为requests不会执行JavaScript也无法模拟出浏览器那么复杂的指纹环境。这就是为什么你会卡在验证页面。网上有些教程会告诉你去硬解它的JS加密算法但那玩意儿混淆得亲妈都不认识而且经常变投入产出比极低。我们这次要走的是一条更“聪明”的路径不完全模拟浏览器而是利用规则找到那个可以绕过最复杂验证、直接获取有效令牌的“后门”并处理掉那个烦人的10秒等待。最终目标是构建一个稳定、高效的Python数据采集方案专门针对沃尔玛的公开页面。2. 核心思路拆解逆向工程与协议模拟面对_pxvid这种客户端生成的令牌我们的核心思路不是去“打败”它而是去“理解”并“加入”它。硬碰硬地破解JS加密是不现实的我们的策略是逆向分析整个令牌获取的HTTP请求流程找到那个最关键的、生成初始令牌或绕过验证的接口然后用Python精确地模拟这个请求。2.1 关键请求链分析首先你需要用Chrome或Edge浏览器的开发者工具F12开启无痕模式避免插件干扰清空Cookie然后访问一个沃尔玛的商品页比如https://www.walmart.com/ip/12345678。在Network网络面板中仔细筛选请求。你会发现一个规律首次访问你会收到一个状态码为202的响应。这个响应体里通常包含一段HTML里面有一个script标签其src指向一个PerimeterX一种常见的人机验证服务提供商的JS文件。同时响应头里会有一个Set-Cookie尝试设置一个名为_pxvid的Cookie但这个初始值通常是无效或待验证的。JS执行与挑战浏览器加载并执行那个PerimeterX的JS文件。这个脚本会收集指纹进行计算然后向一个特定的PX端点通常是/api/v2/collector或类似路径发起POST请求提交指纹数据。令牌获取与重定向如果指纹验证通过或者触发了某种降级策略PX服务会返回一个有效的_pxvid值。浏览器随后可能会带着这个有效的_pxvid重新请求原始页面或者服务器直接接受当前请求。最终访问此时你的请求头或Cookie里携带了有效的_pxvid服务器返回状态码200和正常的商品页面HTML。我们的突破口就在第2步和第3步之间。我们需要找到在什么条件下PX服务会“爽快”地给出一个有效令牌而不弹出图形验证码或进行更复杂的交互2.2 “10秒等待”的奥秘与绕过在分析请求时你可能会注意到有时PX的响应里会包含一个risk等级或者一个delay参数。那个著名的“10秒等待”很可能就是服务端对某些可疑但又不至于完全封杀的请求施加的一个冷却期。它希望你等10秒模拟人类阅读页面的时间然后再进行下一步。但是这个等待是客户端行为还是服务端强制经过大量测试发现这个“10秒”通常是由前端JS根据服务端返回的某个标志比如一个challenge对象里的delay值来计时并阻塞后续请求的。而服务端在发出这个“等待指令”时可能已经预生成或准备了一个有效的令牌只是暂时不给你。那么绕过思路就来了识别等待响应我们通过Python发送模拟请求如果收到包含特定字段如action: challenge,delay: 10000的响应我们就知道触发了等待。提取关键令牌仔细观察这个响应体。很多时候即使是在挑战响应中服务端已经返回了一个加密的_pxvid或者一个cid挑战ID它被包裹在复杂的JSON结构里。模拟等待后请求我们不需要真的在Python里傻等10秒。我们可以直接提取上一步中的关键令牌cid或初始_pxvid然后立即构造一个“挑战完成”的请求发送给PX的另一个端点例如/api/v2/collector带特定参数或一个/challenge完成接口。这个请求的核心是告诉服务器“嘿你刚才给我的那个挑战我已经完成了虽然我实际上没等”。获取清洁令牌如果模拟得当服务器会认为挑战已通过并返回一个干净的、可用于后续浏览的_pxvid。注意这个过程高度依赖对当前沃尔玛所用PerimeterX版本的具体行为分析。不同时期、不同站点的实现可能有细微差别。你必须基于自己抓包到的实时请求进行逆向不能直接套用过去的代码。2.3 工具选型为什么是httpx和rehttpxvsrequests我们选择httpx库。因为它不仅兼容requests的API更重要的是它支持HTTP/2。很多现代反爬系统包括PerimeterX的接口可能默认或优先使用HTTP/2协议requests库不支持。使用httpx能确保我们模拟的请求在协议层就和真实浏览器一致减少被识别的风险。此外httpx的异步客户端对于后续需要高并发的采集场景也更有优势。re(正则表达式)从复杂的JS响应或HTML中提取令牌、参数正则表达式是最直接快速的工具。虽然对于超复杂的HTML解析建议用parsel或bs4但针对反爬接口返回的特定JSON字符串或脚本变量正则往往更精准高效。3. 实操步骤从零构建采集器下面我将带你一步步实现这个绕过流程。请确保你已安装Python 3.8和httpx库pip install httpx。3.1 环境准备与请求会话建立我们首先要建立一个能保持Cookie和Headers的会话并模拟一个合理的浏览器环境。import httpx import re import time import json from urllib.parse import urljoin class WalmartScraper: def __init__(self): # 使用 httpx 客户端启用 http2 并设置合理超时 self.client httpx.Client( http2True, 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: text/html,application/xhtmlxml,application/xml;q0.9,image/avif,image/webp,image/apng,*/*;q0.8, Accept-Language: en-US,en;q0.9, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, Upgrade-Insecure-Requests: 1, Sec-Fetch-Dest: document, Sec-Fetch-Mode: navigate, Sec-Fetch-Site: none, Sec-Fetch-User: ?1, }, timeout30.0, follow_redirectsTrue # 跟随重定向 ) self.base_url https://www.walmart.com # 用于存储最终获取到的有效 _pxvid self.pxvid_cookie None def _update_headers_from_response(self, response): 从响应中更新客户端headers例如获取新的Cookie if set-cookie in response.headers: # 这里简化处理实际应将cookie合并到client的cookies jar中 # httpx.Client会自动管理cookies这里我们主要关注_pxvid pass3.2 首次探测与挑战识别我们定义一个方法用于发起初始请求并判断是否遇到了人机验证挑战。def initial_request(self, product_id): 发起对商品页的初始请求并探测挑战 url f{self.base_url}/ip/{product_id} print(f[*] 正在访问: {url}) try: resp self.client.get(url) print(f[*] 初始状态码: {resp.status_code}) # 情况1: 直接成功 (可能性较低) if resp.status_code 200: print([] 幸运直接访问成功未触发验证。) # 检查是否有_pxvid cookie self._extract_pxvid_from_cookies() return resp.text # 情况2: 收到202或其它包含验证逻辑的响应 # 重点检查响应内容是否包含PerimeterX相关的JS或挑战信息 if resp.status_code 202 or perimeterx in resp.text.lower() or _px in resp.text: print([!] 检测到PerimeterX人机验证。) # 尝试从响应HTML中提取关键的挑战参数例如可能内嵌在JS变量里的appId, uuid, vid等 challenge_data self._parse_challenge_data(resp.text) if challenge_data: print(f[*] 解析到挑战参数: {challenge_data}) # 进入挑战处理流程 return self._handle_px_challenge(challenge_data, url) else: print([-] 未能从页面解析出挑战参数可能需要更复杂的JS模拟。) # 可以尝试备用方案如从Set-Cookie中获取初始_pxvid self._extract_pxvid_from_response(resp) # 即使不完整也返回文本供进一步分析 return resp.text # 情况3: 其他错误 else: print(f[-] 意外的状态码: {resp.status_code}) return resp.text except httpx.RequestError as e: print(f[-] 请求失败: {e}) return None def _parse_challenge_data(self, html): 从包含PX JS的HTML中提取关键参数。这是一个需要根据实际情况调整的正则表达式示例。 data {} # 示例寻找类似 window._pxAppId PXu....; 的变量 app_id_match re.search(rwindow\._pxAppId\s*\s*[\]([^\])[\], html) if app_id_match: data[app_id] app_id_match.group(1) # 示例寻找可能内嵌的配置对象 config_match re.search(rwindow\._pxConfiguration\s*\s*({.*?});, html, re.DOTALL) if config_match: try: config json.loads(config_match.group(1)) data.update(config) # 合并配置中的uuid, vid等 except json.JSONDecodeError: pass # 另一个关键查找首次收集器请求的URL或参数 # 可能出现在类似 px. sendActivity(/api/v2/collector, {...}) 的调用里 collector_match re.search(r[\](/api/v[12]/collector[^\]*)[\], html) if collector_match: data[collector_path] collector_match.group(1) return data if data else None def _extract_pxvid_from_response(self, response): 从响应头或HTML中提取_pxvid的初始值 # 1. 从Set-Cookie头提取 cookies response.headers.get_list(set-cookie) for cookie in cookies: if _pxvid in cookie: # 简单提取实际应使用http.cookies模块解析 match re.search(r_pxvid([^;]), cookie) if match: self.pxvid_cookie match.group(1) print(f[*] 从Set-Cookie获取初始_pxvid: {self.pxvid_cookie[:50]}...) return # 2. 如果Cookie没有有时会作为JS变量或隐藏字段出现在HTML中 # 这里省略具体解析逻辑取决于实际页面结构3.3 核心挑战处理与10秒等待绕过这是整个流程最核心的部分。我们需要模拟那个向/api/v2/collector发送的POST请求。def _handle_px_challenge(self, challenge_data, original_url): 处理PX挑战目标是获取有效的_pxvid print([*] 开始处理PX挑战...) # 步骤1: 构造收集器请求。参数需要根据抓包精确还原。 # 这里是一个示例结构你必须用开发者工具捕获真实请求来填充。 collector_url urljoin(self.base_url, challenge_data.get(collector_path, /api/v2/collector)) # 这些payload字段是PerimeterX收集的指纹数据极其复杂且混淆。 # 我们无法完全生成但关键发现是对于“10秒等待”这种挑战有时发送一个 # 包含特定cid挑战ID和action标记为challenge的简化payload就能触发“挑战完成”流程。 # 这个cid可能在上一步的响应HTML或初始Cookie里。 # 假设我们从初始响应或challenge_data中获取到了cid cid challenge_data.get(cid) or self._guess_cid_from_previous_response() if not cid: print([-] 无法获取挑战ID (cid)绕过失败。) return None # 这是一个模拟的、用于“完成挑战”的payload。**重要此处的具体格式和字段名必须根据你抓包的真实请求来定** payload { appId: challenge_data.get(app_id, PXxxxxxx), # PerimeterX应用ID uuid: challenge_data.get(uuid, ), vid: challenge_data.get(vid, ), cid: cid, # 核心挑战ID action: challenge, # 或可能是 c 等缩写 tag: risk, # 或其它tag data: { cs: [], # 客户端数据通常为空或固定值 ls: {}, # 本地存储可留空 # ... 其他可能需要的字段如 pxcts (客户端时间戳) }, # 一个关键技巧设置一个表示“挑战已解决”的标志 solved: True, # **注意这个字段不一定存在需验证** riskMode: none # 尝试设置风险模式为无 } headers_for_collector { Content-Type: application/json, Origin: self.base_url, Referer: original_url, # 继承主会话的User-Agent等headers **{k: v for k, v in self.client.headers.items() if k.lower() not in [content-type, origin, referer]} } print(f[*] 正在向收集器接口发送请求: {collector_url}) try: # **关键请求模拟挑战完成** collector_resp self.client.post(collector_url, jsonpayload, headersheaders_for_collector) print(f[*] 收集器响应状态码: {collector_resp.status_code}) # 解析响应 if collector_resp.status_code 200: resp_json collector_resp.json() print(f[*] 收集器响应: {json.dumps(resp_json, indent2)[:500]}...) # 打印部分 # 成功的响应可能包含 action: “s“ (success) 和一个新的vid (即_pxvid) if resp_json.get(action) s and vid in resp_json: new_vid resp_json[vid] print(f[] 成功获取有效vid (即_pxvid): {new_vid[:50]}...) # 将这个vid设置为cookie self.client.cookies.set(_pxvid, new_vid, domain.walmart.com) self.pxvid_cookie new_vid # 步骤2: 带着新的_pxvid重新请求原始页面 print([*] 携带新_pxvid重新请求商品页...) final_resp self.client.get(original_url) if final_resp.status_code 200: print([] 成功绕过验证获取到商品页面) return final_resp.text else: print(f[-] 重新请求失败状态码: {final_resp.status_code}) return None # 处理可能返回的“等待”响应 elif resp_json.get(action) challenge and resp_json.get(delay): delay_ms resp_json.get(delay, 10000) print(f[!] 触发 {delay_ms/1000} 秒等待挑战。尝试绕过...) # **绕过关键我们不等直接尝试用响应里可能给的cid或vid进行下一步。** # 有时即使返回了delayvid也已经包含在响应里了。 bypass_vid resp_json.get(vid) if bypass_vid: print(f[*] 从等待响应中提取到vid: {bypass_vid[:50]}...) self.client.cookies.set(_pxvid, bypass_vid, domain.walmart.com) self.pxvid_cookie bypass_vid # 立即重试请求或者构造一个特定的“挑战完成”确认请求如果有对应接口 # 这里我们选择立即重试原页面 time.sleep(0.5) # 象征性短暂等待避免请求过快 final_resp self.client.get(original_url) if final_resp.status_code 200: print([] 成功绕过10秒等待获取页面) return final_resp.text else: print([-] 等待响应中未找到有效vid绕过失败。) return None else: print(f[-] 未预期的收集器响应动作: {resp_json.get(action)}) return None else: print(f[-] 收集器请求失败: {collector_resp.status_code}) return None except Exception as e: print(f[-] 处理挑战过程中发生异常: {e}) return None def _guess_cid_from_previous_response(self): 如果challenge_data里没有尝试从其他位置猜测cid例如初始的_pxvid cookie值本身可能就包含或等同于cid。 # 这是一个启发式方法。有时初始的_pxvid值无效的那个格式里就包含了cid信息。 # 或者需要从之前响应HTML的某个JS变量中提取。 # 此处返回None实际应用中需要你根据抓包分析补充逻辑。 return None3.4 获取数据与会话维持一旦我们成功获取了有效的_pxvid并存储在客户端的Cookie Jar中后续对同一域名下的请求在一定时间内通常就会畅通无阻。def scrape_product(self, product_id): 主爬取方法 # 先尝试直接获取如果触发验证则进入挑战流程 html self.initial_request(product_id) if html and Product Title in html: # 用一个简单的特征判断是否成功 # 成功获取页面开始解析数据... data self._parse_product_page(html) return data else: print([-] 未能成功获取商品页面。) return None def _parse_product_page(self, html): 解析商品页面提取所需数据示例 # 这里使用正则或解析库如parsel, beautifulsoup提取信息 data {} # 示例提取标题 title_match re.search(rh1[^]*itempropname[^]*(.*?)/h1, html, re.DOTALL) if title_match: data[title] re.sub(r[^], , title_match.group(1)).strip() # 示例提取价格 price_match re.search(rprice\s*:\s*([\d\.]), html) if price_match: data[price] price_match.group(1) # 提取其他信息... return data def close(self): 关闭客户端会话 self.client.close() # 使用示例 if __name__ __main__: scraper WalmartScraper() try: product_data scraper.scraper.scrape_product(12345678) # 替换为真实商品ID if product_data: print(f采集到的数据: {product_data}) else: print(采集失败。) finally: scraper.close()4. 关键技巧、注意事项与问题排查4.1 核心技巧与心得动态抓包是唯一真理我提供的代码是一个框架和思路。里面的URL路径、JSON payload结构、字段名称如appId,uuid,cid,action,tag,solved必须通过你自己在目标网站上的实时抓包来获取和确认。PerimeterX的版本和配置可能随时变化。“10秒等待”的绕过本质我们的策略不是“消除”等待而是“欺骗”服务端让它认为等待已经完成。核心在于找到那个能标识一次特定挑战的cid以及告诉服务端挑战已完成的正确接口和参数。有时在触发等待的响应里有效的_pxvid已经给出只是被前端逻辑延迟使用了。Cookie管理至关重要httpx.Client()会自动管理Cookie。确保你的所有请求都在同一个client会话内发起这样获得的_pxvid会自动被后续请求携带。手动设置Cookie时注意域.walmart.com。请求头模拟要细致Origin,Referer,Sec-Fetch-*等头信息在反爬严格的站点中很重要。尽量从浏览器复制完整的请求头。频率控制即使成功绕过也不要过于频繁地请求。模拟人类浏览间隔如3-10秒一个请求并合理使用代理IP池是长期稳定的基础。4.2 常见问题与排查表问题现象可能原因排查与解决思路初始请求一直返回202或验证页面1. 请求头不完整或特征明显。2. IP地址被标记。1. 核对并补全所有请求头特别是User-Agent和Sec-Fetch-*系列。2. 更换代理IP。尝试使用高质量住宅代理。_parse_challenge_data解析不到参数1. 页面结构或JS变量名已变化。2. 挑战逻辑被加载到异步请求中。1. 重新抓包搜索_px,appId,uuid,cid,collector等关键词更新正则表达式。2. 在Network面板查看是否有额外的XHR/Fetch请求在页面加载后发出那可能才是真正的挑战触发点。向/api/v2/collector发送请求后返回403或其它错误1. Payload结构或字段错误。2. 缺少必要的签名或动态参数。3. 请求被频率限制。1.最重要用浏览器精确抓取一次成功的“挑战完成”请求对比你的Python代码生成的Payload逐个字段检查。2. 检查是否有类似pxcts时间戳或一次性Token的字段需要生成。3. 增加请求间隔更换IP。成功获取vid后重新请求商品页仍失败1. 获取的vid并非最终有效的_pxvid。2. Cookie设置不正确域、路径。3. 需要额外的“验证完成”确认步骤。1. 检查获取vid的响应确认action是ssuccess。2. 确保Cookie通过client.cookies.set()正确设置或让httpx自动管理。3. 抓包观察浏览器在获取vid后是否还有另一个请求模拟它。程序运行几次后失效1. 单个IP请求过快触发风控。2. 使用的_pxvid过期或失效。3. PerimeterX策略更新。1. 必须加入随机延迟和代理IP轮换。2. 每次采集新会话时重新运行完整的挑战获取流程不要复用过期的令牌。3. 定期重新抓包分析反爬策略是动态的。4.3 关于代理IP的特别提醒对于沃尔玛这类网站使用代理IP几乎是必须的。但要注意代理质量首选住宅代理其次是高质量的数据中心代理。免费代理或透明代理基本无效。IP关联一个_pxvid令牌通常与请求它的IP地址绑定。如果你获取令牌后切换了IP令牌可能会失效。最佳实践是为每个采集会话从获取令牌到完成一批页面抓取使用同一个代理IP。并发控制即使有多个代理IP也请控制并发度。过高的并发请求本身就是一个爬虫特征。这个方案的稳定性建立在对PerimeterX当前行为模式的准确逆向之上。它可能不会永久有效但其中“分析请求链、模拟关键接口、绕过客户端等待”的思路是应对类似高级反爬机制的通用方法论。你需要保持关注在方法失效时重新拿起开发者工具继续这场“猫鼠游戏”的下一回合。