From 34d6032cf5735f3301cb584b2e9e519a0a13a30c Mon Sep 17 00:00:00 2001 From: Goffi Date: Wed, 8 Apr 2026 10:22:10 +0200 Subject: [PATCH] component pubsub: Handle affiliations setting/retrieval: Let retrieve affiliations for the whole service or a specific node, and set affiliations for a node. rel 476 --- libervia/backend/memory/sqla.py | 106 +++++++++- .../plugins/plugin_comp_pubsub/__init__.py | 193 +++++++++++++++--- 2 files changed, 262 insertions(+), 37 deletions(-) diff --git a/libervia/backend/memory/sqla.py b/libervia/backend/memory/sqla.py index c2132f6b..19d07265 100644 --- a/libervia/backend/memory/sqla.py +++ b/libervia/backend/memory/sqla.py @@ -58,6 +58,7 @@ from libervia.backend.memory import migration from libervia.backend.memory import sqla_config from libervia.backend.memory.sqla_mapping import ( AccessModel, + Affiliation, Base, Component, File, @@ -1788,9 +1789,7 @@ class Storage: # item_ids with RSM doesn't make sense if item_ids is not None and use_rsm: - raise exceptions.BadRequest( - "item_ids cannot be used with RSM parameters" - ) + raise exceptions.BadRequest("item_ids cannot be used with RSM parameters") if use_rsm: return await self._get_items_rsm( @@ -1810,6 +1809,107 @@ class Storage: result = result.scalars().all() return result, metadata + @aio + async def get_pubsub_affiliations( + self, + client: SatXMPPEntity, + service: jid.JID | None = None, + entity: jid.JID | None = None, + node_name: str | None = None, + ) -> list[tuple[str, jid.JID, Affiliation]]: + """Retrieve affiliations from the pubsub service. + + @param client: Affiliation from this client profile will be retrieved. + @param service: If set, filter to affiliations for this pubsub service. + @param entity: If set, filter to affiliations for this entity. + @param node_name: If set, filter to affiliations for this node. + @return: List of (node_name, entity, affiliation) tuples. + """ + stmt = ( + select( + PubsubNode.name, PubsubAffiliation.entity, PubsubAffiliation.affiliation + ) + .join(PubsubNode, PubsubAffiliation.node_id == PubsubNode.id) + .where( + PubsubNode.profile_id == self.profiles[client.profile], + ) + ) + + if service is not None: + stmt = stmt.where(PubsubNode.service == service) + if entity is not None: + stmt = stmt.where(PubsubAffiliation.entity == entity) + if node_name is not None: + stmt = stmt.where(PubsubNode.name == node_name) + + async with self.session() as session: + result = await session.execute(stmt) + + return [(row.name, row.entity, row.affiliation) for row in result.all()] + + @aio + async def set_pubsub_affiliations( + self, + client: SatXMPPEntity, + service: jid.JID, + node_name: str, + affiliations: dict[jid.JID, Affiliation | None], + ) -> None: + """Set affiliations for a node. + + Uses upsert to insert new affiliations or update existing ones. + Affiliations with value None are removed. + + @param client: The profile from this client will be used. + @param service: The pubsub service JID. + @param node_name: The node whose affiliations to set. + @param affiliations: Dict mapping entity JIDs to their new affiliations. + If affiliation is None, the entity's affiliation is removed. + @raise exceptions.NotFound: Node doesn't exist. + """ + node = await as_future(self.get_pubsub_node(client, service, node_name)) + if node is None: + raise exceptions.NotFound(f"Node {node_name!r} not found") + + async with self.session() as session: + async with session.begin(): + to_set = { + entity: aff for entity, aff in affiliations.items() if aff is not None + } + to_delete = [ + entity for entity, aff in affiliations.items() if aff is None + ] + + # Upsert affiliations + if to_set: + for entity, affiliation in to_set.items(): + stmt = ( + insert(PubsubAffiliation) + .values( + node_id=node.id, + entity=entity, + affiliation=affiliation, + ) + .on_conflict_do_update( + index_elements=( + PubsubAffiliation.node_id, + PubsubAffiliation.entity, + ), + set_={PubsubAffiliation.affiliation: affiliation}, + ) + ) + await session.execute(stmt) + + # Delete affiliations marked as None + if to_delete: + delete_stmt = delete(PubsubAffiliation).where( + PubsubAffiliation.node_id == node.id, + PubsubAffiliation.entity.in_(to_delete), + ) + await session.execute(delete_stmt) + + await session.commit() + def _get_sqlite_path(self, path: List[Union[str, int]]) -> str: """generate path suitable to query JSON element with SQLite""" return f"${''.join(f'[{p}]' if isinstance(p, int) else f'.{p}' for p in path)}" diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index 7f85d46c..df2a0473 100644 --- a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py +++ b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py @@ -17,7 +17,6 @@ # along with this program. If not, see . from dataclasses import dataclass, field, fields -from enum import StrEnum, auto from typing import Literal, cast import shortuuid @@ -26,7 +25,7 @@ from twisted.internet import defer from twisted.words.protocols.jabber import error, jid from twisted.words.protocols.jabber.xmlstream import XMPPHandler from twisted.words.xish import domish -from tx_xmpp import data_form, disco, itx_xmpp +from tx_xmpp import disco, itx_xmpp from tx_xmpp import pubsub, rsm from zope.interface import implementer @@ -41,12 +40,11 @@ from libervia.backend.memory.sqla_mapping import ( Affiliation, PublishModel, PubsubAffiliation, + PubsubNode, ) 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__) @@ -226,10 +224,10 @@ class PubsubResource(pubsub.PubSubResource): """ return PubsubNodeConfig.get_field_definitions() - def publish(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def publish(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._publish(request)) - async def _publish(self, request: rsm.PubSubRequest) -> None: + async def _publish(self, request: pubsub.PubSubRequest) -> None: """Publish items to a node. @raise StanzaError: with "bad_request" if no items are provided @@ -294,7 +292,9 @@ class PubsubResource(pubsub.PubSubResource): await G.storage.cache_pubsub_items(client, node, items_data) - def subscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def subscribe( + self, request: pubsub.PubSubRequest + ) -> defer.Deferred[pubsub.Subscription]: return defer.ensureDeferred(self._subscribe(request)) async def _subscribe(self, request: rsm.PubSubRequest) -> None: @@ -312,16 +312,38 @@ class PubsubResource(pubsub.PubSubResource): async def _subscriptions(self, request: rsm.PubSubRequest) -> None: return - def affiliations(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def affiliations( + self, request: pubsub.PubSubRequest + ) -> defer.Deferred[list[tuple[str, str]]]: return defer.ensureDeferred(self._affiliations(request)) - async def _affiliations(self, request: rsm.PubSubRequest) -> None: - return + async def _affiliations(self, request: pubsub.PubSubRequest) -> list[tuple[str, str]]: + """Retrieve the affiliations of the requestor for one or all nodes. - def create(self, request: rsm.PubSubRequest) -> defer.Deferred[str]: + @param request: The affiliations request. + If nodeIdentifier is set, return affiliation for that specific node. + If nodeIdentifier is None, return affiliations for all nodes. + @return: A list of (nodeIdentifier, affiliation) tuples. + """ + client = self.client + assert request.sender is not None + requestor_jid = request.sender.userhostJID() + + affiliations = await G.storage.get_pubsub_affiliations( + client, + client.jid, + entity=requestor_jid, + node_name=request.nodeIdentifier, + ) + + return [ + (node_name, str(affiliation)) for node_name, _, affiliation in affiliations + ] + + def create(self, request: pubsub.PubSubRequest) -> defer.Deferred[str]: return defer.ensureDeferred(self._create(request)) - async def _create(self, request: rsm.PubSubRequest) -> str: + async def _create(self, request: pubsub.PubSubRequest) -> str: """Create a new pubsub node. @param request: Node creation request. @@ -372,16 +394,16 @@ class PubsubResource(pubsub.PubSubResource): return node_id - def default(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def default(self, request: pubsub.PubSubRequest) -> defer.Deferred[dict[str, object]]: return defer.ensureDeferred(self._default(request)) - async def _default(self, request: rsm.PubSubRequest) -> None: - return + async def _default(self, request: pubsub.PubSubRequest) -> dict[str, object]: + raise pubsub.Unsupported("retrieve-default") - def configureGet(self, request: rsm.PubSubRequest) -> defer.Deferred[dict]: + def configureGet(self, request: pubsub.PubSubRequest) -> defer.Deferred[dict]: return defer.ensureDeferred(self._configureGet(request)) - async def _configureGet(self, request: rsm.PubSubRequest) -> dict: + async def _configureGet(self, request: pubsub.PubSubRequest) -> dict: """Retrieve the configuration for a specific node. @param request: The configuration request. @@ -439,10 +461,10 @@ class PubsubResource(pubsub.PubSubResource): return config_values - def configureSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def configureSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._configureSet(request)) - async def _configureSet(self, request: rsm.PubSubRequest) -> None: + async def _configureSet(self, request: pubsub.PubSubRequest) -> None: """Set the configuration of a node. @raise StanzaError: with "item_not_found" if the node does not exist @@ -526,12 +548,12 @@ class PubsubResource(pubsub.PubSubResource): def items( self, - request: rsm.PubSubRequest, + request: pubsub.PubSubRequest, ) -> defer.Deferred[tuple[list[domish.Element], rsm.RSMResponse | None]]: return defer.ensureDeferred(self._items(request)) async def _items( - self, request: rsm.PubSubRequest + self, request: pubsub.PubSubRequest ) -> tuple[list[domish.Element], rsm.RSMResponse | None]: """Retrieve items from a pubsub node. @@ -602,37 +624,140 @@ class PubsubResource(pubsub.PubSubResource): return [item.data for item in items_data], rsm_response - def retract(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def retract(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._retract(request)) - async def _retract(self, request: rsm.PubSubRequest) -> None: + async def _retract(self, request: pubsub.PubSubRequest) -> None: return - def purge(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def purge(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._purge(request)) - async def _purge(self, request: rsm.PubSubRequest) -> None: + async def _purge(self, request: pubsub.PubSubRequest) -> None: return - def delete(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def delete(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._delete(request)) - async def _delete(self, request: rsm.PubSubRequest) -> None: + async def _delete(self, request: pubsub.PubSubRequest) -> None: return - def affiliationsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + def affiliationsGet( + self, request: pubsub.PubSubRequest + ) -> defer.Deferred[dict[jid.JID, str]]: return defer.ensureDeferred(self._affiliationsGet(request)) - async def _affiliationsGet(self, request: rsm.PubSubRequest) -> None: - return + async def _affiliationsGet(self, request: pubsub.PubSubRequest) -> dict[jid.JID, str]: + """Retrieve all affiliations for a specific node. - def affiliationsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + @param request: The affiliations get request with node identifier. + @return: Mapping of entity JIDs to their affiliations. + @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_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") + + return { + affiliation.entity: affiliation.affiliation + for affiliation in node.affiliations + } + + def affiliationsSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: return defer.ensureDeferred(self._affiliationsSet(request)) - async def _affiliationsSet(self, request: rsm.PubSubRequest) -> None: - return + async def _affiliationsSet(self, request: pubsub.PubSubRequest) -> None: + """Set affiliations for entities on a specific node. - def subscriptionsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + @param request: The affiliations set request with node identifier and affiliations + 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 affiliation 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") + + # Prevent owner from changing their own affiliation to non-owner + if request.affiliations is not None: + if requestor_jid in request.affiliations: + new_affiliation = request.affiliations[requestor_jid] + if new_affiliation != "owner": + raise error.StanzaError( + "forbidden", text="You can't change your own affiliation" + ) + + # Convert affiliation strings to format expected by "set_pubsub_affiliations" + affiliations_to_set: dict[jid.JID, Affiliation | None] = {} + if request.affiliations is not None: + for entity, affiliation_str in request.affiliations.items(): + if affiliation_str == "none": + # We set the affiliation to None to mark it for removal. + affiliations_to_set[entity] = None + else: + try: + affiliation = Affiliation(affiliation_str) + affiliations_to_set[entity] = affiliation + except ValueError: + raise error.StanzaError( + "not-acceptable", + text=f"Invalid affiliation value: {affiliation_str!r}", + ) + + if affiliations_to_set: + await G.storage.set_pubsub_affiliations( + client, client.jid, request.nodeIdentifier, affiliations_to_set + ) + + def subscriptionsGet( + self, request: pubsub.PubSubRequest + ) -> defer.Deferred[list[pubsub.Subscription]]: return defer.ensureDeferred(self._subscriptionsGet(request)) async def _subscriptionsGet(self, request: rsm.PubSubRequest) -> None: