component pubsub: PEP presence access model and auto-subscription
- Implement `presence` access model for PEP nodes. - Add PEP auto-subscription (`+notify` handling) - New `get_pep_last_items` storage method to retrieve in a single query items to send to auto-subscribers. - Online resource tracking. This completes the PEP support. rel 476
This commit is contained in:
Goffi
parent
81bd3dcf01
commit
42c2f88b30
9 changed files with 610 additions and 79 deletions
|
|
@ -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')",
|
||||
)
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -106,6 +106,7 @@ class SyncState(enum.Enum):
|
|||
|
||||
class AccessModel(enum.StrEnum):
|
||||
open = enum.auto()
|
||||
presence = enum.auto()
|
||||
whitelist = enum.auto()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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"):
|
||||
|
|
|
|||
|
|
@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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: <presence/> 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 []
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@
|
|||
# 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/>.
|
||||
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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}"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Reference in a new issue