Skip to content

Instantly share code, notes, and snippets.

@ceres-c

ceres-c/twitter-register.py

Created November 1, 2020 19:26
Show Gist options
  • Star 7 You must be signed in to star a gist
  • Fork 2 You must be signed in to fork a gist
  • Save ceres-c/7c16a40c10cb476cce2c4b9023341377 to your computer and use it in GitHub Desktop.
Save ceres-c/7c16a40c10cb476cce2c4b9023341377 to your computer and use it in GitHub Desktop.
Download ZIP
Attempt at registering Twitter accounts with Python Requests. NOT working.
Raw
twitter-register.py
#! /bin/python
import json
import random
import re
import string
import time
import urllib.parse
from threading import Thread
from time import sleep, time_ns
import requests
from requests.api import request
from requests.sessions import session
from selenium import webdriver
init_api_url = "https://twitter.com:443/i/api/1.1/branch/init.json"
onboarding_signup_url = "https://twitter.com:443/i/api/1.1/onboarding/task.json?flow_name=signup"
onboarding_url = "https://twitter.com:443/i/api/1.1/onboarding/task.json"
check_email_url = "https://twitter.com:443/i/api/i/users/email_available.json?email={}"
client_events_url = "https://api.twitter.com:443/1.1/jot/client_event.json"
auth_bearer_js_url = "https://abs.twimg.com:443/responsive-web/client-web/main.6c1aeb65.js"
twitter_sess_url = "https://twitter.com:443/i/js_inst?c_name=ui_metrics"
password_strength_url = "https://twitter.com:443/i/api/1.1/account/password_strength.json"
twitter_sess_header = {
"Connection": "close",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"Accept": "*/*",
"Sec-Fetch-Site": "same-origin",
"Sec-Fetch-Mode": "no-cors",
"Sec-Fetch-Dest": "script",
"Referer": "https://twitter.com/",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
client_event_headers_OPTIONS = {
"Connection": "close",
"Accept": "*/*",
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "authorization,x-csrf-token,x-guest-token,x-twitter-active-user,x-twitter-client-language",
"Origin": "https://twitter.com",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
"Sec-Fetch-Dest": "empty",
"Referer": "https://twitter.com/",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
client_event_headers_POST = {
"Connection": "close",
# "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA",
"x-twitter-client-language": "en",
# "x-csrf-token": "54b9d9e87a4fef1889d060be00437558",
# "x-guest-token": "1321170979104632838",
"x-twitter-active-user": "yes",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"content-type": "application/x-www-form-urlencoded",
"Accept": "*/*",
"Origin": "https://twitter.com",
"Sec-Fetch-Site": "same-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": "https://twitter.com/", # This entry will be changed later on
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
init_api_headers = {
"Connection": "close",
# "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA",
"x-twitter-client-language": "en",
# "x-csrf-token": "54b9d9e87a4fef1889d060be00437558",
# "x-guest-token": "1321170979104632838",
"x-twitter-active-user": "yes",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"content-type": "application/json",
"Accept": "*/*",
"Origin": "https://twitter.com",
"Sec-Fetch-Site": "same-origin",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": "https://twitter.com/",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
personalization_id_headers = {
"Connection": "close",
"Upgrade-Insecure-Requests": "1",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9",
"Sec-Fetch-Site": "none",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-User": "?1",
"Sec-Fetch-Dest": "document",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
auth_bearer_js_headers = {
"Connection": "close",
"Origin": "https://twitter.com",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"Accept": "*/*",
"Sec-Fetch-Site": "cross-site",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "script",
"Referer": "https://twitter.com/",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
onboarding_headers = {
"Connection": "close",
# "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA",
"x-twitter-client-language": "en",
# "x-csrf-token": "54b9d9e87a4fef1889d060be00437558",
# "x-guest-token": "1321170979104632838",
"x-twitter-active-user": "yes",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"content-type": "application/json",
"Accept": "*/*",
"Origin": "https://twitter.com",
"Sec-Fetch-Site": "same-origin",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": "https://twitter.com/i/flow/signup",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
optout_headers = {
"Connection": "close",
# "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA",
"x-twitter-client-language": "en",
# "x-csrf-token": "54b9d9e87a4fef1889d060be00437558",
# "x-guest-token": "1321170979104632838",
"x-twitter-active-user": "yes",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"content-type": "application/x-www-form-urlencoded",
"Accept": "*/*",
"Origin": "https://twitter.com",
"Sec-Fetch-Site": "same-origin",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": "https://twitter.com/i/flow/signup",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
check_email_header = {
"Connection": "close",
# "authorization": "Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA",
"x-twitter-client-language": "en",
# "x-csrf-token": "54b9d9e87a4fef1889d060be00437558",
# "x-guest-token": "1321170979104632838",
"x-twitter-active-user": "yes",
# "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36",
"Accept": "*/*",
"Sec-Fetch-Site": "same-origin",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Dest": "empty",
"Referer": "https://twitter.com/i/flow/signup",
"Accept-Encoding": "gzip, deflate",
"Accept-Language": "en-US,en;q=0.9"
}
def time_ms():
return time.time_ns() // 1000000
def get_random_string(len):
return ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(len))
def get_random_number(len):
return ''.join(random.choice(string.digits) for _ in range(len))
# Add cookies: guest_id, personalization_id, gt, _twitter_sess
def get_twitter_cookies(session):
personalization_id_url = "https://twitter.com:443/"
response = session.get(personalization_id_url, headers=personalization_id_headers) # guest_id, personalization_id
if response is not None:
response_gt = response.text
else:
print("[!] Error downloading guest_id, personalization_id, gt cookies")
return False
gt_pos = response_gt.find("decodeURIComponent")
gt = None
if gt_pos != -1:
gt = response_gt[gt_pos + 23:gt_pos + 23 + 19] # len(decodeURIComponent) = 23. len(gt) = 19
cookie_obj = requests.cookies.create_cookie(domain=".twitter.com", path="/", name="gt", value=gt) # gt
session.cookies.set_cookie(cookie_obj)
if gt is None:
print("[!] Error downloading gt cookie")
return False
for cookie in session.cookies:
if cookie.name not in ["guest_id", "personalization_id", "gt"]: # Cleanup cookies (no mobile ones)
del session.cookies[cookie.name]
# A random cookie
ct0_cookie_obj = requests.cookies.create_cookie(domain=".twitter.com", path="/", name="ct0", value=get_random_string(32))
session.cookies.set_cookie(ct0_cookie_obj)
return True
# Add to session all the random/fixed cookies which do not need requests
def generate_static_cookies(session):
cookie_timestamp = str(int(time.time())) # Used for _ga and _gid cookies
cookie_ver = "GA1"
cookie_subver = "2" # WARNING check for updates in this cookie field for both _ga and _gid cookies
# session.cookies["_ga"] = '.'.join( [cookie_ver, cookie_subver, get_random_number(9), cookie_timestamp] )
ga_cookie = '.'.join( [cookie_ver, cookie_subver, get_random_number(9), cookie_timestamp])
ga_cookie_obj = requests.cookies.create_cookie(domain=".twitter.com", path="/", name="_ga", value=ga_cookie)
session.cookies.set_cookie(ga_cookie_obj)
gid_cookie = '.'.join( [cookie_ver, cookie_subver, get_random_number(9), cookie_timestamp] )
gid_cookie_obj = requests.cookies.create_cookie(domain=".twitter.com", path="/", name="_gid", value=gid_cookie)
session.cookies.set_cookie(gid_cookie_obj)
# Set uniform User-Agent for every header
def configure_user_agent(user_agent):
client_event_headers_POST["User-Agent"] = user_agent
init_api_headers["User-Agent"] = user_agent
onboarding_headers["User-Agent"] = user_agent
optout_headers["User-Agent"] = user_agent
check_email_header["User-Agent"] = user_agent
client_event_headers_OPTIONS["User-Agent"] = user_agent
personalization_id_headers["User-Agent"] = user_agent
auth_bearer_js_headers["User-Agent"] = user_agent
twitter_sess_header["User-Agent"] = user_agent
# Add authorization bearer, x-csrf-token, x-guest-token
def update_headers(session):
response = requests.get(auth_bearer_js_url, headers=auth_bearer_js_headers)
if response is not None:
response_js = response.text
else:
print("[!] Error downloading 'Bearer' identifier")
return False
bearer_pos = response_js.find("Web-12") + 11
bearer = response_js[bearer_pos:bearer_pos+104]
client_event_headers_POST["authorization"] = "Bearer {}".format(bearer)
client_event_headers_POST["x-csrf-token"] = session.cookies['ct0']
client_event_headers_POST["x-guest-token"] = session.cookies['gt']
init_api_headers["authorization"] = "Bearer {}".format(bearer)
init_api_headers["x-csrf-token"] = session.cookies['ct0']
init_api_headers["x-guest-token"] = session.cookies['gt']
onboarding_headers["authorization"] = "Bearer {}".format(bearer)
onboarding_headers["x-csrf-token"] = session.cookies['ct0']
onboarding_headers["x-guest-token"] = session.cookies['gt']
optout_headers["authorization"] = "Bearer {}".format(bearer)
optout_headers["x-csrf-token"] = session.cookies['ct0']
optout_headers["x-guest-token"] = session.cookies['gt']
check_email_header["authorization"] = "Bearer {}".format(bearer)
check_email_header["x-csrf-token"] = session.cookies['ct0']
check_email_header["x-guest-token"] = session.cookies['gt']
# Generate twitter client event logs. Should not be called directly but only by send_twitter_client_event
# Supported event types: storePrepare, empty, heartbeat, client_network_request_event, client_event
def twitter_client_event_log(events, sequence): # TODO description request
twitter_client_events_client_app_id = "3033300"
ret_sequence = sequence
ret_events = []
for event in events:
event_type = event["type"]
if event_type == "storePrepare":
parameters = "category=perftown"
ret_events.extend(
[
{
"description": "rweb:init:storePrepare",
"product": "rweb",
"duration_ms": random.randint(15,25)
},
{
"description": "rweb:ttft:perfSupported",
"product": "rweb",
"duration_ms": random.randint(1,5)
},
{
"description": "rweb:ttft:connect",
"product": "rweb",
"duration_ms": random.randint(30,70)
},
{
"description": "rweb:ttft:process",
"product": "rweb",
"duration_ms": random.randint(500,1000)
},
{
"description": "rweb:ttft:response",
"product": "rweb",
"duration_ms": random.randint(1,5)
},
{
"description": "rweb:ttft:interactivity",
"product": "rweb",
"duration_ms": random.randint(800,1500)
}
]
)
# No changes to ret_sequence in this case
elif event_type == "empty":
return (ret_sequence, "") # They wanted an empty string, so we gave them an empty string
# No changes to ret_sequence in this case
elif event_type == "heartbeat":
parameters = "category=perftown"
ret_events.append(
{
"description": "rweb:heartbeat:health:false",
"product": "rweb"
})
# No changes to ret_sequence in this case
elif event_type == "client_network_request_event": # Event paramteres: triggered_on, start_time_ms, uri_host_name, uri_path, request_body_size
parameters = "" # TODO check this
ret_events.append(
{
"_category_": "client_network_request_event",
"format_version": 2,
"triggered_on": event["triggered_on"], # 1603806896071
"request": {
"uri_scheme": "https:",
"uri_host_name": event["uri_host_name"], # "api.twitter.com"
"uri_path": event["uri_path"], # "/1.1/account/personalization/sync_optout_settings.json"
"uri_query": "",
"http_method": "POST",
"http_status_code": 200,
"start_time_ms": event["start_time_ms"], # 1603806895496
"request_details": {
"duration_ms": random.randint(300,700),
"request_body_size": event["request_body_size"] # 0
}
},
"common_header": {
"commonHeader": {
"clientHeader": {
"timestampMs": event["triggered_on"], # 1603806896071
"timezoneOffsetMin": 60
}
}
},
"network_measurements": {
"connection_type": "unknown",
"speed_class": "3g",
"download_mbps": 1.4,
"rtt_ms": 400,
"reduced_data_usage": False
},
"event_type": "api:all",
"event_source": "rweb",
"client_app_id": twitter_client_events_client_app_id
}
)
ret_sequence += 1
elif event_type == "client_event": # Event paramteres: triggered_on, event_namespace, [items], sequence_start_timestamp
# if "element" in event: # If key "element" is present
# event_namespace = {
# "page": event["page"], # "onboarding"
# "element": event["element"], # "view"
# "action": event["action"], # "show"
# "client": "m5"
# }
# else:
# event_namespace = {
# "page": event["page"], # "onboarding"
# "action": event["action"], # "show"
# "client": "m5"
# }
parameters = "debug=true"
event_namespace = event["event_namespace"]
event_namespace["client"] = "m5"
new_dict = {
"_category_": "client_event",
"format_version": 2,
"triggered_on": event["triggered_on"], # 1603806894895
"event_namespace": event_namespace,
"client_event_sequence_start_timestamp": event["sequence_start_timestamp"], # 1603806886131
"client_event_sequence_number": ret_sequence,
"client_app_id": twitter_client_events_client_app_id
}
if "items" in event: # If key "items" is present
new_dict["items"] = event["items"]
ret_events.append(new_dict)
ret_sequence += 1
json_string = json.dumps(ret_events, separators=(",", ":")) # Me no likey spaces
ret_string = "{}&log={}".format(parameters, urllib.parse.quote(json_string))
return (ret_sequence, ret_string)
# Send twitter client event logs with the appropriate request type
def send_twitter_client_event(session, events, sequence=None):
if len(events) == 1 and events[0]["type"] == "empty":
response = requests.options(client_events_url, headers=client_event_headers_OPTIONS)
ret_sequence = sequence
else:
ret_sequence, data = twitter_client_event_log(events, sequence)
response = session.post(client_events_url, headers=client_event_headers_POST, data=data)
return (True, ret_sequence) if response else (False, ret_sequence)
def solve_js_challenge(js_chal, driver):
functions = re.findall(r'function \w{20}', js_chal)
internal_func_name = functions[1].split(" ")[1]
func_body_begin = js_chal.find("function " + internal_func_name + "()")
func_body_end = js_chal.find("};};")
response_text_body = js_chal[func_body_begin + 33 : func_body_end + 2]
return json.dumps(driver.execute_script(response_text_body), separators=(",", ":")).replace('"', '\\"')
# Send heartbeats every 20 seconds. Must be started by a thread
def keepalive(session):
while True:
time.sleep(20)
if G.kill: # Check right before firing a new heartbeat
G.kill = False
return
send_twitter_client_event(session, [{"type": "heartbeat"}])
def register_twitter(session):
email_address, password = input("Input email address: "), "secure!password"
username = email_address.split("@")[0]
client_event_sequence = 0
configure_user_agent("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.111 Safari/537.36")
cookies_status = get_twitter_cookies(session)
if not cookies_status:
print("[!] Error downloading cookies")
return(1)
update_headers(session)
# client event NaN - OPTIONS
success, client_event_sequence = send_twitter_client_event(session, [{"type": "empty"}], client_event_sequence)
# client event NaN bis - StorePrepare
success, client_event_sequence = send_twitter_client_event(session, [{"type": "storePrepare"}], client_event_sequence)
# client event 0 - front: show
sequence_start = time_ms()
client_events = [
{"type": "client_event", "triggered_on": sequence_start, "items": [], "event_namespace":
{
"page": "front", "action": "show"
}, "sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Download javascript challenge + `_twitter_sess` cookie
js_chal = session.get(twitter_sess_url, headers=twitter_sess_header)
# Add fixed cookies
generate_static_cookies(session)
# client event 1,2 - smartlock: prompt, smartlock: no credentials
triggered_0 = sequence_start
triggered_1 = triggered_0 + random.randrange(20, 100)
client_events = [
{
"type": "client_event", "triggered_on": triggered_0, "event_namespace":
{
"page": "app", "component": "smartlock_prompt", "action": "impression"
},
"items": [], "sequence_start_timestamp": sequence_start
},
{
"type": "client_event", "triggered_on": triggered_1, "event_namespace":
{
"page": "app", "component": "smartlock_prompt", "element": "no_credentials", "action": "cancel"
},
"items": [], "sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Static cookie appearing right here
sl_cookie_obj = requests.cookies.create_cookie(domain=".twitter.com", path="/", name="_sl", value="1")
session.cookies.set_cookie(sl_cookie_obj)
# Init registration flow
session.post(init_api_url, headers=init_api_headers, json={}) # Ignore response
# Update client_event header
client_event_headers_POST["Referer"] = "https://twitter.com/i/flow/signup"
# client event 3 - signup_callout
triggered_3 = time_ms()
client_events = [
{
"type": "client_event", "triggered_on": triggered_3, "event_namespace":
{
"page":"front","section":"front","component":"signup_callout","element":"form","action":"signup"
}, "sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Begin onboarding
onboarding_signup_json={"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}}
response = session.post(onboarding_signup_url, headers=onboarding_headers, json=onboarding_signup_json)
flow_token = response.json()["flow_token"]
# Sync optouts (?)
optout_url = "https://twitter.com:443/i/api/1.1/account/personalization/sync_optout_settings.json"
session.post(optout_url, headers=optout_headers) # Ignore response
# download javascript again (it's different), `_twitter_sess` cookie will not be changed
js_chal = session.get(twitter_sess_url, headers=twitter_sess_header)
# client event 4, 5 - Signup
triggered_4 = time_ms()
triggered_5 = triggered_4 + random.randrange(100, 500)
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_4, "event_namespace": {
"page": "onboarding", "action": "show"
},
"sequence_start_timestamp": sequence_start
},
{
"type": "client_event", "triggered_on": triggered_5, "event_namespace": {
"page": "onboarding", "element": "view", "action": "impression"
}, "items": [
{"token": flow_token, "name": "Signup"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# client event 6 - onboarding
triggered_6 = time_ms()
client_events = [
{
"type": "client_event", "triggered_on": triggered_6, "event_namespace":
{
"page": "onboarding","component": "signup","element": "email","action": "choose"
},
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Check if email is available
session.get(check_email_url.format(email_address), headers=check_email_header) # TODO actually check email status from response
# client event 7 - email_next_link
triggered_7 = time_ms()
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_7, "event_namespace":
{
"page": "onboarding", "element": "link", "action": "click"
},
"items": [
{"token": flow_token, "name": "Signup", "description": "email_next_link"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# client event 8 - SignupSettingsListEmail
triggered_8 = time_ms()
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_8, "event_namespace": {
"page": "onboarding", "element": "view", "action": "impression"
}, "items": [
{"token": flow_token, "name": "SignupSettingsListEmail"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# client event 9 - SignupSettingsListEmail, next_link
triggered_9 = time_ms()
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_9, "event_namespace": {
"page": "onboarding", "element": "link", "action": "click"
}, "items": [
{"token": flow_token, "name": "SignupSettingsListEmail", "description": "next_link"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Ask for verification code
begin_verification_url = "https://twitter.com:443/i/api/1.1/onboarding/begin_verification.json"
begin_verification_json={"email":email_address,"display_name":username,"flow_token":flow_token}
session.post(begin_verification_url, headers=onboarding_headers, json=begin_verification_json) # Ignore response
# client events 10, 11, 12, 13, 14
triggered_10 = time_ms()
triggered_11 = triggered_10 + random.randrange(1500, 2000)
triggered_12 = triggered_11 + random.randrange(1, 10)
triggered_13 = triggered_12 + random.randrange(1, 10)
triggered_14 = triggered_13 + random.randrange(1, 10)
client_events = [
{ # 10
"type": "client_event", "triggered_on": triggered_10, "event_namespace": { # TODO sometimes this event is invoked before asking for verification code
"page": "onboarding", "element": "view", "action": "impression"
}, "items": [
{"token": flow_token, "name": "SignupReview"}
],
"sequence_start_timestamp": sequence_start
},
{ # 11
"type": "client_event", "triggered_on": triggered_11, "event_namespace": {
"page": "onboarding", "element": "link", "action": "click"
}, "items": [
{"token": flow_token, "name": "SignupReview", "description": "signup_with_email_next_link"}
],
"sequence_start_timestamp": sequence_start
},
{ # 12
"type": "client_event", "triggered_on": triggered_12, "event_namespace": {
"page": "onboarding", "element": "view", "action": "impression"
}, "items": [
{"token": flow_token, "name": "EmailVerificationSubtaskLink"}
],
"sequence_start_timestamp": sequence_start
},
{ # 13
"type": "client_event", "triggered_on": triggered_13, "event_namespace": {
"page": "onboarding", "element": "link", "action": "click"
}, "items": [
{"token": flow_token, "name": "EmailVerificationSubtaskLink"}
],
"sequence_start_timestamp": sequence_start
},
{ # 14
"type": "client_event", "triggered_on": triggered_14, "event_namespace": {
"page": "onboarding", "element": "view", "action": "impression"
}, "items": [
{"token": flow_token, "name": "EmailVerification"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Solve javascript challenge
js_chal_response = solve_js_challenge(js_chal.text, driver)
# Prepare heartbeat thread
t = Thread(target=keepalive, args=(session,))
t.start()
# Get email verification code
twitter_verif_code = input("Twitter verification code: ")
G.kill = True # Stop heartbeat
# client event 15 - EmailVerification, next_link
triggered_15 = time_ms()
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_15, "event_namespace": {
"page": "onboarding", "element": "link", "action": "click"
}, "items": [
{"token": flow_token, "name": "EmailVerification", "description": "next_link"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Send verification code + challenge response
onboarding_send_code_string='{{"flow_token": "{}", "subtask_inputs": [{{"subtask_id": "Signup", "sign_up": {{"js_instrumentation": {{"response": "{}"}}, "link": "email_next_link", "name": "{}", "email": "{}", "birthday": {{"day": {}, "month": {}, "year": {}}}}}}}, {{"subtask_id": "SignupSettingsListEmail", "settings_list": {{"setting_responses": [{{"key": "allow_emails_about_activity", "response_data": {{"boolean_data": {{"result": false}}}}}}, {{"key": "find_by_email", "response_data": {{"boolean_data": {{"result": false}}}}}}, {{"key": "personalize_ads", "response_data": {{"boolean_data": {{"result": false}}}}}}], "link": "next_link"}}}}, {{"subtask_id": "SignupReview", "sign_up_review": {{"link": "signup_with_email_next_link"}}}}, {{"subtask_id": "EmailVerification", "email_verification": {{"code": "{}", "email": "{}", "link": "next_link"}}}}]}}'.format(flow_token, js_chal_response, username, email_address, 15, 8, 1996, twitter_verif_code, email_address)
response = session.post(onboarding_url, headers=onboarding_headers, data=onboarding_send_code_string)
# Example of succesful email verification: TODO remove
#{"flow_token":"g;160389851335133523:-1603898515311:Ndo239cQQ7kcizkfOfIdvV1r:2","status":"success","subtasks":[{"subtask_id":"EnterPassword","enter_password":{"primary_text":{"text":"You'll need a password","entities":[]},"next_link":{"link_type":"task","link_id":"next_link","label":"Next"},"secondary_text":{"text":"Make sure it’s 8 characters or more.","entities":[]},"hint":"Password","name":"armando07768","username":"armando07768","email":"armando07768@stempmail.com"},"progress_indication":{"text":{"text":"Step 5 of 5","entities":[]}}}]}
if not response:
print("[!] Error finalizing registration")
print(response.text)
return False
finalize_json = response.json()
if not "subtasks" in finalize_json or len(finalize_json["subtasks"]) < 1 or not "subtask_id" in finalize_json["subtasks"][0]:
print("[!] Missing Subtaks in response json during finalize, got:")
print(response.text)
return False
if finalize_json["subtasks"][0]["subtask_id"] == "EnterPhoneForVerification":
print("[!] SMS confirmation required. Try changing IP address")
return False
# client event 16 - EnterPassword
triggered_16 = time_ms()
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_16, "event_namespace": {
"page": "onboarding", "element": "view", "action": "impression"
}, "items": [
{"token": flow_token, "name": "EnterPassword"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# Check password strength
password_strength_data = {"password": password, "username": username} # TODO check this is actually sent as URL parameters data
session.post(password_strength_url, headers=optout_headers, data=password_strength_data) # TODO do something with this response
# client event 17 - entered_password
triggered_17 = time_ms()
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_17, "event_namespace": {
"page": "onboarding", "component": "password_entry", "element": "entered_password", "action": "valid"
},
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
# client event 18 - entered_password, click
triggered_18 = time_ms()
client_events = [ # triggered_on, page, [element], [items], action, sequence_start_timestamp
{
"type": "client_event", "triggered_on": triggered_18, "event_namespace": {
"page": "onboarding", "element": "link", "action": "click",
}, "items": [
{"token": flow_token, "name": "EnterPassword", "description": "next_link"}
],
"sequence_start_timestamp": sequence_start
}
]
success, client_event_sequence = send_twitter_client_event(session, client_events, client_event_sequence)
enter_password_json={"flow_token":flow_token,"subtask_inputs":[{"subtask_id":"EnterPassword","enter_password":{"password":password,"link":"next_link"}}]}
response = session.post(onboarding_url, headers=onboarding_headers, data=enter_password_json) # WARNING This request always yields an error (see response)
print("SET PASSWORD REQUEST:") # TODO remove
print(enter_password_json) # TODO remove
print("SET PASSWORD RESPONSE:") # TODO remove
print(response.text) # TODO remove
print("[#] Successfully registered as {}:{}".format(email_address, password))
return True
driver = webdriver.Chrome()
print("\t[*] Session URL: {}".format(driver.command_executor._url))
print("\t[*] Session ID: {}".format(driver.session_id))
class globalVars():
pass
G = globalVars() #empty object to pass around global state
G.kill = False
while True:
session = requests.session()
# Debug code
# proxies = {
# 'http': 'http://127.0.0.1:8080',
# 'https': 'http://127.0.0.1:8080'
# }
#session.proxies.update(proxies)
#session.verify = False
registered = register_twitter(session)
if registered:
break
driver.get("http://google.com")
for cookie in session.cookies :
print(cookie)
for cookie in session.cookies :
driver.add_cookie({'name': cookie.name, 'value': cookie.value, 'path': cookie.path, 'domain': cookie.domain})
driver.get("http://twitter.com")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment