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