From 73daf08a347ecb7711605e2e04e49fcbd4f93aa7 Mon Sep 17 00:00:00 2001 From: seriaati Date: Mon, 25 Mar 2024 22:35:57 +0800 Subject: [PATCH] feat: Add OTP-related methods --- genshin/client/components/geetest/client.py | 117 +++++++++++++++++++- genshin/client/components/geetest/server.py | 16 ++- genshin/client/routes.py | 4 + genshin/errors.py | 23 +++- 4 files changed, 153 insertions(+), 7 deletions(-) diff --git a/genshin/client/components/geetest/client.py b/genshin/client/components/geetest/client.py index 318b474d..d8a865a5 100644 --- a/genshin/client/components/geetest/client.py +++ b/genshin/client/components/geetest/client.py @@ -239,6 +239,81 @@ async def _verify_email(self, code: str, ticket: typing.Dict[str, typing.Any]) - return None + async def _send_mobile_otp( + self, + mobile: str, + *, + geetest: typing.Optional[typing.Dict[str, typing.Any]] = None, + ) -> typing.Dict[str, typing.Any] | None: + """Attempt to send OTP to the provided mobile number. + + May return aigis headers if captcha is triggered, None otherwise. + """ + headers = { + **geetest_utility.CN_LOGIN_HEADERS, + "ds": ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"]), + } + if geetest: + mmt_data = geetest["data"] + session_id = geetest["session_id"] + headers["x-rpc-aigis"] = geetest_utility.get_aigis_header(session_id, mmt_data) + + payload = { + "mobile": geetest_utility.encrypt_geetest_credentials(mobile, self._region), + "area_code": geetest_utility.encrypt_geetest_credentials("+86", self._region), + } + + async with aiohttp.ClientSession() as session: + async with session.post( + routes.MOBILE_OTP_URL.get_url(), + json=payload, + headers=headers, + ) as r: + data = await r.json() + + if data["retcode"] == -3101: + # Captcha triggered + aigis = json.loads(r.headers["x-rpc-aigis"]) + aigis["data"] = json.loads(aigis["data"]) + return aigis + + if not data["data"]: + errors.raise_for_retcode(data) + + return None + + async def _login_with_mobile_otp(self, mobile: str, otp: str) -> typing.Dict[str, typing.Any]: + """Login with OTP and mobile number. + + Returns cookies if OTP matches the one sent, raises an error otherwise. + """ + headers = { + **geetest_utility.CN_LOGIN_HEADERS, + "ds": ds_utility.generate_dynamic_secret(constants.DS_SALT["cn_signin"]), + } + + payload = { + "mobile": geetest_utility.encrypt_geetest_credentials(mobile, self._region), + "area_code": geetest_utility.encrypt_geetest_credentials("+86", self._region), + "captcha": otp, + } + + async with aiohttp.ClientSession() as session: + async with session.post( + routes.MOBILE_LOGIN_URL.get_url(), + json=payload, + headers=headers, + ) as r: + data = await r.json() + + if not data["data"]: + errors.raise_for_retcode(data) + + cookies = {cookie.key: cookie.value for cookie in r.cookies.values()} + self.set_cookies(cookies) + + return cookies + async def login_with_password( self, account: str, @@ -286,8 +361,7 @@ async def cn_login_by_password( ) -> typing.Dict[str, str]: """Login with a password via Miyoushe loginByPassword endpoint. - Note that this will start a webserver if captcha is - triggered and `geetest_solver` is not passed. + Note that this will start a webserver if captcha is triggered and `geetest_solver` is not passed. """ result = await self._cn_login_by_password(account, password) @@ -302,6 +376,45 @@ async def cn_login_by_password( return await self._cn_login_by_password(account, password, geetest=geetest) + async def check_mobile_number_validity(self, mobile: str) -> bool: + """Check if a mobile number is valid (it's registered on Miyoushe). + + Returns True if the mobile number is valid, False otherwise. + """ + async with aiohttp.ClientSession() as session: + async with session.get( + routes.CHECK_MOBILE_VALIDITY_URL.get_url(), + params={"mobile": mobile}, + ) as r: + data = await r.json() + + return data["data"]["status"] != data["data"]["is_registable"] + + async def login_with_mobile_number( + self, + mobile: str, + ) -> typing.Dict[str, str]: + """Login with mobile number, returns cookies. + + Only works for Chinese region (Miyoushe) users, do not include area code (+86) in the mobile number. + Steps: + 1. Sends OTP to the provided mobile number. + 1-1. If captcha is triggered, prompts the user to solve it. + 2. Lets user enter the OTP. + 3. Logs in with the OTP. + 4. Returns cookies. + """ + result = await self._send_mobile_otp(mobile) + + if result is not None and "session_id" in result: + # Captcha triggered + geetest = await server.solve_geetest(result) + await self._send_mobile_otp(mobile, geetest=geetest) + + otp = await server.enter_otp() + cookies = await self._login_with_mobile_otp(mobile, otp) + return cookies + async def login_with_app_password( self, account: str, diff --git a/genshin/client/components/geetest/server.py b/genshin/client/components/geetest/server.py index 6f69ae8e..8b9f5205 100644 --- a/genshin/client/components/geetest/server.py +++ b/genshin/client/components/geetest/server.py @@ -14,7 +14,7 @@ __all__ = ["get_page", "launch_webapp", "solve_geetest", "verify_email"] -def get_page(page: typing.Literal["captcha", "verify-email"]) -> str: +def get_page(page: typing.Literal["captcha", "verify-email", "enter-otp"]) -> str: """Get the HTML page.""" return ( """ @@ -84,7 +84,7 @@ def get_page(page: typing.Literal["captcha", "verify-email"]) -> str: async def launch_webapp( - page: typing.Literal["captcha", "verify-email"], + page: typing.Literal["captcha", "verify-email", "enter-otp"], *, port: int = 5000, mmt: typing.Optional[typing.Dict[str, typing.Any]] = None, @@ -101,6 +101,10 @@ async def captcha(request: web.Request) -> web.StreamResponse: async def verify_email(request: web.Request) -> web.StreamResponse: return web.Response(body=get_page("verify-email"), content_type="text/html") + @routes.get("/enter-otp") + async def enter_otp(request: web.Request) -> web.StreamResponse: + return web.Response(body=get_page("enter-otp"), content_type="text/html") + @routes.get("/gt.js") async def gt(request: web.Request) -> web.StreamResponse: async with aiohttp.ClientSession() as session: @@ -162,3 +166,11 @@ async def verify_email( code = data["code"] return await client._verify_email(code, ticket) + + +async def enter_otp(port: int = 5000) -> str: + """Lets user enter the OTP.""" + # The enter-otp page is the same as verify-email page. + data = await launch_webapp("enter-otp", port=port) + code = data["code"] + return code diff --git a/genshin/client/routes.py b/genshin/client/routes.py index ca103bf3..05943ded 100644 --- a/genshin/client/routes.py +++ b/genshin/client/routes.py @@ -224,3 +224,7 @@ def get_url(self, region: types.Region, game: types.Game) -> yarl.URL: "https://sg-public-api.hoyoverse.com/account/ma-verifier/api/createEmailCaptchaByActionTicket" ) VERIFY_EMAIL_URL = Route("https://sg-public-api.hoyoverse.com/account/ma-verifier/api/verifyActionTicketPartly") + +CHECK_MOBILE_VALIDITY_URL = Route("https://webapi.account.mihoyo.com/Api/is_mobile_registrable") +MOBILE_OTP_URL = Route("https://passport-api.miyoushe.com/account/ma-cn-verifier/verifier/createLoginCaptcha") +MOBILE_LOGIN_URL = Route("https://passport-api.miyoushe.com/account/ma-cn-passport/web/loginByMobileCaptcha") diff --git a/genshin/errors.py b/genshin/errors.py index 1ea5e1e2..f62ade2f 100644 --- a/genshin/errors.py +++ b/genshin/errors.py @@ -31,7 +31,11 @@ class GenshinException(Exception): original: str = "" msg: str = "" - def __init__(self, response: typing.Mapping[str, typing.Any] = {}, msg: typing.Optional[str] = None) -> None: + def __init__( + self, + response: typing.Mapping[str, typing.Any] = {}, + msg: typing.Optional[str] = None, + ) -> None: self.retcode = response.get("retcode", self.retcode) self.original = response.get("message", "") self.msg = msg or self.msg or self.original @@ -176,6 +180,12 @@ class AccountHasLocked(GenshinException): msg = "Account has been locked because exceeded password limit. Please wait 20 minute and try again" +class WrongOTP(GenshinException): + """Wrong OTP code.""" + + msg = "The provided OTP code is wrong." + + _TGE = typing.Type[GenshinException] _errors: typing.Dict[int, typing.Union[_TGE, str, typing.Tuple[_TGE, typing.Optional[str]]]] = { # misc hoyolab @@ -189,7 +199,10 @@ class AccountHasLocked(GenshinException): # database game record 10101: TooManyRequests, 10102: DataNotPublic, - 10103: (InvalidCookies, "Cookies are valid but do not have a hoyolab account bound to them."), + 10103: ( + InvalidCookies, + "Cookies are valid but do not have a hoyolab account bound to them.", + ), 10104: "Cannot view real-time notes of other users.", # calculator -500001: "Invalid fields in calculation.", @@ -210,7 +223,10 @@ class AccountHasLocked(GenshinException): -2016: RedemptionCooldown, -2017: RedemptionClaimed, -2018: RedemptionClaimed, - -2021: (RedemptionException, "Cannot claim codes for accounts with adventure rank lower than 10."), + -2021: ( + RedemptionException, + "Cannot claim codes for accounts with adventure rank lower than 10.", + ), # rewards -5003: AlreadyClaimed, # chinese @@ -219,6 +235,7 @@ class AccountHasLocked(GenshinException): # account -3208: AccountLoginFail, -3202: AccountHasLocked, + -3205: WrongOTP, } ERRORS: typing.Dict[int, typing.Tuple[_TGE, typing.Optional[str]]] = {