PaiGramTeam/SIMNet
3
0
Fork 0
mirror of https://github.com/PaiGramTeam/SIMNet.git synced 2024-11-22 14:28:08 +00:00
Code Issues Packages Projects Releases Wiki Activity
1e23a95b08
SIMNet/simnet/client/components/auth.py
xtaodada 5c02e76982
🐛 Fix get all token mid v2'
2024-05-29 15:50:43 +08:00

691 lines
28 KiB
Python
Raw Blame History

import json as jsonlib
import time
from typing import Optional, Tuple, Union
from simnet.client.base import BaseClient
from simnet.client.cookies import CookiesModel
from simnet.client.routes import (
AUTH_URL,
AUTH_KEY_URL,
HK4E_LOGIN_URL,
PASSPORT_URL,
WEB_ACCOUNT_URL,
QRCODE_URL,
URL,
GET_FP_URL,
)
from simnet.errors import RegionNotSupported
from simnet.utils.enums import Region
from simnet.utils.hex import get_random_hex_string_of_length
from simnet.utils.player import recognize_game_biz, recognize_server
__all__ = ("AuthClient",)
class AuthClient(BaseClient):
"""
The AuthClient class is a client for authentication services.
It is derived from the BaseClient class and provides methods for retrieving
different authentication tokens and keys.
"""
async def get_stoken_by_login_ticket(
self, login_ticket: Optional[str] = None, account_id: Optional[int] = None
) -> Optional[str]:
"""
Retrieves a super ticket (`stoken`) using a login ticket (`login_ticket`) .
Args:
login_ticket (Optional[str]): The login ticket to use to retrieve the super ticket. If not provided, the
`login_ticket` cookie value will be used.
account_id (Optional[int]): The account ID to use to retrieve the super ticket. If not provided, the
`account_id` attribute value will be used.
Returns:
Optional[str]: The retrieved super ticket (`stoken`).
Raises:
ValueError: If the `login_ticket` argument is `None`, or if the `account_id` argument is `None`.
"""
url = AUTH_URL.get_url(self.region) / "getMultiTokenByLoginTicket"
login_ticket = login_ticket or self.cookies.get("login_ticket")
account_id = account_id or self.account_id
if login_ticket is None:
raise ValueError("The 'login_ticket' argument cannot be None.")
if account_id is None:
raise ValueError("The 'account_id' argument cannot be None.")
params = {
"login_ticket": login_ticket,
"uid": account_id,
"token_types": 3,
}
data = await self.request_lab(url, params=params)
res_data = data["list"]
for i in res_data:
name = i.get("name")
token = i.get("token")
if name and token:
self.cookies[name] = token
stoken = self.cookies.get("stoken")
stuid = self.cookies.get("stuid")
if stoken and stuid:
self.cookies["stuid"] = str(self.account_id)
return stoken
async def get_cookie_token_by_login_ticket(self, login_ticket: Optional[str] = None) -> Optional[str]:
"""
Retrieves a cookie token (`cookie_token`) using a login ticket (`login_ticket`).
Args:
login_ticket (Optional[str]): The login ticket to use to retrieve the cookie token. If not provided, the
`login_ticket` cookie value will be used.
Returns:
Optional[str]: The retrieved cookie token (`cookie_token`).
Raises:
ValueError: If the `login_ticket` argument is `None`.
"""
url = WEB_ACCOUNT_URL.get_url(self.region) / "cookie_accountinfo_by_loginticket"
login_ticket = login_ticket or self.cookies.get("login_ticket")
if login_ticket is None:
raise ValueError("The 'login_ticket' argument cannot be None.")
params = {"login_ticket": login_ticket}
data = await self.request_lab(url, params=params)
cookie_info = data.get("cookie_info")
if not cookie_info:
raise ValueError("The 'login_ticket' is expired.")
account_id = cookie_info.get("account_id")
cookie_token = cookie_info.get("cookie_token")
if account_id:
self.account_id = account_id
self.cookies["account_id"] = str(account_id)
if cookie_token:
self.cookies["cookie_token"] = cookie_token
return cookie_token
def check_stoken(
self,
stoken: Optional[str] = None,
account_id: Optional[int] = None,
mid: Optional[str] = None,
) -> None:
"""
Checks and sets the stoken, account_id, and mid for a user.
This function retrieves the stoken, account_id, and mid from the provided arguments
or falls back to using existing properties/cookies. It then validates the presence of
'stoken' and 'account_id'. If the stoken starts with 'v2_', the presence of 'mid' is
also validated. Finally, the stoken, account_id, and potentially the mid are set to
the cookies.
Args:
stoken (str, optional): The stoken to check and set. Defaults to the stoken cookie.
account_id (int, optional): The account ID to check and set. Defaults to self.account_id.
mid (str, optional): The mid to check and set. Defaults to the mid cookie.
Raises:
ValueError: If the stoken, account_id, or mid (when stoken starts with 'v2_') is not provided.
Returns:
None: This function modifies instance properties and does not return a value.
"""
stoken = stoken or self.cookies.get("stoken")
account_id = account_id or self.account_id
mid = mid or self.cookies.get("mid")
if stoken is None:
raise ValueError("The 'stoken' argument cannot be None.")
if account_id is None:
raise ValueError("The 'account_id' argument cannot be None.")
if stoken.startswith("v2_"):
if mid is None:
raise ValueError("The 'mid' argument cannot be None.")
self.cookies.set("mid", mid)
self.cookies.set("stuid", str(account_id))
self.cookies.set("stoken", stoken)
async def get_cookie_token_by_stoken(
self,
stoken: Optional[str] = None,
account_id: Optional[int] = None,
mid: Optional[str] = None,
) -> Optional[str]:
"""
Retrieves a cookie token (`cookie_token`) using a super ticket (`stoken`).
Args:
stoken (Optional[str]): The super ticket to use to retrieve the cookie token. If not provided, the
`stoken` cookie value will be used.
account_id (Optional[int]): The account ID to use to retrieve the cookie token. If not provided, the
`account_id` attribute value will be used.
mid (Optional[str]): The machine ID to use to retrieve the cookie token. If not provided, the `mid`
attribute value will be used.
Returns:
Optional[str]: The retrieved cookie token (`cookie_token`).
Raises:
ValueError: If the `login_ticket` argument is `None`, or if the `account_id` argument is `None`.
"""
self.check_stoken(stoken, account_id, mid)
url = PASSPORT_URL.get_url(self.region) / "getCookieAccountInfoBySToken"
method = "GET" if self.region == Region.CHINESE else "POST"
data = await self.request_lab(url, method=method)
cookie_token = data.get("cookie_token")
if cookie_token:
self.cookies["cookie_token"] = cookie_token
self.cookies["account_id"] = str(self.account_id)
return cookie_token
async def get_ltoken_by_stoken(
self,
stoken: Optional[str] = None,
account_id: Optional[int] = None,
mid: Optional[str] = None,
) -> Optional[str]:
"""
Retrieves a login token (`ltoken`) using a super ticket (`stoken`).
Args:
stoken (Optional[str]): The super ticket to use to retrieve the cookie token. If not provided, the
`stoken` cookie value will be used.
account_id (Optional[int]): The account ID to use to retrieve the cookie token. If not provided, the
`account_id` attribute value will be used.
mid (Optional[str]): The machine ID to use to retrieve the cookie token. If not provided, the `mid`
attribute value will be used.
Returns:
Optional[str]: The retrieved cookie token (`cookie_token`).
Raises:
ValueError: If the `login_ticket` argument is `None`, or if the `account_id` argument is `None`.
"""
self.check_stoken(stoken, account_id, mid)
url = PASSPORT_URL.get_url(self.region) / "getLTokenBySToken"
method = "GET" if self.region == Region.CHINESE else "POST"
data = await self.request_lab(url, method=method)
ltoken = data.get("ltoken", "")
if ltoken:
self.cookies["ltoken"] = ltoken
self.cookies["ltuid"] = str(self.account_id)
return ltoken
async def get_authkey_by_stoken(
self,
auth_appid: str,
game_biz: Optional[str] = None,
region: Optional[str] = None,
) -> Optional[str]:
"""
Get the auth key (`authkey`) for a game and region using a super ticket (`stoken`).
Args:
auth_appid (str): The type of application for which the authkey is being requested.
For example, to request wish records, use `webview_gacha`.
game_biz (Optional[str]): The name of the game.
region (Optional[str]): The region in which the game is registered.
Returns:
Optional[str]: The authentication key, or None if not found.
Raises:
ValueError: If `stoken` is not found in the cookies or `player_id` not found.
"""
self.check_stoken()
if not self.player_id:
raise ValueError("player_id not found.")
game_biz = game_biz or recognize_game_biz(self.player_id, self.game)
region = region or recognize_server(self.player_id, self.game)
url = AUTH_KEY_URL.get_url(self.region)
json = {
"auth_appid": auth_appid,
"game_biz": game_biz,
"game_uid": self.player_id,
"region": region,
}
data = await self.request_lab(url, data=json)
return data.get("authkey")
async def get_hk4e_token_by_cookie_token(
self,
game_biz: Optional[str] = None,
region: Optional[str] = None,
player_id: Optional[int] = None,
) -> None:
"""
Get HK4E token (`hk4e_token`) using cookie token (`cookie_token`).
The resulting HK4E token will be automatically saved in self.cookies.
Args:
game_biz (Optional[str]): The name of the game.
region (Optional[str]): The region in which the game is registered.
player_id (Optional[int]): The player ID to use to retrieve the HK4E token. If not provided, the `player_id`
attribute value will be used.
Raises:
ValueError: If `cookie_token` is not found in the cookies.
"""
cookie_token = self.cookies.get("cookie_token")
if cookie_token is None:
raise ValueError("cookie_token not found in cookies.")
uid = self.player_id or player_id
if not uid:
raise ValueError("player_id not found.")
game_biz = game_biz or recognize_game_biz(uid, self.game)
region = region or recognize_server(uid, self.game)
url = HK4E_LOGIN_URL.get_url(self.region)
json = {
"game_biz": game_biz,
"uid": uid,
"region": region,
"lang": self.lang,
}
await self.request_api("POST", url=url, json=json)
async def gen_login_qrcode(
self,
app_id: str = "8",
) -> Tuple[str, str]:
"""
Generate login qrcode and return url and ticket
Args:
app_id (str): The app id to use to generate the qrcode.
If not provided, the `app_id` attribute value will be used.
Returns:
Tuple[str, str]: The url and ticket of the qrcode.
"""
if self.region != Region.CHINESE:
raise RegionNotSupported()
data = {"app_id": app_id, "device": self.get_device_id()}
res_json = await self.request_api("POST", url=QRCODE_URL / "fetch", json=data)
url = res_json.get("url", "")
if not url:
return "", ""
ticket = url.split("ticket=")[1]
return url, ticket
async def check_login_qrcode(self, ticket: str, app_id: str = "8") -> Union[bool, str]:
"""
Check login qrcode and return token if success
Args:
ticket (str): The ticket of the qrcode.
app_id (str): The app id to use to generate the qrcode. If not provided,
the `app_id` attribute value will be used.
Returns:
Union[bool, str]: The token of the qrcode if success, else False.
"""
if self.region != Region.CHINESE:
raise RegionNotSupported()
data = {"app_id": app_id, "ticket": ticket, "device": self.get_device_id()}
res_data = await self.request_api("POST", url=QRCODE_URL / "query", json=data)
if res_data.get("stat", "") != "Confirmed":
return False
info = jsonlib.loads(res_data.get("payload", {}).get("raw", "{}"))
self.account_id = int(info.get("uid", 0))
return info.get("token", "")
async def accept_login_qrcode(self, url: str) -> None:
"""
Accept login qrcode
Args:
url (str): The url of the qrcode.
Returns:
None
"""
if self.region != Region.CHINESE:
raise RegionNotSupported()
self.check_stoken()
if not url.startswith("https://user.mihoyo.com/qr_code_in_game.html"):
raise ValueError("Invalid url")
u = URL(url)
ticket = u.params.get("ticket")
app_id = u.params.get("app_id")
biz_key = u.params.get("biz_key")
scan_url = (QRCODE_URL / "scan").replace("hk4e_cn", biz_key)
data = {"ticket": ticket, "app_id": app_id, "device": self.get_device_id()}
await self.request_lab(url=scan_url, data=data)
game_token = await self.get_game_token_by_stoken()
raw = jsonlib.dumps({"uid": str(self.account_id), "token": game_token}, indent=4, ensure_ascii=False)
data["payload"] = {
"proto": "Account",
"raw": raw,
}
confirm_url = (QRCODE_URL / "confirm").replace("hk4e_cn", biz_key)
await self.request_lab(url=confirm_url, data=data)
async def get_stoken_v2_and_mid_by_game_token(self, game_token: str) -> Tuple[str, str]:
"""
Get stoken_v2 and mid by game token
Args:
game_token (str): The game token to use to retrieve the stoken_v2 and mid.
Returns:
Tuple[str, str]: The stoken_v2 and mid.
"""
if self.region != Region.CHINESE:
raise RegionNotSupported()
url = PASSPORT_URL.get_url(self.region) / "../../ma-cn-session/app/getTokenByGameToken"
data = {
"account_id": self.account_id,
"game_token": game_token,
}
headers = {"x-rpc-app_id": "bll8iq97cem8"}
data = await self.request_lab(url, data=data, headers=headers)
mid = data.get("user_info", {}).get("mid", "")
stoken_v2 = data.get("token", {}).get("token", "")
self.cookies.set("mid", mid)
self.cookies.set("stoken", stoken_v2)
return stoken_v2, mid
async def get_stoken_v2_and_mid_by_by_stoken(
self,
stoken: Optional[str] = None,
account_id: Optional[int] = None,
) -> Tuple[str, str]:
"""
Get stoken_v2 and mid by stoken_v1
Args:
stoken (Optional[str]): The stoken_v1 to use to retrieve the stoken_v2 and mid.
If not provided, the `stoken` attribute value will be used.
account_id (Optional[int]): The account ID to use to retrieve the stoken_v2 and mid.
If not provided, the `account_id` attribute value will be used.
Returns:
Tuple[str, str]: The stoken_v2 and mid.
"""
if self.region != Region.CHINESE:
raise RegionNotSupported()
self.check_stoken(stoken, account_id)
url = PASSPORT_URL.get_url(self.region) / "../../ma-cn-session/app/getTokenBySToken"
headers = {"x-rpc-app_id": "bll8iq97cem8"}
data = await self.request_lab(url, method="POST", headers=headers)
mid = data.get("user_info", {}).get("mid", "")
stoken_v2 = data.get("token", {}).get("token", "")
self.cookies.set("mid", mid)
self.cookies.set("stoken", stoken_v2)
return stoken_v2, mid
async def get_all_token_by_stoken(
self,
stoken: Optional[str] = None,
account_id: Optional[int] = None,
mid: Optional[str] = None,
) -> CookiesModel:
"""
Get stoken_v2, mid, ltoken and cookie_token by stoken v1 or v2
Args:
stoken (Optional[str]): The stoken_v1 to use to retrieve the stoken_v2 and mid.
If not provided, the `stoken` attribute value will be used.
account_id (Optional[int]): The account ID to use to retrieve the stoken_v2 and mid.
If not provided, the `account_id` attribute value will be used.
mid (Optional[str]): The mid to use to retrieve the stoken_v2 and mid.
Returns:
CookiesModel: The stoken_v2, mid, ltoken and cookie_token.
"""
if self.region != Region.OVERSEAS:
raise RegionNotSupported()
self.check_stoken(stoken, account_id, mid)
account_id = account_id or self.account_id
url = AUTH_KEY_URL.get_url(self.region) / "../../../account/ma-passport/token/getBySToken"
headers = {"x-rpc-app_id": "c9oqaq3s3gu8"}
data_ = {"dst_token_types": [1, 2, 4]}
data = await self.request_lab(url, method="POST", headers=headers, data=data_)
tokens = data.get("tokens", [])
token_map = {1: "", 2: "", 4: ""}
for token in tokens:
token_type = token.get("token_type", 0)
token_map[token_type] = token.get("token", "")
mid = data.get("user_info", {}).get("mid", "")
model = CookiesModel(stoken=token_map[1])
ltoken, cookie_token = token_map[2], token_map[4]
if ltoken.startswith("v2_"):
model.ltoken_v2 = ltoken
else:
model.ltoken = ltoken
if cookie_token.startswith("v2_"):
model.cookie_token_v2 = cookie_token
else:
model.cookie_token = cookie_token
model.set_mid(mid)
model.set_uid(account_id)
model.set_v2_uid(account_id)
self.cookies.set("mid", model.mid)
self.cookies.set("stoken", model.stoken)
if model.ltoken:
self.cookies.set("ltoken", model.ltoken)
if model.ltoken_v2:
self.cookies.set("ltoken_v2", model.ltoken_v2)
self.cookies.set("ltmid_v2", model.ltmid_v2)
if model.cookie_token:
self.cookies.set("cookie_token", model.cookie_token)
if model.cookie_token_v2:
self.cookies.set("cookie_token_v2", model.cookie_token_v2)
self.cookies.set("account_mid_v2", model.account_mid_v2)
return model
async def get_game_token_by_stoken(
self,
stoken: Optional[str] = None,
account_id: Optional[int] = None,
mid: Optional[str] = None,
) -> str:
"""
Get game token by stoken
Args:
stoken (Optional[str]): The stoken_v1 to use to retrieve the game token.
If not provided, the `stoken` attribute value will be used.
account_id (Optional[int]): The account ID to use to retrieve the game token.
If not provided, the `account_id` attribute value will be used.
mid (Optional[str]): The mid to use to retrieve the game token.
If not provided, the `mid` attribute value will be used.
Returns:
str: The game token.
"""
if self.region != Region.CHINESE:
raise RegionNotSupported()
self.check_stoken(stoken, account_id, mid)
url = AUTH_URL.get_url(self.region) / "getGameToken"
data = await self.request_lab(url, method="GET")
return data.get("game_token", "")
async def verify_stoken(
self,
stoken: Optional[str] = None,
account_id: Optional[int] = None,
mid: Optional[str] = None,
) -> None:
"""
Verify stoken
Args:
stoken (Optional[str]): The stoken_v1 to use to retrieve the stoken_v2 and mid.
If not provided, the `stoken` attribute value will be used.
account_id (Optional[int]): The account ID to use to retrieve the stoken_v2 and mid.
If not provided, the `account_id` attribute value will be used.
mid (Optional[str]): The mid to use to retrieve the stoken_v2 and mid.
Returns:
None
Raises:
ValueError: If the `stoken` argument is `None`, or if the `account_id` argument is `None`.
InvalidCookies: If the stoken is invalid.
"""
self.check_stoken(stoken, account_id, mid)
if self.region == Region.OVERSEAS:
url = AUTH_KEY_URL.get_url(self.region) / "../../../account/ma-passport/token/verifySToken"
headers = {"x-rpc-app_id": "c9oqaq3s3gu8"}
await self.request_lab(url, method="POST", headers=headers)
else:
await self.get_cookie_token_by_stoken(stoken, account_id, mid)
async def verify_ltoken(
self,
ltoken: Optional[str] = None,
ltuid: Optional[int] = None,
) -> None:
"""
Verify ltoken
Args:
ltoken (Optional[str]): The ltoken to use to verify.
If not provided, the `ltoken` cookie value will be used.
ltuid (Optional[int]): The account ID to use to verify.
If not provided, the `ltuid` cookie value will be used.
Returns:
None
Raises:
ValueError: If the `ltoken` argument is `None`, or if the `account_id` argument is `None`.
InvalidCookies: If the ltoken is invalid.
"""
ltoken = ltoken or self.cookies.get("ltoken")
ltuid = ltuid or self.account_id
if ltoken is None:
raise ValueError("The 'ltoken' argument cannot be None.")
if ltuid is None:
raise ValueError("The 'account_id' argument cannot be None.")
self.cookies.set("ltoken", ltoken)
self.cookies.set("ltuid", str(ltuid))
if self.region == Region.OVERSEAS:
url = AUTH_KEY_URL.get_url(self.region) / "../../../account/ma-passport/token/verifyLToken"
headers = {"x-rpc-app_id": "c9oqaq3s3gu8"}
data = None
else:
url = PASSPORT_URL.get_url(self.region) / "getUserAccountInfoByLToken"
headers = {"x-rpc-app_id": "bll8iq97cem8"}
data = {
"ltoken": ltoken,
"uid": ltuid,
}
await self.request_lab(url, method="POST", headers=headers, data=data)
async def verify_cookie_token(
self,
cookie_token: Optional[str] = None,
account_id: Optional[int] = None,
) -> None:
"""
Verify cookie token
Args:
cookie_token (Optional[str]): The cookie token to use to verify.
If not provided, the `cookie_token` cookie value will be used.
account_id (Optional[int]): The account ID to use to verify.
If not provided, the `account_id` cookie value will be used.
Returns:
None
Raises:
ValueError: If the `cookie_token` argument is `None`, or if the `account_id` argument is `None`.
InvalidCookies: If the cookie_token is invalid.
"""
cookie_token = cookie_token or self.cookies.get("cookie_token")
account_id = account_id or self.account_id
if cookie_token is None:
raise ValueError("The 'cookie_token' argument cannot be None.")
if account_id is None:
raise ValueError("The 'account_id' argument cannot be None.")
self.cookies.set("cookie_token", cookie_token)
self.cookies.set("account_id", str(account_id))
if self.region == Region.OVERSEAS:
url = AUTH_KEY_URL.get_url(self.region) / "../../../account/ma-passport/token/verifyCookieToken"
headers = {"x-rpc-app_id": "c9oqaq3s3gu8"}
data = None
else:
url = PASSPORT_URL.get_url(self.region) / "getUserAccountInfoByCookieToken"
headers = {"x-rpc-app_id": "bll8iq97cem8"}
data = {
"cookie_token": cookie_token,
"uid": account_id,
}
await self.request_lab(url, method="POST", headers=headers, data=data)
async def get_fp(
self,
device_id: Optional[str] = None,
device_fp: Optional[str] = None,
extend_properties: Optional[dict] = None,
app_name: str = "bbs_cn",
platform: int = 2,
) -> str:
"""
Get Device Fingerprint
Args:
device_id (Optional[str]): Device ID, if not provided, use `get_device_id()`.
device_fp (Optional[int]): Device fingerprint, if not provided, use `get_device_fp()`.
extend_properties (Optional[dict]): Device extension information,
defaults to auto-generated if not provided.
app_name (str): APP name, defaults to "bbs_cn" if not provided.
platform (int): Device platform code, same as client_type in the DS algorithm.
Returns:
str: The device fingerprint.
"""
seed_time = int(time.time() * 1000)
seed_id = get_random_hex_string_of_length(13)
if extend_properties is None:
model = get_random_hex_string_of_length(6)
extend_properties = {
"cpuType": "arm64-v8a",
"romCapacity": "512",
"productName": model,
"romRemain": "256",
"manufacturer": "XiaoMi",
"appMemory": "512",
"hostname": "dg02-pool03-kvm87",
"screenSize": "1080x1920",
"osVersion": "13",
"vendor": "中国移动",
"accelerometer": "1.4883357x7.1712894x6.2847486",
"buildTags": "release-keys",
"model": model,
"brand": "XiaoMi",
"oaid": "",
"hardware": "qcom",
"deviceType": "OP5913L1",
"devId": "REL",
"serialNumber": "unknown",
"buildTime": "1687848011000",
"buildUser": "root",
"ramCapacity": "469679",
"magnetometer": "20.081251x-27.487501x2.1937501",
"display": f"{model}_13.1.0.181(CN01)",
"ramRemain": "215344",
"deviceInfo": f"XiaoMi/{model}/OP5913L1:13/SKQ1.221119.001/T.118e6c7-5aa23-73911:user/release-keys",
"gyroscope": "0.030226856x0.014647375x0.010652636",
"vaid": "",
"buildType": "user",
"sdkVersion": "33",
"board": "taro",
}
ext_fields = jsonlib.dumps(extend_properties)
data = {
"app_name": app_name,
"device_fp": device_fp or self.get_device_fp(),
"device_id": device_id or self.get_device_fp(),
"ext_fields": ext_fields,
"platform": platform,
"seed_id": seed_id,
"seed_time": seed_time,
}
new_device_fp = await self.request_lab(GET_FP_URL, method="POST", data=data)
self.device_fp = new_device_fp
return new_device_fp
Reference in New Issue View Git Blame Copy Permalink