Add Miyoushe login methods (#167)

* feat: Add CN password login

* refactor: Refactor parse cookie code

* style: Run nox

* feat: Add OTP-related methods

* refactor: Refactor according to reviews

* refactor: Refactor according to reviews

genshin/client/components/geetest/client.py

@@ -39,8 +39,8 @@ async def _web_login(
             headers["x-rpc-aigis"] = geetest_utility.get_aigis_header(session_id, mmt_data)
 
         payload = {
-            "account": geetest_utility.encrypt_geetest_credentials(account),
-            "password": geetest_utility.encrypt_geetest_credentials(password),
+            "account": geetest_utility.encrypt_geetest_credentials(account, self._region),
+            "password": geetest_utility.encrypt_geetest_credentials(password, self._region),
             "token_type": tokenType,
         }
 
@@ -69,6 +69,54 @@ async def _web_login(
 
         return cookies
 
+    async def _cn_login_by_password(
+        self,
+        account: str,
+        password: str,
+        *,
+        geetest: typing.Optional[typing.Dict[str, typing.Any]] = None,
+    ) -> typing.Dict[str, typing.Any]:
+        """
+        Login with account and password using Miyoushe loginByPassword endpoint.
+
+        Returns data from aigis header or cookies.
+        """
+        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 = {
+            "account": geetest_utility.encrypt_geetest_credentials(account, self._region),
+            "password": geetest_utility.encrypt_geetest_credentials(password, self._region),
+        }
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                routes.CN_WEB_LOGIN_URL.get_url(),
+                json=payload,
+                headers=headers,
+            ) as r:
+                data = await r.json()
+
+        if data["retcode"] == -3102:
+            # 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)
+
+        cookies = {cookie.key: cookie.value for cookie in r.cookies.values()}
+
+        self.set_cookies(cookies)
+        return cookies
+
     async def _app_login(
         self,
         account: str,
@@ -95,8 +143,8 @@ async def _app_login(
             headers["x-rpc-verify"] = json.dumps(ticket)
 
         payload = {
-            "account": geetest_utility.encrypt_geetest_credentials(account),
-            "password": geetest_utility.encrypt_geetest_credentials(password),
+            "account": geetest_utility.encrypt_geetest_credentials(account, self._region),
+            "password": geetest_utility.encrypt_geetest_credentials(password, self._region),
         }
 
         async with aiohttp.ClientSession() as session:
@@ -191,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,
@@ -199,7 +322,10 @@ async def login_with_password(
         port: int = 5000,
         tokenType: typing.Optional[int] = 6,
         geetest_solver: typing.Optional[
-            typing.Callable[[typing.Dict[str, typing.Any]], typing.Awaitable[typing.Dict[str, typing.Any]]]
+            typing.Callable[
+                [typing.Dict[str, typing.Any]],
+                typing.Awaitable[typing.Dict[str, typing.Any]],
+            ]
         ] = None,
     ) -> typing.Dict[str, str]:
         """Login with a password via web endpoint.
@@ -220,14 +346,86 @@ async def login_with_password(
 
         return await self._web_login(account, password, tokenType=tokenType, geetest=geetest)
 
+    async def cn_login_by_password(
+        self,
+        account: str,
+        password: str,
+        *,
+        port: int = 5000,
+        geetest_solver: typing.Optional[
+            typing.Callable[
+                [typing.Dict[str, typing.Any]],
+                typing.Awaitable[typing.Dict[str, typing.Any]],
+            ]
+        ] = None,
+    ) -> 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.
+        """
+        result = await self._cn_login_by_password(account, password)
+
+        if "session_id" not in result:
+            # Captcha not triggered
+            return result
+
+        if geetest_solver:
+            geetest = await geetest_solver(result)
+        else:
+            geetest = await server.solve_geetest(result, port=port)
+
+        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,
         password: str,
         *,
         port: int = 5000,
         geetest_solver: typing.Optional[
-            typing.Callable[[typing.Dict[str, typing.Any]], typing.Awaitable[typing.Dict[str, typing.Any]]]
+            typing.Callable[
+                [typing.Dict[str, typing.Any]],
+                typing.Awaitable[typing.Dict[str, typing.Any]],
+            ]
         ] = None,
     ) -> typing.Dict[str, str]:
         """Login with a password via HoYoLab app endpoint.

genshin/client/components/geetest/server.py

@@ -11,13 +11,10 @@
 
 from . import client
 
-__all__ = ["get_page", "launch_webapp", "solve_geetest", "verify_email"]
+__all__ = ["PAGES", "launch_webapp", "solve_geetest", "verify_email"]
 
-
-def get_page(page: typing.Literal["captcha", "verify-email"]) -> str:
-    """Get the HTML page."""
-    return (
-        """
+PAGES: typing.Final[typing.Dict[typing.Literal["captcha", "verify-email", "enter-otp"], str]] = {
+    "captcha": """
     <!DOCTYPE html>
     <html>
       <body></body>
@@ -55,9 +52,8 @@ def get_page(page: typing.Literal["captcha", "verify-email"]) -> str:
         );
       </script>
     </html>
-    """
-        if page == "captcha"
-        else """
+    """,
+    "verify-email": """
     <!DOCTYPE html>
     <html>
       <body>
@@ -76,15 +72,35 @@ def get_page(page: typing.Literal["captcha", "verify-email"]) -> str:
         };
       </script>
     </html>
-    """
-    )
+    """,
+    "enter-otp": """
+    <!DOCTYPE html>
+    <html>
+      <body>
+        <input id="code" type="number">
+        <button id="verify">Send</button>
+      </body>
+      <script>
+        document.getElementById("verify").onclick = () => {
+          fetch("/send-data", {
+            method: "POST",
+            body: JSON.stringify({
+              code: document.getElementById("code").value
+            }),
+          });
+          document.body.innerHTML = "You may now close this window.";
+        };
+      </script>
+    </html>
+    """,
+}
 
 
 GT_URL = "https://raw.githubusercontent.com/GeeTeam/gt3-node-sdk/master/demo/static/libs/gt.js"
 
 
 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,
@@ -95,11 +111,15 @@ async def launch_webapp(
 
     @routes.get("/captcha")
     async def captcha(request: web.Request) -> web.StreamResponse:
-        return web.Response(body=get_page("captcha"), content_type="text/html")
+        return web.Response(body=PAGES["captcha"], content_type="text/html")
 
     @routes.get("/verify-email")
     async def verify_email(request: web.Request) -> web.StreamResponse:
-        return web.Response(body=get_page("verify-email"), content_type="text/html")
+        return web.Response(body=PAGES["verify-email"], content_type="text/html")
+
+    @routes.get("/enter-otp")
+    async def enter_otp(request: web.Request) -> web.StreamResponse:
+        return web.Response(body=PAGES["enter-otp"], content_type="text/html")
 
     @routes.get("/gt.js")
     async def gt(request: web.Request) -> web.StreamResponse:
@@ -162,3 +182,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

genshin/client/routes.py

@@ -13,6 +13,7 @@
     "BBS_REFERER_URL",
     "BBS_URL",
     "CALCULATOR_URL",
+    "CN_WEB_LOGIN_URL",
     "COMMUNITY_URL",
     "COOKIE_V2_REFRESH_URL",
     "DETAIL_LEDGER_URL",
@@ -217,8 +218,13 @@ def get_url(self, region: types.Region, game: types.Game) -> yarl.URL:
 
 WEB_LOGIN_URL = Route("https://sg-public-api.hoyolab.com/account/ma-passport/api/webLoginByPassword")
 APP_LOGIN_URL = Route("https://sg-public-api.hoyoverse.com/account/ma-passport/api/appLoginByPassword")
+CN_WEB_LOGIN_URL = Route("https://passport-api.miyoushe.com/account/ma-cn-passport/web/loginByPassword")
 
 SEND_VERIFICATION_CODE_URL = Route(
     "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")