libervia-backend/tests/unit/test_plugin_xep_0355.py

#!/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"