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