From 3492baaa65a43886865cf5ca1e55856d310bc899 Mon Sep 17 00:00:00 2001 From: Goffi Date: Fri, 10 Apr 2026 23:05:42 +0200 Subject: [PATCH] component pubsub: integrate pubsub relationships: Pubsub relationships from XEP-0496 are now handled. They can be set in config as explained in the specification, and access are checked with parents when they are set. Deletion cascade as specified in the XEP. rel 476 --- .../plugins/plugin_comp_pubsub/__init__.py | 269 ++++++++++++++---- 1 file changed, 207 insertions(+), 62 deletions(-) diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index e0fcf78e..3f6c1b2c 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.memory.sqla_mapping import ( ) 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.plugins.plugin_xep_0496 import LINK_VAR, PARENT_VAR, XEP_0496 from libervia.backend.tools.config import ComponentSettings from libervia.backend.tools.xml_tools import element_copy from tx_xmpp.pubsub import NS_PUBSUB_EVENT @@ -61,7 +62,7 @@ PLUGIN_INFO = { C.PI_MODES: [C.PLUG_MODE_COMPONENT], C.PI_TYPE: C.PLUG_TYPE_ENTRY_POINT, C.PI_PROTOCOLS: [], - C.PI_DEPENDENCIES: ["XEP-0060"], + C.PI_DEPENDENCIES: ["XEP-0060", "XEP-0496"], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "PubsubComponent", C.PI_HANDLER: C.BOOL_TRUE, @@ -89,6 +90,8 @@ class PubsubNodeConfig: description: str | None = field( default=None, metadata={"label": "A description of the node"} ) + parent: str | None = field(default=None, metadata={"label": "Parent node identifier"}) + link: str | None = field(default=None, metadata={"label": "Linked node identifier"}) _field_var_mapping = { "access_model": "pubsub#access_model", @@ -96,6 +99,8 @@ class PubsubNodeConfig: "max_items": "pubsub#max_items", "title": "pubsub#title", "description": "pubsub#description", + "parent": PARENT_VAR, + "link": LINK_VAR, } @classmethod @@ -109,8 +114,8 @@ class PubsubNodeConfig: """ definitions: dict = {} - for field in fields(cls): - field_name = field.name + for field_ in fields(cls): + field_name = field_.name try: pubsub_var = cls._field_var_mapping[field_name] @@ -118,7 +123,7 @@ class PubsubNodeConfig: log.error("Missing field mapping for {field_name!r}, ignoring.") continue - label = field.metadata.get("label", pubsub_var) + label = field_.metadata.get("label", pubsub_var) match field_name: case "access_model" | "publish_model": @@ -169,6 +174,7 @@ class PubsubResource(pubsub.PubSubResource): def __init__(self, service: "PubsubService") -> None: super().__init__() self.service = service + self.xep_0496 = cast(XEP_0496, service.client.host.plugins.get("XEP-0496")) @property def client(self) -> SatXMPPComponent: @@ -286,29 +292,11 @@ class PubsubResource(pubsub.PubSubResource): 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, + has_publish_rights = await self.xep_0496.has_publish_rights( + self.client, self.client.jid, node, requestor_jid ) - - 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): - is_subscribed = any( - sub.subscriber == requestor_jid - and sub.state == SubscriptionState.SUBSCRIBED - for sub in node.subscriptions - ) - if not is_subscribed: - raise error.StanzaError("forbidden", text="Insufficient privileges") + if not has_publish_rights: + raise error.StanzaError("forbidden", text="Insufficient privileges") ret_pubsub_elt = domish.Element((NS_PUBSUB_EVENT, "pubsub")) ret_publish_elt = ret_pubsub_elt.addElement("publish") @@ -372,6 +360,14 @@ class PubsubResource(pubsub.PubSubResource): if node is None: raise error.StanzaError("item-not-found", text="Node does not exist") + has_access = await self.xep_0496.has_access( + self.client, self.client.jid, node, subscriber_jid + ) + if not has_access: + raise error.StanzaError( + "forbidden", text="Insufficient privileges to subscribe to this node" + ) + affiliation = next( ( aff_data.affiliation @@ -571,13 +567,17 @@ class PubsubResource(pubsub.PubSubResource): except ValueError: pass + # Store relationship fields and normalize empty strings to None + config.parent = request.options.get(PARENT_VAR) or None + config.link = request.options.get(LINK_VAR) or None + owner_affiliation = PubsubAffiliation( entity=requestor_jid, affiliation=Affiliation.owner, ) try: - await G.storage.set_pubsub_node( + node = await G.storage.set_pubsub_node( client=client, service=client.jid, name=node_id, @@ -588,6 +588,35 @@ class PubsubResource(pubsub.PubSubResource): except exceptions.ConflictError: raise error.StanzaError("conflict", text=f"node {node_id!r} already exists.") + if config.parent is not None or config.link is not None: + try: + parent_node, linked_node = await self.xep_0496.check_relationships( + self.client, + self.client.jid, + node, + requestor_jid, + config.parent, + config.link, + ) + except Exception as e: + raise error.StanzaError( + "not-allowed", + appCondition="invalid-option", + text=str(e), + ) + + update_values = {} + + # Set parent if explicitly provided or auto-set from linked node + if parent_node: + update_values["parent_node_id"] = parent_node.id + + if linked_node: + update_values["linked_node_id"] = linked_node.id + + if update_values: + await G.storage.update_pubsub_node(node, update_values) + return node_id def default(self, request: pubsub.PubSubRequest) -> defer.Deferred[dict[str, object]]: @@ -641,7 +670,12 @@ class PubsubResource(pubsub.PubSubResource): assert request.nodeIdentifier is not None node = await G.storage.get_pubsub_node( - client, client.jid, request.nodeIdentifier, with_affiliations=True + client, + client.jid, + request.nodeIdentifier, + with_affiliations=True, + with_parent=True, + with_link=True, ) if node is None: @@ -682,6 +716,12 @@ class PubsubResource(pubsub.PubSubResource): # "max" if node.max_items == -1 else str(node.max_items) # ) + if node.linked_node is not None: + config_values[LINK_VAR] = node.linked_node.name + + if node.parent_node: + config_values[PARENT_VAR] = node.parent_node.name + return config_values def configureSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]: @@ -701,12 +741,25 @@ class PubsubResource(pubsub.PubSubResource): assert request.nodeIdentifier is not None node = await G.storage.get_pubsub_node( - client, client.jid, request.nodeIdentifier, with_affiliations=True + client, + client.jid, + request.nodeIdentifier, + with_affiliations=True, + with_parent=True, + with_link=True, ) if node is None: raise error.StanzaError("item-not-found", text="Node does not exist") + has_access = await self.xep_0496.has_access( + self.client, self.client.jid, node, requestor_jid + ) + if not has_access: + raise error.StanzaError( + "forbidden", text="Insufficient privileges to configure this node" + ) + # Only admins and owner can modifify configuration. if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( @@ -766,6 +819,62 @@ class PubsubResource(pubsub.PubSubResource): if update_extra: update_values["extra"] = extra + # Handle relationship + current_parent = node.parent_node.name if node.parent_node else None + current_link = node.linked_node.name if node.linked_node else None + + # Field present in form = user wants to change it; field absent = no change + parent_in_form = PARENT_VAR in request.options + link_in_form = LINK_VAR in request.options + + new_parent = request.options.get(PARENT_VAR) + new_link = request.options.get(LINK_VAR) + + parent_changed = parent_in_form and new_parent != current_parent + link_changed = link_in_form and new_link != current_link + + if parent_changed or link_changed: + try: + new_parent_node, new_linked_node = ( + await self.xep_0496.check_relationships( + self.client, + self.client.jid, + node, + requestor_jid, + new_parent, + new_link, + ) + ) + except Exception as e: + raise error.StanzaError( + "not-allowed", + appCondition="invalid-option", + text=str(e), + ) + + if parent_changed: + if new_parent_node: + update_values["parent_node_id"] = new_parent_node.id + else: + # Clearing parent + update_values["parent_node_id"] = None + + await self.xep_0496.update_linking_nodes_parent( + node.id, new_parent_node.id if new_parent_node else None + ) + + if link_changed: + if new_linked_node: + update_values["linked_node_id"] = new_linked_node.id + else: + # Clearing link + update_values["linked_node_id"] = None + + # If link was set and check_relationships auto-set the parent, update it too + if link_changed and new_parent_node and not parent_changed: + # Parent was auto-set by check_relationships to match linked node's parent + update_values["parent_node_id"] = new_parent_node.id + if update_values: await G.storage.update_pubsub_node(node, update_values) @@ -778,7 +887,7 @@ class PubsubResource(pubsub.PubSubResource): return defer.ensureDeferred(self._items(request)) async def _items( - self, request: pubsub.PubSubRequest + self, request: rsm.PubSubRequest ) -> list[domish.Element] | tuple[list[domish.Element], rsm.RSMResponse]: """Retrieve items from a pubsub node. @@ -801,35 +910,11 @@ class PubsubResource(pubsub.PubSubResource): 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, + has_access = await self.xep_0496.has_access( + self.client, self.client.jid, node, requestor_jid ) - - 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}", - ) + if not has_access: + raise error.StanzaError("forbidden", text="Insufficient privileges") items_data, metadata = await G.storage.get_items( node=node, @@ -994,6 +1079,14 @@ class PubsubResource(pubsub.PubSubResource): if node is None: raise error.StanzaError("item-not-found", text="Node does not exist") + has_access = await self.xep_0496.has_access( + self.client, self.client.jid, node, requestor_jid + ) + if not has_access: + raise error.StanzaError( + "forbidden", text="Insufficient privileges to delete this node" + ) + if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( ( @@ -1007,10 +1100,46 @@ class PubsubResource(pubsub.PubSubResource): if affiliation != Affiliation.owner: raise error.StanzaError("forbidden", text="Insufficient privileges") - await self._send_notifications(node, "delete", []) - await G.storage.delete_pubsub_node( - profiles=None, services=[client.jid], names=[node.name] - ) + # We'll generate list of nodes to delete due to cascade deletion with + # relationships (cf. XEP-0496). + nodes_to_delete = [node.name] + + linking_nodes = await self.xep_0496.get_linking_nodes_by_id(node.id) + for linking_node in linking_nodes: + nodes_to_delete.append(linking_node.name) + + descendant_nodes = await self.xep_0496.get_descendant_nodes(node) + for descendant in descendant_nodes: + if descendant.name not in nodes_to_delete: + nodes_to_delete.append(descendant.name) + + # Build a map of nodes to delete (reuse already fetched nodes) + nodes_map: dict[str, PubsubNode] = {node.name: node} + nodes_map.update({str(ln.name): ln for ln in linking_nodes}) + nodes_map.update({str(dn.name): dn for dn in descendant_nodes}) + + # Fetch any missing nodes + for node_name in nodes_to_delete: + if node_name not in nodes_map: + fetched_node = await G.storage.get_pubsub_node( + client, + client.jid, + node_name, + with_affiliations=True, + with_subscriptions=True, + ) + if fetched_node: + nodes_map[node_name] = fetched_node + + for node_name in nodes_to_delete: + node_to_delete = nodes_map.get(node_name) + if node_to_delete: + await self._send_notifications(node_to_delete, "delete", []) + + if nodes_to_delete: + await G.storage.delete_pubsub_node( + profiles=None, services=[client.jid], names=nodes_to_delete + ) def affiliationsGet( self, request: pubsub.PubSubRequest @@ -1038,6 +1167,14 @@ class PubsubResource(pubsub.PubSubResource): if node is None: raise error.StanzaError("item-not-found", text="Node does not exist") + has_access = await self.xep_0496.has_access( + self.client, self.client.jid, node, requestor_jid + ) + if not has_access: + raise error.StanzaError( + "forbidden", text="Insufficient privileges to retrieve affiliations" + ) + if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( ( @@ -1081,6 +1218,14 @@ class PubsubResource(pubsub.PubSubResource): if node is None: raise error.StanzaError("item-not-found", text="Node does not exist") + has_access = await self.xep_0496.has_access( + self.client, self.client.jid, node, requestor_jid + ) + if not has_access: + raise error.StanzaError( + "forbidden", text="Insufficient privileges to manage affiliations" + ) + if not G.host.memory.is_admin_jid(requestor_jid): affiliation = next( (