protocol.py

"""Base protocol class and common code."""
import copy
from datetime import datetime, timedelta, timezone
import logging
import os
import re
from threading import Lock
from urllib.parse import urljoin, urlparse

from cachetools import cached, LRUCache
from flask import request
from google.cloud import ndb
from google.cloud.ndb import OR
from google.cloud.ndb.model import _entity_to_protobuf
from granary import as1, as2, source
from granary.source import html_to_text
from oauth_dropins.webutil.appengine_info import DEBUG
from oauth_dropins.webutil.flask_util import cloud_tasks_only
from oauth_dropins.webutil import models
from oauth_dropins.webutil import util
from oauth_dropins.webutil.util import json_dumps, json_loads
from requests import RequestException
import werkzeug.exceptions
from werkzeug.exceptions import BadGateway, HTTPException

import common
from common import (
    DOMAIN_BLOCKLIST,
    DOMAIN_RE,
    DOMAINS,
    PRIMARY_DOMAIN,
    PROTOCOL_DOMAINS,
    report_error,
    subdomain_wrap,
)
import dms
import ids
from ids import (
    BOT_ACTOR_AP_IDS,
    normalize_user_id,
    translate_object_id,
    translate_user_id,
)
import memcache
from models import (
    DM,
    Follower,
    Object,
    PROTOCOLS,
    PROTOCOLS_BY_KIND,
    Target,
    User,
)

OBJECT_REFRESH_AGE = timedelta(days=30)
DELETE_TASK_DELAY = timedelta(minutes=2)
CREATE_MAX_AGE = timedelta(weeks=2)

# require a follow for users on these domains before we deliver anything from
# them other than their profile
LIMITED_DOMAINS = (os.getenv('LIMITED_DOMAINS', '').split()
                   or util.load_file_lines('limited_domains'))

DONT_STORE_AS1_TYPES = as1.CRUD_VERBS | set((
    'accept',
    'reject',
    'stop-following',
    'undo',
))
STORE_AS1_TYPES = (as1.ACTOR_TYPES | as1.POST_TYPES | as1.VERBS_WITH_OBJECT
                   - DONT_STORE_AS1_TYPES)

logger = logging.getLogger(__name__)


def error(*args, status=299, **kwargs):
    """Default HTTP status code to 299 to prevent retrying task."""
    return common.error(*args, status=status, **kwargs)


class ErrorButDoNotRetryTask(HTTPException):
    code = 299
    description = 'ErrorButDoNotRetryTask'

# https://github.com/pallets/flask/issues/1837#issuecomment-304996942
werkzeug.exceptions.default_exceptions.setdefault(299, ErrorButDoNotRetryTask)
werkzeug.exceptions._aborter.mapping.setdefault(299, ErrorButDoNotRetryTask)


def activity_id_memcache_key(id):
    return memcache.key(f'receive-{id}')


