component pubsub: PEP implementation:

New implementations of XEP-0355 and XEP-0356 are integrated to handle PEP. The
implementation is not complete yet, as auto-subscribe and presence permissions are not
implemented yet.

rel 476
This commit is contained in:
Goffi 2026-04-14 23:16:18 +02:00
commit 81bd3dcf01

View file

@ -20,13 +20,13 @@ from dataclasses import dataclass, field, fields
from typing import Literal, cast
import shortuuid
from twisted.internet import defer
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 import pubsub, rsm
from tx_xmpp.pubsub import NS_PUBSUB_EVENT
from zope.interface import implementer
from libervia.backend import G
@ -44,19 +44,20 @@ from libervia.backend.memory.sqla_mapping import (
PubsubSub,
SubscriptionState,
)
from libervia.backend.plugins.plugin_xep_0167 import NS_AV_CONFERENCES
from libervia.backend.plugins.plugin_xep_0060 import NodeMetadata, XEP_0060
from libervia.backend.plugins.plugin_xep_0167 import NS_AV_CONFERENCES
from libervia.backend.plugins.plugin_xep_0355 import XEP_0355, DelegationData
from libervia.backend.plugins.plugin_xep_0356 import XEP_0356
from libervia.backend.plugins.plugin_xep_0496 import LINK_VAR, PARENT_VAR, XEP_0496
from libervia.backend.plugins.plugin_xep_0499 import (
DiscoPubsubItem,
DiscoPubsubNode,
ExtDiscoOptions,
ExtDiscoResult,
DiscoPubsubNode,
XEP_0499,
)
from libervia.backend.tools.config import ComponentSettings
from libervia.backend.tools.xml_tools import element_copy
from tx_xmpp.pubsub import NS_PUBSUB_EVENT
log = getLogger(__name__)
@ -69,7 +70,7 @@ PLUGIN_INFO = {
C.PI_MODES: [C.PLUG_MODE_COMPONENT],
C.PI_TYPE: C.PLUG_TYPE_ENTRY_POINT,
C.PI_PROTOCOLS: [],
C.PI_DEPENDENCIES: ["XEP-0060", "XEP-0496", "XEP-0499"],
C.PI_DEPENDENCIES: ["XEP-0060", "XEP-0355", "XEP-0356", "XEP-0496", "XEP-0499"],
C.PI_RECOMMENDATIONS: [],
C.PI_MAIN: "PubsubComponent",
C.PI_HANDLER: C.BOOL_TRUE,
@ -187,6 +188,8 @@ class PubsubResource(pubsub.PubSubResource):
def __init__(self, service: "PubsubService") -> None:
super().__init__()
self.service = service
self.xep_0355 = self.service.component.xep_0355
self.xep_0356 = cast(XEP_0356, G.host.plugins["XEP-0356"])
self.xep_0496 = cast(XEP_0496, G.host.plugins["XEP-0496"])
@property
@ -260,7 +263,7 @@ class PubsubResource(pubsub.PubSubResource):
# We only return results for requests without a specific node
return []
nodes = await G.storage.get_pubsub_nodes(self.client, self.client.jid)
nodes = await G.storage.get_pubsub_nodes(self.client, service)
return [cast(str, node.name) for node in nodes]
def getConfigurationOptions(self) -> dict[str, dict]:
@ -290,13 +293,14 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
if request.items is None or len(request.items) == 0:
raise error.StanzaError("bad-request", text="no item to publish")
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
@ -307,7 +311,7 @@ class PubsubResource(pubsub.PubSubResource):
raise error.StanzaError("item-not-found", text="Node does not exist")
has_publish_rights = await self.xep_0496.has_publish_rights(
self.client, self.client.jid, node, requestor_jid
self.client, service_jid, node, requestor_jid
)
if not has_publish_rights:
raise error.StanzaError("forbidden", text="Insufficient privileges")
@ -361,6 +365,7 @@ class PubsubResource(pubsub.PubSubResource):
assert request.nodeIdentifier is not None
assert request.subscriber is not None
subscriber_jid = request.subscriber.userhostJID()
service_jid = request.recipient if request.recipient else client.jid
if subscriber_jid != requestor_jid:
raise error.StanzaError(
@ -368,14 +373,14 @@ class PubsubResource(pubsub.PubSubResource):
)
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
client, service_jid, request.nodeIdentifier, with_affiliations=True
)
if node is None:
raise error.StanzaError("item-not-found", text="Node does not exist")
has_access = await self.xep_0496.has_access(
self.client, self.client.jid, node, subscriber_jid
self.client, service_jid, node, subscriber_jid
)
if not has_access:
raise error.StanzaError(
@ -460,6 +465,7 @@ class PubsubResource(pubsub.PubSubResource):
assert request.nodeIdentifier is not None
assert request.subscriber is not None
subscriber_jid = request.subscriber.userhostJID()
service_jid = request.recipient if request.recipient else client.jid
# Check that the requestor matches the subscriber
if subscriber_jid != requestor_jid:
@ -467,7 +473,9 @@ class PubsubResource(pubsub.PubSubResource):
"forbidden", text="Cannot unsubscribe on behalf of another entity"
)
node = await G.storage.get_pubsub_node(client, client.jid, request.nodeIdentifier)
node = await G.storage.get_pubsub_node(
client, service_jid, request.nodeIdentifier
)
if node is None:
raise error.StanzaError("item-not-found", text="Node does not exist")
@ -495,10 +503,11 @@ class PubsubResource(pubsub.PubSubResource):
client = self.client
assert request.sender is not None
requestor_jid = request.sender.userhostJID()
service_jid = request.recipient if request.recipient else client.jid
subscriptions_data = await G.storage.get_pubsub_subscriptions(
client,
service=client.jid,
service=service_jid,
entity=requestor_jid,
node_name=request.nodeIdentifier,
)
@ -537,10 +546,11 @@ class PubsubResource(pubsub.PubSubResource):
client = self.client
assert request.sender is not None
requestor_jid = request.sender.userhostJID()
service_jid = request.recipient if request.recipient else client.jid
affiliations = await G.storage.get_pubsub_affiliations(
client,
client.jid,
service_jid,
entity=requestor_jid,
node_name=request.nodeIdentifier,
)
@ -561,6 +571,19 @@ class PubsubResource(pubsub.PubSubResource):
client = self.client
assert request.sender is not None
requestor_jid = request.sender.userhostJID()
service_jid = request.recipient if request.recipient else client.jid
is_pep = self.xep_0355.is_pep(client, service_jid)
if is_pep and requestor_jid != service_jid.userhostJID():
log.warning(
f"{requestor_jid} tried to create a PEP node on"
f" {service_jid.userhostJID()}'s service, hack attempt?"
)
raise error.StanzaError(
"forbidden",
text="Only the owner of the PEP service can create PEP nodes.",
)
node_id = request.nodeIdentifier or f"generic/{shortuuid.uuid()}"
@ -612,12 +635,13 @@ class PubsubResource(pubsub.PubSubResource):
node = await G.storage.set_pubsub_node(
client=client,
service=client.jid,
service=service_jid,
name=node_id,
access_model=config.access_model,
publish_model=config.publish_model,
affiliations=[owner_affiliation],
max_items=db_max_items,
pep=is_pep,
)
except exceptions.ConflictError:
raise error.StanzaError("conflict", text=f"node {node_id!r} already exists.")
@ -626,7 +650,7 @@ class PubsubResource(pubsub.PubSubResource):
try:
parent_node, linked_node = await self.xep_0496.check_relationships(
self.client,
self.client.jid,
service_jid,
node,
requestor_jid,
config.parent,
@ -702,10 +726,11 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
@ -772,10 +797,11 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
@ -786,7 +812,7 @@ class PubsubResource(pubsub.PubSubResource):
raise error.StanzaError("item-not-found", text="Node does not exist")
has_access = await self.xep_0496.has_access(
self.client, self.client.jid, node, requestor_jid
self.client, service_jid, node, requestor_jid
)
if not has_access:
raise error.StanzaError(
@ -911,7 +937,7 @@ class PubsubResource(pubsub.PubSubResource):
new_parent_node, new_linked_node = (
await self.xep_0496.check_relationships(
self.client,
self.client.jid,
service_jid,
node,
requestor_jid,
new_parent,
@ -975,10 +1001,11 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
@ -988,7 +1015,7 @@ class PubsubResource(pubsub.PubSubResource):
raise error.StanzaError("item-not-found", text="Node does not exist")
has_access = await self.xep_0496.has_access(
self.client, self.client.jid, node, requestor_jid
self.client, service_jid, node, requestor_jid
)
if not has_access:
raise error.StanzaError("forbidden", text="Insufficient privileges")
@ -1027,13 +1054,14 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
if not request.itemIdentifiers:
raise error.StanzaError("bad-request", text="No item identifiers provided")
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
@ -1102,10 +1130,11 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
@ -1146,10 +1175,11 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
@ -1160,7 +1190,7 @@ class PubsubResource(pubsub.PubSubResource):
raise error.StanzaError("item-not-found", text="Node does not exist")
has_access = await self.xep_0496.has_access(
self.client, self.client.jid, node, requestor_jid
self.client, service_jid, node, requestor_jid
)
if not has_access:
raise error.StanzaError(
@ -1203,7 +1233,7 @@ class PubsubResource(pubsub.PubSubResource):
if node_name not in nodes_map:
fetched_node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
node_name,
with_affiliations=True,
with_parent=True,
@ -1219,7 +1249,7 @@ class PubsubResource(pubsub.PubSubResource):
if nodes_to_delete:
await G.storage.delete_pubsub_node(
profiles=None, services=[client.jid], names=nodes_to_delete
profiles=None, services=[service_jid], names=nodes_to_delete
)
def affiliationsGet(
@ -1240,16 +1270,17 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
client, service_jid, request.nodeIdentifier, with_affiliations=True
)
if node is None:
raise error.StanzaError("item-not-found", text="Node does not exist")
has_access = await self.xep_0496.has_access(
self.client, self.client.jid, node, requestor_jid
self.client, service_jid, node, requestor_jid
)
if not has_access:
raise error.StanzaError(
@ -1291,16 +1322,17 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
client, service_jid, request.nodeIdentifier, with_affiliations=True
)
if node is None:
raise error.StanzaError("item-not-found", text="Node does not exist")
has_access = await self.xep_0496.has_access(
self.client, self.client.jid, node, requestor_jid
self.client, service_jid, node, requestor_jid
)
if not has_access:
raise error.StanzaError(
@ -1348,7 +1380,7 @@ class PubsubResource(pubsub.PubSubResource):
if affiliations_to_set:
await G.storage.set_pubsub_affiliations(
client, client.jid, request.nodeIdentifier, affiliations_to_set
client, service_jid, request.nodeIdentifier, affiliations_to_set
)
def subscriptionsGet(
@ -1371,10 +1403,11 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client,
client.jid,
service_jid,
request.nodeIdentifier,
with_subscriptions=True,
with_affiliations=True,
@ -1433,9 +1466,10 @@ class PubsubResource(pubsub.PubSubResource):
requestor_jid = request.sender.userhostJID()
assert request.nodeIdentifier is not None
service_jid = request.recipient if request.recipient else client.jid
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
client, service_jid, request.nodeIdentifier, with_affiliations=True
)
if node is None:
@ -1550,6 +1584,7 @@ class PubsubResource(pubsub.PubSubResource):
async def _send_notification(
self,
node: PubsubNode,
event_type: NotificationEventType,
node_id: str,
subscriber: jid.JID,
@ -1557,13 +1592,14 @@ class PubsubResource(pubsub.PubSubResource):
) -> None:
"""Send a single notification to a subscriber.
@param node: The pubsub node (to check if it's a PEP node).
@param event_type: The type of event.
@param node_id: The node identifier.
@param subscriber: The subscriber JID.
@param item_elts: The items to include in the notification.
"""
message_elt = domish.Element((None, "message"))
message_elt["from"] = self.client.jid.full()
message_elt["from"] = node.service.full()
message_elt["to"] = subscriber.full()
message_elt["id"] = f"{shortuuid.uuid()}"
@ -1594,7 +1630,10 @@ class PubsubResource(pubsub.PubSubResource):
f"Unknown notification event type: {event_type!r}"
)
await self.client.a_send(message_elt)
if node.pep:
await self.xep_0356.send_privileged_message(self.client, message_elt)
else:
await self.client.a_send(message_elt)
async def _send_notifications(
self,
@ -1616,7 +1655,9 @@ class PubsubResource(pubsub.PubSubResource):
deferreds = [
defer.ensureDeferred(
self._send_notification(event_type, node.name, recipient_jid, item_elts)
self._send_notification(
node, event_type, node.name, recipient_jid, item_elts
)
)
for recipient_jid, __ in recipients.items()
]
@ -1631,8 +1672,8 @@ class PubsubResource(pubsub.PubSubResource):
class PubsubService(rsm.PubSubService):
def __init__(self, component: "PubsubComponent"):
super().__init__(resource=PubsubResource(self))
self.component = component
super().__init__(resource=PubsubResource(self))
self.discoidentity = {
"category": "pubsub",
"type": "service",
@ -1654,11 +1695,19 @@ class PubsubService(rsm.PubSubService):
assert query_elt is not None
node_name = query_elt.getAttribute("node")
requestor_jid = client.jid
try:
service_jid = jid.JID(iq_elt["to"])
except (KeyError, RuntimeError):
service_jid = client.jid
try:
requestor_jid = jid.JID(iq_elt["from"])
except (KeyError, RuntimeError):
requestor_jid = client.jid
nodes = await G.storage.get_pubsub_nodes(
client,
service=client.jid,
service=service_jid,
node_name=node_name,
linked_node=ext_disco_options.linked_nodes,
depth=ext_disco_options.depth,
@ -1672,24 +1721,24 @@ class PubsubService(rsm.PubSubService):
for node in nodes:
if ext_disco_options.depth == 0:
disco_node = DiscoPubsubNode.from_sqlalchemy(node, client.jid)
disco_node = DiscoPubsubNode.from_sqlalchemy(node, service_jid)
if "items" in ext_disco_options.type:
try:
disco_node.items = [
DiscoPubsubItem.from_sqlalchemy(item, node.name, client.jid)
DiscoPubsubItem.from_sqlalchemy(item, node.name, service_jid)
for item in node.items
]
except Exception:
pass
else:
disco_node = DiscoPubsubNode.from_sqlalchemy_full_hierarchy(
node, client.jid
node, service_jid
)
filtered_node = (
await self.component.ext_disco.filter_node_hierarchy_by_access(
client,
client.jid,
service_jid,
disco_node,
requestor_jid,
depth=ext_disco_options.depth,
@ -1714,9 +1763,10 @@ class PubsubComponent:
verbose = 0
def __init__(self, host):
self._clientt: SatXMPPComponent | None = None
self.pubsub_service = PubsubService(self)
self._client: SatXMPPComponent | None = None
self.ext_disco = cast(XEP_0499, host.plugins.get("XEP-0499"))
self.xep_0355 = cast(XEP_0355, G.host.plugins["XEP-0355"])
self.pubsub_service = PubsubService(self)
@property
def client(self) -> SatXMPPComponent:
@ -1729,6 +1779,18 @@ class PubsubComponent:
def profile_connecting(self, client: SatXMPPEntity) -> None:
self._client = cast(SatXMPPComponent, client)
self.ext_disco.register_handler(self.pubsub_service.on_extended_disco)
delegation_data = client.get_plugin_data("XEP_0355", DelegationData)
delegation_data.delegation_received.addCallback(self._on_delegation_received)
def _on_delegation_received(self, __) -> None:
"""Check that pubsub is delegated.
NS_PUBSUB is needed to manage PEP nodes.
"""
if pubsub.NS_PUBSUB not in self.xep_0355.delegated_namespaces:
log.warning(
"Didn't get pubsub delegation from server, can't act as a PEP service"
)
@implementer(itx_xmpp.IDisco)