acceptreject.py

""" ActivityPub Accept or Reject json """

__filename__ = "acceptreject.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "1.6.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@libreserver.org"
__status__ = "Production"
__module_group__ = "ActivityPub"

import os
from posts import send_signed_json
from flags import has_group_type
from flags import url_permitted
from status import get_status_number
from utils import get_attributed_to
from utils import get_user_paths
from utils import text_in_file
from utils import has_object_string_object
from utils import has_users_path
from utils import get_full_domain
from utils import get_domain_from_actor
from utils import get_nickname_from_actor
from utils import domain_permitted
from utils import follow_person
from utils import acct_dir
from utils import local_actor_url
from utils import has_actor
from utils import has_object_string_type
from utils import get_actor_from_post
from timeFunctions import get_current_time_int


def _create_quote_accept_reject(receiving_actor: str,
                                sending_actor: str,
                                federation_list: [],
                                debug: bool,
                                quote_request_id: str,
                                quote_request_object: str,
                                quote_request_instrument: str,
                                accept_type: str) -> {}:
    """Creates an Accept or Reject response to QuoteRequest
    https://codeberg.org/fediverse/fep/src/branch/main/fep/044f/fep-044f.md
    """
    if not receiving_actor or \
       not sending_actor or \
       not quote_request_id or \
       not quote_request_object or \
       not quote_request_instrument:
        return None
    if not url_permitted(sending_actor, federation_list):
        return None

    status_number, _ = get_status_number()

    new_accept = {
        "@context": [
            "https://www.w3.org/ns/activitystreams",
            {
                "toot": "http://joinmastodon.org/ns#",
                "Quote": "toot:QuoteRequest"
            }
        ],
        "type": accept_type,
        "to": [sending_actor],
        "id": receiving_actor + "/statuses/" + status_number,
        "actor": receiving_actor,
        "object": {
            "type": "QuoteRequest",
            "id": quote_request_id,
            "actor": sending_actor,
            "object": "https://example.com/users/alice/statuses/1",
            "instrument": "https://example.org/users/bob/statuses/1"
        }
    }
    if debug:
        print('REJECT: QuoteRequest ' + str(new_accept))
    return new_accept


def _create_accept_reject(federation_list: [],
                          nickname: str, domain: str, port: int,
                          to_url: str, cc_url: str, http_prefix: str,
                          object_json: {}, accept_type: str) -> {}:
    """Accepts or rejects something (eg. a follow request or offer)
    Typically to_url will be https://www.w3.org/ns/activitystreams#Public
    and cc_url might be a specific person favorited or repeated and
    the followers url objectUrl is typically the url of the message,
    corresponding to url or atomUri in createPostBase
    """
    if not object_json.get('actor'):
        return None

    actor_url = get_actor_from_post(object_json)
    if not url_permitted(actor_url, federation_list):
        return None

    domain = get_full_domain(domain, port)

    new_accept = {
        "@context": [
            'https://www.w3.org/ns/activitystreams',
            'https://w3id.org/security/v1'
        ],
        'type': accept_type,
        'actor': local_actor_url(http_prefix, nickname, domain),
        'to': [to_url],
        'cc': [],
        'object': object_json
    }
    if cc_url:
        new_accept['cc'] = [cc_url]
    return new_accept


def create_accept(federation_list: [],
                  nickname: str, domain: str, port: int,
                  to_url: str, cc_url: str, http_prefix: str,
                  object_json: {}) -> {}:
    """ Create json for ActivityPub Accept """
    return _create_accept_reject(federation_list,
                                 nickname, domain, port,
                                 to_url, cc_url, http_prefix,
                                 object_json, 'Accept')


def create_reject(federation_list: [],
                  nickname: str, domain: str, port: int,
                  to_url: str, cc_url: str, http_prefix: str,
                  object_json: {}) -> {}:
    """ Create json for ActivityPub Reject """
    return _create_accept_reject(federation_list,
                                 nickname, domain, port,
                                 to_url, cc_url,
                                 http_prefix, object_json, 'Reject')


