from urllib.parse import urlparse from typing import Dict, Any from typing import Optional import logging import requests from .urlutils import check_url logger = logging.getLogger(__name__) def webfinger(resource: str) -> Optional[Dict[str, Any]]: """Mastodon-like WebFinger resolution to retrieve the activity stream Actor URL. """ logger.info(f'performing webfinger resolution for {resource}') protos = ['https', 'http'] if resource.startswith('http://'): protos.reverse() host = urlparse(resource).netloc elif resource.startswith('https://'): host = urlparse(resource).netloc else: if resource.startswith('acct:'): resource = resource[5:] if resource.startswith('@'): resource = resource[1:] _, host = resource.split('@', 1) resource='acct:'+resource # Security check on the url (like not calling localhost) check_url(f'https://{host}') for i, proto in enumerate(protos): try: url = f'{proto}://{host}/.well-known/webfinger' resp = requests.get( url, {'resource': resource} ) except requests.ConnectionError: # If we tried https first and the domain is "http only" if i == 0: continue break if resp.status_code == 404: return None resp.raise_for_status() return resp.json() def get_remote_follow_template(resource: str) -> Optional[str]: data = webfinger(resource) if data is None: return None for link in data['links']: if link.get('rel') == 'http://ostatus.org/schema/1.0/subscribe': return link.get('template') return None def get_actor_url(resource: str) -> Optional[str]: """Mastodon-like WebFinger resolution to retrieve the activity stream Actor URL. Returns: the Actor URL or None if the resolution failed. """ data = webfinger(resource) if data is None: return None for link in data['links']: if link.get('rel') == 'self' and link.get('type') == 'application/activity+json': return link.get('href') return None