diff --git a/TwitchChannelPointsMiner/__init__.py b/TwitchChannelPointsMiner/__init__.py index 11d7d3c..6e84511 100644 --- a/TwitchChannelPointsMiner/__init__.py +++ b/TwitchChannelPointsMiner/__init__.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -__version__ = "1.6.0" +__version__ = "1.7.0" from .TwitchChannelPointsMiner import TwitchChannelPointsMiner __all__ = [ diff --git a/TwitchChannelPointsMiner/classes/TwitchLogin.py b/TwitchChannelPointsMiner/classes/TwitchLogin.py index 7bcd4bf..d8784ad 100644 --- a/TwitchChannelPointsMiner/classes/TwitchLogin.py +++ b/TwitchChannelPointsMiner/classes/TwitchLogin.py @@ -3,12 +3,13 @@ # The MIT License (MIT) import copy -import getpass +# import getpass import logging import os import pickle -import browser_cookie3 +# import webbrowser +# import browser_cookie3 import requests @@ -16,7 +17,10 @@ from TwitchChannelPointsMiner.classes.Exceptions import ( BadCredentialsException, WrongCookiesException, ) -from TwitchChannelPointsMiner.constants import CLIENT_ID, GQLOperations +from TwitchChannelPointsMiner.constants import CLIENT_ID, GQLOperations, USER_AGENTS + +from datetime import datetime, timedelta, timezone +from time import sleep logger = logging.getLogger(__name__) @@ -73,104 +77,103 @@ class TwitchLogin(object): post_data = { "client_id": self.client_id, - "undelete_user": False, - "remember_me": True, + "scopes": ( + "channel_read chat:read user_blocks_edit " + "user_blocks_read user_follows_edit user_read" + ) } # login-fix - # use_backup_flow = False - use_backup_flow = True + use_backup_flow = False + # use_backup_flow = True + while True: + logger.info("Trying the TV login method..") - for attempt in range(0, 25): - password = ( - getpass.getpass(f"Enter Twitch password for {self.username}: ") - if self.password in [None, ""] - else self.password - ) + login_response = self.send_oauth_request( + "https://id.twitch.tv/oauth2/device", post_data) - post_data["username"] = self.username - post_data["password"] = password + # { + # "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" + # } - while True: - # Try login without 2FA - login_response = self.send_login_request(post_data) + if login_response.status_code != 200: + logger.error("TV login response is not 200. Try again") + break - if "captcha_proof" in login_response: - post_data["captcha"] = dict( - proof=login_response["captcha_proof"]) + login_response_json = login_response.json() - if "error_code" in login_response: - err_code = login_response["error_code"] - if err_code in [3011, 3012]: # missing 2fa token - if err_code == 3011: - logger.info( - "Two factor authentication enabled, please enter token below." - ) - else: - logger.info( - "Invalid two factor token, please try again.") + if "user_code" in login_response_json: + user_code: str = login_response_json["user_code"] + now = datetime.now(timezone.utc) + device_code: str = login_response_json["device_code"] + interval: int = login_response_json["interval"] + expires_at = now + \ + timedelta(seconds=login_response_json["expires_in"]) + logger.info( + "Open https://www.twitch.tv/activate" + ) + logger.info( + f"and enter this code: {user_code}" + ) + logger.info( + f"Hurry up! It will expire in {int(login_response_json['expires_in'] / 60)} minutes!" + ) + # twofa = input("2FA token: ") + # webbrowser.open_new_tab("https://www.twitch.tv/activate") - twofa = input("2FA token: ") - post_data["authy_token"] = twofa.strip() - continue + post_data = { + "client_id": CLIENT_ID, + "device_code": device_code, + "grant_type": "urn:ietf:params:oauth:grant-type:device_code", + } - elif err_code in [3022, 3023]: # missing 2fa token - if err_code == 3022: - logger.info("Login Verification code required.") - self.email = login_response["obscured_email"] - else: - logger.info( - "Invalid Login Verification code entered, please try again." - ) - - twofa = input( - f"Please enter the 6-digit code sent to {self.email}: " - ) - post_data["twitchguard_code"] = twofa.strip() - continue - - # invalid password, or password not provided - elif err_code in [3001, 3003]: - logger.info( - "Invalid username or password, please try again.") - - # If the password is loaded from run.py, require the user to fix it there. - if self.password not in [None, ""]: - raise BadCredentialsException( - "Username or password is incorrect." - ) - - # If the user didn't load the password from run.py we can just ask for it again. + while True: + # sleep first, not like the user is gonna enter the code *that* fast + sleep(interval) + login_response = self.send_oauth_request( + "https://id.twitch.tv/oauth2/token", post_data) + if now == expires_at: + logger.error("Code expired. Try again") break - # login-fix - # elif err_code == 1000: - elif err_code in [1000, 5022, 5023, 5024]: - logger.info( - "Console login unavailable (CAPTCHA solving required)." - ) - use_backup_flow = True - break - # https://github.com/rdavydov/Twitch-Channel-Points-Miner-v2/issues/46 -# elif err_code == 5023: -# logger.error( -# "Probably an automatic temporary ban from Twitch. Please wait 24 hours till they lift the ban from the account and try again." -# ) -# use_backup_flow = True -# break + # 200 means success, 400 means the user haven't entered the code yet + if login_response.status_code != 200: + continue + # { + # "access_token": "40 chars [A-Za-z0-9]", + # "refresh_token": "40 chars [A-Za-z0-9]", + # "scope": [...], + # "token_type": "bearer" + # } + login_response_json = login_response.json() + if "access_token" in login_response_json: + self.set_token(login_response_json["access_token"]) + return self.check_login() + # except RequestInvalid: + # the device_code has expired, request a new code + # continue + # invalidate_after is not None + # account for the expiration landing during the request + # and datetime.now(timezone.utc) >= (invalidate_after - session_timeout) + # ): + # raise RequestInvalid() else: + if "error_code" in login_response: + err_code = login_response["error_code"] + logger.error(f"Unknown error: {login_response}") raise NotImplementedError( f"Unknown TwitchAPI error code: {err_code}" ) - if "access_token" in login_response: - self.set_token(login_response["access_token"]) - return self.check_login() - if use_backup_flow: break if use_backup_flow: - self.set_token(self.login_flow_backup(password)) + # self.set_token(self.login_flow_backup(password)) + self.set_token(self.login_flow_backup()) return self.check_login() return False @@ -179,16 +182,30 @@ class TwitchLogin(object): self.token = new_token self.session.headers.update({"Authorization": f"Bearer {self.token}"}) - def send_login_request(self, json_data): + # def send_login_request(self, json_data): + def send_oauth_request(self, url, json_data): # response = self.session.post("https://passport.twitch.tv/protected_login", json=json_data) - response = self.session.post("https://passport.twitch.tv/login", json=json_data, headers={ + """response = self.session.post("https://passport.twitch.tv/login", json=json_data, headers={ 'Accept': 'application/vnd.twitchtv.v3+json', 'Accept-Encoding': 'gzip', 'Accept-Language': 'en-US', 'Content-Type': 'application/json; charset=UTF-8', 'Host': 'passport.twitch.tv' + },)""" + response = self.session.post(url, data=json_data, 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_AGENTS["Android"]["TV"], + "X-Device-Id": self.device_id },) - return response.json() + return response def login_flow_backup(self, password=None): """Backup OAuth Selenium login @@ -208,13 +225,14 @@ class TwitchLogin(object): options.add_argument('--lang=en') options.add_argument('--no-sandbox') options.add_argument('--disable-gpu') - #options.add_argument("--user-agent=\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36\"") - #options.add_argument("--window-size=1920,1080") - #options.set_capability("detach", True) + # options.add_argument("--user-agent=\"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36\"") + # options.add_argument("--window-size=1920,1080") + # options.set_capability("detach", True) - logger.info('Now a browser window will open, it will login with your data.') + logger.info( + 'Now a browser window will open, it will login with your data.') driver = uc.Chrome( - options=options, use_subprocess=True#, executable_path=EXECUTABLE_PATH + options=options, use_subprocess=True # , executable_path=EXECUTABLE_PATH ) driver.request_interceptor = interceptor driver.get('https://www.twitch.tv/login') @@ -233,11 +251,11 @@ class TwitchLogin(object): logger.info("Extracting cookies...") self.cookies = driver.get_cookies() - #print(self.cookies) - #driver.close() + # print(self.cookies) + # driver.close() driver.quit() self.username = self.get_cookie_value("login") - #print(f"self.username: {self.username}") + # print(f"self.username: {self.username}") if not self.username: logger.error("Couldn't extract login, probably bad cookies.") @@ -282,15 +300,16 @@ class TwitchLogin(object): return self.login_check_result def save_cookies(self, cookies_file): - # cookies_dict = self.session.cookies.get_dict() + logger.info("Saving cookies to your computer..") + cookies_dict = self.session.cookies.get_dict() # print(f"cookies_dict2pickle: {cookies_dict}") - # cookies_dict["auth-token"] = self.token - # if "persistent" not in cookies_dict: # saving user id cookies - # cookies_dict["persistent"] = self.user_id + cookies_dict["auth-token"] = self.token + if "persistent" not in cookies_dict: # saving user id cookies + cookies_dict["persistent"] = self.user_id # old way saves only 'auth-token' and 'persistent' self.cookies = [] - cookies_dict = self.shared_cookies + # cookies_dict = self.shared_cookies # print(f"cookies_dict2pickle: {cookies_dict}") for cookie_name, value in cookies_dict.items(): self.cookies.append({"name": cookie_name, "value": value}) diff --git a/TwitchChannelPointsMiner/utils.py b/TwitchChannelPointsMiner/utils.py index 6a4bd9e..79d7c70 100644 --- a/TwitchChannelPointsMiner/utils.py +++ b/TwitchChannelPointsMiner/utils.py @@ -58,11 +58,12 @@ def create_nonce(length=30) -> str: def get_user_agent(browser: str) -> str: - try: + """try: return USER_AGENTS[platform.system()][browser] except KeyError: # return USER_AGENTS["Linux"]["FIREFOX"] - return USER_AGENTS["Windows"]["CHROME"] + # return USER_AGENTS["Windows"]["CHROME"]""" + return USER_AGENTS["Android"]["TV"] # return USER_AGENTS["Android"]["App"] diff --git a/requirements.txt b/requirements.txt index 9e67bfe..5a011c2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,7 +9,3 @@ colorama flask irc pandas -browser_cookie3 -selenium -selenium-wire -undetected_chromedriver diff --git a/setup.py b/setup.py index 8626e5d..30afcec 100644 --- a/setup.py +++ b/setup.py @@ -36,11 +36,7 @@ setuptools.setup( "colorama", "flask", "irc", - "pandas", - "browser_cookie3", - "selenium", - "selenium-wire", - "undetected_chromedriver" + "pandas" ], long_description=read("README.md"), long_description_content_type="text/markdown",