304 lines
12 KiB
Python
304 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""应用宝 QQ 二维码登录（纯 Python，无浏览器自动化）。"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import secrets
|
||
import tempfile
|
||
import time
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from http.cookiejar import CookieJar
|
||
from pathlib import Path
|
||
from urllib.error import HTTPError
|
||
from urllib.parse import urlencode
|
||
from urllib.request import HTTPRedirectHandler, HTTPCookieProcessor, Request, build_opener
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
OPEN_APPID = "102033112"
|
||
PT_APPID = "716027609"
|
||
PT_DAID = "383"
|
||
CALLBACK = "https://yybadaccess.3g.qq.com/pc_yyb/pcyyb_oauth?login_type=QC"
|
||
GRAPH_SHOW = "https://graph.qq.com/oauth2.0/show"
|
||
XLOGIN = "https://xui.ptlogin2.qq.com/cgi-bin/xlogin"
|
||
QR_SHOW = "https://xui.ptlogin2.qq.com/ssl/ptqrshow"
|
||
QR_POLL = "https://xui.ptlogin2.qq.com/ssl/ptqrlogin"
|
||
USER_INFO = "https://yybadaccess.3g.qq.com/pc_yyb/pcyyb_get_user_info"
|
||
LOGIN_JUMP = "https://graph.qq.com/oauth2.0/login_jump"
|
||
USER_AGENT = (
|
||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
|
||
class NoRedirect(HTTPRedirectHandler):
|
||
"""Keep OAuth redirects visible while CookieJar receives Set-Cookie headers."""
|
||
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D401
|
||
return None
|
||
|
||
|
||
@dataclass
|
||
class Response:
|
||
status: int
|
||
body: bytes
|
||
headers: object
|
||
|
||
@property
|
||
def text(self) -> str:
|
||
return self.body.decode("utf-8", "replace")
|
||
|
||
def location(self) -> str:
|
||
return str(self.headers.get("Location", ""))
|
||
|
||
|
||
class Client:
|
||
def __init__(self) -> None:
|
||
self.jar = CookieJar()
|
||
self.opener = build_opener(NoRedirect, HTTPCookieProcessor(self.jar))
|
||
|
||
def request(self, url: str, *, method: str = "GET", body: bytes | None = None,
|
||
referer: str = "", headers: dict[str, str] | None = None) -> Response:
|
||
request_headers = {"User-Agent": USER_AGENT, "Accept": "*/*"}
|
||
if referer:
|
||
request_headers["Referer"] = referer
|
||
request_headers.update(headers or {})
|
||
request = Request(url, data=body, headers=request_headers, method=method)
|
||
try:
|
||
response = self.opener.open(request, timeout=30)
|
||
except HTTPError as error:
|
||
response = error
|
||
return Response(response.status, response.read(), response.headers)
|
||
|
||
def cookie(self, name: str) -> str:
|
||
for cookie in self.jar:
|
||
if cookie.name == name:
|
||
return cookie.value
|
||
return ""
|
||
|
||
def cookies(self) -> dict[str, str]:
|
||
return {cookie.name: cookie.value for cookie in self.jar}
|
||
|
||
|
||
def js_int32(value: int) -> int:
|
||
value &= 0xFFFFFFFF
|
||
return value - 0x100000000 if value >= 0x80000000 else value
|
||
|
||
|
||
def ptqr_token(qrsig: str) -> int:
|
||
"""QQ QR's zero-seeded JS 32-bit hash used as ptqrtoken."""
|
||
result = 0
|
||
for char in qrsig:
|
||
result += (js_int32(result) << 5) + ord(char)
|
||
return js_int32(result) & 0x7FFFFFFF
|
||
|
||
|
||
def g_tk(p_skey: str) -> int:
|
||
"""QQ OAuth's 5381-seeded hash used as g_tk."""
|
||
result = 5381
|
||
for char in p_skey:
|
||
result += (js_int32(result) << 5) + ord(char)
|
||
return js_int32(result) & 0x7FFFFFFF
|
||
|
||
|
||
def parse_poll(body: str) -> tuple[int, str]:
|
||
match = re.search(r"ptuiCB\((.*)\)", body, re.S)
|
||
if not match:
|
||
raise ValueError("QQ 二维码轮询响应缺少 ptuiCB")
|
||
values = re.findall(r"'([^']*)'", match.group(1))
|
||
if not values or not values[0].lstrip("-").isdigit():
|
||
raise ValueError("QQ 二维码轮询响应格式异常")
|
||
callback = next((value for value in values[1:] if value.startswith("https://")), "")
|
||
return int(values[0]), callback
|
||
|
||
|
||
def login_type_header(value: str) -> str:
|
||
try:
|
||
return {"QC": "1", "MOBILEQ": "1", "WX": "2"}[value]
|
||
except KeyError as exc:
|
||
raise ValueError(f"未知应用宝登录类型: {value!r}") from exc
|
||
|
||
|
||
def authorize_params(state: str, p_skey: str) -> dict[str, str]:
|
||
return {
|
||
"auth_time": str(int(time.time() * 1000)),
|
||
"client_id": OPEN_APPID,
|
||
"from_ptlogin": "1",
|
||
"g_tk": str(g_tk(p_skey)),
|
||
"openapi": "1010",
|
||
"redirect_uri": CALLBACK,
|
||
"response_type": "code",
|
||
"scope": "",
|
||
"src": "1",
|
||
"state": state,
|
||
"switch": "",
|
||
"ui": str(uuid.uuid4()).upper(),
|
||
"update_auth": "1",
|
||
}
|
||
|
||
|
||
def write_session(path: Path, cookies: dict[str, str]) -> None:
|
||
"""Merge QQ OAuth results while retaining mall data in the session file."""
|
||
document: dict = {}
|
||
if path.exists():
|
||
try:
|
||
document = json.loads(path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"会话文件不是有效 JSON: {path}") from exc
|
||
prior = document.get("cookies", {})
|
||
if not isinstance(prior, dict):
|
||
prior = {}
|
||
document["cookies"] = {**prior, **cookies}
|
||
document["login_type"] = cookies.get("logintype", "QC")
|
||
document["login_updated_at"] = int(time.time())
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
fd, temporary = tempfile.mkstemp(prefix=".mall-session-", suffix=".tmp", dir=path.parent)
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||
json.dump(document, handle, ensure_ascii=False, indent=2)
|
||
handle.write("\n")
|
||
handle.flush()
|
||
os.fsync(handle.fileno())
|
||
os.chmod(temporary, 0o600)
|
||
os.replace(temporary, path)
|
||
os.chmod(path, 0o600)
|
||
except BaseException:
|
||
Path(temporary).unlink(missing_ok=True)
|
||
raise
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="应用宝 QQ 扫码登录（纯 Python）")
|
||
parser.add_argument("--session", type=Path, default=ROOT / "config/mall-session.json")
|
||
parser.add_argument("--qr", type=Path, default=ROOT / "config/qq-login.jpg")
|
||
parser.add_argument("--timeout", type=float, default=300, help="二维码等待秒数")
|
||
parser.add_argument("--interval", type=float, default=3, help="轮询间隔秒数")
|
||
args = parser.parse_args()
|
||
if args.timeout <= 0 or args.interval <= 0:
|
||
raise ValueError("--timeout 和 --interval 必须为正数")
|
||
|
||
client = Client()
|
||
state = secrets.token_urlsafe(14)
|
||
show_query = {
|
||
"which": "Login", "display": "pc", "response_type": "code", "client_id": OPEN_APPID,
|
||
"redirect_uri": CALLBACK, "state": state,
|
||
}
|
||
show_url = f"{GRAPH_SHOW}?{urlencode(show_query)}"
|
||
show = client.request(show_url, referer="https://m.yyb.qq.com/")
|
||
if show.status != 200:
|
||
raise RuntimeError(f"QQ OAuth 页面请求失败: HTTP {show.status}")
|
||
|
||
xlogin_query = {
|
||
"appid": PT_APPID, "daid": PT_DAID, "style": "33", "login_text": "登录",
|
||
"hide_title_bar": "1", "hide_border": "1", "target": "self", "s_url": LOGIN_JUMP,
|
||
"pt_3rd_aid": OPEN_APPID,
|
||
"pt_feedback_link": f"https://support.qq.com/products/77942?customInfo=.appid{OPEN_APPID}",
|
||
"theme": "2", "verify_theme": "",
|
||
}
|
||
xlogin_url = f"{XLOGIN}?{urlencode(xlogin_query)}"
|
||
xlogin = client.request(xlogin_url, referer=show_url)
|
||
if xlogin.status != 200:
|
||
raise RuntimeError(f"QQ 登录页初始化失败: HTTP {xlogin.status}")
|
||
login_sig = client.cookie("pt_login_sig")
|
||
if not login_sig:
|
||
raise RuntimeError("QQ 登录页未写入 pt_login_sig")
|
||
|
||
qr_query = {
|
||
"appid": PT_APPID, "e": "2", "l": "M", "s": "3", "d": "72", "v": "4",
|
||
"t": str(secrets.randbelow(1_000_000) / 1_000_000), "daid": PT_DAID,
|
||
"pt_3rd_aid": OPEN_APPID, "u1": LOGIN_JUMP,
|
||
}
|
||
qr_url = f"{QR_SHOW}?{urlencode(qr_query)}"
|
||
image = client.request(qr_url, referer=xlogin_url)
|
||
if image.status != 200 or not image.body:
|
||
raise RuntimeError(f"QQ 二维码请求失败: HTTP {image.status}")
|
||
qrsig = client.cookie("qrsig")
|
||
if not qrsig:
|
||
raise RuntimeError("QQ 二维码响应未写入 qrsig")
|
||
args.qr.parent.mkdir(parents=True, exist_ok=True)
|
||
args.qr.write_bytes(image.body)
|
||
print(f"QQ 登录二维码: {args.qr}")
|
||
print("请用 QQ 扫码并在手机确认；本终端将自动继续。")
|
||
|
||
deadline = time.monotonic() + args.timeout
|
||
callback = ""
|
||
o1v_id = secrets.token_hex(16)
|
||
while time.monotonic() < deadline:
|
||
poll_query = {
|
||
"u1": LOGIN_JUMP, "ptqrtoken": str(ptqr_token(qrsig)), "ptredirect": "0", "h": "1", "t": "1",
|
||
"g": "1", "from_ui": "1", "ptlang": "2052", "action": f"0-0-{int(time.time() * 1000)}",
|
||
"js_ver": "26071711", "js_type": "1", "login_sig": login_sig, "pt_uistyle": "40",
|
||
"aid": PT_APPID, "daid": PT_DAID, "pt_3rd_aid": OPEN_APPID, "o1vId": o1v_id,
|
||
"pt_js_version": "c1987b96",
|
||
}
|
||
poll = client.request(f"{QR_POLL}?{urlencode(poll_query)}", referer=xlogin_url)
|
||
if poll.status != 200:
|
||
raise RuntimeError(f"QQ 二维码轮询失败: HTTP {poll.status}")
|
||
code, callback = parse_poll(poll.text)
|
||
if code == 0:
|
||
break
|
||
if code in (65, 68):
|
||
raise RuntimeError("QQ 二维码已失效，请重新执行登录")
|
||
if code not in (66, 67):
|
||
raise RuntimeError(f"QQ 二维码登录失败: ptuiCB={code}")
|
||
time.sleep(args.interval)
|
||
else:
|
||
raise TimeoutError("QQ 二维码轮询超时")
|
||
if not callback:
|
||
raise RuntimeError("QQ 登录成功响应缺少 check_sig 回调")
|
||
|
||
check_sig = client.request(callback, referer=xlogin_url)
|
||
login_jump = check_sig.location()
|
||
if check_sig.status != 302 or not login_jump.startswith("https://graph.qq.com/oauth2.0/login_jump"):
|
||
raise RuntimeError("QQ check_sig 未跳转到 OAuth login_jump")
|
||
jump = client.request(login_jump, referer=callback)
|
||
if jump.status != 200:
|
||
raise RuntimeError(f"QQ OAuth login_jump 失败: HTTP {jump.status}")
|
||
|
||
p_skey = client.cookie("p_skey")
|
||
if not p_skey:
|
||
raise RuntimeError("QQ check_sig 未写入 p_skey")
|
||
authorize = client.request(
|
||
"https://graph.qq.com/oauth2.0/authorize", method="POST",
|
||
body=urlencode(authorize_params(state, p_skey)).encode("utf-8"), referer=login_jump,
|
||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||
)
|
||
oauth_callback = authorize.location()
|
||
if authorize.status != 302 or not oauth_callback.startswith(CALLBACK):
|
||
raise RuntimeError("QQ OAuth authorize 未跳转到 YYB 回调")
|
||
yyb_callback = client.request(oauth_callback, referer="https://graph.qq.com/")
|
||
if yyb_callback.status != 302:
|
||
raise RuntimeError(f"YYB QQ OAuth 回调失败: HTTP {yyb_callback.status}")
|
||
cookies = client.cookies()
|
||
if not cookies.get("openid") or not cookies.get("accesstoken"):
|
||
raise RuntimeError("YYB QQ OAuth 回调未写入 openid/accesstoken")
|
||
login_type = cookies.get("logintype", "QC")
|
||
info = client.request(USER_INFO, headers={
|
||
"Ual-Access-Login-Type": login_type_header(login_type),
|
||
"Ual-Access-Access-Token": cookies["accesstoken"],
|
||
"Ual-Access-Openid": cookies["openid"],
|
||
"Origin": "https://m.yyb.qq.com", "Referer": "https://m.yyb.qq.com/",
|
||
})
|
||
if info.status != 200:
|
||
raise RuntimeError(f"YYB QQ 登录态校验失败: HTTP {info.status}")
|
||
try:
|
||
value = json.loads(info.text)
|
||
except json.JSONDecodeError as exc:
|
||
raise RuntimeError("YYB QQ 登录态校验返回非 JSON") from exc
|
||
if isinstance(value, dict) and value.get("ret") not in (None, 0):
|
||
raise RuntimeError(f"YYB QQ 登录态校验失败: ret={value.get('ret')}")
|
||
write_session(args.session, cookies)
|
||
print(f"QQ 登录成功，cookies 已写入: {args.session}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
raise SystemExit(main())
|
||
except (OSError, TimeoutError, ValueError, RuntimeError) as error:
|
||
print(f"QQ 登录失败: {error}")
|
||
raise SystemExit(1) from error
|