diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index 11339224..7f85d46c 100644 --- a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py +++ b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py @@ -46,6 +46,7 @@ from libervia.backend.plugins.plugin_xep_0167 import NS_AV_CONFERENCES from libervia.backend.plugins.plugin_xep_0060 import NodeMetadata, XEP_0060 from libervia.backend.tools.config import ComponentSettings from libervia.backend.tools.utils import ensure_deferred +from libervia.backend.tools.xml_tools import element_copy log = getLogger(__name__) @@ -229,7 +230,69 @@ class PubsubResource(pubsub.PubSubResource): return defer.ensureDeferred(self._publish(request)) async def _publish(self, request: rsm.PubSubRequest) -> None: - return + """Publish items to a node. + + @raise StanzaError: with "bad_request" if no items are provided + @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor lacks publish privileges + @raise StanzaError: with "not_allowed" if item format is invalid + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + assert request.nodeIdentifier is not None + + if request.items is None or len(request.items) == 0: + raise error.StanzaError("bad_request", text="no item to publish") + + node = await G.storage.get_pubsub_node( + client, client.jid, request.nodeIdentifier, with_affiliations=True + ) + + if node is None: + raise error.StanzaError("item_not_found", text="Node does not exist") + + affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == requestor_jid + ), + None, + ) + + publish_model = node.publish_model or PublishModel.publishers + + if publish_model == PublishModel.publishers: + if affiliation not in (Affiliation.owner, Affiliation.publisher): + raise error.StanzaError("forbidden", text="Insufficient privileges") + elif publish_model == PublishModel.subscribers: + if affiliation not in (Affiliation.owner, Affiliation.publisher): + # FIXME + raise NotImplementedError("FIXME: need to handle subscribers") + # is_subscribed = any( + # sub.subscriber == requestor_jid and sub.state is not None + # for sub in node.subscriptions + # ) + # if not is_subscribed: + # raise error.StanzaError("forbidden", text="Insufficient privileges") + + items_data = [] + for item in request.items: + if item.name != "item" or item.uri != pubsub.NS_PUBSUB: + raise error.StanzaError( + "not_allowed", text=f"Invalid item: {item.toXml()}" + ) + + item["publisher"] = request.sender.full() + + if not item.getAttribute("id"): + item["id"] = f"{shortuuid.uuid()}" + + items_data.append(item) + + await G.storage.cache_pubsub_items(client, node, items_data) def subscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._subscribe(request)) @@ -470,13 +533,74 @@ class PubsubResource(pubsub.PubSubResource): async def _items( self, request: rsm.PubSubRequest ) -> tuple[list[domish.Element], rsm.RSMResponse | None]: + """Retrieve items from a pubsub node. + + @param request: The items request containing node identifier, item identifiers, + max items limit, and optional RSM parameters. + @return: A tuple of (list of item elements, RSM response or None). + @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor lacks retrieve privileges + """ client = self.client assert request.sender is not None requestor_jid = request.sender.userhostJID() - if not client.is_local(requestor_jid): - raise error.StanzaError("forbidden") - return [], None + assert request.nodeIdentifier is not None + + node = await G.storage.get_pubsub_node( + client, client.jid, request.nodeIdentifier, with_affiliations=True + ) + + if node is None: + raise error.StanzaError("item_not_found", text="Node does not exist") + + affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == requestor_jid + ), + None, + ) + + access_model = node.access_model or AccessModel.open + + if access_model == AccessModel.open: + pass + elif access_model == AccessModel.whitelist: + if affiliation not in ( + Affiliation.owner, + Affiliation.publisher, + Affiliation.member, + ): + raise error.StanzaError("forbidden", text="Insufficient privileges") + else: + log.error( + f"Unknown access model {access_model!r} for node" + f" {request.nodeIdentifier!r}" + ) + raise error.StanzaError( + "internal-server-error", + text=f"Unknown access model {access_model!r}", + ) + + items_data, metadata = await G.storage.get_items( + node=node, + max_items=request.rsm.max if request.rsm else request.maxItems, + item_ids=request.itemIdentifiers or None, + before=request.rsm.before if request.rsm else None, + after=request.rsm.after if request.rsm else None, + from_index=request.rsm.index if request.rsm else None, + desc=request.rsm is None, + force_rsm=request.rsm is not None, + ) + + if "rsm" in metadata: + rsm_response = rsm.RSMResponse(**metadata["rsm"]) + else: + rsm_response = None + + return [item.data for item in items_data], rsm_response def retract(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._retract(request))