def _reject_quote_request(message_json: {}, domain_full: str,
                          federation_list: [],
                          debug: bool,
                          session, session_onion, session_i2p,
                          base_dir: str,
                          http_prefix: str,
                          send_threads: [], post_log: [],
                          cached_webfingers: {},
                          person_cache: {}, project_version: str,
                          signing_priv_key_pem: str,
                          onion_domain: str, i2p_domain: str,
                          extra_headers: {},
                          sites_unavailable: {},
                          system_language: str,
                          mitm_servers: []) -> bool:
    """ Rejects a QuoteRequest
    https://codeberg.org/fediverse/fep/src/branch/main/fep/044f/fep-044f.md
    """
    sending_actor = None
    receiving_actor = None
    quote_request_id = None
    quote_request_object = None
    quote_request_instrument = None
    if message_json.get('actor'):
        sending_actor = message_json['actor']
    elif message_json.get('instrument'):
        if isinstance(message_json['instrument'], dict):
            instrument_dict = message_json['instrument']
            if instrument_dict.get('attributedTo'):
                sending_actor = \
                    get_attributed_to(instrument_dict['attributedTo'])
            if instrument_dict.get('to'):
                if isinstance(instrument_dict['to'], str):
                    receiving_actor = instrument_dict['to']
                elif isinstance(instrument_dict['to'], list):
                    for receiver in instrument_dict['to']:
                        if '#Public' not in receiver and \
                           '://' + domain_full + '/' in receiver:
                            receiving_actor = receiver
                            break
            if instrument_dict.get('id'):
                quote_request_instrument = instrument_dict['id']
            if instrument_dict.get('object'):
                quote_request_object = instrument_dict['object']
    if message_json.get('id'):
        quote_request_id = message_json['id']
    if not sending_actor:
        return False
    if receiving_actor:
        quote_request_object = receiving_actor
    reject_json = \
        _create_quote_accept_reject(receiving_actor,
                                    sending_actor,
                                    federation_list,
                                    debug,
                                    quote_request_id,
                                    quote_request_object,
                                    quote_request_instrument,
                                    'Reject')
    if reject_json:
        print('REJECT: QuoteRequest from ' + sending_actor)
        nickname = get_nickname_from_actor(receiving_actor)
        domain, from_port = get_domain_from_actor(receiving_actor)
        nickname_to_follow = get_nickname_from_actor(sending_actor)
        domain_to_follow, port = get_domain_from_actor(sending_actor)
        group_account = \
            has_group_type(base_dir, receiving_actor, person_cache)
        if nickname and domain and \
           nickname_to_follow and domain_to_follow:
            if debug:
                print('REJECT: QuoteRequest sending reject ' +
                      str(reject_json))

            curr_session = session
            curr_domain = domain
            curr_port = from_port
            curr_http_prefix = http_prefix
            if onion_domain and \
               not curr_domain.endswith('.onion') and \
               domain_to_follow.endswith('.onion'):
                curr_session = session_onion
                curr_http_prefix = 'http'
                curr_domain = onion_domain
                curr_port = 80
                port = 80
                if debug:
                    print('Domain switched from ' + domain +
                          ' to ' + curr_domain)
            elif (i2p_domain and
                  not curr_domain.endswith('.i2p') and
                  domain_to_follow.endswith('.i2p')):
                curr_session = session_i2p
                curr_http_prefix = 'http'
                curr_domain = i2p_domain
                curr_port = 80
                port = 80
                if debug:
                    print('Domain switched from ' + domain +
                          ' to ' + curr_domain)

            client_to_server = False
            send_signed_json(reject_json, curr_session, base_dir,
                             nickname_to_follow, domain_to_follow, port,
                             nickname, domain, curr_port,
                             curr_http_prefix, client_to_server,
                             federation_list,
                             send_threads, post_log, cached_webfingers,
                             person_cache, debug, project_version, None,
                             group_account, signing_priv_key_pem,
                             726235284, curr_domain, onion_domain, i2p_domain,
                             extra_headers, sites_unavailable,
                             system_language, mitm_servers)
        return True
    return False


