plugin XEP-0355: Namespace Delegation implementation:
The admin mode is implemented. It is needed to handle PEP with Pubsub Component. rel 476
This commit is contained in:
Goffi
parent
da2c2c49ea
commit
f5e0152b32
5 changed files with 970 additions and 21 deletions
425
tests/unit/test_xep_0355.py
Normal file
425
tests/unit/test_xep_0355.py
Normal file
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
"""
|
||||
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"
|
||||
Loading…
Reference in a new issue