__filename__ = "mastoapiv1.py" __author__ = "Bob Mottram" __license__ = "AGPL3+" __version__ = "1.5.0" __maintainer__ = "Bob Mottram" __email__ = "bob@libreserver.org" __status__ = "Production" __module_group__ = "API" import os from utils import string_contains from utils import get_url_from_post from utils import load_json from utils import get_config_param from utils import acct_dir from utils import remove_html from utils import get_attachment_property_value from utils import no_of_accounts from utils import get_status_count from utils import lines_in_file def _meta_data_instance_v1(show_accounts: bool, instance_title: str, instance_description_short: str, instance_description: str, http_prefix: str, base_dir: str, admin_nickname: str, domain: str, domain_full: str, registration: bool, system_language: str, version: str) -> {}: """ /api/v1/instance endpoint """ admin_actor_filename = \ base_dir + '/accounts/' + admin_nickname + '@' + domain + '.json' if not os.path.isfile(admin_actor_filename): return {} admin_actor = load_json(admin_actor_filename, 0) if not admin_actor: print('WARN: json load exception _meta_data_instance_v1') return {} rules_list = [] rules_filename = \ base_dir + '/accounts/tos.md' if os.path.isfile(rules_filename): with open(rules_filename, 'r', encoding='utf-8') as fp_rules: rules_lines = fp_rules.readlines() rule_ctr = 1 for line in rules_lines: line = line.strip() if not line: continue if line.startswith('#'): continue rules_list.append({ 'id': str(rule_ctr), 'text': line }) rule_ctr += 1 is_bot = False is_group = False if admin_actor['type'] == 'Group': is_group = True elif admin_actor['type'] != 'Person': is_bot = True url = \ http_prefix + '://' + domain_full + '/@' + \ admin_actor['preferredUsername'] if show_accounts: active_accounts = no_of_accounts(base_dir) local_posts = get_status_count(base_dir) else: active_accounts = 1 local_posts = 1 created_at = '' if admin_actor.get('published'): created_at = admin_actor['published'] url_str = get_url_from_post(admin_actor['icon']['url']) icon_url = remove_html(url_str) url_str = get_url_from_post(admin_actor['image']['url']) image_url = remove_html(url_str) instance = { 'approval_required': False, 'invites_enabled': False, 'registrations': registration, 'contact_account': { 'acct': admin_actor['preferredUsername'], 'created_at': created_at, 'avatar': icon_url, 'avatar_static': icon_url, 'header': image_url, 'header_static': image_url, 'bot': is_bot, 'discoverable': True, 'group': is_group, 'display_name': admin_actor['name'], 'locked': admin_actor['manuallyApprovesFollowers'], 'note': '

Admin of ' + domain + '

', 'url': url, 'username': admin_actor['preferredUsername'] }, 'description': instance_description, 'languages': [system_language], 'short_description': instance_description_short, 'stats': { 'domain_count': 2, 'status_count': local_posts, 'user_count': active_accounts }, 'thumbnail': http_prefix + '://' + domain_full + '/login.png', 'title': instance_title, 'uri': domain_full, 'urls': {}, 'version': version, 'rules': rules_list, 'configuration': { 'statuses': { 'max_media_attachments': 1 }, 'media_attachments': { 'supported_mime_types': [ 'image/jpeg', 'image/jxl', 'image/png', 'image/gif', 'image/webp', 'image/avif', 'image/heic', 'image/svg+xml', 'video/webm', 'video/mp4', 'video/ogv', 'audio/ogg', 'audio/wav', 'audio/x-wav', 'audio/x-pn-wave', 'audio/vnd.wave', 'audio/opus', 'audio/speex', 'audio/x-speex', 'audio/flac', 'audio/mpeg' ], 'image_size_limit': 10485760, 'image_matrix_limit': 16777216, 'video_size_limit': 41943040, 'video_frame_rate_limit': 60, 'video_matrix_limit': 2304000 } } } return instance def _get_mast_api_v1id(path: str) -> int: """Extracts the mastodon Id number from the given path """ masto_id = None id_path = '/api/v1/accounts/:' if not path.startswith(id_path): return None masto_id_str = path.replace(id_path, '') if '/' in masto_id_str: masto_id_str = masto_id_str.split('/')[0] if masto_id_str.isdigit(): masto_id = int(masto_id_str) return masto_id return None def get_masto_api_v1id_from_nickname(nickname: str) -> int: """Given an account nickname return the corresponding mastodon id """ return int.from_bytes(nickname.encode('utf-8'), 'little') def _int_to_bytes(num: int) -> str: """Integer conversion """ if num == 0: return b"" return _int_to_bytes(num // 256) + bytes([num % 256]) def get_nickname_from_masto_api_v1id(masto_id: int) -> str: """Given the mastodon Id return the nickname """ nickname = _int_to_bytes(masto_id).decode() return nickname[::-1] def _get_masto_api_v1account(base_dir: str, nickname: str, domain: str, show_accounts: bool, broch_mode: bool) -> {}: """See https://github.com/McKael/mastodon-documentation/ blob/master/Using-the-API/API.md#account Authorization has already been performed """ account_dir = acct_dir(base_dir, nickname, domain) account_filename = account_dir + '.json' if not os.path.isfile(account_filename): return {} account_json = load_json(account_filename) if not account_json: return {} url_str = get_url_from_post(account_json['icon']['url']) avatar_url = remove_html(url_str) url_str = get_url_from_post(account_json['image']['url']) image_url = remove_html(url_str) joined_date = "2016-10-05T10:30:00Z" if account_json.get('published'): joined_date = account_json['published'] noindex = True if 'indexable' in account_json: if account_json['indexable'] is True: noindex = False discoverable = True if 'discoverable' in account_json: if account_json['discoverable'] is False: discoverable = False group = False bot = False if account_json['type'] == 'Group': group = True elif account_json['type'] != 'Person': bot = True no_of_statuses = 0 no_of_followers = 0 no_of_following = 0 fields = [] published = None if show_accounts and not broch_mode: no_of_followers = lines_in_file(account_dir + '/followers.txt') no_of_following = lines_in_file(account_dir + '/following.txt') # count the number of posts for _, _, files2 in os.walk(account_dir + '/outbox'): no_of_statuses = len(files2) break # get account fields from attachments if account_json.get('attachment'): if isinstance(account_json['attachment'], list): for tag in account_json['attachment']: if not isinstance(tag, dict): continue if not tag.get('name'): continue if not isinstance(tag['name'], str): continue prop_value_name, _ = \ get_attachment_property_value(tag) if not prop_value_name: continue if not tag.get(prop_value_name): continue if not isinstance(tag[prop_value_name], str): continue fields.append({ "name": tag['name'], "value": tag[prop_value_name], "verified_at": None }) published_filename = \ acct_dir(base_dir, nickname, domain) + '/.last_published' if os.path.isfile(published_filename): try: with open(published_filename, 'r', encoding='utf-8') as fp_pub: published = fp_pub.read() except OSError: print('EX: unable to read last published time 1 ' + published_filename) masto_account_json = { "id": get_masto_api_v1id_from_nickname(nickname), "username": nickname, "acct": nickname, "display_name": account_json['name'], "locked": account_json['manuallyApprovesFollowers'], "created_at": joined_date, "followers_count": no_of_followers, "following_count": no_of_following, "statuses_count": no_of_statuses, "note": account_json['summary'], "url": account_json['id'], "avatar": avatar_url, "avatar_static": avatar_url, "header": image_url, "header_static": image_url, "noindex": noindex, "discoverable": discoverable, "group": group, "bot": bot, "emojis": [], "roles": [], "fields": fields } if published: masto_account_json['last_status_at'] = published return masto_account_json def masto_api_v1_response(path: str, calling_domain: str, ua_str: str, authorized: bool, http_prefix: str, base_dir: str, nickname: str, domain: str, domain_full: str, onion_domain: str, i2p_domain: str, translate: {}, registration: bool, system_language: str, project_version: str, custom_emoji: [], show_node_info_accounts: bool, broch_mode: bool) -> ({}, str): """This is a vestigil mastodon API for the purpose of returning a result to sites like https://mastopeek.app-dist.eu """ send_json = None send_json_str = '' if not ua_str: ua_str = '' # parts of the api needing authorization if authorized and nickname: if path == '/api/v1/accounts/verify_credentials': send_json = \ _get_masto_api_v1account(base_dir, nickname, domain, show_node_info_accounts, broch_mode) send_json_str = \ 'masto API account sent for ' + nickname + ' ' + ua_str # information about where the request is coming from calling_info = ' ' + ua_str + ', ' + calling_domain # Parts of the api which don't need authorization masto_id = _get_mast_api_v1id(path) if masto_id is not None: path_nickname = get_nickname_from_masto_api_v1id(masto_id) if path_nickname: original_path = path if string_contains(path, ('/followers?', '/following?', '/streaming/', '/search?', '/relationships?', '/statuses?')): path = path.split('?')[0] if '/streaming/' in path: streaming_msg = \ "Error: Streaming API not implemented on this instance" send_json = { "error": streaming_msg } send_json_str = 'masto API streaming response' if path.endswith('/followers'): send_json = [] send_json_str = \ 'masto API followers sent for ' + nickname + \ calling_info elif path.endswith('/following'): send_json = [] send_json_str = \ 'masto API following sent for ' + nickname + \ calling_info elif path.endswith('/statuses'): send_json = [] send_json_str = \ 'masto API statuses sent for ' + nickname + \ calling_info elif path.endswith('/search'): send_json = [] send_json_str = \ 'masto API search sent ' + original_path + \ calling_info elif path.endswith('/relationships'): send_json = [] send_json_str = \ 'masto API relationships sent ' + original_path + \ calling_info else: send_json = \ _get_masto_api_v1account(base_dir, path_nickname, domain, show_node_info_accounts, broch_mode) send_json_str = \ 'masto API account sent for ' + nickname + \ calling_info # NOTE: adding support for '/api/v1/directory seems to create # federation problems, so avoid implementing that if path.startswith('/api/v1/blocks'): send_json = [] send_json_str = \ 'masto API instance blocks sent ' + path + calling_info elif path.startswith('/api/v1/favorites'): send_json = [] send_json_str = 'masto API favorites sent ' + path + calling_info elif path.startswith('/api/v1/follow_requests'): send_json = [] send_json_str = \ 'masto API follow requests sent ' + path + calling_info elif path.startswith('/api/v1/mutes'): send_json = [] send_json_str = \ 'masto API mutes sent ' + path + calling_info elif path.startswith('/api/v1/notifications'): send_json = [] send_json_str = \ 'masto API notifications sent ' + path + calling_info elif path.startswith('/api/v1/reports'): send_json = [] send_json_str = 'masto API reports sent ' + path + calling_info elif path.startswith('/api/v1/statuses'): send_json = [] send_json_str = 'masto API statuses sent ' + path + calling_info elif path.startswith('/api/v1/timelines'): send_json = { 'error': 'This method requires an authenticated user' } send_json_str = 'masto API timelines sent ' + path + calling_info elif path.startswith('/api/v1/custom_emojis'): send_json = custom_emoji send_json_str = \ 'masto API custom emojis sent ' + path + calling_info admin_nickname = get_config_param(base_dir, 'admin') if admin_nickname and path == '/api/v1/instance': instance_description_short = \ get_config_param(base_dir, 'instanceDescriptionShort') if not instance_description_short: instance_description_short = \ translate['Yet another Epicyon Instance'] instance_description = \ get_config_param(base_dir, 'instanceDescription') instance_title = get_config_param(base_dir, 'instanceTitle') if calling_domain.endswith('.onion') and onion_domain: domain_full = onion_domain http_prefix = 'http' elif (calling_domain.endswith('.i2p') and i2p_domain): domain_full = i2p_domain http_prefix = 'http' if broch_mode: show_node_info_accounts = False send_json = \ _meta_data_instance_v1(show_node_info_accounts, instance_title, instance_description_short, instance_description, http_prefix, base_dir, admin_nickname, domain, domain_full, registration, system_language, project_version) send_json_str = 'masto API instance metadata sent ' + ua_str elif path.startswith('/api/v1/instance/peers'): # This is just a dummy result. # Showing the full list of peers would have privacy implications. # On a large instance you are somewhat lost in the crowd, but on # small instances a full list of peers would convey a lot of # information about the interests of a small number of accounts send_json = ['mastodon.social', domain_full] send_json_str = 'masto API peers metadata sent ' + ua_str elif path.startswith('/api/v1/instance/activity'): send_json = [] send_json_str = 'masto API activity metadata sent ' + ua_str return send_json, send_json_str