class Protocol:
    """Base protocol class. Not to be instantiated; classmethods only."""
    ABBREV = None
    """str: lower case abbreviation, used in URL paths"""
    PHRASE = None
    """str: human-readable name or phrase. Used in phrases like ``Follow this person on {PHRASE}``"""
    OTHER_LABELS = ()
    """sequence of str: label aliases"""
    LOGO_HTML = ''
    """str: logo emoji or ``<img>`` tag"""
    CONTENT_TYPE = None
    """str: MIME type of this protocol's native data format, appropriate for the ``Content-Type`` HTTP header."""
    HAS_COPIES = False
    """bool: whether this protocol is push and needs us to proactively create "copy" users and objects, as opposed to pulling converted objects on demand"""
    REQUIRES_AVATAR = False
    """bool: whether accounts on this protocol are required to have a profile picture. If they don't, their ``User.status`` will be ``blocked``."""
    REQUIRES_NAME = False
    """bool: whether accounts on this protocol are required to have a profile name that's different than their handle or id. If they don't, their ``User.status`` will be ``blocked``."""
    REQUIRES_OLD_ACCOUNT = False
    """bool: whether accounts on this protocol are required to be at least :const:`common.OLD_ACCOUNT_AGE` old. If their profile includes creation date and it's not old enough, their ``User.status`` will be ``blocked``."""
    DEFAULT_ENABLED_PROTOCOLS = ()
    """sequence of str: labels of other protocols that are automatically enabled for this protocol to bridge into"""
    DEFAULT_SERVE_USER_PAGES = False
    """bool: whether to serve user pages for all of this protocol's users on the fed.brid.gy. If ``False``, user pages will only be served for users who have explictly opted in."""
    SUPPORTED_AS1_TYPES = ()
    """sequence of str: AS1 objectTypes and verbs that this protocol supports receiving and sending"""
    SUPPORTS_DMS = False
    """bool: whether this protocol can receive DMs (chat messages)"""

    def __init__(self):
        assert False

    @classmethod
    @property
    def LABEL(cls):
        """str: human-readable lower case name of this protocol, eg ``'activitypub``"""
        return cls.__name__.lower()

    @staticmethod
    def for_request(fed=None):
        """Returns the protocol for the current request.

        ...based on the request's hostname.

        Args:
          fed (str or protocol.Protocol): protocol to return if the current
            request is on ``fed.brid.gy``

        Returns:
          Protocol: protocol, or None if the provided domain or request hostname
          domain is not a subdomain of ``brid.gy`` or isn't a known protocol
        """
        return Protocol.for_bridgy_subdomain(request.host, fed=fed)

    @staticmethod
    def for_bridgy_subdomain(domain_or_url, fed=None):
        """Returns the protocol for a brid.gy subdomain.

        Args:
          domain_or_url (str)
          fed (str or protocol.Protocol): protocol to return if the current
            request is on ``fed.brid.gy``

        Returns:
          class: :class:`Protocol` subclass, or None if the provided domain or request
          hostname domain is not a subdomain of ``brid.gy`` or isn't a known
          protocol
        """
        domain = (util.domain_from_link(domain_or_url, minimize=False)
                  if util.is_web(domain_or_url)
                  else domain_or_url)

        if domain == common.PRIMARY_DOMAIN or domain in common.LOCAL_DOMAINS:
            return PROTOCOLS[fed] if isinstance(fed, str) else fed
        elif domain and domain.endswith(common.SUPERDOMAIN):
            label = domain.removesuffix(common.SUPERDOMAIN)
            return PROTOCOLS.get(label)

    @classmethod
    def owns_id(cls, id):
        """Returns whether this protocol owns the id, or None if it's unclear.

        To be implemented by subclasses.

        IDs are string identities that uniquely identify users, and are intended
        primarily to be machine readable and usable. Compare to handles, which
        are human-chosen, human-meaningful, and often but not always unique.

        Some protocols' ids are more or less deterministic based on the id
        format, eg AT Protocol owns ``at://`` URIs. Others, like http(s) URLs,
        could be owned by eg Web or ActivityPub.

        This should be a quick guess without expensive side effects, eg no
        external HTTP fetches to fetch the id itself or otherwise perform
        discovery.

        Returns False if the id's domain is in :const:`common.DOMAIN_BLOCKLIST`.

        Args:
          id (str)

        Returns:
          bool or None:
        """
        return False

    @classmethod
    def owns_handle(cls, handle, allow_internal=False):
        """Returns whether this protocol owns the handle, or None if it's unclear.

        To be implemented by subclasses.

        Handles are string identities that are human-chosen, human-meaningful,
        and often but not always unique. Compare to IDs, which uniquely identify
        users, and are intended primarily to be machine readable and usable.

        Some protocols' handles are more or less deterministic based on the id
        format, eg ActivityPub (technically WebFinger) handles are
        ``@user@instance.com``. Others, like domains, could be owned by eg Web,
        ActivityPub, AT Protocol, or others.

        This should be a quick guess without expensive side effects, eg no
        external HTTP fetches to fetch the id itself or otherwise perform
        discovery.

        Args:
          handle (str)
          allow_internal (bool): whether to return False for internal domains
            like ``fed.brid.gy``, ``bsky.brid.gy``, etc

        Returns:
          bool or None
        """
        return False

    @classmethod
    def handle_to_id(cls, handle):
        """Converts a handle to an id.

        To be implemented by subclasses.

        May incur network requests, eg DNS queries or HTTP requests. Avoids
        blocked or opted out users.

        Args:
          handle (str)

        Returns:
          str: corresponding id, or None if the handle can't be found
        """
        raise NotImplementedError()

    @classmethod
    def key_for(cls, id, allow_opt_out=False):
        """Returns the :class:`google.cloud.ndb.Key` for a given id's :class:`models.User`.

        To be implemented by subclasses. Canonicalizes the id if necessary.

        If called via `Protocol.key_for`, infers the appropriate protocol with
        :meth:`for_id`. If called with a concrete subclass, uses that subclass
        as is.

        Args:
          id (str):
          allow_opt_out (bool): whether to allow users who are currently opted out

        Returns:
          google.cloud.ndb.Key: matching key, or None if the given id is not a
          valid :class:`User` id for this protocol.
        """
        if cls == Protocol:
            proto = Protocol.for_id(id)
            return proto.key_for(id, allow_opt_out=allow_opt_out) if proto else None

        # load user so that we follow use_instead
        existing = cls.get_by_id(id, allow_opt_out=True)
        if existing:
            if existing.status and not allow_opt_out:
                return None
            return existing.key

        return cls(id=id).key

    @staticmethod
    def _for_id_memcache_key(id, remote=None):
        """If id is a URL, uses its domain, otherwise returns None.

        Args:
          id (str)

        Returns:
          (str domain, bool remote) or None
        """
        if remote and util.is_web(id):
            return util.domain_from_link(id)

    @cached(LRUCache(20000), lock=Lock())
    @memcache.memoize(key=_for_id_memcache_key, write=lambda id, remote: remote,
                      version=3)
    @staticmethod
    def for_id(id, remote=True):
        """Returns the protocol for a given id.

        Args:
          id (str)
          remote (bool): whether to perform expensive side effects like fetching
            the id itself over the network, or other discovery.

        Returns:
          Protocol subclass: matching protocol, or None if no single known
          protocol definitively owns this id
        """
        logger.debug(f'Determining protocol for id {id}')
        if not id:
            return None

        # remove our synthetic id fragment, if any
        #
        # will this eventually cause false positives for other services that
        # include our full ids inside their own ids, non-URL-encoded? guess
        # we'll figure that out if/when it happens.
        id = id.partition('#bridgy-fed-')[0]
        if not id:
            return None

        if util.is_web(id):
            # step 1: check for our per-protocol subdomains
            try:
                is_homepage = urlparse(id).path.strip('/') == ''
            except ValueError as e:
                logger.info(f'urlparse ValueError: {e}')
                return None

            by_subdomain = Protocol.for_bridgy_subdomain(id)
            if by_subdomain and not is_homepage and id not in BOT_ACTOR_AP_IDS:
                logger.debug(f'  {by_subdomain.LABEL} owns id {id}')
                return by_subdomain

        # step 2: check if any Protocols say conclusively that they own it
        # sort to be deterministic
        protocols = sorted(set(p for p in PROTOCOLS.values() if p),
                           key=lambda p: p.LABEL)
        candidates = []
        for protocol in protocols:
            owns = protocol.owns_id(id)
            if owns:
                logger.debug(f'  {protocol.LABEL} owns id {id}')
                return protocol
            elif owns is not False:
                candidates.append(protocol)

        if len(candidates) == 1:
            logger.debug(f'  {candidates[0].LABEL} owns id {id}')
            return candidates[0]

        # step 3: look for existing Objects in the datastore
        #
        # note that we don't currently see if this is a copy id because I have FUD
        # over which Protocol for_id should return in that case...and also because a
        # protocol may already say definitively above that it owns the id, eg ATProto
        # with DIDs and at:// URIs.
        obj = Protocol.load(id, remote=False)
        if obj and obj.source_protocol:
            logger.debug(f'  {obj.key.id()} owned by source_protocol {obj.source_protocol}')
            return PROTOCOLS[obj.source_protocol]

        # step 4: fetch over the network, if necessary
        if not remote:
            return None

        for protocol in candidates:
            logger.debug(f'Trying {protocol.LABEL}')
            try:
                obj = protocol.load(id, local=False, remote=True)

                if protocol.ABBREV == 'web':
                    # for web, if we fetch and get HTML without microformats,
                    # load returns False but the object will be stored in the
                    # datastore with source_protocol web, and in cache. load it
                    # again manually to check for that.
                    obj = Object.get_by_id(id)
                    if obj and obj.source_protocol != 'web':
                        obj = None

                if obj:
                    logger.debug(f'  {protocol.LABEL} owns id {id}')
                    return protocol
            except BadGateway:
                # we tried and failed fetching the id over the network.
                # this depends on ActivityPub.fetch raising this!
                return None
            except HTTPException as e:
                # internal error we generated ourselves; try next protocol
                pass
            except Exception as e:
                code, _ = util.interpret_http_exception(e)
                if code:
                    # we tried and failed fetching the id over the network
                    return None
                raise

        logger.info(f'No matching protocol found for {id} !')
        return None

    @cached(LRUCache(20000), lock=Lock())
    @staticmethod
    def for_handle(handle):
        """Returns the protocol for a given handle.

        May incur expensive side effects like resolving the handle itself over
        the network or other discovery.

        Args:
          handle (str)

        Returns:
          (Protocol subclass, str) tuple: matching protocol and optional id (if
          resolved), or ``(None, None)`` if no known protocol owns this handle
        """
        # TODO: normalize, eg convert domains to lower case
        logger.debug(f'Determining protocol for handle {handle}')
        if not handle:
            return (None, None)

        # step 1: check if any Protocols say conclusively that they own it.
        # sort to be deterministic.
        protocols = sorted(set(p for p in PROTOCOLS.values() if p),
                           key=lambda p: p.LABEL)
        candidates = []
        for proto in protocols:
            owns = proto.owns_handle(handle)
            if owns:
                logger.debug(f'  {proto.LABEL} owns handle {handle}')
                return (proto, None)
            elif owns is not False:
                candidates.append(proto)

        if len(candidates) == 1:
            logger.debug(f'  {candidates[0].LABEL} owns handle {handle}')
            return (candidates[0], None)

        # step 2: look for matching User in the datastore
        for proto in candidates:
            user = proto.query(proto.handle == handle).get()
            if user:
                if user.status:
                    return (None, None)
                logger.debug(f'  user {user.key} handle {handle}')
                return (proto, user.key.id())

        # step 3: resolve handle to id
        for proto in candidates:
            id = proto.handle_to_id(handle)
            if id:
                logger.debug(f'  {proto.LABEL} resolved handle {handle} to id {id}')
                return (proto, id)

        logger.info(f'No matching protocol found for handle {handle} !')
        return (None, None)

    @classmethod
    def bridged_web_url_for(cls, user, fallback=False):
        """Returns the web URL for a user's bridged profile in this protocol.

        For example, for Web user ``alice.com``, :meth:`ATProto.bridged_web_url_for`
        returns ``https://bsky.app/profile/alice.com.web.brid.gy``

        Args:
          user (models.User)
          fallback (bool): if True, and bridged users have no canonical user
            profile URL in this protocol, return the native protocol's profile URL

        Returns:
          str, or None if there isn't a canonical URL
        """
        if fallback:
            return user.web_url()

    @classmethod
    def actor_key(cls, obj, allow_opt_out=False):
        """Returns the :class:`User`: key for a given object's author or actor.

        Args:
          obj (models.Object)
          allow_opt_out (bool): whether to return a user key if they're opted out

        Returns:
          google.cloud.ndb.key.Key or None:
        """
        owner = as1.get_owner(obj.as1)
        if owner:
            return cls.key_for(owner, allow_opt_out=allow_opt_out)

    @classmethod
    def bot_user_id(cls):
        """Returns the Web user id for the bot user for this protocol.

        For example, ``'bsky.brid.gy'`` for ATProto.

        Returns:
          str:
        """
        return f'{cls.ABBREV}{common.SUPERDOMAIN}'

    @classmethod
    def create_for(cls, user):
        """Creates or re-activate a copy user in this protocol.

        Should add the copy user to :attr:`copies`.

        If the copy user already exists and active, should do nothing.

        Args:
          user (models.User): original source user. Shouldn't already have a
            copy user for this protocol in :attr:`copies`.

        Raises:
          ValueError: if we can't create a copy of the given user in this protocol
        """
        raise NotImplementedError()

    @classmethod
    def send(to_cls, obj, url, from_user=None, orig_obj_id=None):
        """Sends an outgoing activity.

        To be implemented by subclasses.

        NOTE: if this protocol's ``HAS_COPIES`` is True, and this method creates
        a copy and sends it, it *must* add that copy to the *object*'s (not
        activity's) :attr:`copies`!

        Args:
          obj (models.Object): with activity to send
          url (str): destination URL to send to
          from_user (models.User): user (actor) this activity is from
          orig_obj_id (str): :class:`models.Object` key id of the "original object"
            that this object refers to, eg replies to or reposts or likes

        Returns:
          bool: True if the activity is sent successfully, False if it is
          ignored or otherwise unsent due to protocol logic, eg no webmention
          endpoint, protocol doesn't support the activity type. (Failures are
          raised as exceptions.)

        Raises:
          werkzeug.HTTPException if the request fails
        """
        raise NotImplementedError()

    @classmethod
    def fetch(cls, obj, **kwargs):
        """Fetches a protocol-specific object and populates it in an :class:`Object`.

        Errors are raised as exceptions. If this method returns False, the fetch
        didn't fail but didn't succeed either, eg the id isn't valid for this
        protocol, or the fetch didn't return valid data for this protocol.

        To be implemented by subclasses.

        Args:
          obj (models.Object): with the id to fetch. Data is filled into one of
            the protocol-specific properties, eg ``as2``, ``mf2``, ``bsky``.
          kwargs: subclass-specific

        Returns:
          bool: True if the object was fetched and populated successfully,
          False otherwise

        Raises:
          requests.RequestException or werkzeug.HTTPException: if the fetch fails
        """
        raise NotImplementedError()

    @classmethod
    def convert(cls, obj, from_user=None, **kwargs):
        """Converts an :class:`Object` to this protocol's data format.

        For example, an HTML string for :class:`Web`, or a dict with AS2 JSON
        and ``application/activity+json`` for :class:`ActivityPub`.

        Just passes through to :meth:`_convert`, then does minor
        protocol-independent postprocessing.

        Args:
          obj (models.Object):
          from_user (models.User): user (actor) this activity/object is from
          kwargs: protocol-specific, passed through to :meth:`_convert`

        Returns:
          converted object in the protocol's native format, often a dict
        """
        if not obj or not obj.as1:
            return {}

        id = obj.key.id() if obj.key else obj.as1.get('id')
        is_activity = obj.as1.get('verb') in ('post', 'update')
        base_obj = as1.get_object(obj.as1) if is_activity else obj.as1
        orig_our_as1 = obj.our_as1

        # mark bridged actors as bots and add "bridged by Bridgy Fed" to their bios
        if (from_user and base_obj
            and base_obj.get('objectType') in as1.ACTOR_TYPES
            and PROTOCOLS.get(obj.source_protocol) != cls
            and Protocol.for_bridgy_subdomain(id) not in DOMAINS
            # Web users are special cased, they don't get the label if they've
            # explicitly enabled Bridgy Fed with redirects or webmentions
            and not (from_user.LABEL == 'web'
                     and (from_user.last_webmention_in or from_user.has_redirects))):

            obj.our_as1 = copy.deepcopy(obj.as1)
            actor = as1.get_object(obj.as1) if is_activity else obj.as1
            actor['objectType'] = 'person'
            cls.add_source_links(actor=actor, obj=obj, from_user=from_user)

        converted = cls._convert(obj, from_user=from_user, **kwargs)
        obj.our_as1 = orig_our_as1
        return converted

    @classmethod
    def _convert(cls, obj, from_user=None, **kwargs):
        """Converts an :class:`Object` to this protocol's data format.

        To be implemented by subclasses. Implementations should generally call
        :meth:`Protocol.translate_ids` (as their own class) before converting to
        their format.

        Args:
          obj (models.Object):
          from_user (models.User): user (actor) this activity/object is from
          kwargs: protocol-specific

        Returns:
          converted object in the protocol's native format, often a dict. May
            return the ``{}`` empty dict if the object can't be converted.
        """
        raise NotImplementedError()

    @classmethod
    def add_source_links(cls, actor, obj, from_user):
        """Adds "bridged from ... by Bridgy Fed" HTML to ``actor['summary']``.

        Default implementation; subclasses may override.

        Args:
          actor (dict): AS1 actor
          obj (models.Object):
          from_user (models.User): user (actor) this activity/object is from
        """
        assert from_user
        summary = actor.setdefault('summary', '')
        if 'Bridgy Fed]' in html_to_text(summary, ignore_links=True):
            return

        id = actor.get('id')
        proto_phrase = (PROTOCOLS[obj.source_protocol].PHRASE
                        if obj.source_protocol else '')
        if proto_phrase:
            proto_phrase = f' on {proto_phrase}'

        if from_user.key and id in (from_user.key.id(), from_user.profile_id()):
            source_links = f'[<a href="https://{PRIMARY_DOMAIN}{from_user.user_page_path()}">bridged</a> from <a href="{from_user.web_url()}">{from_user.handle}</a>{proto_phrase} by <a href="https://{PRIMARY_DOMAIN}/">Bridgy Fed</a>]'

        else:
            url = as1.get_url(actor) or id
            source = util.pretty_link(url) if url else '?'
            source_links = f'[bridged from {source}{proto_phrase} by <a href="https://{PRIMARY_DOMAIN}/">Bridgy Fed</a>]'

        if summary:
            summary += '<br><br>'
        actor['summary'] = summary + source_links

    @classmethod
    def set_username(to_cls, user, username):
        """Sets a custom username for a user's bridged account in this protocol.

        Args:
          user (models.User)
          username (str)

        Raises:
          ValueError: if the username is invalid
          RuntimeError: if the username could not be set
        """
        raise NotImplementedError()

    @classmethod
    def migrate_out(cls, user, to_user_id):
        """Migrates a bridged account out to be a native account.

        Args:
          user (models.User)
          to_user_id (str)

        Raises:
          ValueError: eg if this protocol doesn't own ``to_user_id``, or if
            ``user`` is on this protocol or not bridged to this protocol
        """
        raise NotImplementedError()

    @classmethod
    def migrate_in(cls, user, from_user_id, **kwargs):
        """Migrates a native account in to be a bridged account.

        Args:
          user (models.User): native user on another protocol to attach the
            newly imported bridged account to
          from_user_id (str)
          kwargs: additional protocol-specific parameters

        Raises:
          ValueError: eg if this protocol doesn't own ``from_user_id``, or if
            ``user`` is on this protocol or already bridged to this protocol
        """
        raise NotImplementedError()

    @classmethod
    def target_for(cls, obj, shared=False):
        """Returns an :class:`Object`'s delivery target (endpoint).

        To be implemented by subclasses.

        Examples:

        * If obj has ``source_protocol`` ``web``, returns its URL, as a
          webmention target.
        * If obj is an ``activitypub`` actor, returns its inbox.
        * If obj is an ``activitypub`` object, returns it's author's or actor's
          inbox.

        Args:
          obj (models.Object):
          shared (bool): optional. If True, returns a common/shared
            endpoint, eg ActivityPub's ``sharedInbox``, that can be reused for
            multiple recipients for efficiency

        Returns:
          str: target endpoint, or None if not available.
        """
        raise NotImplementedError()

    @classmethod
    def is_blocklisted(cls, url, allow_internal=False):
        """Returns True if we block the given URL and shouldn't deliver to it.

        Default implementation here, subclasses may override.

        Args:
          url (str):
          allow_internal (bool): whether to return False for internal domains
            like ``fed.brid.gy``, ``bsky.brid.gy``, etc
        """
        blocklist = DOMAIN_BLOCKLIST
        if not allow_internal:
            blocklist += DOMAINS
        return util.domain_or_parent_in(util.domain_from_link(url), blocklist)

    @classmethod
    def translate_ids(to_cls, obj):
        """Translates all ids in an AS1 object to a specific protocol.

        Infers source protocol for each id value separately.

        For example, if ``proto`` is :class:`ActivityPub`, the ATProto URI
        ``at://did:plc:abc/coll/123`` will be converted to
        ``https://bsky.brid.gy/ap/at://did:plc:abc/coll/123``.

        Wraps these AS1 fields:

        * ``id``
        * ``actor``
        * ``author``
        * ``bcc``
        * ``bto``
        * ``cc``
        * ``object``
        * ``object.actor``
        * ``object.author``
        * ``object.id``
        * ``object.inReplyTo``
        * ``object.object``
        * ``attachments[].id``
        * ``tags[objectType=mention].url``
        * ``to``

        This is the inverse of :meth:`models.Object.resolve_ids`. Much of the
        same logic is duplicated there!

        TODO: unify with :meth:`Object.resolve_ids`,
        :meth:`models.Object.normalize_ids`.

        Args:
          to_proto (Protocol subclass)
          obj (dict): AS1 object or activity (not :class:`models.Object`!)

        Returns:
          dict: wrapped AS1 version of ``obj``
        """
        assert to_cls != Protocol
        if not obj:
            return obj

        outer_obj = copy.deepcopy(obj)
        inner_objs = outer_obj['object'] = as1.get_objects(outer_obj)

        def translate(elem, field, fn, uri=False):
            elem[field] = as1.get_objects(elem, field)
            for obj in elem[field]:
                if id := obj.get('id'):
                    if field in ('to', 'cc', 'bcc', 'bto') and as1.is_audience(id):
                        continue
                    from_cls = Protocol.for_id(id)
                    # TODO: what if from_cls is None? relax translate_object_id,
                    # make it a noop if we don't know enough about from/to?
                    if from_cls and from_cls != to_cls:
                        obj['id'] = fn(id=id, from_=from_cls, to=to_cls)
                    if obj['id'] and uri:
                        obj['id'] = to_cls(id=obj['id']).id_uri()

            elem[field] = [o['id'] if o.keys() == {'id'} else o
                           for o in elem[field]]

            if len(elem[field]) == 1:
                elem[field] = elem[field][0]

        type = as1.object_type(outer_obj)
        translate(outer_obj, 'id',
                  translate_user_id if type in as1.ACTOR_TYPES
                  else translate_object_id)

        for o in inner_objs:
            is_actor = (as1.object_type(o) in as1.ACTOR_TYPES
                        or as1.get_owner(outer_obj) == o.get('id')
                        or type in ('follow', 'stop-following'))
            translate(o, 'id', translate_user_id if is_actor else translate_object_id)
            obj_is_actor = o.get('verb') in as1.VERBS_WITH_ACTOR_OBJECT
            translate(o, 'object', translate_user_id if obj_is_actor
                      else translate_object_id)

        for o in [outer_obj] + inner_objs:
            translate(o, 'inReplyTo', translate_object_id)
            for field in 'actor', 'author', 'to', 'cc', 'bto', 'bcc':
                translate(o, field, translate_user_id)
            for tag in as1.get_objects(o, 'tags'):
                if tag.get('objectType') == 'mention':
                    translate(tag, 'url', translate_user_id, uri=True)
            for att in as1.get_objects(o, 'attachments'):
                translate(att, 'id', translate_object_id)
                url = att.get('url')
                if url and not att.get('id'):
                    if from_cls := Protocol.for_id(url):
                        att['id'] = translate_object_id(from_=from_cls, to=to_cls,
                                                        id=url)

        outer_obj = util.trim_nulls(outer_obj)

        if objs := util.get_list(outer_obj ,'object'):
            outer_obj['object'] = [o['id'] if o.keys() == {'id'} else o for o in objs]
            if len(outer_obj['object']) == 1:
                outer_obj['object'] = outer_obj['object'][0]

        return outer_obj

    @classmethod
    def receive(from_cls, obj, authed_as=None, internal=False, received_at=None):
        """Handles an incoming activity.

        If ``obj``'s key is unset, ``obj.as1``'s id field is used. If both are
        unset, returns HTTP 299.

        Args:
          obj (models.Object)
          authed_as (str): authenticated actor id who sent this activity
          internal (bool): whether to allow activity ids on internal domains,
            from opted out/blocked users, etc.
          received_at (datetime): when we first saw (received) this activity.
            Right now only used for monitoring.

        Returns:
          (str, int) tuple: (response body, HTTP status code) Flask response

        Raises:
          werkzeug.HTTPException: if the request is invalid
        """
        # check some invariants
        assert from_cls != Protocol
        assert isinstance(obj, Object), obj

        if not obj.as1:
            error('No object data provided')

        id = None
        if obj.key and obj.key.id():
            id = obj.key.id()

        if not id:
            id = obj.as1.get('id')
            obj.key = ndb.Key(Object, id)

        if not id:
            error('No id provided')
        elif from_cls.owns_id(id) is False:
            error(f'Protocol {from_cls.LABEL} does not own id {id}')
        elif from_cls.is_blocklisted(id, allow_internal=internal):
            error(f'Activity {id} is blocklisted')
        # check that this activity is public. only do this for some activities,
        # not eg likes or follows, since Mastodon doesn't currently mark those
        # as explicitly public.
        elif (obj.type in set(('post', 'update')) | as1.POST_TYPES | as1.ACTOR_TYPES
                  and not as1.is_public(obj.as1, unlisted=False)
                  and not as1.is_dm(obj.as1)):
              logger.info('Dropping non-public activity')
              return ('OK', 200)

        # lease this object, atomically
        memcache_key = activity_id_memcache_key(id)
        leased = memcache.memcache.add(memcache_key, 'leased', noreply=False,
                                     expire=5 * 60)  # 5 min
        # short circuit if we've already seen this activity id.
        # (don't do this for bare objects since we need to check further down
        # whether they've been updated since we saw them last.)
        if (obj.as1.get('objectType') == 'activity'
            and 'force' not in request.values
            and (not leased
                 or (obj.new is False and obj.changed is False))):
            error(f'Already seen this activity {id}', status=204)

        pruned = {k: v for k, v in obj.as1.items()
                  if k not in ('contentMap', 'replies', 'signature')}
        delay = ''
        if (received_at and request.headers.get('X-AppEngine-TaskRetryCount') == '0'
                and obj.type != 'delete'):  # we delay deletes for 2m
            delay_s = int((util.now().replace(tzinfo=None)
                           - received_at.replace(tzinfo=None)
                           ).total_seconds())
            delay = f'({delay_s} s behind)'
        logger.info(f'Receiving {from_cls.LABEL} {obj.type} {id} {delay} AS1: {json_dumps(pruned, indent=2)}')

        # does this protocol support this activity/object type?
        from_cls.check_supported(obj)

        # check authorization
        # https://www.w3.org/wiki/ActivityPub/Primer/Authentication_Authorization
        actor = as1.get_owner(obj.as1)
        if not actor:
            error('Activity missing actor or author')
        elif from_cls.owns_id(actor) is False:
            error(f"{from_cls.LABEL} doesn't own actor {actor}, this is probably a bridged activity. Skipping.", status=204)

        assert authed_as
        assert isinstance(authed_as, str)
        authed_as = normalize_user_id(id=authed_as, proto=from_cls)
        actor = normalize_user_id(id=actor, proto=from_cls)
        if actor != authed_as:
            report_error("Auth: receive: authed_as doesn't match owner",
                         user=f'{id} authed_as {authed_as} owner {actor}')
            error(f"actor {actor} isn't authed user {authed_as}")

        # update copy ids to originals
        obj.normalize_ids()
        obj.resolve_ids()

        if (obj.type == 'follow'
                and Protocol.for_bridgy_subdomain(as1.get_object(obj.as1).get('id'))):
            # follows of bot user; refresh user profile first
            logger.info(f'Follow of bot user, reloading {actor}')
            from_user = from_cls.get_or_create(id=actor, allow_opt_out=True)
            from_user.reload_profile()
        else:
            # load actor user
            from_user = from_cls.get_or_create(id=actor, allow_opt_out=internal)

        if not internal and (not from_user or from_user.manual_opt_out):
            error(f'Actor {actor} is manually opted out', status=204)

        # if this is an object, ie not an activity, wrap it in a create or update
        obj = from_cls.handle_bare_object(obj, authed_as=authed_as)
        obj.add('users', from_user.key)

        inner_obj_as1 = as1.get_object(obj.as1)
        inner_obj_id = inner_obj_as1.get('id')
        if obj.type in as1.CRUD_VERBS | as1.VERBS_WITH_OBJECT:
            if not inner_obj_id:
                error(f'{obj.type} object has no id!')

        # check age. we support backdated posts, but if they're over 2w old, we
        # don't deliver them
        if obj.type == 'post':
            if published := inner_obj_as1.get('published'):
                try:
                    published_dt = util.parse_iso8601(published)
                    if not published_dt.tzinfo:
                        published_dt = published_dt.replace(tzinfo=timezone.utc)
                    age = util.now() - published_dt
                    if age > CREATE_MAX_AGE:
                        error(f'Ignoring, too old, {age} is over {CREATE_MAX_AGE}',
                              status=204)
                except ValueError:  # from parse_iso8601
                    logger.debug(f"Couldn't parse published {published}")

        # write Object to datastore
        obj.source_protocol = from_cls.LABEL
        if obj.type in STORE_AS1_TYPES:
            obj.put()

        # store inner object
        # TODO: unify with big obj.type conditional below. would have to merge
        # this with the DM handling block lower down.