Add Support to Resolve MiyousheGeetestError

genshin/client/components/auth/client.py

@@ -19,7 +19,7 @@
     QRLoginResult,
     WebLoginResult,
 )
-from genshin.models.auth.geetest import MMT, RiskyCheckMMT, RiskyCheckMMTResult, SessionMMT, SessionMMTResult
+from genshin.models.auth.geetest import MMT, MMTResult, RiskyCheckMMT, RiskyCheckMMTResult, SessionMMT, SessionMMTResult
 from genshin.models.auth.qrcode import QRCodeStatus
 from genshin.models.auth.verification import ActionTicket
 from genshin.types import Game
@@ -231,12 +231,12 @@ async def login_with_qrcode(self) -> QRLoginResult:
     async def create_mmt(self) -> MMT:
         """Create a geetest challenge."""
         is_genshin = self.game is Game.GENSHIN
+        ds_headers = ds_utility.get_ds_headers(self.region, params={"is_high": "false"})
         headers = {
-            "DS": ds_utility.generate_create_geetest_ds(),
             "x-rpc-challenge_game": "2" if is_genshin else "6",
             "x-rpc-page": "v4.1.5-ys_#ys" if is_genshin else "v1.4.1-rpg_#/rpg",
             "x-rpc-tool-verison": "v4.1.5-ys" if is_genshin else "v1.4.1-rpg",
-            **auth_utility.CREATE_MMT_HEADERS,
+            **ds_headers,
         }
 
         assert isinstance(self.cookie_manager, managers.CookieManager)
@@ -251,6 +251,56 @@ async def create_mmt(self) -> MMT:
 
         return MMT(**data["data"])
 
+    @base.region_specific(types.Region.CHINESE)
+    @managers.no_multi
+    async def verify_mmt(self, mmt_result: MMTResult) -> None:
+        """Verify a geetest challenge."""
+        is_genshin = self.game is Game.GENSHIN
+        ds_headers = ds_utility.get_ds_headers(self.region, data=mmt_result.get_data())
+        headers = {
+            "x-rpc-challenge_game": "2" if is_genshin else "6",
+            "x-rpc-page": "v4.1.5-ys_#ys" if is_genshin else "v1.4.1-rpg_#/rpg",
+            "x-rpc-tool-verison": "v4.1.5-ys" if is_genshin else "v1.4.1-rpg",
+            **ds_headers,
+        }
+
+        assert isinstance(self.cookie_manager, managers.CookieManager)
+        async with self.cookie_manager.create_session() as session:
+            async with session.post(
+                routes.VERIFY_MMT_URL.get_url(),
+                headers=headers,
+                json=mmt_result.get_data(),
+                cookies=self.cookie_manager.cookies,
+            ) as r:
+                data = await r.json()
+
+        if not data["data"]:
+            errors.raise_for_retcode(data)
+
+    @staticmethod
+    async def generate_device_fp() -> str:
+        """Generate a device fingerprint through the API."""
+        payload = {
+            "device_id": auth_utility.generate_device_fp(length=16),
+            "device_fp": auth_utility.generate_device_fp(),
+            "seed_id": auth_utility.generate_device_id(),
+            "seed_time": str(int(time.time() * 1000)),
+            "app_name": "bbs_cn",
+            "bbs_device_id": auth_utility.generate_device_id(),
+        }
+
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                routes.GET_FP_URL.get_url(),
+                json=payload,
+            ) as resp:
+                data = await resp.json()
+
+        if not data["data"]:
+            errors.raise_for_retcode(data)
+
+        return data["data"]["device_fp"]
+
     @base.region_specific(types.Region.OVERSEAS)
     async def os_game_login(
         self,

genshin/client/components/base.py

@@ -20,6 +20,7 @@
 from genshin.client.manager import managers
 from genshin.models import hoyolab as hoyolab_models
 from genshin.models import model as base_model
+from genshin.models.auth.geetest import MMTResult
 from genshin.utility import concurrency, deprecation, ds
 
 __all__ = ["BaseClient"]
@@ -423,6 +424,7 @@ async def request_hoyolab(
         params: typing.Optional[typing.Mapping[str, typing.Any]] = None,
         data: typing.Any = None,
         headers: typing.Optional[aiohttp.typedefs.LooseHeaders] = None,
+        mmt_result: typing.Optional[MMTResult] = None,
         **kwargs: typing.Any,
     ) -> typing.Mapping[str, typing.Any]:
         """Make a request any hoyolab endpoint."""
@@ -435,6 +437,8 @@ async def request_hoyolab(
         url = routes.TAKUMI_URL.get_url(region).join(yarl.URL(url))
 
         headers = dict(headers or {})
+        if mmt_result is not None:
+            headers["x-rpc-challenge"] = mmt_result.geetest_challenge
         headers.update(ds.get_ds_headers(data=data, params=params, region=region, lang=lang or self.lang))
 
         data = await self.request(url, method=method, params=params, data=data, headers=headers, **kwargs)

genshin/client/components/chronicle/base.py

@@ -10,6 +10,7 @@
 from genshin.client.components import base
 from genshin.client.manager import managers
 from genshin.models import hoyolab as hoyolab_models
+from genshin.models.auth.geetest import MMTResult
 from genshin.utility import deprecation
 
 __all__ = ["BaseBattleChronicleClient"]
@@ -44,6 +45,7 @@ async def request_game_record(
         lang: typing.Optional[str] = None,
         region: typing.Optional[types.Region] = None,
         game: typing.Optional[types.Game] = None,
+        mmt_result: typing.Optional[MMTResult] = None,
         **kwargs: typing.Any,
     ) -> typing.Mapping[str, typing.Any]:
         """Make a request towards the game record endpoint."""
@@ -57,7 +59,7 @@ async def request_game_record(
         mi18n_task = asyncio.create_task(self._fetch_mi18n("bbs", lang=lang or self.lang))
         update_task = asyncio.create_task(utility.update_characters_any(lang or self.lang, lenient=True))
 
-        data = await self.request_hoyolab(url, lang=lang, region=region, **kwargs)
+        data = await self.request_hoyolab(url, lang=lang, region=region, mmt_result=mmt_result, **kwargs)
 
         await mi18n_task
         try:

genshin/client/components/chronicle/genshin.py

@@ -5,6 +5,7 @@
 import typing
 
 from genshin import errors, paginators, types, utility
+from genshin.models.auth.geetest import MMTResult
 from genshin.models.genshin import character as character_models
 from genshin.models.genshin import chronicle as models
 
@@ -25,6 +26,7 @@ async def _request_genshin_record(
         lang: typing.Optional[str] = None,
         payload: typing.Optional[typing.Mapping[str, typing.Any]] = None,
         cache: bool = True,
+        mmt_result: typing.Optional[MMTResult] = None,
     ) -> typing.Mapping[str, typing.Any]:
         """Get an arbitrary honkai object."""
         payload = dict(payload or {})
@@ -57,6 +59,7 @@ async def _request_genshin_record(
             params=params,
             data=data,
             cache=cache_key,
+            mmt_result=mmt_result,
         )
 
     async def get_partial_genshin_user(
@@ -113,10 +116,11 @@ async def get_genshin_notes(
         *,
         lang: typing.Optional[str] = None,
         autoauth: bool = True,
+        mmt_result: typing.Optional[MMTResult] = None,
     ) -> models.Notes:
         """Get genshin real-time notes."""
         try:
-            data = await self._request_genshin_record("dailyNote", uid, lang=lang, cache=False)
+            data = await self._request_genshin_record("dailyNote", uid, lang=lang, cache=False, mmt_result=mmt_result)
         except errors.DataNotPublic as e:
             # error raised only when real-time notes are not enabled
             if uid and (await self._get_uid(types.Game.GENSHIN)) != uid:
@@ -125,7 +129,7 @@ async def get_genshin_notes(
                 raise errors.GenshinException(e.response, "Real-time notes are not enabled.") from e
 
             await self.update_settings(3, True, game=types.Game.GENSHIN)
-            data = await self._request_genshin_record("dailyNote", uid, lang=lang, cache=False)
+            data = await self._request_genshin_record("dailyNote", uid, lang=lang, cache=False, mmt_result=mmt_result)
 
         return models.Notes(**data)
 

genshin/client/manager/managers.py

@@ -156,7 +156,7 @@ async def _request(
         errors.check_for_geetest(data)
 
         if data["retcode"] in MIYOUSHE_GEETEST_RETCODES:
-            raise errors.MiyousheGeetestError(data, {k: morsel.value for k, morsel in response.cookies.items()})
+            raise errors.MiyousheGeetestError(data)
 
         if data["retcode"] == 0:
             return data["data"]

genshin/client/routes.py

@@ -242,6 +242,9 @@ def get_url(self, region: types.Region, game: types.Game) -> yarl.URL:
 CREATE_MMT_URL = Route(
     "https://api-takumi-record.mihoyo.com/game_record/app/card/wapi/createVerification?is_high=false"
 )
+VERIFY_MMT_URL = Route("https://api-takumi-record.mihoyo.com/game_record/app/card/wapi/verifyVerification")
+
+GET_FP_URL = Route("https://public-data-api.mihoyo.com/device-fp/api/getFp")
 
 GAME_RISKY_CHECK_URL = InternationalRoute(
     overseas="https://api-account-os.hoyoverse.com/account/risky/api/check",

genshin/constants.py

@@ -75,3 +75,6 @@
     },
 }
 """App IDs used for game login."""
+
+DEVICE_HEADERS = {"x-rpc-device_id": "441058ee-2c18-4a4b-94f0-1081d12eda92", "x-rpc-device_fp": "38d7f191c78a0"}
+"""Headers used for device information."""

genshin/errors.py

@@ -193,9 +193,7 @@ class MiyousheGeetestError(GenshinException):
     def __init__(
         self,
         response: typing.Dict[str, typing.Any],
-        cookies: typing.Mapping[str, str],
     ) -> None:
-        self.cookies = cookies
         super().__init__(response)
 
     msg = "Geetest triggered during Miyoushe API request."

genshin/utility/auth.py

@@ -3,7 +3,9 @@
 import base64
 import hmac
 import json
+import random
 import typing
+import uuid
 from hashlib import sha256
 
 from genshin import constants
@@ -69,11 +71,6 @@
     "x-rpc-client_type": "2",
 }
 
-CREATE_MMT_HEADERS = {
-    "x-rpc-app_version": "2.60.1",
-    "x-rpc-client_type": "5",
-}
-
 DEVICE_ID = "D6AF5103-D297-4A01-B86A-87F87DS5723E"
 
 RISKY_CHECK_HEADERS = {
@@ -157,3 +154,14 @@ def generate_risky_header(
 ) -> str:
     """Generate risky header for geetest verification."""
     return f"id={check_id};c={challenge};s={validate}|jordan;v={validate}"
+
+
+def generate_device_id() -> str:
+    """Generate a random device ID."""
+    return str(uuid.uuid4()).lower()
+
+
+def generate_device_fp(length: int = 13) -> str:
+    """Generate a random device fingerprint."""
+    characters = "abcdef0123456789"
+    return "".join(random.choices(characters, k=length))

genshin/utility/ds.py

@@ -11,7 +11,6 @@
 
 __all__ = [
     "generate_cn_dynamic_secret",
-    "generate_create_geetest_ds",
     "generate_dynamic_secret",
     "generate_passport_ds",
     "get_ds_headers",
@@ -34,7 +33,7 @@ def generate_cn_dynamic_secret(
 ) -> str:
     """Create a new chinese dynamic secret."""
     t = int(time.time())
-    r = random.randint(100001, 200000)
+    r = random.randint(100000, 200000)
     b = json.dumps(body) if body else ""
     q = "&".join(f"{k}={v}" for k, v in sorted(query.items())) if query else ""
 
@@ -58,7 +57,7 @@ def get_ds_headers(
         }
     elif region == types.Region.CHINESE:
         ds_headers = {
-            "x-rpc-app_version": "2.11.1",
+            "x-rpc-app_version": "2.60.1",
             "x-rpc-client_type": "5",
             "ds": generate_cn_dynamic_secret(data, params),
         }
@@ -76,12 +75,3 @@ def generate_passport_ds(body: typing.Mapping[str, typing.Any]) -> str:
     h = hashlib.md5(f"salt={salt}&t={t}&r={r}&b={b}&q=".encode()).hexdigest()
     result = f"{t},{r},{h}"
     return result
-
-
-def generate_create_geetest_ds() -> str:
-    """Create a dynamic secret for Miyoushe createVerification API endpoint."""
-    salt = constants.DS_SALT[types.Region.CHINESE]
-    t = int(time.time())
-    r = random.randint(100000, 200000)
-    h = hashlib.md5(f"salt={salt}&t={t}&r={r}&b=&q=is_high=false".encode()).hexdigest()
-    return f"{t},{r},{h}"