From 504c6b06c0fea91bd67cd042706d18b420bd8b6f Mon Sep 17 00:00:00 2001 From: DevilXD Date: Sat, 19 Nov 2022 19:03:52 +0100 Subject: [PATCH] Reimplement standard login procedure, keep chrome as backup --- constants.py | 6 +- gui.py | 215 +++++++------------------------ twitch.py | 353 +++++++++++++++++++++++++++++++++++++++++++++++---- utils.py | 11 ++ 4 files changed, 388 insertions(+), 197 deletions(-) diff --git a/constants.py b/constants.py index 2cffc29..717d1fe 100644 --- a/constants.py +++ b/constants.py @@ -64,9 +64,9 @@ DEFAULT_LANG = "English" BASE_URL = URL("https://twitch.tv") CLIENT_ID = "kd1unb4b3q4t58fwlpcbzcbnm76a8fp" # CLIENT_ID = "uo6dggojyb8d6soh92zknwmi5ej1q2" -USER_AGENT = ( - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/105.0.0.0 Safari/537.36" +ANDROID_USER_AGENT = ( + "Dalvik/2.1.0 (Linux; U; Android 7.1.2; SM-G975N Build/N2G48C) " + "tv.twitch.android.app/13.4.1/1304010" ) # Intervals and Delays PING_INTERVAL = timedelta(minutes=3) diff --git a/gui.py b/gui.py index c8e245a..33acd6d 100644 --- a/gui.py +++ b/gui.py @@ -1,8 +1,6 @@ from __future__ import annotations -import re import sys -import json import ctypes import asyncio import logging @@ -10,7 +8,7 @@ import webbrowser import tkinter as tk from collections import abc from math import log10, ceil -from subprocess import CREATE_NO_WINDOW +from dataclasses import dataclass from tkinter.font import Font, nametofont from functools import partial, cached_property from datetime import datetime, timedelta, timezone @@ -23,31 +21,13 @@ import win32con import win32gui from yarl import URL from PIL import Image as Image_module -try: - from seleniumwire.request import Request - from selenium.common.exceptions import WebDriverException - from seleniumwire.undetected_chromedriver import Chrome, ChromeOptions -except ImportError as exc: - raise ImportError( - "You need to install Visual C++ Redist (x86 and x64): " - "https://support.microsoft.com/en-gb/help/2977003/the-latest-supported-visual-c-downloads" - ) from exc from translate import _ from cache import ImageCache +from exceptions import ExitRequest from utils import resource_path, Game, _T from registry import RegistryKey, ValueType -from exceptions import MinerException, ExitRequest, LoginException -from constants import ( - CLIENT_ID, - SELF_PATH, - # CACHE_PATH, - FORMATTER, - WS_TOPICS_LIMIT, - MAX_WEBSOCKETS, - WINDOW_TITLE, - State, -) +from constants import SELF_PATH, FORMATTER, WS_TOPICS_LIMIT, MAX_WEBSOCKETS, WINDOW_TITLE, State if TYPE_CHECKING: from twitch import Twitch @@ -424,6 +404,13 @@ class WebsocketStatus: self._topics_var.set('\n'.join(topic_lines)) +@dataclass +class LoginData: + username: str + password: str + token: str + + class LoginForm: def __init__(self, manager: GUIManager, master: ttk.Widget): self._manager = manager @@ -435,14 +422,14 @@ class LoginForm: frame.rowconfigure(4, weight=1) 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._pass_entry = PlaceholderEntry( - # frame, placeholder=_("gui", "login", "password"), show='•' - # ) - # 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._login_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "username")) + 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._token_entry = PlaceholderEntry(frame, placeholder=_("gui", "login", "twofa_code")) + 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" @@ -450,150 +437,46 @@ class LoginForm: self._button.grid(column=0, row=4, columnspan=2) self.update(_("gui", "login", "logged_out"), None) - @staticmethod - def interceptor(request: Request) -> None: - if ( - request.method == "POST" - and request.url == "https://passport.twitch.tv/protected_login" - ): - body = request.body.decode("utf-8") - data = json.loads(body) - data["client_id"] = CLIENT_ID - request.body = json.dumps(data).encode("utf-8") - del request.headers["Content-Length"] - request.headers["Content-Length"] = str(len(request.body)) + def clear(self, login: bool = False, password: bool = False, token: bool = False): + clear_all = not login and not password and not token + if login or clear_all: + self._login_entry.clear() + if password or clear_all: + self._pass_entry.clear() + if token or clear_all: + self._token_entry.clear() - async def ask_login(self) -> str: - self.update(_("gui", "login", "required"), None) - self._manager.print(_("gui", "login", "request")) - # ensure the window isn't hidden into tray when this runs - self._manager.tray.restore() - # the user needs to press on the login button to open the browser + async def wait_for_login_press(self) -> None: self._confirm.clear() try: self._button.config(state="normal") await self._manager.coro_unless_closed(self._confirm.wait()) finally: self._button.config(state="disabled") - self._confirm.clear() - self._manager.print("Opening Chrome...") - # open the chrome browser on the Twitch's login page - # use a separate executor to void blocking the event loop - loop = asyncio.get_running_loop() - driver = None - try: - version_main = None - for attempt in range(2): - options = ChromeOptions() - options.add_argument("--log-level=3") - options.add_argument("--disable-web-security") - options.add_argument("--allow-running-insecure-content") - options.add_argument("--lang=en") - options.add_argument("--no-sandbox") - options.add_argument("--test-type") - options.add_argument("--disable-gpu") - options.set_capability("pageLoadStrategy", "eager") - try: - wire_options: dict[str, Any] = {"proxy": {}} - if self._manager._twitch.settings.proxy: - wire_options["proxy"]["http"] = str(self._manager._twitch.settings.proxy) - driver_coro = loop.run_in_executor( - None, - lambda: Chrome( - options=options, - suppress_welcome=True, - version_main=version_main, - seleniumwire_options=wire_options, - service_creationflags=CREATE_NO_WINDOW, - # user_data_dir=str(CACHE_PATH.joinpath("ChromeProfile")), - ) - ) - driver = await self._manager.coro_unless_closed(driver_coro) - break - except WebDriverException as exc: - message = exc.msg - if ( - message is not None - and (match := re.search( - r'Chrome version ([\d]+)\nCurrent browser version is ((\d+)\.[\d.]+)', - message, - )) is not None - ): - if not attempt: - version_main = int(match.group(3)) - continue - else: - raise MinerException( - "Your Chrome browser is out of date\n" - f"Required version: {match.group(1)}\n" - f"Current version: {match.group(2)}" - ) from None - raise MinerException( - "An error occured while boostrapping the Chrome browser" - ) from exc - assert driver is not None - driver.request_interceptor = self.interceptor - # driver.set_page_load_timeout(30) - page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv/login") - await self._manager.coro_unless_closed(page_coro) - # enable the login button once the page finishes opening - self._button.config(state="normal") - self._manager.print( - "Complete the login procedure manually by pressing the Login button again." + async def ask_login(self) -> LoginData: + self.update(_("gui", "login", "required"), None) + # ensure the window isn't hidden into tray when this runs + self._manager.tray.restore() + while True: + self._manager.print(_("gui", "login", "request")) + await self.wait_for_login_press() + login_data = LoginData( + self._login_entry.get().strip(), + self._pass_entry.get(), + self._token_entry.get().strip(), ) - - # auto login - # driver.find_element("id", "login-username").send_keys(username) - # driver.find_element("id", "password-input").send_keys(password) - # driver.find_element( - # "css selector", '[data-a-target="passport-login-button"]' - # ).click() - # token submit button css selectors - # Button: "screen="two_factor" target="submit_button" - # Input: - - # wait for the user to navigate away from the URL, indicating successful login - # alternatively, they can press on the login button again - while ( - driver.current_url != "https://www.twitch.tv/?no-reload=true" - and not self._confirm.is_set() - ): - if self._manager.close_requested: - raise ExitRequest() - await asyncio.sleep(0.5) - - # cookies = [ - # { - # "domain": ".twitch.tv", - # "expiry": 1700000000, - # "httpOnly": False, - # "name": "auth-token", - # "path": "/", - # "sameSite": "None", - # "secure": True, - # "value": "..." - # }, - # ..., - # ] - cookies = driver.get_cookies() - for cookie in cookies: - if "twitch.tv" in cookie["domain"] and cookie["name"] == "auth-token": - auth_token = cookie["value"] - break - else: - raise LoginException("Unable to extract authorization token") - except WebDriverException: - driver = None - raise LoginException( - "Chrome window was closed before the login procedure could complete." - ) - finally: - self._button.config(state="disabled") - if driver is not None: - driver.quit() - return auth_token + # basic input data validation + if len(login_data.username) < 3: + self.clear(login=True) + continue + if len(login_data.password) < 8: + self.clear(password=True) + continue + if login_data.token and len(login_data.token) < 6: + self.clear(token=True) + continue + return login_data def update(self, status: str, user_id: int | None): if user_id is not None: diff --git a/twitch.py b/twitch.py index 03835dc..cec421d 100644 --- a/twitch.py +++ b/twitch.py @@ -1,14 +1,16 @@ from __future__ import annotations import re -# import json +import sys +import json +import random import asyncio import logging from math import ceil from time import time from itertools import chain from functools import partial -# from base64 import urlsafe_b64decode +from subprocess import CREATE_NO_WINDOW from collections import abc, OrderedDict from datetime import datetime, timedelta, timezone from contextlib import suppress, asynccontextmanager @@ -16,18 +18,34 @@ from typing import Any, Literal, Final, NoReturn, cast, TYPE_CHECKING import aiohttp from yarl import URL +try: + from seleniumwire.request import Request + from selenium.common.exceptions import WebDriverException + from seleniumwire.undetected_chromedriver import Chrome, ChromeOptions +except ImportError as exc: + raise ImportError( + "You need to install Visual C++ Redist (x86 and x64): " + "https://support.microsoft.com/en-gb/help/2977003/the-latest-supported-visual-c-downloads" + ) from exc from translate import _ from gui import GUIManager from channel import Channel from websocket import WebsocketPool from inventory import DropsCampaign -from exceptions import MinerException, ExitRequest, LoginException, ReloadRequest, RequestInvalid +from exceptions import ( + MinerException, + ExitRequest, + LoginException, + ReloadRequest, + RequestInvalid, +) from utils import ( CHARS_HEX_LOWER, timestamp, create_nonce, task_wrapper, + first_to_complete, OrderedSet, AwaitableValue, ExponentialBackoff, @@ -35,13 +53,13 @@ from utils import ( from constants import ( BASE_URL, CLIENT_ID, - USER_AGENT, COOKIES_PATH, GQL_OPERATIONS, MAX_WEBSOCKETS, WATCH_INTERVAL, WS_TOPICS_LIMIT, DROPS_ENABLED_TAG, + ANDROID_USER_AGENT, State, WebsocketTopic, ) @@ -58,6 +76,16 @@ logger = logging.getLogger("TwitchDrops") gql_logger = logging.getLogger("TwitchDrops.gql") +class SkipExtraJsonDecoder(json.JSONDecoder): + def decode(self, s: str, *args): + # skip whitespace check + obj, end = self.raw_decode(s) + return obj + + +SAFE_LOADS = lambda s: json.loads(s, cls=SkipExtraJsonDecoder) + + class _AuthState: def __init__(self, twitch: Twitch): self._twitch: Twitch = twitch @@ -98,37 +126,292 @@ class _AuthState: ) self._logged_in.clear() + @staticmethod + def interceptor(request: Request) -> None: + if ( + request.method == "POST" + and request.url == "https://passport.twitch.tv/protected_login" + ): + body = request.body.decode("utf-8") + data = json.loads(body) + data["client_id"] = CLIENT_ID + request.body = json.dumps(data).encode("utf-8") + del request.headers["Content-Length"] + request.headers["Content-Length"] = str(len(request.body)) + + async def _chrome_login(self) -> None: + gui_print = self._twitch.gui.print + login_form: LoginForm = self._twitch.gui.login + coro_unless_closed = self._twitch.gui.coro_unless_closed + + # open the chrome browser on the Twitch's login page + # use a separate executor to void blocking the event loop + loop = asyncio.get_running_loop() + driver: Chrome | None = None + while True: + gui_print("Opening Chrome...") + try: + version_main = None + for attempt in range(2): + options = ChromeOptions() + options.add_argument("--log-level=3") + options.add_argument("--disable-web-security") + options.add_argument("--allow-running-insecure-content") + options.add_argument("--lang=en") + options.add_argument("--no-sandbox") + options.add_argument("--test-type") + options.add_argument("--disable-gpu") + options.set_capability("pageLoadStrategy", "eager") + try: + wire_options: dict[str, Any] = {"proxy": {}} + if self._twitch.settings.proxy: + wire_options["proxy"]["http"] = str(self._twitch.settings.proxy) + driver_coro = loop.run_in_executor( + None, + lambda: Chrome( + options=options, + suppress_welcome=True, + version_main=version_main, + seleniumwire_options=wire_options, + service_creationflags=CREATE_NO_WINDOW, + ) + ) + driver = await coro_unless_closed(driver_coro) + break + except WebDriverException as exc: + message = exc.msg + if ( + message is not None + and ( + match := re.search( + ( + r'Chrome version ([\d]+)\n' + r'Current browser version is ((\d+)\.[\d.]+)' + ), + message, + ) + ) is not None + ): + if not attempt: + version_main = int(match.group(3)) + continue + else: + raise MinerException( + "Your Chrome browser is out of date\n" + f"Required version: {match.group(1)}\n" + f"Current version: {match.group(2)}" + ) from None + raise MinerException( + "An error occured while boostrapping the Chrome browser" + ) from exc + assert driver is not None + driver.request_interceptor = self.interceptor + # driver.set_page_load_timeout(30) + # page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv") + # await coro_unless_closed(page_coro) + page_coro = loop.run_in_executor(None, driver.get, "https://twitch.tv/login") + await coro_unless_closed(page_coro) + + # auto login + # if login_data.username and login_data.password: + # driver.find_element("id", "login-username").send_keys(login_data.username) + # driver.find_element("id", "password-input").send_keys(login_data.password) + # driver.find_element( + # "css selector", '[data-a-target="passport-login-button"]' + # ).click() + # token submit button css selectors + # Button: "screen="two_factor" target="submit_button" + # Input: + + # wait for the user to navigate away from the URL, indicating successful login + # alternatively, they can press on the login button again + async def url_waiter(driver=driver): + while driver.current_url != "https://www.twitch.tv/?no-reload=true": + await asyncio.sleep(0.5) + + gui_print( + "Complete the login procedure manually by pressing the Login button again." + ) + await first_to_complete([ + url_waiter(), + coro_unless_closed(login_form.wait_for_login_press()), + ]) + + # cookies = [ + # { + # "domain": ".twitch.tv", + # "expiry": 1700000000, + # "httpOnly": False, + # "name": "auth-token", + # "path": "/", + # "sameSite": "None", + # "secure": True, + # "value": "..." + # }, + # ..., + # ] + cookies = driver.get_cookies() + for cookie in cookies: + if "twitch.tv" in cookie["domain"] and cookie["name"] == "auth-token": + self.access_token = cookie["value"] + break + else: + gui_print("Unable to extract authorization token") + except WebDriverException: + gui_print( + "Chrome window was closed before the login procedure could complete." + ) + finally: + if driver is not None: + driver.quit() + driver = None + await coro_unless_closed(login_form.wait_for_login_press()) + async def _login(self) -> str: logger.debug("Login flow started") + gui_print = self._twitch.gui.print login_form: LoginForm = self._twitch.gui.login - for attempt in range(1, 6): # 5 attempts - try: - self.access_token = await login_form.ask_login() - break - except LoginException as exc: - if attempt < 5: - self._twitch.gui.print(exc.args[0]) - continue - raise # reraise on and past the 5th attempt - logger.debug("Access token granted") - return self.access_token + token_kind: str = '' + use_chrome: bool = False + payload: JsonType = { + "client_id": CLIENT_ID, # client ID to-be associated with the access token + "undelete_user": False, # purpose unknown + "remember_me": True, # persist the session via the cookie + # "authy_token": str, # 2FA token + # "twitchguard_code": str, # email code + # "captcha": str, # self-fed captcha + # 'force_twitchguard': False, # force email code confirmation + } - def gql_headers(self, *, integrity: bool) -> JsonType: + while True: + login_data = await login_form.ask_login() + payload["username"] = login_data.username + payload["password"] = login_data.password + # reinstate the 2FA token, if present + payload.pop("authy_token", None) + payload.pop("twitchguard_code", None) + if login_data.token: + # if there's no token kind set yet, and the user has entered a token, + # we can immediately assume it's an authenticator token and not an email one + if not token_kind: + token_kind = "authy" + if token_kind == "authy": + payload["authy_token"] = login_data.token + elif token_kind == "email": + payload["twitchguard_code"] = login_data.token + + # use fancy headers to mimic the twitch android app + headers = { + "Accept": "application/vnd.twitchtv.v3+json", + "Accept-Encoding": "gzip", + "Accept-Language": "en-US", + "Client-Id": CLIENT_ID, + "Content-Type": "application/json; charset=UTF-8", + "Host": "passport.twitch.tv", + "User-Agent": ANDROID_USER_AGENT, + "X-Device-Id": self.device_id, + # "X-Device-Id": ''.join(random.choices('0123456789abcdef', k=32)), + } + async with self._twitch.request( + "POST", "https://passport.twitch.tv/login", headers=headers, json=payload + ) as response: + login_response: JsonType = await response.json(loads=SAFE_LOADS) + + # Feed this back in to avoid running into CAPTCHA if possible + if "captcha_proof" in login_response: + payload["captcha"] = {"proof": login_response["captcha_proof"]} + + # Error handling + if "error_code" in login_response: + error_code: int = login_response["error_code"] + logger.debug(f"Login error code: {error_code}") + if error_code == 1000: + logger.debug("1000: CAPTCHA is required") + use_chrome = True + break + elif error_code == 3001: + logger.debug("3001: Login failed due to incorrect username or password") + gui_print(_("login", "incorrect_login_pass")) + login_form.clear(password=True) + continue + elif error_code in ( + 3012, # Invalid authy token + 3023, # Invalid email code + ): + logger.debug("3012/23: Login failed due to incorrect 2FA code") + if error_code == 3023: + token_kind = "email" + gui_print(_("login", "incorrect_email_code")) + else: + token_kind = "authy" + gui_print(_("login", "incorrect_twofa_code")) + login_form.clear(token=True) + continue + elif error_code in ( + 3011, # Authy token needed + 3022, # Email code needed + ): + # 2FA handling + logger.debug("3011/22: 2FA token required") + # user didn't provide a token, so ask them for it + if error_code == 3022: + token_kind = "email" + gui_print(_("login", "email_code_required")) + else: + token_kind = "authy" + gui_print(_("login", "twofa_code_required")) + continue + elif error_code == 5027: + # client blocked from this operation - use chrome + logger.debug("5027: Client blocked from this operation") + use_chrome = True + break + else: + msg = str(login_response["error"]) + logger.debug(msg) + raise LoginException(msg) + # Success handling + if "access_token" in login_response: + self.access_token = cast(str, login_response["access_token"]) + logger.debug("Access token granted") + login_form.clear() + break + + if use_chrome: + await self._chrome_login() + + if hasattr(self, "access_token"): + logger.debug("Access token granted") + return self.access_token + raise MinerException("Login flow finished without setting the access token") + + def headers( + self, *, user_agent: str = '', gql: bool = False, integrity: bool = False + ) -> JsonType: headers = { - "Authorization": f"OAuth {self.access_token}", "Accept": "*/*", + "Accept-Encoding": "gzip", + "Accept-Language": "en-US", "Pragma": "no-cache", "Cache-Control": "no-cache", - "Origin": "https://www.twitch.tv", - "Referer": "https://www.twitch.tv/", "Client-Id": CLIENT_ID, - "Client-Session-Id": self.session_id, - "Client-Version": self.client_version, - "X-Device-Id": self.device_id, } - # if integrity: - # headers["Client-Integrity"] = self.integrity_token + if user_agent: + headers["User-Agent"] = user_agent + if hasattr(self, "session_id"): + headers["Client-Session-Id"] = self.session_id + if hasattr(self, "client_version"): + headers["Client-Version"] = self.client_version + if hasattr(self, "device_id"): + headers["X-Device-Id"] = self.device_id + if gql: + headers["Origin"] = "https://www.twitch.tv" + headers["Referer"] = "https://www.twitch.tv/" + headers["Authorization"] = f"OAuth {self.access_token}" + if integrity: + headers["Client-Integrity"] = self.integrity_token return headers async def validate(self): @@ -142,8 +425,11 @@ class _AuthState: session = await self._twitch.get_session() jar = cast(aiohttp.CookieJar, session.cookie_jar) if not self._hasattrs("client_version", "device_id"): - async with self._twitch.request("GET", BASE_URL) as response: - match = re.search(r'twilightBuildID="([-a-z0-9]+)"', await response.text("utf8")) + async with self._twitch.request( + "GET", BASE_URL, headers=self.headers() # (user_agent=ANDROID_USER_AGENT) + ) as response: + page_html = await response.text("utf8") + match = re.search(r'twilightBuildID="([-a-z0-9]+)"', page_html) if match is None: raise MinerException("Unable to extract client_version") self.client_version = match.group(1) @@ -253,12 +539,23 @@ class Twitch: if session.closed: raise RuntimeError("Session is closed") return session + # obtain the latest Chrome user agent from a Github project + async with aiohttp.request( + "GET", "https://jnrbsn.github.io/user-agents/user-agents.json" + ) as response: + agents = await response.json() + if sys.platform == "win32": + chrome_agent = random.choice(agents[:3]) + elif sys.platform == "linux": + chrome_agent = random.choice(agents[7:11]) + # load in cookies cookie_jar = aiohttp.CookieJar() if COOKIES_PATH.exists(): cookie_jar.load(COOKIES_PATH) + # create session self._session = aiohttp.ClientSession( cookie_jar=cookie_jar, - headers={"User-Agent": USER_AGENT}, + headers={"User-Agent": chrome_agent}, timeout=aiohttp.ClientTimeout(connect=5, total=10), ) return self._session @@ -974,7 +1271,7 @@ class Twitch: "POST", "https://gql.twitch.tv/gql", json=op, - headers=auth_state.gql_headers(integrity=True), + headers=auth_state.headers(gql=True), invalidate_after=getattr(auth_state, "integrity_expires", None), ) as response: response_json: JsonType = await response.json() diff --git a/utils.py b/utils.py index c021c78..f4d3bd7 100644 --- a/utils.py +++ b/utils.py @@ -38,6 +38,17 @@ _JSON_T = TypeVar("_JSON_T", bound=Mapping[Any, Any]) logger = logging.getLogger("TwitchDrops") +async def first_to_complete(coros: abc.Iterable[abc.Coroutine[Any, Any, _T]]) -> _T: + # In Python 3.11, we need to explicitly wrap awaitables + tasks = [asyncio.ensure_future(coro) for coro in coros] + done: set[asyncio.Task[Any]] + pending: set[asyncio.Task[Any]] + done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + for task in pending: + task.cancel() + return next(iter(done)).result() + + def format_traceback(exc: BaseException, **kwargs: Any) -> str: """ Like `traceback.print_exc` but returns a string. Uses the passed-in exception.