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"