From f5e0152b3204c82e963035e687e373db365c381e Mon Sep 17 00:00:00 2001 From: Goffi Date: Tue, 14 Apr 2026 23:04:04 +0200 Subject: [PATCH] plugin XEP-0355: Namespace Delegation implementation: The admin mode is implemented. It is needed to handle PEP with Pubsub Component. rel 476 --- libervia/backend/core/patches.py | 7 +- libervia/backend/plugins/plugin_xep_0297.py | 67 ++- libervia/backend/plugins/plugin_xep_0355.py | 483 ++++++++++++++++++++ libervia/backend/tools/xml_tools.py | 9 +- tests/unit/test_xep_0355.py | 425 +++++++++++++++++ 5 files changed, 970 insertions(+), 21 deletions(-) create mode 100644 libervia/backend/plugins/plugin_xep_0355.py create mode 100644 tests/unit/test_xep_0355.py diff --git a/libervia/backend/core/patches.py b/libervia/backend/core/patches.py index 8e8948b9..da7f9d24 100644 --- a/libervia/backend/core/patches.py +++ b/libervia/backend/core/patches.py @@ -16,6 +16,7 @@ from tx_xmpp import client from zope.interface import implementer from libervia.backend.core.constants import Const as C +from libervia.backend.core.exceptions import CancelError from libervia.backend.core.log import getLogger log = getLogger(__name__) @@ -239,7 +240,11 @@ class XmlStream(xmlstream.XmlStream): def send(self, obj): for hook in self._sendHooks: - hook(obj) + try: + hook(obj) + except CancelError: + # Hook requested to prevent stanza from being sent + return xmlstream.XmlStream.send(self, obj) diff --git a/libervia/backend/plugins/plugin_xep_0297.py b/libervia/backend/plugins/plugin_xep_0297.py index da9d5b4f..4ec96b77 100644 --- a/libervia/backend/plugins/plugin_xep_0297.py +++ b/libervia/backend/plugins/plugin_xep_0297.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 -# SAT plugin for Stanza Forwarding (XEP-0297) +# Libervia plugin for Stanza Forwarding (XEP-0297) # Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) # Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org) @@ -47,6 +47,7 @@ PLUGIN_INFO = { C.PI_NAME: "Stanza Forwarding", C.PI_IMPORT_NAME: "XEP-0297", C.PI_TYPE: "XEP", + C.PI_MODES: [C.PLUG_MODE_COMPONENT], C.PI_PROTOCOLS: ["XEP-0297"], C.PI_DEPENDENCIES: ["XEP-0428"], C.PI_MAIN: "XEP_0297", @@ -79,11 +80,11 @@ class XEP_0297: return XEP_0297_handler(self, client.profile) @classmethod - def update_uri(cls, element, uri): + def update_uri(cls, element: domish.Element, uri: str) -> None: """Update recursively the element URI. - @param element (domish.Element): element to update - @param uri (unicode): new URI + @param element: element to update + @param uri: new URI """ # XXX: we need this because changing the URI of an existing element # containing children doesn't update the children's blank URI. @@ -93,6 +94,25 @@ class XEP_0297: if isinstance(child, domish.Element) and not child.uri: XEP_0297.update_uri(child, uri) + @staticmethod + def build_forwarded( + stanza: domish.Element, timestamp: datetime | None = None + ) -> domish.Element: + """Build a element containing a stanza. + + @param stanza: The stanza to wrap in forwarded element. + @param timestamp: Timestamp for delay element (XEP-0203). + @return: The element with the stanza as child. + """ + forwarded_elt = domish.Element((C.NS_FORWARD, "forwarded")) + if timestamp: + delay_elt = G.host.plugins["XEP-0203"].delay(timestamp) + forwarded_elt.addChild(delay_elt) + if not stanza.uri: + XEP_0297.update_uri(stanza, "jabber:client") + forwarded_elt.addChild(stanza) + return forwarded_elt + def _forward( self, message_id: str, recipient_jid_s: str, profile_key: str ) -> defer.Deferred[None]: @@ -101,7 +121,10 @@ class XEP_0297: return defer.ensureDeferred(self.forward_by_id(client, message_id, recipient_jid)) async def forward_by_id( - self, client: SatXMPPEntity, message_id: str, recipient_jid: jid.JID + self, + client: SatXMPPEntity, + message_id: str, + recipient_jid: jid.JID, ) -> None: history = cast( History, @@ -156,7 +179,7 @@ class XEP_0297: to_jid: jid.JID, timestamp: datetime | None, fallback_msg: str | None = None, - ): + ) -> defer.Deferred: """Forward a message to the given JID. @param client: client instance. @@ -170,13 +193,7 @@ class XEP_0297: message_elt["to"] = to_jid.full() message_elt["type"] = stanza["type"] - forwarded_elt = domish.Element((C.NS_FORWARD, "forwarded")) - if timestamp: - delay_elt = self.host.plugins["XEP-0203"].delay(timestamp) - forwarded_elt.addChild(delay_elt) - if not stanza.uri: - XEP_0297.update_uri(stanza, "jabber:client") - forwarded_elt.addChild(stanza) + forwarded_elt = XEP_0297.build_forwarded(stanza, timestamp) message_elt.addChild(domish.Element((None, "body"))) message_elt.addChild(forwarded_elt) @@ -189,13 +206,31 @@ class XEP_0297: @implementer(itx_xmpp.IDisco) class XEP_0297_handler(XMPPHandler): - def __init__(self, plugin_parent, profile): + def __init__(self, plugin_parent: XEP_0297, profile: str) -> None: self.plugin_parent = plugin_parent self.host = plugin_parent.host self.profile = profile - def getDiscoInfo(self, requestor, target, nodeIdentifier=""): + def getDiscoInfo( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoFeature]: + """Get disco info for pubsub extended discovery + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: optional node identifier + @return: list of disco features + """ return [disco.DiscoFeature(C.NS_FORWARD)] - def getDiscoItems(self, requestor, target, nodeIdentifier=""): + def getDiscoItems( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoItem]: + """Get disco items with extended discovery support + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: optional node identifier + @return: list of disco items + """ return [] diff --git a/libervia/backend/plugins/plugin_xep_0355.py b/libervia/backend/plugins/plugin_xep_0355.py new file mode 100644 index 00000000..766e6888 --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0355.py @@ -0,0 +1,483 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Namespace Delegation (XEP-0355). +# 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 +from functools import partial +from typing import TYPE_CHECKING, NamedTuple, cast + +from twisted.internet import defer, reactor +from twisted.internet.interfaces import IDelayedCall +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 disco, itx_xmpp, pubsub +from tx_xmpp import data_form +from zope.interface import implementer + +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.plugins.plugin_xep_0297 import NS_FORWARD, XEP_0297 +from libervia.backend.tools.utils import as_deferred, ensure_deferred +from libervia.backend.tools.xml_tools import element_copy + +if TYPE_CHECKING: + from libervia.backend.core.main import LiberviaBackend + + +log = getLogger(__name__) + +PLUGIN_INFO = { + C.PI_NAME: "Namespace Delegation", + C.PI_IMPORT_NAME: "XEP-0355", + C.PI_TYPE: "XEP", + C.PI_MODES: [C.PLUG_MODE_COMPONENT], + C.PI_PROTOCOLS: ["XEP-0355"], + C.PI_DEPENDENCIES: ["XEP-0297"], + C.PI_RECOMMENDATIONS: [], + C.PI_MAIN: "XEP_0355", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _("""Namespace delegation implementation."""), +} + +NS_DELEGATION = "urn:xmpp:delegation:2" + +DELEGATION_MAIN_SEP = "::" +DELEGATION_BARE_SEP = ":bare:" + +SEND_HOOK_TIMEOUT = 300 + + +@dataclass +class DelegationData: + delegation_received: defer.Deferred[None] = field( + default_factory=lambda: defer.Deferred() + ) + + +class DelegatedIQ(NamedTuple): + """Data of IQ request being tracked for delegation.""" + + #: The client that received the delegation. + client: SatXMPPComponent + #: The original IQ request. + iq_request_elt: domish.Element + #: The wrapping delegation IQ from the server. + wrapping_iq_elt: domish.Element + #: Timer for the request. + timeout: IDelayedCall + + +class XEP_0355: + """XEP-0355: Namespace Delegation handler.""" + + namespace = NS_DELEGATION + + def __init__(self, host: "LiberviaBackend") -> None: + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + self.host = host + self._forward = cast(XEP_0297, host.plugins["XEP-0297"]) + self.delegated_namespaces: set[str] = set() + self._current_iqs: dict[str, DelegatedIQ] = {} + + host.register_namespace("delegation", NS_DELEGATION) + host.trigger.add("stream_hooks", self.add_hooks) + + def add_hooks(self, client, receive_hooks, send_hooks): + """Register send hook to intercept outgoing IQ results/errors.""" + send_hooks.append(partial(self.on_send, client=client)) + return True + + def on_send(self, obj, client): + """Intercept outgoing stanzas to catch delegated IQ responses.""" + if not client.is_component: + return + + if not self._current_iqs: + return + + if not isinstance(obj, domish.Element): + return + + if obj.name != "iq": + return + + if obj.getAttribute("type") not in ("result", "error"): + return + + self.handle_outgoing_iq(obj) + + def handle_outgoing_iq(self, iq_elt: domish.Element) -> None: + """Check if outgoing IQ is a response to a delegated stanza. + + @param iq_elt: The outgoing IQ to check. + """ + iq_id = iq_elt.getAttribute("id") + if not iq_id or iq_id not in self._current_iqs: + return + + delegated_iq = self._current_iqs[iq_id] + + iq_to = iq_elt.getAttribute("to") + iq_request_from = delegated_iq.iq_request_elt.getAttribute("from") + + # Response's "to" must match original request's "from" + if iq_to != iq_request_from: + log.error( + f"IQ id conflict for {iq_id}: expected response to {iq_request_from!r}, " + f"got {iq_to!r}" + ) + del self._current_iqs[iq_id] + return + + if not delegated_iq.timeout.called: + delegated_iq.timeout.cancel() + + del self._current_iqs[iq_id] + self.delegated_result( + delegated_iq.client, + delegated_iq.iq_request_elt, + iq_elt, + delegated_iq.wrapping_iq_elt, + ) + raise exceptions.CancelError() + + def is_pep(self, client: SatXMPPComponent, entity_jid: jid.JID) -> bool: + """Check if a request is a PEP request. + + @param client: Client instance. + @param to_jid: The JID to check. + @return: True if this is a PEP request to a local user. + @raise exceptions.InternalError: Client is not a component (XEP-0355 can be used + only with components). + """ + if not client.is_component: + raise exceptions.InternalError("XEP-0355 used with a non-component.") + + if entity_jid == client.jid: + return False + + if entity_jid.user is None: + return False + + if entity_jid.resource is not None: + return False + + if not client.is_local(entity_jid): + return False + + return True + + def get_handler(self, client: SatXMPPEntity) -> "XEP_0355_Handler": + return XEP_0355_Handler(self) + + def on_advertise(self, message_elt: domish.Element, client: SatXMPPComponent) -> None: + """Handle delegation advertisement from server. + + @param client: Client instance. + @param message_elt: Delegation advertisement message. + """ + server_jid = client.server_jid + if jid.JID(message_elt["from"]) != server_jid: + log.warning( + f"Delegation advertising message received from {message_elt['from']}," + f" while expected server JID is {server_jid}, this may be a security" + " threat, please check your configuration and network." + ) + raise error.StanzaError("not-allowed") + + try: + delegation_elt = next(message_elt.elements(NS_DELEGATION, "delegation")) + except StopIteration: + log.warning("Invalid delegation advertisement: missing delegation element") + raise error.StanzaError("not-acceptable") + + delegated = {} + for child in delegation_elt.elements(NS_DELEGATION): + if child.name != "delegated": + log.warning( + f"Invalid delegation advertisement: unexpected element {child.name!r}" + ) + raise error.StanzaError("not-acceptable") + + try: + namespace = child["namespace"] + except KeyError: + log.warning( + "Invalid delegation advertisement: missing namespace attribute in" + " delegated element" + ) + raise error.StanzaError("not-acceptable") + + delegated[namespace] = [] + for attribute_elt in child.elements(NS_DELEGATION, "attribute"): + try: + delegated[namespace].append(attribute_elt["name"]) + except KeyError: + log.warning( + "Invalid delegation advertisement: missing name attribute in" + " attribute element" + ) + raise error.StanzaError("not-acceptable") + + self.delegated_namespaces = set(delegated.keys()) + + log_lines = [f"Delegations updated for {client.profile}:"] + for ns, attrs in sorted(delegated.items()): + if attrs: + log_lines.append( + f" - namespace {ns!r} with filtering on {len(attrs)} attribute(s):" + f" {', '.join(map(repr, attrs))}" + ) + else: + log_lines.append(f" - namespace {ns!r}") + + log.info("\n".join(log_lines)) + + delegation_data = client.get_plugin_data(C.PI_IMPORT_NAME, DelegationData) + if not delegation_data.delegation_received.called: + delegation_data.delegation_received.callback(None) + + def on_forward( + self, delegation_iq_elt: domish.Element, client: SatXMPPComponent + ) -> None: + """Handle forwarded stanza from server. + + @param client: Client instance. + @param delegation_iq_elt: Full delegation stanza. + """ + server_jid = client.server_jid + if jid.JID(delegation_iq_elt["from"]) != server_jid: + log.warning( + "SECURITY: forwarded stanza from unexpected server:" + f" {delegation_iq_elt.toXml()}" + ) + raise error.StanzaError("not-allowed") + + try: + delegation_elt = next(delegation_iq_elt.elements(NS_DELEGATION, "delegation")) + forwarded_elt = next(delegation_elt.elements(C.NS_FORWARD, "forwarded")) + forwarded_iq_elt = forwarded_elt.firstChildElement() + if forwarded_iq_elt is None: + raise error.StanzaError("not-acceptable") + except StopIteration: + raise error.StanzaError("not-acceptable") + + iq_id = forwarded_iq_elt["id"] + iq_to = forwarded_iq_elt.getAttribute("to") + timeout = reactor.callLater(SEND_HOOK_TIMEOUT, self.on_timeout, iq_id, iq_to) + self._current_iqs[iq_id] = DelegatedIQ( + client, forwarded_iq_elt, delegation_iq_elt, timeout + ) + + if not forwarded_iq_elt.hasAttribute("to"): + forwarded_iq_elt["to"] = jid.JID(forwarded_iq_elt["from"]).userhost() + + # This attribute may be used by plugins to track that the stanza is delegated. + forwarded_iq_elt.delegated = True + + # Prevent further processing. + delegation_iq_elt.handled = True + + client.xmlstream.dispatch(forwarded_iq_elt) + + def delegated_result( + self, + client: SatXMPPComponent, + iq_req_elt: domish.Element, + iq_resp_elt: domish.Element, + wrapping_iq_elt: domish.Element, + ) -> None: + """Wrap result/error stanza and send back to server. + + @param client: Client instance. + @param iq_req_elt: The original forwarded IQ request. + @param iq_resp_elt: The response IQ (result or error). + @param wrapping_iq_elt: The original delegation IQ from the server. + """ + iq_result_elt = domish.Element((None, "iq")) + iq_result_elt["from"] = wrapping_iq_elt["to"] + iq_result_elt["to"] = wrapping_iq_elt["from"] + iq_result_elt["id"] = wrapping_iq_elt["id"] + iq_result_elt["type"] = "result" + + delegation_elt = domish.Element((NS_DELEGATION, "delegation")) + + wrapped_iq_elt = domish.Element(("jabber:client", "iq")) + # According to the specs, we don't set a "from" here + wrapped_iq_elt["to"] = iq_req_elt["from"] + wrapped_iq_elt["id"] = iq_req_elt["id"] + wrapped_iq_elt["type"] = iq_resp_elt["type"] + + for child in iq_resp_elt.children: + wrapped_iq_elt.addChild(element_copy(child)) + + forwarded_elt = XEP_0297.build_forwarded(wrapped_iq_elt) + delegation_elt.addChild(forwarded_elt) + iq_result_elt.addChild(delegation_elt) + + client.xmlstream.send(iq_result_elt) + + def on_timeout(self, iq_id: str, iq_to: str | None) -> None: + """Handle timeout for a delegated IQ request. + + @param iq_id: The ID of the timed-out IQ. + @param iq_to: The 'to' attribute of the timed-out IQ. + """ + self._current_iqs.pop(iq_id, None) + log.error(f"Timeout waiting for response to delegated IQ {iq_id!r} to {iq_to!r}") + + +@implementer(itx_xmpp.IDisco) +class XEP_0355_Handler(XMPPHandler): + """Handler for delegation stanzas on a per-client basis.""" + + def __init__(self, plugin_parent: XEP_0355) -> None: + self.plugin_parent = plugin_parent + + @property + def client(self) -> SatXMPPComponent: + assert self.parent is not None + return self.parent + + def connectionInitialized(self) -> None: + """Handle connection initialization for delegation stanzas.""" + self.client.xmlstream.addObserver( + f"/message/delegation[@xmlns='{NS_DELEGATION}']", + self.plugin_parent.on_advertise, + client=self.client, + ) + self.client.xmlstream.addObserver( + f"/iq[@type='set']/delegation[@xmlns='{NS_DELEGATION}']" + f"/forwarded[@xmlns='{NS_FORWARD}']", + self.plugin_parent.on_forward, + client=self.client, + ) + + @ensure_deferred + async def getDiscoInfo( + self, + requestor: jid.JID, + target: jid.JID, + nodeIdentifier: str = "", + ) -> list[disco.DiscoFeature | disco.DiscoIdentity | data_form.Form]: + """Get disco info for delegation nesting. + + Server queries this handler for features/identities to nest in its own disco info. + Node format: urn:xmpp:delegation:2:: for server + urn:xmpp:delegation:2:bare: for bare JID + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: Node identifier + @return: Disco features/identities for the delegated namespace + """ + if not nodeIdentifier.startswith(NS_DELEGATION): + return [] + + # Extract namespace from delegation node identifier + namespace: str | None = None + if DELEGATION_MAIN_SEP in nodeIdentifier: + _, namespace = nodeIdentifier.split(DELEGATION_MAIN_SEP, 1) + elif DELEGATION_BARE_SEP in nodeIdentifier: + _, namespace = nodeIdentifier.split(DELEGATION_BARE_SEP, 1) + else: + log.error(f"Unexpected disco node: {nodeIdentifier}") + + if not namespace: + return [] + + return await self._get_disco_info_for_namespace(namespace, requestor, target) + + def getDiscoItems( + self, + requestor: jid.JID, + target: jid.JID, + nodeIdentifier: str = "", + ) -> list[disco.DiscoItem]: + """Get disco items. + + Actual disco#items delegation is handled via stanza forwarding. + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: Node identifier + @return: list of disco items + """ + return [] + + async def _get_disco_info_for_namespace( + self, + namespace: str, + requestor: jid.JID, + target: jid.JID, + ) -> list[disco.DiscoFeature | disco.DiscoIdentity | data_form.Form]: + """Collect disco info for a delegated namespace from all handlers. + + @param namespace: The delegated namespace + @return: List of features/identities matching the namespace + """ + # Normalize pubsub namespace: pubsub uses several namespaces starting with + # NS_PUBSUB; e.g. http://jabber.org/protocol/pubsub#owner. We return the same + # disco info for all pubsub variants (matches SàT Pubsub reference implementation) + normalized_namespace = namespace + if namespace.startswith(pubsub.NS_PUBSUB): + normalized_namespace = pubsub.NS_PUBSUB + + disco_info: list[disco.DiscoFeature | disco.DiscoIdentity | data_form.Form] = [] + + for handler in self.client.handlers: + if not hasattr(handler, "getDiscoInfo"): + continue + + try: + handler_disco_info = await as_deferred( + handler.getDiscoInfo, requestor, target, "" + ) + if not handler_disco_info: + continue + + for info in handler_disco_info: + if isinstance(info, disco.DiscoFeature): + if info.startswith(normalized_namespace): + disco_info.append(info) + elif isinstance(info, disco.DiscoIdentity): + # Check if identity matches namespace (pubsub category for pubsub + # namespace) + # FIXME: we should handle other identities + if ( + normalized_namespace == pubsub.NS_PUBSUB + and info.category == "pubsub" + ): + disco_info.append(info) + elif isinstance(info, data_form.Form): + if info.formNamespace and info.formNamespace.startswith( + normalized_namespace + ): + disco_info.append(info) + except Exception: + log.exception(f"Error collecting disco info from handler") + + # Add PEP identity for pubsub namespace + # FIXME: to be moved to Pubsub component + if normalized_namespace == pubsub.NS_PUBSUB: + disco_info.append(disco.DiscoIdentity("pubsub", "pep", "PEP")) + + return disco_info diff --git a/libervia/backend/tools/xml_tools.py b/libervia/backend/tools/xml_tools.py index 1451a2d5..da539aa6 100644 --- a/libervia/backend/tools/xml_tools.py +++ b/libervia/backend/tools/xml_tools.py @@ -1822,11 +1822,12 @@ def element_copy( ) -> domish.Element: """Make a copy of a domish.Element - The copy will have its own children list, so other elements - can be added as direct children without modifying orignal one. - Children are not deeply copied, so if an element is added to a child or grandchild, - it will also affect original element. + The copy will have its own children list, so other elements can be added as direct + children without modifying orignal one. Children are not deeply copied, so if an + element is added to a child or grandchild, it will also affect original element. + @param element: Element to clone + @return: Copy of the element """ new_elt = domish.Element( (element.uri, element.name), diff --git a/tests/unit/test_xep_0355.py b/tests/unit/test_xep_0355.py new file mode 100644 index 00000000..643c61e6 --- /dev/null +++ b/tests/unit/test_xep_0355.py @@ -0,0 +1,425 @@ +#!/usr/bin/env python3 + +# Libervia: an XMPP client +# 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 . + +""" +Tests for XEP-0355 Namespace Delegation implementation. + +Focus: Security and delegation request handling. +""" + +from unittest.mock import Mock + +from pytest import fixture, raises +from pytest_twisted import ensureDeferred as ed +from twisted.words.protocols.jabber import jid +from twisted.words.protocols.jabber.error import StanzaError +from twisted.words.xish import domish +from tx_xmpp import disco, pubsub + +from libervia.backend.core.constants import Const as C +from libervia.backend.core.exceptions import CancelError +from libervia.backend.plugins.plugin_xep_0297 import XEP_0297 +from libervia.backend.plugins.plugin_xep_0355 import ( + XEP_0355, + NS_DELEGATION, + DelegatedIQ, +) +from libervia.backend.tools.xml_tools import element_copy + + +def make_delegation_advertisement(server_jid, namespaces): + """Helper to create delegation advertisement message.""" + msg = domish.Element( + (None, "message"), + attribs={"from": server_jid.full(), "to": "pubsub.test.example"}, + ) + delegation = domish.Element((NS_DELEGATION, "delegation")) + for ns in namespaces: + delegated = domish.Element( + (NS_DELEGATION, "delegated"), attribs={"namespace": ns} + ) + delegation.addChild(delegated) + msg.addChild(delegation) + return msg + + +def make_forwarded_stanza(server_jid, component_jid, inner_stanza): + """Helper to create forwarded stanza IQ.""" + iq = domish.Element( + (None, "iq"), + attribs={ + "from": server_jid.full(), + "to": component_jid.full(), + "type": "set", + "id": "test123", + }, + ) + delegation = domish.Element((NS_DELEGATION, "delegation")) + forwarded = XEP_0297.build_forwarded(element_copy(inner_stanza)) + delegation.addChild(forwarded) + iq.addChild(delegation) + return iq + + +@fixture +def xep_0355(host): + """Create XEP-0355 plugin instance.""" + return XEP_0355(host) + + +@fixture(autouse=True) +def setup_component_client(client): + """Set up client as a component for XEP-0355 tests.""" + client.is_component = True + + +@fixture +def delegation_handler(xep_0355, client): + """Create delegation handler for testing.""" + handler = xep_0355.get_handler(client) + handler.parent = client + return handler + + +class TestXEP0355Security: + """Security tests for delegation advertisement and forwarding.""" + + @ed + async def test_rejects_advertisement_from_wrong_server(self, xep_0355, client): + """Advertisement from wrong server raises StanzaError.""" + wrong_server = jid.JID("other.example") + msg = make_delegation_advertisement(wrong_server, [pubsub.NS_PUBSUB]) + + with raises(StanzaError, match="not-allowed"): + xep_0355.on_advertise(msg, client) + + @ed + async def test_rejects_advertisement_missing_delegation_element( + self, xep_0355, client + ): + """Advertisement without delegation element raises StanzaError.""" + msg = domish.Element( + (None, "message"), + attribs={"from": client.server_jid.full(), "to": "pubsub.test.example"}, + ) + + with raises(StanzaError, match="not-acceptable"): + xep_0355.on_advertise(msg, client) + + @ed + async def test_rejects_delegated_without_namespace(self, xep_0355, client): + """Delegated elements without namespace attribute raise StanzaError.""" + msg = make_delegation_advertisement(client.server_jid, [pubsub.NS_PUBSUB]) + + # Add a delegated element without namespace + delegation = next(msg.elements(NS_DELEGATION, "delegation")) + bad_delegated = domish.Element((NS_DELEGATION, "delegated")) + delegation.addChild(bad_delegated) + + with raises(StanzaError, match="not-acceptable"): + xep_0355.on_advertise(msg, client) + + @ed + async def test_rejects_unexpected_element_in_delegation(self, xep_0355, client): + """Unexpected elements in delegation raise StanzaError.""" + msg = make_delegation_advertisement(client.server_jid, [pubsub.NS_PUBSUB]) + + # Add an unexpected element + delegation = next(msg.elements(NS_DELEGATION, "delegation")) + bad_element = domish.Element((NS_DELEGATION, "unknown")) + delegation.addChild(bad_element) + + with raises(StanzaError, match="not-acceptable"): + xep_0355.on_advertise(msg, client) + + @ed + async def test_rejects_forwarded_stanza_from_wrong_server(self, xep_0355, client): + """Forwarded stanza from non-server JID raises StanzaError.""" + wrong_server = jid.JID("other.example") + inner_iq = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "user@test.example", + "to": "pubsub.test.example", + "id": "inner1", + }, + ) + iq = make_forwarded_stanza(wrong_server, client.jid, inner_iq) + + with raises(StanzaError, match="not-allowed"): + xep_0355.on_forward(iq, client) + + @ed + async def test_rejects_malformed_forwarding(self, xep_0355, client): + """Malformed forwarding structure raises StanzaError.""" + # Create IQ without proper structure + iq = domish.Element( + (None, "iq"), + attribs={ + "from": client.server_jid.full(), + "to": client.jid.full(), + "type": "set", + "id": "test", + }, + ) + # Missing delegation element + iq.addChild(domish.Element((None, "query"))) + + with raises(StanzaError, match="not-acceptable"): + xep_0355.on_forward(iq, client) + + @ed + async def test_cleans_up_on_iq_id_conflict(self, xep_0355, client): + """IQ with wrong 'to' address is cleaned up.""" + inner_req = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "pubsub.test.example", + "to": "user@test.example", + "id": "inner1", + }, + ) + wrapping_iq = domish.Element( + (None, "iq"), + attribs={ + "from": client.server_jid.full(), + "to": client.jid.full(), + "id": "wrap1", + }, + ) + + mock_timeout = Mock() + mock_timeout.called = False + mock_timeout.cancel = Mock() + + xep_0355._current_iqs["inner1"] = DelegatedIQ( + client, inner_req, wrapping_iq, mock_timeout + ) + + # Outgoing IQ with wrong 'to' + outgoing_iq = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "pubsub.test.example", + "to": "wrong@test.example", + "id": "inner1", + "type": "result", + }, + ) + + xep_0355.handle_outgoing_iq(outgoing_iq) + + # Entry should be cleaned up + assert "inner1" not in xep_0355._current_iqs + # Should not have sent wrapped result + assert not client.xmlstream.send.called + + +class TestXEP0355DelegationHandling: + """Tests for delegation request handling and response wrapping.""" + + @ed + async def test_dispatches_inner_stanza_from_forwarded(self, xep_0355, client): + """Forwarded stanza inner IQ is dispatched to xmlstream.""" + inner_iq = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "user@test.example/resource", + "to": "pubsub.test.example", + "type": "set", + "id": "inner1", + }, + ) + inner_iq.addChild(domish.Element((pubsub.NS_PUBSUB, "publish"))) + + iq = make_forwarded_stanza(client.server_jid, client.jid, inner_iq) + + xep_0355.on_forward(iq, client) + + # Verify dispatch was called + client.xmlstream.dispatch.assert_called_once() + dispatched = client.xmlstream.dispatch.call_args.args[0] + assert dispatched["id"] == "inner1" + + @ed + async def test_sets_to_bare_jid_if_missing(self, xep_0355, client): + """Dispatched stanza gets to=bare_jid if not present.""" + inner_iq = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "user@test.example/resource", + "id": "inner1", + "type": "set", + }, + ) + # No 'to' attribute + + iq = make_forwarded_stanza(client.server_jid, client.jid, inner_iq) + + xep_0355.on_forward(iq, client) + + dispatched = client.xmlstream.dispatch.call_args.args[0] + assert dispatched["to"] == "user@test.example" + + @ed + async def test_wraps_result_in_delegation_format(self, xep_0355, client): + """Result IQ is properly wrapped in delegation format.""" + inner_req = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "pubsub.test.example", + "to": "user@test.example", + "id": "inner1", + }, + ) + wrapping_iq = domish.Element( + (None, "iq"), + attribs={ + "from": client.server_jid.full(), + "to": client.jid.full(), + "id": "wrap1", + }, + ) + + # Create response + inner_resp = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "pubsub.test.example", + "to": "user@test.example", + "id": "inner1", + "type": "result", + }, + ) + result_data = domish.Element( + (pubsub.NS_PUBSUB, "published"), attribs={"node": "test_node"} + ) + inner_resp.addChild(result_data) + + xep_0355.delegated_result(client, inner_req, inner_resp, wrapping_iq) + + # Verify send was called with wrapped result + sent = client.xmlstream.send.call_args.args[0] + assert sent["type"] == "result" + assert sent["id"] == "wrap1" + assert sent["from"] == wrapping_iq["to"] + assert sent["to"] == wrapping_iq["from"] + + # Verify structure + delegation_elt = next(sent.elements(NS_DELEGATION, "delegation")) + forwarded_elt = next(delegation_elt.elements(C.NS_FORWARD, "forwarded")) + wrapped_iq = next(forwarded_elt.elements("jabber:client", "iq")) + + assert wrapped_iq["id"] == "inner1" + assert wrapped_iq["type"] == "result" + + @ed + async def test_intercepts_outgoing_result_iq(self, xep_0355, client): + """Outgoing result IQ matching tracked ID is intercepted and wrapped.""" + inner_req = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "pubsub.test.example", + "to": "user@test.example", + "id": "inner1", + }, + ) + wrapping_iq = domish.Element( + (None, "iq"), + attribs={ + "from": client.server_jid.full(), + "to": client.jid.full(), + "id": "wrap1", + }, + ) + + mock_timeout = Mock() + mock_timeout.called = False + mock_timeout.cancel = Mock() + + xep_0355._current_iqs["inner1"] = DelegatedIQ( + client, inner_req, wrapping_iq, mock_timeout + ) + + # Simulate outgoing response + outgoing_iq = domish.Element( + ("jabber:client", "iq"), + attribs={ + "from": "pubsub.test.example", + "to": "pubsub.test.example", + "id": "inner1", + "type": "result", + }, + ) + + # Should raise CancelError to prevent normal send + with raises(CancelError): + xep_0355.handle_outgoing_iq(outgoing_iq) + + # Verify callback was invoked (send was called) + assert client.xmlstream.send.called + + +class TestXEP0355DiscoNesting: + """Tests for disco nesting support per XEP-0355 spec.""" + + @ed + async def test_server_queries_pubsub_namespace_gets_pep_identity( + self, delegation_handler + ): + """Server queries for pubsub namespace, component returns PEP identity.""" + node_id = f"{NS_DELEGATION}::{pubsub.NS_PUBSUB}" + result = await delegation_handler.getDiscoInfo( + jid.JID("juliet@capulet.lit/balcony"), + jid.JID("capulet.lit"), + node_id, + ) + + pep_identities = [r for r in result if isinstance(r, disco.DiscoIdentity)] + assert len(pep_identities) >= 1, "Should return at least PEP identity" + assert pep_identities[0].category == "pubsub" + assert pep_identities[0].type == "pep" + + @ed + async def test_normalizes_pubsub_namespace_variants(self, delegation_handler): + """Component normalizes pubsub namespace variants to base namespace.""" + node_id_owner = f"{NS_DELEGATION}::{pubsub.NS_PUBSUB}#owner" + result_owner = await delegation_handler.getDiscoInfo( + jid.JID("juliet@capulet.lit"), + jid.JID("capulet.lit"), + node_id_owner, + ) + + node_id_base = f"{NS_DELEGATION}::{pubsub.NS_PUBSUB}" + result_base = await delegation_handler.getDiscoInfo( + jid.JID("juliet@capulet.lit"), + jid.JID("capulet.lit"), + node_id_base, + ) + + # Both should return PEP identity + pep_owner = [r for r in result_owner if isinstance(r, disco.DiscoIdentity)] + pep_base = [r for r in result_base if isinstance(r, disco.DiscoIdentity)] + + assert len(pep_owner) >= 1, "pubsub#owner should return PEP identity" + assert pep_owner[0].category == "pubsub" + assert pep_owner[0].type == "pep" + + assert len(pep_base) >= 1, "pubsub base should return PEP identity" + assert pep_base[0].category == "pubsub" + assert pep_base[0].type == "pep"