libervia-backend: sat/plugins/plugin_comp_ap_gateway/__init_

comparison sat/plugins/plugin_comp_ap_gateway/init.py @ 3804:36b167ddbfca

component AP gateway: AP delete activity => message retract: handle retractation of messages. As it is not possible to know from the AP item alone if we need to to a message retractation (XEP-0424) or a pubsub retractation (XEP-0060), we now cache sent message, and decide which method to use according to how the item is cached (i.e. in message history or in pubsub cache). rel 367

author	Goffi <goffi@goffi.org>
date	Fri, 17 Jun 2022 14:15:23 +0200
parents	d5f343939239
children	2032826cfbcf

comparison

equal deleted inserted replaced

-:d5f343939239
+:36b167ddbfca
 from sat.core import exceptions
 from sat.core.constants import Const as C
 from sat.core.core_types import SatXMPPEntity
 from sat.core.i18n import _
 from sat.core.log import getLogger
-from sat.memory.sqla_mapping import SubscriptionState
+from sat.memory.sqla_mapping import SubscriptionState, History
 from sat.tools import utils
 from sat.tools.common import data_format, tls, uri
 from sat.tools.common.async_utils import async_lru
 from .constants import (
 context_factory = tls.getTLSContextFactory(options)
 reactor.listenSSL(self.http_port, self.server, context_factory)
 async def profileConnecting(self, client):
 self.client = client
+client.sendHistory = True
 await self.init(client)
 async def _itemsReceived(self, client, itemsEvent):
 """Callback called when pubsub items are received
 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.getJIDAndNode(parent_item_account)
+parent_item_service, parent_item_node = await self.getJIDAndNode(
+parent_item_account
+)
 if parent_item_node is None:
 parent_item_node = self._m.namespace
 items, __ = await self._p.getItems(
 client, parent_item_service, parent_item_node, item_ids=[parent_item_id]
 )
 raise NotImplemented()
 else:
 __, item_elt = await self.apItem2MbDataAndElt(ap_item)
 await self._p.publish(client, comment_service, comment_node, [item_elt])
-async def newAPItem(
+def getAPItemTargets(self, item: Dict[str, Any]) -> Tuple[bool, Set[str], Set[str]]:
-self,
+"""Retrieve targets of an AP item, and indicate if it's a public one
-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
+@return: Are returned:
+- is_public flag, indicating if the item is world-readable
+- targets of the item
+- targets of the items
 """
 targets: Set[str] = set()
 is_public = False
 # TODO: handle "audience"
 for key in ("to", "bto", "cc", "bcc"):
 if not self.isLocalURL(value):
 continue
 targets.add(value)
 targets_types = {self.parseAPURL(t)[0] for t in targets}
+return is_public, targets, targets_types
+async def newAPItem(
+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, targets_types = self.getAPItemTargets(item)
 if not is_public and targets_types == {TYPE_ACTOR}:
 # this is a direct message
 await self.handleMessageAPItem(
 client, targets, destinee, item
 )
 defer_l.append(
 client.sendMessage(
 target_jid,
 {'': mb_data.get("content", "")},
 mb_data.get("title"),
+extra={"origin_id": mb_data["id"]}
 )
 )
 await defer.DeferredList(defer_l)
 async def handlePubsubAPItem(
 async def newAPDeleteItem(
 self,
 client: SatXMPPEntity,
 destinee: Optional[jid.JID],
 node: str,
+activity: dict,
 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
 """
 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.isLocalURL(item_id):
 raise ValueError("Local IDs should not be used")
-cached_node = await self.host.memory.storage.getPubsubNode(
+# we have no way to know if a deleted item is a direct one (thus a message) or one
-client, client.jid, node, with_subscriptions=True
+# converted to pubsub. We check if the id is in message history to decide what to
-)
+# do.
-if cached_node is None:
+history = await self.host.memory.storage.get(
-log.warning(
+client,
-f"Received an item retract for node {node!r} at {client.jid} which is "
+History,
-"not cached"
+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"retractation 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.retractByHistory(client, history)
 else:
+# no history in cache with this ID, it's probably a pubsub item
+cached_node = await self.host.memory.storage.getPubsubNode(
+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.deletePubsubItems(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:

Mercurial > libervia-backend

comparison sat/plugins/plugin_comp_ap_gateway/__init__.py @ 3804:36b167ddbfca

comparison sat/plugins/plugin_comp_ap_gateway/init.py @ 3804:36b167ddbfca