532 lines
19 KiB
Python
532 lines
19 KiB
Python
"""虎牙短信改密流程。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import random
|
|
import string
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from urllib.parse import quote, urlsplit, urlunsplit
|
|
|
|
import requests
|
|
from loguru import logger
|
|
|
|
from core.sms_provider import SmsLine, SmsProviderClient
|
|
|
|
from .login import (
|
|
APP_ID,
|
|
APP_SIGN,
|
|
DEFAULT_UA,
|
|
DF_COLLECT_URL,
|
|
DF_TOKEN_URL,
|
|
HuyaLoginError,
|
|
cookie_mapping,
|
|
generate_context,
|
|
generate_request_id,
|
|
)
|
|
|
|
|
|
CHANGE_PASSWORD_VERSION = "2.5"
|
|
CHANGE_PASSWORD_CHECK_URI = "60011"
|
|
CHANGE_PASSWORD_SEND_SMS_URI = "60003"
|
|
CHANGE_PASSWORD_SUBMIT_URI = "60007"
|
|
CHANGE_PASSWORD_CHECK_URL = "https://udbreg.huya.com/web/v2/modifyPswByCheck"
|
|
CHANGE_PASSWORD_SEND_SMS_URL = "https://udbreg.huya.com/sms/v2/send/changePsw"
|
|
CHANGE_PASSWORD_SUBMIT_URL = "https://udbreg.huya.com/web/v2/changePswBySms"
|
|
CHANGE_PASSWORD_PAGE_URL = "https://aq.huya.com/i/modify.html"
|
|
CHANGE_PASSWORD_FROM_URL = "https://i.huya.com/"
|
|
CHANGE_PASSWORD_PAGE = quote(CHANGE_PASSWORD_FROM_URL, safe="~()*!.'")
|
|
LCID = "2052"
|
|
|
|
|
|
@dataclass
|
|
class HuyaChangePasswordCodeResult:
|
|
"""虎牙改密短信发送结果。"""
|
|
|
|
success: bool
|
|
session_data: str = ""
|
|
message: str = ""
|
|
raw: dict | None = None
|
|
sdid: str = ""
|
|
context: str = ""
|
|
request_id: str = ""
|
|
|
|
|
|
@dataclass
|
|
class HuyaChangePasswordResult:
|
|
"""虎牙改密结果。"""
|
|
|
|
success: bool
|
|
message: str = ""
|
|
code: str = ""
|
|
raw: dict | None = None
|
|
sdid: str = ""
|
|
context: str = ""
|
|
request_id: str = ""
|
|
attempts: int = 0
|
|
|
|
|
|
def generate_huya_password(prefix: str = "hy", random_length: int = 8) -> str:
|
|
"""生成适合虎牙账号使用的随机密码。"""
|
|
clean_prefix = "".join(ch for ch in str(prefix or "hy") if ch.isalnum())[:8] or "hy"
|
|
alphabet = string.ascii_lowercase + string.digits
|
|
suffix = "".join(random.SystemRandom().choice(alphabet) for _ in range(max(6, random_length)))
|
|
return f"{clean_prefix}{suffix}"
|
|
|
|
|
|
def encode_change_password_behavior(stage: str) -> str:
|
|
"""生成改密页行为轨迹。"""
|
|
now = int(time.time() * 1000) - random.randint(2500, 9000)
|
|
elapsed = random.randint(900, 1800)
|
|
actions: list[dict] = []
|
|
|
|
if stage in {"send", "submit"}:
|
|
actions.append({
|
|
"id": "pmodify.btn.getsms",
|
|
"x": random.randint(470, 520),
|
|
"y": random.randint(120, 145),
|
|
"d": elapsed,
|
|
"time": now,
|
|
})
|
|
if stage == "submit":
|
|
for action_id in ("pmodify.input.sms", "pmodify.input.pw", "pmodify.input.sms"):
|
|
now += random.randint(600, 3200)
|
|
elapsed += random.randint(600, 3200)
|
|
actions.append({"id": action_id, "d": elapsed, "time": now})
|
|
now += random.randint(800, 2400)
|
|
elapsed += random.randint(800, 2400)
|
|
actions.append({
|
|
"id": "pmodify.btn.submit",
|
|
"x": random.randint(395, 445),
|
|
"y": random.randint(220, 245),
|
|
"d": elapsed,
|
|
"time": now,
|
|
})
|
|
|
|
value = {
|
|
"furl": CHANGE_PASSWORD_FROM_URL,
|
|
"curl": CHANGE_PASSWORD_PAGE_URL,
|
|
"user_action": actions,
|
|
}
|
|
return quote(json.dumps(value, separators=(",", ":"), ensure_ascii=False), safe="~()*!.'")
|
|
|
|
|
|
def _sleep_or_stop(stop_event: threading.Event | None, seconds: float) -> bool:
|
|
if seconds <= 0:
|
|
return not (stop_event and stop_event.is_set())
|
|
if stop_event:
|
|
return not stop_event.wait(seconds)
|
|
time.sleep(seconds)
|
|
return True
|
|
|
|
|
|
def _payload_data(payload: dict) -> dict:
|
|
data = payload.get("data")
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def _session_data(payload: dict, fallback: str = "") -> str:
|
|
data = _payload_data(payload)
|
|
return str(data.get("sessionData") or data.get("sessiondata") or payload.get("sessionData") or fallback or "")
|
|
|
|
|
|
class HuyaPasswordChanger:
|
|
"""虎牙已登录账号短信改密。"""
|
|
|
|
timeout = (8, 20)
|
|
|
|
def __init__(
|
|
self,
|
|
uid: str,
|
|
cookie: Mapping[str, str] | str | None = None,
|
|
ua: str = DEFAULT_UA,
|
|
session: requests.Session | None = None,
|
|
proxies: Mapping[str, str] | None = None,
|
|
timeout: tuple[float, float] | None = None,
|
|
):
|
|
self.uid = str(uid or "").strip()
|
|
if not self.uid:
|
|
raise HuyaLoginError("缺少虎牙 uid，无法修改密码")
|
|
self.ua = ua or DEFAULT_UA
|
|
self.timeout = timeout or self.timeout
|
|
self.request_id = generate_request_id()
|
|
self.exchange = self.request_id
|
|
self.device_id = uuid.uuid4().hex
|
|
self.mid = uuid.uuid4().hex.upper()
|
|
self.context = generate_context(self.device_id, self.mid)
|
|
self.middle_url = (
|
|
f"https://udbreg.huya.com/web/middle/{CHANGE_PASSWORD_VERSION}/"
|
|
f"{self.exchange}/https/{self.device_id}"
|
|
)
|
|
self.sdid = ""
|
|
|
|
self.session = session or requests.Session()
|
|
self.session.trust_env = False
|
|
if cookie:
|
|
self.session.cookies.update(cookie_mapping(cookie))
|
|
if proxies:
|
|
self.session.proxies = dict(proxies)
|
|
self._setup_headers()
|
|
|
|
def _setup_headers(self) -> None:
|
|
self.session.headers.update({
|
|
"Accept": "application/json, text/plain, */*",
|
|
"Accept-Language": "zh-CN,zh;q=0.9",
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"Content-Type": "application/json;charset=UTF-8",
|
|
"Origin": "https://udbreg.huya.com",
|
|
"Pragma": "no-cache",
|
|
"Referer": self.middle_url,
|
|
"Sec-Fetch-Dest": "empty",
|
|
"Sec-Fetch-Mode": "cors",
|
|
"Sec-Fetch-Site": "same-origin",
|
|
"User-Agent": self.ua,
|
|
"sec-ch-ua": '"Not;A=Brand";v="8", "Chromium";v="150", "Google Chrome";v="150"',
|
|
"sec-ch-ua-mobile": "?0",
|
|
"sec-ch-ua-platform": '"macOS"',
|
|
})
|
|
|
|
@staticmethod
|
|
def _safe_url(url: str) -> str:
|
|
parsed = urlsplit(url)
|
|
return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
|
|
|
|
def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict:
|
|
response = self.session.request(method, url, timeout=self.timeout, **kwargs)
|
|
logger.debug(f"{method.upper()} {self._safe_url(url)} -> {response.status_code}")
|
|
response.raise_for_status()
|
|
try:
|
|
return response.json()
|
|
except json.JSONDecodeError as exc:
|
|
preview = response.text[:200].replace("\n", "\\n")
|
|
raise HuyaLoginError(f"{source} 返回不是 JSON: {preview}") from exc
|
|
|
|
def init_udb_middle(self) -> None:
|
|
"""初始化改密 middle 页面。"""
|
|
for host in ("udblgn.huya.com", "udbreg.huya.com"):
|
|
url = f"https://{host}/web/middle/{CHANGE_PASSWORD_VERSION}/{self.exchange}/https/{self.device_id}"
|
|
try:
|
|
response = self.session.get(url, timeout=self.timeout)
|
|
logger.debug(f"GET {self._safe_url(url)} -> {response.status_code}")
|
|
except requests.RequestException as exc:
|
|
logger.debug(f"虎牙改密 middle 初始化失败: {host}: {exc}")
|
|
self.session.headers.update({
|
|
"Origin": "https://udbreg.huya.com",
|
|
"Referer": self.middle_url,
|
|
})
|
|
|
|
def prepare_device(self) -> str:
|
|
"""获取虎牙风控 sdid。"""
|
|
token_payload = {"encryptVersion": "1.0.1", "fingerprintVersion": "1.2.41"}
|
|
token_res = self._request_json("post", DF_TOKEN_URL, "获取虎牙 df token", json=token_payload)
|
|
token = token_res.get("data", {}).get("token")
|
|
if not token:
|
|
raise HuyaLoginError(f"获取虎牙 df token 失败: {token_res}")
|
|
|
|
collect_res = self._request_json("post", DF_COLLECT_URL, "获取虎牙 sdid", json={"token": token})
|
|
self.sdid = collect_res.get("data", {}).get("sdid", "")
|
|
if not self.sdid:
|
|
raise HuyaLoginError(f"获取虎牙 sdid 失败: {collect_res}")
|
|
return self.sdid
|
|
|
|
def _solve_verification(self, payload: dict) -> None:
|
|
"""处理虎牙风控验证。"""
|
|
from .verification import HuyaVerificationSolver
|
|
|
|
solver = HuyaVerificationSolver(
|
|
cookie=self.session.cookies.get_dict(),
|
|
ua=self.ua,
|
|
sdid=self.sdid,
|
|
session=self.session,
|
|
proxies=self.session.proxies,
|
|
timeout=self.timeout,
|
|
)
|
|
solver.solve(payload)
|
|
|
|
def _common_payload(self, uri: str, data: dict) -> dict:
|
|
return {
|
|
"uri": uri,
|
|
"version": CHANGE_PASSWORD_VERSION,
|
|
"context": self.context,
|
|
"appId": APP_ID,
|
|
"appSign": APP_SIGN,
|
|
"authId": "",
|
|
"sdid": self.sdid,
|
|
"lcid": LCID,
|
|
"byPass": "3",
|
|
"requestId": self.request_id,
|
|
"data": data,
|
|
}
|
|
|
|
def _headers_for_uri(self, uri: str) -> dict:
|
|
return {
|
|
"Origin": "https://udbreg.huya.com",
|
|
"Referer": self.middle_url,
|
|
"context": self.context,
|
|
"lcid": LCID,
|
|
"reqid": self.request_id,
|
|
"uri": uri,
|
|
}
|
|
|
|
def _check_once(self) -> dict:
|
|
data = {
|
|
"uid": self.uid,
|
|
"behavior": encode_change_password_behavior("check"),
|
|
"page": CHANGE_PASSWORD_PAGE,
|
|
}
|
|
return self._request_json(
|
|
"post",
|
|
CHANGE_PASSWORD_CHECK_URL,
|
|
"虎牙改密前置校验",
|
|
json=self._common_payload(CHANGE_PASSWORD_CHECK_URI, data),
|
|
headers=self._headers_for_uri(CHANGE_PASSWORD_CHECK_URI),
|
|
)
|
|
|
|
def _send_sms_once(self, session_data: str) -> dict:
|
|
data = {
|
|
"phone": "",
|
|
"sessionData": session_data,
|
|
"behavior": encode_change_password_behavior("send"),
|
|
"page": CHANGE_PASSWORD_PAGE,
|
|
}
|
|
return self._request_json(
|
|
"post",
|
|
CHANGE_PASSWORD_SEND_SMS_URL,
|
|
"虎牙改密短信发码",
|
|
json=self._common_payload(CHANGE_PASSWORD_SEND_SMS_URI, data),
|
|
headers=self._headers_for_uri(CHANGE_PASSWORD_SEND_SMS_URI),
|
|
)
|
|
|
|
def _submit_once(self, password: str, sms_code: str, session_data: str) -> dict:
|
|
data = {
|
|
"password": password,
|
|
"smsCode": sms_code,
|
|
"sessionData": session_data,
|
|
"behavior": encode_change_password_behavior("submit"),
|
|
"page": CHANGE_PASSWORD_PAGE,
|
|
}
|
|
return self._request_json(
|
|
"post",
|
|
CHANGE_PASSWORD_SUBMIT_URL,
|
|
"虎牙短信改密",
|
|
json=self._common_payload(CHANGE_PASSWORD_SUBMIT_URI, data),
|
|
headers=self._headers_for_uri(CHANGE_PASSWORD_SUBMIT_URI),
|
|
)
|
|
|
|
def send_code(self) -> HuyaChangePasswordCodeResult:
|
|
"""发送改密短信，返回提交改密需要的 sessionData。"""
|
|
self.prepare_device()
|
|
self.init_udb_middle()
|
|
|
|
payload = self._check_once()
|
|
for index in range(3):
|
|
return_code = int(payload.get("returnCode") or 0)
|
|
if return_code in (10030, 10039):
|
|
logger.info(f"虎牙改密前置校验触发风控: {return_code} ({index + 1}/3)")
|
|
self._solve_verification(payload)
|
|
time.sleep(0.5)
|
|
payload = self._check_once()
|
|
continue
|
|
if return_code != 0:
|
|
return HuyaChangePasswordCodeResult(
|
|
success=False,
|
|
message=f"虎牙改密前置校验失败: {payload}",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
break
|
|
else:
|
|
return HuyaChangePasswordCodeResult(
|
|
success=False,
|
|
message=f"虎牙改密前置校验风控后仍失败: {payload}",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
session_data = _session_data(payload)
|
|
if not session_data:
|
|
return HuyaChangePasswordCodeResult(
|
|
success=False,
|
|
message=f"虎牙改密前置校验未返回 sessionData: {payload}",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
payload = self._send_sms_once(session_data)
|
|
for index in range(3):
|
|
return_code = int(payload.get("returnCode") or 0)
|
|
session_data = _session_data(payload, fallback=session_data)
|
|
if return_code == 0 and session_data:
|
|
return HuyaChangePasswordCodeResult(
|
|
success=True,
|
|
session_data=session_data,
|
|
message="改密短信已发送",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
if return_code not in (10030, 10039):
|
|
return HuyaChangePasswordCodeResult(
|
|
success=False,
|
|
message=f"虎牙改密短信发码失败: {payload}",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
logger.info(f"虎牙改密短信发码触发风控: {return_code} ({index + 1}/3)")
|
|
self._solve_verification(payload)
|
|
time.sleep(0.5)
|
|
payload = self._send_sms_once(session_data)
|
|
|
|
return HuyaChangePasswordCodeResult(
|
|
success=False,
|
|
message=f"虎牙改密短信发码风控后仍失败: {payload}",
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
)
|
|
|
|
def submit_code(self, password: str, sms_code: str, session_data: str) -> HuyaChangePasswordResult:
|
|
"""提交改密短信验证码。"""
|
|
if not session_data:
|
|
raise HuyaLoginError("缺少改密 sessionData，请先发送改密短信")
|
|
|
|
payload = self._submit_once(password=password, sms_code=sms_code, session_data=session_data)
|
|
for index in range(3):
|
|
return_code = int(payload.get("returnCode") or 0)
|
|
if return_code == 0:
|
|
return HuyaChangePasswordResult(
|
|
success=True,
|
|
message="密码修改成功",
|
|
code=sms_code,
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
attempts=index + 1,
|
|
)
|
|
if return_code not in (10030, 10039):
|
|
return HuyaChangePasswordResult(
|
|
success=False,
|
|
message=f"虎牙短信改密失败: {payload}",
|
|
code=sms_code,
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
attempts=index + 1,
|
|
)
|
|
|
|
logger.info(f"虎牙短信改密触发风控: {return_code} ({index + 1}/3)")
|
|
self._solve_verification(payload)
|
|
time.sleep(0.5)
|
|
payload = self._submit_once(password=password, sms_code=sms_code, session_data=session_data)
|
|
|
|
return HuyaChangePasswordResult(
|
|
success=False,
|
|
message=f"虎牙短信改密风控后仍失败: {payload}",
|
|
code=sms_code,
|
|
raw=payload,
|
|
sdid=self.sdid,
|
|
context=self.context,
|
|
request_id=self.request_id,
|
|
attempts=3,
|
|
)
|
|
|
|
|
|
def change_huya_password_with_sms_line(
|
|
uid: str,
|
|
cookie: Mapping[str, str] | str,
|
|
password: str,
|
|
item: SmsLine,
|
|
wait_seconds: float = 180,
|
|
poll_interval: float = 5,
|
|
ignore_codes: set[str] | None = None,
|
|
proxies: Mapping[str, str] | None = None,
|
|
stop_event: threading.Event | None = None,
|
|
) -> HuyaChangePasswordResult:
|
|
"""使用同一手机号接码链接完成改密短信验证。"""
|
|
changer = HuyaPasswordChanger(uid=uid, cookie=cookie, proxies=proxies)
|
|
sent_at = datetime.now()
|
|
code_result = changer.send_code()
|
|
if not code_result.success or not code_result.session_data:
|
|
return HuyaChangePasswordResult(
|
|
success=False,
|
|
message=code_result.message or "虎牙改密短信发码失败",
|
|
raw=code_result.raw,
|
|
sdid=code_result.sdid,
|
|
context=code_result.context,
|
|
request_id=code_result.request_id,
|
|
)
|
|
|
|
client = SmsProviderClient()
|
|
deadline = time.monotonic() + max(15.0, float(wait_seconds or 180))
|
|
interval = max(1.0, float(poll_interval or 5))
|
|
attempts = 0
|
|
ignored = {str(code) for code in (ignore_codes or set()) if code}
|
|
last_message = "等待改密验证码"
|
|
|
|
while time.monotonic() < deadline:
|
|
if stop_event and stop_event.is_set():
|
|
return HuyaChangePasswordResult(
|
|
success=False,
|
|
message="已停止",
|
|
sdid=code_result.sdid,
|
|
context=code_result.context,
|
|
request_id=code_result.request_id,
|
|
attempts=attempts,
|
|
)
|
|
|
|
attempts += 1
|
|
poll_result = client.poll(item, after_time=sent_at)
|
|
last_message = poll_result.message or last_message
|
|
if poll_result.has_code:
|
|
if poll_result.code in ignored:
|
|
last_message = "忽略注册阶段旧验证码"
|
|
else:
|
|
result = changer.submit_code(
|
|
password=password,
|
|
sms_code=poll_result.code,
|
|
session_data=code_result.session_data,
|
|
)
|
|
result.attempts = attempts
|
|
return result
|
|
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
break
|
|
if not _sleep_or_stop(stop_event, min(interval, remaining)):
|
|
return HuyaChangePasswordResult(
|
|
success=False,
|
|
message="已停止",
|
|
sdid=code_result.sdid,
|
|
context=code_result.context,
|
|
request_id=code_result.request_id,
|
|
attempts=attempts,
|
|
)
|
|
|
|
return HuyaChangePasswordResult(
|
|
success=False,
|
|
message=f"等待改密验证码超时: {last_message}",
|
|
sdid=code_result.sdid,
|
|
context=code_result.context,
|
|
request_id=code_result.request_id,
|
|
attempts=attempts,
|
|
)
|