diff --git a/libervia/backend/memory/sqla.py b/libervia/backend/memory/sqla.py index 6767fb92..08293355 100644 --- a/libervia/backend/memory/sqla.py +++ b/libervia/backend/memory/sqla.py @@ -1408,9 +1408,16 @@ class Storage: parent_node=parent_node, linked_node=linked_node, ) - async with self.session() as session: - async with session.begin(): - session.add(node) + try: + async with self.session() as session: + async with session.begin(): + session.add(node) + except IntegrityError as e: + if "unique" in str(e.orig).lower(): + raise exceptions.ConflictError( + f"Node {name!r} already exists at {service}" + ) from e + raise return node @aio @@ -1428,6 +1435,18 @@ class Storage: ) ) + @aio + async def update_pubsub_node( + self, node: PubsubNode, values: dict + ) -> None: + async with self.session() as session: + async with session.begin(): + await session.execute( + update(PubsubNode) + .filter_by(id=node.id) + .values(**values) + ) + @aio async def delete_pubsub_node( self, @@ -1462,10 +1481,17 @@ class Storage: self, client: SatXMPPEntity, node: PubsubNode, - items: List[domish.Element], - parsed_items: Optional[List[dict]] = None, + items: list[domish.Element], + parsed_items: list[dict] | None = None, ) -> None: - """Add items to database, using an upsert taking care of "updated" field""" + """Add items to database, using an upsert taking care of "updated" field + + @param client: the client instance + @param node: the pubsub node + @param items: list of domish.Element items to cache + @param parsed_items: list of parsed item, i.e.; dictionaries with parsed data that + will be stored as JSON. Can be used later to filter requests on JSON values. + """ if parsed_items is not None and len(items) != len(parsed_items): raise exceptions.InternalError( "parsed_items must have the same lenght as items" diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py new file mode 100644 index 00000000..11339224 --- /dev/null +++ b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py @@ -0,0 +1,564 @@ +#!/usr/bin/env python3 + +# Libervia Pubsub Component +# Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# You should have received a copy of the GNU Affero General Public License +# 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 + +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 pubsub, rsm +from zope.interface import implementer + +from libervia.backend import G +from libervia.backend.core import exceptions +from libervia.backend.core.constants import Const as C +from libervia.backend.core.core_types import SatXMPPComponent, SatXMPPEntity +from libervia.backend.core.i18n import _ +from libervia.backend.core.log import getLogger +from libervia.backend.memory.sqla_mapping import ( + AccessModel, + Affiliation, + PublishModel, + PubsubAffiliation, +) +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 + +log = getLogger(__name__) + +IMPORT_NAME = "pubsub" +NAME = "Libervia Pubsub" + +PLUGIN_INFO = { + C.PI_NAME: NAME, + C.PI_IMPORT_NAME: IMPORT_NAME, + 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_RECOMMENDATIONS: [], + C.PI_MAIN: "PubsubComponent", + C.PI_HANDLER: C.BOOL_TRUE, + C.PI_DESCRIPTION: _("Libervia advanced Pubsub service."), +} + + +@dataclass +class PubsubNodeConfig: + max_items: int | Literal["max"] = field( + default="max", metadata={"label": "Maximum number of items to persist"} + ) + access_model: AccessModel = field( + default=AccessModel.open, + metadata={"label": "Who may subscribe and retrieve items"}, + ) + publish_model: PublishModel = field( + default=PublishModel.publishers, metadata={"label": "The publisher model"} + ) + title: str | None = field( + default=None, metadata={"label": "A short name for the node"} + ) + description: str | None = field( + default=None, metadata={"label": "A description of the node"} + ) + + _field_var_mapping = { + "access_model": "pubsub#access_model", + "publish_model": "pubsub#publish_model", + "max_items": "pubsub#max_items", + "title": "pubsub#title", + "description": "pubsub#description", + } + + @classmethod + def get_field_definitions(cls) -> dict[str, dict]: + """Return field definitions for Tx-XMPP configuration forms. + + This method dynamically builds the schema by introspecting dataclass fields + and their metadata. + + @return: A dict mapping pubsub field vars to their definitions. + """ + definitions: dict = {} + + for field in fields(cls): + field_name = field.name + + try: + pubsub_var = cls._field_var_mapping[field_name] + except KeyError: + log.error("Missing field mapping for {field_name!r}, ignoring.") + continue + + label = field.metadata.get("label", pubsub_var) + + match field_name: + case "access_model" | "publish_model": + enum_type = ( + AccessModel if field_name == "access_model" else PublishModel + ) + definitions[pubsub_var] = { + "type": "list-single", + "label": label, + "options": {value.value: None for value in enum_type}, + } + case _: + definitions[pubsub_var] = { + "type": "text-single", + "label": label, + } + + return definitions + + +class PubsubSettings(ComponentSettings, plugin_info=PLUGIN_INFO): + pass + + +class PubsubResource(pubsub.PubSubResource): + + def __init__(self, service: "PubsubService") -> None: + super().__init__() + self.service = service + + @property + def client(self) -> SatXMPPComponent: + return self.service.client + + def getInfo( + self, + requestor: jid.JID, + service: jid.JID, + nodeIdentifier: str, + ) -> defer.Deferred[dict | None]: + return defer.ensureDeferred(self.get_info(requestor, service, nodeIdentifier)) + + async def get_info( + self, + requestor: jid.JID, + service: jid.JID, + nodeIdentifier: str, + ) -> dict | None: + """Get node information for disco#info. + + @param requestor: The entity the request originated from. + @param service: The publish-subscribe service entity. + @param nodeIdentifier: Identifier of the node to get info for. + @return: Node type and metadata, as expected by ``Tx-XMPP``. + """ + node = await G.storage.get_pubsub_node(self.client, service, nodeIdentifier) + + if node is None: + return None + + metadata = NodeMetadata( + access_model=str(node.access_model) if node.access_model else None, + publish_model=str(node.publish_model) if node.publish_model else None, + ) + + # We convert to the format expected by Tx-XMPP. + form = metadata.to_data_form() + metadata_list = [ + { + "var": field.var, + "type": field.fieldType, + "value": field.value, + "label": field.label, + } + for field in form.fieldList + ] + + return { + "type": "leaf", + "meta-data": metadata_list if metadata_list else None, + } + + def getNodes( + self, requestor: jid.JID, service: jid.JID, nodeIdentifier: str + ) -> defer.Deferred[list[str]]: + return defer.ensureDeferred(self._get_nodes(requestor, service, nodeIdentifier)) + + async def _get_nodes( + self, requestor: jid.JID, service: jid.JID, node_id: str + ) -> list[str]: + """Get list of child nodes for a given node (async implementation). + + @param requestor: The entity the request originated from. + @param service: The publish-subscribe service entity. + @param node_id: Identifier of the node to request the children for. + @return: A list of node identifiers. + """ + if node_id: + # We only return results for requests without a specific node + return [] + + nodes = await G.storage.get_pubsub_nodes(self.client, self.client.jid) + return [cast(str, node.name) for node in nodes] + + def getConfigurationOptions(self) -> dict[str, dict]: + """Return the available configuration options for pubsub nodes. + + @return: A map field vars to their definition in the format expected by Tx-XMPP. + """ + return PubsubNodeConfig.get_field_definitions() + + def publish(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._publish(request)) + + async def _publish(self, request: rsm.PubSubRequest) -> None: + return + + def subscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._subscribe(request)) + + async def _subscribe(self, request: rsm.PubSubRequest) -> None: + return + + def unsubscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._unsubscribe(request)) + + async def _unsubscribe(self, request: rsm.PubSubRequest) -> None: + return + + def subscriptions(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._subscriptions(request)) + + async def _subscriptions(self, request: rsm.PubSubRequest) -> None: + return + + def affiliations(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._affiliations(request)) + + async def _affiliations(self, request: rsm.PubSubRequest) -> None: + return + + def create(self, request: rsm.PubSubRequest) -> defer.Deferred[str]: + return defer.ensureDeferred(self._create(request)) + + async def _create(self, request: rsm.PubSubRequest) -> str: + """Create a new pubsub node. + + @param request: Node creation request. + @return: The identifier of the created node. + """ + 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") + + node_id = request.nodeIdentifier or f"generic/{shortuuid.uuid()}" + + config = PubsubNodeConfig() + + if request.options is not None: + access_model_value = request.options.get(XEP_0060.OPT_ACCESS_MODEL) + if access_model_value: + try: + config.access_model = AccessModel(access_model_value) + except ValueError: + pass + + publish_model_value = request.options.get(XEP_0060.OPT_PUBLISH_MODEL) + if publish_model_value: + try: + config.publish_model = PublishModel(publish_model_value) + except ValueError: + pass + + owner_affiliation = PubsubAffiliation( + entity=requestor_jid, + affiliation=Affiliation.owner, + ) + + try: + await G.storage.set_pubsub_node( + client=client, + service=client.jid, + name=node_id, + access_model=config.access_model, + publish_model=config.publish_model, + affiliations=[owner_affiliation], + ) + except exceptions.ConflictError: + raise error.StanzaError("conflict", text=f"node {node_id!r} already exists.") + + return node_id + + def default(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._default(request)) + + async def _default(self, request: rsm.PubSubRequest) -> None: + return + + def configureGet(self, request: rsm.PubSubRequest) -> defer.Deferred[dict]: + return defer.ensureDeferred(self._configureGet(request)) + + async def _configureGet(self, request: rsm.PubSubRequest) -> dict: + """Retrieve the configuration for a specific node. + + @param request: The configuration request. + @return: Current configuration values for 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_affiliations=True + ) + + if node is None: + raise error.StanzaError("item_not_found", text="Node does not exist") + + # Only admins and owner can get configuration. + 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, + ) + + # Only owners can configure nodes + if affiliation != Affiliation.owner: + raise error.StanzaError("forbidden", text="Insufficient privileges") + + config_values: dict[str, str | bool | list[str]] = {} + + if node.extra: + if "title" in node.extra: + config_values["pubsub#title"] = node.extra["title"] + if "description" in node.extra: + config_values["pubsub#description"] = node.extra["description"] + + if node.access_model: + config_values["pubsub#access_model"] = str(node.access_model) + + if node.publish_model: + config_values["pubsub#publish_model"] = str(node.publish_model) + + # TODO: Add max_items when it's stored in the database + # if node.max_items is not None: + # config_values["pubsub#max_items"] = ( + # "max" if node.max_items == -1 else str(node.max_items) + # ) + + return config_values + + def configureSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._configureSet(request)) + + async def _configureSet(self, request: rsm.PubSubRequest) -> None: + """Set the configuration of a 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 + @raise StanzaError: with "not_acceptable" if configuration 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") + + # Only admins and owner can modifify configuration. + 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.options is None: + return + + config = PubsubNodeConfig() + + access_model_value = request.options.get(XEP_0060.OPT_ACCESS_MODEL) + if access_model_value: + try: + config.access_model = AccessModel(access_model_value) + except ValueError: + raise error.StanzaError( + "not-acceptable", + text=f"Invalid access_model value: {access_model_value!r}", + ) + + publish_model_value = request.options.get(XEP_0060.OPT_PUBLISH_MODEL) + if publish_model_value: + try: + config.publish_model = PublishModel(publish_model_value) + except ValueError: + raise error.StanzaError( + "not-acceptable", + text=f"Invalid publish_model value: {publish_model_value!r}", + ) + + title_value = request.options.get("pubsub#title") + description_value = request.options.get("pubsub#description") + + extra = node.extra.copy() if node.extra else {} + update_extra = False + if title_value is not None: + extra["title"] = title_value + update_extra = True + if description_value is not None: + extra["description"] = description_value + update_extra = True + + update_values = {} + if config.access_model != node.access_model: + update_values["access_model"] = config.access_model + if config.publish_model != node.publish_model: + update_values["publish_model"] = config.publish_model + if update_extra: + update_values["extra"] = extra + + if update_values: + await G.storage.update_pubsub_node(node, update_values) + + def items( + self, + request: rsm.PubSubRequest, + ) -> defer.Deferred[tuple[list[domish.Element], rsm.RSMResponse | None]]: + return defer.ensureDeferred(self._items(request)) + + async def _items( + self, request: rsm.PubSubRequest + ) -> tuple[list[domish.Element], rsm.RSMResponse | None]: + 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 + + def retract(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._retract(request)) + + async def _retract(self, request: rsm.PubSubRequest) -> None: + return + + def purge(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._purge(request)) + + async def _purge(self, request: rsm.PubSubRequest) -> None: + return + + def delete(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._delete(request)) + + async def _delete(self, request: rsm.PubSubRequest) -> None: + return + + def affiliationsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._affiliationsGet(request)) + + async def _affiliationsGet(self, request: rsm.PubSubRequest) -> None: + return + + def affiliationsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._affiliationsSet(request)) + + async def _affiliationsSet(self, request: rsm.PubSubRequest) -> None: + return + + def subscriptionsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._subscriptionsGet(request)) + + async def _subscriptionsGet(self, request: rsm.PubSubRequest) -> None: + return + + def subscriptionsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]: + return defer.ensureDeferred(self._subscriptionsSet(request)) + + async def _subscriptionsSet(self, request: rsm.PubSubRequest) -> None: + return + + +class PubsubService(rsm.PubSubService): + def __init__(self, component: "PubsubComponent"): + super().__init__(resource=PubsubResource(self)) + self.component = component + self.discoidentity = { + "category": "pubsub", + "type": "service", + "name": "Libervia Pubsub", + } + + @property + def client(self) -> SatXMPPComponent: + return self.component._client + + +class PubsubComponent: + IMPORT_NAME = IMPORT_NAME + verbose = 0 + + def __init__(self, host): + self._clientt: SatXMPPComponent | None = None + self.pubsub_service = PubsubService(self) + + def client(self) -> SatXMPPComponent: + assert self._client is not None + return self._client + + def get_handler(self, __) -> PubsubService: + return self.pubsub_service + + def profile_connecting(self, client: SatXMPPEntity) -> None: + self._client = cast(SatXMPPComponent, client) + + +@implementer(itx_xmpp.IDisco) +class PubsubHandler(XMPPHandler): + + def getDiscoInfo( + self, requestor, target, nodeIdentifier="" + ) -> list[disco.DiscoFeature]: + return [disco.DiscoFeature(NS_AV_CONFERENCES)]