From 6605e523d07f5fe35ae90c0d7bd3c6235ec82050 Mon Sep 17 00:00:00 2001 From: Goffi Date: Tue, 14 Apr 2026 23:08:21 +0200 Subject: [PATCH] plugin XEP-0356: Privileged Entity implementation: This is the second XEP needed to implement PEP in Pubsub Component. This implementation should be more or less complete. rel 476 --- libervia/backend/plugins/plugin_xep_0356.py | 581 ++++++++++++++++ tests/unit/conftest.py | 44 ++ tests/unit/test_xep_0356.py | 708 ++++++++++++++++++++ 3 files changed, 1333 insertions(+) create mode 100644 libervia/backend/plugins/plugin_xep_0356.py create mode 100644 tests/unit/test_xep_0356.py diff --git a/libervia/backend/plugins/plugin_xep_0356.py b/libervia/backend/plugins/plugin_xep_0356.py new file mode 100644 index 00000000..40acc340 --- /dev/null +++ b/libervia/backend/plugins/plugin_xep_0356.py @@ -0,0 +1,581 @@ +#!/usr/bin/env python3 + +# Libervia plugin for Privileged Entity (XEP-0356). +# 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 textwrap import dedent +from typing import TYPE_CHECKING, Final, Literal, cast +import uuid + +from twisted.internet import reactor +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 +from tx_xmpp.xmppim import RosterItem +from zope.interface import implementer + +from libervia.backend import G +from libervia.backend.core import exceptions +from libervia.backend.core.constants import Const as C +from libervia.backend.core.core_types import SatXMPPComponent, SatXMPPEntity +from libervia.backend.core.i18n import _ +from libervia.backend.core.log import getLogger +from libervia.backend.plugins.plugin_xep_0297 import NS_FORWARD, XEP_0297 + +if TYPE_CHECKING: + from libervia.backend.core.main import LiberviaBackend + +log = getLogger(__name__) + +PLUGIN_INFO = { + C.PI_NAME: "Privileged Entity", + C.PI_IMPORT_NAME: "XEP-0356", + C.PI_TYPE: "XEP", + C.PI_MODES: [C.PLUG_MODE_COMPONENT], + C.PI_PROTOCOLS: ["XEP-0356"], + C.PI_DEPENDENCIES: ["XEP-0297"], + C.PI_RECOMMENDATIONS: [], + C.PI_MAIN: "XEP_0356", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: dedent(_("""\ + Privileged entity implementation for components. + + This plugin provides privileged entity capabilities to components, allowing them to: + - Access and modify user rosters (with roster push support) + - Send messages on behalf of server + - Send IQ stanzas on behalf of server + - Receive presence information for managed entities and roster contacts + """)), +} + +NS_PRIVILEGE: Final = "urn:xmpp:privilege:2" +NS_ROSTER = "jabber:iq:roster" +ROSTER_TTL = 3600 +IMPORT_NAME = "XEP-0356" + +RosterPrivilegeType = Literal["none", "get", "set", "both"] +MessagePrivilegeType = Literal["none", "outgoing"] +PresencePrivilegeType = Literal["none", "managed_entity", "roster"] + + +@dataclass +class IQPermission: + """IQ permission for a namespace.""" + + get: bool + set: bool + + +@dataclass +class RosterCacheEntry: + """Cached roster data for a managed entity.""" + + roster: dict[jid.JID, RosterItem] + version: str | None + timestamp: float + + +@dataclass +class PrivilegeData: + """Data structure for privilege tracking.""" + + roster_access: RosterPrivilegeType = "none" + roster_push: bool = False + message_privilege: MessagePrivilegeType = "none" + presence_privilege: PresencePrivilegeType = "none" + iq_permissions: dict[str, IQPermission] = field(default_factory=dict) + + +class PrivilegeError(exceptions.PermissionError): + """A privileged operation failed""" + + pass + + +@dataclass +class ClientPrivilegeData: + privileges: PrivilegeData = field(default_factory=lambda: PrivilegeData()) + roster_cache: dict[jid.JID, RosterCacheEntry] = field(default_factory=dict) + + +class XEP_0356: + """XEP-0356: Privileged Entity handler.""" + + namespace = NS_PRIVILEGE + + def __init__(self, host: "LiberviaBackend") -> None: + log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + self._forward = cast(XEP_0297, host.plugins["XEP-0297"]) + host.register_namespace("privilege", NS_PRIVILEGE) + + def get_handler(self, client: SatXMPPEntity) -> "XEP0356Handler": + """Return the handler for this plugin. + + @param client: The client entity. + @return: The XEP0356Handler instance. + """ + return XEP0356Handler(self) + + def profile_connected(self, client: SatXMPPEntity) -> None: + """Called when a client is connected. + + Observers are now registered in XEP0356Handler.connectionInitialized() + to avoid race conditions with privilege advertisements. + + @param client: The client connected. + """ + if not client.is_component: + return + log.debug(f"XEP-0356 profile_connected called for {client.profile}") + + def can_access_roster(self, client: SatXMPPComponent) -> bool: + """Check if this entity can access user rosters.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + return data.privileges.roster_access in ("get", "both") + + def can_modify_roster(self, client: SatXMPPComponent) -> bool: + """Check if this entity can modify user rosters.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + return data.privileges.roster_access in ("set", "both") + + def can_receive_roster_pushes(self, client: SatXMPPComponent) -> bool: + """Check if this entity receives roster pushes.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + return data.privileges.roster_push + + def can_send_messages(self, client: SatXMPPComponent) -> bool: + """Check if this entity can send privileged messages.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + return data.privileges.message_privilege == "outgoing" + + def can_receive_managed_presence(self, client: SatXMPPComponent) -> bool: + """Check if this entity can receive presence for managed entities.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + return data.privileges.presence_privilege in ("managed_entity", "roster") + + def can_receive_roster_presence(self, client: SatXMPPComponent) -> bool: + """Check if this entity can receive presence for roster contacts.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + return data.privileges.presence_privilege == "roster" + + def can_send_iq(self, client: SatXMPPComponent, namespace: str, iq_type: str) -> bool: + """Check if this entity can send privileged IQ for a namespace/type. + + @param client: The client to check. + @param namespace: The namespace of the IQ payload. + @param iq_type: The type of IQ ("get" or "set"). + @return: True if the entity can send this IQ. + """ + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + perms = data.privileges.iq_permissions.get(namespace) + if perms is None: + return False + return perms.get if iq_type == "get" else perms.set + + def on_advertise(self, msg: domish.Element, client: SatXMPPComponent) -> None: + """Handle privilege advertisement from the server. + + @param msg: The message stanza containing privilege information. + @param client: The client receiving the advertisement. + """ + try: + privilege_elt = next(msg.elements(NS_PRIVILEGE, "privilege")) + except StopIteration: + log.warning("Privilege advertisement without privilege element") + return + + server_jid = client.server_jid + if jid.JID(msg["from"]) != server_jid: + log.warning( + f"SECURITY: privilege advertisement from unexpected server: {msg['from']}" + ) + raise error.StanzaError("not-allowed") + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + data.privileges = PrivilegeData() + + for perm in privilege_elt.elements(NS_PRIVILEGE, "perm"): + try: + access = perm["access"] + except KeyError: + log.warning(f"perm element without access attribute: {perm.toXml()}") + continue + + if access == "roster": + self._handle_roster_perm(data, perm) + elif access == "message": + self._handle_message_perm(data, perm) + elif access == "presence": + self._handle_presence_perm(data, perm) + elif access == "iq": + self._handle_iq_perm(data, perm) + else: + log.warning(f"Unknown privilege access type: {access}") + + log.info( + f"Privileges updated for {client.profile}:\n" + f" roster: {data.privileges.roster_access} " + f"(push={data.privileges.roster_push})\n" + f" message: {data.privileges.message_privilege}\n" + f" presence: {data.privileges.presence_privilege}\n" + f" iq: {list(data.privileges.iq_permissions.keys())}" + ) + + def _handle_roster_perm( + self, data: ClientPrivilegeData, perm: domish.Element + ) -> None: + """Handle roster permission element.""" + try: + perm_type = perm["type"] + except KeyError: + log.warning(f"roster perm without type: {perm.toXml()}") + return + + if perm_type not in ("none", "get", "set", "both"): + log.warning(f"Invalid roster permission type: {perm_type}") + return + + data.privileges.roster_access = cast(RosterPrivilegeType, perm_type) + + push_attr = perm.getAttribute("push") + if push_attr is not None: + data.privileges.roster_push = push_attr.lower() == "true" + else: + data.privileges.roster_push = perm_type in ("get", "both") + + def _handle_message_perm( + self, data: ClientPrivilegeData, perm: domish.Element + ) -> None: + """Handle message permission element.""" + try: + perm_type = perm["type"] + except KeyError: + log.warning(f"message perm without type: {perm.toXml()}") + return + + if perm_type not in ("none", "outgoing"): + log.warning(f"Invalid message permission type: {perm_type}") + return + + data.privileges.message_privilege = cast(MessagePrivilegeType, perm_type) + + def _handle_presence_perm( + self, data: ClientPrivilegeData, perm: domish.Element + ) -> None: + """Handle presence permission element.""" + try: + perm_type = perm["type"] + except KeyError: + log.warning(f"presence perm without type: {perm.toXml()}") + return + + if perm_type not in ("none", "managed_entity", "roster"): + log.warning(f"Invalid presence permission type: {perm_type}") + return + + data.privileges.presence_privilege = cast(PresencePrivilegeType, perm_type) + + def _handle_iq_perm(self, data: ClientPrivilegeData, perm: domish.Element) -> None: + """Handle IQ permission element.""" + for ns_elt in perm.elements(NS_PRIVILEGE, "namespace"): + try: + ns = ns_elt["ns"] + perm_type = ns_elt["type"] + except KeyError: + log.warning( + f"namespace element without required attributes: {ns_elt.toXml()}" + ) + continue + + if perm_type not in ("get", "set", "both"): + log.warning(f"Invalid IQ permission type: {perm_type}") + continue + + data.privileges.iq_permissions[ns] = IQPermission( + get=perm_type in ("get", "both"), + set=perm_type in ("set", "both"), + ) + + def on_roster_push(self, client: SatXMPPComponent, iq_elt: domish.Element) -> None: + """Handle roster push from server. + + @param client: The client receiving the roster push. + @param iq_elt: The roster push IQ stanza. + """ + if not self.can_receive_roster_pushes(client): + log.debug("Roster push received but not configured") + return + + from_jid_str = iq_elt.getAttribute("from") + if not from_jid_str: + log.warning(f"Roster push without 'from' attribute: {iq_elt.toXml()}") + return + + from_jid = jid.JID(from_jid_str) + if from_jid.resource: + log.warning( + f"Roster push from full JID instead of bare JID: {iq_elt.toXml()}" + ) + return + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + cache = data.roster_cache + + # Parse roster item using tx-xmpp + query = next(iq_elt.elements(NS_ROSTER, "query")) + item_elt = next(query.elements(NS_ROSTER, "item")) + roster_item = RosterItem.fromElement(item_elt) + item_jid = roster_item.entity + + cached_entry = cache.get(from_jid) + roster = cached_entry.roster if cached_entry else {} + if roster_item.remove: + roster.pop(item_jid, None) + else: + roster[item_jid] = roster_item + + version = query.getAttribute("ver") + cache[from_jid] = RosterCacheEntry( + roster=roster, + version=version, + timestamp=reactor.seconds(), + ) + + log.debug(f"Roster push for {from_jid}: {item_jid}") + + async def get_roster( + self, client: SatXMPPComponent, user_jid: jid.JID + ) -> dict[jid.JID, RosterItem]: + """Get the roster of a managed entity. + + @param client: The client making the request. + @param user_jid: The bare JID of the user whose roster to retrieve. + @return: Dict mapping JIDs to RosterItem objects. + @raise PrivilegeError: If roster access is not permitted. + """ + if not self.can_access_roster(client): + raise PrivilegeError(f"Roster access not permitted for {user_jid}") + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + cache = data.roster_cache + cached = cache.get(user_jid) + if cached and (reactor.seconds() - cached.timestamp) < ROSTER_TTL: + return cached.roster + + iq = client.IQ("get") + iq["to"] = user_jid.full() + query = domish.Element((NS_ROSTER, "query")) + iq.addChild(query) + + result = await iq.send() + + roster = {} + for item_elt in result.elements(NS_ROSTER, "item"): + roster_item = RosterItem.fromElement(item_elt) + roster[roster_item.entity] = roster_item + + version = query.getAttribute("ver") + cache[user_jid] = RosterCacheEntry( + roster=roster, + version=version, + timestamp=reactor.seconds(), + ) + + return roster + + async def send_privileged_message( + self, client: SatXMPPComponent, message_elt: domish.Element + ) -> None: + """Send a message on behalf of a user. + + @param client: The client sending the message. + @param message_elt: The message stanza to send (will be wrapped in forwarded). + @raise PrivilegeError: If message privilege is not permitted. + """ + if not self.can_send_messages(client): + raise PrivilegeError("Message privilege not permitted") + + try: + from_attr = message_elt["from"] + except KeyError: + raise PrivilegeError('Message must have a "from" attribute') + + from_jid = jid.JID(from_attr) + if from_jid.resource or jid.JID(from_jid.host) != client.server_jid: + raise PrivilegeError( + f"from_jid must be a bare JID from our server: {from_jid}" + ) + + wrapper = domish.Element((None, "message")) + wrapper["from"] = client.jid.full() + wrapper["to"] = client.server_jid.full() + wrapper["id"] = str(uuid.uuid4()) + + privilege = domish.Element((NS_PRIVILEGE, "privilege")) + forwarded = self._forward.build_forwarded(message_elt) + privilege.addChild(forwarded) + wrapper.addChild(privilege) + + client.xmlstream.send(wrapper) + + async def send_privileged_iq( + self, client: SatXMPPComponent, iq_elt: domish.Element + ) -> domish.Element: + """Send an IQ stanza on behalf of a user. + + @param client: The client sending the IQ. + @param iq_elt: The IQ stanza to send (will be wrapped). + @return: The IQ response. + @raise PrivilegeError: If IQ privilege is not permitted. + """ + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + if not data.privileges.iq_permissions: + raise PrivilegeError("IQ privilege not permitted") + + to_attr = iq_elt.getAttribute("to") + if not to_attr: + raise PrivilegeError("IQ must have a 'to' attribute") + + user_jid = jid.JID(to_attr) + if user_jid.resource or jid.JID(user_jid.host) != client.server_jid: + raise PrivilegeError( + f"user_jid must be a bare JID from our server: {user_jid}" + ) + + iq_type = cast(str, iq_elt.getAttribute("type", "")) + first_child = iq_elt.firstChildElement() + if first_child is None: + raise PrivilegeError("IQ has no payload") + + namespace = first_child.uri + if not self.can_send_iq(client, namespace, iq_type): + raise PrivilegeError( + f"IQ privilege not permitted for namespace {namespace} type {iq_type}" + ) + + wrapper = client.IQ(iq_type) + wrapper["to"] = user_jid.full() + + priv_iq = domish.Element((NS_PRIVILEGE, "privileged_iq")) + iq_copy = domish.Element(("jabber:client", "iq")) + iq_copy["type"] = iq_type + iq_copy["id"] = iq_elt.getAttribute("id", str(uuid.uuid4())) + if iq_elt.getAttribute("to"): + iq_copy["to"] = iq_elt["to"] + iq_copy.addChild(first_child) + priv_iq.addChild(iq_copy) + wrapper.addChild(priv_iq) + + result = await wrapper.send() + return self._unwrap_privileged_iq_result(result) + + def _unwrap_privileged_iq_result( + self, iq_result_elt: domish.Element + ) -> domish.Element: + """Unwrap a privileged IQ response. + + @param iq_result_elt: The IQ result stanza. + @return: The unwrapped inner IQ. + @raise PrivilegeError: If the response format is invalid. + """ + try: + privilege = next(iq_result_elt.elements(NS_PRIVILEGE, "privilege")) + forwarded = next(privilege.elements(NS_FORWARD, "forwarded")) + return next(forwarded.elements("jabber:client", "iq")) + except StopIteration: + log.error(f"Invalid privileged IQ response: {iq_result_elt.toXml()}") + raise PrivilegeError("Invalid response format") + + def _on_presence( + self, presence_elt: domish.Element, client: SatXMPPComponent + ) -> None: + """Handle incoming presence stanza. + + @param presence_elt: The presence stanza. + @param client: The client instance. + """ + try: + from_jid = jid.JID(presence_elt["from"]) + except: + log.exception("Can't parse 'from' attribute.") + raise + + G.host.trigger.point("XEP-0356_presence_received", client, from_jid, presence_elt) + + +@implementer(itx_xmpp.IDisco) +class XEP0356Handler(XMPPHandler): + + def __init__(self, plugin_parent: XEP_0356) -> None: + self.plugin_parent = plugin_parent + + @property + def client(self) -> SatXMPPComponent: + """Get the client (component) this handler is attached to.""" + assert self.parent is not None + return cast(SatXMPPComponent, self.parent) + + def connectionInitialized(self) -> None: + """Register observers after stream initialization. + + This is called immediately after authentication, ensuring we don't miss + privilege advertisements that may be sent by the server right after + the component authenticates. + """ + if not self.client.is_component: + return + + self.client.xmlstream.addObserver( + f"/message/privilege[@xmlns='{NS_PRIVILEGE}']", + self.plugin_parent.on_advertise, + client=self.client, + ) + self.client.xmlstream.addObserver( + "/presence", self.plugin_parent._on_presence, client=self.client + ) + log.debug(f"XEP-0356 observers registered for {self.client.profile}") + + def getDiscoInfo( + self, + requestor: jid.JID, + target: jid.JID, + nodeIdentifier: str = "", + ) -> list[disco.DiscoFeature | disco.DiscoIdentity]: + """Get disco info. + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: Node identifier + @return: Disco features + """ + return [disco.DiscoFeature(NS_PRIVILEGE)] + + def getDiscoItems( + self, + requestor: jid.JID, + target: jid.JID, + nodeIdentifier: str = "", + ) -> list[disco.DiscoItem]: + """Get disco items. + + @param requestor: JID of the requesting entity + @param target: JID of the target entity + @param nodeIdentifier: Node identifier + @return: Disco items + """ + return [] diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index f30a84ae..5fe73b7d 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -20,6 +20,7 @@ from unittest.mock import MagicMock, AsyncMock from pytest import fixture from twisted.internet import defer from twisted.words.protocols.jabber import jid +from twisted.words.xish import domish from libervia.backend import G from tx_xmpp.data_form import NS_X_DATA from libervia.backend.core.main import LiberviaBackend @@ -94,6 +95,49 @@ def host(bridge, storage): return host +@fixture +def client_IQ(): + """Factory fixture to create a mock IQ method for testing. + + Usage: + def test_something(client, client_IQ): + response_elt = domish.Element((None, "iq")) + response_elt["type"] = "result" + client.IQ = client_IQ(response_elt=response_elt) + + # Later in test: + assert len(client.IQ.sent_iqs) == 1 + + @param response: Optional response to return from send(). If None, returns empty IQ. + @return: The mock IQ method with attached sent_iqs list. + """ + + def factory(response_elt=None): + sent_iqs = [] + + def client_IQ(type_="set", timeout=60): + iq_elt = domish.Element((None, "iq")) + iq_elt["type"] = type_ + + async def mock_send(): + sent_iqs.append(iq_elt) + if response_elt is not None: + return response_elt + # Return empty IQ by default + empty_response = domish.Element((None, "iq")) + empty_response["type"] = "result" + return empty_response + + iq_elt.send = mock_send + return iq_elt + + # Attach list for test assertions + client_IQ.sent_iqs = sent_iqs + return client_IQ + + return factory + + @fixture def client(): client = MagicMock() diff --git a/tests/unit/test_xep_0356.py b/tests/unit/test_xep_0356.py new file mode 100644 index 00000000..567033a3 --- /dev/null +++ b/tests/unit/test_xep_0356.py @@ -0,0 +1,708 @@ +#!/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-0356 Privileged Entity implementation. +""" + +from unittest.mock import MagicMock + +from pytest import fixture, raises +from pytest_twisted import ensureDeferred as ed +from twisted.internet import reactor +from twisted.words.protocols.jabber import jid +from twisted.words.xish import domish + +from libervia.backend import G +from libervia.backend.plugins.plugin_xep_0356 import ( + XEP_0356, + NS_PRIVILEGE, + ClientPrivilegeData, + RosterCacheEntry, + IMPORT_NAME, + PrivilegeError, +) +from tx_xmpp import disco +from tx_xmpp.xmppim import RosterItem, NS_ROSTER + + +@fixture +def xep_0356(host): + """Create XEP-0356 plugin instance.""" + return XEP_0356(host) + + +@fixture(autouse=True) +def setup_component_client(client): + """Set up client as a component for XEP-0356 tests.""" + client.is_component = True + _priv_data = ClientPrivilegeData() + client.get_plugin_data = lambda import_name, data_class: _priv_data + + +def make_privilege_advertisement(privileges): + """Helper to create privilege advertisement message. + + @param privileges: Dict of privilege type -> access level + e.g., {"roster": "both", "message": "outgoing"} + """ + msg = domish.Element( + (None, "message"), + attribs={"from": "test.example", "to": "pubsub.test.example"}, + ) + privilege = domish.Element((NS_PRIVILEGE, "privilege")) + + for access, perm_type in privileges.items(): + perm_attrs = {"access": access} + if perm_type: + perm_attrs["type"] = perm_type + if access == "roster" and perm_type in ("get", "both"): + perm_attrs["push"] = "true" + perm = domish.Element((NS_PRIVILEGE, "perm"), attribs=perm_attrs) + privilege.addChild(perm) + + msg.addChild(privilege) + return msg + + +def make_iq_privilege_advertisement(namespaces): + """Helper to create IQ privilege advertisement. + + @param namespaces: Dict of namespace -> type + e.g., {"http://jabber.org/protocol/pubsub": "set"} + """ + msg = domish.Element( + (None, "message"), + attribs={"from": "test.example", "to": "pubsub.test.example"}, + ) + privilege = domish.Element((NS_PRIVILEGE, "privilege")) + perm = domish.Element((NS_PRIVILEGE, "perm"), attribs={"access": "iq"}) + + for ns, perm_type in namespaces.items(): + ns_elt = domish.Element( + (NS_PRIVILEGE, "namespace"), + attribs={"ns": ns, "type": perm_type}, + ) + perm.addChild(ns_elt) + + privilege.addChild(perm) + msg.addChild(privilege) + return msg + + +class TestXEP0356Advertisement: + """Tests for privilege advertisement handling.""" + + @ed + async def test_parses_roster_privilege_both(self, xep_0356, client): + """Roster privilege with type='both' sets access correctly.""" + msg = make_privilege_advertisement({"roster": "both"}) + xep_0356.on_advertise(msg, client) + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert data.privileges.roster_access == "both" + assert xep_0356.can_access_roster(client) + assert xep_0356.can_modify_roster(client) + assert xep_0356.can_receive_roster_pushes(client) + + @ed + async def test_parses_roster_privilege_get(self, xep_0356, client): + """Roster privilege with type='get' allows read-only access.""" + msg = make_privilege_advertisement({"roster": "get"}) + xep_0356.on_advertise(msg, client) + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert data.privileges.roster_access == "get" + assert xep_0356.can_access_roster(client) + assert not xep_0356.can_modify_roster(client) + assert xep_0356.can_receive_roster_pushes(client) + + @ed + async def test_parses_roster_privilege_set(self, xep_0356, client): + """Roster privilege with type='set' allows write-only access.""" + msg = make_privilege_advertisement({"roster": "set"}) + xep_0356.on_advertise(msg, client) + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert data.privileges.roster_access == "set" + assert not xep_0356.can_access_roster(client) + assert xep_0356.can_modify_roster(client) + assert not xep_0356.can_receive_roster_pushes(client) + + @ed + async def test_parses_message_privilege(self, xep_0356, client): + """Message privilege is correctly set.""" + msg = make_privilege_advertisement({"message": "outgoing"}) + xep_0356.on_advertise(msg, client) + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert data.privileges.message_privilege == "outgoing" + assert xep_0356.can_send_messages(client) + + @ed + async def test_parses_presence_privilege_managed_entity(self, xep_0356, client): + """Presence privilege with managed_entity type.""" + msg = make_privilege_advertisement({"presence": "managed_entity"}) + xep_0356.on_advertise(msg, client) + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert data.privileges.presence_privilege == "managed_entity" + assert xep_0356.can_receive_managed_presence(client) + assert not xep_0356.can_receive_roster_presence(client) + + @ed + async def test_parses_presence_privilege_roster(self, xep_0356, client): + """Presence privilege with roster type.""" + msg = make_privilege_advertisement({"presence": "roster"}) + xep_0356.on_advertise(msg, client) + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert data.privileges.presence_privilege == "roster" + assert xep_0356.can_receive_managed_presence(client) + assert xep_0356.can_receive_roster_presence(client) + + @ed + async def test_parses_iq_privilege(self, xep_0356, client): + """IQ privilege with namespace is correctly set.""" + msg = make_iq_privilege_advertisement( + {"http://jabber.org/protocol/pubsub": "set"} + ) + xep_0356.on_advertise(msg, client) + + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert "http://jabber.org/protocol/pubsub" in data.privileges.iq_permissions + assert xep_0356.can_send_iq(client, "http://jabber.org/protocol/pubsub", "set") + assert not xep_0356.can_send_iq( + client, "http://jabber.org/protocol/pubsub", "get" + ) + + @ed + async def test_parses_iq_privilege_both(self, xep_0356, client): + """IQ privilege with type='both' allows get and set.""" + msg = make_iq_privilege_advertisement( + {"http://jabber.org/protocol/pubsub": "both"} + ) + xep_0356.on_advertise(msg, client) + + assert xep_0356.can_send_iq(client, "http://jabber.org/protocol/pubsub", "get") + assert xep_0356.can_send_iq(client, "http://jabber.org/protocol/pubsub", "set") + + @ed + async def test_rejects_advertisement_from_wrong_server(self, xep_0356, client): + """Advertisement from wrong server raises StanzaError.""" + from twisted.words.protocols.jabber.error import StanzaError + + msg = make_privilege_advertisement({"roster": "both"}) + msg["from"] = "other.example" + + with raises(StanzaError, match="not-allowed"): + xep_0356.on_advertise(msg, client) + + +class TestXEP0356RosterAccess: + """Tests for roster access operations.""" + + @ed + async def test_get_roster_without_privilege_raises(self, xep_0356, client): + """get_roster raises PrivilegeError without roster privilege.""" + user_jid = jid.JID("user@test.example") + + with raises(PrivilegeError, match="Roster access not permitted"): + await xep_0356.get_roster(client, user_jid) + + @ed + async def test_get_roster_sends_request(self, xep_0356, client, client_IQ): + """get_roster sends proper roster get request.""" + # Grant privilege + msg = make_privilege_advertisement({"roster": "both"}) + xep_0356.on_advertise(msg, client) + + user_jid = jid.JID("user@test.example") + contact_jid = jid.JID("contact@test.example") + + # Create mock IQ with roster response + response_elt = domish.Element((None, "iq")) + response_elt["type"] = "result" + item = domish.Element((NS_ROSTER, "item")) + item["jid"] = contact_jid.full() + item["subscription"] = "both" + response_elt.addChild(item) + + client.IQ = client_IQ(response_elt=response_elt) + + # Call get_roster + roster = await xep_0356.get_roster(client, user_jid) + + # We check that the request was sent correctly + assert len(client.IQ.sent_iqs) == 1 + sent_iq = client.IQ.sent_iqs[0] + assert sent_iq["type"] == "get" + assert sent_iq["to"] == user_jid.full() + assert len(list(sent_iq.elements("jabber:iq:roster", "query"))) == 1 + + # We check that roster was parsed correctly + assert len(roster) == 1 + assert contact_jid in roster + assert isinstance(roster[contact_jid], RosterItem) + assert roster[contact_jid].subscriptionTo + assert roster[contact_jid].subscriptionFrom + + @ed + async def test_get_roster_uses_cache(self, xep_0356, client, client_IQ): + """get_roster uses cached roster if not expired.""" + # Grant privilege + msg = make_privilege_advertisement({"roster": "both"}) + xep_0356.on_advertise(msg, client) + + user_jid = jid.JID("user@test.example") + + # Pre-populate cache + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + cached_item = RosterItem(user_jid) + cached_item.subscriptionTo = True + data.roster_cache[user_jid] = RosterCacheEntry( + roster={user_jid: cached_item}, + version="v1", + # Fresh timestamp + timestamp=reactor.seconds(), + ) + + client.IQ = client_IQ() + + roster = await xep_0356.get_roster(client, user_jid) + + # We check that cache was used (no IQ sent) + assert len(client.IQ.sent_iqs) == 0 + assert len(roster) == 1 + + @ed + async def test_on_roster_push_updates_cache(self, xep_0356, client): + """on_roster_push updates the roster cache.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + # We want to be sure that roster is empty + assert len(data.roster_cache) == 0 + + # Grant roster push privilege + msg = make_privilege_advertisement({"roster": "both"}) + xep_0356.on_advertise(msg, client) + + owner_jid = jid.JID("user@test.example") + + # Create roster push stanza + iq = domish.Element((None, "iq")) + iq["from"] = owner_jid.full() + iq["type"] = "set" + query = domish.Element(("jabber:iq:roster", "query")) + item = domish.Element(("jabber:iq:roster", "item")) + item["jid"] = "contact@test.example" + item["subscription"] = "both" + query.addChild(item) + iq.addChild(query) + + xep_0356.on_roster_push(client, iq) + + # We check that cache was updated + assert owner_jid in data.roster_cache + cached_entry = data.roster_cache[owner_jid] + assert isinstance(cached_entry, RosterCacheEntry) + cached_roster = cached_entry.roster + assert len(cached_roster) == 1 + contact_jid = jid.JID("contact@test.example") + assert contact_jid in cached_roster + assert isinstance(cached_roster[contact_jid], RosterItem) + + @ed + async def test_on_roster_push_remove_item(self, xep_0356, client): + """on_roster_push removes item when subscription='remove'.""" + # Grant roster push privilege + msg = make_privilege_advertisement({"roster": "both"}) + xep_0356.on_advertise(msg, client) + + owner_jid = jid.JID("user@test.example") + + # Pre-populate cache with an item + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + existing_item = RosterItem(jid.JID("contact@test.example")) + data.roster_cache[owner_jid] = RosterCacheEntry( + roster={jid.JID("contact@test.example"): existing_item}, + version="v1", + timestamp=reactor.seconds(), + ) + + # Create roster push with remove + iq = domish.Element((None, "iq")) + iq["from"] = owner_jid.full() + iq["type"] = "set" + query = domish.Element(("jabber:iq:roster", "query")) + item = domish.Element(("jabber:iq:roster", "item")) + item["jid"] = "contact@test.example" + item["subscription"] = "remove" + query.addChild(item) + iq.addChild(query) + + xep_0356.on_roster_push(client, iq) + + # We check that item was removed + cached_entry = data.roster_cache[owner_jid] + cached_roster = cached_entry.roster + assert len(cached_roster) == 0 + + @ed + async def test_on_roster_push_without_privilege(self, xep_0356, client): + """on_roster_push is ignored without roster push privilege.""" + # Don't grant privilege + owner_jid = jid.JID("user@test.example") + + iq = domish.Element((None, "iq")) + iq["from"] = owner_jid.full() + iq["type"] = "set" + query = domish.Element(("jabber:iq:roster", "query")) + iq.addChild(query) + + xep_0356.on_roster_push(client, iq) + + # We check that cache was not updated + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + assert owner_jid not in data.roster_cache + + +class TestXEP0356PrivilegedMessages: + """Tests for privileged message sending.""" + + @ed + async def test_send_privileged_message_without_privilege_raises( + self, xep_0356, client + ): + """send_privileged_message raises PrivilegeError without message privilege.""" + message = domish.Element((None, "message")) + message["from"] = "user@test.example" + + with raises(PrivilegeError, match="Message privilege not permitted"): + await xep_0356.send_privileged_message(client, message) + + @ed + async def test_send_privileged_message_without_from_raises(self, xep_0356, client): + """send_privileged_message raises PrivilegeError if message has no 'from'.""" + # Grant privilege + msg = make_privilege_advertisement({"message": "outgoing"}) + xep_0356.on_advertise(msg, client) + + message = domish.Element((None, "message")) + + with raises(PrivilegeError, match='Message must have a "from" attribute'): + await xep_0356.send_privileged_message(client, message) + + @ed + async def test_send_privileged_message_wrong_server_raises(self, xep_0356, client): + """send_privileged_message raises if from is not from our server.""" + # Grant privilege + msg = make_privilege_advertisement({"message": "outgoing"}) + xep_0356.on_advertise(msg, client) + + message = domish.Element((None, "message")) + message["from"] = "user@other.example" + + with raises(PrivilegeError, match="from_jid must be a bare JID from our server"): + await xep_0356.send_privileged_message(client, message) + + @ed + async def test_send_privileged_message_with_resource_raises(self, xep_0356, client): + """send_privileged_message raises if from has a resource.""" + # Grant privilege + msg = make_privilege_advertisement({"message": "outgoing"}) + xep_0356.on_advertise(msg, client) + + message = domish.Element((None, "message")) + message["from"] = "user@test.example/resource" + + with raises(PrivilegeError, match="from_jid must be a bare JID from our server"): + await xep_0356.send_privileged_message(client, message) + + @ed + async def test_send_privileged_message_sends_wrapped_message(self, xep_0356, client): + """send_privileged_message sends properly wrapped message.""" + # Grant privilege + msg = make_privilege_advertisement({"message": "outgoing"}) + xep_0356.on_advertise(msg, client) + + from_jid = jid.JID("user@test.example") + to_jid = jid.JID("other@test.example") + inner_msg = domish.Element( + (None, "message"), + attribs={"from": from_jid.full(), "to": to_jid.full()}, + ) + body = domish.Element((None, "body")) + body.textContent = "Hello" + inner_msg.addChild(body) + + # Track sent messages + sent_messages = [] + + def mock_send(elt): + sent_messages.append(elt) + + client.xmlstream.send = mock_send + + # Call send_privileged_message + await xep_0356.send_privileged_message(client, inner_msg) + + # We check that message was sent + assert len(sent_messages) == 1 + wrapper = sent_messages[0] + + assert wrapper.name == "message" + assert wrapper["from"] == client.jid.full() + assert wrapper["to"] == client.server_jid.full() + assert "id" in wrapper.attributes + + # We check that privilege wrapper + privilege = next(wrapper.elements(NS_PRIVILEGE, "privilege")) + assert privilege is not None + + +class TestXEP0356PrivilegedIQ: + """Tests for privileged IQ sending.""" + + @ed + async def test_send_privileged_iq_without_privilege_raises(self, xep_0356, client): + """send_privileged_iq raises PrivilegeError without IQ privilege.""" + iq = domish.Element(("jabber:client", "iq")) + iq["type"] = "set" + iq["to"] = "user@test.example" + payload = domish.Element(("http://jabber.org/protocol/pubsub", "publish")) + iq.addChild(payload) + + with raises(PrivilegeError, match="IQ privilege not permitted"): + await xep_0356.send_privileged_iq(client, iq) + + @ed + async def test_send_privileged_iq_no_to_raises(self, xep_0356, client): + """send_privileged_iq raises if IQ has no 'to' attribute.""" + # Grant privilege + msg = make_iq_privilege_advertisement( + {"http://jabber.org/protocol/pubsub": "set"} + ) + xep_0356.on_advertise(msg, client) + + iq = domish.Element(("jabber:client", "iq")) + iq["type"] = "set" + payload = domish.Element(("http://jabber.org/protocol/pubsub", "publish")) + iq.addChild(payload) + + with raises(PrivilegeError, match="IQ must have a 'to' attribute"): + await xep_0356.send_privileged_iq(client, iq) + + @ed + async def test_send_privileged_iq_wrong_server_raises(self, xep_0356, client): + """send_privileged_iq raises if to is not from our server.""" + # Grant privilege + msg = make_iq_privilege_advertisement( + {"http://jabber.org/protocol/pubsub": "set"} + ) + xep_0356.on_advertise(msg, client) + + iq = domish.Element(("jabber:client", "iq")) + iq["type"] = "set" + iq["to"] = "user@other.example" + payload = domish.Element(("http://jabber.org/protocol/pubsub", "publish")) + iq.addChild(payload) + + with raises(PrivilegeError, match="user_jid must be a bare JID from our server"): + await xep_0356.send_privileged_iq(client, iq) + + @ed + async def test_send_privileged_iq_no_payload_raises(self, xep_0356, client): + """send_privileged_iq raises if IQ has no payload.""" + # Grant privilege + msg = make_iq_privilege_advertisement( + {"http://jabber.org/protocol/pubsub": "set"} + ) + xep_0356.on_advertise(msg, client) + + iq = domish.Element(("jabber:client", "iq")) + iq["type"] = "set" + iq["to"] = "user@test.example" + + with raises(PrivilegeError, match="IQ has no payload"): + await xep_0356.send_privileged_iq(client, iq) + + @ed + async def test_send_privileged_iq_unauthorized_namespace_raises( + self, xep_0356, client + ): + """send_privileged_iq raises for unauthorized namespace.""" + # Grant privilege for pubsub only + msg = make_iq_privilege_advertisement( + {"http://jabber.org/protocol/pubsub": "set"} + ) + xep_0356.on_advertise(msg, client) + + iq = domish.Element(("jabber:client", "iq")) + iq["type"] = "set" + iq["to"] = "user@test.example" + payload = domish.Element(("http://jabber.org/protocol/bookmarks", "bookmarks")) + iq.addChild(payload) + + with raises(PrivilegeError, match="IQ privilege not permitted"): + await xep_0356.send_privileged_iq(client, iq) + + @ed + async def test_send_privileged_iq_sends_wrapped_iq(self, xep_0356, client, client_IQ): + """send_privileged_iq sends properly wrapped IQ.""" + # Grant privilege + msg = make_iq_privilege_advertisement( + {"http://jabber.org/protocol/pubsub": "set"} + ) + xep_0356.on_advertise(msg, client) + + user_jid = jid.JID("user@test.example") + inner_iq = domish.Element( + ("jabber:client", "iq"), attribs={"type": "set", "id": "inner1"} + ) + pubsub = domish.Element(("http://jabber.org/protocol/pubsub", "publish")) + pubsub["node"] = "test_node" + inner_iq.addChild(pubsub) + inner_iq["to"] = user_jid.full() + + # Create mock response + response_elt = domish.Element((None, "iq")) + response_elt["type"] = "result" + privilege = domish.Element((NS_PRIVILEGE, "privilege")) + forwarded = domish.Element(("urn:xmpp:forward:0", "forwarded")) + inner_response = domish.Element(("jabber:client", "iq")) + inner_response["type"] = "result" + inner_response["id"] = "inner1" + forwarded.addChild(inner_response) + privilege.addChild(forwarded) + response_elt.addChild(privilege) + + client.IQ = client_IQ(response_elt=response_elt) + + result = await xep_0356.send_privileged_iq(client, inner_iq) + + # We check that the request was sent correctly + assert len(client.IQ.sent_iqs) == 1 + sent_iq = client.IQ.sent_iqs[0] + assert sent_iq["type"] == "set" + assert sent_iq["to"] == user_jid.full() + + # We check that privilege wrapper + priv_iq = next(sent_iq.elements(NS_PRIVILEGE, "privileged_iq")) + assert priv_iq is not None + + # We check that result was unwrapped + assert result["type"] == "result" + assert result["id"] == "inner1" + + @ed + async def test_unwrap_privileged_iq_result_invalid_raises(self, xep_0356): + """_unwrap_privileged_iq_result raises for invalid response format.""" + # Invalid response (no privilege wrapper) + invalid_response = domish.Element((None, "iq")) + invalid_response["type"] = "result" + + with raises(PrivilegeError, match="Invalid response format"): + xep_0356._unwrap_privileged_iq_result(invalid_response) + + +class TestXEP0356PresenceHandling: + """Tests for presence handling.""" + + @ed + async def test_profile_connected_does_not_register_observers(self, xep_0356, client): + """profile_connected no longer registers observers (done by handler instead).""" + assert client.is_component + + # Mock addObserver to track calls + observer_calls = [] + + def mock_add_observer(xpath, handler, client=None): + observer_calls.append((xpath, handler, client)) + + client.xmlstream.addObserver = mock_add_observer + + # Call profile_connected + xep_0356.profile_connected(client) + + # We check that no observers were registered (handler does it instead) + assert len(observer_calls) == 0 + + @ed + async def test_profile_connected_skips_non_component(self, xep_0356, host): + """profile_connected skips non-component clients.""" + # Create a non-component client + non_component_client = MagicMock() + non_component_client.is_component = False + non_component_client.jid = jid.JID("user@test.example/resource") + + # Mock addObserver to track calls + observer_calls = [] + + def mock_add_observer(xpath, handler, client=None): + observer_calls.append((xpath, handler, client)) + + non_component_client.xmlstream.addObserver = mock_add_observer + + xep_0356.profile_connected(non_component_client) + + # We check that observer was NOT registered + assert len(observer_calls) == 0 + + @ed + async def test_on_presence_triggers_async_point(self, xep_0356, client): + """_on_presence triggers async_point for any presence.""" + # Create presence stanza + presence = domish.Element((None, "presence")) + presence["from"] = "user@test.example" + + trigger_point_calls = [] + + def mock_trigger_point(point_name, *args): + trigger_point_calls.append((point_name, args)) + + G.host.trigger.point = mock_trigger_point + + xep_0356._on_presence(presence, client) + + assert len(trigger_point_calls) == 1 + point_name, args = trigger_point_calls[0] + assert point_name == "XEP-0356_presence_received" + assert len(args) == 3 + assert args[0] == client + assert args[1] == jid.JID("user@test.example") + assert args[2] == presence + + @ed + async def test_on_presence_with_unavailable_type(self, xep_0356, client): + """_on_presence triggers async_point for unavailable presence.""" + presence = domish.Element((None, "presence")) + presence["from"] = "user.test.example" + presence["type"] = "unavailable" + + trigger_point_calls = [] + + def mock_trigger_point(point_name, *args): + trigger_point_calls.append((point_name, args)) + + G.host.trigger.point = mock_trigger_point + + xep_0356._on_presence(presence, client) + + # We check that async_point was triggered + assert len(trigger_point_calls) == 1 + assert trigger_point_calls[0][0] == "XEP-0356_presence_received"