itsulu/FediFetcher
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #187 from nanos/identify-access-token

Browse source 
Identify a (masked) Access Token when an error occurs
This commit is contained in:
Michael 2024-12-04 07:31:53 +00:00 committed by GitHub
commit 3fba054257
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 42 additions and 58 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

100
find_posts.py
View file

@ -367,19 +367,12 @@ def get_timeline(server, access_token, max):
if response.status_code == 200: if response.status_code == 200:
toots = response.json() toots = response.json()
elif response.status_code == 401:
raise Exception(
f"Error getting URL {url}. Status code: {response.status_code}. "
"Ensure your access token is correct"
)
elif response.status_code == 403:
raise Exception(
f"Error getting URL {url}. Status code: {response.status_code}. "
"Make sure you have the read:statuses scope enabled for your access token."
)
else: else:
raise Exception( report_mastodon_error(
f"Error getting URL {url}. Status code: {response.status_code}" f"Error getting URL {url}",
response.status_code,
access_token,
"read:statuses"
) )
# Paginate as needed # Paginate as needed
@ -401,19 +394,12 @@ def get_toots(url, access_token):
if response.status_code == 200: if response.status_code == 200:
return response return response
elif response.status_code == 401:
raise Exception(
f"Error getting URL {url}. Status code: {response.status_code}. "
"It looks like your access token is incorrect."
)
elif response.status_code == 403:
raise Exception(
f"Error getting URL {url}. Status code: {response.status_code}. "
"Make sure you have the read:statuses scope enabled for your access token."
)
else: else:
raise Exception( report_mastodon_error(
f"Error getting URL {url}. Status code: {response.status_code}" f"Error getting URL {url}",
response.status_code,
access_token,
"read:statuses"
) )
def get_active_user_ids(server, access_token, reply_interval_hours): def get_active_user_ids(server, access_token, reply_interval_hours):
@ -432,19 +418,12 @@ def get_active_user_ids(server, access_token, reply_interval_hours):
if last_active > since: if last_active > since:
logger.info(f"Found active user: {user['username']}") logger.info(f"Found active user: {user['username']}")
yield user["id"] yield user["id"]
elif resp.status_code == 401:
raise Exception(
f"Error getting user IDs on server {server}. Status code: {resp.status_code}. "
"Ensure your access token is correct"
)
elif resp.status_code == 403:
raise Exception(
f"Error getting user IDs on server {server}. Status code: {resp.status_code}. "
"Make sure you have the admin:read:accounts scope enabled for your access token."
)
else: else:
raise Exception( report_mastodon_error(
f"Error getting user IDs on server {server}. Status code: {resp.status_code}" f"Error getting user IDs on server {server}",
resp.status_code,
access_token,
"admin:read:accounts"
) )
@ -491,16 +470,14 @@ def get_reply_toots(user_id, server, access_token, seen_urls, reply_since):
for toot in toots: for toot in toots:
logger.debug(f"Found reply toot: {toot['url']}") logger.debug(f"Found reply toot: {toot['url']}")
return toots return toots
elif resp.status_code == 403: else:
raise Exception( report_mastodon_error(
f"Error getting replies for user {user_id} on server {server}. Status code: {resp.status_code}. " f"Error getting replies for user {user_id} on server {server}",
"Make sure you have the read:statuses scope enabled for your access token." resp.status_code,
access_token,
"read:statuses"
) )
raise Exception(
f"Error getting replies for user {user_id} on server {server}. Status code: {resp.status_code}"
)
def toot_context_can_be_fetched(toot): def toot_context_can_be_fetched(toot):
fetchable = toot["visibility"] in ["public", "unlisted"] fetchable = toot["visibility"] in ["public", "unlisted"]
@ -1033,20 +1010,11 @@ def get_paginated_mastodon(url, max, headers = {}, timeout = 0, max_tries = 5):
response = get(furl, headers, timeout, max_tries) response = get(furl, headers, timeout, max_tries)
if response.status_code != 200: if response.status_code != 200:
if response.status_code == 401: report_mastodon_error(
raise Exception( f"Error getting URL {url}",
f"Error getting URL {url}. Status code: {response.status_code}. " response.status_code,
"Ensure your access token is correct" headers.get('Authorization', '').replace("Bearer ", ""),
) )
elif response.status_code == 403:
raise Exception(
f"Error getting URL {url}. Status code: {response.status_code}. "
"Make sure you have the correct scopes enabled for your access token."
)
else:
raise Exception(
f"Error getting URL {url}. Status code: {response.status_code}"
)
result = response.json() result = response.json()
@ -1470,6 +1438,22 @@ def fetch_timeline_context(timeline_posts, token, parsed_urls, seen_hosts, seen_
add_user_posts(arguments.server, token, filter_known_users(mentioned_users, all_known_users), recently_checked_users, all_known_users, seen_urls, seen_hosts) add_user_posts(arguments.server, token, filter_known_users(mentioned_users, all_known_users), recently_checked_users, all_known_users, seen_urls, seen_hosts)
def report_mastodon_error(error_message, error_code, access_token, required_scope = ''):
subline = ""
match error_code:
case 401:
subline = "\nIt looks like your access token is incorrect. Consider generating a new access token, and/or ensure you have copy and pasted the whole token correctly."
case 403:
if(required_scope != ""):
subline = f"\nAdd the {required_scope} scope to your access token, and regenerate the token."
else:
subline = "\nMake sure you have enabled the required scope(s) for your token."
raise Exception(
f"{error_message} with token {access_token[:+5]}{'*' * (len(access_token) - 10)}{access_token[-5:]}. Status code: {error_code} "
f"{subline}"
)
if __name__ == "__main__": if __name__ == "__main__":
start = datetime.now() start = datetime.now()