From 3892dfe59e8835486966cd668c9e128c36b7cd95 Mon Sep 17 00:00:00 2001 From: DevilXD Date: Fri, 23 Dec 2022 13:18:23 +0100 Subject: [PATCH] Implement SmartTV OAuth2 login flow --- gui.py | 16 ++++++++--- twitch.py | 79 +++++++++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 90 insertions(+), 5 deletions(-) diff --git a/gui.py b/gui.py index ab47ba7..88a512f 100644 --- a/gui.py +++ b/gui.py @@ -431,13 +431,14 @@ class LoginForm: ttk.Label(frame, text=_("gui", "login", "labels")).grid(column=0, row=0) ttk.Label(frame, textvariable=self._var, justify="center").grid(column=1, row=0) self._login_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "username")) - self._login_entry.grid(column=0, row=1, columnspan=2) + # self._login_entry.grid(column=0, row=1, columnspan=2) self._pass_entry = PlaceholderEntry( frame, placeholder=_("gui", "login", "password"), show='•' ) - self._pass_entry.grid(column=0, row=2, columnspan=2) + # self._pass_entry.grid(column=0, row=2, columnspan=2) self._token_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "twofa_code")) - self._token_entry.grid(column=0, row=3, columnspan=2) + # self._token_entry.grid(column=0, row=3, columnspan=2) + self._confirm = asyncio.Event() self._button = ttk.Button( frame, text=_("gui", "login", "button"), command=self._confirm.set, state="disabled" @@ -489,6 +490,15 @@ class LoginForm: continue return login_data + async def ask_enter_code(self, user_code: str) -> None: + self.update(_("gui", "login", "required"), None) + # ensure the window isn't hidden into tray when this runs + self._manager.tray.restore() + self._manager.print(_("gui", "login", "request")) + await self.wait_for_login_press() + self._manager.print(f"Enter this code on the Twitch's device activation page: {user_code}") + webbrowser.open_new_tab("https://www.twitch.tv/activate") + def update(self, status: str, user_id: int | None): if user_id is not None: user_str = str(user_id) diff --git a/twitch.py b/twitch.py index 721e453..876a200 100644 --- a/twitch.py +++ b/twitch.py @@ -86,7 +86,7 @@ class SkipExtraJsonDecoder(json.JSONDecoder): return obj -CLIENT_ID, USER_AGENT = ClientType.ANDROID +CLIENT_ID, USER_AGENT = ClientType.SMARTBOX SAFE_LOADS = lambda s: json.loads(s, cls=SkipExtraJsonDecoder) @@ -268,6 +268,81 @@ class _AuthState: driver = None await coro_unless_closed(login_form.wait_for_login_press()) + async def _oauth_login(self) -> str: + login_form: LoginForm = self._twitch.gui.login + headers = { + "Accept": "application/json", + "Accept-Encoding": "gzip", + "Accept-Language": "en-US", + "Cache-Control": "no-cache", + "Client-Id": CLIENT_ID, + "Host": "id.twitch.tv", + "Origin": "https://android.tv.twitch.tv", + "Pragma": "no-cache", + "Referer": "https://android.tv.twitch.tv/", + "User-Agent": USER_AGENT, + "X-Device-Id": self.device_id, + } + payload = { + "client_id": CLIENT_ID, + "scopes": ( + "channel_read chat:read user_blocks_edit " + "user_blocks_read user_follows_edit user_read" + ), + } + while True: + try: + async with self._twitch.request( + "POST", "https://id.twitch.tv/oauth2/device", headers=headers, data=payload + ) as response: + # { + # "device_code": "40 chars [A-Za-z0-9]", + # "expires_in": 1800, + # "interval": 5, + # "user_code": "8 chars [A-Z]", + # "verification_uri": "https://www.twitch.tv/activate" + # } + now = datetime.now(timezone.utc) + response_json: JsonType = await response.json() + device_code: str = response_json["device_code"] + user_code: str = response_json["user_code"] + interval: int = response_json["interval"] + expires_at = now + timedelta(seconds=response_json["expires_in"]) + + # Print the code to the user, open them the activate page so they can type it in + await login_form.ask_enter_code(user_code) + + payload = { + "client_id": CLIENT_ID, + "device_code": device_code, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + } + while True: + # sleep first, not like the user is gonna enter the code *that* fast + await asyncio.sleep(interval) + async with self._twitch.request( + "POST", + "https://id.twitch.tv/oauth2/token", + headers=headers, + data=payload, + invalidate_after=expires_at, + ) as response: + # 200 means success, 400 means the user haven't entered the code yet + if response.status != 200: + continue + response_json = await response.json() + # { + # "access_token": "40 chars [A-Za-z0-9]", + # "refresh_token": "40 chars [A-Za-z0-9]", + # "scope": [...], + # "token_type": "bearer" + # } + self.access_token = cast(str, response_json["access_token"]) + return self.access_token + except RequestInvalid: + # the device_code has expired, request a new code + continue + async def _login(self) -> str: logger.info("Login flow started") gui_print = self._twitch.gui.print @@ -462,7 +537,7 @@ class _AuthState: for attempt in range(2): cookie = jar.filter_cookies(BASE_URL) if "auth-token" not in cookie: - self.access_token = await self._login() + self.access_token = await self._oauth_login() cookie["auth-token"] = self.access_token elif not hasattr(self, "access_token"): logger.info("Restoring session from cookie")