From 5e3d4ca581b3abb5ac386a98ea72bc9181c66025 Mon Sep 17 00:00:00 2001 From: Goffi Date: Wed, 31 Aug 2022 17:07:03 +0200 Subject: [PATCH] component AP gateway: AP <=> XMPP reactions conversions: - Pubsub Attachments plugin has been renamed to XEP-0470 following publication - XEP-0470 has been updated to follow 0.2 changes - AP reactions (as implemented in Pleroma) are converted to XEP-0470 - XEP-0470 events are converted to AP reactions (again, using "EmojiReact" from Pleroma) - AP activities related to attachments (like/reactions) are cached in Libervia because it's not possible to retrieve them from Pleroma instances once they have been emitted (doing an HTTP get on their ID returns a 404). For now those cache are not flushed, this should be improved in the future. - `sharedInbox` is used when available. Pleroma returns a 500 HTTP error when ``to`` or ``cc`` are used in a direct inbox. - reactions and like are not currently used for direct messages, because they can't be emitted from Pleroma in this case, thus there is no point in implementing them for the moment. rel 371 --- .../plugin_comp_ap_gateway/__init__.py | 61 +++++++++--- .../plugin_comp_ap_gateway/constants.py | 10 +- .../plugin_comp_ap_gateway/http_server.py | 95 +++++++++++++------ .../plugin_comp_ap_gateway/pubsub_service.py | 2 +- ...bsub_attachments.py => plugin_xep_0470.py} | 74 +++++++++------ 5 files changed, 173 insertions(+), 69 deletions(-) rename sat/plugins/{plugin_pubsub_attachments.py => plugin_xep_0470.py} (87%) diff --git a/sat/plugins/plugin_comp_ap_gateway/__init__.py b/sat/plugins/plugin_comp_ap_gateway/__init__.py index 7372aa50..6e0c7e28 100644 --- a/sat/plugins/plugin_comp_ap_gateway/__init__.py +++ b/sat/plugins/plugin_comp_ap_gateway/__init__.py @@ -71,6 +71,7 @@ from .constants import ( TYPE_TOMBSTONE, TYPE_MENTION, TYPE_LIKE, + TYPE_REACTION, NS_AP, NS_AP_PUBLIC, PUBLIC_TUPLE @@ -92,8 +93,8 @@ PLUGIN_INFO = { C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: [ "XEP-0054", "XEP-0060", "XEP-0084", "XEP-0106", "XEP-0277", "XEP-0292", - "XEP-0329", "XEP-0372", "XEP-0424", "XEP-0465", "PUBSUB_CACHE", "TEXT_SYNTAXES", - "IDENTITY", "PUBSUB_ATTACHMENTS" + "XEP-0329", "XEP-0372", "XEP-0424", "XEP-0465", "XEP-0470", "PUBSUB_CACHE", + "TEXT_SYNTAXES", "IDENTITY" ], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "APGateway", @@ -132,7 +133,7 @@ class APGateway: self._c = host.plugins["PUBSUB_CACHE"] self._t = host.plugins["TEXT_SYNTAXES"] self._i = host.plugins["IDENTITY"] - self._pa = host.plugins["PUBSUB_ATTACHMENTS"] + self._pa = host.plugins["XEP-0470"] self._p.addManagedNode( "", items_cb=self._itemsReceived, @@ -335,7 +336,7 @@ class APGateway: if resp.code >= 300: text = await resp.text() if resp.code == 404: - raise exceptions.NotFound() + raise exceptions.NotFound(f"Can't find resource at {url}") else: msg = f"HTTP error {resp.code}: {text}" raise exceptions.ExternalRequestError(msg) @@ -1104,6 +1105,7 @@ class APGateway: old_attachment_pubsub_items = await self.host.memory.storage.searchPubsubItems({ "profiles": [self.client.profile], "services": [service], + "nodes": [node], "names": [item_elt["id"]] }) if not old_attachment_pubsub_items: @@ -1127,6 +1129,7 @@ class APGateway: # 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 @@ -1134,7 +1137,8 @@ class APGateway: like = self.createActivity( TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id ) - like["to"] = [NS_AP_PUBLIC] + like["to"] = [ap_account] + like["cc"] = [NS_AP_PUBLIC] await self.signAndPost(inbox, publisher_actor_id, like) else: if "noticed" in old_attachment: @@ -1143,10 +1147,36 @@ class APGateway: like = self.createActivity( TYPE_LIKE, publisher_actor_id, item_url, activity_id=activity_id ) - like["to"] = [NS_AP_PUBLIC] + like["to"] = [ap_account] + like["cc"] = [NS_AP_PUBLIC] undo = self.createActivity("Undo", publisher_actor_id, like) await self.signAndPost(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.buildAPURL( + "reaction", item_account, item_id, reaction.encode().hex() + ) + reaction_activity = self.createActivity( + 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.createActivity( + "Undo", publisher_actor_id, reaction_activity + ) + else: + activy = reaction_activity + await self.signAndPost(inbox, publisher_actor_id, activy) + if service.user and self.isVirtualJID(service): # the item is on a virtual service, we need to store it in cache log.debug("storing attachments item in cache") @@ -1262,9 +1292,18 @@ class APGateway: href = await self.getAPActorIdFromAccount(account) return await self.apGet(href) - async def getAPInboxFromId(self, actor_id: str) -> str: - """Retrieve inbox of an actor_id""" + async def getAPInboxFromId(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.getActorData(actor_id) + if use_shared: + try: + return data["endpoints"]["sharedInbox"] + except KeyError: + pass return data["inbox"] @async_lru(maxsize=LRU_MAX_SIZE) @@ -2039,7 +2078,7 @@ class APGateway: actor_account = self._e.unescape(mess_data["to"].user) actor_id = await self.getAPActorIdFromAccount(actor_account) - inbox = await self.getAPInboxFromId(actor_id) + inbox = await self.getAPInboxFromId(actor_id, use_shared=False) try: language, message = next(iter(mess_data["message"].items())) @@ -2085,7 +2124,7 @@ class APGateway: ) ap_account = self._e.unescape(to_jid.user) actor_id = await self.getAPActorIdFromAccount(ap_account) - inbox = await self.getAPInboxFromId(actor_id) + inbox = await self.getAPInboxFromId(actor_id, use_shared=False) url_actor, ap_item = await self.apDeleteItem( from_jid.userhostJID(), None, fastened_elts.id, public=False ) @@ -2175,7 +2214,7 @@ class APGateway: "name": ap_account, }) - inbox = await self.getAPInboxFromId(actor_id) + inbox = await self.getAPInboxFromId(actor_id, use_shared=False) resp = await self.signAndPost(inbox, ap_item["actor"], ap_item) diff --git a/sat/plugins/plugin_comp_ap_gateway/constants.py b/sat/plugins/plugin_comp_ap_gateway/constants.py index 75705c7a..204f73f4 100644 --- a/sat/plugins/plugin_comp_ap_gateway/constants.py +++ b/sat/plugins/plugin_comp_ap_gateway/constants.py @@ -30,6 +30,7 @@ TYPE_ITEM = "item" TYPE_TOMBSTONE = "Tombstone" TYPE_MENTION = "Mention" TYPE_LIKE = "Like" +TYPE_REACTION = "EmojiReact" MEDIA_TYPE_AP = "application/activity+json" NS_AP = "https://www.w3.org/ns/activitystreams" NS_AP_PUBLIC = f"{NS_AP}#Public" @@ -60,7 +61,9 @@ ACTIVITY_TYPES = ( "Accept", "Add", "Announce", "Arrive", "Block", "Create", "Delete", "Dislike", "Flag", "Follow", "Ignore", "Invite", "Join", "Leave", "Like", "Listen", "Move", "Offer", "Question", "Reject", "Read", "Remove", "TentativeReject", "TentativeAccept", - "Travel", "Undo", "Update", "View" + "Travel", "Undo", "Update", "View", + # non-standard activities + "EmojiReact" ) ACTIVITY_TYPES_LOWER = [a.lower() for a in ACTIVITY_TYPES] ACTIVITY_OBJECT_MANDATORY = ( @@ -69,7 +72,9 @@ ACTIVITY_OBJECT_MANDATORY = ( ACTIVITY_TARGET_MANDATORY = ("Add", "Remove") # activities which can be used with Shared Inbox (i.e. with no account specified) # must be lowercase -ACTIVIY_NO_ACCOUNT_ALLOWED = ("create", "delete", "announce", "undo") +ACTIVIY_NO_ACCOUNT_ALLOWED = ( + "create", "delete", "announce", "undo", "like", "emojireact" +) # maximum number of parents to retrieve when comments_max_depth option is set COMMENTS_MAX_PARENTS = 100 # maximum size of avatar, in bytes @@ -77,3 +82,4 @@ MAX_AVATAR_SIZE = 1024 * 1024 * 5 # storage prefixes ST_AVATAR = "[avatar]" +ST_AP_CACHE = "[AP_item_cache]" diff --git a/sat/plugins/plugin_comp_ap_gateway/http_server.py b/sat/plugins/plugin_comp_ap_gateway/http_server.py index 0919b49b..34b0bf74 100644 --- a/sat/plugins/plugin_comp_ap_gateway/http_server.py +++ b/sat/plugins/plugin_comp_ap_gateway/http_server.py @@ -42,7 +42,8 @@ from sat.memory.sqla_mapping import SubscriptionState from .constants import ( NS_AP, CONTENT_TYPE_AP, TYPE_ACTOR, TYPE_INBOX, TYPE_SHARED_INBOX, TYPE_OUTBOX, AP_REQUEST_TYPES, PAGE_SIZE, ACTIVITY_TYPES_LOWER, ACTIVIY_NO_ACCOUNT_ALLOWED, - SIGN_HEADERS, HS2019, SIGN_EXP, TYPE_FOLLOWERS, TYPE_FOLLOWING, TYPE_ITEM, TYPE_LIKE + SIGN_HEADERS, HS2019, SIGN_EXP, TYPE_FOLLOWERS, TYPE_FOLLOWING, TYPE_ITEM, TYPE_LIKE, + TYPE_REACTION, ST_AP_CACHE ) from .regex import RE_SIG_PARAM @@ -124,7 +125,19 @@ class HTTPAPGServer(web_resource.Resource): if node is None: node = self.apg._m.namespace client = await self.apg.getVirtualClient(signing_actor) - objects = await self.apg.apGetList(data, "object") + object_ = data.get("object") + if isinstance(object_, str): + # we check first if it's not a cached object + ap_cache_key = f"{ST_AP_CACHE}{object_}" + value = await self.apg.client._ap_storage.get(ap_cache_key) + else: + value = None + if value is not None: + objects = [value] + # because we'll undo the activity, we can remove it from cache + await self.apg.client._ap_storage.remove(ap_cache_key) + else: + objects = await self.apg.apGetList(data, "object") for obj in objects: type_ = obj.get("type") actor = await self.apg.apGetSenderActor(obj) @@ -152,7 +165,11 @@ class HTTPAPGServer(web_resource.Resource): # needed await self.apg.newAPDeleteItem(client, None, node, obj) elif type_ == TYPE_LIKE: - await self.handleNewLikeItem(client, obj, True) + await self.handleAttachmentItem(client, obj, {"noticed": False}) + elif type_ == TYPE_REACTION: + await self.handleAttachmentItem(client, obj, { + "reactions": {"operation": "update", "remove": [obj["content"]]} + }) else: log.warning(f"Unmanaged undo type: {type_!r}") @@ -184,7 +201,7 @@ class HTTPAPGServer(web_resource.Resource): if subscription.state != "subscribed": # other states should raise an Exception raise exceptions.InternalError('"subscribed" state was expected') - inbox = await self.apg.getAPInboxFromId(signing_actor) + inbox = await self.apg.getAPInboxFromId(signing_actor, use_shared=False) actor_id = self.apg.buildAPURL(TYPE_ACTOR, ap_account) accept_data = self.apg.createActivity( "Accept", actor_id, object_=data @@ -370,40 +387,47 @@ class HTTPAPGServer(web_resource.Resource): repeated=True ) - async def handleNewLikeItem( + async def handleAttachmentItem( self, client: SatXMPPEntity, data: dict, - undo: bool = False, + attachment_data: dict ) -> None: - liked_ids = data.get("object") - if not liked_ids: + target_ids = data.get("object") + if not target_ids: raise exceptions.DataError("object should be set") - elif isinstance(liked_ids, list): + elif isinstance(target_ids, list): try: - liked_ids = [o["id"] for o in liked_ids] + target_ids = [o["id"] for o in target_ids] except (KeyError, TypeError): - raise exceptions.DataError(f"invalid object: {liked_ids!r}") - elif isinstance(liked_ids, dict): - obj_id = liked_ids.get("id") + raise exceptions.DataError(f"invalid object: {target_ids!r}") + elif isinstance(target_ids, dict): + obj_id = target_ids.get("id") if not obj_id or not isinstance(obj_id, str): - raise exceptions.DataError(f"invalid object: {liked_ids!r}") - liked_ids = [obj_id] - elif isinstance(liked_ids, str): - liked_ids = [liked_ids] + raise exceptions.DataError(f"invalid object: {target_ids!r}") + target_ids = [obj_id] + elif isinstance(target_ids, str): + target_ids = [target_ids] - for liked_id in liked_ids: - if not self.apg.isLocalURL(liked_id): - log.debug(f"ignoring non local liked ID: {liked_id}") + # XXX: we have to cache AP items because some implementation (Pleroma notably) + # don't keep object accessible, and we need to be able to retrieve them for + # UNDO. Current implementation will grow, we need to add a way to flush it after + # a while. + # TODO: add a way to flush old cached AP items. + await client._ap_storage.aset(f"{ST_AP_CACHE}{data['id']}", data) + + for target_id in target_ids: + if not self.apg.isLocalURL(target_id): + log.debug(f"ignoring non local target ID: {target_id}") continue - url_type, url_args = self.apg.parseAPURL(liked_id) + url_type, url_args = self.apg.parseAPURL(target_id) if url_type != TYPE_ITEM: - log.warning(f"unexpected local URL for liked item: {liked_id}") + log.warning(f"unexpected local URL for attachment on item {target_id}") continue try: account, item_id = url_args except ValueError: - raise ValueError(f"invalid URL: {liked_id}") + raise ValueError(f"invalid URL: {target_id}") author_jid, item_node = await self.apg.getJIDAndNode(account) if item_node is None: item_node = self.apg._m.namespace @@ -418,7 +442,7 @@ class HTTPAPGServer(web_resource.Resource): create=True ) found_items, __ = await self.apg.host.memory.storage.getItems( - cached_node, item_ids=[item_id] + cached_node, item_ids=[client.jid.userhost()] ) if not found_items: old_item_elt = None @@ -428,9 +452,9 @@ class HTTPAPGServer(web_resource.Resource): item_elt = self.apg._pa.applySetHandler( client, - {"extra": {"noticed": not undo}}, + {"extra": attachment_data}, old_item_elt, - [("noticed", self.apg._pa.namespace)] + None ) # we reparse the element, as there can be other attachments attachments_data = self.apg._pa.items2attachmentData(client, [item_elt]) @@ -468,9 +492,24 @@ class HTTPAPGServer(web_resource.Resource): ap_account: Optional[str], ap_url: str, signing_actor: str - ): + ) -> None: client = await self.apg.getVirtualClient(signing_actor) - await self.handleNewLikeItem(client, data) + await self.handleAttachmentItem(client, data, {"noticed": True}) + + async def handleEmojireactActivity( + self, + request: "HTTPRequest", + data: dict, + account_jid: Optional[jid.JID], + node: Optional[str], + ap_account: Optional[str], + ap_url: str, + signing_actor: str + ) -> None: + client = await self.apg.getVirtualClient(signing_actor) + await self.handleAttachmentItem(client, data, { + "reactions": {"operation": "update", "add": [data["content"]]} + }) async def APActorRequest( self, diff --git a/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py b/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py index d010a959..f694fff6 100644 --- a/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py +++ b/sat/plugins/plugin_comp_ap_gateway/pubsub_service.py @@ -95,7 +95,7 @@ class APPubsubService(rsm.PubSubService): requestor_actor_id = self.apg.buildAPURL(TYPE_ACTOR, requestor.userhost()) recipient_account = self.apg._e.unescape(recipient.user) recipient_actor_id = await self.apg.getAPActorIdFromAccount(recipient_account) - inbox = await self.apg.getAPInboxFromId(recipient_actor_id) + inbox = await self.apg.getAPInboxFromId(recipient_actor_id, use_shared=False) return requestor_actor_id, recipient_actor_id, inbox diff --git a/sat/plugins/plugin_pubsub_attachments.py b/sat/plugins/plugin_xep_0470.py similarity index 87% rename from sat/plugins/plugin_pubsub_attachments.py rename to sat/plugins/plugin_xep_0470.py index 2b9f9a9c..9f48d9a8 100644 --- a/sat/plugins/plugin_pubsub_attachments.py +++ b/sat/plugins/plugin_xep_0470.py @@ -35,12 +35,12 @@ from sat.tools.utils import xmpp_date log = getLogger(__name__) -IMPORT_NAME = "PUBSUB_ATTACHMENTS" +IMPORT_NAME = "XEP-0470" PLUGIN_INFO = { C.PI_NAME: "Pubsub Attachments", C.PI_IMPORT_NAME: IMPORT_NAME, - C.PI_TYPE: C.PLUG_TYPE_EXP, + C.PI_TYPE: C.PLUG_TYPE_XEP, C.PI_MODES: C.PLUG_MODE_BOTH, C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: ["XEP-0060"], @@ -49,15 +49,15 @@ PLUGIN_INFO = { C.PI_DESCRIPTION: _("""Pubsub Attachments implementation"""), } NS_PREFIX = "urn:xmpp:pubsub-attachments:" -NS_PUBSUB_ATTACHMENTS = f"{NS_PREFIX}0" -NS_PUBSUB_ATTACHMENTS_SUM = f"{NS_PREFIX}summary:0" +NS_PUBSUB_ATTACHMENTS = f"{NS_PREFIX}1" +NS_PUBSUB_ATTACHMENTS_SUM = f"{NS_PREFIX}summary:1" class PubsubAttachments: namespace = NS_PUBSUB_ATTACHMENTS def __init__(self, host): - log.info(_("Pubsub Attachments plugin initialization")) + log.info(_("XEP-0470 (Pubsub Attachments) plugin initialization")) host.registerNamespace("pubsub-attachments", NS_PUBSUB_ATTACHMENTS) self.host = host self._p = host.plugins["XEP-0060"] @@ -67,7 +67,7 @@ class PubsubAttachments: "noticed", NS_PUBSUB_ATTACHMENTS, self.noticedGet, self.noticedSet ) self.registerAttachmentHandler( - "reaction", NS_PUBSUB_ATTACHMENTS, self.reactionGet, self.reactionSet + "reactions", NS_PUBSUB_ATTACHMENTS, self.reactionsGet, self.reactionsSet ) host.bridge.addMethod( "psAttachmentsGet", @@ -252,6 +252,22 @@ class PubsubAttachments: senders: Optional[List[jid.JID]], extra: Optional[dict] = None ) -> Tuple[List[Dict[str, Any]], dict]: + """Retrieve data attached to a pubsub item + + @param service: pubsub service where the node is + @param node: pubsub node containing the item + @param item: ID of the item for which attachments will be retrieved + @param senders: bare JIDs of entities that are checked. Attachments from those + entities will be retrieved. + If None, attachments from all entities will be retrieved + @param extra: extra data, will be used as ``extra`` argument when doing + ``getItems`` call. + @return: A tuple with: + - the list of attachments data, one item per found sender. The attachments + data are dict containing attachment, no ``extra`` field is used here + (contrarily to attachments data used with ``setAttachments``). + - metadata returned by the call to ``getItems`` + """ if extra is None: extra = {} attachment_node = self.getAttachmentNodeName(service, node, item) @@ -470,54 +486,58 @@ class PubsubAttachments: else: return None - def reactionGet( + def reactionsGet( self, client: SatXMPPEntity, attachments_elt: domish.Element, data: Dict[str, Any], ) -> None: try: - reaction_elt = next( - attachments_elt.elements(NS_PUBSUB_ATTACHMENTS, "reaction") + reactions_elt = next( + attachments_elt.elements(NS_PUBSUB_ATTACHMENTS, "reactions") ) except StopIteration: pass else: - reaction_data = { - "reactions": str(reaction_elt) - } - self.setTimestamp(reaction_elt, reaction_data) - data["reaction"] = reaction_data + reactions_data = {"reactions": []} + reactions = reactions_data["reactions"] + for reaction_elt in reactions_elt.elements(NS_PUBSUB_ATTACHMENTS, "reaction"): + reactions.append(str(reaction_elt)) + self.setTimestamp(reactions_elt, reactions_data) + data["reactions"] = reactions_data - def reactionSet( + def reactionsSet( self, client: SatXMPPEntity, data: Dict[str, Any], former_elt: Optional[domish.Element] ) -> Optional[domish.Element]: """update the attachment""" - reaction = data["extra"].get("reaction") - if reaction is None: + reactions_data = data["extra"].get("reactions") + if reactions_data is None: return former_elt - operation_type = reaction.get("operation", "update") + operation_type = reactions_data.get("operation", "update") if operation_type == "update": - reactions = "".join( - set(str(former_elt or "")) - | set(reaction.get("reactions") or "") - ) + former_reactions = { + str(r) for r in former_elt.elements(NS_PUBSUB_ATTACHMENTS, "reaction") + } if former_elt is not None else set() + added_reactions = set(reactions_data.get("add") or []) + removed_reactions = set(reactions_data.get("remove") or []) + reactions = list((former_reactions | added_reactions) - removed_reactions) elif operation_type == "replace": - reactions = reaction.get("reactions", "") + reactions = reactions_data.get("reactions") or [] else: raise exceptions.DataError(f"invalid reaction operation: {operation_type!r}") if reactions: - reaction_elt = domish.Element( - (NS_PUBSUB_ATTACHMENTS, "reaction"), + reactions_elt = domish.Element( + (NS_PUBSUB_ATTACHMENTS, "reactions"), attribs = { "timestamp": xmpp_date() } ) - reaction_elt.addContent(reactions) - return reaction_elt + for reactions_data in reactions: + reactions_elt.addElement("reaction", content=reactions_data) + return reactions_elt else: return None