From 79d108610dd418cb84274616c846158ced714174 Mon Sep 17 00:00:00 2001 From: moo611 Date: Sat, 29 Apr 2023 13:44:35 +0800 Subject: [PATCH 1/4] login --- src/twitter_video_dl/login.py | 136 ++++++++++++++++++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 src/twitter_video_dl/login.py diff --git a/src/twitter_video_dl/login.py b/src/twitter_video_dl/login.py new file mode 100644 index 0000000..bc2ab6f --- /dev/null +++ b/src/twitter_video_dl/login.py @@ -0,0 +1,136 @@ +import requests +import json + +class TwitterLoginClient: + def __init__(self, bear_token,proxies={}): + self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36" + self.bear_token = bear_token + self.proxies = proxies + self.session = requests.session() + self.__twitter() + self.x_guest_token = self.get_guest_token() + self.flow_token = None + + def login(self,username,password): + + + + + def __twitter(self): + headers = { + "User-Agent":self.USER_AGENT, + } + response = self.session.get( + "https://twitter.com/", headers=headers, proxies=self.proxies + ) + + + + def get_guest_token(self): + headers = { + "authorization": self.AUTHORIZATION, + "User-Agent": self.USER_AGENT, + } + response = self.session.post( + "https://api.twitter.com/1.1/guest/activate.json", + headers=headers, + proxies=self.proxies, + ).json() + return response["guest_token"] + + + def login_flow(self): + data = { + "input_flow_data": { + "flow_context": { + "debug_overrides": {}, + "start_location": {"location": "splash_screen"}, + } + }, + "subtask_versions": { + "contacts_live_sync_permission_prompt": 0, + "email_verification": 1, + "topics_selector": 1, + "wait_spinner": 1, + "cta": 4, + }, + } + params = {"flow_name": "login"} + response = self.session.post( + "https://twitter.com/i/api/1.1/onboarding/task.json", + headers=self.__get_headers(), + json=data, + params=params, + proxies=self.proxies, + ).json() + self.__error_check(response) + self.flow_token = response.get("flow_token") + self.content = response + return self + + def LoginJsInstrumentationSubtask(self): + self.__flow_token_check() + self.__method_check("LoginJsInstrumentationSubtask") + data = { + "flow_token": self.flow_token, + "subtask_inputs": [ + { + "subtask_id": "LoginJsInstrumentationSubtask", + "js_instrumentation": { + "response": json.dumps( + { + "rf": { + "af07339bbc6d24ced887d705eb0c9fd29b4a7d7ddc21136c9f94d53a4bc774d2": 88, + "a6ce87d6481c6ec4a823548be3343437888441d2a453061c54f8e2eb325856f7": 250, + "a0062ad06384a8afd38a41cd83f31b0dbfdea0eff4b24c69f0dd9095b2fb56d6": 16, + "a929e5913a5715d93491eaffaa139ba4977cbc826a5e2dbcdc81cae0f093db25": 186, + }, + "s": "Q-H-53m1uXImK0F0ogrxRQtCWTH1KIlPbIy0MloowlMa4WNK5ZCcDoXyRs1q_cPbynK73w_wfHG_UVRKKBWRoh6UJtlPS5kMa1p8fEvTYi76hwdzBEzovieR8t86UpeSkSBFYcL8foYKSp6Nop5mQR_QHGyEeleclCPUvzS0HblBJqZZdtUo-6by4BgCyu3eQ4fY5nOF8fXC85mu6k34wo982LMK650NsoPL96DBuloqSZvSHU47wq2uA4xy24UnI2WOc6U9KTvxumtchSYNnXq1HV662B8U2-jWrzvIU4yUHV3JYUO6sbN6j8Ho9JaUNJpJSK7REwqCBQ3yG7iwMAAAAX2Vqcbs", + } + ), + "link": "next_link", + }, + } + ], + } + response = self.session.post( + "https://twitter.com/i/api/1.1/onboarding/task.json", + headers=self.__get_headers(), + json=data, + proxies=self.proxies, + ).json() + self.__error_check(response) + self.flow_token = response.get("flow_token") + self.content = response + return self + + def LoginEnterUserIdentifierSSO(self, user_id): + self.__flow_token_check() + self.__method_check("LoginEnterUserIdentifierSSO") + data = { + "flow_token": self.flow_token, + "subtask_inputs": [ + { + "subtask_id": "LoginEnterUserIdentifierSSO", + "settings_list": { + "setting_responses": [ + { + "key": "user_identifier", + "response_data": {"text_data": {"result": user_id}}, + } + ], + "link": "next_link", + }, + } + ], + } + response = self.session.post( + "https://twitter.com/i/api/1.1/onboarding/task.json", + headers=self.__get_headers(), + json=data, + proxies=self.proxies, + ).json() + self.__error_check(response) + self.flow_token = response.get("flow_token") + self.content = response + return self From 85a3273a710282a668b4fa54d7a49a5b2a756a6a Mon Sep 17 00:00:00 2001 From: moo611 Date: Sat, 29 Apr 2023 20:49:48 +0800 Subject: [PATCH 2/4] login --- src/twitter_video_dl/login.py | 228 ++++++++++++++++++++++------------ 1 file changed, 150 insertions(+), 78 deletions(-) diff --git a/src/twitter_video_dl/login.py b/src/twitter_video_dl/login.py index bc2ab6f..013ce52 100644 --- a/src/twitter_video_dl/login.py +++ b/src/twitter_video_dl/login.py @@ -1,50 +1,64 @@ import requests import json -class TwitterLoginClient: - def __init__(self, bear_token,proxies={}): - self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36" - self.bear_token = bear_token - self.proxies = proxies - self.session = requests.session() - self.__twitter() - self.x_guest_token = self.get_guest_token() - self.flow_token = None - - def login(self,username,password): - +class TwitterLoginClient: + def __init__(self, bear_token, proxies={}): + self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36" + self.authorization = bear_token + self.proxies = proxies + self.session = requests.session() + #self.__twitter() + self.x_guest_token = self.__get_guest_token() + self.flow_token = None + self.cookie = None - def __twitter(self): - headers = { - "User-Agent":self.USER_AGENT, + def __get_headers(self): + return { + "authorization": self.authorization, + "User-Agent": self.user_agent, + "Content-type": "application/json", + "x-guest-token": self.x_guest_token, + # "x-csrf-token": self.session.cookies.get("ct0"), + "x-twitter-active-user": "yes", + "x-twitter-client-language": "en", } - response = self.session.get( - "https://twitter.com/", headers=headers, proxies=self.proxies - ) - + def __flow_token_check(self): + if self.flow_token is None: + raise Exception("not found token") + + def __error_check(self, content): + if content.get("errors"): + return True + return False - def get_guest_token(self): - headers = { - "authorization": self.AUTHORIZATION, - "User-Agent": self.USER_AGENT, + def __get_guest_token(self): + """get guest token""" + headers = { + "authorization": self.authorization, + "User-Agent": self.user_agent, } - response = self.session.post( + response = self.session.post( "https://api.twitter.com/1.1/guest/activate.json", headers=headers, proxies=self.proxies, ).json() - return response["guest_token"] - + return response["guest_token"] - def login_flow(self): + def login(self, username=None, password=None): + """launch login flow""" + print(username+"&"+password) + self.username = username + self.password = password data = { "input_flow_data": { "flow_context": { "debug_overrides": {}, - "start_location": {"location": "splash_screen"}, + "start_location": { + "location": "splash_screen" + }, } }, "subtask_versions": { @@ -63,35 +77,41 @@ def login_flow(self): params=params, proxies=self.proxies, ).json() - self.__error_check(response) - self.flow_token = response.get("flow_token") - self.content = response - return self + print(response) + if not self.__error_check(response): + + self.flow_token = response.get("flow_token") + self.content = response + self.__login_js() - def LoginJsInstrumentationSubtask(self): + def __login_js(self): self.__flow_token_check() - self.__method_check("LoginJsInstrumentationSubtask") + print("--------flow_login_js") data = { - "flow_token": self.flow_token, - "subtask_inputs": [ - { - "subtask_id": "LoginJsInstrumentationSubtask", - "js_instrumentation": { - "response": json.dumps( - { - "rf": { - "af07339bbc6d24ced887d705eb0c9fd29b4a7d7ddc21136c9f94d53a4bc774d2": 88, - "a6ce87d6481c6ec4a823548be3343437888441d2a453061c54f8e2eb325856f7": 250, - "a0062ad06384a8afd38a41cd83f31b0dbfdea0eff4b24c69f0dd9095b2fb56d6": 16, - "a929e5913a5715d93491eaffaa139ba4977cbc826a5e2dbcdc81cae0f093db25": 186, - }, - "s": "Q-H-53m1uXImK0F0ogrxRQtCWTH1KIlPbIy0MloowlMa4WNK5ZCcDoXyRs1q_cPbynK73w_wfHG_UVRKKBWRoh6UJtlPS5kMa1p8fEvTYi76hwdzBEzovieR8t86UpeSkSBFYcL8foYKSp6Nop5mQR_QHGyEeleclCPUvzS0HblBJqZZdtUo-6by4BgCyu3eQ4fY5nOF8fXC85mu6k34wo982LMK650NsoPL96DBuloqSZvSHU47wq2uA4xy24UnI2WOc6U9KTvxumtchSYNnXq1HV662B8U2-jWrzvIU4yUHV3JYUO6sbN6j8Ho9JaUNJpJSK7REwqCBQ3yG7iwMAAAAX2Vqcbs", - } - ), - "link": "next_link", - }, - } - ], + "flow_token": + self.flow_token, + "subtask_inputs": [{ + "subtask_id": "LoginJsInstrumentationSubtask", + "js_instrumentation": { + "response": + json.dumps({ + "rf": { + "af07339bbc6d24ced887d705eb0c9fd29b4a7d7ddc21136c9f94d53a4bc774d2": + 88, + "a6ce87d6481c6ec4a823548be3343437888441d2a453061c54f8e2eb325856f7": + 250, + "a0062ad06384a8afd38a41cd83f31b0dbfdea0eff4b24c69f0dd9095b2fb56d6": + 16, + "a929e5913a5715d93491eaffaa139ba4977cbc826a5e2dbcdc81cae0f093db25": + 186, + }, + "s": + "Q-H-53m1uXImK0F0ogrxRQtCWTH1KIlPbIy0MloowlMa4WNK5ZCcDoXyRs1q_cPbynK73w_wfHG_UVRKKBWRoh6UJtlPS5kMa1p8fEvTYi76hwdzBEzovieR8t86UpeSkSBFYcL8foYKSp6Nop5mQR_QHGyEeleclCPUvzS0HblBJqZZdtUo-6by4BgCyu3eQ4fY5nOF8fXC85mu6k34wo982LMK650NsoPL96DBuloqSZvSHU47wq2uA4xy24UnI2WOc6U9KTvxumtchSYNnXq1HV662B8U2-jWrzvIU4yUHV3JYUO6sbN6j8Ho9JaUNJpJSK7REwqCBQ3yG7iwMAAAAX2Vqcbs", + }), + "link": + "next_link", + }, + }], } response = self.session.post( "https://twitter.com/i/api/1.1/onboarding/task.json", @@ -99,30 +119,82 @@ def LoginJsInstrumentationSubtask(self): json=data, proxies=self.proxies, ).json() - self.__error_check(response) - self.flow_token = response.get("flow_token") - self.content = response - return self - def LoginEnterUserIdentifierSSO(self, user_id): + if not self.__error_check(response): + self.flow_token = response.get("flow_token") + self.content = response + self.__login_sso() + + def __login_sso(self): self.__flow_token_check() - self.__method_check("LoginEnterUserIdentifierSSO") + print("--------flow_login_sso") data = { - "flow_token": self.flow_token, - "subtask_inputs": [ - { - "subtask_id": "LoginEnterUserIdentifierSSO", - "settings_list": { - "setting_responses": [ - { - "key": "user_identifier", - "response_data": {"text_data": {"result": user_id}}, + "flow_token": + self.flow_token, + "subtask_inputs": [{ + "subtask_id": "LoginEnterUserIdentifierSSO", + "settings_list": { + "setting_responses": [{ + "key": "user_identifier", + "response_data": { + "text_data": { + "result": self.username } - ], - "link": "next_link", - }, - } - ], + }, + }], + "link": + "next_link", + }, + }], + } + response = self.session.post( + "https://twitter.com/i/api/1.1/onboarding/task.json", + headers=self.__get_headers(), + json=data, + proxies=self.proxies, + ).json() + if not self.__error_check(response): + self.flow_token = response.get("flow_token") + self.content = response + self.__login_password() + + def __login_password(self): + self.__flow_token_check() + print("--------flow_login_password") + data = { + "flow_token": + self.flow_token, + "subtask_inputs": [{ + "subtask_id": "LoginEnterPassword", + "enter_password": { + "password": self.password, + "link": "next_link" + }, + }], + } + response = self.session.post( + "https://twitter.com/i/api/1.1/onboarding/task.json", + headers=self.__get_headers(), + json=data, + proxies=self.proxies, + ).json() + if not self.__error_check(response): + self.flow_token = response.get("flow_token") + self.content = response + self.__login_check() + + def __login_check(self): + self.__flow_token_check() + print("--------flow_login_check") + data = { + "flow_token": + self.flow_token, + "subtask_inputs": [{ + "subtask_id": "AccountDuplicationCheck", + "check_logged_in_account": { + "link": "AccountDuplicationCheck_false" + }, + }], } response = self.session.post( "https://twitter.com/i/api/1.1/onboarding/task.json", @@ -130,7 +202,7 @@ def LoginEnterUserIdentifierSSO(self, user_id): json=data, proxies=self.proxies, ).json() - self.__error_check(response) - self.flow_token = response.get("flow_token") - self.content = response - return self + if not self.__error_check(response): + self.flow_token = response.get("flow_token") + self.content = response + self.cookie = self.session.cookies From 0a81e90d3c344832becb88e5a434b298721faf8e Mon Sep 17 00:00:00 2001 From: moo611 Date: Sat, 29 Apr 2023 21:03:14 +0800 Subject: [PATCH 3/4] login --- src/twitter_video_dl/login.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/src/twitter_video_dl/login.py b/src/twitter_video_dl/login.py index 013ce52..fa142be 100644 --- a/src/twitter_video_dl/login.py +++ b/src/twitter_video_dl/login.py @@ -3,6 +3,11 @@ class TwitterLoginClient: + """ + name:TwitterLoginClient + author:Desong + date:2023-04-29 + """ def __init__(self, bear_token, proxies={}): self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.45 Safari/537.36" @@ -13,6 +18,7 @@ def __init__(self, bear_token, proxies={}): self.x_guest_token = self.__get_guest_token() self.flow_token = None self.cookie = None + self.csrftoken = None def __get_headers(self): return { @@ -48,8 +54,10 @@ def __get_guest_token(self): return response["guest_token"] def login(self, username=None, password=None): - """launch login flow""" - print(username+"&"+password) + """ + launch login flow + """ + print(username + "&" + password) self.username = username self.password = password data = { @@ -79,7 +87,7 @@ def login(self, username=None, password=None): ).json() print(response) if not self.__error_check(response): - + self.flow_token = response.get("flow_token") self.content = response self.__login_js() @@ -205,4 +213,8 @@ def __login_check(self): if not self.__error_check(response): self.flow_token = response.get("flow_token") self.content = response - self.cookie = self.session.cookies + self.cookie = self.session.cookies.get_dict() + self.csrftoken = self.__extract_csrf_token() + + def __extract_csrf_token(self): + return self.cookie["ct0"] From 7d7ca438ab4cdc5e985bf9b7223155ea30f79884 Mon Sep 17 00:00:00 2001 From: moo611 Date: Sat, 29 Apr 2023 21:20:13 +0800 Subject: [PATCH 4/4] login test --- tests/TestLogin.py | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) create mode 100644 tests/TestLogin.py diff --git a/tests/TestLogin.py b/tests/TestLogin.py new file mode 100644 index 0000000..bd9fa22 --- /dev/null +++ b/tests/TestLogin.py @@ -0,0 +1,33 @@ +import sys +import os + +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from src.twitter_video_dl.login import TwitterLoginClient +import argparse + +#this temp bearertoken can be changed by the token from main.js. +BEARER_TOKEN = "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" + +if __name__ == "__main__": + + parser = argparse.ArgumentParser(description="Test for Twiiter login") + + parser.add_argument( + "-a", + required=True, + type=str, + help="username, email or phone number. e.g. +8613300000000") + + parser.add_argument("-p", + required=True, + type=str, + help="The password of the account") + + args = parser.parse_args() + #proxy = {"http": "http://127.0.0.1:7890"} + login_client = TwitterLoginClient(bear_token=BEARER_TOKEN) + login_client.login(username=args.a, password=args.p) + + print(login_client.cookie) + print(login_client.csrftoken) \ No newline at end of file