diff --git a/genshin/__main__.py b/genshin/__main__.py index 05134196..629ed987 100644 --- a/genshin/__main__.py +++ b/genshin/__main__.py @@ -228,7 +228,7 @@ async def lineups(client: genshin.Client, scenario: typing.Optional[str]) -> Non click.echo(f"{click.style('Characters:', bold=True)}") for group, characters in enumerate(lineup.characters, 1): if len(lineup.characters) > 1: - click.echo(f"- Group {group }:") + click.echo(f"- Group {group}:") for character in characters: click.echo(f" - {character.name} ({character.role})") diff --git a/genshin/client/components/geetest/client.py b/genshin/client/components/geetest/client.py index 8e3a14a6..898a4d23 100644 --- a/genshin/client/components/geetest/client.py +++ b/genshin/client/components/geetest/client.py @@ -5,7 +5,7 @@ import aiohttp import aiohttp.web -from genshin import errors, constants +from genshin import constants, errors from genshin.client import routes from genshin.client.components import base from genshin.utility import ds as ds_utility @@ -18,32 +18,31 @@ class GeetestClient(base.BaseClient): """Geetest client component.""" - + async def web_login( self, - account: str, + account: str, password: str, *, tokenType: typing.Optional[int] = 6, - geetest: typing.Optional[typing.Dict[str, typing.Any]] = None, + geetest: typing.Optional[typing.Dict[str, typing.Any]] = None, ) -> typing.Dict[str, typing.Any]: - """Login with a password and a solved geetest (if triggered) - using web endpoint. - + """Login with a password using web endpoint. + Returns either data from aigis header or cookies. """ - headers = { **geetest_utility.WEB_LOGIN_HEADERS } + headers = {**geetest_utility.WEB_LOGIN_HEADERS} 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), "password": geetest_utility.encrypt_geetest_credentials(password), "token_type": tokenType, } - + async with aiohttp.ClientSession() as session: async with session.post( routes.WEB_LOGIN_URL.get_url(), @@ -52,23 +51,23 @@ async def web_login( ) as r: data = await r.json() cookies = {cookie.key: cookie.value for cookie in r.cookies.values()} - + 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) - + if data["data"].get("stoken"): cookies["stoken"] = data["data"]["stoken"] self.set_cookies(cookies) return cookies - + async def app_login( self, account: str, @@ -76,12 +75,11 @@ async def app_login( *, geetest: typing.Optional[typing.Dict[str, typing.Any]] = None, ) -> typing.Dict[str, typing.Any]: - """Login with a password and a solved geetest (if triggered) - using HoYoLab app endpoint. - + """Login with a password using HoYoLab app endpoint. + Returns data from aigis header or action_ticket or cookies. """ - headers = { + headers = { **geetest_utility.APP_LOGIN_HEADERS, "ds": ds_utility.generate_dynamic_secret(constants.DS_SALT["app_login"]), } @@ -89,12 +87,12 @@ async def app_login( 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), "password": geetest_utility.encrypt_geetest_credentials(password), } - + async with aiohttp.ClientSession() as session: async with session.post( routes.APP_LOGIN_URL.get_url(), @@ -102,20 +100,20 @@ async def app_login( 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 data["retcode"] == -3239: # Email verification required return json.loads(r.headers["x-rpc-verify"]) - + if not data["data"]: errors.raise_for_retcode(data) - + cookies = { "stoken": data["data"]["token"]["token"], "ltuid_v2": data["data"]["user_info"]["aid"], @@ -126,7 +124,7 @@ async def app_login( self.set_cookies(cookies) return cookies - + async def send_verification_email( self, ticket: typing.Dict[str, typing.Any], @@ -134,15 +132,15 @@ async def send_verification_email( geetest: typing.Optional[typing.Dict[str, typing.Any]] = None, ) -> typing.Union[None, typing.Dict[str, typing.Any]]: """Send verification email. - + Returns None if success, aigis headers (mmt/aigis) otherwise. """ - headers = { **geetest_utility.EMAIL_SEND_HEADERS } + headers = {**geetest_utility.EMAIL_SEND_HEADERS} if geetest: mmt_data = geetest["data"] session_id = geetest["session_id"] headers["x-rpc-aigis"] = geetest_utility.get_aigis_header(session_id, mmt_data) - + async with aiohttp.ClientSession() as session: async with session.post( routes.SEND_VERIFICATION_CODE_URL.get_url(), @@ -153,18 +151,18 @@ async def send_verification_email( 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 data["retcode"] != 0: errors.raise_for_retcode(data) - + return None - + async def verify_email(self, code: str, ticket: typing.Dict[str, typing.Any]) -> None: """Verify email.""" async with aiohttp.ClientSession() as session: @@ -179,82 +177,81 @@ async def verify_email(self, code: str, ticket: typing.Dict[str, typing.Any]) -> headers=geetest_utility.EMAIL_VERIFY_HEADERS, ) as r: data = await r.json() - + if data["retcode"] != 0: errors.raise_for_retcode(data) - + return None - + async def login_with_password( - self, - account: str, - password: str, - *, + self, + account: str, + password: str, + *, 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]]] + 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 web endpoint. - Note that this will start a webserver if captcha is + Note that this will start a webserver if captcha is triggered and `geetest_solver` is not passed. """ result = await self.web_login(account, password, tokenType=tokenType) - + if "session_id" not in result: # Captcha not triggered return result - - if (geetest_solver): + + if geetest_solver: geetest = await geetest_solver(result) else: geetest = await server.solve_geetest(result, port=port) - + return await self.web_login(account, password, tokenType=tokenType, geetest=geetest) - + async def login_with_app_password( - self, - account: str, - password: str, - *, + 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, + 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 HoYoLab app endpoint. - Note that this will start a webserver if either of the + Note that this will start a webserver if either of the following happens: - + 1. Captcha is triggered and `geetest_solver` is not passed. 2. Email verification is triggered (can happen if you first login with a new device). """ result = await self.app_login(account, password) - + if "session_id" in result: # Captcha triggered - if (geetest_solver): + if geetest_solver: geetest = await geetest_solver(result) else: geetest = await server.solve_geetest(result, port=port) - + result = await self.app_login(account, password, geetest=geetest) - + if "risk_ticket" in result: # Email verification required mmt = await self.send_verification_email(result) - if (mmt): - if (geetest_solver): + if mmt: + if geetest_solver: geetest = await geetest_solver(mmt) else: geetest = await server.solve_geetest(mmt, port=port) - + await server.verify_email(self, result, port=port) result = await self.app_login(account, password) - + return result - \ No newline at end of file diff --git a/genshin/client/components/geetest/server.py b/genshin/client/components/geetest/server.py index aefd46bd..639c3fc9 100644 --- a/genshin/client/components/geetest/server.py +++ b/genshin/client/components/geetest/server.py @@ -12,65 +12,71 @@ __all__ = ["get_page", "launch_webapp", "solve_geetest", "verify_email"] + def get_page(page: typing.Literal["captcha", "verify-email"]) -> str: - return """ - - - - - - - """ if page == "captcha" else """ - - - + """Get the HTML page.""" + return ( + """ + + + + + + + """ + if page == "captcha" + else """ + + + - - - - """ + + + + """ + ) GT_URL = "https://raw.githubusercontent.com/GeeTeam/gt3-node-sdk/master/demo/static/libs/gt.js" @@ -78,7 +84,7 @@ def get_page(page: typing.Literal["captcha", "verify-email"]) -> str: async def launch_webapp( page: typing.Literal["captcha", "verify-email"], - *, + *, port: int = 5000, mmt: typing.Optional[typing.Dict[str, typing.Any]] = None, ) -> typing.Any: @@ -89,7 +95,7 @@ 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") - + @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") @@ -135,22 +141,22 @@ async def send_data_endpoint(request: web.Request) -> web.Response: async def solve_geetest( - mmt: typing.Dict[str, typing.Any], - *, - port: int = 5000, + mmt: typing.Dict[str, typing.Any], + *, + port: int = 5000, ) -> typing.Dict[str, typing.Any]: - """Solve a geetest captcha manually.""" - return await launch_webapp("captcha", port=port, mmt=mmt) + """Solve a geetest captcha manually.""" + return await launch_webapp("captcha", port=port, mmt=mmt) async def verify_email( client: client.GeetestClient, ticket: typing.Dict[str, typing.Any], *, - port: int = 5000, + port: int = 5000, ) -> None: - """Verify email to login via HoYoLab app endpoint.""" - data = await launch_webapp("verify-email", port=port) - code = data["code"] - - return await client.verify_email(code, ticket) + """Verify email to login via HoYoLab app endpoint.""" + data = await launch_webapp("verify-email", port=port) + code = data["code"] + + return await client.verify_email(code, ticket) diff --git a/genshin/client/manager/cookie.py b/genshin/client/manager/cookie.py index ee591388..1780de6e 100644 --- a/genshin/client/manager/cookie.py +++ b/genshin/client/manager/cookie.py @@ -22,17 +22,17 @@ import aiohttp import aiohttp.typedefs -from genshin import errors, types, constants +from genshin import constants, errors, types from genshin.client import routes from genshin.client.manager import managers from genshin.utility import ds as ds_utility __all__ = [ - "complete_cookies", - "refresh_cookie_token", - "fetch_cookie_token_info", - "fetch_cookie_with_cookie", + "complete_cookies", + "fetch_cookie_token_info", + "fetch_cookie_with_cookie", "fetch_cookie_with_stoken_v2", + "refresh_cookie_token", ] @@ -79,30 +79,28 @@ async def fetch_cookie_with_stoken_v2( if "ltmid_v2" in cookies: # The endpoint requires ltmid_v2 to be named mid cookies["mid"] = cookies["ltmid_v2"] - + url = routes.COOKIE_V2_REFRESH_URL.get_url() - + headers = { "ds": ds_utility.generate_dynamic_secret(constants.DS_SALT["app_login"]), "x-rpc-app_id": "c9oqaq3s3gu8", } - body = { "dst_token_types": token_types } - + body = {"dst_token_types": token_types} + async with aiohttp.ClientSession() as session: r = await session.request("POST", url, json=body, headers=headers, cookies=cookies) data = await r.json(content_type=None) - + if data["retcode"] != 0: errors.raise_for_retcode(data) - + cookies = dict() for token in data["data"]["tokens"]: if token["token_type"] == 2: cookies["ltoken_v2"] = token["token"] elif token["token_type"] == 4: cookies["cookie_token_v2"] = token["token"] - else: - raise ValueError(f"Unknown token type: {token["token_type"]}") return cookies diff --git a/genshin/client/routes.py b/genshin/client/routes.py index 915567a9..e4804618 100644 --- a/genshin/client/routes.py +++ b/genshin/client/routes.py @@ -8,10 +8,12 @@ __all__ = [ "ACCOUNT_URL", + "APP_LOGIN_URL", "BBS_REFERER_URL", "BBS_URL", "CALCULATOR_URL", "COMMUNITY_URL", + "COOKIE_V2_REFRESH_URL", "DETAIL_LEDGER_URL", "GACHA_URL", "HK4E_URL", @@ -23,13 +25,11 @@ "Route", "TAKUMI_URL", "TEAPOT_URL", + "VERIFY_EMAIL_URL", "WEBAPI_URL", "WEBSTATIC_URL", - "YSULOG_URL", - "COOKIE_V2_REFRESH_URL", "WEB_LOGIN_URL", - "APP_LOGIN_URL", - "VERIFY_EMAIL_URL", + "YSULOG_URL", ] @@ -217,5 +217,7 @@ 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") -SEND_VERIFICATION_CODE_URL = Route("https://sg-public-api.hoyoverse.com/account/ma-verifier/api/createEmailCaptchaByActionTicket") +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") diff --git a/genshin/utility/geetest.py b/genshin/utility/geetest.py index 86ad3859..4d535d7d 100644 --- a/genshin/utility/geetest.py +++ b/genshin/utility/geetest.py @@ -1,7 +1,7 @@ """Geetest utilities.""" import base64 -import typing import json +import typing __all__ = ["encrypt_geetest_credentials"] @@ -58,6 +58,7 @@ def encrypt_geetest_credentials(text: str) -> str: crypto = rsa.encrypt(text.encode("utf-8"), public_key) return base64.b64encode(crypto).decode("utf-8") + def get_aigis_header(session_id: str, mmt_data: typing.Dict[str, typing.Any]) -> str: """Get aigis header.""" - return f"{session_id};{base64.b64encode(json.dumps(mmt_data).encode()).decode()}" \ No newline at end of file + return f"{session_id};{base64.b64encode(json.dumps(mmt_data).encode()).decode()}"