Allow feature requests from mutuals

This can be refined later
main
bashrc 2026-07-18 18:41:23 +01:00
parent 86fc1b921b
commit 93753cafc5
21 changed files with 261 additions and 145 deletions

View File

@ -29,6 +29,7 @@ from src.utils import has_actor
from src.utils import has_object_string_type from src.utils import has_object_string_type
from src.utils import get_actor_from_post from src.utils import get_actor_from_post
from src.utils import is_yggdrasil_address from src.utils import is_yggdrasil_address
from src.utils import url_text_to_number
from src.timeFunctions import get_current_time_int from src.timeFunctions import get_current_time_int
from src.data import is_a_file from src.data import is_a_file
@ -490,11 +491,11 @@ def receive_quote_request(message_json: {}, federation_list: [],
def create_feature_reject(federation_list: [], def create_feature_reject(federation_list: [],
nickname: str, domain: str, port: int, nickname: str, domain: str, port: int,
to_url: str, cc_url: str, http_prefix: str, to_url: str, cc_url: str, http_prefix: str,
object_json: {}) -> {}: request_object_json: {}) -> {}:
""" Create json for ActivityPub FeatureRequest Reject """ Create json for ActivityPub FeatureRequest Reject
https://codeberg.org/fediverse/fep/src/branch/main/fep/7aa9/fep-7aa9.md https://codeberg.org/fediverse/fep/src/branch/main/fep/7aa9/fep-7aa9.md
""" """
if not object_json.get('id'): if not request_object_json.get('id'):
return None return None
if not url_permitted(to_url, federation_list): if not url_permitted(to_url, federation_list):
@ -502,6 +503,8 @@ def create_feature_reject(federation_list: [],
domain = get_full_domain(domain, port) domain = get_full_domain(domain, port)
request_id = request_object_json['id']
new_reject = { new_reject = {
"@context": [ "@context": [
'https://www.w3.org/ns/activitystreams', 'https://www.w3.org/ns/activitystreams',
@ -511,8 +514,67 @@ def create_feature_reject(federation_list: [],
'actor': local_actor_url(http_prefix, nickname, domain), 'actor': local_actor_url(http_prefix, nickname, domain),
'to': [to_url], 'to': [to_url],
'cc': [], 'cc': [],
'object': object_json['id'] 'object': request_id
} }
if cc_url: if cc_url:
new_reject['cc'] = [cc_url] new_reject['cc'] = [cc_url]
return new_reject return new_reject
def create_feature_accept(federation_list: [],
nickname: str, domain: str, port: int,
to_url: str, cc_url: str, http_prefix: str,
request_object_json: {}) -> ({}, {}):
""" Create json for ActivityPub FeatureRequest Accept and
featured item to be stored
https://codeberg.org/fediverse/fep/src/branch/main/fep/7aa9/fep-7aa9.md
"""
if not request_object_json.get('id'):
return None, None
if not isinstance(request_object_json['id'], str):
return None, None
if not request_object_json.get('object'):
return None, None
if not isinstance(request_object_json['object'], str):
return None, None
if not url_permitted(to_url, federation_list):
return None, None
domain = get_full_domain(domain, port)
request_id = request_object_json['id']
stamp_id = url_text_to_number(request_object_json['object'])
stamp_url = \
http_prefix + '://' + domain + '/stamps/' + stamp_id
new_accept = {
"@context": [
'https://www.w3.org/ns/activitystreams',
'https://w3id.org/security/v1'
],
'type': 'Accept',
'actor': local_actor_url(http_prefix, nickname, domain),
'to': [to_url],
'cc': [],
'object': request_id,
'result': stamp_url
}
if cc_url:
new_accept['cc'] = [cc_url]
_, published = get_status_number()
new_featured_item = {
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/fep/7aa9"
],
"id": request_id,
"type": "FeaturedItem",
"object": request_object_json['object'],
"featureAuthorization": stamp_url,
"published": published
}
return new_accept, new_featured_item

View File

@ -18,12 +18,15 @@ from src.utils import file_last_modified
from src.utils import file_created_date from src.utils import file_created_date
from src.utils import get_nickname_from_actor from src.utils import get_nickname_from_actor
from src.utils import get_domain_from_actor from src.utils import get_domain_from_actor
from src.utils import url_text_to_number
from src.utils import save_json
from src.data import is_a_file from src.data import is_a_file
from src.data import load_list from src.data import load_list
from src.data import is_a_dir
from src.data import makedir
FEATURED_COLLECTIONS_ENDING = '/featured_collections' FEATURED_COLLECTIONS_ENDING = '/featured_collections'
AUTHORIZATIONS_ENDING = '/featured_authorizations' AUTHORIZATIONS_ENDING = '/featured_authorizations'
URL_TEXT_MAGIC_NUMBER = 7439
def _get_no_of_featured_collections(base_dir: str, def _get_no_of_featured_collections(base_dir: str,
@ -47,22 +50,6 @@ def _get_no_of_featured_collections(base_dir: str,
return ctr return ctr
def _url_text_to_number(url: str) -> str:
"""converts url text or collection name to a number string used as an id
"""
result = ''
for char in url:
num = ord(char)
if num > 99:
continue
if num > 9:
result = str(num) + result
else:
result = '0' + str(num) + result
num = (URL_TEXT_MAGIC_NUMBER + int(result)) % 99999999999999999999
return str(num)
def _update_collections(collection_name: str, collection_items: [], def _update_collections(collection_name: str, collection_items: [],
collection: [], collection: [],
http_prefix: str, domain: str, http_prefix: str, domain: str,
@ -74,7 +61,7 @@ def _update_collections(collection_name: str, collection_items: [],
collection_items.clear() collection_items.clear()
return return
# create an id number for the collection # create an id number for the collection
collection_id = _url_text_to_number(collection_name) collection_id = url_text_to_number(collection_name)
collection_url = \ collection_url = \
http_prefix + '://' + domain + '/collections/' + collection_id http_prefix + '://' + domain + '/collections/' + collection_id
collection_dict = { collection_dict = {
@ -275,20 +262,8 @@ def get_featured_collections_feed(base_dir: str,
if item_nickname and item_domain: if item_nickname and item_domain:
item_url = text item_url = text
if item_url: if item_url:
# authorization link which should be from the same domain
# as item_url
if authorizations.get(item_url): if authorizations.get(item_url):
feature_authorization = authorizations[item_url] collection_items.append(authorizations[item_url])
# id for collection item
collection_item_id = _url_text_to_number(item_url)
collection_item_dict = {
'featureAuthorization': feature_authorization,
'featuredObject': item_url,
'id': actor + '/collection_items/' + collection_item_id,
'published': published,
'type': 'FeaturedItem'
}
collection_items.append(collection_item_dict)
# store the current collection # store the current collection
_update_collections(collection_name, collection_items, _update_collections(collection_name, collection_items,
collection, collection,
@ -302,3 +277,17 @@ def get_featured_collections_feed(base_dir: str,
collection['items'].append(collection_dict) collection['items'].append(collection_dict)
collection['totalItems'] = collection['totalItems'] + 1 collection['totalItems'] = collection['totalItems'] + 1
return collection return collection
def store_feature_authorization(base_dir: str, nickname: str, domain: str,
featured_item: {}) -> bool:
"""Stores a feature authorization
"""
accounts_dir: str = acct_dir(base_dir, nickname, domain)
if not is_a_dir(accounts_dir + '/stamps'):
makedir(accounts_dir + '/stamps')
stamp_id = featured_item['featureAuthorization'].split('/')[1]
feature_authorization_filename = accounts_dir + '/' + stamp_id
if not is_a_file(feature_authorization_filename):
return False
return save_json(featured_item, feature_authorization_filename)

View File

@ -241,7 +241,8 @@ def announce_button(self, calling_domain: str, path: str,
if self.post_to_nickname in min_images_for_accounts: if self.post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, mutuals_list: list[str] = \
get_mutuals_of_person(base_dir,
self.post_to_nickname, domain) self.post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, recent_posts_cache,

View File

@ -193,7 +193,7 @@ def bookmark_button(self, calling_domain: str, path: str,
if self.post_to_nickname in min_images_for_accounts: if self.post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, self.post_to_nickname, domain) get_mutuals_of_person(base_dir, self.post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
False, False,
@ -422,7 +422,7 @@ def bookmark_button_undo(self, calling_domain: str, path: str,
if self.post_to_nickname in min_images_for_accounts: if self.post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, self.post_to_nickname, domain) get_mutuals_of_person(base_dir, self.post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
False, False,

View File

@ -240,7 +240,7 @@ def like_button(self, calling_domain: str, path: str,
if self.post_to_nickname in min_images_for_accounts: if self.post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, self.post_to_nickname, domain) get_mutuals_of_person(base_dir, self.post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
False, False,
@ -513,7 +513,7 @@ def like_button_undo(self, calling_domain: str, path: str,
if self.post_to_nickname in min_images_for_accounts: if self.post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, self.post_to_nickname, domain) get_mutuals_of_person(base_dir, self.post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
False, False,

View File

@ -149,7 +149,8 @@ def mute_button(self, calling_domain: str, path: str,
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
allow_downloads, allow_downloads,
recent_posts_cache, recent_posts_cache,
@ -341,7 +342,8 @@ def mute_button_undo(self, calling_domain: str, path: str,
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
allow_downloads, allow_downloads,
recent_posts_cache, recent_posts_cache,

View File

@ -270,7 +270,7 @@ def reaction_button(self, calling_domain: str, path: str,
if self.post_to_nickname in min_images_for_accounts: if self.post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, self.post_to_nickname, domain) get_mutuals_of_person(base_dir, self.post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
False, False,
@ -566,7 +566,7 @@ def reaction_button_undo(self, calling_domain: str, path: str,
if self.post_to_nickname in min_images_for_accounts: if self.post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, self.post_to_nickname, domain) get_mutuals_of_person(base_dir, self.post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
False, False,

View File

@ -141,7 +141,8 @@ def _show_post_from_file(self, post_filename: str, liked_by: str,
bold_reading = True bold_reading = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
msg = \ msg = \
html_individual_post(recent_posts_cache, html_individual_post(recent_posts_cache,
@ -1053,7 +1054,8 @@ def show_replies_to_post(self, authorized: bool,
if bold_reading_nicknames.get(nickname): if bold_reading_nicknames.get(nickname):
bold_reading = True bold_reading = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
msg = \ msg = \
html_post_replies(recent_posts_cache, html_post_replies(recent_posts_cache,
max_recent_posts, max_recent_posts,
@ -1176,7 +1178,8 @@ def show_replies_to_post(self, authorized: bool,
if bold_reading_nicknames.get(nickname): if bold_reading_nicknames.get(nickname):
bold_reading = True bold_reading = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
msg = \ msg = \
html_post_replies(recent_posts_cache, html_post_replies(recent_posts_cache,
max_recent_posts, max_recent_posts,

View File

@ -2020,7 +2020,8 @@ def _inbox_after_initial(server, inbox_start_time,
return False return False
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, handle_name, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, handle_name, domain)
if receive_actor_status(base_dir, person_cache, message_json, if receive_actor_status(base_dir, person_cache, message_json,
debug): debug):

View File

@ -98,7 +98,9 @@ from src.data import erase_file
from src.data import is_a_file from src.data import is_a_file
from src.data import is_a_dir from src.data import is_a_dir
from src.acceptreject import create_feature_reject from src.acceptreject import create_feature_reject
from src.acceptreject import create_feature_accept
from src.posts import send_signed_json from src.posts import send_signed_json
from src.collections import store_feature_authorization
def inbox_update_index(boxname: str, base_dir: str, handle: str, def inbox_update_index(boxname: str, base_dir: str, handle: str,
@ -520,7 +522,8 @@ def receive_edit_to_post(recent_posts_cache: {}, message_json: {},
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,
@ -1099,7 +1102,8 @@ def receive_like(recent_posts_cache: {},
if handle_name in min_images_for_accounts: if handle_name in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, handle_name, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, handle_name, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,
@ -1179,10 +1183,10 @@ def receive_feature_request(session,
print('DEBUG: FeatureRequest has no actor domain ' + actor_url) print('DEBUG: FeatureRequest has no actor domain ' + actor_url)
return False return False
actor_domain_full = get_full_domain(actor_domain, actor_port) actor_domain_full = get_full_domain(actor_domain, actor_port)
handle = actor_nickname + '@' + actor_domain_full handle2 = actor_nickname + '@' + actor_domain_full
handle_dir = acct_handle_dir(base_dir, handle) handle2_dir = acct_handle_dir(base_dir, handle2)
if not is_a_dir(handle_dir): if not is_a_dir(handle2_dir):
print('DEBUG: unknown recipient of FeatureRequest - ' + handle) print('DEBUG: unknown recipient of FeatureRequest - ' + handle2)
return True return True
sender_url = message_json['id'] sender_url = message_json['id']
@ -1201,23 +1205,40 @@ def receive_feature_request(session,
print('DEBUG: FeatureRequest has no sender domain ' + sender_url) print('DEBUG: FeatureRequest has no sender domain ' + sender_url)
return False return False
sender_domain_full = get_full_domain(sender_domain, sender_port) sender_domain_full = get_full_domain(sender_domain, sender_port)
sender_handle = sender_nickname + '@' + sender_domain_full
if is_blocked(base_dir, actor_nickname, actor_domain_full, if is_blocked(base_dir, actor_nickname, actor_domain_full,
sender_nickname, sender_domain_full, sender_nickname, sender_domain_full,
blocked_cache, block_federated): blocked_cache, block_federated):
sender_handle = sender_nickname + '@' + sender_domain_full
print('BLOCK: FeatureRequest sender actor is blocked ' + print('BLOCK: FeatureRequest sender actor is blocked ' +
sender_handle + ' by ' + actor_url) sender_handle + ' by ' + actor_url)
return False return False
reject_json = \ handle_nickname = handle.split('@')[0]
handle_domain = handle.split('@')[0]
mutuals: list[str] = \
get_mutuals_of_person(base_dir, handle_nickname, handle_domain)
new_featured_item = None
if sender_handle not in mutuals:
acceptreject_json = \
create_feature_reject(federation_list, create_feature_reject(federation_list,
actor_nickname, actor_domain, actor_port, actor_nickname, actor_domain, actor_port,
sender_url, '', http_prefix, message_json) sender_url, '', http_prefix, message_json)
if debug: if debug:
pprint(reject_json) pprint(acceptreject_json)
print('REJECT: sending FeatureRequest Reject from ' + print('REJECT: sending FeatureRequest Reject from ' +
actor_nickname + '@' + actor_domain_full + actor_nickname + '@' + actor_domain_full +
' to ' + sender_nickname + '@' + sender_domain_full) ' to ' + sender_nickname + '@' + sender_domain_full)
else:
acceptreject_json, new_featured_item = \
create_feature_accept(federation_list,
actor_nickname, actor_domain, actor_port,
sender_url, '', http_prefix, message_json)
if debug:
pprint(acceptreject_json)
print('ACCEPT: sending FeatureRequest Accept from ' +
actor_nickname + '@' + actor_domain_full +
' to ' + sender_nickname + '@' + sender_domain_full)
curr_session = session curr_session = session
curr_domain = domain curr_domain = domain
@ -1262,7 +1283,7 @@ def receive_feature_request(session,
group_account: bool = False group_account: bool = False
if has_group_type(base_dir, actor_url, person_cache): if has_group_type(base_dir, actor_url, person_cache):
group_account = True group_account = True
send_signed_json(reject_json, curr_session, base_dir, if send_signed_json(acceptreject_json, curr_session, base_dir,
sender_nickname, sender_domain, sender_port, sender_nickname, sender_domain, sender_port,
actor_nickname, domain, curr_port, actor_nickname, domain, curr_port,
curr_http_prefix, client_to_server, curr_http_prefix, client_to_server,
@ -1273,7 +1294,11 @@ def receive_feature_request(session,
326735214, curr_domain, 326735214, curr_domain,
onion_domain, i2p_domain, yggdrasil_domain, onion_domain, i2p_domain, yggdrasil_domain,
extra_headers, sites_unavailable, extra_headers, sites_unavailable,
system_language, mitm_servers) system_language, mitm_servers) == 0:
if new_featured_item:
store_feature_authorization(base_dir,
handle_nickname, handle_domain,
new_featured_item)
return True return True
@ -1498,7 +1523,8 @@ def receive_reaction(recent_posts_cache: {},
if handle_name in min_images_for_accounts: if handle_name in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, handle_name, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, handle_name, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,
@ -1700,7 +1726,8 @@ def receive_zot_reaction(recent_posts_cache: {},
if handle_name in min_images_for_accounts: if handle_name in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, handle_name, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, handle_name, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,
@ -1845,7 +1872,8 @@ def receive_bookmark(recent_posts_cache: {},
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,
@ -2171,7 +2199,8 @@ def receive_announce(recent_posts_cache: {},
show_vote_posts = False show_vote_posts = False
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
announce_html = \ announce_html = \
individual_post_as_html(signing_priv_key_pem, True, individual_post_as_html(signing_priv_key_pem, True,
@ -2401,7 +2430,8 @@ def receive_question_vote(server, base_dir: str, nickname: str, domain: str,
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,

View File

@ -275,7 +275,8 @@ def receive_undo_like(recent_posts_cache: {},
if handle_name in min_images_for_accounts: if handle_name in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, handle_name, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, handle_name, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,
@ -436,7 +437,8 @@ def receive_undo_reaction(recent_posts_cache: {},
if handle_name in min_images_for_accounts: if handle_name in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, handle_name, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, handle_name, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,
@ -583,7 +585,8 @@ def receive_undo_bookmark(recent_posts_cache: {},
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
individual_post_as_html(signing_priv_key_pem, False, individual_post_as_html(signing_priv_key_pem, False,
recent_posts_cache, max_recent_posts, recent_posts_cache, max_recent_posts,
translate, page_number, base_dir, translate, page_number, base_dir,

View File

@ -704,7 +704,7 @@ def post_message_to_outbox(session, translate: {},
if post_to_nickname in min_images_for_accounts: if post_to_nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, post_to_nickname, domain) get_mutuals_of_person(base_dir, post_to_nickname, domain)
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
False, recent_posts_cache, False, recent_posts_cache,

View File

@ -178,7 +178,8 @@ def search_box_posts(base_dir: str, nickname: str, domain: str,
following_list = get_followers_list(base_dir, nickname, domain, following_list = get_followers_list(base_dir, nickname, domain,
'following.txt') 'following.txt')
# create a list containing all of the mutuals # create a list containing all of the mutuals
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
res: list[str] = [] res: list[str] = []
for root, _, fnames in os.walk(path): for root, _, fnames in os.walk(path):

View File

@ -58,6 +58,8 @@ INVALID_ACTOR_URL_CHARACTERS = (
';', '=' ';', '='
) )
URL_TEXT_MAGIC_NUMBER = 7439
def is_account_dir(dir_name: str) -> bool: def is_account_dir(dir_name: str) -> bool:
"""Is the given directory an account within /accounts ? """Is the given directory an account within /accounts ?
@ -735,6 +737,31 @@ def set_memorials(base_dir: str, domain: str, memorial_str) -> None:
'EX: unable to write ' + memorial_file) 'EX: unable to write ' + memorial_file)
def save_json(json_object: {}, filename: str) -> bool:
"""Saves json to a file
"""
if not isinstance(json_object, dict):
if not isinstance(json_object, list):
print('EX: save_json object is not json ' + str(json_object))
return False
tries: int = 1
savestr = json.dumps(json_object)
while tries <= 5:
success, errno = \
save_with_err(savestr, filename,
'EX: save_json ' + str(tries) + ' ' +
str(filename) + ' [ex]')
if success:
return True
if errno == 36:
# filename too long
break
time.sleep(1)
tries += 1
return False
def _create_config(base_dir: str) -> None: def _create_config(base_dir: str) -> None:
"""Creates a configuration file """Creates a configuration file
""" """
@ -745,6 +772,32 @@ def _create_config(base_dir: str) -> None:
save_json(config_json, config_filename) save_json(config_json, config_filename)
def load_json(filename: str) -> {}:
"""Makes a few attempts to load a json formatted file
"""
if '/Actor@' in filename:
filename = filename.replace('/Actor@', '/inbox@')
json_object = None
data: str = load_string(filename,
'EX: load_json exception ' +
str(filename) + ' [ex]')
if data is None:
return json_object
# check that something was loaded
if not data:
print('EX: load_json no data ' + str(filename))
return json_object
# convert to json
try:
json_object = json.loads(data)
except BaseException as exc:
print('EX: load_json exception ' + str(filename) + ' ' + str(exc))
return json_object
def set_config_param(base_dir: str, variable_name: str, def set_config_param(base_dir: str, variable_name: str,
variable_value) -> None: variable_value) -> None:
"""Sets a configuration value """Sets a configuration value
@ -908,57 +961,6 @@ def get_link_prefixes() -> []:
'hyper://', 'gemini://', 'gopher://', 'briar:') 'hyper://', 'gemini://', 'gopher://', 'briar:')
def save_json(json_object: {}, filename: str) -> bool:
"""Saves json to a file
"""
if not isinstance(json_object, dict):
if not isinstance(json_object, list):
print('EX: save_json object is not json ' + str(json_object))
return False
tries: int = 1
savestr = json.dumps(json_object)
while tries <= 5:
success, errno = \
save_with_err(savestr, filename,
'EX: save_json ' + str(tries) + ' ' +
str(filename) + ' [ex]')
if success:
return True
if errno == 36:
# filename too long
break
time.sleep(1)
tries += 1
return False
def load_json(filename: str) -> {}:
"""Makes a few attempts to load a json formatted file
"""
if '/Actor@' in filename:
filename = filename.replace('/Actor@', '/inbox@')
json_object = None
data: str = load_string(filename,
'EX: load_json exception ' +
str(filename) + ' [ex]')
if data is None:
return json_object
# check that something was loaded
if not data:
print('EX: load_json no data ' + str(filename))
return json_object
# convert to json
try:
json_object = json.loads(data)
except BaseException as exc:
print('EX: load_json exception ' + str(filename) + ' ' + str(exc))
return json_object
def load_json_onionify(filename: str, domain: str, onion_domain: str, def load_json_onionify(filename: str, domain: str, onion_domain: str,
delay_sec: int = 2) -> {}: delay_sec: int = 2) -> {}:
"""Makes a few attempts to load a json formatted file """Makes a few attempts to load a json formatted file
@ -4374,3 +4376,19 @@ def get_preferred_username(actor: {}, system_language: str) -> str:
if isinstance(username, str): if isinstance(username, str):
return username return username
return actor['preferredUsername'] return actor['preferredUsername']
def url_text_to_number(url: str) -> str:
"""converts url text or collection name to a number string used as an id
"""
result = ''
for char in url:
num = ord(char)
if num > 99:
continue
if num > 9:
result = str(num) + result
else:
result = '0' + str(num) + result
num = (URL_TEXT_MAGIC_NUMBER + int(result)) % 9999999999999999999999999999
return str(num)

View File

@ -96,7 +96,8 @@ def html_confirm_delete(server,
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
delete_post_str += \ delete_post_str += \
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
True, recent_posts_cache, max_recent_posts, True, recent_posts_cache, max_recent_posts,

View File

@ -140,7 +140,8 @@ def html_conversation_view(authorized: bool, post_id: str,
show_individual_post_icons = False show_individual_post_icons = False
allow_deletion: bool = False allow_deletion: bool = False
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
post_str = \ post_str = \
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
True, recent_posts_cache, True, recent_posts_cache,

View File

@ -443,7 +443,7 @@ def html_new_post(edit_post_params: {},
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain) get_mutuals_of_person(base_dir, nickname, domain)
replied_to_post = \ replied_to_post = \
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,

View File

@ -84,7 +84,7 @@ def _html_front_screen_posts(recent_posts_cache: {}, max_recent_posts: int,
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = \ mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain) get_mutuals_of_person(base_dir, nickname, domain)
post_str = \ post_str = \
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,

View File

@ -103,7 +103,8 @@ def html_likers_of_post(base_dir: str, nickname: str,
if nickname in min_images_for_accounts: if nickname in min_images_for_accounts:
minimize_all_images = True minimize_all_images = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
html_str += \ html_str += \
individual_post_as_html(signing_priv_key_pem, individual_post_as_html(signing_priv_key_pem,
True, recent_posts_cache, True, recent_posts_cache,
@ -160,7 +161,8 @@ def html_likers_of_post(base_dir: str, nickname: str,
'<center><h2>' + translate['Repeated by'] + '</h2></center>\n' '<center><h2>' + translate['Repeated by'] + '</h2></center>\n'
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
is_text_mode = text_mode_browser(ua_str) is_text_mode = text_mode_browser(ua_str)
likers_list: str = '' likers_list: str = ''

View File

@ -3567,7 +3567,8 @@ def html_individual_post(recent_posts_cache: {}, max_recent_posts: int,
actor = '/users/' + nickname actor = '/users/' + nickname
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
mutual_prefix: str = '' mutual_prefix: str = ''
if by_str_handle in mutuals_list: if by_str_handle in mutuals_list:
if not text_mode_browser(ua_str): if not text_mode_browser(ua_str):

View File

@ -1020,7 +1020,8 @@ def html_timeline(default_timeline: str,
no_seen_posts = True no_seen_posts = True
# get the list of mutuals for the current account # get the list of mutuals for the current account
mutuals_list = get_mutuals_of_person(base_dir, nickname, domain) mutuals_list: list[str] = \
get_mutuals_of_person(base_dir, nickname, domain)
# show each post in the timeline # show each post in the timeline
tl_items_str: str = '' tl_items_str: str = ''