450 lines
17 KiB
Python
450 lines
17 KiB
Python
import json
|
||
import random
|
||
import string
|
||
from typing import Optional, Dict, Tuple
|
||
import time
|
||
import httpx
|
||
import re
|
||
import requests
|
||
import logging
|
||
|
||
from anyio import sleep
|
||
|
||
logging.basicConfig(
|
||
level=logging.DEBUG,
|
||
format='%(asctime)s - %(levelname)s - %(message)s'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# --- 配置信息 ---
|
||
BASE_URL = "https://mail.copy.qzz.io/api"
|
||
AUTH_TOKEN = 'ks4cPQfAvMQI1iqhFCDJTGpuc5ez6B95Iaf9SzA47MP0LiH5Pv7urHjjvVsHJlCY'
|
||
|
||
# --- 公共变量 ---
|
||
headers = {
|
||
'Authorization': 'Bearer ' + AUTH_TOKEN
|
||
}
|
||
|
||
|
||
def extract_verification_link(content):
|
||
"""从邮件内容提取验证链接"""
|
||
match = re.search(r'href="(https://services\.sheerid\.com/verify/[^"]+emailToken=[^"]+)"', content)
|
||
if match:
|
||
return match.group(1).replace('&', '&')
|
||
|
||
match = re.search(r'https://services\.sheerid\.com/verify/[^\s<>"]+emailToken=\d+', content)
|
||
if match:
|
||
return match.group(0)
|
||
|
||
return None
|
||
|
||
def generate_new_email(length: int = 8, domain_index: int = 0) -> str | None:
|
||
"""
|
||
向 API 请求生成一个新的临时邮箱地址。
|
||
|
||
Args:
|
||
length (int): 邮箱地址的用户名长度。
|
||
domain_index (int): 使用的域名索引。
|
||
|
||
Returns:
|
||
str | None: 成功时返回邮箱地址字符串，失败时返回 None。
|
||
"""
|
||
# 1. 明确操作意图，并记录请求参数
|
||
operation = "生成新邮箱地址"
|
||
endpoint = f"{BASE_URL}/generate"
|
||
params = {'length': length, 'domainIndex': domain_index}
|
||
|
||
logger.info(f"开始操作: {operation} (参数: {params})")
|
||
|
||
try:
|
||
# 使用 params 参数，让 requests 自动处理 URL 查询字符串
|
||
logger.debug(f"发送 GET 请求至: {endpoint}")
|
||
response = requests.get(endpoint, headers=headers, params=params)
|
||
|
||
# 2. 关联请求与响应
|
||
logger.info(f"操作 '{operation}' 收到响应，状态码: {response.status_code}")
|
||
# 检查 HTTP 错误状态码
|
||
response.raise_for_status()
|
||
|
||
# 尝试解析 JSON
|
||
data = response.json()
|
||
logger.debug(f"成功解析 JSON 响应: \n{json.dumps(data, indent=2)}")
|
||
|
||
email = data.get('email')
|
||
if email:
|
||
# 3. 增加成功日志
|
||
logger.info(f"成功 {operation}，获取到邮箱: {email}")
|
||
return email
|
||
else:
|
||
logger.error(f"操作 '{operation}' 失败: 响应 JSON 中缺少 'email' 键。")
|
||
return None
|
||
except requests.exceptions.HTTPError as err:
|
||
# 4. 丰富错误上下文
|
||
logger.error(f"操作 '{operation}' 失败，发生 HTTP 错误: {err}")
|
||
# 在 debug 级别记录完整的服务器响应，避免在 INFO/ERROR 级别刷屏
|
||
logger.debug(f"服务器错误响应全文: {err.response.text}")
|
||
return None
|
||
|
||
except requests.exceptions.RequestException as err:
|
||
logger.critical(f"操作 '{operation}' 失败，发生严重请求错误 (如网络问题): {err}")
|
||
return None
|
||
|
||
except json.JSONDecodeError:
|
||
logger.error(f"操作 '{operation}' 失败: 无法解析服务器响应为 JSON。")
|
||
logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'")
|
||
return None
|
||
|
||
|
||
def get_mail_detail_from_mailbox(email_address: str, email_id: int) -> str | None:
|
||
"""
|
||
从指定的邮箱中获取所有邮件，并提取出 SheerID 验证链接。
|
||
Args:
|
||
email_address (str): 要查询的临时邮箱地址。
|
||
auth_token (str): 用于 API 认证的 Bearer Token。
|
||
Returns:
|
||
list[str]: 一个包含所有找到的 SheerID 链接的列表。
|
||
如果找不到链接或发生错误，则返回一个空列表。
|
||
"""
|
||
base_url = "https://mail.copy.qzz.io/api"
|
||
endpoint = f"{base_url}/email/{email_id}"
|
||
operation = f"获取邮箱 <{email_address}> 内的 SheerID 链接"
|
||
logger.info(f"开始操作: {operation}")
|
||
try:
|
||
# --- 发送 API 请求 ---
|
||
response = requests.get(endpoint, headers=headers)
|
||
|
||
|
||
|
||
# --- 解析 JSON ---
|
||
data = response.json()
|
||
logger.debug(f"成功解析 JSON 响应: \n{json.dumps(data, indent=2)}")
|
||
extract_verification_link(data.get('html_content'))
|
||
return data.get('html_content')
|
||
|
||
except requests.exceptions.HTTPError as err:
|
||
logger.error(f"操作 '{operation}' 失败: 发生 HTTP 错误: {err}")
|
||
logger.debug(f"服务器错误响应全文: {err.response.text}")
|
||
return None # 发生错误时返回空列表
|
||
|
||
except requests.exceptions.RequestException as err:
|
||
logger.critical(f"操作 '{operation}' 失败: 发生严重请求错误 (如网络问题): {err}")
|
||
return None
|
||
|
||
except json.JSONDecodeError:
|
||
logger.error(f"操作 '{operation}' 失败: 无法解析服务器响应为 JSON。")
|
||
logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'")
|
||
return None
|
||
|
||
|
||
def get_sheerid_links_from_mailbox(email_address: str) -> str | None:
|
||
"""
|
||
从指定的邮箱中获取所有邮件，并提取出 SheerID 验证链接。
|
||
Args:
|
||
email_address (str): 要查询的临时邮箱地址。
|
||
auth_token (str): 用于 API 认证的 Bearer Token。
|
||
Returns:
|
||
list[str]: 一个包含所有找到的 SheerID 链接的列表。
|
||
如果找不到链接或发生错误，则返回一个空列表。
|
||
"""
|
||
base_url = "https://mail.copy.qzz.io/api"
|
||
endpoint = f"{base_url}/emails"
|
||
|
||
params = {
|
||
"mailbox": email_address
|
||
}
|
||
|
||
operation = f"获取邮箱 <{email_address}> 内的 SheerID 链接"
|
||
logger.info(f"开始操作: {operation}")
|
||
try:
|
||
# --- 发送 API 请求 ---
|
||
response = requests.get(endpoint, headers=headers, params=params)
|
||
response.raise_for_status()
|
||
logger.info(f"操作 '{operation}' 收到响应，状态码: {response.status_code}")
|
||
# --- 解析 JSON ---
|
||
data = response.json()
|
||
logger.debug(f"成功解析 JSON 响应: \n{json.dumps(data, indent=2)}")
|
||
if not data[0].get('sender') == "Verify@SheerID.com":
|
||
return None
|
||
id = data[0].get('id')
|
||
logger.debug(id)
|
||
content = get_mail_detail_from_mailbox(email_address, id)
|
||
url = extract_verification_link(content)
|
||
return url
|
||
|
||
except requests.exceptions.HTTPError as err:
|
||
logger.error(f"操作 '{operation}' 失败: 发生 HTTP 错误: {err}")
|
||
logger.debug(f"服务器错误响应全文: {err.response.text}")
|
||
return None # 发生错误时返回空列表
|
||
|
||
except requests.exceptions.RequestException as err:
|
||
logger.critical(f"操作 '{operation}' 失败: 发生严重请求错误 (如网络问题): {err}")
|
||
return None
|
||
|
||
except json.JSONDecodeError:
|
||
logger.error(f"操作 '{operation}' 失败: 无法解析服务器响应为 JSON。")
|
||
logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'")
|
||
return None
|
||
|
||
|
||
def delete_mailbox(email_address: str) -> bool:
|
||
"""
|
||
通过 API 删除一个临时邮箱地址（邮箱）。
|
||
Args:
|
||
email_address (str): 要删除的邮箱地址。
|
||
auth_token (str): 用于 API 认证的 Bearer Token。
|
||
Returns:
|
||
bool: 如果成功删除则返回 True，否则返回 False。
|
||
"""
|
||
# 稍微修改了函数名，因为 API 端点是 /mailboxes，更准确
|
||
operation = f"删除邮箱 <{email_address}>"
|
||
logger.info(f"开始操作: {operation}")
|
||
# 1. 输入验证
|
||
if not email_address:
|
||
logger.warning("尝试删除一个空的邮箱地址，操作已取消。")
|
||
return False
|
||
endpoint = f"{BASE_URL}/mailboxes"
|
||
params = {
|
||
'address': email_address
|
||
}
|
||
|
||
try:
|
||
# --- 发送 API 请求 ---
|
||
response = requests.delete(endpoint, headers=headers, params=params)
|
||
|
||
logger.info(f"操作 '{operation}' 收到响应，状态码: {response.status_code}")
|
||
response.raise_for_status() # 如果状态码是 4xx 或 5xx，将在此处引发 HTTPError
|
||
# 2. 检查成功响应
|
||
# 成功的 DELETE 请求通常返回 204 No Content (无响应体)
|
||
if response.status_code == 204:
|
||
logger.info(f"成功 {operation} (状态码 204)。")
|
||
return True
|
||
|
||
# 如果返回 200 OK 并带有 JSON 体
|
||
data = response.json()
|
||
logger.debug(f"成功 {operation}，收到的 JSON 详情: \n{json.dumps(data, indent=2)}")
|
||
logger.info(f"成功 {operation} (状态码 {response.status_code})。")
|
||
return True
|
||
except requests.exceptions.HTTPError as err:
|
||
logger.error(f"操作 '{operation}' 失败: 发生 HTTP 错误: {err}")
|
||
# 使用 debug 级别记录详细响应，避免刷屏
|
||
logger.debug(f"服务器错误响应全文: {err.response.text}")
|
||
return False
|
||
|
||
except requests.exceptions.RequestException as err:
|
||
logger.critical(f"操作 '{operation}' 失败: 发生网络等严重请求错误: {err}")
|
||
return False
|
||
|
||
except json.JSONDecodeError:
|
||
logger.error(f"操作 '{operation}' 失败: 服务器返回了成功的状态码，但响应不是有效的 JSON。")
|
||
logger.debug(f"服务器返回的非 JSON 内容: '{response.text}'")
|
||
return False
|
||
|
||
|
||
class SheerIDVerifier:
|
||
def __init__(self, verification_id: str,program_id:str) -> None:
|
||
self.program_id = program_id
|
||
self.verification_id = verification_id
|
||
self.device_fingerprint = self._generate_device_fingerprint()
|
||
self.http_client = httpx.Client(timeout=30.0)
|
||
|
||
def __del__(self):
|
||
if hasattr(self, "http_client"):
|
||
self.http_client.close()
|
||
|
||
@staticmethod
|
||
def _generate_device_fingerprint() -> str:
|
||
chars = '0123456789abcdef'
|
||
return ''.join(random.choice(chars) for _ in range(32))
|
||
|
||
@staticmethod
|
||
def normalize_url(url: str) -> str:
|
||
"""规范化 URL（保留原样）"""
|
||
return url
|
||
|
||
@staticmethod
|
||
def parse_verification_id(url: str) -> Optional[str]:
|
||
match = re.search(r"verificationId=([a-f0-9]+)", url, re.IGNORECASE)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
@staticmethod
|
||
def parse_program_id(url: str) -> Optional[str]:
|
||
match = re.search(r"verify/([a-f0-9]+)/", url, re.IGNORECASE)
|
||
if match:
|
||
return match.group(1)
|
||
return None
|
||
|
||
def _sheerid_request(
|
||
self, method: str, url: str, body: Optional[Dict] = None
|
||
) -> Tuple[Dict, int]:
|
||
"""发送 SheerID API 请求"""
|
||
headers = {
|
||
'host': 'services.sheerid.com',
|
||
'sec-ch-ua-platform': '"Windows"',
|
||
'sec-ch-ua': f'"Chromium";v="142", "Google Chrome";v="142", "Not_A Brand";v="99"',
|
||
'clientversion': '2.157.0',
|
||
'sec-ch-ua-mobile': '?0',
|
||
'clientname': 'jslib',
|
||
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36',
|
||
'accept': 'application/json',
|
||
'content-type': 'application/json',
|
||
'origin': 'https://services.sheerid.com',
|
||
'sec-fetch-site': 'same-origin',
|
||
'sec-fetch-mode': 'cors',
|
||
'sec-fetch-dest': 'empty',
|
||
'referer': f'https://services.sheerid.com/verify/{self.program_id}/?verificationId={self.verification_id}',
|
||
'accept-encoding': 'gzip, deflate, br, zstd',
|
||
'accept-language': 'en-US,en-GB;q=0.9,en;q=0.8',
|
||
'priority': 'u=1, i',
|
||
}
|
||
|
||
try:
|
||
response = self.http_client.request(
|
||
method=method, url=url, json=body, headers=headers
|
||
)
|
||
try:
|
||
data = response.json()
|
||
except Exception:
|
||
data = response.text
|
||
return data, response.status_code
|
||
except Exception as e:
|
||
logger.error(f"SheerID 请求失败: {e}")
|
||
raise
|
||
|
||
def verify(self) -> Dict:
|
||
try:
|
||
|
||
email_address = generate_new_email(10,0)
|
||
time.sleep(30)
|
||
submit_data, submit_status = self._sheerid_request(
|
||
"POST",
|
||
f"https://services.sheerid.com/rest/v2/verification/{self.verification_id}/step/collectMilitaryStatus",
|
||
{'status': 'VETERAN'},
|
||
)
|
||
|
||
if submit_status != 200:
|
||
raise Exception(f"提交状态 失败 (状态码 {submit_status}): {submit_data}")
|
||
if submit_data.get("currentStep") != "collectInactiveMilitaryPersonalInfo":
|
||
error_msg = ", ".join(submit_data.get("errorIds", ["Unknown error"]))
|
||
raise Exception(f"提交状态 错误: {error_msg}")
|
||
|
||
|
||
submissionUrl = submit_data.get("submissionUrl")
|
||
verificationId = re.search(r"verificationId=([a-f0-9]+)", submissionUrl, re.IGNORECASE)
|
||
|
||
random_string = ''.join(random.choices(string.ascii_uppercase, k=5))
|
||
|
||
verify_body = {
|
||
"firstName": "KELLY " + random_string,
|
||
"lastName": "BURKHART",
|
||
"birthDate": "1981-05-14",
|
||
"email": email_address,
|
||
"phoneNumber": "",
|
||
"organization": {
|
||
"id": 4070,
|
||
"name": "Army"
|
||
},
|
||
"dischargeDate": "2025-10-29",
|
||
"locale": "en-US",
|
||
"country": "US",
|
||
"metadata": {
|
||
"marketConsentValue": False,
|
||
"refererUrl": f"https://services.sheerid.com/verify/{self.program_id}/?verificationId={self.verification_id}",
|
||
"verificationId": verificationId,
|
||
"flags": "{\"doc-upload-considerations\":\"default\",\"doc-upload-may24\":\"default\",\"doc-upload-redesign-use-legacy-message-keys\":false,\"docUpload-assertion-checklist\":\"default\",\"include-cvec-field-france-student\":\"not-labeled-optional\",\"org-search-overlay\":\"default\",\"org-selected-display\":\"default\"}",
|
||
"submissionOptIn": "By submitting the personal information above, I acknowledge that my personal information is being collected under the <a target=\"_blank\" rel=\"noopener noreferrer\" class=\"sid-privacy-policy sid-link\" href=\"https://openai.com/policies/privacy-policy/\">privacy policy</a> of the business from which I am seeking a discount, and I understand that my personal information will be shared with SheerID as a processor/third-party service provider in order for SheerID to confirm my eligibility for a special offer. Contact OpenAI Support for further assistance at support@openai.com"
|
||
}
|
||
}
|
||
|
||
info_data, info_status = self._sheerid_request(
|
||
"POST",
|
||
submissionUrl,
|
||
verify_body,
|
||
)
|
||
|
||
if info_status != 200:
|
||
raise Exception(f"提交信息 失败 (状态码 {info_status}): {info_data}")
|
||
if info_data.get("currentStep") == "error":
|
||
error_msg = ", ".join(info_data.get("errorIds", ["Unknown error"]))
|
||
raise Exception(f"提交信息 错误: {error_msg}")
|
||
|
||
time.sleep(120)
|
||
|
||
finish_verifying_url = get_sheerid_links_from_mailbox(email_address)
|
||
while True:
|
||
if finish_verifying_url:
|
||
break
|
||
time.sleep(120)
|
||
finish_verifying_url = get_sheerid_links_from_mailbox(email_address)
|
||
logger.debug("完成验证 " + finish_verifying_url)
|
||
|
||
final_data, final_status = self._sheerid_request(
|
||
"POST",
|
||
finish_verifying_url,
|
||
)
|
||
if final_status != 200:
|
||
raise Exception(f"完成验证 失败 (状态码 {submit_status}): {submit_data}")
|
||
|
||
logger.debug(f"完成验证 {json.dumps(final_data, indent=2)}")
|
||
|
||
time.sleep(30)
|
||
delete_mailbox(email_address)
|
||
# 不做状态轮询，直接返回等待审核
|
||
return {
|
||
"success": True,
|
||
"pending": True,
|
||
"verification_id": self.verification_id,
|
||
"redirect_url": final_data.get("redirectUrl"),
|
||
"status": final_data,
|
||
}
|
||
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 验证失败: {e}")
|
||
return {"success": False, "message": str(e), "verification_id": self.verification_id}
|
||
|
||
|
||
def main():
|
||
"""主函数 - 命令行界面"""
|
||
import sys
|
||
|
||
print("=" * 60)
|
||
print("SheerID 学生身份验证工具 (Python版)")
|
||
print("=" * 60)
|
||
print()
|
||
|
||
if len(sys.argv) > 1:
|
||
url = sys.argv[1]
|
||
else:
|
||
url = input("请输入 SheerID 验证 URL: ").strip()
|
||
|
||
if not url:
|
||
print("❌ 错误: 未提供 URL")
|
||
sys.exit(1)
|
||
|
||
verification_id = SheerIDVerifier.parse_verification_id(url)
|
||
program_id = SheerIDVerifier.parse_program_id(url)
|
||
if not verification_id and not program_id:
|
||
print("❌ 错误: 无效的验证 ID 格式")
|
||
sys.exit(1)
|
||
|
||
print(f"✅ 解析到验证 ID: {verification_id}\n✅ 解析到验证 ID: {program_id}", end="\n")
|
||
|
||
verifier = SheerIDVerifier(verification_id,program_id)
|
||
result = verifier.verify()
|
||
|
||
print()
|
||
print("=" * 60)
|
||
print("验证结果:")
|
||
print("=" * 60)
|
||
print(f"状态: {'✅ 成功' if result['success'] else '❌ 失败'}")
|
||
print(f"消息: {result['message']}")
|
||
if result.get("redirect_url"):
|
||
print(f"跳转 URL: {result['redirect_url']}")
|
||
print("=" * 60)
|
||
|
||
return 0 if result["success"] else 1
|
||
|
||
if __name__ == '__main__':
|
||
exit(main()) |