def _accept_follow(base_dir: str, message_json: {},
                   federation_list: [], debug: bool,
                   curr_domain: str,
                   onion_domain: str, i2p_domain: str) -> None:
    """ Receiving an ActivityPub follow Accept activity
    Your follow was accepted
    """
    if not has_object_string_type(message_json, debug):
        return
    if message_json['object']['type'] not in ('Follow', 'Join'):
        return
    if debug:
        print('DEBUG: receiving Follow activity')
    if not message_json['object'].get('actor'):
        print('DEBUG: no actor in Follow activity')
        return
    # no, this isn't a mistake
    if not has_object_string_object(message_json, debug):
        return
    if not message_json.get('to'):
        if debug:
            print('DEBUG: No "to" parameter in follow Accept')
        return
    if debug:
        print('DEBUG: follow Accept received ' + str(message_json))
    this_actor = get_actor_from_post(message_json['object'])
    nickname = get_nickname_from_actor(this_actor)
    if not nickname:
        print('WARN: no nickname found in ' + this_actor)
        return
    accepted_domain, accepted_port = get_domain_from_actor(this_actor)
    if not accepted_domain:
        if debug:
            print('DEBUG: domain not found in ' + this_actor)
        return
    if not nickname:
        if debug:
            print('DEBUG: nickname not found in ' + this_actor)
        return
    if accepted_port:
        if '/' + accepted_domain + ':' + str(accepted_port) + \
           '/users/' + nickname not in this_actor:
            if debug:
                print('Port: ' + str(accepted_port))
                print('Expected: /' + accepted_domain + ':' +
                      str(accepted_port) + '/users/' + nickname)
                print('Actual:   ' + this_actor)
                print('DEBUG: unrecognized actor ' + this_actor)
            return
    else:
        actor_found = False
        users_list = get_user_paths()
        for users_str in users_list:
            if '/' + accepted_domain + users_str + nickname in this_actor:
                actor_found = True
                break

        if not actor_found:
            if debug:
                print('Expected: /' + accepted_domain + '/users/' + nickname)
                print('Actual:   ' + this_actor)
                print('DEBUG: unrecognized actor ' + this_actor)
            return
    followed_actor = message_json['object']['object']
    followed_domain, port = get_domain_from_actor(followed_actor)
    if not followed_domain:
        print('DEBUG: no domain found within Follow activity object ' +
              followed_actor)
        return
    followed_domain_full = followed_domain
    if port:
        followed_domain_full = followed_domain + ':' + str(port)
    followed_nickname = get_nickname_from_actor(followed_actor)
    if not followed_nickname:
        print('DEBUG: no nickname found within Follow activity object ' +
              followed_actor)
        return

    # convert from onion/i2p to clearnet accepted domain
    if onion_domain:
        if accepted_domain.endswith('.onion') and \
           not curr_domain.endswith('.onion'):
            accepted_domain = curr_domain
    if i2p_domain:
        if accepted_domain.endswith('.i2p') and \
           not curr_domain.endswith('.i2p'):
            accepted_domain = curr_domain

    accepted_domain_full = accepted_domain
    if accepted_port:
        accepted_domain_full = accepted_domain + ':' + str(accepted_port)

    # has this person already been unfollowed?
    unfollowed_filename = \
        acct_dir(base_dir, nickname, accepted_domain_full) + '/unfollowed.txt'
    if os.path.isfile(unfollowed_filename):
        if text_in_file(followed_nickname + '@' + followed_domain_full,
                        unfollowed_filename):
            if debug:
                print('DEBUG: follow accept arrived for ' +
                      nickname + '@' + accepted_domain_full +
                      ' from ' +
                      followed_nickname + '@' + followed_domain_full +
                      ' but they have been unfollowed')
            return

    # does the url path indicate that this is a group actor
    group_account = has_group_type(base_dir, followed_actor, None, debug)
    if debug:
        print('Accepted follow is a group: ' + str(group_account) +
              ' ' + followed_actor + ' ' + base_dir)

    if follow_person(base_dir,
                     nickname, accepted_domain_full,
                     followed_nickname, followed_domain_full,
                     federation_list, debug, group_account,
                     'following.txt'):
        if debug:
            print('DEBUG: ' + nickname + '@' + accepted_domain_full +
                  ' followed ' +
                  followed_nickname + '@' + followed_domain_full)
    else:
        if debug:
            print('DEBUG: Unable to create follow - ' +
                  nickname + '@' + accepted_domain + ' -> ' +
                  followed_nickname + '@' + followed_domain)


def receive_accept_reject(base_dir: str, domain: str, message_json: {},
                          federation_list: [], debug: bool, curr_domain: str,
                          onion_domain: str, i2p_domain: str) -> bool:
    """Receives an Accept or Reject within the POST section of HTTPServer
    """
    if message_json['type'] not in ('Accept', 'Reject'):
        return False
    if not has_actor(message_json, debug):
        return False
    actor_url = get_actor_from_post(message_json)
    if not has_users_path(actor_url):
        if debug:
            print('DEBUG: "users" or "profile" missing from actor in ' +
                  message_json['type'] + '. Assuming single user instance.')
    domain, _ = get_domain_from_actor(actor_url)
    if not domain_permitted(domain, federation_list):
        if debug:
            print('DEBUG: ' + message_json['type'] +
                  ' from domain not permitted - ' + domain)
        return False
    nickname = get_nickname_from_actor(actor_url)
    if not nickname:
        # single user instance
        nickname = 'dev'
        if debug:
            print('DEBUG: ' + message_json['type'] +
                  ' does not contain a nickname. ' +
                  'Assuming single user instance.')
    # receive follow accept
    _accept_follow(base_dir, message_json, federation_list, debug,
                   curr_domain, onion_domain, i2p_domain)
    if debug:
        print('DEBUG: Uh, ' + message_json['type'] + ', I guess')
    return True


def receive_quote_request(message_json: {}, federation_list: [],
                          debug: bool,
                          domain_full: str,
                          session, session_onion, session_i2p,
                          base_dir: str,
                          http_prefix: str,
                          send_threads: [], post_log: [],
                          cached_webfingers: {},
                          person_cache: {}, project_version: str,
                          signing_priv_key_pem: str,
                          onion_domain: str,
                          i2p_domain: str,
                          extra_headers: {},
                          sites_unavailable: {},
                          system_language: str,
                          mitm_servers: [],
                          last_quote_request) -> bool:
    """Receives a QuoteRequest within the POST section of HTTPServer
    https://codeberg.org/fediverse/fep/src/branch/main/fep/044f/fep-044f.md
    """
    if message_json['type'] != 'QuoteRequest':
        return False
    curr_time = get_current_time_int()
    seconds_since_last_quote_request = curr_time - last_quote_request
    if seconds_since_last_quote_request < 30:
        # don't handle quote requests too often
        return True
    _reject_quote_request(message_json, domain_full,
                          federation_list, debug,
                          session, session_onion, session_i2p, base_dir,
                          http_prefix,
                          send_threads, post_log,
                          cached_webfingers,
                          person_cache, project_version,
                          signing_priv_key_pem,
                          onion_domain,
                          i2p_domain,
                          extra_headers,
                          sites_unavailable,
                          system_language,
                          mitm_servers)

    return True