soopy.moe
1# nitter-guest-account.py
2# cross-platform port of https://github.com/zedeus/nitter/issues/983#issuecomment-1681199357
3# thank you!
4import sys
5import json
6import time
7import random
8import typing
9import traceback
10from base64 import b64encode
11from argparse import ArgumentParser
12
13try:
14 import requests
15except ImportError:
16 print("\x1b[31m[!] Could not import `requests`.")
17 print("\x1b[31m[!] This script requires the requests module to be installed.")
18 print("\x1b[31m[!] We apologize but using plain http.client is way too painful."
19 " Please reach out with a PR if you would like to change that!")
20 sys.exit(1)
21
22verbose = False
23noprettyprint = False
24
25# Constants
26CONSUMER_KEY = "3nVuSoBZnx6U4vzUxf5w"
27CONSUMER_SECRET = "Bcs59EFbbsdF6Sl9Ng71smgStWEGwXXKSjYvPVt7qys"
28EXPECTED_BEARER_TOKEN = "Bearer AAAAAAAAAAAAAAAAAAAAAFXzAwAAAAAAMHCxpeSDG1gLNLghVe8d74hl6k4%3DRUMF4xAQLsbeBhTSRrCiQpJtxoGWeyHrDb5te2jpGskWDFW82F"
29BASE_REQUEST_HEADERS = {
30 'Content-Type': 'application/json',
31 'User-Agent': 'TwitterAndroid/9.95.0-release.0 (29950000-r-0) ONEPLUS+A3010/9 (OnePlus;ONEPLUS+A3010;OnePlus;OnePlus3;0;;1;2016)',
32 'X-Twitter-API-Version': '5',
33 'X-Twitter-Client': 'TwitterAndroid',
34 'X-Twitter-Client-Version': '9.95.0-release.0',
35 'OS-Version': '28',
36 'System-User-Agent': 'Dalvik/2.1.0 (Linux; U; Android 9; ONEPLUS A3010 Build/PKQ1.181203.001)',
37 'X-Twitter-Active-User': 'yes',
38}
39
40BASE_URL = "https://api.twitter.com"
41BEARER_TOKEN_ENDPOINT = "/oauth2/token"
42GUEST_TOKEN_ENDPOINT = "/1.1/guest/activate.json"
43FLOW_TOKEN_ENDPOINT = "/1.1/onboarding/task.json?flow_name=welcome&api_version=1&known_device_token=&sim_country_code=us"
44TASKS_ENDPOINT = "/1.1/onboarding/task.json"
45
46def send_req(method, endpoint, **kwargs) -> requests.Response:
47 debug(f"attempting `{endpoint}`")
48 try:
49 res = requests.request(method, BASE_URL + endpoint, **kwargs)
50 res.raise_for_status()
51 except requests.HTTPError:
52 error("HTTP request failed (non 2xx), unable to proceed.")
53 error('Please try again in a bit')
54 debug(f"request headers => {res.request.headers}")
55 debug(f"response headers => {res.headers}")
56 debug(f'response body => {res.content}')
57 sys.exit(1)
58 except Exception:
59 error(f"an unhandled error while sending a request to {endpoint} occurred")
60 raise
61
62 debug(f'got response body {res.content}')
63 return res
64
65B_SUBTASK_VERSIONS = {
66 "generic_urt": 3,
67 "standard": 1,
68 "open_home_timeline": 1,
69 "app_locale_update": 1,
70 "enter_date": 1,
71 "email_verification": 3,
72 "enter_password": 5,
73 "enter_text": 5,
74 "one_tap": 2,
75 "cta": 7,
76 "single_sign_on": 1,
77 "fetch_persisted_data": 1,
78 "enter_username": 3,
79 "web_modal": 2,
80 "fetch_temporary_password": 1,
81 "menu_dialog": 1,
82 "sign_up_review": 5,
83 "interest_picker": 4,
84 "user_recommendations_urt": 3,
85 "in_app_notification": 1,
86 "sign_up": 2,
87 "typeahead_search": 1,
88 "user_recommendations_list": 4,
89 "cta_inline": 1,
90 "contacts_live_sync_permission_prompt": 3,
91 "choice_selection": 5,
92 "js_instrumentation": 1,
93 "alert_dialog_suppress_client_events": 1,
94 "privacy_options": 1,
95 "topics_selector": 1,
96 "wait_spinner": 3,
97 "tweet_selection_urt": 1,
98 "end_flow": 1,
99 "settings_list": 7,
100 "open_external_link": 1,
101 "phone_verification": 5,
102 "security_key": 3,
103 "select_banner": 2,
104 "upload_media": 1,
105 "web": 2,
106 "alert_dialog": 1,
107 "open_account": 2,
108 "action_list": 2,
109 "enter_phone": 2,
110 "open_link": 1,
111 "show_code": 1,
112 "update_users": 1,
113 "check_logged_in_account": 1,
114 "enter_email": 2,
115 "select_avatar": 4,
116 "location_permission_prompt": 2,
117 "notifications_permission_prompt": 4
118}
119
120def get_flow_token_body():
121 return {
122 "flow_token": None,
123 "input_flow_data": {
124 "country_code": None,
125 "flow_context": {
126 "start_location": {
127 "location": "splash_screen"
128 }
129 },
130 "requested_variant": None,
131 "target_user_id": 0
132 },
133 "subtask_versions": B_SUBTASK_VERSIONS
134 }
135
136def get_tasks_body(flow_token: str) -> dict:
137 return {
138 "flow_token": flow_token,
139 "subtask_inputs": [{
140 "open_link": {
141 "link": "next_link"
142 },
143 "subtask_id": "NextTaskOpenLink"
144 }],
145 "subtask_versions": B_SUBTASK_VERSIONS
146 }
147
148# Functions
149def format_json(object) -> str:
150 global noprettyprint
151 return json.dumps(object, indent=None if noprettyprint else 4)
152
153def debug(msg, *arg, override=False, **kwarg) -> None:
154 global verbose
155 if verbose or override:
156 print("\x1b[37m[*]", msg, *arg, "\x1b[0m", file=sys.stderr, **kwarg)
157
158def info(msg, *arg, **kwarg) -> None:
159 print("\x1b[34m[i]", msg, *arg, "\x1b[0m", file=sys.stderr, **kwarg)
160
161
162def success(msg, *arg, **kwarg) -> None:
163 print("\x1b[32m[i]", msg, *arg, "\x1b[0m", file=sys.stderr, **kwarg)
164
165
166def warn(msg, *arg, **kwarg) -> None:
167 print("\x1b[33m[!]", msg, *arg, "\x1b[0m", file=sys.stderr, **kwarg)
168
169
170def error(msg, *arg, **kwarg) -> None:
171 print("\x1b[31m[x]", msg, *arg, "\x1b[0m", file=sys.stderr, **kwarg)
172
173
174def prompt_bool(msg, default: typing.Optional[bool] = True) -> bool:
175 """
176 Prompt the user for a y/n value until a valid input is entered.
177
178 :param msg: Message to display
179 :param default: What should the default value be if the user pressed enter. Pass in None to force user to pick one.
180 """
181 resolved = False
182 p_string = f"({'Y' if default else 'y'}/{'N' if not default else 'n'})" if default is not None else "(y/n)"
183 while not resolved:
184 print("\x1b[35m[?]", msg, f"\x1b[0m{p_string}", file=sys.stderr, end=" ")
185 try:
186 r = input()
187 except EOFError:
188 debug("^D", override=True)
189 debug("gracefully handling EOF")
190 error("invalid input, please try again.")
191 continue
192
193 if default is None and r.strip() == "":
194 error('a response is required. please enter either y or n.')
195 continue
196 if r.strip() == "":
197 return default
198 # if r.strip()[0].lower() not in ['y', 'n']:
199 # error('invalid input. please enter either y or n.')
200 # continue
201 match r.strip()[0].lower():
202 case "y":
203 return True
204 case "n":
205 return False
206 case _:
207 error('invalid input. please enter either y or n.')
208 continue
209
210
211# Arguments
212parser = ArgumentParser()
213parser.add_argument('-v', '--verbose', action='store_true', help="be more noisy")
214parser.add_argument('-P', '--no-pretty', action='store_true', help="disable pretty-printing of json data")
215parser.add_argument(
216 'outfile',
217 nargs="?",
218 default="-",
219 help="the json output file to put/append received account data to."
220)
221
222
223def main() -> int:
224 global verbose, noprettyprint
225 args = parser.parse_args()
226 verbose = args.verbose
227 noprettyprint = args.no_pretty
228
229 info("nitter-guest-account.py (2023-08-25)")
230 info("This is free software: you are free to change and redistribute it, under the terms of the Apache-2.0 license")
231 info("There is NO WARRANTY, to the extent permitted by law.")
232
233 info("Fetching bearer token...")
234 bt_raw = send_req('post', BEARER_TOKEN_ENDPOINT,
235 auth=requests.auth.HTTPBasicAuth(CONSUMER_KEY, CONSUMER_SECRET),
236 data={'grant_type': "client_credentials"}
237 ).json()
238 bearer_token = ' '.join(bt_raw.values())
239 if bearer_token.lower() != EXPECTED_BEARER_TOKEN.lower():
240 warn('Received bearer token does not match expected value. Continuing anyways, but beware of errors.')
241 info(f'bearer token => {bearer_token}')
242 else:
243 success('Received bearer token matches expected value.')
244
245 info("Fetching guest token...")
246 guest_token = send_req('post', GUEST_TOKEN_ENDPOINT, headers={'Authorization': bearer_token}).json()['guest_token']
247 success(f'guest token => {guest_token}')
248
249 debug('updating header with acquried credentials')
250 request_headers = BASE_REQUEST_HEADERS.copy()
251 request_headers.update({
252 "authorization": bearer_token,
253 "X-Guest-Token": guest_token
254 })
255
256 info('Fetching flow token...')
257 flow_token = send_req('post', FLOW_TOKEN_ENDPOINT, headers=request_headers, json=get_flow_token_body()).json()['flow_token']
258 success(f'flow token => {flow_token}')
259
260 info('Fetching final account object...')
261 backoff_time = 0
262 exception = None
263 for attempt in range(0, 6):
264 debug(f"Attempt #{attempt + 1}")
265 tasks = send_req('post', TASKS_ENDPOINT, headers=request_headers, json=get_tasks_body(flow_token)).json()
266 try:
267 try:
268 open_account_task = next(filter(lambda i: i.get('subtask_id') == "OpenAccount", tasks['subtasks']))
269 account = open_account_task['open_account']
270 except StopIteration as e:
271 backoff_time += random.randint(5000,10000) / 1000
272 exception = e
273 warn(f"attempt #{attempt + 1} failed to acquire token, retrying in {backoff_time}s.")
274 time.sleep(backoff_time)
275 continue
276
277 info(f"Attempt #{attempt + 1} succeeded")
278 if args.outfile == "-":
279 debug("outfile is `-`, printing to stdout")
280 print(format_json(account))
281 return 0
282
283 # Sanity checks
284 try:
285 debug(f"attempting to read file: {args.outfile}")
286 with open(args.outfile) as f:
287 old_data = json.load(f)
288 except FileNotFoundError:
289 # that's okay, we might be able to create it later.
290 old_data = []
291 except PermissionError:
292 # that's not okay. we will need to access the file later anyways.
293 error("unable to read file due to a permission error.")
294 error("please make sure this script has read and write access to the file.")
295 print(format_json(account))
296 return 1
297 except json.JSONDecodeError:
298 error("could not parse the provided JSON file.")
299 if not prompt_bool("Do you want to overwrite the file?", default=None):
300 warn("Not overwriting file, printing to stdout instead.")
301 print(format_json(account))
302 return 1
303 debug('assuming old data is an empty array because we are overwriting')
304 old_data = []
305 if type(old_data) != list:
306 error("top-level object of the existing JSON file is not a list.")
307 error("due to the implementation, the file must be overwritten.")
308 if not (prompt_bool("Do you want to overwrite?", default=None)):
309 warn("Not overwriting existing data, printing to stdout instead.")
310 print(format_json(account))
311 return 1
312 debug("assuming old data is an empty array because we are overwriting")
313 old_data = []
314
315 old_data.append(account)
316
317 try:
318 debug("attempting to write file")
319 with open(args.outfile, 'w+') as f:
320 f.write(format_json(old_data))
321 success(f"successfully written to file {args.outfile}")
322 return 0
323 except PermissionError:
324 error("unable to write to file due to permission error.")
325 error("please make sure this script has write access to the file.")
326 print(format_json(account))
327 return 1
328 except Exception as e:
329 error("Unable to write to file due to an uncaught error:", e)
330 tb = ''.join(traceback.TracebackException.from_exception(e).format())
331 debug("exception stacktrace\n" + tb)
332 print(format_json(account))
333
334 except Exception:
335 debug("resulting tasks =>", format_json(tasks), override=True)
336 error("an unhandled error occurred. the tasks object is printed to avoid losing otherwise successful data.")
337 error("please file a bug report and attach the traceback below.")
338 raise
339
340 if exception != None:
341 debug("resulting tasks =>", format_json(tasks))
342 error("Unable to acquire guest account credentials with 5 attempts as it isn't present in any of the API responses.")
343 error("This might be because of a wide variety of reasons, but it most likely is due to your IP being rate-limited.")
344 error("Try again with a new IP address or in 24 hours after this attempt.")
345 return 1
346
347 return 0
348
349
350if __name__ == "__main__":
351 rc = main()
352 sys.exit(rc)