yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复虎牙手机号登录并统一部署项目名

Browse Source
This commit is contained in:
yml2213
2026-08-31 16:11:08 +08:00
parent 09ab80e062
commit 76b14b69b7
10 changed files with 223 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/app_login.py
+47 -5
View File
@@ -37,7 +37,7 @@ from loguru import logger
from .cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1
from .cookie_utils import normalize_huya_cookie
from .device_fingerprint import account_state_dir, get_huya_sdid
from .device_fingerprint import account_state_dir, get_huya_sdid, reset_account_state
from .device_profile import get_profile, mobile_user_agent
from .dfp_register import DfpRegistrationError, register_device
from .envelope_forge import Envelope
@@ -61,6 +61,8 @@ APP_UA_MOBILE = (
"Chrome/149.0.7827.159 Mobile Safari/537.36 huya adr/13.4.22/xiaomi/30"
)
SessionAssets = tuple[dict, str, str]
RISK_URL_RE = re.compile(rb"https://aq\.huya\.com/p/safe_auth/[^\x00-\x20\"'\\<>]+")
_URL_TAIL_KEEP = set(
"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-._~:/?#[]@!$&'()*+,;=%"
@@ -123,6 +125,7 @@ def wup_password_login_raw(
safedeviceid: str | None = None,
hdid: str | None = None,
proxies: dict | None = None,
session_assets: SessionAssets | None = None,
) -> bytes:
"""发送 WUP 密码登录，返回原始响应字节。
@@ -132,7 +135,7 @@ def wup_password_login_raw(
"""
uid_str = account.removeprefix("hy_")
dev = dict(device_info) if device_info is not None else get_profile(account)
mj, ua, _old_sd = _golden_session_assets()
mj, ua, _old_sd = session_assets or _golden_session_assets()
if not safedeviceid:
try:
_t1, safedeviceid, registered_device_id = register_device(
@@ -202,6 +205,16 @@ def parse_risk_url(resp: bytes) -> str | None:
return (pt or urls)[0]
def _response_markers(resp: bytes) -> str:
"""提取 WUP 错误响应中的短 ASCII 字段，便于区分签名和凭据失败。"""
markers = []
for raw in re.findall(rb"[ -~]{4,}", resp):
value = raw.decode("ascii", "ignore")
if value not in markers and len(value) <= 160:
markers.append(value)
return ", ".join(markers[:8])
def solve_safe_auth(
risk_url: str,
proxies=None,
@@ -282,19 +295,27 @@ def login_cred_with_flow(
except DfpRegistrationError as exc:
raise HuyaAppLoginError(f"新设备注册失败: {exc}") from exc
dev["device_id"] = registered_device_id
# 风控验证绑定首次 WUP 请求的 session/traceId；所有重发必须复用它们。
session_assets = _golden_session_assets()
for rnd in range(max_rounds):
logger.debug(f"[huya-app] WUP 登录请求第 {rnd + 1} 轮...")
resp = wup_password_login_raw(
account,
password,
device_info=dev,
safedeviceid=safedeviceid,
proxies=proxies,
session_assets=session_assets,
)
cred = parse_cred(resp)
risk_url = parse_risk_url(resp) if not cred else None
logger.debug(
f"[huya-app] WUP 第 {rnd + 1} 轮响应: {len(resp)}B, "
f"cred={bool(cred)}, risk={bool(risk_url)}"
)
if cred:
uid = parse_real_uid(resp)
return cred, uid
risk_url = parse_risk_url(resp)
if risk_url:
kind = (
"pt_auth(滑块)"
@@ -309,7 +330,18 @@ def login_cred_with_flow(
solve_safe_auth(risk_url, proxies=proxies, device_info=dev)
logger.info("[huya-app] safe_auth 滑块过验成功，重发 WUP 登录...")
continue
raise HuyaAppLoginError("登录未返回凭据也无风控URL（密码错误或账号状态异常）")
markers = _response_markers(resp)
detail = f"，服务端字段: {markers}" if markers else ""
logger.warning(f"[huya-app] WUP 响应未包含 cred/风控: {len(resp)}B{detail}")
if "LGN_INFO_INVALID_USER_OR_PASSWORD" in markers:
raise HuyaAppLoginError(
"虎牙账号或密码错误（服务端: LGN_INFO_INVALID_USER_OR_PASSWORD）"
)
if "APP_SIGN_NOT_MATCH" in markers:
raise HuyaAppLoginError("虎牙设备签名不匹配（服务端: APP_SIGN_NOT_MATCH）")
raise HuyaAppLoginError(
f"登录未返回凭据也无风控URL（响应 {len(resp)}B{detail}）"
)
raise HuyaAppLoginError(f"{max_rounds} 轮内未取得登录凭据")
@@ -414,6 +446,8 @@ class HuyaAppPasswordLogin:
self.proxies = dict(proxies) if proxies else None
self.timeout = timeout or (10.0, 25.0)
self.force_new_device = force_new_device
if force_new_device:
reset_account_state(self.username)
self.device_info = device_info or get_profile(
self.username, force_new=force_new_device
)
@@ -495,6 +529,7 @@ class HuyaAppPasswordLogin:
device_hint=self.device_info,
)
sdid = sdid_obj.sdid if sdid_obj else ""
logger.info("[huya-app] cred 已获取，开始二维码绑定流程")
pc = QrRole(pc=True, sdid=sdid, proxies=self.proxies)
ph = QrRole(
pc=False,
@@ -511,6 +546,7 @@ class HuyaAppPasswordLogin:
{"behavior": beh, "type": "", "domainList": "", "page": page},
)
qrid = (resp.get("data") or {}).get("qrId")
logger.debug(f"[huya-app] getQrId 响应: qrid={bool(qrid)}")
if not qrid:
return HuyaLoginResult(
success=False,
@@ -540,6 +576,9 @@ class HuyaAppPasswordLogin:
"page": quote(cp, safe=""),
},
)
logger.debug(
f"[huya-app] bindQrLoginUser 响应: returnCode={r2.get('returnCode')}"
)
if r2.get("returnCode") not in (0, "0", None) and r2.get("returnCode") != 0:
logger.warning(
f"[huya-app] bind 返回码: {r2.get('returnCode')} msg: {r2.get('message')}"
@@ -547,7 +586,7 @@ class HuyaAppPasswordLogin:
# 4.3 轮询 tryQrLogin
biztoken = None
for _ in range(12):
for index in range(12):
rt = pc.call(
"/qrLgn/tryQrLogin",
"70003",
@@ -560,6 +599,9 @@ class HuyaAppPasswordLogin:
},
)
dt = rt.get("data") or {}
logger.debug(
f"[huya-app] tryQrLogin 第 {index + 1}/12 轮: stage={dt.get('stage')}"
)
if dt.get("stage") == 2:
biztoken = dt.get("biztoken")
break
core/huya/device_fingerprint.py
+11
View File
@@ -37,6 +37,17 @@ def account_state_dir(account: str) -> Path:
return FP_STATE_ROOT / safe
def reset_account_state(account: str) -> None:
"""清理账号的 hydevice 持久化状态，供真正的全新设备登录使用。"""
state_dir = account_state_dir(account)
for name in ("localstorage.json", "device.json"):
path = state_dir / name
try:
path.unlink()
except FileNotFoundError:
pass
DEFAULT_TIMEOUT = (8, 40)
core/huya/fingerprint/env.js
+4
View File
@@ -13,6 +13,10 @@ function makeEnv(overrides) {
const VW = ov.screenWidth ? Math.round(ov.screenWidth / 2.75) : 393;
const VH = ov.screenHeight ? Math.round(ov.screenHeight / 2.75) : 851;
const W = globalThis;
// Node 18/20 do not expose a browser navigator; create one before defining
// the properties consumed by hydevice. Newer Node versions already expose
// a Navigator instance, which is retained.
W.navigator = W.navigator || {};
W.screen = {width:VW, height:VH, availWidth:VW, availHeight:VH,
colorDepth:24, pixelDepth:24, availLeft:0, availTop:0, orientation:{type:'portrait-primary', angle:0}};
W.devicePixelRatio = 2.75;
core/huya/fingerprint/runner.js
+8 -2
View File
@@ -6,12 +6,18 @@ const path = require('path');
const https = require('https');
globalThis.window = globalThis;
const stateDir = process.argv[2] || '.';
let _deviceOverrides = null;
try { _deviceOverrides = JSON.parse(fs.readFileSync(process.argv[3] || '', 'utf8')); } catch (e) {}
// Prefer an explicitly supplied JSON path; otherwise use the device hint that
// get_huya_sdid writes into the per-account state directory.
try {
const hintPath = process.argv[3] && process.argv[3].endsWith('.json')
? process.argv[3] : path.join(stateDir, 'device.json');
_deviceOverrides = JSON.parse(fs.readFileSync(hintPath, 'utf8'));
} catch (e) {}
require(path.join(__dirname, 'env.js')).makeEnv(_deviceOverrides);
// ---- localStorage 持久化(设备稳定性) ----
const stateDir = process.argv[2] || '.';
try { fs.mkdirSync(stateDir, {recursive: true}); } catch (e) {}
const lsFile = path.join(stateDir, 'localstorage.json');
let _ls = {};
core/huya/wup_encoder.py
+6 -1
View File
@@ -7,6 +7,7 @@ from __future__ import annotations
import json
import random
import re
import struct
import time as _time
from typing import Any
@@ -114,7 +115,11 @@ def _build_meta_json(session: int, trace_id: str) -> str:
def _make_name(uid_str: str) -> str:
"""登录名 = "hy_" + 虎牙号。"""
"""构造 App 登录名：手机号原样提交，虎牙号使用 ``hy_`` 前缀。"""
# App 协议对手机号和虎牙号使用不同的账号命名空间。手机号登录时
# name 就是 11 位手机号；只有数字虎牙号才需要 hy_ 前缀。
if re.fullmatch(r"1\d{10}", uid_str):
return uid_str
if uid_str.startswith("hy_"):
return uid_str
return "hy_" + uid_str
deploy.sh
+33
View File
@@ -13,6 +13,21 @@ if [ -f "$ROOT_DIR/.env" ]; then
set +a
fi
# 使用新的项目标识；可通过环境变量覆盖。
COMPOSE_PROJECT_NAME="${COMPOSE_PROJECT_NAME:-live-hub-py}"
export COMPOSE_PROJECT_NAME
# 迁移期间优先复用旧项目卷，避免目录改名导致创建空数据库。
MYSQL_VOLUME_NAME="${MYSQL_VOLUME_NAME:-}"
MYSQL_VOLUME_EXTERNAL="${MYSQL_VOLUME_EXTERNAL:-false}"
if [ -z "$MYSQL_VOLUME_NAME" ] && command -v docker >/dev/null 2>&1 \
&& docker volume inspect douyu_login_py_mysql-data >/dev/null 2>&1; then
MYSQL_VOLUME_NAME="douyu_login_py_mysql-data"
MYSQL_VOLUME_EXTERNAL="true"
fi
export MYSQL_VOLUME_NAME
export MYSQL_VOLUME_EXTERNAL
APP_PORT="${APP_PORT:-8000}"
# ── 工具函数 ──────────────────────────────────────────────
@@ -27,6 +42,20 @@ detect_compose() {
fi
}
stop_legacy_containers() {
local name project
for name in douyu-login-db douyu-login; do
if ! docker inspect "$name" >/dev/null 2>&1; then
continue
fi
project="$(docker inspect "$name" --format '{{index .Config.Labels "com.docker.compose.project"}}' 2>/dev/null || true)"
if [ -n "$project" ] && [ "$project" != "$COMPOSE_PROJECT_NAME" ]; then
echo "停止旧 Compose 项目容器: $name (项目 $project)"
docker stop "$name" >/dev/null
fi
done
}
# ── 子命令 ──────────────────────────────────────────────
cmd_deploy() {
@@ -65,6 +94,7 @@ cmd_deploy() {
# 构建并启动
echo "正在构建应用镜像（首次构建约3-5分钟）..."
stop_legacy_containers
# 生产部署不构建含大型开发依赖的 test 镜像；测试镜像仅在本地按需构建。
$COMPOSE build douyu-login
@@ -190,6 +220,7 @@ cmd_migrate_mysql() {
fi
echo "正在启动 MySQL..."
stop_legacy_containers
$COMPOSE up -d mysql
echo "等待 MySQL 就绪..."
@@ -222,6 +253,7 @@ cmd_migrate() {
fi
echo "正在启动 MySQL..."
stop_legacy_containers
$COMPOSE up -d mysql
echo "等待 MySQL 就绪..."
@@ -261,6 +293,7 @@ cmd_restart() {
echo "❌ 未检测到 docker compose"
exit 1
fi
stop_legacy_containers
# 重建镜像和应用容器，使新迁移文件与入口迁移逻辑都能生效。
$COMPOSE up -d --build --force-recreate douyu-login
echo "✅ 服务已重建并重启（已自动检查数据库迁移）"
dev.sh
+40 -5
View File
@@ -6,6 +6,15 @@ set -euo pipefail
ROOT_DIR="$(cd "$(dirname "$0")" && pwd)"
cd "$ROOT_DIR"
# 目录改名后，旧 shell 可能仍携带 douyu_login_py/.venv；交给 uv 选择当前项目环境。
if [ "${VIRTUAL_ENV:-}" != "$ROOT_DIR/.venv" ]; then
unset VIRTUAL_ENV
fi
# 使用新的项目标识；可通过环境变量覆盖。
COMPOSE_PROJECT_NAME="${COMPOSE_PROJECT_NAME:-live-hub-py}"
export COMPOSE_PROJECT_NAME
if [ -f "$ROOT_DIR/.env" ]; then
set -a
# 调试模式复用 Docker 部署的环境变量，尤其是 JWT 和敏感字段加密密钥。
@@ -14,6 +23,17 @@ if [ -f "$ROOT_DIR/.env" ]; then
set +a
fi
# 迁移期间优先复用旧项目卷，避免目录改名导致创建空数据库。
MYSQL_VOLUME_NAME="${MYSQL_VOLUME_NAME:-}"
MYSQL_VOLUME_EXTERNAL="${MYSQL_VOLUME_EXTERNAL:-false}"
if [ -z "$MYSQL_VOLUME_NAME" ] && command -v docker >/dev/null 2>&1 \
&& docker volume inspect douyu_login_py_mysql-data >/dev/null 2>&1; then
MYSQL_VOLUME_NAME="douyu_login_py_mysql-data"
MYSQL_VOLUME_EXTERNAL="true"
fi
export MYSQL_VOLUME_NAME
export MYSQL_VOLUME_EXTERNAL
BACKEND_HOST="${BACKEND_HOST:-0.0.0.0}"
# 8800 是项目默认开发端口并由脚本独占；其他显式端口发生冲突时只报错。
BACKEND_PORT_EXPLICIT=false
@@ -58,22 +78,22 @@ fi
# --group dev 确保 pytest 等 dev 依赖已安装。
if [ "${1:-}" = "test" ]; then
shift
exec uv run --group dev pytest "$@"
exec uv run --group dev python -m pytest "$@"
fi
if [ "${1:-}" = "format" ]; then
shift
exec uv run --group dev ruff format "$@"
exec uv run --group dev python -m ruff format "$@"
fi
if [ "${1:-}" = "format-check" ]; then
shift
exec uv run --group dev ruff format --check "$@"
exec uv run --group dev python -m ruff format --check "$@"
fi
if [ "${1:-}" = "type-check" ]; then
shift
exec uv run --group dev pyright "$@"
exec uv run --group dev python -m pyright "$@"
fi
BACKEND_PID=""
@@ -121,6 +141,20 @@ detect_compose() {
fi
}
stop_legacy_containers() {
local name project
for name in douyu-login-db douyu-login; do
if ! docker inspect "$name" >/dev/null 2>&1; then
continue
fi
project="$(docker inspect "$name" --format '{{index .Config.Labels "com.docker.compose.project"}}' 2>/dev/null || true)"
if [ -n "$project" ] && [ "$project" != "$COMPOSE_PROJECT_NAME" ]; then
echo "停止旧 Compose 项目容器: $name (项目 $project)"
docker stop "$name" >/dev/null
fi
done
}
detect_backend_proxy_host() {
if [ "$BACKEND_HOST" != "0.0.0.0" ] && [ "$BACKEND_HOST" != "::" ]; then
echo "$BACKEND_HOST"
@@ -263,6 +297,7 @@ start_yyb_worker() {
}
echo "正在启动本地 MySQL..."
stop_legacy_containers
MYSQL_IMAGE="$MYSQL_IMAGE" \
MYSQL_BIND_HOST="$MYSQL_BIND_HOST" \
MYSQL_HOST_PORT="$MYSQL_HOST_PORT" \
@@ -313,7 +348,7 @@ echo ""
export DB_POOL_SIZE DB_MAX_OVERFLOW DB_POOL_TIMEOUT DB_POOL_RECYCLE
export YYB_WORKER_KEY LOG_LEVEL LOG_DIR
export YYB_WORKER_URL="$DEV_YYB_WORKER_URL"
uv run uvicorn web.backend.main:app \
uv run python -m uvicorn web.backend.main:app \
--host "$BACKEND_HOST" \
--port "$BACKEND_PORT" \
--reload \
docker-compose.yml
+3 -3
View File
@@ -7,7 +7,6 @@ services:
args:
# 服务器默认走阿里云镜像；本地测试设为 false。
USE_CHINA_MIRRORS: ${USE_CHINA_MIRRORS:-true}
container_name: douyu-login
restart: unless-stopped
ports:
# 保持现有服务端口；可通过 APP_PORT 覆盖。
@@ -86,12 +85,10 @@ services:
args:
USE_CHINA_MIRRORS: ${USE_CHINA_MIRRORS:-true}
profiles: ["test"]
container_name: douyu-login-test
entrypoint: ["python", "-m", "pytest"]
mysql:
image: ${MYSQL_IMAGE:-docker.m.daocloud.io/library/mysql:8.4}
container_name: douyu-login-db
restart: unless-stopped
ports:
# 仅绑定宿主机本地地址，供 ./dev.sh 的本地后端连接，不对外网开放。
@@ -121,6 +118,9 @@ services:
volumes:
mysql-data:
# 目录/项目改名时优先复用旧项目的数据卷；新环境默认使用新项目名卷。
name: ${MYSQL_VOLUME_NAME:-live-hub-py_mysql-data}
external: ${MYSQL_VOLUME_EXTERNAL:-false}
networks:
order-site-net:
docs/虎牙App密码登录-WUP明文抓包.md
+3 -3
View File
@@ -16,7 +16,7 @@
| **msgType** | 4097 (0x1001) = `MsgLogin.mMsgId` |
| **请求大小** | 1356 字节 |
| **密码** | SHA1(明文) hex，无盐（`772ed992b0e161276f44ec63671e60155c506294`） |
| **name** | `hy_300023887`（`hy_` + 虎牙号） |
| **name** | 虎牙号为 `hy_<虎牙号>`；手机号登录时为 11 位手机号原值 |
---
@@ -92,7 +92,7 @@ _wup_data = STRUCT {
tag8 = "7c5387e0539c023c31c4ff0e807e7256117385ee" # 另一设备ID(40 hex)
}
field3 = STRUCT { # 登录凭证
tag3 = "hy_300023887" # name (hy_虎牙号)
tag3 = "hy_300023887" # name (虎牙号示例；手机号登录时直接放手机号)
tag4 = "772ed992b0e161276f44ec63671e60155c506294" # password = SHA1(明文)
tag5 = LIST["5008"] # appid列表
tag6 = 1 (INT8)
@@ -151,7 +151,7 @@ userAction JSON:
7c5387e0539c023c31c4ff0e807e7256117385ee ← 另一个设备ID
登录凭证:
hy_300023887 ← name
hy_300023887 ← name (虎牙号示例；手机号登录时直接放手机号)
772ed992b0e161276f44ec63671e60155c506294 ← password (SHA1)
结尾:
tests/test_huya_app_login.py
+68 -1
View File
@@ -17,13 +17,14 @@ from core.huya.app_login import (
wup_password_login_raw,
)
from core.huya.cert_forge import build_p1, decrypt_cert, forge_cert, parse_p1
from core.huya.device_fingerprint import account_state_dir, reset_account_state
from core.huya.device_profile import generate_profile, get_profile
from core.huya.dfp_register import DfpRegistrationError
from core.huya.envelope_forge import Envelope
from core.huya.login import HuyaLoginResult
from core.huya.nonce_forge import K1_DEFAULT, gen_nonce
from core.huya.udb_aes import udb_decrypt, udb_encrypt
from core.huya.wup_encoder import build_password_login_wup
from core.huya.wup_encoder import _make_name, build_password_login_wup
from web.backend.database import Base
from web.backend.models import HuyaAccount, User
from web.backend.routers.huya import (
@@ -104,6 +105,18 @@ class TestHuyaAppLogin:
p3 = get_profile("test_user_account_123")
assert p2["fingerprint"] == p3["fingerprint"]
def test_force_new_device_clears_hydevice_state(self, tmp_path, monkeypatch):
monkeypatch.setattr(
"core.huya.device_fingerprint.FP_STATE_ROOT", tmp_path / "fp"
)
state = account_state_dir("hy_test")
state.mkdir(parents=True)
(state / "localstorage.json").write_text("{}", encoding="utf-8")
(state / "device.json").write_text("{}", encoding="utf-8")
reset_account_state("hy_test")
assert not (state / "localstorage.json").exists()
assert not (state / "device.json").exists()
def test_wup_encoder_output(self):
dev = generate_profile()
pkt = build_password_login_wup(
@@ -121,6 +134,12 @@ class TestHuyaAppLogin:
total_len = struct.unpack(">I", pkt[:4])[0]
assert total_len == len(pkt)
def test_app_login_name_keeps_mobile_number(self):
"""App 协议手机号登录名不能套用虎牙号的 hy_ 前缀。"""
assert _make_name("15197635967") == "15197635967"
assert _make_name("300023887") == "hy_300023887"
assert _make_name("hy_300023887") == "hy_300023887"
# ---- 新设备注册链 (core/huya/dfp_register) 生产接入测试 ----
def test_wup_login_skips_registration_when_safedeviceid_given(self):
@@ -192,6 +211,54 @@ class TestHuyaAppLogin:
):
login_cred_with_flow("300023887", "pw")
def test_login_cred_flow_reuses_wup_session_after_slider(self):
"""滑块通过后的 WUP 重发必须沿用首次请求的会话上下文。"""
assets = ({"session": 123}, "ua", "")
responses = [b"risk", b"success"]
calls = []
def fake_wup(*args, **kwargs):
calls.append(kwargs["session_assets"])
return responses.pop(0)
with (
patch(
"core.huya.app_login.register_device",
return_value=("a" * 32, "A" * 180, "c" * 40),
),
patch("core.huya.app_login._golden_session_assets", return_value=assets),
patch("core.huya.app_login.wup_password_login_raw", side_effect=fake_wup),
patch("core.huya.app_login.parse_cred", side_effect=[None, b"c" * 114]),
patch(
"core.huya.app_login.parse_risk_url",
return_value="https://aq.huya.com/p/safe_auth/pt_auth.html?param=x",
),
patch("core.huya.app_login.solve_safe_auth", return_value={"authId": "id"}),
patch("core.huya.app_login.parse_real_uid", return_value=1199666914671),
):
cred, uid = login_cred_with_flow("300023887", "pw")
assert cred == b"c" * 114
assert uid == 1199666914671
assert calls == [assets, assets]
def test_login_cred_flow_maps_invalid_password_response(self):
"""服务端明确返回账号密码错误时，不再显示泛化的无凭据提示。"""
with (
patch(
"core.huya.app_login.register_device",
return_value=("a" * 32, "A" * 180, "c" * 40),
),
patch(
"core.huya.app_login.wup_password_login_raw",
return_value=b"F!LGN_INFO_INVALID_USER_OR_PASSWORDV",
),
patch("core.huya.app_login.parse_cred", return_value=None),
patch("core.huya.app_login.parse_risk_url", return_value=None),
pytest.raises(HuyaAppLoginError, match="账号或密码错误"),
):
login_cred_with_flow("300023887", "pw")
def test_router_functions(self):
mock_res = HuyaLoginResult(
success=True,