libervia-backend: libervia/backend/plugins/plugin_comp_ap_gateway/__init_

comparison libervia/backend/plugins/plugin_comp_ap_gateway/init.py @ 4071:4b842c1fb686

refactoring: renamed `sat` package to `libervia.backend`

author	Goffi <goffi@goffi.org>
date	Fri, 02 Jun 2023 11:49:51 +0200
parents	sat/plugins/plugin_comp_ap_gateway/__init__.py@c23cad65ae99
children	c3b68fdc2de7

comparison

equal deleted inserted replaced

-:d10748475025
+:4b842c1fb686
+#!/usr/bin/env python3
+# Libervia ActivityPub Gateway
+# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import base64
+import calendar
+import hashlib
+import json
+from pathlib import Path
+from pprint import pformat
+import re
+from typing import (
+Any,
+Awaitable,
+Callable,
+Dict,
+List,
+Optional,
+Set,
+Tuple,
+Union,
+overload,
+)
+from urllib import parse
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.asymmetric import padding
+import dateutil
+from dateutil.parser import parserinfo
+import shortuuid
+from sqlalchemy.exc import IntegrityError
+import treq
+from treq.response import _Response as TReqResponse
+from twisted.internet import defer, reactor, threads
+from twisted.web import http
+from twisted.words.protocols.jabber import error, jid
+from twisted.words.xish import domish
+from wokkel import pubsub, rsm
+from libervia.backend.core import exceptions
+from libervia.backend.core.constants import Const as C
+from libervia.backend.core.core_types import SatXMPPEntity
+from libervia.backend.core.i18n import _
+from libervia.backend.core.log import getLogger
+from libervia.backend.memory import persistent
+from libervia.backend.memory.sqla_mapping import History, SubscriptionState
+from libervia.backend.tools import utils
+from libervia.backend.tools.common import data_format, date_utils, tls, uri
+from libervia.backend.tools.common.async_utils import async_lru
+from .ad_hoc import APAdHocService
+from .events import APEvents
+from .constants import (
+ACTIVITY_OBJECT_MANDATORY,
+ACTIVITY_TARGET_MANDATORY,
+ACTIVITY_TYPES,
+ACTIVITY_TYPES_LOWER,
+COMMENTS_MAX_PARENTS,
+CONF_SECTION,
+IMPORT_NAME,
+LRU_MAX_SIZE,
+MEDIA_TYPE_AP,
+NS_AP,
+NS_AP_PUBLIC,
+PUBLIC_TUPLE,
+TYPE_ACTOR,
+TYPE_EVENT,
+TYPE_FOLLOWERS,
+TYPE_ITEM,
+TYPE_LIKE,
+TYPE_MENTION,
+TYPE_REACTION,
+TYPE_TOMBSTONE,
+TYPE_JOIN,
+TYPE_LEAVE
+)
+from .http_server import HTTPServer
+from .pubsub_service import APPubsubService
+from .regex import RE_MENTION
+log = getLogger(__name__)
+IMPORT_NAME = "ap-gateway"
+PLUGIN_INFO = {
+C.PI_NAME: "ActivityPub Gateway component",
+C.PI_IMPORT_NAME: IMPORT_NAME,
+C.PI_MODES: [C.PLUG_MODE_COMPONENT],
+C.PI_TYPE: C.PLUG_TYPE_ENTRY_POINT,
+C.PI_PROTOCOLS: [],
+C.PI_DEPENDENCIES: [
+"XEP-0050", "XEP-0054", "XEP-0060", "XEP-0084", "XEP-0106", "XEP-0277",
+"XEP-0292", "XEP-0329", "XEP-0372", "XEP-0424", "XEP-0465", "XEP-0470",
+"XEP-0447", "XEP-0471", "PUBSUB_CACHE", "TEXT_SYNTAXES", "IDENTITY"
+],
+C.PI_RECOMMENDATIONS: [],
+C.PI_MAIN: "APGateway",
+C.PI_HANDLER: C.BOOL_TRUE,
+C.PI_DESCRIPTION: _(
+"Gateway for bidirectional communication between XMPP and ActivityPub."
+),
+}
+HEXA_ENC = r"(?P<hex>[0-9a-fA-f]{2})"
+RE_PERIOD_ENC = re.compile(f"\\.{HEXA_ENC}")
+RE_PERCENT_ENC = re.compile(f"%{HEXA_ENC}")
+RE_ALLOWED_UNQUOTED = re.compile(r"^[a-zA-Z0-9_-]+$")
+class APGateway:
+IMPORT_NAME = IMPORT_NAME
+# show data send or received through HTTP, used for debugging
+# 1: log POST objects
+# 2: log POST and GET objects
+# 3: log POST and GET objects with HTTP headers for GET requests
+verbose = 0
+def __init__(self, host):
+self.host = host
+self.initialised = False
+self.client = None
+self._p = host.plugins["XEP-0060"]
+self._a = host.plugins["XEP-0084"]
+self._e = host.plugins["XEP-0106"]
+self._m = host.plugins["XEP-0277"]
+self._v = host.plugins["XEP-0292"]
+self._refs = host.plugins["XEP-0372"]
+self._r = host.plugins["XEP-0424"]
+self._sfs = host.plugins["XEP-0447"]
+self._pps = host.plugins["XEP-0465"]
+self._pa = host.plugins["XEP-0470"]
+self._c = host.plugins["PUBSUB_CACHE"]
+self._t = host.plugins["TEXT_SYNTAXES"]
+self._i = host.plugins["IDENTITY"]
+self._events = host.plugins["XEP-0471"]
+self._p.add_managed_node(
+"",
+items_cb=self._items_received,
+# we want to be sure that the callbacks are launched before pubsub cache's
+# one, as we need to inspect items before they are actually removed from cache
+# or updated
+priority=1000
+)
+self.pubsub_service = APPubsubService(self)
+self.ad_hoc = APAdHocService(self)
+self.ap_events = APEvents(self)
+host.trigger.add("message_received", self._message_received_trigger, priority=-1000)
+host.trigger.add("XEP-0424_retractReceived", self._on_message_retract)
+host.trigger.add("XEP-0372_ref_received", self._on_reference_received)
+host.bridge.add_method(
+"ap_send",
+".plugin",
+in_sign="sss",
+out_sign="",
+method=self._publish_message,
+async_=True,
+)
+def get_handler(self, __):
+return self.pubsub_service
+async def init(self, client):
+if self.initialised:
+return
+self.initialised = True
+log.info(_("ActivityPub Gateway initialization"))
+# RSA keys
+stored_data = await self.host.memory.storage.get_privates(
+IMPORT_NAME, ["rsa_key"], profile=client.profile
+)
+private_key_pem = stored_data.get("rsa_key")
+if private_key_pem is None:
+self.private_key = await threads.deferToThread(
+rsa.generate_private_key,
+public_exponent=65537,
+key_size=4096,
+)
+private_key_pem = self.private_key.private_bytes(
+encoding=serialization.Encoding.PEM,
+format=serialization.PrivateFormat.PKCS8,
+encryption_algorithm=serialization.NoEncryption()
+).decode()
+await self.host.memory.storage.set_private_value(
+IMPORT_NAME, "rsa_key", private_key_pem, profile=client.profile
+)
+else:
+self.private_key = serialization.load_pem_private_key(
+private_key_pem.encode(),
+password=None,
+)
+self.public_key = self.private_key.public_key()
+self.public_key_pem = self.public_key.public_bytes(
+encoding=serialization.Encoding.PEM,
+format=serialization.PublicFormat.SubjectPublicKeyInfo
+).decode()
+# params
+# URL and port
+self.public_url = self.host.memory.config_get(
+CONF_SECTION, "public_url"
+) or self.host.memory.config_get(
+CONF_SECTION, "xmpp_domain"
+)
+if self.public_url is None:
+log.error(
+'"public_url" not set in configuration, this is mandatory to have'
+"ActivityPub Gateway running. Please set this option it to public facing "
+f"url in {CONF_SECTION!r} configuration section."
+)
+return
+if parse.urlparse(self.public_url).scheme:
+log.error(
+"Scheme must not be specified in \"public_url\", please remove it from "
+"\"public_url\" configuration option. ActivityPub Gateway won't be run."
+)
+return
+self.http_port = int(self.host.memory.config_get(
+CONF_SECTION, 'http_port', 8123))
+connection_type = self.host.memory.config_get(
+CONF_SECTION, 'http_connection_type', 'https')
+if connection_type not in ('http', 'https'):
+raise exceptions.ConfigError(
+'bad ap-gateay http_connection_type, you must use one of "http" or '
+'"https"'
+)
+self.max_items = int(self.host.memory.config_get(
+CONF_SECTION, 'new_node_max_items', 50
+))
+self.comments_max_depth = int(self.host.memory.config_get(
+CONF_SECTION, 'comments_max_depth', 0
+))
+self.ap_path = self.host.memory.config_get(CONF_SECTION, 'ap_path', '_ap')
+self.base_ap_url = parse.urljoin(f"https://{self.public_url}", f"{self.ap_path}/")
+# True (default) if we provide gateway only to entities/services from our server
+self.local_only = C.bool(
+self.host.memory.config_get(CONF_SECTION, 'local_only', C.BOOL_TRUE)
+)
+# if True (default), mention will be parsed in non-private content coming from
+# XMPP. This is necessary as XEP-0372 are coming separately from item where the
+# mention is done, which is hard to impossible to translate to ActivityPub (where
+# mention specified inside the item directly). See documentation for details.
+self.auto_mentions = C.bool(
+self.host.memory.config_get(CONF_SECTION, "auto_mentions", C.BOOL_TRUE)
+)
+html_redirect: Dict[str, Union[str, dict]] = self.host.memory.config_get(
+CONF_SECTION, 'html_redirect_dict', {}
+)
+self.html_redirect: Dict[str, List[dict]] = {}
+for url_type, target in html_redirect.items():
+if isinstance(target, str):
+target = {"url": target}
+elif not isinstance(target, dict):
+raise exceptions.ConfigError(
+f"html_redirect target must be a URL or a dict, not {target!r}"
+)
+filters = target.setdefault("filters", {})
+if "url" not in target:
+log.warning(f"invalid HTML redirection, missing target URL: {target}")
+continue
+# a slash in the url_type is a syntactic shortcut to have a node filter
+if "/" in url_type:
+url_type, node_filter = url_type.split("/", 1)
+filters["node"] = node_filter
+self.html_redirect.setdefault(url_type, []).append(target)
+# HTTP server launch
+self.server = HTTPServer(self)
+if connection_type == 'http':
+reactor.listenTCP(self.http_port, self.server)
+else:
+options = tls.get_options_from_config(
+self.host.memory.config, CONF_SECTION)
+tls.tls_options_check(options)
+context_factory = tls.get_tls_context_factory(options)
+reactor.listenSSL(self.http_port, self.server, context_factory)
+async def profile_connecting(self, client):
+self.client = client
+client.sendHistory = True
+client._ap_storage = persistent.LazyPersistentBinaryDict(
+IMPORT_NAME,
+client.profile
+)
+await self.init(client)
+def profile_connected(self, client):
+self.ad_hoc.init(client)
+async def _items_received(
+self,
+client: SatXMPPEntity,
+itemsEvent: pubsub.ItemsEvent
+) -> None:
+"""Callback called when pubsub items are received
+if the items are adressed to a JID corresponding to an AP actor, they are
+converted to AP items and sent to the corresponding AP server.
+If comments nodes are linked, they are automatically subscribed to get new items
+from there too.
+"""
+if client != self.client:
+return
+# we need recipient as JID and not gateway own JID to be able to use methods such
+# as "subscribe"
+client = self.client.get_virtual_client(itemsEvent.sender)
+recipient = itemsEvent.recipient
+if not recipient.user:
+log.debug("ignoring items event without local part specified")
+return
+ap_account = self._e.unescape(recipient.user)
+if self._pa.is_attachment_node(itemsEvent.nodeIdentifier):
+await self.convert_and_post_attachments(
+client, ap_account, itemsEvent.sender, itemsEvent.nodeIdentifier,
+itemsEvent.items
+)
+else:
+await self.convert_and_post_items(
+client, ap_account, itemsEvent.sender, itemsEvent.nodeIdentifier,
+itemsEvent.items
+)
+async def get_virtual_client(self, actor_id: str) -> SatXMPPEntity:
+"""Get client for this component with a specified jid
+This is needed to perform operations with the virtual JID corresponding to the AP
+actor instead of the JID of the gateway itself.
+@param actor_id: ID of the actor
+@return: virtual client
+"""
+local_jid = await self.get_jid_from_id(actor_id)
+return self.client.get_virtual_client(local_jid)
+def is_activity(self, data: dict) -> bool:
+"""Return True if the data has an activity type"""
+try:
+return (data.get("type") or "").lower() in ACTIVITY_TYPES_LOWER
+except (KeyError, TypeError):
+return False
+async def ap_get(self, url: str) -> dict:
+"""Retrieve AP JSON from given URL
+@raise error.StanzaError: "service-unavailable" is sent when something went wrong
+with AP server
+"""
+resp = await treq.get(
+url,
+headers = {
+"Accept": [MEDIA_TYPE_AP],
+"Content-Type": [MEDIA_TYPE_AP],
+}
+)
+if resp.code >= 300:
+text = await resp.text()
+if resp.code == 404:
+raise exceptions.NotFound(f"Can't find resource at {url}")
+else:
+msg = f"HTTP error {resp.code} (url: {url}): {text}"
+raise exceptions.ExternalRequestError(msg)
+try:
+return await treq.json_content(resp)
+except Exception as e:
+raise error.StanzaError(
+"service-unavailable",
+text=f"Can't get AP data at {url}: {e}"
+)
+@overload
+async def ap_get_object(self, data: dict, key: str) -> Optional[dict]:
+...
+@overload
+async def ap_get_object(
+self, data: Union[str, dict], key: None = None
+) -> dict:
+...
+async def ap_get_object(self, data, key = None):
+"""Retrieve an AP object, dereferencing when necessary
+This method is to be used with attributes marked as "Functional" in
+https://www.w3.org/TR/activitystreams-vocabulary
+@param data: AP object where an other object is looked for, or the object itself
+@param key: name of the object to look for, or None if data is the object directly
+@return: found object if any
+"""
+if key is not None:
+value = data.get(key)
+else:
+value = data
+if value is None:
+if key is None:
+raise ValueError("None can't be used with ap_get_object is key is None")
+return None
+elif isinstance(value, dict):
+return value
+elif isinstance(value, str):
+if self.is_local_url(value):
+return await self.ap_get_local_object(value)
+else:
+return await self.ap_get(value)
+else:
+raise NotImplementedError(
+"was expecting a string or a dict, got {type(value)}: {value!r}}"
+)
+async def ap_get_local_object(
+self,
+url: str
+) -> dict:
+"""Retrieve or generate local object
+for now, only handle XMPP items to convert to AP
+"""
+url_type, url_args = self.parse_apurl(url)
+if url_type == TYPE_ITEM:
+try:
+account, item_id = url_args
+except ValueError:
+raise ValueError(f"invalid URL: {url}")
+author_jid, node = await self.get_jid_and_node(account)
+if node is None:
+node = self._m.namespace
+cached_node = await self.host.memory.storage.get_pubsub_node(
+self.client, author_jid, node
+)
+if not cached_node:
+log.debug(f"node {node!r} at {author_jid} is not found in cache")
+found_item = None
+else:
+cached_items, __ = await self.host.memory.storage.get_items(
+cached_node, item_ids=[item_id]
+)
+if not cached_items:
+log.debug(
+f"item {item_id!r} of {node!r} at {author_jid} is not found in "
+"cache"
+)
+found_item = None
+else:
+found_item = cached_items[0].data
+if found_item is None:
+# the node is not in cache, we have to make a request to retrieve the item
+# If the node doesn't exist, get_items will raise a NotFound exception
+found_items, __ = await self._p.get_items(
+self.client, author_jid, node, item_ids=[item_id]
+)
+try:
+found_item = found_items[0]
+except IndexError:
+raise exceptions.NotFound(f"requested item at {url} can't be found")
+if node.startswith(self._events.namespace):
+# this is an event
+event_data = self._events.event_elt_2_event_data(found_item)
+ap_item = await self.ap_events.event_data_2_ap_item(
+event_data, author_jid
+)
+# the URL must return the object and not the activity
+ap_item["object"]["@context"] = ap_item["@context"]
+return ap_item["object"]
+else:
+# this is a blog item
+mb_data = await self._m.item_2_mb_data(
+self.client, found_item, author_jid, node
+)
+ap_item = await self.mb_data_2_ap_item(self.client, mb_data)
+# the URL must return the object and not the activity
+return ap_item["object"]
+else:
+raise NotImplementedError(
+'only object from "item" URLs can be retrieved for now'
+)
+async def ap_get_list(
+self,
+data: dict,
+key: str,
+only_ids: bool = False
+) -> Optional[List[Dict[str, Any]]]:
+"""Retrieve a list of objects from AP data, dereferencing when necessary
+This method is to be used with non functional vocabularies. Use ``ap_get_object``
+otherwise.
+If the value is a dictionary, it will be wrapped in a list
+@param data: AP object where a list of objects is looked for
+@param key: key of the list to look for
+@param only_ids: if Trye, only items IDs are retrieved
+@return: list of objects, or None if the key is not present
+"""
+value = data.get(key)
+if value is None:
+return None
+elif isinstance(value, str):
+if self.is_local_url(value):
+value = await self.ap_get_local_object(value)
+else:
+value = await self.ap_get(value)
+if isinstance(value, dict):
+return [value]
+if not isinstance(value, list):
+raise ValueError(f"A list was expected, got {type(value)}: {value!r}")
+if only_ids:
+return [
+{"id": v["id"]} if isinstance(v, dict) else {"id": v}
+for v in value
+]
+else:
+return [await self.ap_get_object(i) for i in value]
+async def ap_get_actors(
+self,
+data: dict,
+key: str,
+as_account: bool = True
+) -> List[str]:
+"""Retrieve AP actors from data
+@param data: AP object containing a field with actors
+@param key: field to use to retrieve actors
+@param as_account: if True returns account handles, otherwise will return actor
+IDs
+@raise exceptions.DataError: there is not actor data or it is invalid
+"""
+value = data.get(key)
+if value is None:
+raise exceptions.DataError(
+f"no actor associated to object {data.get('id')!r}"
+)
+elif isinstance(value, dict):
+actor_id = value.get("id")
+if actor_id is None:
+raise exceptions.DataError(
+f"invalid actor associated to object {data.get('id')!r}: {value!r}"
+)
+value = [actor_id]
+elif isinstance(value, str):
+value = [value]
+elif isinstance(value, list):
+try:
+value = [a if isinstance(a, str) else a["id"] for a in value]
+except (TypeError, KeyError):
+raise exceptions.DataError(
+f"invalid actors list to object {data.get('id')!r}: {value!r}"
+)
+if not value:
+raise exceptions.DataError(
+f"list of actors is empty"
+)
+if as_account:
+return [await self.get_ap_account_from_id(actor_id) for actor_id in value]
+else:
+return value
+async def ap_get_sender_actor(
+self,
+data: dict,
+) -> str:
+"""Retrieve actor who sent data
+This is done by checking "actor" field first, then "attributedTo" field.
+Only the first found actor is taken into account
+@param data: AP object
+@return: actor id of the sender
+@raise exceptions.NotFound: no actor has been found in data
+"""
+try:
+actors = await self.ap_get_actors(data, "actor", as_account=False)
+except exceptions.DataError:
+actors = None
+if not actors:
+try:
+actors = await self.ap_get_actors(data, "attributedTo", as_account=False)
+except exceptions.DataError:
+raise exceptions.NotFound(
+'actor not specified in "actor" or "attributedTo"'
+)
+try:
+return actors[0]
+except IndexError:
+raise exceptions.NotFound("list of actors is empty")
+def must_encode(self, text: str) -> bool:
+"""Indicate if a text must be period encoded"""
+return (
+not RE_ALLOWED_UNQUOTED.match(text)
+or text.startswith("___")
+or "---" in text
+)
+def period_encode(self, text: str) -> str:
+"""Period encode a text
+see [get_jid_and_node] for reasons of period encoding
+"""
+return (
+parse.quote(text, safe="")
+.replace("---", "%2d%2d%2d")
+.replace("___", "%5f%5f%5f")
+.replace(".", "%2e")
+.replace("~", "%7e")
+.replace("%", ".")
+)
+async def get_ap_account_from_jid_and_node(
+self,
+jid_: jid.JID,
+node: Optional[str]
+) -> str:
+"""Construct AP account from JID and node
+The account construction will use escaping when necessary
+"""
+if not node or node == self._m.namespace:
+node = None
+if self.client is None:
+raise exceptions.InternalError("Client is not set yet")
+if self.is_virtual_jid(jid_):
+# this is an proxy JID to an AP Actor
+return self._e.unescape(jid_.user)
+if node and not jid_.user and not self.must_encode(node):
+is_pubsub = await self.is_pubsub(jid_)
+# when we have a pubsub service, the user part can be used to set the node
+# this produces more user-friendly AP accounts
+if is_pubsub:
+jid_.user = node
+node = None
+is_local = self.is_local(jid_)
+user = jid_.user if is_local else jid_.userhost()
+if user is None:
+user = ""
+account_elts = []
+if node and self.must_encode(node) or self.must_encode(user):
+account_elts = ["___"]
+if node:
+node = self.period_encode(node)
+user = self.period_encode(user)
+if not user:
+raise exceptions.InternalError("there should be a user part")
+if node:
+account_elts.extend((node, "---"))
+account_elts.extend((
+user, "@", jid_.host if is_local else self.client.jid.userhost()
+))
+return "".join(account_elts)
+def is_local(self, jid_: jid.JID) -> bool:
+"""Returns True if jid_ use a domain or subdomain of gateway's host"""
+local_host = self.client.host.split(".")
+assert local_host
+return jid_.host.split(".")[-len(local_host):] == local_host
+async def is_pubsub(self, jid_: jid.JID) -> bool:
+"""Indicate if a JID is a Pubsub service"""
+host_disco = await self.host.get_disco_infos(self.client, jid_)
+return (
+("pubsub", "service") in host_disco.identities
+and not ("pubsub", "pep") in host_disco.identities
+)
+async def get_jid_and_node(self, ap_account: str) -> Tuple[jid.JID, Optional[str]]:
+"""Decode raw AP account handle to get XMPP JID and Pubsub Node
+Username are case insensitive.
+By default, the username correspond to local username (i.e. username from
+component's server).
+If local name's domain is a pubsub service (and not PEP), the username is taken as
+a pubsub node.
+If ``---`` is present in username, the part before is used as pubsub node, and the
+rest as a JID user part.
+If username starts with ``___``, characters are encoded using period encoding
+(i.e. percent encoding where a ``.`` is used instead of ``%``).
+This horror is necessary due to limitation in some AP implementation (notably
+Mastodon), cf. https://github.com/mastodon/mastodon/issues/17222
+examples:
+``toto@example.org`` => JID = toto@example.org, node = None
+``___toto.40example.net@example.org`` => JID = toto@example.net (this one is a
+non-local JID, and will work only if setings ``local_only`` is False), node = None
+``toto@pubsub.example.org`` (with pubsub.example.org being a pubsub service) =>
+JID = pubsub.example.org, node = toto
+``tata---toto@example.org`` => JID = toto@example.org, node = tata
+``___urn.3axmpp.3amicroblog.3a0@pubsub.example.org`` (with pubsub.example.org
+being a pubsub service) ==> JID = pubsub.example.org, node = urn:xmpp:microblog:0
+@param ap_account: ActivityPub account handle (``username@domain.tld``)
+@return: service JID and pubsub node
+if pubsub node is None, default microblog pubsub node (and possibly other
+nodes that plugins may hanlde) will be used
+@raise ValueError: invalid account
+@raise PermissionError: non local jid is used when gateway doesn't allow them
+"""
+if ap_account.count("@") != 1:
+raise ValueError("Invalid AP account")
+if ap_account.startswith("___"):
+encoded = True
+ap_account = ap_account[3:]
+else:
+encoded = False
+username, domain = ap_account.split("@")
+if "---" in username:
+node, username = username.rsplit("---", 1)
+else:
+node = None
+if encoded:
+username = parse.unquote(
+RE_PERIOD_ENC.sub(r"%\g<hex>", username),
+errors="strict"
+)
+if node:
+node = parse.unquote(
+RE_PERIOD_ENC.sub(r"%\g<hex>", node),
+errors="strict"
+)
+if "@" in username:
+username, domain = username.rsplit("@", 1)
+if not node:
+# we need to check host disco, because disco request to user may be
+# blocked for privacy reason (see
+# https://xmpp.org/extensions/xep-0030.html#security)
+is_pubsub = await self.is_pubsub(jid.JID(domain))
+if is_pubsub:
+# if the host is a pubsub service and not a PEP, we consider that username
+# is in fact the node name
+node = username
+username = None
+jid_s = f"{username}@{domain}" if username else domain
+try:
+jid_ = jid.JID(jid_s)
+except RuntimeError:
+raise ValueError(f"Invalid jid: {jid_s!r}")
+if self.local_only and not self.is_local(jid_):
+raise exceptions.PermissionError(
+"This gateway is configured to map only local entities and services"
+)
+return jid_, node
+def get_local_jid_from_account(self, account: str) -> jid.JID:
+"""Compute JID linking to an AP account
+The local jid is computer by escaping AP actor handle and using it as local part
+of JID, where domain part is this gateway own JID
+"""
+return jid.JID(
+None,
+(
+self._e.escape(account),
+self.client.jid.host,
+None
+)
+)
+async def get_jid_from_id(self, actor_id: str) -> jid.JID:
+"""Compute JID linking to an AP Actor ID
+The local jid is computer by escaping AP actor handle and using it as local part
+of JID, where domain part is this gateway own JID
+If the actor_id comes from local server (checked with self.public_url), it means
+that we have an XMPP entity, and the original JID is returned
+"""
+if self.is_local_url(actor_id):
+request_type, extra_args = self.parse_apurl(actor_id)
+if request_type != TYPE_ACTOR or len(extra_args) != 1:
+raise ValueError(f"invalid actor id: {actor_id!r}")
+actor_jid, __ = await self.get_jid_and_node(extra_args[0])
+return actor_jid
+account = await self.get_ap_account_from_id(actor_id)
+return self.get_local_jid_from_account(account)
+def parse_apurl(self, url: str) -> Tuple[str, List[str]]:
+"""Parse an URL leading to an AP endpoint
+@param url: URL to parse (schema is not mandatory)
+@return: endpoint type and extra arguments
+"""
+path = parse.urlparse(url).path.lstrip("/")
+type_, *extra_args = path[len(self.ap_path):].lstrip("/").split("/")
+return type_, [parse.unquote(a) for a in extra_args]
+def build_apurl(self, type_:str , *args: str) -> str:
+"""Build an AP endpoint URL
+@param type_: type of AP endpoing
+@param arg: endpoint dependant arguments
+"""
+return parse.urljoin(
+self.base_ap_url,
+str(Path(type_).joinpath(*(parse.quote_plus(a, safe="@") for a in args)))
+)
+def is_local_url(self, url: str) -> bool:
+"""Tells if an URL link to this component
+``public_url`` and ``ap_path`` are used to check the URL
+"""
+return url.startswith(self.base_ap_url)
+def is_virtual_jid(self, jid_: jid.JID) -> bool:
+"""Tell if a JID is an AP actor mapped through this gateway"""
+return jid_.host == self.client.jid.userhost()
+def build_signature_header(self, values: Dict[str, str]) -> str:
+"""Build key="<value>" signature header from signature data"""
+fields = []
+for key, value in values.items():
+if key not in ("(created)", "(expired)"):
+if '"' in value:
+raise NotImplementedError(
+"string escaping is not implemented, double-quote can't be used "
+f"in {value!r}"
+)
+value = f'"{value}"'
+fields.append(f"{key}={value}")
+return ",".join(fields)
+def get_digest(self, body: bytes, algo="SHA-256") -> Tuple[str, str]:
+"""Get digest data to use in header and signature
+@param body: body of the request
+@return: hash name and digest
+"""
+if algo != "SHA-256":
+raise NotImplementedError("only SHA-256 is implemented for now")
+return algo, base64.b64encode(hashlib.sha256(body).digest()).decode()
+@async_lru(maxsize=LRU_MAX_SIZE)
+async def get_actor_data(self, actor_id) -> dict:
+"""Retrieve actor data with LRU cache"""
+return await self.ap_get(actor_id)
+@async_lru(maxsize=LRU_MAX_SIZE)
+async def get_actor_pub_key_data(
+self,
+actor_id: str
+) -> Tuple[str, str, rsa.RSAPublicKey]:
+"""Retrieve Public Key data from actor ID
+@param actor_id: actor ID (url)
+@return: key_id, owner and public_key
+@raise KeyError: publicKey is missing from actor data
+"""
+actor_data = await self.get_actor_data(actor_id)
+pub_key_data = actor_data["publicKey"]
+key_id = pub_key_data["id"]
+owner = pub_key_data["owner"]
+pub_key_pem = pub_key_data["publicKeyPem"]
+pub_key = serialization.load_pem_public_key(pub_key_pem.encode())
+return key_id, owner, pub_key
+def create_activity(
+self,
+activity: str,
+actor_id: str,
+object_: Optional[Union[str, dict]] = None,
+target: Optional[Union[str, dict]] = None,
+activity_id: Optional[str] = None,
+**kwargs,
+) -> Dict[str, Any]:
+"""Generate base data for an activity
+@param activity: one of ACTIVITY_TYPES
+@param actor_id: AP actor ID of the sender
+@param object_: content of "object" field
+@param target: content of "target" field
+@param activity_id: ID to use for the activity
+if not set it will be automatically generated, but it is usually desirable to
+set the ID manually so it can be retrieved (e.g. for Undo)
+"""
+if activity not in ACTIVITY_TYPES:
+raise exceptions.InternalError(f"invalid activity: {activity!r}")
+if object_ is None and activity in ACTIVITY_OBJECT_MANDATORY:
+raise exceptions.InternalError(
+f'"object_" is mandatory for activity {activity!r}'
+)
+if target is None and activity in ACTIVITY_TARGET_MANDATORY:
+raise exceptions.InternalError(
+f'"target" is mandatory for activity {activity!r}'
+)
+if activity_id is None:
+activity_id = f"{actor_id}#{activity.lower()}_{shortuuid.uuid()}"
+data: Dict[str, Any] = {
+"@context": [NS_AP],
+"actor": actor_id,
+"id": activity_id,
+"type": activity,
+}
+data.update(kwargs)
+if object_ is not None:
+data["object"] = object_
+if target is not None:
+data["target"] = target
+return data
+def get_key_id(self, actor_id: str) -> str:
+"""Get local key ID from actor ID"""
+return f"{actor_id}#main-key"
+async def check_signature(
+self,
+signature: str,
+key_id: str,
+headers: Dict[str, str]
+) -> str:
+"""Verify that signature matches given headers
+see https://datatracker.ietf.org/doc/html/draft-cavage-http-signatures-06#section-3.1.2
+@param signature: Base64 encoded signature
+@param key_id: ID of the key used to sign the data
+@param headers: headers and their values, including pseudo-headers
+@return: id of the signing actor
+@raise InvalidSignature: signature doesn't match headers
+"""
+to_sign = "\n".join(f"{k.lower()}: {v}" for k,v in headers.items())
+if key_id.startswith("acct:"):
+actor = key_id[5:]
+actor_id = await self.get_ap_actor_id_from_account(actor)
+else:
+actor_id = key_id.split("#", 1)[0]
+pub_key_id, pub_key_owner, pub_key = await self.get_actor_pub_key_data(actor_id)
+if pub_key_id != key_id or pub_key_owner != actor_id:
+raise exceptions.EncryptionError("Public Key mismatch")
+try:
+pub_key.verify(
+base64.b64decode(signature),
+to_sign.encode(),
+# we have to use PKCS1v15 padding to be compatible with Mastodon
+padding.PKCS1v15(),  # type: ignore
+hashes.SHA256()  # type: ignore
+)
+except InvalidSignature:
+raise exceptions.EncryptionError(
+"Invalid signature (using PKC0S1 v1.5 and SHA-256)"
+)
+return actor_id
+def get_signature_data(
+self,
+key_id: str,
+headers: Dict[str, str]
+) -> Tuple[Dict[str, str], Dict[str, str]]:
+"""Generate and return signature and corresponding headers
+@param parsed_url: URL where the request is sent/has been received
+@param key_id: ID of the key (URL linking to the data with public key)
+@param date: HTTP datetime string of signature generation
+@param body: body of the HTTP request
+@param headers: headers to sign and their value:
+default value will be used if not specified
+@return: headers and signature data
+``headers`` is an updated copy of ``headers`` arguments, with pseudo-headers
+removed, and ``Signature`` added.
+"""
+# headers must be lower case
+l_headers: Dict[str, str] = {k.lower(): v for k, v in headers.items()}
+to_sign = "\n".join(f"{k}: {v}" for k,v in l_headers.items())
+signature = base64.b64encode(self.private_key.sign(
+to_sign.encode(),
+# we have to use PKCS1v15 padding to be compatible with Mastodon
+padding.PKCS1v15(),  # type: ignore
+hashes.SHA256()  # type: ignore
+)).decode()
+sign_data = {
+"keyId": key_id,
+"Algorithm": "rsa-sha256",
+"headers": " ".join(l_headers.keys()),
+"signature": signature
+}
+new_headers = {k: v for k,v in headers.items() if not k.startswith("(")}
+new_headers["Signature"] = self.build_signature_header(sign_data)
+return new_headers, sign_data
+async def convert_and_post_items(
+self,
+client: SatXMPPEntity,
+ap_account: str,
+service: jid.JID,
+node: str,
+items: List[domish.Element],
+subscribe_extra_nodes: bool = True,
+) -> None:
+"""Convert XMPP items to AP items and post them to actor inbox
+@param ap_account: account of ActivityPub actor receiving the item
+@param service: JID of the (virtual) pubsub service where the item has been
+published
+@param node: (virtual) node corresponding where the item has been published
+@param subscribe_extra_nodes: if True, extra data nodes will be automatically
+subscribed, that is comment nodes if present and attachments nodes.
+"""
+actor_id = await self.get_ap_actor_id_from_account(ap_account)
+inbox = await self.get_ap_inbox_from_id(actor_id)
+for item in items:
+if item.name == "item":
+cached_item = await self.host.memory.storage.search_pubsub_items({
+"profiles": [self.client.profile],
+"services": [service],
+"nodes": [node],
+"names": [item["id"]]
+})
+is_new = not bool(cached_item)
+if node.startswith(self._events.namespace):
+# event item
+event_data = self._events.event_elt_2_event_data(item)
+try:
+author_jid = jid.JID(item["publisher"]).userhostJID()
+except (KeyError, RuntimeWarning):
+root_elt = item
+while root_elt.parent is not None:
+root_elt = root_elt.parent
+author_jid = jid.JID(root_elt["from"]).userhostJID()
+if subscribe_extra_nodes and not self.is_virtual_jid(author_jid):
+# we subscribe automatically to comment nodes if any
+recipient_jid = self.get_local_jid_from_account(ap_account)
+recipient_client = self.client.get_virtual_client(recipient_jid)
+comments_data = event_data.get("comments")
+if comments_data:
+comment_service = jid.JID(comments_data["jid"])
+comment_node = comments_data["node"]
+await self._p.subscribe(
+recipient_client, comment_service, comment_node
+)
+try:
+await self._pa.subscribe(
+recipient_client, service, node, event_data["id"]
+)
+except exceptions.NotFound:
+log.debug(
+f"no attachment node found for item {event_data['id']!r} "
+f"on {node!r} at {service}"
+)
+ap_item = await self.ap_events.event_data_2_ap_item(
+event_data, author_jid, is_new=is_new
+)
+else:
+# blog item
+mb_data = await self._m.item_2_mb_data(client, item, service, node)
+author_jid = jid.JID(mb_data["author_jid"])
+if subscribe_extra_nodes and not self.is_virtual_jid(author_jid):
+# we subscribe automatically to comment nodes if any
+recipient_jid = self.get_local_jid_from_account(ap_account)
+recipient_client = self.client.get_virtual_client(recipient_jid)
+for comment_data in mb_data.get("comments", []):
+comment_service = jid.JID(comment_data["service"])
+if self.is_virtual_jid(comment_service):
+log.debug(
+f"ignoring virtual comment service: {comment_data}"
+)
+continue
+comment_node = comment_data["node"]
+await self._p.subscribe(
+recipient_client, comment_service, comment_node
+)
+try:
+await self._pa.subscribe(
+recipient_client, service, node, mb_data["id"]
+)
+except exceptions.NotFound:
+log.debug(
+f"no attachment node found for item {mb_data['id']!r} on "
+f"{node!r} at {service}"
+)
+ap_item = await self.mb_data_2_ap_item(client, mb_data, is_new=is_new)
+url_actor = ap_item["actor"]
+elif item.name == "retract":
+url_actor, ap_item = await self.ap_delete_item(
+client.jid, node, item["id"]
+)
+else:
+raise exceptions.InternalError(f"unexpected element: {item.toXml()}")
+await self.sign_and_post(inbox, url_actor, ap_item)
+async def convert_and_post_attachments(
+self,
+client: SatXMPPEntity,
+ap_account: str,
+service: jid.JID,
+node: str,
+items: List[domish.Element],
+publisher: Optional[jid.JID] = None
+) -> None:
+"""Convert XMPP item attachments to AP activities and post them to actor inbox
+@param ap_account: account of ActivityPub actor receiving the item
+@param service: JID of the (virtual) pubsub service where the item has been
+published
+@param node: (virtual) node corresponding where the item has been published
+subscribed, that is comment nodes if present and attachments nodes.
+@param items: attachments items
+@param publisher: publisher of the attachments item (it's NOT the PEP/Pubsub
+service, it's the publisher of the item). To be filled only when the publisher
+is known for sure, otherwise publisher will be determined either if
+"publisher" attribute is set by pubsub service, or as a last resort, using
+item's ID (which MUST be publisher bare JID according to pubsub-attachments
+specification).
+"""
+if len(items) != 1:
+log.warning(
+"we should get exactly one attachment item for an entity, got "
+f"{len(items)})"
+)
+actor_id = await self.get_ap_actor_id_from_account(ap_account)
+inbox = await self.get_ap_inbox_from_id(actor_id)
+item_elt = items[0]
+item_id = item_elt["id"]
+if publisher is None:
+item_pub_s = item_elt.getAttribute("publisher")
+publisher = jid.JID(item_pub_s) if item_pub_s else jid.JID(item_id)
+if publisher.userhost() != item_id:
+log.warning(
+"attachments item ID must be publisher's bare JID, ignoring: "
+f"{item_elt.toXml()}"
+)
+return
+if self.is_virtual_jid(publisher):
+log.debug(f"ignoring item coming from local virtual JID {publisher}")
+return
+if publisher is not None:
+item_elt["publisher"] = publisher.userhost()
+item_service, item_node, item_id = self._pa.attachment_node_2_item(node)
+item_account = await self.get_ap_account_from_jid_and_node(item_service, item_node)
+if self.is_virtual_jid(item_service):
+# it's a virtual JID mapping to an external AP actor, we can use the
+# item_id directly
+item_url = item_id
+if not item_url.startswith("https:"):
+log.warning(
+"item ID of external AP actor is not an https link, ignoring: "
+f"{item_id!r}"
+)
+return
+else:
+item_url = self.build_apurl(TYPE_ITEM, item_account, item_id)
+old_attachment_pubsub_items = await self.host.memory.storage.search_pubsub_items({
+"profiles": [self.client.profile],
+"services": [service],
+"nodes": [node],
+"names": [item_elt["id"]]
+})
+if not old_attachment_pubsub_items:
+old_attachment = {}
+else:
+old_attachment_items = [i.data for i in old_attachment_pubsub_items]
+old_attachments = self._pa.items_2_attachment_data(client, old_attachment_items)
+try:
+old_attachment = old_attachments[0]
+except IndexError:
+# no known element was present in attachments
+old_attachment = {}
+publisher_account = await self.get_ap_account_from_jid_and_node(
+publisher,
+None
+)
+publisher_actor_id = self.build_apurl(TYPE_ACTOR, publisher_account)
+try:
+attachments = self._pa.items_2_attachment_data(client, [item_elt])[0]
+except IndexError:
+# no known element was present in attachments
+attachments = {}
+# noticed
+if "noticed" in attachments:
+if not "noticed" in old_attachment:
+# new "noticed" attachment, we translate to "Like" activity
+activity_id = self.build_apurl("like", item_account, item_id)
+activity = self.create_activity(
+TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id
+)
+activity["to"] = [ap_account]
+activity["cc"] = [NS_AP_PUBLIC]
+await self.sign_and_post(inbox, publisher_actor_id, activity)
+else:
+if "noticed" in old_attachment:
+# "noticed" attachment has been removed, we undo the "Like" activity
+activity_id = self.build_apurl("like", item_account, item_id)
+activity = self.create_activity(
+TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id
+)
+activity["to"] = [ap_account]
+activity["cc"] = [NS_AP_PUBLIC]
+undo = self.create_activity("Undo", publisher_actor_id, activity)
+await self.sign_and_post(inbox, publisher_actor_id, undo)
+# reactions
+new_reactions = set(attachments.get("reactions", {}).get("reactions", []))
+old_reactions = set(old_attachment.get("reactions", {}).get("reactions", []))
+reactions_remove = old_reactions - new_reactions
+reactions_add = new_reactions - old_reactions
+for reactions, undo in ((reactions_remove, True), (reactions_add, False)):
+for reaction in reactions:
+activity_id = self.build_apurl(
+"reaction", item_account, item_id, reaction.encode().hex()
+)
+reaction_activity = self.create_activity(
+TYPE_REACTION, publisher_actor_id, item_url,
+activity_id=activity_id
+)
+reaction_activity["content"] = reaction
+reaction_activity["to"] = [ap_account]
+reaction_activity["cc"] = [NS_AP_PUBLIC]
+if undo:
+activy = self.create_activity(
+"Undo", publisher_actor_id, reaction_activity
+)
+else:
+activy = reaction_activity
+await self.sign_and_post(inbox, publisher_actor_id, activy)
+# RSVP
+if "rsvp" in attachments:
+attending = attachments["rsvp"].get("attending", "no")
+old_attending = old_attachment.get("rsvp", {}).get("attending", "no")
+if attending != old_attending:
+activity_type = TYPE_JOIN if attending == "yes" else TYPE_LEAVE
+activity_id = self.build_apurl(activity_type.lower(), item_account, item_id)
+activity = self.create_activity(
+activity_type, publisher_actor_id, item_url, activity_id=activity_id
+)
+activity["to"] = [ap_account]
+activity["cc"] = [NS_AP_PUBLIC]
+await self.sign_and_post(inbox, publisher_actor_id, activity)
+else:
+if "rsvp" in old_attachment:
+old_attending = old_attachment.get("rsvp", {}).get("attending", "no")
+if old_attending == "yes":
+activity_id = self.build_apurl(TYPE_LEAVE.lower(), item_account, item_id)
+activity = self.create_activity(
+TYPE_LEAVE, publisher_actor_id, item_url, activity_id=activity_id
+)
+activity["to"] = [ap_account]
+activity["cc"] = [NS_AP_PUBLIC]
+await self.sign_and_post(inbox, publisher_actor_id, activity)
+if service.user and self.is_virtual_jid(service):
+# the item is on a virtual service, we need to store it in cache
+log.debug("storing attachments item in cache")
+cached_node = await self.host.memory.storage.get_pubsub_node(
+client, service, node, with_subscriptions=True, create=True
+)
+await self.host.memory.storage.cache_pubsub_items(
+self.client,
+cached_node,
+[item_elt],
+[attachments]
+)
+async def sign_and_post(self, url: str, actor_id: str, doc: dict) -> TReqResponse:
+"""Sign a documentent and post it to AP server
+@param url: AP server endpoint
+@param actor_id: originating actor ID (URL)
+@param doc: document to send
+"""
+if self.verbose:
+__, actor_args = self.parse_apurl(actor_id)
+actor_account = actor_args[0]
+to_log = [
+"",
+f">>> {actor_account} is signing and posting to {url}:\n{pformat(doc)}"
+]
+p_url = parse.urlparse(url)
+body = json.dumps(doc).encode()
+digest_algo, digest_hash = self.get_digest(body)
+digest = f"{digest_algo}={digest_hash}"
+headers = {
+"(request-target)": f"post {p_url.path}",
+"Host": p_url.hostname,
+"Date": http.datetimeToString().decode(),
+"Digest": digest
+}
+headers["Content-Type"] = (
+'application/activity+json'
+)
+headers, __ = self.get_signature_data(self.get_key_id(actor_id), headers)
+if self.verbose:
+if self.verbose>=3:
+h_to_log = "\n".join(f"    {k}: {v}" for k,v in headers.items())
+to_log.append(f"  headers:\n{h_to_log}")
+to_log.append("---")
+log.info("\n".join(to_log))
+resp = await treq.post(
+url,
+body,
+headers=headers,
+)
+if resp.code >= 300:
+text = await resp.text()
+log.warning(f"POST request to {url} failed [{resp.code}]: {text}")
+elif self.verbose:
+log.info(f"==> response code: {resp.code}")
+return resp
+def _publish_message(self, mess_data_s: str, service_s: str, profile: str):
+mess_data: dict = data_format.deserialise(mess_data_s) # type: ignore
+service = jid.JID(service_s)
+client = self.host.get_client(profile)
+return defer.ensureDeferred(self.publish_message(client, mess_data, service))
+@async_lru(maxsize=LRU_MAX_SIZE)
+async def get_ap_actor_id_from_account(self, account: str) -> str:
+"""Retrieve account ID from it's handle using WebFinger
+Don't use this method to get local actor id from a local account derivated for
+JID: in this case, the actor ID is retrieve with
+``self.build_apurl(TYPE_ACTOR, ap_account)``
+@param account: AP handle (user@domain.tld)
+@return: Actor ID (which is an URL)
+"""
+if account.count("@") != 1 or "/" in account:
+raise ValueError(f"Invalid account: {account!r}")
+host = account.split("@")[1]
+try:
+finger_data = await treq.json_content(await treq.get(
+f"https://{host}/.well-known/webfinger?"
+f"resource=acct:{parse.quote_plus(account)}",
+))
+except Exception as e:
+raise exceptions.DataError(f"Can't get webfinger data for {account!r}: {e}")
+for link in finger_data.get("links", []):
+if (
+link.get("type") == "application/activity+json"
+and link.get("rel") == "self"
+):
+href = link.get("href", "").strip()
+if not href:
+raise ValueError(
+f"Invalid webfinger data for {account:r}: missing href"
+)
+break
+else:
+raise ValueError(
+f"No ActivityPub link found for {account!r}"
+)
+return href
+async def get_ap_actor_data_from_account(self, account: str) -> dict:
+"""Retrieve ActivityPub Actor data
+@param account: ActivityPub Actor identifier
+"""
+href = await self.get_ap_actor_id_from_account(account)
+return await self.ap_get(href)
+async def get_ap_inbox_from_id(self, actor_id: str, use_shared: bool = True) -> str:
+"""Retrieve inbox of an actor_id
+@param use_shared: if True, and a shared inbox exists, it will be used instead of
+the user inbox
+"""
+data = await self.get_actor_data(actor_id)
+if use_shared:
+try:
+return data["endpoints"]["sharedInbox"]
+except KeyError:
+pass
+return data["inbox"]
+@async_lru(maxsize=LRU_MAX_SIZE)
+async def get_ap_account_from_id(self, actor_id: str) -> str:
+"""Retrieve AP account from the ID URL
+Works with external or local actor IDs.
+@param actor_id: AP ID of the actor (URL to the actor data)
+@return: AP handle
+"""
+if self.is_local_url(actor_id):
+url_type, url_args = self.parse_apurl(actor_id)
+if url_type != "actor" or not url_args:
+raise exceptions.DataError(
+f"invalid local actor ID: {actor_id}"
+)
+account = url_args[0]
+try:
+account_user, account_host = account.split('@')
+except ValueError:
+raise exceptions.DataError(
+f"invalid account from url: {actor_id}"
+)
+if not account_user or account_host != self.public_url:
+raise exceptions.DataError(
+f"{account!r} is not a valid local account (from {actor_id})"
+)
+return account
+url_parsed = parse.urlparse(actor_id)
+actor_data = await self.get_actor_data(actor_id)
+username = actor_data.get("preferredUsername")
+if not username:
+raise exceptions.DataError(
+'No "preferredUsername" field found, can\'t retrieve actor account'
+)
+account = f"{username}@{url_parsed.hostname}"
+# we try to retrieve the actor ID from the account to check it
+found_id = await self.get_ap_actor_id_from_account(account)
+if found_id != actor_id:
+# cf. https://socialhub.activitypub.rocks/t/how-to-retrieve-user-server-tld-handle-from-actors-url/2196
+msg = (
+f"Account ID found on WebFinger {found_id!r} doesn't match our actor ID "
+f"({actor_id!r}). This AP instance doesn't seems to use "
+'"preferredUsername" as we expect.'
+)
+log.warning(msg)
+raise exceptions.DataError(msg)
+return account
+async def get_ap_items(
+self,
+collection: dict,
+max_items: Optional[int] = None,
+chronological_pagination: bool = True,
+after_id: Optional[str] = None,
+start_index: Optional[int] = None,
+parser: Optional[Callable[[dict], Awaitable[domish.Element]]] = None,
+only_ids: bool = False,
+) -> Tuple[List[domish.Element], rsm.RSMResponse]:
+"""Retrieve AP items and convert them to XMPP items
+@param account: AP account handle to get items from
+@param max_items: maximum number of items to retrieve
+retrieve all items by default
+@param chronological_pagination: get pages in chronological order
+AP use reversed chronological order for pagination, "first" page returns more
+recent items. If "chronological_pagination" is True, "last" AP page will be
+retrieved first.
+@param after_id: if set, retrieve items starting from given ID
+Due to ActivityStream Collection Paging limitations, this is inefficient and
+if ``after_id`` is not already in cache, we have to retrieve every page until
+we find it.
+In most common cases, ``after_id`` should be in cache though (client usually
+use known ID when in-order pagination is used).
+@param start_index: start retrieving items from the one with given index
+Due to ActivityStream Collection Paging limitations, this is inefficient and
+all pages before the requested index will be retrieved to count items.
+@param parser: method to use to parse AP items and get XMPP item elements
+if None, use default generic parser
+@param only_ids: if True, only retrieve items IDs
+Retrieving only item IDs avoid HTTP requests to retrieve items, it may be
+sufficient in some use cases (e.g. when retrieving following/followers
+collections)
+@return: XMPP Pubsub items and corresponding RSM Response
+Items are always returned in chronological order in the result
+"""
+if parser is None:
+parser = self.ap_item_2_mb_elt
+rsm_resp: Dict[str, Union[bool, int]] = {}
+try:
+count = collection["totalItems"]
+except KeyError:
+log.warning(
+f'"totalItems" not found in collection {collection.get("id")}, '
+"defaulting to 20"
+)
+count = 20
+else:
+log.info(f"{collection.get('id')} has {count} item(s)")
+rsm_resp["count"] = count
+if start_index is not None:
+assert chronological_pagination and after_id is None
+if start_index >= count:
+return [], rsm_resp
+elif start_index == 0:
+# this is the default behaviour
+pass
+elif start_index > 5000:
+raise error.StanzaError(
+"feature-not-implemented",
+text="Maximum limit for previous_index has been reached, this limit"
+"is set to avoid DoS"
+)
+else:
+# we'll convert "start_index" to "after_id", thus we need the item just
+# before "start_index"
+previous_index = start_index - 1
+retrieved_items = 0
+current_page = collection["last"]
+while retrieved_items < count:
+page_data, items = await self.parse_ap_page(
+current_page, parser, only_ids
+)
+if not items:
+log.warning(f"found an empty AP page at {current_page}")
+return [], rsm_resp
+page_start_idx = retrieved_items
+retrieved_items += len(items)
+if previous_index <= retrieved_items:
+after_id = items[previous_index - page_start_idx]["id"]
+break
+try:
+current_page = page_data["prev"]
+except KeyError:
+log.warning(
+f"missing previous page link at {current_page}: {page_data!r}"
+)
+raise error.StanzaError(
+"service-unavailable",
+"Error while retrieving previous page from AP service at "
+f"{current_page}"
+)
+init_page = "last" if chronological_pagination else "first"
+page = collection.get(init_page)
+if not page:
+raise exceptions.DataError(
+f"Initial page {init_page!r} not found for collection "
+f"{collection.get('id')})"
+)
+items = []
+page_items = []
+retrieved_items = 0
+found_after_id = False
+while retrieved_items < count:
+__, page_items = await self.parse_ap_page(page, parser, only_ids)
+if not page_items:
+break
+retrieved_items += len(page_items)
+if after_id is not None and not found_after_id:
+# if we have an after_id, we ignore all items until the requested one is
+# found
+try:
+limit_idx = [i["id"] for i in page_items].index(after_id)
+except ValueError:
+# if "after_id" is not found, we don't add any item from this page
+page_id = page.get("id") if isinstance(page, dict) else page
+log.debug(f"{after_id!r} not found at {page_id}, skipping")
+else:
+found_after_id = True
+if chronological_pagination:
+start_index = retrieved_items - len(page_items) + limit_idx + 1
+page_items = page_items[limit_idx+1:]
+else:
+start_index = count - (retrieved_items - len(page_items) +
+limit_idx + 1)
+page_items = page_items[:limit_idx]
+items.extend(page_items)
+else:
+items.extend(page_items)
+if max_items is not None and len(items) >= max_items:
+if chronological_pagination:
+items = items[:max_items]
+else:
+items = items[-max_items:]
+break
+page = collection.get("prev" if chronological_pagination else "next")
+if not page:
+break
+if after_id is not None and not found_after_id:
+raise error.StanzaError("item-not-found")
+if items:
+if after_id is None:
+rsm_resp["index"] = 0 if chronological_pagination else count - len(items)
+if start_index is not None:
+rsm_resp["index"] = start_index
+elif after_id is not None:
+log.warning("Can't determine index of first element")
+elif chronological_pagination:
+rsm_resp["index"] = 0
+else:
+rsm_resp["index"] = count - len(items)
+rsm_resp.update({
+"first": items[0]["id"],
+"last": items[-1]["id"]
+})
+return items, rsm.RSMResponse(**rsm_resp)
+async def ap_item_2_mb_data_and_elt(self, ap_item: dict) -> Tuple[dict, domish.Element]:
+"""Convert AP item to parsed microblog data and corresponding item element"""
+mb_data = await self.ap_item_2_mb_data(ap_item)
+item_elt = await self._m.mb_data_2_entry_elt(
+self.client, mb_data, mb_data["id"], None, self._m.namespace
+)
+if "repeated" in mb_data["extra"]:
+item_elt["publisher"] = mb_data["extra"]["repeated"]["by"]
+else:
+item_elt["publisher"] = mb_data["author_jid"]
+return mb_data, item_elt
+async def ap_item_2_mb_elt(self, ap_item: dict) -> domish.Element:
+"""Convert AP item to XMPP item element"""
+__, item_elt = await self.ap_item_2_mb_data_and_elt(ap_item)
+return item_elt
+async def parse_ap_page(
+self,
+page: Union[str, dict],
+parser: Callable[[dict], Awaitable[domish.Element]],
+only_ids: bool = False
+) -> Tuple[dict, List[domish.Element]]:
+"""Convert AP objects from an AP page to XMPP items
+@param page: Can be either url linking and AP page, or the page data directly
+@param parser: method to use to parse AP items and get XMPP item elements
+@param only_ids: if True, only retrieve items IDs
+@return: page data, pubsub items
+"""
+page_data = await self.ap_get_object(page)
+if page_data is None:
+log.warning('No data found in collection')
+return {}, []
+ap_items = await self.ap_get_list(page_data, "orderedItems", only_ids=only_ids)
+if ap_items is None:
+ap_items = await self.ap_get_list(page_data, "items", only_ids=only_ids)
+if not ap_items:
+log.warning(f'No item field found in collection: {page_data!r}')
+return page_data, []
+else:
+log.warning(
+"Items are not ordered, this is not spec compliant"
+)
+items = []
+# AP Collections are in antichronological order, but we expect chronological in
+# Pubsub, thus we reverse it
+for ap_item in reversed(ap_items):
+try:
+items.append(await parser(ap_item))
+except (exceptions.DataError, NotImplementedError, error.StanzaError):
+continue
+return page_data, items
+async def get_comments_nodes(
+self,
+item_id: str,
+parent_id: Optional[str]
+) -> Tuple[Optional[str], Optional[str]]:
+"""Get node where this item is and node to use for comments
+if config option "comments_max_depth" is set, a common node will be used below the
+given depth
+@param item_id: ID of the reference item
+@param parent_id: ID of the parent item if any (the ID set in "inReplyTo")
+@return: a tuple with parent_node_id, comments_node_id:
+- parent_node_id is the ID of the node where reference item must be. None is
+returned when the root node (i.e. not a comments node) must be used.
+- comments_node_id: is the ID of the node to use for comments. None is
+returned when no comment node must be used (happens when we have reached
+"comments_max_depth")
+"""
+if parent_id is None or not self.comments_max_depth:
+return (
+self._m.get_comments_node(parent_id) if parent_id is not None else None,
+self._m.get_comments_node(item_id)
+)
+parent_url = parent_id
+parents = []
+for __ in range(COMMENTS_MAX_PARENTS):
+parent_item = await self.ap_get(parent_url)
+parents.insert(0, parent_item)
+parent_url = parent_item.get("inReplyTo")
+if parent_url is None:
+break
+parent_limit = self.comments_max_depth-1
+if len(parents) <= parent_limit:
+return (
+self._m.get_comments_node(parents[-1]["id"]),
+self._m.get_comments_node(item_id)
+)
+else:
+last_level_item = parents[parent_limit]
+return (
+self._m.get_comments_node(last_level_item["id"]),
+None
+)
+async def ap_item_2_mb_data(self, ap_item: dict) -> dict:
+"""Convert AP activity or object to microblog data
+@param ap_item: ActivityPub item to convert
+Can be either an activity of an object
+@return: AP Item's Object and microblog data
+@raise exceptions.DataError: something is invalid in the AP item
+@raise NotImplementedError: some AP data is not handled yet
+@raise error.StanzaError: error while contacting the AP server
+"""
+is_activity = self.is_activity(ap_item)
+if is_activity:
+ap_object = await self.ap_get_object(ap_item, "object")
+if not ap_object:
+log.warning(f'No "object" found in AP item {ap_item!r}')
+raise exceptions.DataError
+else:
+ap_object = ap_item
+item_id = ap_object.get("id")
+if not item_id:
+log.warning(f'No "id" found in AP item: {ap_object!r}')
+raise exceptions.DataError
+mb_data = {"id": item_id, "extra": {}}
+# content
+try:
+language, content_xhtml = ap_object["contentMap"].popitem()
+except (KeyError, AttributeError):
+try:
+mb_data["content_xhtml"] = ap_object["content"]
+except KeyError:
+log.warning(f"no content found:\n{ap_object!r}")
+raise exceptions.DataError
+else:
+mb_data["language"] = language
+mb_data["content_xhtml"] = content_xhtml
+mb_data["content"] = await self._t.convert(
+mb_data["content_xhtml"],
+self._t.SYNTAX_XHTML,
+self._t.SYNTAX_TEXT,
+False,
+)
+if "attachment" in ap_object:
+attachments = mb_data["extra"][C.KEY_ATTACHMENTS] = []
+for ap_attachment in ap_object["attachment"]:
+try:
+url = ap_attachment["url"]
+except KeyError:
+log.warning(
+f'"url" missing in AP attachment, ignoring: {ap_attachment}'
+)
+continue
+if not url.startswith("http"):
+log.warning(f"non HTTP URL in attachment, ignoring: {ap_attachment}")
+continue
+attachment = {"url": url}
+for ap_key, key in (
+("mediaType", "media_type"),
+# XXX: as weird as it seems, "name" is actually used for description
+#   in AP world
+("name", "desc"),
+):
+value = ap_attachment.get(ap_key)
+if value:
+attachment[key] = value
+attachments.append(attachment)
+# author
+if is_activity:
+authors = await self.ap_get_actors(ap_item, "actor")
+else:
+authors = await self.ap_get_actors(ap_object, "attributedTo")
+if len(authors) > 1:
+# we only keep first item as author
+# TODO: handle multiple actors
+log.warning("multiple actors are not managed")
+account = authors[0]
+author_jid = self.get_local_jid_from_account(account).full()
+mb_data["author"] = account.split("@", 1)[0]
+mb_data["author_jid"] = author_jid
+# published/updated
+for field in ("published", "updated"):
+value = ap_object.get(field)
+if not value and field == "updated":
+value = ap_object.get("published")
+if value:
+try:
+mb_data[field] = calendar.timegm(
+dateutil.parser.parse(str(value)).utctimetuple()
+)
+except dateutil.parser.ParserError as e:
+log.warning(f"Can't parse {field!r} field: {e}")
+# repeat
+if "_repeated" in ap_item:
+mb_data["extra"]["repeated"] = ap_item["_repeated"]
+# comments
+in_reply_to = ap_object.get("inReplyTo")
+__, comments_node = await self.get_comments_nodes(item_id, in_reply_to)
+if comments_node is not None:
+comments_data = {
+"service": author_jid,
+"node": comments_node,
+"uri": uri.build_xmpp_uri(
+"pubsub",
+path=author_jid,
+node=comments_node
+)
+}
+mb_data["comments"] = [comments_data]
+return mb_data
+async def get_reply_to_id_from_xmpp_node(
+self,
+client: SatXMPPEntity,
+ap_account: str,
+parent_item: str,
+mb_data: dict
+) -> str:
+"""Get URL to use for ``inReplyTo`` field in AP item.
+There is currently no way to know the parent service of a comment with XEP-0277.
+To work around that, we try to check if we have this item in the cache (we
+should). If there is more that one item with this ID, we first try to find one
+with this author_jid. If nothing is found, we use ap_account to build `inReplyTo`.
+@param ap_account: AP account corresponding to the publication author
+@param parent_item: ID of the node where the publication this item is replying to
+has been posted
+@param mb_data: microblog data of the publication
+@return: URL to use in ``inReplyTo`` field
+"""
+# FIXME: propose a protoXEP to properly get parent item, node and service
+found_items = await self.host.memory.storage.search_pubsub_items({
+"profiles": [client.profile],
+"names": [parent_item]
+})
+if not found_items:
+log.warning(f"parent item {parent_item!r} not found in cache")
+parent_ap_account = ap_account
+elif len(found_items) == 1:
+cached_node = found_items[0].node
+parent_ap_account = await self.get_ap_account_from_jid_and_node(
+cached_node.service,
+cached_node.name
+)
+else:
+# we found several cached item with given ID, we check if there is one
+# corresponding to this author
+try:
+author = jid.JID(mb_data["author_jid"]).userhostJID()
+cached_item = next(
+i for i in found_items
+if jid.JID(i.data["publisher"]).userhostJID()
+== author
+)
+except StopIteration:
+# no item corresponding to this author, we use ap_account
+log.warning(
+"Can't find a single cached item for parent item "
+f"{parent_item!r}"
+)
+parent_ap_account = ap_account
+else:
+cached_node = cached_item.node
+parent_ap_account = await self.get_ap_account_from_jid_and_node(
+cached_node.service,
+cached_node.name
+)
+return self.build_apurl(
+TYPE_ITEM, parent_ap_account, parent_item
+)
+async def repeated_mb_2_ap_item(
+self,
+mb_data: dict
+) -> dict:
+"""Convert repeated blog item to suitable AP Announce activity
+@param mb_data: microblog metadata of an item repeating an other blog post
+@return: Announce activity linking to the repeated item
+"""
+repeated = mb_data["extra"]["repeated"]
+repeater = jid.JID(repeated["by"])
+repeater_account = await self.get_ap_account_from_jid_and_node(
+repeater,
+None
+)
+repeater_id = self.build_apurl(TYPE_ACTOR, repeater_account)
+repeated_uri = repeated["uri"]
+if not repeated_uri.startswith("xmpp:"):
+log.warning(
+"Only xmpp: URL are handled for repeated item at the moment, ignoring "
+f"item {mb_data}"
+)
+raise NotImplementedError
+parsed_url = uri.parse_xmpp_uri(repeated_uri)
+if parsed_url["type"] != "pubsub":
+log.warning(
+"Only pubsub URL are handled for repeated item at the moment, ignoring "
+f"item {mb_data}"
+)
+raise NotImplementedError
+rep_service = jid.JID(parsed_url["path"])
+rep_item = parsed_url["item"]
+activity_id = self.build_apurl("item", repeater.userhost(), mb_data["id"])
+if self.is_virtual_jid(rep_service):
+# it's an AP actor linked through this gateway
+# in this case we can simply use the item ID
+if not rep_item.startswith("https:"):
+log.warning(
+f"Was expecting an HTTPS url as item ID and got {rep_item!r}\n"
+f"{mb_data}"
+)
+announced_uri = rep_item
+repeated_account = self._e.unescape(rep_service.user)
+else:
+# the repeated item is an XMPP publication, we build the corresponding ID
+rep_node = parsed_url["node"]
+repeated_account = await self.get_ap_account_from_jid_and_node(
+rep_service, rep_node
+)
+announced_uri = self.build_apurl("item", repeated_account, rep_item)
+announce = self.create_activity(
+"Announce", repeater_id, announced_uri, activity_id=activity_id
+)
+announce["to"] = [NS_AP_PUBLIC]
+announce["cc"] = [
+self.build_apurl(TYPE_FOLLOWERS, repeater_account),
+await self.get_ap_actor_id_from_account(repeated_account)
+]
+return announce
+async def mb_data_2_ap_item(
+self,
+client: SatXMPPEntity,
+mb_data: dict,
+public: bool =True,
+is_new: bool = True,
+) -> dict:
+"""Convert Libervia Microblog Data to ActivityPub item
+@param mb_data: microblog data (as used in plugin XEP-0277) to convert
+If ``public`` is True, ``service`` and ``node`` keys must be set.
+If ``published`` is not set, current datetime will be used
+@param public: True if the message is not a private/direct one
+if True, the AP Item will be marked as public, and AP followers of target AP
+account (which retrieve from ``service``) will be put in ``cc``.
+``inReplyTo`` will also be set if suitable
+if False, no destinee will be set (i.e., no ``to`` or ``cc`` or public flag).
+This is usually used for direct messages.
+@param is_new: if True, the item is a new one (no instance has been found in
+cache).
+If True, a "Create" activity will be generated, otherwise an "Update" one will
+be.
+@return: Activity item
+"""
+extra = mb_data.get("extra", {})
+if "repeated" in extra:
+return await self.repeated_mb_2_ap_item(mb_data)
+if not mb_data.get("id"):
+mb_data["id"] = shortuuid.uuid()
+if not mb_data.get("author_jid"):
+mb_data["author_jid"] = client.jid.userhost()
+ap_account = await self.get_ap_account_from_jid_and_node(
+jid.JID(mb_data["author_jid"]),
+None
+)
+url_actor = self.build_apurl(TYPE_ACTOR, ap_account)
+url_item = self.build_apurl(TYPE_ITEM, ap_account, mb_data["id"])
+ap_object = {
+"id": url_item,
+"type": "Note",
+"published": utils.xmpp_date(mb_data.get("published")),
+"attributedTo": url_actor,
+"content": mb_data.get("content_xhtml") or mb_data["content"],
+}
+language = mb_data.get("language")
+if language:
+ap_object["contentMap"] = {language: ap_object["content"]}
+attachments = extra.get(C.KEY_ATTACHMENTS)
+if attachments:
+ap_attachments = ap_object["attachment"] = []
+for attachment in attachments:
+try:
+url = next(
+s['url'] for s in attachment["sources"] if 'url' in s
+)
+except (StopIteration, KeyError):
+log.warning(
+f"Ignoring attachment without URL: {attachment}"
+)
+continue
+ap_attachment = {
+"url": url
+}
+for key, ap_key in (
+("media_type", "mediaType"),
+# XXX: yes "name", cf. [ap_item_2_mb_data]
+("desc", "name"),
+):
+value = attachment.get(key)
+if value:
+ap_attachment[ap_key] = value
+ap_attachments.append(ap_attachment)
+if public:
+ap_object["to"] = [NS_AP_PUBLIC]
+if self.auto_mentions:
+for m in RE_MENTION.finditer(ap_object["content"]):
+mention = m.group()
+mentioned = mention[1:]
+__, m_host = mentioned.split("@", 1)
+if m_host in (self.public_url, self.client.jid.host):
+# we ignore mention of local users, they should be sent as XMPP
+# references
+continue
+try:
+mentioned_id = await self.get_ap_actor_id_from_account(mentioned)
+except Exception as e:
+log.warning(f"Can't add mention to {mentioned!r}: {e}")
+else:
+ap_object["to"].append(mentioned_id)
+ap_object.setdefault("tag", []).append({
+"type": TYPE_MENTION,
+"href": mentioned_id,
+"name": mention,
+})
+try:
+node = mb_data["node"]
+service = jid.JID(mb_data["service"])
+except KeyError:
+# node and service must always be specified when this method is used
+raise exceptions.InternalError(
+"node or service is missing in mb_data"
+)
+target_ap_account = await self.get_ap_account_from_jid_and_node(
+service, node
+)
+if self.is_virtual_jid(service):
+# service is a proxy JID for AP account
+actor_data = await self.get_ap_actor_data_from_account(target_ap_account)
+followers = actor_data.get("followers")
+else:
+# service is a real XMPP entity
+followers = self.build_apurl(TYPE_FOLLOWERS, target_ap_account)
+if followers:
+ap_object["cc"] = [followers]
+if self._m.is_comment_node(node):
+parent_item = self._m.get_parent_item(node)
+if self.is_virtual_jid(service):
+# the publication is on a virtual node (i.e. an XMPP node managed by
+# this gateway and linking to an ActivityPub actor)
+ap_object["inReplyTo"] = parent_item
+else:
+# the publication is from a followed real XMPP node
+ap_object["inReplyTo"] = await self.get_reply_to_id_from_xmpp_node(
+client,
+ap_account,
+parent_item,
+mb_data
+)
+return self.create_activity(
+"Create" if is_new else "Update", url_actor, ap_object, activity_id=url_item
+)
+async def publish_message(
+self,
+client: SatXMPPEntity,
+mess_data: dict,
+service: jid.JID
+) -> None:
+"""Send an AP message
+.. note::
+This is a temporary method used for development only
+@param mess_data: message data. Following keys must be set:
+``node``
+identifier of message which is being replied (this will
+correspond to pubsub node in the future)
+``content_xhtml`` or ``content``
+message body (respectively in XHTML or plain text)
+@param service: JID corresponding to the AP actor.
+"""
+if not service.user:
+raise ValueError("service must have a local part")
+account = self._e.unescape(service.user)
+ap_actor_data = await self.get_ap_actor_data_from_account(account)
+try:
+inbox_url = ap_actor_data["endpoints"]["sharedInbox"]
+except KeyError:
+raise exceptions.DataError("Can't get ActivityPub actor inbox")
+item_data = await self.mb_data_2_ap_item(client, mess_data)
+url_actor = item_data["actor"]
+resp = await self.sign_and_post(inbox_url, url_actor, item_data)
+async def ap_delete_item(
+self,
+jid_: jid.JID,
+node: Optional[str],
+item_id: str,
+public: bool = True
+) -> Tuple[str, Dict[str, Any]]:
+"""Build activity to delete an AP item
+@param jid_: JID of the entity deleting an item
+@param node: node where the item is deleted
+None if it's microblog or a message
+@param item_id: ID of the item to delete
+it's the Pubsub ID or message's origin ID
+@param public: if True, the activity will be addressed to public namespace
+@return: actor_id of the entity deleting the item, activity to send
+"""
+if node is None:
+node = self._m.namespace
+author_account = await self.get_ap_account_from_jid_and_node(jid_, node)
+author_actor_id = self.build_apurl(TYPE_ACTOR, author_account)
+items = await self.host.memory.storage.search_pubsub_items({
+"profiles": [self.client.profile],
+"services": [jid_],
+"names": [item_id]
+})
+if not items:
+log.warning(
+f"Deleting an unknown item at service {jid_}, node {node} and id "
+f"{item_id}"
+)
+else:
+try:
+mb_data = await self._m.item_2_mb_data(self.client, items[0].data, jid_, node)
+if "repeated" in mb_data["extra"]:
+# we are deleting a repeated item, we must translate this to an
+# "Undo" of the "Announce" activity instead of a "Delete" one
+announce = await self.repeated_mb_2_ap_item(mb_data)
+undo = self.create_activity("Undo", author_actor_id, announce)
+return author_actor_id, undo
+except Exception as e:
+log.debug(
+f"Can't parse item, maybe it's not a blog item: {e}\n"
+f"{items[0].toXml()}"
+)
+url_item = self.build_apurl(TYPE_ITEM, author_account, item_id)
+ap_item = self.create_activity(
+"Delete",
+author_actor_id,
+{
+"id": url_item,
+"type": TYPE_TOMBSTONE
+}
+)
+if public:
+ap_item["to"] = [NS_AP_PUBLIC]
+return author_actor_id, ap_item
+def _message_received_trigger(
+self,
+client: SatXMPPEntity,
+message_elt: domish.Element,
+post_treat: defer.Deferred
+) -> bool:
+"""add the gateway workflow on post treatment"""
+if self.client is None:
+log.debug(f"no client set, ignoring message: {message_elt.toXml()}")
+return True
+post_treat.addCallback(
+lambda mess_data: defer.ensureDeferred(self.onMessage(client, mess_data))
+)
+return True
+async def onMessage(self, client: SatXMPPEntity, mess_data: dict) -> dict:
+"""Called once message has been parsed
+this method handle the conversion to AP items and posting
+"""
+if client != self.client:
+return mess_data
+if mess_data["type"] not in ("chat", "normal"):
+log.warning(f"ignoring message with unexpected type: {mess_data}")
+return mess_data
+if not self.is_local(mess_data["from"]):
+log.warning(f"ignoring non local message: {mess_data}")
+return mess_data
+if not mess_data["to"].user:
+log.warning(
+f"ignoring message addressed to gateway itself: {mess_data}"
+)
+return mess_data
+actor_account = self._e.unescape(mess_data["to"].user)
+actor_id = await self.get_ap_actor_id_from_account(actor_account)
+inbox = await self.get_ap_inbox_from_id(actor_id, use_shared=False)
+try:
+language, message = next(iter(mess_data["message"].items()))
+except (KeyError, StopIteration):
+log.warning(f"ignoring empty message: {mess_data}")
+return mess_data
+mb_data = {
+"content": message,
+}
+if language:
+mb_data["language"] = language
+origin_id = mess_data["extra"].get("origin_id")
+if origin_id:
+# we need to use origin ID when present to be able to retract the message
+mb_data["id"] = origin_id
+attachments = mess_data["extra"].get(C.KEY_ATTACHMENTS)
+if attachments:
+mb_data["extra"] = {
+C.KEY_ATTACHMENTS: attachments
+}
+client = self.client.get_virtual_client(mess_data["from"])
+ap_item = await self.mb_data_2_ap_item(client, mb_data, public=False)
+ap_object = ap_item["object"]
+ap_object["to"] = ap_item["to"] = [actor_id]
+# we add a mention to direct message, otherwise peer is not notified in some AP
+# implementations (notably Mastodon), and the message may be missed easily.
+ap_object.setdefault("tag", []).append({
+"type": TYPE_MENTION,
+"href": actor_id,
+"name": f"@{actor_account}",
+})
+await self.sign_and_post(inbox, ap_item["actor"], ap_item)
+return mess_data
+async def _on_message_retract(
+self,
+client: SatXMPPEntity,
+message_elt: domish.Element,
+retract_elt: domish.Element,
+fastened_elts
+) -> bool:
+if client != self.client:
+return True
+from_jid = jid.JID(message_elt["from"])
+if not self.is_local(from_jid):
+log.debug(
+f"ignoring retract request from non local jid {from_jid}"
+)
+return False
+to_jid = jid.JID(message_elt["to"])
+if (to_jid.host != self.client.jid.full() or not to_jid.user):
+# to_jid should be a virtual JID from this gateway
+raise exceptions.InternalError(
+f"Invalid destinee's JID: {to_jid.full()}"
+)
+ap_account = self._e.unescape(to_jid.user)
+actor_id = await self.get_ap_actor_id_from_account(ap_account)
+inbox = await self.get_ap_inbox_from_id(actor_id, use_shared=False)
+url_actor, ap_item = await self.ap_delete_item(
+from_jid.userhostJID(), None, fastened_elts.id, public=False
+)
+resp = await self.sign_and_post(inbox, url_actor, ap_item)
+return False
+async def _on_reference_received(
+self,
+client: SatXMPPEntity,
+message_elt: domish.Element,
+reference_data: Dict[str, Union[str, int]]
+) -> bool:
+parsed_uri: dict = reference_data.get("parsed_uri")
+if not parsed_uri:
+log.warning(f"no parsed URI available in reference {reference_data}")
+return False
+try:
+mentioned = jid.JID(parsed_uri["path"])
+except RuntimeError:
+log.warning(f"invalid target: {reference_data['uri']}")
+return False
+if mentioned.host != self.client.jid.full() or not mentioned.user:
+log.warning(
+f"ignoring mentioned user {mentioned}, it's not a JID mapping an AP "
+"account"
+)
+return False
+ap_account = self._e.unescape(mentioned.user)
+actor_id = await self.get_ap_actor_id_from_account(ap_account)
+parsed_anchor: dict = reference_data.get("parsed_anchor")
+if not parsed_anchor:
+log.warning(f"no XMPP anchor, ignoring reference {reference_data!r}")
+return False
+if parsed_anchor["type"] != "pubsub":
+log.warning(
+f"ignoring reference with non pubsub anchor, this is not supported: "
+"{reference_data!r}"
+)
+return False
+try:
+pubsub_service = jid.JID(parsed_anchor["path"])
+except RuntimeError:
+log.warning(f"invalid anchor: {reference_data['anchor']}")
+return False
+pubsub_node = parsed_anchor.get("node")
+if not pubsub_node:
+log.warning(f"missing pubsub node in anchor: {reference_data['anchor']}")
+return False
+pubsub_item = parsed_anchor.get("item")
+if not pubsub_item:
+log.warning(f"missing pubsub item in anchor: {reference_data['anchor']}")
+return False
+cached_node = await self.host.memory.storage.get_pubsub_node(
+client, pubsub_service, pubsub_node
+)
+if not cached_node:
+log.warning(f"Anchored node not found in cache: {reference_data['anchor']}")
+return False
+cached_items, __ = await self.host.memory.storage.get_items(
+cached_node, item_ids=[pubsub_item]
+)
+if not cached_items:
+log.warning(
+f"Anchored pubsub item not found in cache: {reference_data['anchor']}"
+)
+return False
+cached_item = cached_items[0]
+mb_data = await self._m.item_2_mb_data(
+client, cached_item.data, pubsub_service, pubsub_node
+)
+ap_item = await self.mb_data_2_ap_item(client, mb_data)
+ap_object = ap_item["object"]
+ap_object["to"] = [actor_id]
+ap_object.setdefault("tag", []).append({
+"type": TYPE_MENTION,
+"href": actor_id,
+"name": ap_account,
+})
+inbox = await self.get_ap_inbox_from_id(actor_id, use_shared=False)
+resp = await self.sign_and_post(inbox, ap_item["actor"], ap_item)
+return False
+async def new_reply_to_xmpp_item(
+self,
+client: SatXMPPEntity,
+ap_item: dict,
+targets: Dict[str, Set[str]],
+mentions: List[Dict[str, str]],
+) -> None:
+"""We got an AP item which is a reply to an XMPP item"""
+in_reply_to = ap_item["inReplyTo"]
+url_type, url_args = self.parse_apurl(in_reply_to)
+if url_type != "item":
+log.warning(
+"Ignoring AP item replying to an XMPP item with an unexpected URL "
+f"type({url_type!r}):\n{pformat(ap_item)}"
+)
+return
+try:
+parent_item_account, parent_item_id = url_args[0], '/'.join(url_args[1:])
+except (IndexError, ValueError):
+log.warning(
+"Ignoring AP item replying to an XMPP item with invalid inReplyTo URL "
+f"({in_reply_to!r}):\n{pformat(ap_item)}"
+)
+return
+parent_item_service, parent_item_node = await self.get_jid_and_node(
+parent_item_account
+)
+if parent_item_node is None:
+parent_item_node = self._m.namespace
+items, __ = await self._p.get_items(
+client, parent_item_service, parent_item_node, item_ids=[parent_item_id]
+)
+try:
+parent_item_elt = items[0]
+except IndexError:
+log.warning(
+f"Can't find parent item at {parent_item_service} (node "
+f"{parent_item_node!r})\n{pformat(ap_item)}")
+return
+parent_item_parsed = await self._m.item_2_mb_data(
+client, parent_item_elt, parent_item_service, parent_item_node
+)
+try:
+comment_service = jid.JID(parent_item_parsed["comments"][0]["service"])
+comment_node = parent_item_parsed["comments"][0]["node"]
+except (KeyError, IndexError):
+# we don't have a comment node set for this item
+from libervia.backend.tools.xml_tools import pp_elt
+log.info(f"{pp_elt(parent_item_elt.toXml())}")
+raise NotImplementedError()
+else:
+__, item_elt = await self.ap_item_2_mb_data_and_elt(ap_item)
+await self._p.publish(client, comment_service, comment_node, [item_elt])
+await self.notify_mentions(
+targets, mentions, comment_service, comment_node, item_elt["id"]
+)
+def get_ap_item_targets(
+self,
+item: Dict[str, Any]
+) -> Tuple[bool, Dict[str, Set[str]], List[Dict[str, str]]]:
+"""Retrieve targets of an AP item, and indicate if it's a public one
+@param item: AP object payload
+@return: Are returned:
+- is_public flag, indicating if the item is world-readable
+- a dict mapping target type to targets
+"""
+targets: Dict[str, Set[str]] = {}
+is_public = False
+# TODO: handle "audience"
+for key in ("to", "bto", "cc", "bcc"):
+values = item.get(key)
+if not values:
+continue
+if isinstance(values, str):
+values = [values]
+for value in values:
+if value in PUBLIC_TUPLE:
+is_public = True
+continue
+if not value:
+continue
+if not self.is_local_url(value):
+continue
+target_type = self.parse_apurl(value)[0]
+if target_type != TYPE_ACTOR:
+log.debug(f"ignoring non actor type as a target: {href}")
+else:
+targets.setdefault(target_type, set()).add(value)
+mentions = []
+tags = item.get("tag")
+if tags:
+for tag in tags:
+if tag.get("type") != TYPE_MENTION:
+continue
+href = tag.get("href")
+if not href:
+log.warning('Missing "href" field from mention object: {tag!r}')
+continue
+if not self.is_local_url(href):
+continue
+uri_type = self.parse_apurl(href)[0]
+if uri_type != TYPE_ACTOR:
+log.debug(f"ignoring non actor URI as a target: {href}")
+continue
+mention = {"uri": href}
+mentions.append(mention)
+name = tag.get("name")
+if name:
+mention["content"] = name
+return is_public, targets, mentions
+async def new_ap_item(
+self,
+client: SatXMPPEntity,
+destinee: Optional[jid.JID],
+node: str,
+item: dict,
+) -> None:
+"""Analyse, cache and send notification for received AP item
+@param destinee: jid of the destinee,
+@param node: XMPP pubsub node
+@param item: AP object payload
+"""
+is_public, targets, mentions = self.get_ap_item_targets(item)
+if not is_public and targets.keys() == {TYPE_ACTOR}:
+# this is a direct message
+await self.handle_message_ap_item(
+client, targets, mentions, destinee, item
+)
+else:
+await self.handle_pubsub_ap_item(
+client, targets, mentions, destinee, node, item, is_public
+)
+async def handle_message_ap_item(
+self,
+client: SatXMPPEntity,
+targets: Dict[str, Set[str]],
+mentions: List[Dict[str, str]],
+destinee: Optional[jid.JID],
+item: dict,
+) -> None:
+"""Parse and deliver direct AP items translating to XMPP messages
+@param targets: actors where the item must be delivered
+@param destinee: jid of the destinee,
+@param item: AP object payload
+"""
+targets_jids = {
+await self.get_jid_from_id(t)
+for t_set in targets.values()
+for t in t_set
+}
+if destinee is not None:
+targets_jids.add(destinee)
+mb_data = await self.ap_item_2_mb_data(item)
+extra = {
+"origin_id": mb_data["id"]
+}
+attachments = mb_data["extra"].get(C.KEY_ATTACHMENTS)
+if attachments:
+extra[C.KEY_ATTACHMENTS] = attachments
+defer_l = []
+for target_jid in targets_jids:
+defer_l.append(
+client.sendMessage(
+target_jid,
+{'': mb_data.get("content", "")},
+mb_data.get("title"),
+extra=extra
+)
+)
+await defer.DeferredList(defer_l)
+async def notify_mentions(
+self,
+targets: Dict[str, Set[str]],
+mentions: List[Dict[str, str]],
+service: jid.JID,
+node: str,
+item_id: str,
+) -> None:
+"""Send mention notifications to recipients and mentioned entities
+XEP-0372 (References) is used.
+Mentions are also sent to recipients as they are primary audience (see
+https://www.w3.org/TR/activitystreams-vocabulary/#microsyntaxes).
+"""
+anchor = uri.build_xmpp_uri("pubsub", path=service.full(), node=node, item=item_id)
+seen = set()
+# we start with explicit mentions because mentions' content will be used in the
+# future to fill "begin" and "end" reference attributes (we can't do it at the
+# moment as there is no way to specify the XML element to use in the blog item).
+for mention in mentions:
+mentioned_jid = await self.get_jid_from_id(mention["uri"])
+self._refs.send_reference(
+self.client,
+to_jid=mentioned_jid,
+anchor=anchor
+)
+seen.add(mentioned_jid)
+remaining = {
+await self.get_jid_from_id(t)
+for t_set in targets.values()
+for t in t_set
+} - seen
+for target in remaining:
+self._refs.send_reference(
+self.client,
+to_jid=target,
+anchor=anchor
+)
+async def handle_pubsub_ap_item(
+self,
+client: SatXMPPEntity,
+targets: Dict[str, Set[str]],
+mentions: List[Dict[str, str]],
+destinee: Optional[jid.JID],
+node: str,
+item: dict,
+public: bool
+) -> None:
+"""Analyse, cache and deliver AP items translating to Pubsub
+@param targets: actors/collections where the item must be delivered
+@param destinee: jid of the destinee,
+@param node: XMPP pubsub node
+@param item: AP object payload
+@param public: True if the item is public
+"""
+# XXX: "public" is not used for now
+service = client.jid
+in_reply_to = item.get("inReplyTo")
+if in_reply_to and isinstance(in_reply_to, list):
+in_reply_to = in_reply_to[0]
+if in_reply_to and isinstance(in_reply_to, str):
+if self.is_local_url(in_reply_to):
+# this is a reply to an XMPP item
+await self.new_reply_to_xmpp_item(client, item, targets, mentions)
+return
+# this item is a reply to an AP item, we use or create a corresponding node
+# for comments
+parent_node, __ = await self.get_comments_nodes(item["id"], in_reply_to)
+node = parent_node or node
+cached_node = await self.host.memory.storage.get_pubsub_node(
+client, service, node, with_subscriptions=True, create=True,
+create_kwargs={"subscribed": True}
+)
+else:
+# it is a root item (i.e. not a reply to an other item)
+create = node == self._events.namespace
+cached_node = await self.host.memory.storage.get_pubsub_node(
+client, service, node, with_subscriptions=True, create=create
+)
+if cached_node is None:
+log.warning(
+f"Received item in unknown node {node!r} at {service}. This may be "
+f"due to a cache purge. We synchronise the node\n{item}"
+)
+return
+if item.get("type") == TYPE_EVENT:
+data, item_elt = await self.ap_events.ap_item_2_event_data_and_elt(item)
+else:
+data, item_elt = await self.ap_item_2_mb_data_and_elt(item)
+await self.host.memory.storage.cache_pubsub_items(
+client,
+cached_node,
+[item_elt],
+[data]
+)
+for subscription in cached_node.subscriptions:
+if subscription.state != SubscriptionState.SUBSCRIBED:
+continue
+self.pubsub_service.notifyPublish(
+service,
+node,
+[(subscription.subscriber, None, [item_elt])]
+)
+await self.notify_mentions(targets, mentions, service, node, item_elt["id"])
+async def new_ap_delete_item(
+self,
+client: SatXMPPEntity,
+destinee: Optional[jid.JID],
+node: str,
+item: dict,
+) -> None:
+"""Analyse, cache and send notification for received AP item
+@param destinee: jid of the destinee,
+@param node: XMPP pubsub node
+@param activity: parent AP activity
+@param item: AP object payload
+only the "id" field is used
+"""
+item_id = item.get("id")
+if not item_id:
+raise exceptions.DataError('"id" attribute is missing in item')
+if not item_id.startswith("http"):
+raise exceptions.DataError(f"invalid id: {item_id!r}")
+if self.is_local_url(item_id):
+raise ValueError("Local IDs should not be used")
+# we have no way to know if a deleted item is a direct one (thus a message) or one
+# converted to pubsub. We check if the id is in message history to decide what to
+# do.
+history = await self.host.memory.storage.get(
+client,
+History,
+History.origin_id,
+item_id,
+(History.messages, History.subjects)
+)
+if history is not None:
+# it's a direct message
+if history.source_jid != client.jid:
+log.warning(
+f"retraction received from an entity ''{client.jid}) which is "
+f"not the original sender of the message ({history.source_jid}), "
+"hack attemps?"
+)
+raise exceptions.PermissionError("forbidden")
+await self._r.retract_by_history(client, history)
+else:
+# no history in cache with this ID, it's probably a pubsub item
+cached_node = await self.host.memory.storage.get_pubsub_node(
+client, client.jid, node, with_subscriptions=True
+)
+if cached_node is None:
+log.warning(
+f"Received an item retract for node {node!r} at {client.jid} "
+"which is not cached"
+)
+raise exceptions.NotFound
+await self.host.memory.storage.delete_pubsub_items(cached_node, [item_id])
+# notifyRetract is expecting domish.Element instances
+item_elt = domish.Element((None, "item"))
+item_elt["id"] = item_id
+for subscription in cached_node.subscriptions:
+if subscription.state != SubscriptionState.SUBSCRIBED:
+continue
+self.pubsub_service.notifyRetract(
+client.jid,
+node,
+[(subscription.subscriber, None, [item_elt])]
+)

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_comp_ap_gateway/__init__.py @ 4071:4b842c1fb686

comparison libervia/backend/plugins/plugin_comp_ap_gateway/init.py @ 4071:4b842c1fb686