1537 lines
62 KiB
Python
1537 lines
62 KiB
Python
"""
|
||
Stripe Checkout 自动化支付脚本
|
||
用法:
|
||
python pay.py <session_id> [--card N] [--config path] [--token TOKEN]
|
||
|
||
示例:
|
||
python pay.py cs_live_a12H3g13P9TH6udPmljRCpWsmHiKRFH7VUiZBbcA1U60eMzFFI2wp3rtXL
|
||
"""
|
||
|
||
import argparse
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import random
|
||
import re
|
||
import string
|
||
import sys
|
||
import time
|
||
import urllib.parse
|
||
import uuid
|
||
from datetime import datetime
|
||
|
||
import requests
|
||
|
||
|
||
LOG_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "log.txt")
|
||
|
||
def _init_log():
|
||
"""清空并初始化 log.txt"""
|
||
with open(LOG_FILE, "w", encoding="utf-8") as f:
|
||
f.write(f"{'='*80}\n")
|
||
f.write(f" Stripe 自动化支付 日志 — {datetime.now().isoformat()}\n")
|
||
f.write(f"{'='*80}\n\n")
|
||
|
||
def _log(msg: str):
|
||
"""追加一行到 log.txt 并同时 print"""
|
||
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
||
line = f"[{ts}] {msg}"
|
||
print(line)
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(line + "\n")
|
||
|
||
def _log_raw(text: str):
|
||
"""追加原始文本到 log.txt（不 print）"""
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(text + "\n")
|
||
|
||
def _log_request(method: str, url: str, data=None, params=None, tag: str = ""):
|
||
"""记录 HTTP 请求详情"""
|
||
_log_raw(f"\n{'─'*70}")
|
||
_log_raw(f">>> REQUEST {tag}")
|
||
_log_raw(f" {method} {url}")
|
||
if params:
|
||
_log_raw(f" PARAMS: {json.dumps(params, ensure_ascii=False, indent=6)}")
|
||
if data:
|
||
# 脱敏卡号
|
||
safe = dict(data) if isinstance(data, dict) else {}
|
||
if "card[number]" in safe:
|
||
safe["card[number]"] = "****" + str(safe["card[number]"])[-4:]
|
||
if "card[cvc]" in safe:
|
||
safe["card[cvc]"] = "***"
|
||
_log_raw(f" BODY: {json.dumps(safe, ensure_ascii=False, indent=6)}")
|
||
|
||
def _log_response(resp: requests.Response, tag: str = ""):
|
||
"""记录 HTTP 响应详情"""
|
||
_log_raw(f"<<< RESPONSE {tag} status={resp.status_code}")
|
||
try:
|
||
body = resp.json()
|
||
_log_raw(f" BODY: {json.dumps(body, ensure_ascii=False, indent=6)}")
|
||
except Exception:
|
||
_log_raw(f" BODY(raw): {resp.text[:2000]}")
|
||
_log_raw(f"{'─'*70}\n")
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 常量
|
||
# ---------------------------------------------------------------------------
|
||
STRIPE_API = "https://api.stripe.com"
|
||
STRIPE_VERSION_FULL = "2025-03-31.basil; checkout_server_update_beta=v1; checkout_manual_approval_preview=v1"
|
||
STRIPE_VERSION_BASE = "2025-03-31.basil"
|
||
USER_AGENT = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/146.0.0.0 Safari/537.36"
|
||
)
|
||
HCAPTCHA_SITE_KEY_FALLBACK = "c7faac4c-1cd7-4b1b-b2d4-42ba98d09c7a"
|
||
|
||
KNOWN_PUBLISHABLE_KEYS = {
|
||
"1HOrSwC6h1nxGoI3": "pk_live_51HOrSwC6h1nxGoI3lTAgRjYVrz4dU3fVOabyCcKR3pbEJguCVAlqCxdxCUvoRh1XWwRacViovU3kLKvpkjh7IqkW00iXQsjo3n",
|
||
}
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 地域 / 浏览器配置 — 必须和代理 IP 出口一致
|
||
# ---------------------------------------------------------------------------
|
||
LOCALE_PROFILES = {
|
||
"US": {
|
||
"browser_locale": "en-US",
|
||
"browser_timezone": "America/Los_Angeles",
|
||
"browser_tz_offset": 360, # CST = UTC-6 → 360
|
||
"browser_language": "en-US",
|
||
"color_depth": 24,
|
||
"screen_w": 2560, "screen_h": 1440, "dpr": 1,
|
||
}
|
||
}
|
||
|
||
|
||
APATA_RBA_ORG_ID = "8t63q4n4"
|
||
|
||
def _build_browser_fingerprint(locale_profile: dict) -> dict:
|
||
"""构建 RecordBrowserInfo 的完整设备指纹 payload"""
|
||
sw = locale_profile["screen_w"]
|
||
sh = locale_profile["screen_h"]
|
||
dpr = locale_profile["dpr"]
|
||
cd = locale_profile["color_depth"]
|
||
lang = locale_profile["browser_language"]
|
||
tz_name = locale_profile["browser_timezone"]
|
||
tz_offset = locale_profile["browser_tz_offset"]
|
||
|
||
# 可用高度 = 屏幕高度 - 任务栏 (48-60px)
|
||
avail_h = sh - random.randint(40, 60)
|
||
|
||
return {
|
||
"navigator": {
|
||
"mediaDevices": {"audioinput": random.randint(1, 3), "videoinput": random.randint(0, 2),
|
||
"audiooutput": random.randint(1, 3)},
|
||
"battery": {"charging": True, "chargingTime": 0, "dischargingTime": None,
|
||
"level": round(random.uniform(0.5, 1.0), 2)},
|
||
"appCodeName": "Mozilla", "appName": "Netscape",
|
||
"appVersion": USER_AGENT.replace("Mozilla/", ""),
|
||
"cookieEnabled": True, "doNotTrack": None,
|
||
"hardwareConcurrency": random.choice([8, 12, 16, 32]),
|
||
"language": lang,
|
||
"languages": [lang, lang.split("-")[0]],
|
||
"maxTouchPoints": 0, "onLine": True,
|
||
"platform": "Win32", "product": "Gecko", "productSub": "20030107",
|
||
"userAgent": USER_AGENT,
|
||
"vendor": "Google Inc.", "vendorSub": "",
|
||
"webdriver": False,
|
||
"deviceMemory": random.choice([4, 8, 16]),
|
||
"pdfViewerEnabled": True, "javaEnabled": False,
|
||
"plugins": "PDF Viewer,Chrome PDF Viewer,Chromium PDF Viewer,Microsoft Edge PDF Viewer,WebKit built-in PDF",
|
||
"connections": {
|
||
"effectiveType": "4g",
|
||
"downlink": round(random.uniform(1.0, 10.0), 2),
|
||
"rtt": random.choice([50, 100, 150, 200, 250, 300, 350, 400]),
|
||
"saveData": False,
|
||
},
|
||
},
|
||
"screen": {
|
||
"availHeight": avail_h, "availWidth": sw,
|
||
"availLeft": 0, "availTop": 0,
|
||
"colorDepth": cd, "height": sh, "width": sw,
|
||
"pixelDepth": cd,
|
||
"orientation": "landscape-primary",
|
||
"devicePixelRatio": dpr,
|
||
},
|
||
"timezone": {"offset": tz_offset, "timezone": tz_name},
|
||
"canvas": hashlib.sha256(os.urandom(32)).hexdigest(),
|
||
"permissions": {
|
||
"geolocation": "denied", "notifications": "denied",
|
||
"midi": "denied", "camera": "denied", "microphone": "denied",
|
||
"background-fetch": "prompt", "background-sync": "granted",
|
||
"persistent-storage": "granted", "accelerometer": "granted",
|
||
"gyroscope": "granted", "magnetometer": "granted",
|
||
"clipboard-read": "denied", "clipboard-write": "denied",
|
||
"screen-wake-lock": "denied", "display-capture": "denied",
|
||
"idle-detection": "denied",
|
||
},
|
||
"audio": {"sum": 124.04347527516074},
|
||
"browserBars": {
|
||
"locationbar": True, "menubar": True, "personalbar": True,
|
||
"statusbar": True, "toolbar": True, "scrollbars": True,
|
||
},
|
||
"sensors": {
|
||
"accelerometer": True, "gyroscope": True, "linearAcceleration": True,
|
||
"absoluteOrientation": True, "relativeOrientation": True,
|
||
"magnetometer": False, "ambientLight": False, "proximity": False,
|
||
},
|
||
"storage": {
|
||
"localStorage": True, "sessionStorage": True,
|
||
"indexedDB": True, "openDatabase": False,
|
||
},
|
||
"webGl": {
|
||
"dataHash": hashlib.sha256(os.urandom(32)).hexdigest(),
|
||
"vendor": "Google Inc. (NVIDIA)",
|
||
"renderer": "ANGLE (NVIDIA, NVIDIA GeForce RTX 4060 (0x00002882) Direct3D11 vs_5_0 ps_5_0, D3D11)",
|
||
},
|
||
"adblock": False,
|
||
"clientRects": {
|
||
"x": round(-10004 + random.uniform(-1, 1), 10),
|
||
"y": round(2.35 + random.uniform(-0.01, 0.01), 10),
|
||
"width": round(111.29 + random.uniform(-0.01, 0.01), 10),
|
||
"height": round(111.29 + random.uniform(-0.01, 0.01), 10),
|
||
"top": round(2.35 + random.uniform(-0.01, 0.01), 10),
|
||
"bottom": round(113.64 + random.uniform(-0.01, 0.01), 10),
|
||
"left": round(-10004 + random.uniform(-1, 1), 10),
|
||
"right": round(-9893 + random.uniform(-1, 1), 10),
|
||
},
|
||
"fonts": {"installed_count": random.randint(40, 60), "not_installed_count": 0},
|
||
}
|
||
|
||
|
||
def _gen_fingerprint():
|
||
def _id():
|
||
return str(uuid.uuid4()).replace("-", "") + uuid.uuid4().hex[:6]
|
||
return _id(), _id(), _id()
|
||
|
||
|
||
|
||
_PLUGINS_STR = (
|
||
"PDF Viewer,internal-pdf-viewer,application/pdf,pdf++text/pdf,pdf, "
|
||
"Chrome PDF Viewer,internal-pdf-viewer,application/pdf,pdf++text/pdf,pdf, "
|
||
"Chromium PDF Viewer,internal-pdf-viewer,application/pdf,pdf++text/pdf,pdf, "
|
||
"Microsoft Edge PDF Viewer,internal-pdf-viewer,application/pdf,pdf++text/pdf,pdf, "
|
||
"WebKit built-in PDF,internal-pdf-viewer,application/pdf,pdf++text/pdf,pdf"
|
||
)
|
||
_CANVAS_FPS = [
|
||
"0100100101111111101111101111111001110010110111110111111",
|
||
"0100100101111111101111101111111001110010110111110111110",
|
||
"0100100101111111101111101111111001110010110111110111101",
|
||
]
|
||
_AUDIO_FPS = [
|
||
"d331ca493eb692cfcd19ae5db713ad4b",
|
||
"a7c5f72e1b3d4e8f9c0d2a6b7e8f1c3d",
|
||
"e4b8d6f2a0c3d5e7f9b1c3d5e7f9a0b2",
|
||
]
|
||
|
||
|
||
def _encode_m6(payload: dict) -> str:
|
||
"""JSON → urlencode → base64 (m.stripe.com/6 编码格式)"""
|
||
raw = json.dumps(payload, separators=(",", ":"))
|
||
return base64.b64encode(urllib.parse.quote(raw, safe="").encode()).decode()
|
||
|
||
|
||
def _b64url_seg(n: int = 32) -> str:
|
||
return base64.urlsafe_b64encode(os.urandom(n)).rstrip(b"=").decode()
|
||
|
||
|
||
def register_fingerprint(http: "requests.Session") -> tuple[str, str, str]:
|
||
"""向 m.stripe.com/6 发送 4 次指纹上报, 返回服务端分配的 (guid, muid, sid)。
|
||
如果请求失败, 返回本地随机生成的值。
|
||
"""
|
||
# 本地备用值
|
||
guid, muid, sid = _gen_fingerprint()
|
||
fp_id = uuid.uuid4().hex
|
||
|
||
# 屏幕参数 (US 常见配置)
|
||
screens = [(1920, 1080, 1), (1536, 864, 1.25), (2560, 1440, 1), (1440, 900, 1)]
|
||
sw, sh, dpr = random.choice(screens)
|
||
vh = sh - random.randint(40, 70) # viewport = screen - chrome
|
||
cpu = random.choice([4, 8, 12, 16])
|
||
canvas_fp = random.choice(_CANVAS_FPS)
|
||
audio_fp = random.choice(_AUDIO_FPS)
|
||
|
||
def _build_full(v2: int, inc_ids: bool) -> dict:
|
||
s1, s2, s3, s4, s5 = (_b64url_seg() for _ in range(5))
|
||
ts_now = int(time.time() * 1000)
|
||
return {
|
||
"v2": v2, "id": fp_id,
|
||
"t": round(random.uniform(3, 120), 1),
|
||
"tag": "$npm_package_version", "src": "js",
|
||
"a": {
|
||
"a": {"v": "true", "t": 0},
|
||
"b": {"v": "true", "t": 0},
|
||
"c": {"v": "en-US", "t": 0},
|
||
"d": {"v": "Win32", "t": 0},
|
||
"e": {"v": _PLUGINS_STR, "t": round(random.uniform(0, 0.5), 1)},
|
||
"f": {"v": f"{sw}w_{vh}h_24d_{dpr}r", "t": 0},
|
||
"g": {"v": str(cpu), "t": 0},
|
||
"h": {"v": "false", "t": 0},
|
||
"i": {"v": "sessionStorage-enabled, localStorage-enabled", "t": round(random.uniform(0.5, 2), 1)},
|
||
"j": {"v": canvas_fp, "t": round(random.uniform(5, 120), 1)},
|
||
"k": {"v": "", "t": 0},
|
||
"l": {"v": USER_AGENT, "t": 0},
|
||
"m": {"v": "", "t": 0},
|
||
"n": {"v": "false", "t": round(random.uniform(3, 50), 1)},
|
||
"o": {"v": audio_fp, "t": round(random.uniform(20, 30), 1)},
|
||
},
|
||
"b": {
|
||
"a": f"https://{s1}.{s2}.{s3}/",
|
||
"b": f"https://{s1}.{s3}/{s4}/{s5}/{_b64url_seg()}",
|
||
"c": _b64url_seg(),
|
||
"d": muid if inc_ids else "NA",
|
||
"e": sid if inc_ids else "NA",
|
||
"f": False, "g": True, "h": True,
|
||
"i": ["location"], "j": [],
|
||
"n": round(random.uniform(800, 2000), 1),
|
||
"u": "chatgpt.com", "v": "auth.openai.com",
|
||
"w": f"{ts_now}:{hashlib.sha256(os.urandom(32)).hexdigest()}",
|
||
},
|
||
"h": os.urandom(10).hex(),
|
||
}
|
||
|
||
def _build_mouse(source: str) -> dict:
|
||
return {
|
||
"muid": muid, "sid": sid,
|
||
"url": f"https://{_b64url_seg()}.{_b64url_seg()}/{_b64url_seg()}/{_b64url_seg()}/{_b64url_seg()}",
|
||
"source": source,
|
||
"data": [random.randint(1, 8) for _ in range(10)],
|
||
}
|
||
|
||
m6_headers = {
|
||
"User-Agent": USER_AGENT,
|
||
"Content-Type": "text/plain;charset=UTF-8",
|
||
"Accept": "*/*",
|
||
"Origin": "https://m.stripe.network",
|
||
"Referer": "https://m.stripe.network/",
|
||
}
|
||
m6_url = "https://m.stripe.com/6"
|
||
_log(" [指纹] 向 m.stripe.com/6 注册设备指纹 ...")
|
||
|
||
# #1 完整指纹 (v2=1, 无 ID)
|
||
try:
|
||
r1 = http.post(m6_url, data=_encode_m6(_build_full(1, False)), headers=m6_headers, timeout=10)
|
||
if r1.status_code == 200:
|
||
j = r1.json()
|
||
muid = j.get("muid", muid)
|
||
guid = j.get("guid", guid)
|
||
sid = j.get("sid", sid)
|
||
_log(f" [指纹] #1 OK → muid={muid[:20]}...")
|
||
except Exception as e:
|
||
_log(f" [指纹] #1 失败: {e}")
|
||
|
||
# #2 完整指纹 (v2=2, 带 ID)
|
||
try:
|
||
r2 = http.post(m6_url, data=_encode_m6(_build_full(2, True)), headers=m6_headers, timeout=10)
|
||
if r2.status_code == 200:
|
||
j = r2.json()
|
||
guid = j.get("guid", guid)
|
||
_log(f" [指纹] #2 OK → guid={guid[:20]}...")
|
||
except Exception as e:
|
||
_log(f" [指纹] #2 失败: {e}")
|
||
|
||
# #3 鼠标行为 (mouse-timings-10-v2)
|
||
try:
|
||
http.post(m6_url, data=_encode_m6(_build_mouse("mouse-timings-10-v2")), headers=m6_headers, timeout=10)
|
||
_log(" [指纹] #3 OK (mouse-timings-v2)")
|
||
except Exception:
|
||
pass
|
||
|
||
# #4 鼠标行为 (mouse-timings-10)
|
||
try:
|
||
http.post(m6_url, data=_encode_m6(_build_mouse("mouse-timings-10")), headers=m6_headers, timeout=10)
|
||
_log(" [指纹] #4 OK (mouse-timings)")
|
||
except Exception:
|
||
pass
|
||
|
||
_log(f" [指纹] 完成 → guid={guid[:25]}... muid={muid[:25]}... sid={sid[:25]}...")
|
||
return guid, muid, sid
|
||
|
||
|
||
def _gen_elements_session_id():
|
||
"""生成类似 elements_session_15hfldlRpSm 的 session id"""
|
||
import random, string
|
||
chars = string.ascii_letters + string.digits
|
||
return "elements_session_" + "".join(random.choices(chars, k=11))
|
||
|
||
|
||
def _stripe_headers():
|
||
return {
|
||
"User-Agent": USER_AGENT,
|
||
"Accept": "application/json",
|
||
"Origin": "https://js.stripe.com",
|
||
"Referer": "https://js.stripe.com/",
|
||
}
|
||
def parse_checkout_url(raw: str) -> tuple[str, str]:
|
||
"""解析输入，返回 (session_id, stripe_checkout_url)
|
||
|
||
支持以下格式:
|
||
- 裸 session_id: cs_live_xxx / cs_test_xxx
|
||
- Stripe URL: https://checkout.stripe.com/c/pay/cs_live_xxx
|
||
- ChatGPT URL: https://chatgpt.com/checkout/openai_llc/cs_live_xxx
|
||
"""
|
||
raw = raw.strip()
|
||
m = re.search(r"(cs_(?:live|test)_[A-Za-z0-9]+)", raw)
|
||
if not m:
|
||
raise ValueError(f"无法从输入中提取 checkout_session_id: {raw[:120]}...")
|
||
session_id = m.group(1)
|
||
|
||
# 构建用于 Playwright 等回退方案的 Stripe checkout URL
|
||
# 如果输入是 checkout.stripe.com 的链接则直接使用，否则用标准格式构建
|
||
if "checkout.stripe.com" in raw:
|
||
stripe_url = raw
|
||
else:
|
||
stripe_url = f"https://checkout.stripe.com/c/pay/{session_id}"
|
||
|
||
return session_id, stripe_url
|
||
|
||
def fetch_publishable_key(session: requests.Session, session_id: str, stripe_checkout_url: str) -> str:
|
||
checkout_url = stripe_checkout_url
|
||
|
||
_log("[2/6] 获取 publishable_key ...")
|
||
|
||
for acct_id_part, known_pk in KNOWN_PUBLISHABLE_KEYS.items():
|
||
try:
|
||
url = f"{STRIPE_API}/v1/payment_pages/{session_id}/init"
|
||
post_data = {"key": known_pk, "_stripe_version": STRIPE_VERSION_BASE,
|
||
"browser_locale": "en-US"}
|
||
_log_request("POST", url, data=post_data, tag="[2/6] pk探测")
|
||
test_resp = session.post(url, data=post_data, headers=_stripe_headers(), timeout=15)
|
||
_log_response(test_resp, tag="[2/6] pk探测")
|
||
if test_resp.status_code == 200:
|
||
_log(f" publishable_key: {known_pk[:30]}... (已知)")
|
||
return known_pk
|
||
except Exception as e:
|
||
_log(f" pk探测异常: {e}")
|
||
|
||
pk = _fetch_pk_playwright(checkout_url)
|
||
if pk:
|
||
_log(f" publishable_key: {pk[:30]}... (playwright)")
|
||
return pk
|
||
|
||
raise RuntimeError("无法提取 publishable_key")
|
||
|
||
|
||
def _fetch_pk_playwright(checkout_url: str) -> str | None:
|
||
try:
|
||
from playwright.sync_api import sync_playwright
|
||
except ImportError:
|
||
return None
|
||
|
||
pk = None
|
||
|
||
def on_request(request):
|
||
nonlocal pk
|
||
if pk:
|
||
return
|
||
if "api.stripe.com" in request.url and "init" in request.url:
|
||
post = request.post_data or ""
|
||
m = re.search(r"key=(pk_(?:live|test)_[A-Za-z0-9]+)", post)
|
||
if m:
|
||
pk = m.group(1)
|
||
|
||
try:
|
||
with sync_playwright() as p:
|
||
browser = p.chromium.launch(headless=True)
|
||
page = browser.new_page()
|
||
page.on("request", on_request)
|
||
try:
|
||
page.goto(checkout_url, wait_until="domcontentloaded", timeout=20000)
|
||
for _ in range(10):
|
||
if pk:
|
||
break
|
||
page.wait_for_timeout(1000)
|
||
except Exception:
|
||
pass
|
||
browser.close()
|
||
except Exception:
|
||
return None
|
||
|
||
return pk
|
||
|
||
|
||
def init_checkout(session: requests.Session, session_id: str, pk: str, locale_profile: dict = None) -> tuple[dict, str, dict]:
|
||
"""返回 (init_resp, stripe_ver, ctx) — ctx 包含后续步骤需要的上下文"""
|
||
locale_profile = locale_profile or LOCALE_PROFILES["US"]
|
||
url = f"{STRIPE_API}/v1/payment_pages/{session_id}/init"
|
||
stripe_js_id = str(uuid.uuid4())
|
||
elements_session_id = _gen_elements_session_id()
|
||
|
||
for version in [STRIPE_VERSION_BASE, STRIPE_VERSION_FULL]:
|
||
data = {
|
||
"browser_locale": locale_profile["browser_locale"],
|
||
"browser_timezone": locale_profile["browser_timezone"],
|
||
"elements_session_client[elements_init_source]": "custom_checkout",
|
||
"elements_session_client[referrer_host]": "chatgpt.com",
|
||
"elements_session_client[stripe_js_id]": stripe_js_id,
|
||
"elements_session_client[locale]": locale_profile["browser_locale"],
|
||
"elements_session_client[is_aggregation_expected]": "false",
|
||
"key": pk,
|
||
"_stripe_version": version,
|
||
}
|
||
if version == STRIPE_VERSION_FULL:
|
||
data["elements_session_client[client_betas][0]"] = "custom_checkout_server_updates_1"
|
||
data["elements_session_client[client_betas][1]"] = "custom_checkout_manual_approval_1"
|
||
|
||
_log(f" 初始化结账会话 (init) ... version={version[:30]}")
|
||
_log_request("POST", url, data=data, tag="[2b/6] init")
|
||
resp = session.post(url, data=data, headers=_stripe_headers())
|
||
_log_response(resp, tag="[2b/6] init")
|
||
if resp.status_code == 200:
|
||
ctx = {
|
||
"stripe_js_id": stripe_js_id,
|
||
"elements_session_id": elements_session_id,
|
||
}
|
||
return resp.json(), version, ctx
|
||
if resp.status_code == 400 and "beta" in resp.text.lower():
|
||
_log(f" 版本 {version[:20]}... 不支持 beta, 尝试下一个 ...")
|
||
continue
|
||
raise RuntimeError(f"init 失败 [{resp.status_code}]: {resp.text[:500]}")
|
||
|
||
raise RuntimeError("init 失败: 所有 Stripe API 版本均不可用")
|
||
|
||
|
||
def extract_hcaptcha_config(init_resp: dict) -> dict:
|
||
raw = json.dumps(init_resp)
|
||
result = {"site_key": HCAPTCHA_SITE_KEY_FALLBACK, "rqdata": ""}
|
||
|
||
if init_resp.get("site_key"):
|
||
result["site_key"] = init_resp["site_key"]
|
||
m = re.search(r'"hcaptcha_site_key"\s*:\s*"([^"]+)"', raw)
|
||
if m and not init_resp.get("site_key"):
|
||
result["site_key"] = m.group(1)
|
||
|
||
m = re.search(r'"hcaptcha_rqdata"\s*:\s*"([^"]+)"', raw)
|
||
if m:
|
||
result["rqdata"] = m.group(1)
|
||
|
||
return result
|
||
|
||
|
||
def fetch_elements_session(
|
||
session: requests.Session,
|
||
pk: str,
|
||
session_id: str,
|
||
ctx: dict,
|
||
stripe_ver: str = STRIPE_VERSION_FULL,
|
||
locale_profile: dict = None,
|
||
) -> dict:
|
||
"""调用 elements/sessions, 返回响应 dict 并更新 ctx 中的 elements_session_id"""
|
||
locale_profile = locale_profile or LOCALE_PROFILES["US"]
|
||
locale_short = locale_profile["browser_locale"].split("-")[0] # HAR: "zh" 而非 "zh-CN"
|
||
stripe_js_id = ctx.get("stripe_js_id", str(uuid.uuid4()))
|
||
url = f"{STRIPE_API}/v1/elements/sessions"
|
||
params = {
|
||
"client_betas[0]": "custom_checkout_server_updates_1",
|
||
"client_betas[1]": "custom_checkout_manual_approval_1",
|
||
"deferred_intent[mode]": "subscription",
|
||
"deferred_intent[amount]": "0",
|
||
"deferred_intent[currency]": "usd",
|
||
"deferred_intent[setup_future_usage]": "off_session",
|
||
"deferred_intent[payment_method_types][0]": "card",
|
||
"currency": "usd",
|
||
"key": pk,
|
||
"_stripe_version": stripe_ver,
|
||
"elements_init_source": "custom_checkout",
|
||
"referrer_host": "chatgpt.com",
|
||
"stripe_js_id": stripe_js_id,
|
||
"locale": locale_short,
|
||
"type": "deferred_intent",
|
||
"checkout_session_id": session_id,
|
||
}
|
||
_log(" [elements] GET /v1/elements/sessions ...")
|
||
_log_request("GET", url, params=params, tag="[2c] elements/sessions")
|
||
resp = session.get(url, params=params, headers=_stripe_headers())
|
||
_log_response(resp, tag="[2c] elements/sessions")
|
||
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
# 提取真实的 elements_session_id (如果有)
|
||
real_es_id = data.get("session_id") or data.get("id")
|
||
if real_es_id:
|
||
ctx["elements_session_id"] = real_es_id
|
||
_log(f" [elements] 真实 session_id: {real_es_id}")
|
||
# 提取 config_id
|
||
config_id = data.get("config_id")
|
||
if config_id:
|
||
ctx["config_id"] = config_id
|
||
_log(f" [elements] config_id: {config_id}")
|
||
return data
|
||
else:
|
||
_log(f" [elements] 请求失败 [{resp.status_code}], 继续使用本地生成的 ID")
|
||
return {}
|
||
|
||
|
||
|
||
def lookup_consumer(
|
||
session: requests.Session,
|
||
pk: str,
|
||
email: str,
|
||
stripe_ver: str = STRIPE_VERSION_FULL,
|
||
):
|
||
"""查询 Stripe Link 消费者会话，模拟真实浏览器的两次 lookup"""
|
||
url = f"{STRIPE_API}/v1/consumers/sessions/lookup"
|
||
surfaces = [
|
||
("web_link_authentication_in_payment_element", "default_value"),
|
||
("web_elements_controller", "default_value"),
|
||
]
|
||
for surface, source in surfaces:
|
||
data = {
|
||
"request_surface": surface,
|
||
"email_address": email,
|
||
"email_source": source,
|
||
"session_id": str(uuid.uuid4()),
|
||
"key": pk,
|
||
"_stripe_version": stripe_ver,
|
||
}
|
||
if surface == "web_elements_controller":
|
||
data["do_not_log_consumer_funnel_event"] = "true"
|
||
try:
|
||
_log(f" [link] lookup ({surface[:30]}...) ...")
|
||
_log_request("POST", url, data=data, tag="[2d] consumer/lookup")
|
||
resp = session.post(url, data=data, headers=_stripe_headers(), timeout=10)
|
||
_log_response(resp, tag="[2d] consumer/lookup")
|
||
except Exception as e:
|
||
_log(f" [link] lookup 异常: {e}")
|
||
time.sleep(random.uniform(0.3, 0.8))
|
||
|
||
|
||
def update_payment_page_address(
|
||
session: requests.Session,
|
||
pk: str,
|
||
session_id: str,
|
||
card: dict,
|
||
ctx: dict,
|
||
stripe_ver: str = STRIPE_VERSION_FULL,
|
||
):
|
||
"""模拟浏览器逐字段提交地址/税区信息, 共 6 次 POST"""
|
||
url = f"{STRIPE_API}/v1/payment_pages/{session_id}"
|
||
addr = card.get("address", {})
|
||
elements_session_id = ctx.get("elements_session_id", _gen_elements_session_id())
|
||
stripe_js_id = ctx.get("stripe_js_id", str(uuid.uuid4()))
|
||
|
||
# 基础字段 — 每次 update 都要带
|
||
base = {
|
||
"elements_session_client[client_betas][0]": "custom_checkout_server_updates_1",
|
||
"elements_session_client[client_betas][1]": "custom_checkout_manual_approval_1",
|
||
"elements_session_client[elements_init_source]": "custom_checkout",
|
||
"elements_session_client[referrer_host]": "chatgpt.com",
|
||
"elements_session_client[session_id]": elements_session_id,
|
||
"elements_session_client[stripe_js_id]": stripe_js_id,
|
||
"elements_session_client[locale]": "en-US",
|
||
"elements_session_client[is_aggregation_expected]": "false",
|
||
"client_attribution_metadata[merchant_integration_additional_elements][0]": "payment",
|
||
"client_attribution_metadata[merchant_integration_additional_elements][1]": "address",
|
||
"key": pk,
|
||
"_stripe_version": stripe_ver,
|
||
}
|
||
|
||
# HAR 中的逐字段提交顺序: country → (重复一次) → line1 → city → state → postal_code
|
||
address_steps = [
|
||
{"tax_region[country]": addr.get("country", "US")},
|
||
{}, # 重复提交 (无新字段, 模拟用户切换焦点)
|
||
{"tax_region[line1]": addr.get("line1", "")},
|
||
{"tax_region[city]": addr.get("city", "")},
|
||
{"tax_region[state]": addr.get("state", "")},
|
||
{"tax_region[postal_code]": addr.get("postal_code", "")},
|
||
]
|
||
|
||
_log(" [address] 逐字段提交税区地址 ...")
|
||
accumulated = {}
|
||
for step_idx, new_fields in enumerate(address_steps):
|
||
accumulated.update(new_fields)
|
||
data = dict(base)
|
||
data.update(accumulated)
|
||
|
||
step_name = list(new_fields.keys())[0] if new_fields else "(焦点变更)"
|
||
_log(f" [address] step {step_idx + 1}/6: {step_name}")
|
||
_log_request("POST", url, data=data, tag=f"[2e] update_address({step_idx + 1}/6)")
|
||
resp = session.post(url, data=data, headers=_stripe_headers())
|
||
_log_response(resp, tag=f"[2e] update_address({step_idx + 1}/6)")
|
||
|
||
if resp.status_code != 200:
|
||
_log(f" [address] step {step_idx + 1} 返回 {resp.status_code}, 继续 ...")
|
||
|
||
# 模拟人类输入间隔 (2-5 秒)
|
||
time.sleep(random.uniform(2.0, 4.5))
|
||
|
||
def send_telemetry(
|
||
session: requests.Session,
|
||
event_type: str,
|
||
session_id: str,
|
||
ctx: dict,
|
||
):
|
||
"""向 r.stripe.com/b 发送遥测事件, 模拟 stripe.js 行为上报"""
|
||
url = "https://r.stripe.com/b"
|
||
muid = ctx.get("muid", "")
|
||
sid = ctx.get("sid", "")
|
||
guid = ctx.get("guid", "")
|
||
|
||
payload = {
|
||
"v2": 1,
|
||
"tag": event_type,
|
||
"src": "js",
|
||
"pid": "checkout_" + session_id[:20],
|
||
"muid": muid,
|
||
"sid": sid,
|
||
"guid": guid,
|
||
}
|
||
headers = {
|
||
"User-Agent": USER_AGENT,
|
||
"Content-Type": "text/plain;charset=UTF-8",
|
||
"Accept": "*/*",
|
||
"Origin": "https://js.stripe.com",
|
||
"Referer": "https://js.stripe.com/",
|
||
}
|
||
try:
|
||
body = base64.b64encode(json.dumps(payload, separators=(",", ":")).encode()).decode()
|
||
session.post(url, data=body, headers=headers, timeout=5)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def send_telemetry_batch(
|
||
session: requests.Session,
|
||
session_id: str,
|
||
ctx: dict,
|
||
phase: str = "init",
|
||
):
|
||
"""按阶段批量发送遥测事件"""
|
||
events_map = {
|
||
"init": ["checkout.init", "elements.create", "payment_element.mount"],
|
||
"address": ["address.update", "address.focus", "address.blur"],
|
||
"card_input": ["card.focus", "card.input", "card.blur", "cvc.input"],
|
||
"confirm": ["checkout.confirm.start", "payment_method.create", "checkout.confirm.intent"],
|
||
"3ds": ["three_ds2.start", "three_ds2.fingerprint", "three_ds2.authenticate"],
|
||
"poll": ["checkout.poll", "checkout.complete"],
|
||
}
|
||
events = events_map.get(phase, [])
|
||
for evt in events:
|
||
send_telemetry(session, evt, session_id, ctx)
|
||
time.sleep(random.uniform(0.05, 0.2))
|
||
|
||
|
||
def submit_apata_fingerprint(
|
||
session: requests.Session,
|
||
three_ds_server_trans_id: str,
|
||
three_ds_method_url: str,
|
||
notification_url: str,
|
||
locale_profile: dict,
|
||
ctx: dict,
|
||
):
|
||
|
||
|
||
# 1) POST acs-method.apata.io/v1/houston/method — 提交 threeDSMethodData
|
||
_log(" [apata] POST houston/method ...")
|
||
method_data = base64.b64encode(json.dumps({
|
||
"threeDSServerTransID": three_ds_server_trans_id,
|
||
"threeDSMethodNotificationURL": notification_url,
|
||
}, separators=(",", ":")).encode()).decode()
|
||
|
||
try:
|
||
method_url = three_ds_method_url or "https://acs-method.apata.io/v1/houston/method"
|
||
resp = session.post(
|
||
method_url,
|
||
data={"threeDSMethodData": method_data},
|
||
headers={
|
||
"User-Agent": USER_AGENT,
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Origin": "https://js.stripe.com",
|
||
"Referer": "https://js.stripe.com/",
|
||
},
|
||
timeout=15,
|
||
)
|
||
_log(f" [apata] houston/method → {resp.status_code}")
|
||
except Exception as e:
|
||
_log(f" [apata] houston/method 异常: {e}")
|
||
|
||
time.sleep(random.uniform(0.5, 1.0))
|
||
|
||
# 2) POST acs-method.apata.io/v1/RecordBrowserInfo — 设备指纹上报
|
||
_log(" [apata] POST RecordBrowserInfo ...")
|
||
# 生成 possessionDeviceId (localStorage acsRbaDeviceId 模拟)
|
||
possession_device_id = ctx.get("apata_device_id") or str(uuid.uuid4())
|
||
ctx["apata_device_id"] = possession_device_id
|
||
|
||
fp_data = _build_browser_fingerprint(locale_profile)
|
||
record_payload = {
|
||
"threeDSServerTransID": three_ds_server_trans_id,
|
||
"computedValue": hashlib.sha256(os.urandom(32)).hexdigest()[:20],
|
||
"possessionDeviceId": possession_device_id,
|
||
}
|
||
record_payload.update(fp_data)
|
||
|
||
try:
|
||
record_url = "https://acs-method.apata.io/v1/RecordBrowserInfo"
|
||
resp = session.post(
|
||
record_url,
|
||
json=record_payload,
|
||
headers={
|
||
"User-Agent": USER_AGENT,
|
||
"Content-Type": "application/json",
|
||
"Origin": "https://acs-method.apata.io",
|
||
"Referer": "https://acs-method.apata.io/",
|
||
},
|
||
timeout=15,
|
||
)
|
||
_log(f" [apata] RecordBrowserInfo → {resp.status_code}")
|
||
except Exception as e:
|
||
_log(f" [apata] RecordBrowserInfo 异常: {e}")
|
||
|
||
time.sleep(random.uniform(0.5, 1.0))
|
||
|
||
# 3) GET rba.apata.io/xxx.js — 模拟 RBA profile 脚本加载
|
||
_log(" [apata] GET rba profile script ...")
|
||
rba_session_id = ctx.get("rba_session_id") or str(uuid.uuid4())
|
||
ctx["rba_session_id"] = rba_session_id
|
||
try:
|
||
# HAR 中的 URL 格式: rba.apata.io/<random>.js?<random_param>=<org_id>&<random_param>=<session_id>
|
||
rba_script_name = ''.join(random.choices(string.ascii_lowercase + string.digits, k=16)) + ".js"
|
||
rba_param1 = ''.join(random.choices(string.ascii_lowercase + string.digits, k=16))
|
||
rba_param2 = ''.join(random.choices(string.ascii_lowercase + string.digits, k=16))
|
||
rba_url = f"https://rba.apata.io/{rba_script_name}?{rba_param1}={APATA_RBA_ORG_ID}&{rba_param2}={rba_session_id}"
|
||
resp = session.get(rba_url, headers={"User-Agent": USER_AGENT}, timeout=10)
|
||
_log(f" [apata] rba profile → {resp.status_code}")
|
||
except Exception as e:
|
||
_log(f" [apata] rba profile 异常: {e}")
|
||
|
||
# 4) 模拟 aa.online-metrix.net CONNECT (WebRTC beacon 不可模拟, 仅日志标记)
|
||
_log(" [apata] online-metrix beacon (WebRTC, 已跳过 — 无法在 requests 中模拟)")
|
||
|
||
# 总等待: 让 Apata 有时间处理指纹结果 (HAR 中这个窗口约 8-12 秒)
|
||
wait = random.uniform(5.0, 8.0)
|
||
_log(f" [apata] 等待指纹处理完成 ({wait:.1f}s) ...")
|
||
time.sleep(wait)
|
||
|
||
def solve_hcaptcha(captcha_cfg: dict, hcaptcha_config: dict, max_retries: int = 3) -> tuple[str, str]:
|
||
"""返回 (token, ekey) 元组"""
|
||
api_url = captcha_cfg.get("api_url", "https://api.yescaptcha.com")
|
||
client_key = captcha_cfg["api_key"]
|
||
site_key = hcaptcha_config["site_key"]
|
||
rqdata = hcaptcha_config.get("rqdata", "")
|
||
|
||
for retry in range(max_retries):
|
||
if retry > 0:
|
||
_log(f" --- 重试第 {retry + 1}/{max_retries} 次 ---")
|
||
|
||
_log(f" 解 hCaptcha (siteKey: {site_key[:20]}...)")
|
||
|
||
# 创建 1 个任务
|
||
task_body = {
|
||
"type": "HCaptchaTaskProxyless",
|
||
"websiteURL": "https://b.stripecdn.com/stripethirdparty-srv/assets/v32.1/HCaptchaInvisible.html",
|
||
"websiteKey": site_key,
|
||
"isEnterprise": True,
|
||
"userAgent": USER_AGENT,
|
||
}
|
||
|
||
|
||
create_payload = {"clientKey": client_key, "task": task_body}
|
||
try:
|
||
create_url = f"{api_url}/createTask"
|
||
_log_request("POST", create_url, data=create_payload, tag="[captcha] createTask")
|
||
create_resp = requests.post(create_url, json=create_payload, timeout=15)
|
||
_log_response(create_resp, tag="[captcha] createTask")
|
||
data = create_resp.json()
|
||
if data.get("errorId", 1) != 0:
|
||
_log(f" 任务创建失败: {data.get('errorDescription', '?')}")
|
||
time.sleep(3)
|
||
continue
|
||
task_id = data["taskId"]
|
||
except Exception as e:
|
||
_log(f" 任务创建异常: {e}")
|
||
time.sleep(3)
|
||
continue
|
||
|
||
_log(f" 任务: {task_id} 等待解题 ...")
|
||
|
||
|
||
for attempt in range(60):
|
||
time.sleep(3)
|
||
try:
|
||
result_url = f"{api_url}/getTaskResult"
|
||
result_payload = {"clientKey": client_key, "taskId": task_id}
|
||
result_resp = requests.post(result_url, json=result_payload, timeout=10)
|
||
result_data = result_resp.json()
|
||
except Exception:
|
||
continue
|
||
|
||
if result_data.get("errorId", 0) != 0:
|
||
error_code = result_data.get("errorCode", "")
|
||
if error_code == "ERROR_TASK_TIMEOUT":
|
||
_log(" 任务超时, 重新发起 ...")
|
||
break
|
||
continue
|
||
|
||
if result_data.get("status") == "ready":
|
||
solution = result_data["solution"]
|
||
_log_raw(f" solution keys: {list(solution.keys())}")
|
||
_log_raw(f" solution full: {json.dumps(solution, ensure_ascii=False)[:500]}")
|
||
token = solution["gRecaptchaResponse"]
|
||
# eKey 可能在不同字段名下
|
||
ekey = solution.get("eKey", "") or solution.get("respKey", "") or solution.get("ekey", "")
|
||
_log(f" 已解决 (token: {len(token)} chars, ekey: {len(ekey)} chars)")
|
||
_log_raw(f" captcha_token(前100): {token[:100]}...")
|
||
if ekey:
|
||
_log_raw(f" captcha_ekey(前100): {ekey[:100]}...")
|
||
return token, ekey
|
||
|
||
if attempt % 5 == 4:
|
||
_log(f" 等待中 ... ({attempt + 1}/60)")
|
||
|
||
raise RuntimeError(f"YesCaptcha 解题失败 (已重试 {max_retries} 轮)")
|
||
|
||
|
||
def create_payment_method(
|
||
session: requests.Session,
|
||
pk: str,
|
||
card: dict,
|
||
captcha_token: str,
|
||
session_id: str,
|
||
stripe_ver: str = STRIPE_VERSION_BASE,
|
||
ctx: dict = None,
|
||
) -> str:
|
||
ctx = ctx or {}
|
||
guid = ctx.get("guid") or _gen_fingerprint()[0]
|
||
muid = ctx.get("muid") or _gen_fingerprint()[0]
|
||
sid = ctx.get("sid") or _gen_fingerprint()[0]
|
||
addr = card.get("address", {})
|
||
|
||
data = {
|
||
"billing_details[name]": card["name"],
|
||
"billing_details[email]": card["email"],
|
||
"billing_details[address][country]": addr.get("country", "US"),
|
||
"billing_details[address][line1]": addr.get("line1", ""),
|
||
"billing_details[address][city]": addr.get("city", ""),
|
||
"billing_details[address][postal_code]": addr.get("postal_code", ""),
|
||
"billing_details[address][state]": addr.get("state", ""),
|
||
"type": "card",
|
||
"card[number]": card["number"],
|
||
"card[cvc]": card["cvc"],
|
||
"card[exp_year]": card["exp_year"],
|
||
"card[exp_month]": card["exp_month"],
|
||
"allow_redisplay": "unspecified",
|
||
|
||
"payment_user_agent": "stripe.js/5412f474d5; stripe-js-v3/5412f474d5; payment-element; deferred-intent",
|
||
"referrer": "https://chatgpt.com",
|
||
# time_on_page: 模拟从页面加载到提交的真实耗时 (HAR: 31368ms / 249421ms)
|
||
"time_on_page": str(ctx.get("time_on_page", random.randint(25000, 55000))),
|
||
"client_attribution_metadata[client_session_id]": str(uuid.uuid4()),
|
||
"client_attribution_metadata[checkout_session_id]": session_id,
|
||
"client_attribution_metadata[merchant_integration_source]": "elements",
|
||
"client_attribution_metadata[merchant_integration_subtype]": "payment-element",
|
||
"client_attribution_metadata[merchant_integration_version]": "2021",
|
||
"client_attribution_metadata[payment_intent_creation_flow]": "deferred",
|
||
"client_attribution_metadata[payment_method_selection_flow]": "automatic",
|
||
"guid": guid,
|
||
"muid": muid,
|
||
"sid": sid,
|
||
"key": pk,
|
||
"_stripe_version": stripe_ver,
|
||
}
|
||
if captcha_token:
|
||
data["radar_options[hcaptcha_token]"] = captcha_token
|
||
|
||
url = f"{STRIPE_API}/v1/payment_methods"
|
||
_log("[4/6] 创建支付方式 (payment_method) ...")
|
||
_log_request("POST", url, data=data, tag="[4/6] create_payment_method")
|
||
resp = session.post(url, data=data, headers=_stripe_headers())
|
||
_log_response(resp, tag="[4/6] create_payment_method")
|
||
if resp.status_code != 200:
|
||
raise RuntimeError(f"创建 payment_method 失败 [{resp.status_code}]: {resp.text[:500]}")
|
||
|
||
pm = resp.json()
|
||
pm_id = pm["id"]
|
||
brand = pm.get("card", {}).get("display_brand", "unknown")
|
||
last4 = pm.get("card", {}).get("last4", "????")
|
||
_log(f" 成功: {pm_id} ({brand} ****{last4})")
|
||
return pm_id
|
||
|
||
def confirm_payment(
|
||
session: requests.Session,
|
||
pk: str,
|
||
session_id: str,
|
||
pm_id: str,
|
||
captcha_token: str,
|
||
init_resp: dict,
|
||
stripe_ver: str = STRIPE_VERSION_BASE,
|
||
captcha_cfg: dict = None,
|
||
captcha_ekey: str = "",
|
||
ctx: dict = None,
|
||
locale_profile: dict = None,
|
||
) -> dict:
|
||
ctx = ctx or {}
|
||
locale_profile = locale_profile or LOCALE_PROFILES["US"]
|
||
guid = ctx.get("guid") or _gen_fingerprint()[0]
|
||
muid = ctx.get("muid") or _gen_fingerprint()[0]
|
||
sid = ctx.get("sid") or _gen_fingerprint()[0]
|
||
|
||
expected_amount = "0"
|
||
line_items = init_resp.get("line_items", [])
|
||
if line_items:
|
||
total = sum(item.get("amount", 0) for item in line_items)
|
||
expected_amount = str(total)
|
||
|
||
|
||
init_checksum = init_resp.get("init_checksum", "")
|
||
config_id = init_resp.get("config_id", "")
|
||
stripe_js_id = ctx.get("stripe_js_id", str(uuid.uuid4()))
|
||
elements_session_id = ctx.get("elements_session_id", _gen_elements_session_id())
|
||
checkout_url = init_resp.get("url") or init_resp.get("stripe_hosted_url") or ""
|
||
|
||
|
||
ver = STRIPE_VERSION_FULL
|
||
|
||
data = {
|
||
"guid": guid,
|
||
"muid": muid,
|
||
"sid": sid,
|
||
"payment_method": pm_id,
|
||
"expected_amount": expected_amount,
|
||
"expected_payment_method_type": "card",
|
||
"consent[terms_of_service]": "accepted",
|
||
"key": pk,
|
||
"_stripe_version": ver,
|
||
|
||
"init_checksum": init_checksum,
|
||
|
||
"version": "5412f474d5",
|
||
|
||
"return_url": checkout_url,
|
||
|
||
"elements_session_client[elements_init_source]": "custom_checkout",
|
||
"elements_session_client[referrer_host]": "chatgpt.com",
|
||
"elements_session_client[stripe_js_id]": stripe_js_id,
|
||
"elements_session_client[locale]": locale_profile.get("browser_locale", "en-US"),
|
||
"elements_session_client[is_aggregation_expected]": "false",
|
||
"elements_session_client[session_id]": elements_session_id,
|
||
"elements_session_client[client_betas][0]": "custom_checkout_server_updates_1",
|
||
"elements_session_client[client_betas][1]": "custom_checkout_manual_approval_1",
|
||
|
||
"client_attribution_metadata[client_session_id]": stripe_js_id,
|
||
"client_attribution_metadata[checkout_session_id]": session_id,
|
||
"client_attribution_metadata[checkout_config_id]": config_id,
|
||
"client_attribution_metadata[elements_session_id]": elements_session_id,
|
||
"client_attribution_metadata[elements_session_config_id]": str(uuid.uuid4()),
|
||
"client_attribution_metadata[merchant_integration_source]": "checkout",
|
||
"client_attribution_metadata[merchant_integration_subtype]": "payment-element",
|
||
"client_attribution_metadata[merchant_integration_version]": "custom",
|
||
"client_attribution_metadata[payment_intent_creation_flow]": "deferred",
|
||
"client_attribution_metadata[payment_method_selection_flow]": "automatic",
|
||
"client_attribution_metadata[merchant_integration_additional_elements][0]": "payment",
|
||
"client_attribution_metadata[merchant_integration_additional_elements][1]": "address",
|
||
}
|
||
|
||
|
||
if captcha_token:
|
||
data["passive_captcha_token"] = captcha_token
|
||
|
||
|
||
url = f"{STRIPE_API}/v1/payment_pages/{session_id}/confirm"
|
||
_log("[5/6] 确认支付 (confirm) ...")
|
||
_log_request("POST", url, data=data, tag="[5/6] confirm")
|
||
resp = session.post(url, data=data, headers=_stripe_headers())
|
||
_log_response(resp, tag="[5/6] confirm")
|
||
if resp.status_code != 200:
|
||
raise RuntimeError(f"confirm 失败 [{resp.status_code}]: {resp.text[:500]}")
|
||
|
||
confirm_data = resp.json()
|
||
|
||
next_action = confirm_data.get("next_action")
|
||
if not next_action:
|
||
seti = _find_setup_intent(confirm_data)
|
||
if seti and seti.get("next_action"):
|
||
next_action = seti["next_action"]
|
||
|
||
if next_action and next_action.get("type") == "use_stripe_sdk":
|
||
_log(" 触发 3DS 验证，正在处理 ...")
|
||
_handle_3ds(session, pk, confirm_data, captcha_token, stripe_ver, captcha_cfg,
|
||
locale_profile=locale_profile, ctx=ctx)
|
||
|
||
return confirm_data
|
||
|
||
|
||
def _find_setup_intent(data: dict) -> dict | None:
|
||
si = data.get("setup_intent")
|
||
if si:
|
||
return si
|
||
pm_obj = data.get("payment_method_object")
|
||
if pm_obj and isinstance(pm_obj, dict):
|
||
return pm_obj.get("setup_intent")
|
||
raw = json.dumps(data)
|
||
m = re.search(r"seti_[A-Za-z0-9]+", raw)
|
||
if m:
|
||
return {"id": m.group(0)}
|
||
return None
|
||
|
||
|
||
def _handle_3ds(
|
||
session: requests.Session,
|
||
pk: str,
|
||
confirm_data: dict,
|
||
captcha_token: str,
|
||
stripe_ver: str = STRIPE_VERSION_BASE,
|
||
captcha_cfg: dict = None,
|
||
locale_profile: dict = None,
|
||
ctx: dict = None,
|
||
):
|
||
"""处理 3DS2 认证流程 (模拟浏览器: captcha → verify_challenge → Apata指纹 → 3ds2/authenticate)"""
|
||
locale_profile = locale_profile or LOCALE_PROFILES["US"]
|
||
ctx = ctx or {}
|
||
raw = json.dumps(confirm_data)
|
||
|
||
# 查找 setatt_ (直接在 confirm 响应中)
|
||
source_match = re.search(r"(setatt_[A-Za-z0-9]+)", raw)
|
||
source = source_match.group(1) if source_match else None
|
||
_log(f" 3DS: setatt_ = {source}")
|
||
|
||
# 查找 seti_ 和 client_secret
|
||
seti_match = re.search(r"(seti_[A-Za-z0-9]+)", raw)
|
||
seti_id = seti_match.group(1) if seti_match else None
|
||
_log(f" 3DS: seti_id = {seti_id}")
|
||
|
||
client_secret = None
|
||
if seti_id:
|
||
cs_match = re.search(rf"({re.escape(seti_id)}_secret_[A-Za-z0-9]+)", raw)
|
||
if cs_match:
|
||
client_secret = cs_match.group(1)
|
||
_log(f" 3DS: client_secret = {client_secret[:40] + '...' if client_secret else None}")
|
||
|
||
|
||
challenge_site_key = None
|
||
challenge_rqdata = ""
|
||
# 从 setup_intent.next_action.use_stripe_sdk.stripe_js 提取
|
||
seti_obj = _find_setup_intent(confirm_data)
|
||
if seti_obj and isinstance(seti_obj, dict):
|
||
na = seti_obj.get("next_action", {})
|
||
sdk_info = na.get("use_stripe_sdk", {})
|
||
stripe_js = sdk_info.get("stripe_js", {})
|
||
if stripe_js.get("site_key"):
|
||
challenge_site_key = stripe_js["site_key"]
|
||
challenge_rqdata = stripe_js.get("rqdata", "")
|
||
_log(f" 检测到 confirmation challenge (site_key: {challenge_site_key[:20]}...)")
|
||
|
||
CHALLENGE_MAX_ATTEMPTS = 5
|
||
if challenge_site_key and seti_id and client_secret and captcha_cfg:
|
||
challenge_hcaptcha_cfg = {
|
||
"site_key": challenge_site_key,
|
||
"rqdata": challenge_rqdata,
|
||
}
|
||
|
||
for challenge_attempt in range(1, CHALLENGE_MAX_ATTEMPTS + 1):
|
||
_log(f" 解 challenge captcha (第 {challenge_attempt}/{CHALLENGE_MAX_ATTEMPTS} 次) ...")
|
||
challenge_token, challenge_ekey = solve_hcaptcha(captcha_cfg, challenge_hcaptcha_cfg, max_retries=3)
|
||
|
||
verify_url = f"{STRIPE_API}/v1/setup_intents/{seti_id}/verify_challenge"
|
||
_log(f" verify_challenge (seti: {seti_id[:30]}...) ...")
|
||
verify_data = {
|
||
"client_secret": client_secret,
|
||
"challenge_response_token": challenge_token,
|
||
"captcha_vendor_name": "hcaptcha",
|
||
"key": pk,
|
||
"_stripe_version": STRIPE_VERSION_FULL,
|
||
}
|
||
|
||
_log_request("POST", verify_url, data=verify_data, tag=f"[5/6] verify_challenge({challenge_attempt}/{CHALLENGE_MAX_ATTEMPTS})")
|
||
resp = session.post(verify_url, data=verify_data, headers=_stripe_headers())
|
||
_log_response(resp, tag=f"[5/6] verify_challenge({challenge_attempt}/{CHALLENGE_MAX_ATTEMPTS})")
|
||
|
||
if resp.status_code != 200:
|
||
err_text = resp.text[:300]
|
||
_log(f" verify_challenge 返回 {resp.status_code}: {err_text}")
|
||
if "no valid challenge" in err_text.lower():
|
||
raise RuntimeError(f"challenge 已失效 (Stripe 返回 {resp.status_code}), 需要重新 confirm 获取新的 challenge")
|
||
break # 其他非 200 退出循环
|
||
|
||
verify_result = resp.json()
|
||
verify_status = verify_result.get("status", "unknown")
|
||
_log(f" verify_challenge 状态: {verify_status}")
|
||
|
||
# 检测 captcha challenge 失败
|
||
setup_error = verify_result.get("last_setup_error", {})
|
||
if setup_error:
|
||
err_code = setup_error.get("code", "")
|
||
err_msg = setup_error.get("message", "")
|
||
if "captcha" in err_msg.lower() or "authentication_failure" in err_code:
|
||
if challenge_attempt < CHALLENGE_MAX_ATTEMPTS:
|
||
_log(f" challenge captcha 被拒, 重试 ...")
|
||
continue # 重新解 captcha 再试
|
||
else:
|
||
raise RuntimeError(f"challenge captcha 连续 {CHALLENGE_MAX_ATTEMPTS} 次被 Stripe 拒绝: [{err_code}] {err_msg}")
|
||
|
||
# verify 成功, 从响应中提取 setatt_
|
||
verify_raw = json.dumps(verify_result)
|
||
new_source = re.search(r"(setatt_[A-Za-z0-9]+)", verify_raw)
|
||
if new_source:
|
||
source = new_source.group(1)
|
||
_log(f" 从 verify 响应中获取 setatt_: {source[:30]}...")
|
||
break # 成功, 退出循环
|
||
|
||
elif seti_id and client_secret and not source:
|
||
# 没有 challenge 但也没有 setatt_, 尝试原始 verify_challenge
|
||
verify_url = f"{STRIPE_API}/v1/setup_intents/{seti_id}/verify_challenge"
|
||
_log(f" verify_challenge (seti: {seti_id[:30]}...) ...")
|
||
verify_data = {
|
||
"client_secret": client_secret,
|
||
"challenge_response_token": captcha_token,
|
||
"captcha_vendor_name": "hcaptcha",
|
||
"key": pk,
|
||
"_stripe_version": STRIPE_VERSION_FULL,
|
||
}
|
||
_log_request("POST", verify_url, data=verify_data, tag="[5/6] verify_challenge(fallback)")
|
||
resp = session.post(verify_url, data=verify_data, headers=_stripe_headers())
|
||
_log_response(resp, tag="[5/6] verify_challenge(fallback)")
|
||
if resp.status_code == 200:
|
||
si_result = resp.json()
|
||
_log(f" verify_challenge 状态: {si_result.get('status', 'unknown')}")
|
||
# 检测 captcha challenge 失败
|
||
setup_error = si_result.get("last_setup_error", {})
|
||
if setup_error:
|
||
err_code = setup_error.get("code", "")
|
||
err_msg = setup_error.get("message", "")
|
||
if "captcha" in err_msg.lower() or "authentication_failure" in err_code:
|
||
raise RuntimeError(f"challenge captcha 被 Stripe 拒绝: [{err_code}] {err_msg}")
|
||
verify_raw = json.dumps(si_result)
|
||
new_source = re.search(r"(setatt_[A-Za-z0-9]+)", verify_raw)
|
||
if new_source:
|
||
source = new_source.group(1)
|
||
else:
|
||
_log(f" verify_challenge 返回 {resp.status_code}: {resp.text[:300]}")
|
||
|
||
|
||
send_telemetry_batch(session, "", ctx, phase="3ds")
|
||
|
||
|
||
three_ds_trans_id = None
|
||
three_ds_method_url = None
|
||
notification_url = None
|
||
|
||
for search_blob in [raw]:
|
||
m_tid = re.search(r'"server_transaction_id"\s*:\s*"([^"]+)"', search_blob)
|
||
if m_tid:
|
||
three_ds_trans_id = m_tid.group(1)
|
||
m_murl = re.search(r'"three_ds_method_url"\s*:\s*"([^"]+)"', search_blob)
|
||
if m_murl:
|
||
three_ds_method_url = m_murl.group(1)
|
||
|
||
if source and three_ds_trans_id:
|
||
# 构建 notification URL (HAR 中的格式)
|
||
acct_match = re.search(r'(acct_[A-Za-z0-9]+)', raw)
|
||
acct_id = acct_match.group(1) if acct_match else "acct_unknown"
|
||
notification_url = f"https://hooks.stripe.com/3d_secure_2/fingerprint/{acct_id}/{source}"
|
||
|
||
submit_apata_fingerprint(
|
||
session=session,
|
||
three_ds_server_trans_id=three_ds_trans_id,
|
||
three_ds_method_url=three_ds_method_url,
|
||
notification_url=notification_url,
|
||
locale_profile=locale_profile,
|
||
ctx=ctx,
|
||
)
|
||
elif source:
|
||
|
||
wait = random.uniform(8.0, 12.0)
|
||
_log(f" [3ds] 未找到 server_transaction_id, 等待 {wait:.1f}s 模拟指纹窗口 ...")
|
||
time.sleep(wait)
|
||
|
||
|
||
if source:
|
||
auth_url = f"{STRIPE_API}/v1/3ds2/authenticate"
|
||
_log(f" 3DS2 authenticate (source: {source[:30]}...) ...")
|
||
auth_data = {
|
||
"source": source,
|
||
"browser": json.dumps({
|
||
"fingerprintAttempted": True,
|
||
"fingerprintData": None,
|
||
"challengeWindowSize": None,
|
||
"threeDSCompInd": "Y",
|
||
"browserJavaEnabled": False,
|
||
"browserJavascriptEnabled": True,
|
||
"browserLanguage": locale_profile.get("browser_language", "en-US"),
|
||
"browserColorDepth": str(locale_profile.get("color_depth", 24)),
|
||
"browserScreenHeight": str(locale_profile.get("screen_h", 1080)),
|
||
"browserScreenWidth": str(locale_profile.get("screen_w", 1920)),
|
||
"browserTZ": str(locale_profile.get("browser_tz_offset", 360)),
|
||
"browserUserAgent": USER_AGENT,
|
||
}),
|
||
"one_click_authn_device_support[hosted]": "false",
|
||
"one_click_authn_device_support[same_origin_frame]": "false",
|
||
"one_click_authn_device_support[spc_eligible]": "true",
|
||
"one_click_authn_device_support[webauthn_eligible]": "true",
|
||
"one_click_authn_device_support[publickey_credentials_get_allowed]": "true",
|
||
"key": pk,
|
||
"_stripe_version": STRIPE_VERSION_FULL,
|
||
}
|
||
_log_request("POST", auth_url, data=auth_data, tag="[5/6] 3ds2/authenticate")
|
||
resp = session.post(auth_url, data=auth_data, headers=_stripe_headers())
|
||
_log_response(resp, tag="[5/6] 3ds2/authenticate")
|
||
if resp.status_code == 200:
|
||
result = resp.json()
|
||
state = result.get("state", "unknown")
|
||
trans_status = result.get("ares", {}).get("transStatus", "?")
|
||
_log(f" 3DS2 结果: state={state}, transStatus={trans_status}")
|
||
else:
|
||
_log(f" 3DS2 authenticate 返回 {resp.status_code}: {resp.text[:200]}")
|
||
else:
|
||
_log(" ⚠ 没有 setatt_ source, 跳过 3DS2 authenticate")
|
||
raise RuntimeError("3DS 验证失败: 未获取到 setatt_ source, 无法完成认证")
|
||
|
||
|
||
if seti_id and client_secret:
|
||
time.sleep(3)
|
||
poll_url = f"{STRIPE_API}/v1/setup_intents/{seti_id}"
|
||
poll_params = {
|
||
"client_secret": client_secret,
|
||
"is_stripe_sdk": "false",
|
||
"key": pk,
|
||
"_stripe_version": STRIPE_VERSION_FULL,
|
||
}
|
||
_log(f" 查询 setup_intent 最终状态 ...")
|
||
_log_request("GET", poll_url, params=poll_params, tag="[5/6] setup_intent状态")
|
||
poll_resp = session.get(poll_url, params=poll_params, headers=_stripe_headers())
|
||
_log_response(poll_resp, tag="[5/6] setup_intent状态")
|
||
if poll_resp.status_code == 200:
|
||
si_status = poll_resp.json().get("status", "unknown")
|
||
_log(f" setup_intent 状态: {si_status}")
|
||
else:
|
||
_log(" ⚠ 无 seti_id / client_secret, 跳过 setup_intent 查询")
|
||
|
||
|
||
def poll_result(session: requests.Session, pk: str, session_id: str, stripe_ver: str = STRIPE_VERSION_BASE) -> dict:
|
||
url = f"{STRIPE_API}/v1/payment_pages/{session_id}/poll"
|
||
params = {
|
||
"key": pk,
|
||
"_stripe_version": stripe_ver,
|
||
}
|
||
|
||
_log("[6/6] 轮询支付结果 ...")
|
||
for attempt in range(30):
|
||
time.sleep(2)
|
||
_log_request("GET", url, params=params, tag=f"[6/6] poll({attempt+1}/30)")
|
||
resp = session.get(url, params=params, headers=_stripe_headers())
|
||
_log_response(resp, tag=f"[6/6] poll({attempt+1}/30)")
|
||
if resp.status_code != 200:
|
||
_log(f" poll 返回 {resp.status_code}, 重试 ...")
|
||
continue
|
||
|
||
data = resp.json()
|
||
state = data.get("state", "unknown")
|
||
payment_status = data.get("payment_object_status", "unknown")
|
||
|
||
if state == "succeeded":
|
||
return_url = data.get("return_url", "")
|
||
_log(f"\n{'='*60}")
|
||
_log(f" 支付成功!")
|
||
_log(f" state: {state}")
|
||
_log(f" payment_status: {payment_status}")
|
||
_log(f" mode: {data.get('mode', '?')}")
|
||
_log(f" return_url: {return_url}")
|
||
_log(f"{'='*60}\n")
|
||
return data
|
||
|
||
if state in ("failed", "expired", "canceled"):
|
||
_log(f"\n 支付失败: state={state}")
|
||
_log_raw(f" 完整 poll 响应: {json.dumps(data, ensure_ascii=False, indent=4)}")
|
||
return data
|
||
|
||
_log(f" state={state}, payment_status={payment_status} ({attempt + 1}/30)")
|
||
|
||
raise TimeoutError("轮询超时 (60s)")
|
||
|
||
def load_config(path: str) -> dict:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def run(checkout_input: str, card_index: int = 0, config_path: str = "config.json", manual_token: str = ""):
|
||
_init_log() # 初始化日志文件
|
||
|
||
cfg = load_config(config_path)
|
||
cards = cfg["cards"]
|
||
if card_index >= len(cards):
|
||
raise ValueError(f"卡索引 {card_index} 超出范围，共 {len(cards)} 张卡")
|
||
card = cards[card_index]
|
||
captcha_cfg = cfg["captcha"]
|
||
|
||
|
||
_FIRST_NAMES = ["JAMES", "JOHN", "ROBERT", "MICHAEL", "WILLIAM", "DAVID", "RICHARD", "JOSEPH",
|
||
"THOMAS", "CHARLES", "DANIEL", "MATTHEW", "ANTHONY", "MARK", "STEVEN",
|
||
"MARY", "PATRICIA", "JENNIFER", "LINDA", "ELIZABETH", "BARBARA", "SUSAN",
|
||
"JESSICA", "SARAH", "KAREN", "NANCY", "LISA", "BETTY", "MARGARET", "SANDRA"]
|
||
_LAST_NAMES = ["SMITH", "JOHNSON", "WILLIAMS", "BROWN", "JONES", "GARCIA", "MILLER",
|
||
"DAVIS", "RODRIGUEZ", "MARTINEZ", "WILSON", "ANDERSON", "TAYLOR", "THOMAS",
|
||
"MOORE", "JACKSON", "MARTIN", "LEE", "THOMPSON", "WHITE", "HARRIS", "CLARK"]
|
||
card["name"] = f"{random.choice(_FIRST_NAMES)} {random.choice(_LAST_NAMES)}"
|
||
|
||
email_user = ''.join(random.choices(string.ascii_lowercase + string.digits, k=random.randint(8, 12)))
|
||
_EMAIL_DOMAINS = ["gmail.com", "yahoo.com", "outlook.com", "hotmail.com", "icloud.com", "protonmail.com"]
|
||
card["email"] = f"{email_user}@{random.choice(_EMAIL_DOMAINS)}"
|
||
|
||
addr = card.get("address", {})
|
||
line1 = addr.get("line1", "")
|
||
|
||
new_line1 = re.sub(r"^\d+", str(random.randint(100, 999)), line1)
|
||
if new_line1 == line1 and line1:
|
||
new_line1 = f"{random.randint(100, 999)} {line1}"
|
||
addr["line1"] = new_line1
|
||
card["address"] = addr
|
||
|
||
|
||
locale_key = cfg.get("locale", addr.get("country", "US")).upper()
|
||
locale_profile = LOCALE_PROFILES.get(locale_key, LOCALE_PROFILES["US"])
|
||
_log(f" 地域: {locale_key} (tz={locale_profile['browser_timezone']}, lang={locale_profile['browser_locale']})")
|
||
|
||
_log(f"\n{'='*60}")
|
||
_log(f" Stripe 自动化支付")
|
||
_log(f" 使用卡: ****{card['number'][-4:]} ({card['name']})")
|
||
_log(f" 邮箱: {card['email']}")
|
||
_log(f" 地址: {addr.get('line1', '')}")
|
||
_log(f" 配置文件: {config_path}")
|
||
_log(f"{'='*60}\n")
|
||
|
||
|
||
_log("[1/6] 解析 checkout session ID ...")
|
||
session_id, stripe_checkout_url = parse_checkout_url(checkout_input)
|
||
_log(f" session_id: {session_id}")
|
||
if "chatgpt.com" in checkout_input:
|
||
_log(f" 输入格式: ChatGPT 嵌入式链接 → 转换为 Stripe URL")
|
||
_log(f" stripe_url: {stripe_checkout_url}")
|
||
|
||
http = requests.Session()
|
||
http.headers.update({"User-Agent": USER_AGENT})
|
||
|
||
# 代理配置
|
||
proxy_cfg = cfg.get("proxy")
|
||
if proxy_cfg:
|
||
host = proxy_cfg["host"]
|
||
port = proxy_cfg["port"]
|
||
user = proxy_cfg.get("user", "")
|
||
pwd = proxy_cfg.get("pass", "")
|
||
if user and pwd:
|
||
proxy_url = f"http://{user}:{pwd}@{host}:{port}"
|
||
else:
|
||
proxy_url = f"http://{host}:{port}"
|
||
http.proxies = {"http": proxy_url, "https": proxy_url}
|
||
_log(f" 代理: {host}:{port} (user={user})")
|
||
else:
|
||
_log(" 代理: 无 (直连)")
|
||
|
||
|
||
reg_guid, reg_muid, reg_sid = register_fingerprint(http)
|
||
|
||
|
||
pk = fetch_publishable_key(http, session_id, stripe_checkout_url)
|
||
|
||
|
||
init_resp, stripe_ver, init_ctx = init_checkout(http, session_id, pk, locale_profile=locale_profile)
|
||
init_ctx["guid"] = reg_guid
|
||
init_ctx["muid"] = reg_muid
|
||
init_ctx["sid"] = reg_sid
|
||
|
||
init_ctx["page_load_ts"] = int(time.time() * 1000)
|
||
mode = init_resp.get("mode", "unknown")
|
||
display_name = init_resp.get("account_settings", {}).get("display_name", "?")
|
||
_log(f" 商户: {display_name} | 模式: {mode}")
|
||
|
||
|
||
send_telemetry_batch(http, session_id, init_ctx, phase="init")
|
||
|
||
|
||
_log("[2c/6] 获取 elements session ...")
|
||
fetch_elements_session(http, pk, session_id, init_ctx, stripe_ver=stripe_ver, locale_profile=locale_profile)
|
||
|
||
|
||
_log("[2d/6] 查询 Link 消费者 ...")
|
||
lookup_consumer(http, pk, card["email"], stripe_ver=stripe_ver)
|
||
|
||
|
||
_log("[2e/6] 逐字段提交地址 ...")
|
||
update_payment_page_address(http, pk, session_id, card, init_ctx, stripe_ver=stripe_ver)
|
||
|
||
|
||
send_telemetry_batch(http, session_id, init_ctx, phase="address")
|
||
|
||
|
||
init_ctx["time_on_page"] = int(time.time() * 1000) - init_ctx.get("page_load_ts", int(time.time() * 1000))
|
||
|
||
hcaptcha_cfg = extract_hcaptcha_config(init_resp)
|
||
_log(f" hCaptcha site_key: {hcaptcha_cfg['site_key']}")
|
||
if hcaptcha_cfg.get("rqdata"):
|
||
_log(f" hCaptcha rqdata: {hcaptcha_cfg['rqdata'][:50]}...")
|
||
|
||
--
|
||
send_telemetry_batch(http, session_id, init_ctx, phase="card_input")
|
||
|
||
if manual_token:
|
||
_log(f"[3/6] 使用手动传入的 token (长度: {len(manual_token)})")
|
||
captcha_token = manual_token
|
||
captcha_ekey = ""
|
||
pm_id = create_payment_method(http, pk, card, captcha_token, session_id, stripe_ver, ctx=init_ctx)
|
||
# ---- 遥测: confirm 阶段 ----
|
||
send_telemetry_batch(http, session_id, init_ctx, phase="confirm")
|
||
confirm_payment(http, pk, session_id, pm_id, captcha_token, init_resp, stripe_ver, captcha_cfg,
|
||
captcha_ekey=captcha_ekey, ctx=init_ctx, locale_profile=locale_profile)
|
||
else:
|
||
|
||
_log("[3/6] 尝试不带 hCaptcha 直接提交 ...")
|
||
try:
|
||
pm_id = create_payment_method(http, pk, card, "", session_id, stripe_ver, ctx=init_ctx)
|
||
send_telemetry_batch(http, session_id, init_ctx, phase="confirm")
|
||
confirm_payment(http, pk, session_id, pm_id, "", init_resp, stripe_ver, captcha_cfg,
|
||
captcha_ekey="", ctx=init_ctx, locale_profile=locale_profile)
|
||
except RuntimeError as e:
|
||
err_msg = str(e).lower()
|
||
if any(kw in err_msg for kw in ["captcha", "hcaptcha", "blocked", "denied", "radar"]):
|
||
_log(f" 需要 hCaptcha，开始解题 ...")
|
||
captcha_token, captcha_ekey = solve_hcaptcha(captcha_cfg, hcaptcha_cfg)
|
||
|
||
init_ctx["time_on_page"] = int(time.time() * 1000) - init_ctx.get("page_load_ts", int(time.time() * 1000))
|
||
pm_id = create_payment_method(http, pk, card, captcha_token, session_id, stripe_ver, ctx=init_ctx)
|
||
send_telemetry_batch(http, session_id, init_ctx, phase="confirm")
|
||
confirm_payment(http, pk, session_id, pm_id, captcha_token, init_resp, stripe_ver, captcha_cfg,
|
||
captcha_ekey=captcha_ekey, ctx=init_ctx, locale_profile=locale_profile)
|
||
else:
|
||
raise
|
||
|
||
|
||
send_telemetry_batch(http, session_id, init_ctx, phase="poll")
|
||
|
||
# Step 6
|
||
result = poll_result(http, pk, session_id, stripe_ver)
|
||
_log(f"\n日志已保存到: {LOG_FILE}")
|
||
return result
|
||
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description="Stripe Checkout 自动化支付",
|
||
epilog="示例: python pay.py cs_live_a12H3g13P9TH6udPmljRCpWsmHiKRFH7VUiZBbcA1U60eMzFFI2wp3rtXL",
|
||
)
|
||
parser.add_argument("session_id", help="Checkout Session ID (cs_live_xxx 或 cs_test_xxx)")
|
||
parser.add_argument("--card", type=int, default=0, help="使用第 N 张卡 (0-based, 默认 0)")
|
||
parser.add_argument("--config", default="config.json", help="配置文件路径 (默认 config.json)")
|
||
parser.add_argument("--token", default="", help="手动传入 hCaptcha token (跳过打码平台)")
|
||
args = parser.parse_args()
|
||
|
||
try:
|
||
run(args.session_id, card_index=args.card, config_path=args.config, manual_token=args.token)
|
||
except Exception as e:
|
||
err_msg = f"\n[ERROR] {type(e).__name__}: {e}"
|
||
print(err_msg, file=sys.stderr)
|
||
# 也写入日志
|
||
try:
|
||
import traceback
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(f"\n{'!'*60}\n")
|
||
f.write(err_msg + "\n")
|
||
f.write(traceback.format_exc())
|
||
f.write(f"{'!'*60}\n")
|
||
except Exception:
|
||
pass
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |