diff --git a/libervia/backend/memory/migration/versions/3ddac15a0a39_add_presence_access_model_to_pubsubnode.py b/libervia/backend/memory/migration/versions/3ddac15a0a39_add_presence_access_model_to_pubsubnode.py new file mode 100644 index 00000000..f4ca2ef4 --- /dev/null +++ b/libervia/backend/memory/migration/versions/3ddac15a0a39_add_presence_access_model_to_pubsubnode.py @@ -0,0 +1,51 @@ +"""add "presence" access model to PubsubNode + cleanup + +Revision ID: 3ddac15a0a39 +Revises: 007439b3b0ce +Create Date: 2026-04-15 08:55:32.891237 + +""" + +from alembic import op +from sqlalchemy import inspect +from sqlalchemy.schema import conv + +# revision identifiers, used by Alembic. +revision = "3ddac15a0a39" +down_revision = "007439b3b0ce" +branch_labels = None +depends_on = None + + +def _existing_check_constraints(table_name: str) -> set[str | None]: + insp = inspect(op.get_bind()) + return {c["name"] for c in insp.get_check_constraints(table_name) if c.get("name")} + + +def upgrade(): + existing = _existing_check_constraints("pubsub_nodes") + + with op.batch_alter_table("pubsub_nodes") as batch_op: + # remove legacy duplicate constraints if present + if "accessmodel" in existing: + batch_op.drop_constraint(conv("accessmodel"), type_="check") + if "publishmodel" in existing: + batch_op.drop_constraint(conv("publishmodel"), type_="check") + + if "ck_pubsub_nodes_accessmodel" in existing: + batch_op.drop_constraint(conv("ck_pubsub_nodes_accessmodel"), type_="check") + + batch_op.create_check_constraint( + "accessmodel", + "access_model IN ('open', 'presence', 'whitelist')", + ) + + +def downgrade(): + with op.batch_alter_table("pubsub_nodes") as batch_op: + batch_op.drop_constraint("accessmodel", type_="check") + + batch_op.create_check_constraint( + "accessmodel", + "access_model IN ('open', 'whitelist')", + ) diff --git a/libervia/backend/memory/sqla.py b/libervia/backend/memory/sqla.py index 643a430a..55f41bb4 100644 --- a/libervia/backend/memory/sqla.py +++ b/libervia/backend/memory/sqla.py @@ -28,7 +28,7 @@ from typing import Any, Dict, List, Optional, Tuple, Union from alembic import config as al_config, script as al_script from alembic.runtime import migration as al_migration -from sqlalchemy import and_, delete, event, exists, func, literal, or_, update +from sqlalchemy import and_, delete, desc, event, exists, func, literal, or_, true, update from sqlalchemy import Integer, literal_column, text from sqlalchemy.dialects.sqlite import insert from sqlalchemy.engine import Connection, Engine @@ -81,8 +81,8 @@ from libervia.backend.memory.sqla_mapping import ( PubsubItem, PubsubNode, PubsubSub, - SubscriptionState, Subject, + SubscriptionState, SyncState, Thread, get_profile_by_id, @@ -1211,8 +1211,8 @@ class Storage: @param with_parent: retrieve parent_node in the same query @param with_link: retrieve linked_node in the same query @param create: if the node doesn't exist in DB, create it - @param create_kwargs: keyword arguments to use with ``set_pubsub_node`` if the node - needs to be created. + @param create_kwargs: keyword arguments to use with ``set_pubsub_node`` if the + node needs to be created. @raise ValueError: if both name and node_id are provided, or neither. """ if node_id is not None: @@ -1430,6 +1430,70 @@ class Storage: result = await session.execute(stmt) return result.scalars().all() + @aio + async def get_pep_last_items( + self, + client: SatXMPPEntity, + publishers: list[jid.JID], + node_names: list[str], + access_models: list[AccessModel], + ) -> list[tuple[PubsubNode, PubsubItem]]: + """Get last items for PEP nodes. + + This retrieves the last item for each (publisher, node_name) combination in a + single query. + + @param client: Client instance for profile filtering. + @param publishers: List of bare JIDs that own PEP nodes. + @param node_names: List of node names to query (e.g., from +notify features). + @param access_models: Allowed node access models. + @return: List of (node, last_item) tuples. + """ + if not publishers or not node_names: + return [] + + profile_id = self.profiles.get(client.profile) + + async with self.session() as session: + # Use window function to rank items by node, ordered by updated DESC, id DESC + # ROW_NUMBER() assigns 1 to the latest item per node + ranked_cte = ( + select( + PubsubNode, + PubsubItem, + func.row_number() + .over( + partition_by=PubsubItem.node_id, + order_by=( + desc(PubsubItem.updated), + desc(PubsubItem.id), # Tie-breaker + ), + ) + .label("row_num"), + ) + .join( + PubsubNode, + PubsubItem.node_id == PubsubNode.id, + ) + .where( + PubsubNode.profile_id == profile_id, + PubsubNode.service.in_(publishers), + PubsubNode.name.in_(node_names), + PubsubNode.pep == true(), + PubsubNode.access_model.in_(access_models), + ) + .cte() + ) + + # Select only the latest item (row_num = 1) for each node + stmt = select( + ranked_cte.c.PubsubNode, + ranked_cte.c.PubsubItem, + ).where(ranked_cte.c.row_num == 1) + + result = await session.execute(stmt) + return result.all() + @aio async def set_pubsub_node( self, diff --git a/libervia/backend/memory/sqla_mapping.py b/libervia/backend/memory/sqla_mapping.py index 9eb93379..2e18d9c7 100644 --- a/libervia/backend/memory/sqla_mapping.py +++ b/libervia/backend/memory/sqla_mapping.py @@ -106,6 +106,7 @@ class SyncState(enum.Enum): class AccessModel(enum.StrEnum): open = enum.auto() + presence = enum.auto() whitelist = enum.auto() diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index 91ea33f4..6ea8e576 100644 --- a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py +++ b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py @@ -192,6 +192,8 @@ class PubsubResource(pubsub.PubSubResource): self.xep_0356 = cast(XEP_0356, G.host.plugins["XEP-0356"]) self.xep_0496 = cast(XEP_0496, G.host.plugins["XEP-0496"]) + G.host.trigger.add("XEP-0356_presence_received", self._on_presence_received) + @property def client(self) -> SatXMPPComponent: return self.service.client @@ -1669,6 +1671,82 @@ class PubsubResource(pubsub.PubSubResource): if not success: log.error(f"Failed to send notification: {result}") + async def _on_presence_received( + self, + client: SatXMPPComponent, + from_jid: jid.JID, + presence_elt: domish.Element, + ) -> None: + """Handle presence for PEP auto-subscription. + + When a subscriber comes online with +notify capability, send them the last item + from each relevant PEP node where they have presence subscription to the node + owner. + + @param client: The pubsub component client. + @param from_jid: Full JID of the subscriber sending presence. + @param presence_elt: The presence stanza. + """ + if not client.is_component: + return + + requestor_jid = from_jid.userhostJID() + + caps_hash = G.host.memory.get_entity_datum(client, from_jid, C.ENTITY_CAP_HASH) + if not caps_hash: + return + + try: + disco_infos = G.host.memory.disco.hashes[caps_hash] + except KeyError: + return + + notify_nodes = [f[:-7] for f in disco_infos.features if f.endswith("+notify")] + if not notify_nodes: + return + + # Build list of publishers (entities the requestor can receive notifications from) + + # Self-subscription + publishers = {requestor_jid} + + try: + roster = await self.xep_0356.get_roster(client, requestor_jid) + except exceptions.PermissionError: + roster = {} + + # Add publishers that the requestor has subscribed to. + publishers.update( + roster_jid + for roster_jid, roster_item in roster.items() + if roster_item.subscriptionTo + ) + + try: + last_items = await G.storage.get_pep_last_items( + client, + list(publishers), + notify_nodes, + [AccessModel.open, AccessModel.presence], + ) + except Exception: + log.warning(f"Failed to get PEP last items for {from_jid}") + return + + # Send notifications + for node, item in last_items: + # For self-subscription, no need to check access + if node.service != requestor_jid: + if not await self.xep_0496.has_access( + client, node.service, node, requestor_jid + ): + continue + + item_elt = cast(domish.Element, item.data) + await self._send_notification( + node, "items", str(node.name), from_jid, [item_elt] + ) + class PubsubService(rsm.PubSubService): def __init__(self, component: "PubsubComponent"): diff --git a/libervia/backend/plugins/plugin_xep_0115.py b/libervia/backend/plugins/plugin_xep_0115.py index 5059b205..aeec5b52 100644 --- a/libervia/backend/plugins/plugin_xep_0115.py +++ b/libervia/backend/plugins/plugin_xep_0115.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 -# SAT plugin for managing xep-0115 +# Libervia plugin for managing xep-0115 # Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify @@ -17,6 +17,8 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from libervia.backend import G +from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import _ from libervia.backend.core.constants import Const as C from libervia.backend.core.log import getLogger @@ -28,10 +30,7 @@ from twisted.internet import defer, error from zope.interface import implementer from tx_xmpp import disco, itx_xmpp -try: - from twisted.words.protocols.xmlstream import XMPPHandler -except ImportError: - from tx_xmpp.subprotocols import XMPPHandler +from twisted.words.protocols.jabber.xmlstream import XMPPHandler PRESENCE = "/presence" NS_ENTITY_CAPABILITY = "http://jabber.org/protocol/caps" @@ -51,7 +50,7 @@ PLUGIN_INFO = { } -class XEP_0115(object): +class XEP_0115: cap_hash = None # capabilities hash is class variable as it is common to all profiles def __init__(self, host): @@ -62,13 +61,12 @@ class XEP_0115(object): def get_handler(self, client): return XEP_0115_handler(self) - @defer.inlineCallbacks - def _prepare_caps(self, client): + async def _prepare_caps(self, client: SatXMPPEntity): # we have to calculate hash for client # because disco infos/identities may change between clients # optimize check - client._caps_optimize = yield self.host.has_feature(client, NS_CAPS_OPTIMIZE) + client._caps_optimize = await self.host.has_feature(client, NS_CAPS_OPTIMIZE) if client._caps_optimize: log.info(_("Caps optimisation enabled")) client._caps_sent = False @@ -76,7 +74,7 @@ class XEP_0115(object): log.warning(_("Caps optimisation not available")) # hash generation - _infos = yield client.discoHandler.info(client.jid, client.jid, "") + _infos = await client.discoHandler.info(client.jid, client.jid, "") disco_infos = disco.DiscoInfo() for item in _infos: disco_infos.append(item) @@ -109,43 +107,27 @@ class XEP_0115(object): def _presence_trigger(self, client, obj, presence_d): if not hasattr(client, "_caps_optimize"): - presence_d.addCallback(lambda __: self._prepare_caps(client)) + presence_d.addCallback( + lambda __: defer.ensureDeferred(self._prepare_caps(client)) + ) presence_d.addCallback(lambda __: self._presence_add_elt(client, obj)) return True + def _update(self, presence_elt: domish.Element, client: SatXMPPEntity) -> None: + defer.ensureDeferred(self.update(client, presence_elt)) -@implementer(itx_xmpp.IDisco) -class XEP_0115_handler(XMPPHandler): - - def __init__(self, plugin_parent): - self.plugin_parent = plugin_parent - self.host = plugin_parent.host - - @property - def client(self): - return self.parent - - def connectionInitialized(self): - self.xmlstream.addObserver(CAPABILITY_UPDATE, self.update) - - def getDiscoInfo(self, requestor, target, nodeIdentifier=""): - return [ - disco.DiscoFeature(NS_ENTITY_CAPABILITY), - disco.DiscoFeature(NS_CAPS_OPTIMIZE), - ] - - def getDiscoItems(self, requestor, target, nodeIdentifier=""): - return [] - - def update(self, presence): + async def update(self, client: SatXMPPEntity, presence_elt: domish.Element) -> None: """ Manage the capabilities of the entity - Check if we know the version of this capabilities and get the capabilities if necessary + Check if we know the version of this capabilities and get the capabilities if + necessary + + @param presence_elt: element to parse. """ - from_jid = jid.JID(presence["from"]) - c_elem = next(presence.elements(NS_ENTITY_CAPABILITY, "c")) + from_jid = jid.JID(presence_elt["from"]) + c_elem = next(presence_elt.elements(NS_ENTITY_CAPABILITY, "c")) try: c_ver = c_elem["ver"] c_hash = c_elem["hash"] @@ -161,7 +143,7 @@ class XEP_0115_handler(XMPPHandler): % {"hash": c_ver, "jid": from_jid.full()} ) self.host.memory.update_entity_data( - self.client, from_jid, C.ENTITY_CAP_HASH, c_ver + client, from_jid, C.ENTITY_CAP_HASH, c_ver ) return @@ -173,9 +155,19 @@ class XEP_0115_handler(XMPPHandler): ).format(hash_method=c_hash, entity_jid=from_jid, node=c_node) ) - def cb(__): + try: + await G.host.get_disco_infos(client, from_jid) + except error.ConnectionDone: + return + except Exception as e: + log.error( + _("Couldn't retrieve disco info for {jid}: {error}").format( + jid=from_jid.full(), error=str(e) + ) + ) + else: computed_hash = self.host.memory.get_entity_datum( - self.client, from_jid, C.ENTITY_CAP_HASH + client, from_jid, C.ENTITY_CAP_HASH ) if computed_hash != c_ver: log.warning( @@ -192,20 +184,30 @@ class XEP_0115_handler(XMPPHandler): ) ) - def eb(failure): - if isinstance(failure.value, error.ConnectionDone): - return - msg = ( - failure.value.condition - if hasattr(failure.value, "condition") - else failure.getErrorMessage() - ) - log.error( - _("Couldn't retrieve disco info for {jid}: {error}").format( - jid=from_jid.full(), error=msg - ) - ) - - d = self.host.get_disco_infos(self.parent, from_jid) - d.addCallbacks(cb, eb) # TODO: me must manage the full algorithm described at XEP-0115 #5.4 part 3 + + +@implementer(itx_xmpp.IDisco) +class XEP_0115_handler(XMPPHandler): + + def __init__(self, plugin_parent): + self.plugin_parent = plugin_parent + + @property + def client(self): + return self.parent + + def connectionInitialized(self): + assert self.xmlstream is not None + self.xmlstream.addObserver( + CAPABILITY_UPDATE, self.plugin_parent._update, client=self.client + ) + + def getDiscoInfo(self, requestor, target, nodeIdentifier=""): + return [ + disco.DiscoFeature(NS_ENTITY_CAPABILITY), + disco.DiscoFeature(NS_CAPS_OPTIMIZE), + ] + + def getDiscoItems(self, requestor, target, nodeIdentifier=""): + return [] diff --git a/libervia/backend/plugins/plugin_xep_0356.py b/libervia/backend/plugins/plugin_xep_0356.py index 40acc340..527a8cb9 100644 --- a/libervia/backend/plugins/plugin_xep_0356.py +++ b/libervia/backend/plugins/plugin_xep_0356.py @@ -21,7 +21,7 @@ from textwrap import dedent from typing import TYPE_CHECKING, Final, Literal, cast import uuid -from twisted.internet import reactor +from twisted.internet import defer, reactor from twisted.words.protocols.jabber import error, jid from twisted.words.protocols.jabber.xmlstream import XMPPHandler from twisted.words.xish import domish @@ -35,6 +35,7 @@ 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_0115 import XEP_0115 from libervia.backend.plugins.plugin_xep_0297 import NS_FORWARD, XEP_0297 if TYPE_CHECKING: @@ -48,7 +49,7 @@ PLUGIN_INFO = { C.PI_TYPE: "XEP", C.PI_MODES: [C.PLUG_MODE_COMPONENT], C.PI_PROTOCOLS: ["XEP-0356"], - C.PI_DEPENDENCIES: ["XEP-0297"], + C.PI_DEPENDENCIES: ["XEP-0115", "XEP-0297"], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "XEP_0356", C.PI_HANDLER: "yes", @@ -107,10 +108,21 @@ class PrivilegeError(exceptions.PermissionError): pass +@dataclass +class ClientPresenceData: + """Presence tracking data for a client.""" + + # bare_jid -> set of online resource strings + online_resources: dict[jid.JID, set[str]] = field(default_factory=dict) + + @dataclass class ClientPrivilegeData: privileges: PrivilegeData = field(default_factory=lambda: PrivilegeData()) roster_cache: dict[jid.JID, RosterCacheEntry] = field(default_factory=dict) + presence_data: ClientPresenceData = field( + default_factory=lambda: ClientPresenceData() + ) class XEP_0356: @@ -120,6 +132,7 @@ class XEP_0356: def __init__(self, host: "LiberviaBackend") -> None: log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") + self._xep_0115 = cast(XEP_0115, host.plugins["XEP-0115"]) self._forward = cast(XEP_0297, host.plugins["XEP-0297"]) host.register_namespace("privilege", NS_PRIVILEGE) @@ -501,21 +514,61 @@ class XEP_0356: log.error(f"Invalid privileged IQ response: {iq_result_elt.toXml()}") raise PrivilegeError("Invalid response format") - def _on_presence( + async def _on_presence( + self, presence_elt: domish.Element, client: SatXMPPComponent + ) -> None: + defer.ensureDeferred(self.on_presence(presence_elt, client)) + + async def on_presence( self, presence_elt: domish.Element, client: SatXMPPComponent ) -> None: """Handle incoming presence stanza. + Capabilities are parsed with XEP-0115. + @param presence_elt: The presence stanza. @param client: The client instance. """ try: from_jid = jid.JID(presence_elt["from"]) - except: + except (KeyError, RuntimeError): log.exception("Can't parse 'from' attribute.") raise - G.host.trigger.point("XEP-0356_presence_received", client, from_jid, presence_elt) + self._track_presence(from_jid, presence_elt, client) + + try: + await self._xep_0115.update(client, presence_elt) + except Exception: + log.exception("Failed to parse capabilities in XEP-0115") + + await G.host.trigger.async_point( + "XEP-0356_presence_received", client, from_jid, presence_elt + ) + + def _track_presence( + self, from_jid: jid.JID, presence_elt: domish.Element, client: SatXMPPComponent + ) -> None: + """Track online resources per bare JID. + + @param from_jid: The JID sending presence. + @param presence_elt: The presence stanza. + @param client: The client instance. + """ + resource = from_jid.resource + if resource is None: + # Bare JID presence, ignore + return + + bare_jid = from_jid.userhostJID() + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + online_resources = data.presence_data.online_resources + + presence_type = presence_elt.getAttribute("type") + if presence_type == "unavailable": + online_resources.setdefault(bare_jid, set()).discard(resource) + else: + online_resources.setdefault(bare_jid, set()).add(resource) @implementer(itx_xmpp.IDisco) diff --git a/libervia/backend/plugins/plugin_xep_0496.py b/libervia/backend/plugins/plugin_xep_0496.py index f3dc2673..773aa518 100644 --- a/libervia/backend/plugins/plugin_xep_0496.py +++ b/libervia/backend/plugins/plugin_xep_0496.py @@ -16,7 +16,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, cast from sqlalchemy import select, update from twisted.words.protocols.jabber import jid @@ -38,6 +38,7 @@ from libervia.backend.memory.sqla_mapping import ( PubsubNode, SubscriptionState, ) +from libervia.backend.plugins.plugin_xep_0356 import XEP_0356 from libervia.backend.tools.utils import aio if TYPE_CHECKING: @@ -53,6 +54,7 @@ PLUGIN_INFO = { C.PI_PROTOCOLS: ["XEP-0496"], C.PI_DEPENDENCIES: [ "XEP-0060", + "XEP-0356", ], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "XEP_0496", @@ -69,7 +71,7 @@ class XEP_0496: def __init__(self, host: "LiberviaBackend") -> None: log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") - self.host = host + self._xep_0356 = cast(XEP_0356, host.plugins["XEP-0356"]) host.register_namespace("pubsub-relationships", NS_PUBSUB_RELATIONSHIPS) def get_handler(self, client) -> "XEP_0496_handler": @@ -240,9 +242,17 @@ class XEP_0496: return parent_node, linked_node - def _check_node_access(self, node: PubsubNode, entity_jid: jid.JID) -> bool: + async def _check_node_access( + self, + client: SatXMPPEntity, + service: JID, + node: PubsubNode, + entity_jid: jid.JID, + ) -> bool: """Check if an entity has access to a single node. + @param client: Client instance. + @param service: The pubsub service JID. @param node: The node to check access for. It must have ``affiliations`` loaded. @param entity_jid: The entity JID to check access for. @return: True if the entity has access to the node. @@ -260,6 +270,8 @@ class XEP_0496: if access_model == AccessModel.open: return True + elif access_model == AccessModel.presence: + return await self._check_presence_access(client, service, node, entity_jid) elif access_model == AccessModel.whitelist: return affiliation in ( Affiliation.owner, @@ -268,6 +280,42 @@ class XEP_0496: ) return False + async def _check_presence_access( + self, + client: SatXMPPEntity, + service: JID, + node: PubsubNode, + entity_jid: jid.JID, + ) -> bool: + """Check presence-based access for a node. + + @param client: Client instance. + @param service: The pubsub service JID. + @param node: The pubsub node. + @param entity_jid: The entity JID to check access for. + @return: True if requester has "from" or "both" subscription with publisher. + """ + if not node.pep: + raise NotImplementedError("Presence access model is only implemented for PEP") + + # For PEP, owner/publisher is the service JID. + publisher_jid = service.userhostJID() + + try: + roster = await self._xep_0356.get_roster(client, publisher_jid) + except Exception: + log.warning( + f"Failed to get roster for {publisher_jid} when checking presence access." + ) + return False + + roster_item = roster.get(entity_jid) + if roster_item is None: + return False + + # According to XEP-0060, subscription "from" or "both" are accepted. + return roster_item.subscriptionFrom + async def has_access( self, client: SatXMPPEntity, service: JID, node: PubsubNode, entity_jid: jid.JID ) -> bool: @@ -279,7 +327,7 @@ class XEP_0496: @param entity_jid: The entity JID to check access for. @return: True if the entity has access to the node and all its parents. """ - if not self._check_node_access(node, entity_jid): + if not await self._check_node_access(client, service, node, entity_jid): return False current_node = node @@ -295,7 +343,9 @@ class XEP_0496: if parent_node is None: break - if not self._check_node_access(parent_node, entity_jid): + if not await self._check_node_access( + client, service, parent_node, entity_jid + ): return False current_node = parent_node diff --git a/tests/unit/test_component_pubsub.py b/tests/unit/test_component_pubsub.py index 87c826e9..dbd8de49 100644 --- a/tests/unit/test_component_pubsub.py +++ b/tests/unit/test_component_pubsub.py @@ -28,10 +28,12 @@ from twisted.words.protocols.jabber.error import StanzaError from twisted.words.xish import domish from tx_xmpp import pubsub from tx_xmpp import rsm +from tx_xmpp.xmppim import RosterItem from libervia.backend.memory.sqla_mapping import ( AccessModel, Affiliation, PublishModel, + PubsubItem, PubsubNode, PubsubAffiliation, PubsubSub, @@ -109,6 +111,7 @@ def make_pubsub_node( publish_model: PublishModel = PublishModel.publishers, affiliations: list[PubsubAffiliation] | None = None, subscriptions: list[PubsubSub] | None = None, + pep: bool = False, ) -> PubsubNode: """Helper to create a PubsubNode for testing.""" node = PubsubNode(name=name) @@ -116,6 +119,7 @@ def make_pubsub_node( node.publish_model = publish_model node.affiliations = affiliations or [] node.subscriptions = subscriptions or [] + node.pep = pep return node @@ -721,6 +725,103 @@ class TestPubsubResourceItems: host.memory.storage.get_items.assert_called_once() + @ed + async def test_presence_model_rejects_non_contacts( + self, host, client, pubsub_resource + ): + """presence access model rejects non-contacts""" + resource = pubsub_resource + request = make_pubsub_request( + sender=jid.JID("stranger@test.example"), + nodeIdentifier="test_node", + ) + + node = make_pubsub_node( + name="test_node", + access_model=AccessModel.presence, + pep=True, + ) + host.memory.storage.get_pubsub_node = AsyncMock(return_value=node) + + # Mock XEP-0356 roster to not include the stranger + mock_xep_0356 = MagicMock() + mock_xep_0356.get_roster = AsyncMock(return_value={}) + host.plugins["XEP-0356"] = mock_xep_0356 + + with raises(StanzaError, match="forbidden"): + await resource._items(request) + + @ed + async def test_presence_model_rejects_contact_from_subscription( + self, host, client, pubsub_resource + ): + """presence access model rejects when "to" subscription is missing""" + requestor_jid = jid.JID("contact@test.example") + + resource = pubsub_resource + request = make_pubsub_request( + sender=requestor_jid, + nodeIdentifier="test_node", + ) + + node = make_pubsub_node( + name="test_node", + access_model=AccessModel.presence, + pep=True, + ) + host.memory.storage.get_pubsub_node = AsyncMock(return_value=node) + + mock_xep_0356 = MagicMock() + roster_item = RosterItem( + entity=requestor_jid, + subscriptionTo=False, + subscriptionFrom=True, + ) + mock_xep_0356.get_roster = AsyncMock( + return_value={requestor_jid: roster_item} + ) + host.plugins["XEP-0356"] = mock_xep_0356 + + with raises(StanzaError, match="forbidden"): + await resource._items(request) + + @ed + async def test_presence_model_allows_contact_from_subscription( + self, host, client, pubsub_resource + ): + """presence access model allows when requestor has "from" subscription""" + resource = pubsub_resource + requestor_jid = jid.JID("contact@test.example") + request = make_pubsub_request( + sender=requestor_jid, + nodeIdentifier="test_node", + recipient=client.pubsub_service, + ) + + node = make_pubsub_node( + name="test_node", + access_model=AccessModel.presence, + pep=True, + ) + host.memory.storage.get_pubsub_node = AsyncMock(return_value=node) + host.memory.storage.get_items = AsyncMock(return_value=([], {})) + mock_xep_0356 = MagicMock() + host.plugins["XEP-0356"] = mock_xep_0356 + host.plugins["XEP-0496"]._xep_0356 = mock_xep_0356 + + roster_item = RosterItem( + entity=requestor_jid, + subscriptionTo=False, + subscriptionFrom=True, + ) + mock_xep_0356.get_roster = AsyncMock( + return_value={requestor_jid: roster_item} + ) + + await resource._items(request) + + host.memory.storage.get_items.assert_called_once() + class TestPubsubResourceRetract: """Tests for item retraction.""" @@ -1295,7 +1396,6 @@ class TestPubsubResourceSubscriptionsGetSet: ) ], ) - node = make_pubsub_node( name="test_node", affiliations=[ @@ -1308,3 +1408,76 @@ class TestPubsubResourceSubscriptionsGetSet: with raises(StanzaError, match="not-acceptable"): await resource._subscriptionsSet(request) + + +class TestPubsubAutoSubscribe: + """Tests for PEP auto-subscription.""" + + @ed + async def test_auto_subscribe_skips_no_notify_capability( + self, host, client, pubsub_resource + ): + """auto-subscription skips contacts without +notify capability""" + client.is_component = True + + contact_jid = jid.JID("contact@test.example/resource1") + presence_elt = domish.Element((None, "presence")) + presence_elt["from"] = contact_jid.full() + + # Mock caps hash + caps_hash = "abc123" + host.memory.get_entity_datum = MagicMock(return_value=caps_hash) + + # Mock disco info WITHOUT +notify feature + mock_disco_info = MagicMock() + mock_disco_info.features = [ + "http://jabber.org/protocol/caps", + "http://jabber.org/protocol/tune", # No +notify + ] + host.memory.disco.hashes = {caps_hash: mock_disco_info} + + await pubsub_resource._on_presence_received(client, contact_jid, presence_elt) + + # Check that no query was made (no +notify) + host.memory.storage.get_pep_last_items.assert_not_called() + + @ed + async def test_auto_subscribe_skips_no_presence_subscription( + self, host, client, pubsub_resource + ): + """auto-subscription skips contacts without presence subscription""" + client.is_component = True + + contact_jid = jid.JID("stranger@test.example/resource1") + requestor_jid = contact_jid.userhostJID() + presence_elt = domish.Element((None, "presence")) + presence_elt["from"] = contact_jid.full() + + # Mock caps hash with +notify + caps_hash = "abc123" + host.memory.get_entity_datum = MagicMock(return_value=caps_hash) + mock_disco_info = MagicMock() + mock_disco_info.features = ["http://jabber.org/protocol/tune+notify"] + host.memory.disco.hashes = {caps_hash: mock_disco_info} + + # Mock get_roster to return empty roster (no presence subscription) + mock_xep_0356 = MagicMock() + mock_xep_0356.get_roster = AsyncMock(return_value={}) + host.plugins["XEP-0356"] = mock_xep_0356 + pubsub_resource.xep_0356 = mock_xep_0356 + + host.memory.storage.get_pep_last_items = AsyncMock(return_value=[]) + + await pubsub_resource._on_presence_received(client, contact_jid, presence_elt) + + mock_xep_0356.get_roster.assert_called_once_with(client, requestor_jid) + + # Verify get_pep_last_items was called with only self-subscription, not the + # stranger + host.memory.storage.get_pep_last_items.assert_called_once() + call_args = host.memory.storage.get_pep_last_items.call_args + # We want the second positional argument (list of publishers) + publishers_arg = call_args[0][1] + assert publishers_arg == [requestor_jid], ( + f"Expected only self-subscription, got {publishers_arg}" + ) diff --git a/tests/unit/test_xep_0356.py b/tests/unit/test_xep_0356.py index 567033a3..8314eba3 100644 --- a/tests/unit/test_xep_0356.py +++ b/tests/unit/test_xep_0356.py @@ -37,7 +37,6 @@ from libervia.backend.plugins.plugin_xep_0356 import ( IMPORT_NAME, PrivilegeError, ) -from tx_xmpp import disco from tx_xmpp.xmppim import RosterItem, NS_ROSTER @@ -672,12 +671,12 @@ class TestXEP0356PresenceHandling: trigger_point_calls = [] - def mock_trigger_point(point_name, *args): + async def mock_async_point(point_name, *args): trigger_point_calls.append((point_name, args)) - G.host.trigger.point = mock_trigger_point + G.host.trigger.async_point = mock_async_point - xep_0356._on_presence(presence, client) + await xep_0356._on_presence(presence, client) assert len(trigger_point_calls) == 1 point_name, args = trigger_point_calls[0] @@ -696,13 +695,73 @@ class TestXEP0356PresenceHandling: trigger_point_calls = [] - def mock_trigger_point(point_name, *args): + async def mock_async_point(point_name, *args): trigger_point_calls.append((point_name, args)) - G.host.trigger.point = mock_trigger_point + G.host.trigger.async_point = mock_async_point - xep_0356._on_presence(presence, client) + await 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" + + @ed + async def test_presence_tracking_add_and_remove_resource(self, xep_0356, client): + """_track_presence adds and removes resources from the presence data.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + bare_jid = jid.JID("user@test.example") + + assert len(data.presence_data.online_resources) == 0 + + presence_available = domish.Element((None, "presence")) + presence_available["from"] = "user@test.example/resource1" + await xep_0356._on_presence(presence_available, client) + + assert bare_jid in data.presence_data.online_resources + assert "resource1" in data.presence_data.online_resources[bare_jid] + + presence_unavailable = domish.Element((None, "presence")) + presence_unavailable["from"] = "user@test.example/resource1" + presence_unavailable["type"] = "unavailable" + await xep_0356._on_presence(presence_unavailable, client) + + assert bare_jid in data.presence_data.online_resources + assert "resource1" not in data.presence_data.online_resources[bare_jid] + + @ed + async def test_presence_tracking_multiple_resources(self, xep_0356, client): + """_track_presence tracks multiple resources for the same bare JID.""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + bare_jid = jid.JID("user@test.example") + + assert len(data.presence_data.online_resources) == 0 + + presence1 = domish.Element((None, "presence")) + presence1["from"] = "user@test.example/resource1" + await xep_0356._on_presence(presence1, client) + + assert bare_jid in data.presence_data.online_resources + assert "resource1" in data.presence_data.online_resources[bare_jid] + + presence2 = domish.Element((None, "presence")) + presence2["from"] = "user@test.example/resource2" + await xep_0356._on_presence(presence2, client) + + assert "resource1" in data.presence_data.online_resources[bare_jid] + assert "resource2" in data.presence_data.online_resources[bare_jid] + + @ed + async def test_presence_tracking_ignores_bare_jid(self, xep_0356, client): + """_track_presence ignores presence from bare JIDs (no resource).""" + data = client.get_plugin_data(IMPORT_NAME, ClientPrivilegeData) + bare_jid = jid.JID("user@test.example") + + assert len(data.presence_data.online_resources) == 0 + + presence = domish.Element((None, "presence")) + presence["from"] = "user@test.example" + + await xep_0356._on_presence(presence, client) + + assert bare_jid not in data.presence_data.online_resources