From 5038fdb8f0d56f6774523480bc672680cc9ccb20 Mon Sep 17 00:00:00 2001 From: Goffi Date: Wed, 8 Apr 2026 16:18:40 +0200 Subject: [PATCH] component pubsub: Implement subscription handling methods: rel 476 --- .../plugins/plugin_comp_pubsub/__init__.py | 325 +++++++++++++++++- 1 file changed, 311 insertions(+), 14 deletions(-) diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index df2a0473..d3dee5bd 100644 --- a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py +++ b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py @@ -40,7 +40,7 @@ from libervia.backend.memory.sqla_mapping import ( Affiliation, PublishModel, PubsubAffiliation, - PubsubNode, + SubscriptionState, ) from libervia.backend.plugins.plugin_xep_0167 import NS_AV_CONFERENCES from libervia.backend.plugins.plugin_xep_0060 import NodeMetadata, XEP_0060 @@ -297,20 +297,174 @@ class PubsubResource(pubsub.PubSubResource): ) -> defer.Deferred[pubsub.Subscription]: return defer.ensureDeferred(self._subscribe(request)) - async def _subscribe(self, request: rsm.PubSubRequest) -> None: - return + async def _subscribe(self, request: pubsub.PubSubRequest) -> pubsub.Subscription: + """Handle a subscribe request. - def unsubscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + @param request: The subscribe request containing node identifier, subscriber JID, + and optional subscription options. + @return: A Subscription object representing the resulting subscription state. + @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor lacks subscribe privileges + @raise StanzaError: with "not-authorized" if there's a pending subscription + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + assert request.nodeIdentifier is not None + assert request.subscriber is not None + subscriber_jid = request.subscriber.userhostJID() + + if subscriber_jid != requestor_jid: + raise error.StanzaError( + "forbidden", text="Cannot subscribe on behalf of another entity" + ) + + 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 == subscriber_jid + ), + None, + ) + + if affiliation == Affiliation.outcast: + raise error.StanzaError("forbidden", text="You can't subscribe this node") + + 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( + "not-authorized", text="You can't subscribe to this node" + ) + 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}", + ) + + # Check for existing pending subscription + existing_subs = await G.storage.get_pubsub_subscriptions( + client, + node=node, + entity=subscriber_jid, + ) + existing_sub = existing_subs[0] if existing_subs else None + + if existing_sub and existing_sub.state == SubscriptionState.PENDING: + raise error.StanzaError( + "not-authorized", text="You already have a pending subscription" + ) + + # For now, we auto-approve subscriptions + # TODO: Implement pending subscription approval workflow + await G.storage.set_pubsub_subscriptions( + node, + {subscriber_jid: SubscriptionState.SUBSCRIBED}, + ) + + return pubsub.Subscription( + nodeIdentifier=request.nodeIdentifier, + subscriber=request.subscriber, + state="subscribed", + ) + + def unsubscribe(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._unsubscribe(request)) - async def _unsubscribe(self, request: rsm.PubSubRequest) -> None: - return + async def _unsubscribe(self, request: pubsub.PubSubRequest) -> None: + """Handle an unsubscribe request. - def subscriptions(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + @param request: The unsubscribe request + @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor is not the subscriber + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + assert request.nodeIdentifier is not None + assert request.subscriber is not None + subscriber_jid = request.subscriber.userhostJID() + + # Check that the requestor matches the subscriber + if subscriber_jid != requestor_jid: + raise error.StanzaError( + "forbidden", text="Cannot unsubscribe on behalf of another entity" + ) + + node = await G.storage.get_pubsub_node(client, client.jid, request.nodeIdentifier) + + if node is None: + raise error.StanzaError("item_not_found", text="Node does not exist") + + await G.storage.set_pubsub_subscriptions( + node, + {subscriber_jid: None}, + ) + + def subscriptions( + self, request: pubsub.PubSubRequest + ) -> defer.Deferred[list[pubsub.Subscription]]: return defer.ensureDeferred(self._subscriptions(request)) - async def _subscriptions(self, request: rsm.PubSubRequest) -> None: - return + async def _subscriptions( + self, request: pubsub.PubSubRequest + ) -> list[pubsub.Subscription]: + """Retrieve the subscriptions of the requestor for one or all nodes. + + @param request: The subscriptions request. + If nodeIdentifier is set, return subscription for that specific node. + If nodeIdentifier is None, return subscriptions for all nodes. + @return: A list of Subscription objects. + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + subscriptions_data = await G.storage.get_pubsub_subscriptions( + client, + service=client.jid, + entity=requestor_jid, + node_name=request.nodeIdentifier, + ) + + # Convert to Subscription objects + subscriptions = [] + for pubsub_sub in subscriptions_data: + state_str = ( + "subscribed" + if pubsub_sub.state == SubscriptionState.SUBSCRIBED + else "pending" + ) + subscriptions.append( + pubsub.Subscription( + nodeIdentifier=pubsub_sub.node.name, + subscriber=pubsub_sub.subscriber, + state=state_str, + ) + ) + + return subscriptions def affiliations( self, request: pubsub.PubSubRequest @@ -760,14 +914,157 @@ class PubsubResource(pubsub.PubSubResource): ) -> defer.Deferred[list[pubsub.Subscription]]: return defer.ensureDeferred(self._subscriptionsGet(request)) - async def _subscriptionsGet(self, request: rsm.PubSubRequest) -> None: - return + async def _subscriptionsGet( + self, request: pubsub.PubSubRequest + ) -> list[pubsub.Subscription]: + """Retrieve all subscriptions for a specific node. - def subscriptionsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + @param request: The subscriptions get request with node identifier. + @return: List of Subscription objects for all subscribers of the node. + @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor is not an owner or admin + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + assert request.nodeIdentifier is not None + + node = await G.storage.get_pubsub_node( + client, client.jid, request.nodeIdentifier, with_subscriptions=True, with_affiliations=True + ) + + if node is None: + raise error.StanzaError("item_not_found", text="Node does not exist") + + if not G.host.memory.is_admin_jid(requestor_jid): + affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == requestor_jid + ), + None, + ) + + if affiliation != Affiliation.owner: + raise error.StanzaError("forbidden", text="Insufficient privileges") + + # Convert subscriptions to Subscription objects + subscriptions = [] + for sub in node.subscriptions: + state_str = ( + "subscribed" if sub.state == SubscriptionState.SUBSCRIBED else "pending" + ) + subscriptions.append( + pubsub.Subscription( + nodeIdentifier=request.nodeIdentifier, + subscriber=sub.subscriber, + state=state_str, + ) + ) + + return subscriptions + + def subscriptionsSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._subscriptionsSet(request)) - async def _subscriptionsSet(self, request: rsm.PubSubRequest) -> None: - return + async def _subscriptionsSet(self, request: pubsub.PubSubRequest) -> None: + """Set subscriptions for entities on a specific node. + + This allows node owners to manage subscriptions for entities. + Owner can approve pending subscriptions or set entities as subscribed. + + @param request: The subscriptions set request with node identifier and + subscriptions dict. + @raise StanzaError: with "item_not_found" if the node does not exist + @raise StanzaError: with "forbidden" if the requestor is not an owner or admin + @raise StanzaError: with "not_acceptable" if subscription values are invalid + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + 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") + + if not G.host.memory.is_admin_jid(requestor_jid): + affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == requestor_jid + ), + None, + ) + + if affiliation != Affiliation.owner: + raise error.StanzaError("forbidden", text="Insufficient privileges") + + if request.subscriptions is None: + return + + # Validate subscriptions from the request and convert them to the format expected + # by "set_pubsub_subscriptions" + subscriptions_to_set: dict[jid.JID, SubscriptionState | None] = {} + for subscription in request.subscriptions: + assert subscription.subscriber is not None + assert subscription.state is not None + entity = subscription.subscriber + subscription_state = subscription.state + + match subscription_state: + case "none": + subscriptions_to_set[entity] = None + case "subscribed": + entity_affiliation = next( + ( + aff_data.affiliation + for aff_data in node.affiliations + if aff_data.entity == entity + ), + None, + ) + + if entity_affiliation == Affiliation.outcast: + raise error.StanzaError( + "not-acceptable", + text=f"Cannot subscribe outcast entity {entity!r}", + ) + + access_model = node.access_model or AccessModel.open + if access_model == AccessModel.whitelist: + if entity_affiliation not in ( + Affiliation.owner, + Affiliation.publisher, + Affiliation.member, + ): + raise error.StanzaError( + "not-acceptable", + text=f"Entity {entity!r} is not on the whitelist", + ) + + subscriptions_to_set[entity] = SubscriptionState.SUBSCRIBED + case "pending": + subscriptions_to_set[entity] = SubscriptionState.PENDING + case "unconfigured": + # FIXME: unconfigured is treated as subscribed for now + subscriptions_to_set[entity] = SubscriptionState.SUBSCRIBED + case _: + raise error.StanzaError( + "not-acceptable", + text=f"Invalid subscription state: {subscription_state!r}", + ) + + await G.storage.set_pubsub_subscriptions( + node, subscriptions_to_set + ) class PubsubService(rsm.PubSubService):