erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

component pubsub: integrate pubsub relationships:

Browse source 
Pubsub relationships from XEP-0496 are now handled. They can be set in config as explained
in the specification, and access are checked with parents when they are set.

Deletion cascade as specified in the XEP.

rel 476
This commit is contained in:
Goffi 2026-04-10 23:05:42 +02:00
commit 3492baaa65
1 changed files with 207 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

269
libervia/backend/plugins/plugin_comp_pubsub/__init__.py
View file

@ -46,6 +46,7 @@ from libervia.backend.memory.sqla_mapping import (
)
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_0496 import LINK_VAR, PARENT_VAR, XEP_0496
from libervia.backend.tools.config import ComponentSettings
from libervia.backend.tools.xml_tools import element_copy
from tx_xmpp.pubsub import NS_PUBSUB_EVENT
@ -61,7 +62,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"],
C.PI_DEPENDENCIES: ["XEP-0060", "XEP-0496"],
C.PI_RECOMMENDATIONS: [],
C.PI_MAIN: "PubsubComponent",
C.PI_HANDLER: C.BOOL_TRUE,
@ -89,6 +90,8 @@ class PubsubNodeConfig:
description: str | None = field(
default=None, metadata={"label": "A description of the node"}
)
parent: str | None = field(default=None, metadata={"label": "Parent node identifier"})
link: str | None = field(default=None, metadata={"label": "Linked node identifier"})
_field_var_mapping = {
"access_model": "pubsub#access_model",
@ -96,6 +99,8 @@ class PubsubNodeConfig:
"max_items": "pubsub#max_items",
"title": "pubsub#title",
"description": "pubsub#description",
"parent": PARENT_VAR,
"link": LINK_VAR,
}
@classmethod
@ -109,8 +114,8 @@ class PubsubNodeConfig:
"""
definitions: dict = {}
for field in fields(cls):
field_name = field.name
for field_ in fields(cls):
field_name = field_.name
try:
pubsub_var = cls._field_var_mapping[field_name]
@ -118,7 +123,7 @@ class PubsubNodeConfig:
log.error("Missing field mapping for {field_name!r}, ignoring.")
continue
label = field.metadata.get("label", pubsub_var)
label = field_.metadata.get("label", pubsub_var)
match field_name:
case "access_model" | "publish_model":
@ -169,6 +174,7 @@ class PubsubResource(pubsub.PubSubResource):
def __init__(self, service: "PubsubService") -> None:
super().__init__()
self.service = service
self.xep_0496 = cast(XEP_0496, service.client.host.plugins.get("XEP-0496"))
@property
def client(self) -> SatXMPPComponent:
@ -286,29 +292,11 @@ class PubsubResource(pubsub.PubSubResource):
if node is None:
raise error.StanzaError("item-not-found", text="Node does not exist")
affiliation = next(
(
aff_data.affiliation
for aff_data in node.affiliations
if aff_data.entity == requestor_jid
),
None,
has_publish_rights = await self.xep_0496.has_publish_rights(
self.client, self.client.jid, node, requestor_jid
)
publish_model = node.publish_model or PublishModel.publishers
if publish_model == PublishModel.publishers:
if affiliation not in (Affiliation.owner, Affiliation.publisher):
raise error.StanzaError("forbidden", text="Insufficient privileges")
elif publish_model == PublishModel.subscribers:
if affiliation not in (Affiliation.owner, Affiliation.publisher):
is_subscribed = any(
sub.subscriber == requestor_jid
and sub.state == SubscriptionState.SUBSCRIBED
for sub in node.subscriptions
)
if not is_subscribed:
raise error.StanzaError("forbidden", text="Insufficient privileges")
if not has_publish_rights:
raise error.StanzaError("forbidden", text="Insufficient privileges")
ret_pubsub_elt = domish.Element((NS_PUBSUB_EVENT, "pubsub"))
ret_publish_elt = ret_pubsub_elt.addElement("publish")
@ -372,6 +360,14 @@ class PubsubResource(pubsub.PubSubResource):
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
)
if not has_access:
raise error.StanzaError(
"forbidden", text="Insufficient privileges to subscribe to this node"
)
affiliation = next(
(
aff_data.affiliation
@ -571,13 +567,17 @@ class PubsubResource(pubsub.PubSubResource):
except ValueError:
pass
# Store relationship fields and normalize empty strings to None
config.parent = request.options.get(PARENT_VAR) or None
config.link = request.options.get(LINK_VAR) or None
owner_affiliation = PubsubAffiliation(
entity=requestor_jid,
affiliation=Affiliation.owner,
)
try:
await G.storage.set_pubsub_node(
node = await G.storage.set_pubsub_node(
client=client,
service=client.jid,
name=node_id,
@ -588,6 +588,35 @@ class PubsubResource(pubsub.PubSubResource):
except exceptions.ConflictError:
raise error.StanzaError("conflict", text=f"node {node_id!r} already exists.")
if config.parent is not None or config.link is not None:
try:
parent_node, linked_node = await self.xep_0496.check_relationships(
self.client,
self.client.jid,
node,
requestor_jid,
config.parent,
config.link,
)
except Exception as e:
raise error.StanzaError(
"not-allowed",
appCondition="invalid-option",
text=str(e),
)
update_values = {}
# Set parent if explicitly provided or auto-set from linked node
if parent_node:
update_values["parent_node_id"] = parent_node.id
if linked_node:
update_values["linked_node_id"] = linked_node.id
if update_values:
await G.storage.update_pubsub_node(node, update_values)
return node_id
def default(self, request: pubsub.PubSubRequest) -> defer.Deferred[dict[str, object]]:
@ -641,7 +670,12 @@ class PubsubResource(pubsub.PubSubResource):
assert request.nodeIdentifier is not None
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
client,
client.jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
with_link=True,
)
if node is None:
@ -682,6 +716,12 @@ class PubsubResource(pubsub.PubSubResource):
# "max" if node.max_items == -1 else str(node.max_items)
# )
if node.linked_node is not None:
config_values[LINK_VAR] = node.linked_node.name
if node.parent_node:
config_values[PARENT_VAR] = node.parent_node.name
return config_values
def configureSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
@ -701,12 +741,25 @@ class PubsubResource(pubsub.PubSubResource):
assert request.nodeIdentifier is not None
node = await G.storage.get_pubsub_node(
client, client.jid, request.nodeIdentifier, with_affiliations=True
client,
client.jid,
request.nodeIdentifier,
with_affiliations=True,
with_parent=True,
with_link=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
)
if not has_access:
raise error.StanzaError(
"forbidden", text="Insufficient privileges to configure this node"
)
# Only admins and owner can modifify configuration.
if not G.host.memory.is_admin_jid(requestor_jid):
affiliation = next(
@ -766,6 +819,62 @@ class PubsubResource(pubsub.PubSubResource):
if update_extra:
update_values["extra"] = extra
# Handle relationship
current_parent = node.parent_node.name if node.parent_node else None
current_link = node.linked_node.name if node.linked_node else None
# Field present in form = user wants to change it; field absent = no change
parent_in_form = PARENT_VAR in request.options
link_in_form = LINK_VAR in request.options
new_parent = request.options.get(PARENT_VAR)
new_link = request.options.get(LINK_VAR)
parent_changed = parent_in_form and new_parent != current_parent
link_changed = link_in_form and new_link != current_link
if parent_changed or link_changed:
try:
new_parent_node, new_linked_node = (
await self.xep_0496.check_relationships(
self.client,
self.client.jid,
node,
requestor_jid,
new_parent,
new_link,
)
)
except Exception as e:
raise error.StanzaError(
"not-allowed",
appCondition="invalid-option",
text=str(e),
)
if parent_changed:
if new_parent_node:
update_values["parent_node_id"] = new_parent_node.id
else:
# Clearing parent
update_values["parent_node_id"] = None
await self.xep_0496.update_linking_nodes_parent(
node.id, new_parent_node.id if new_parent_node else None
)
if link_changed:
if new_linked_node:
update_values["linked_node_id"] = new_linked_node.id
else:
# Clearing link
update_values["linked_node_id"] = None
# If link was set and check_relationships auto-set the parent, update it too
if link_changed and new_parent_node and not parent_changed:
# Parent was auto-set by check_relationships to match linked node's parent
update_values["parent_node_id"] = new_parent_node.id
if update_values:
await G.storage.update_pubsub_node(node, update_values)
@ -778,7 +887,7 @@ class PubsubResource(pubsub.PubSubResource):
return defer.ensureDeferred(self._items(request))
async def _items(
self, request: pubsub.PubSubRequest
self, request: rsm.PubSubRequest
) -> list[domish.Element] | tuple[list[domish.Element], rsm.RSMResponse]:
"""Retrieve items from a pubsub node.
@ -801,35 +910,11 @@ class PubsubResource(pubsub.PubSubResource):
if node is None:
raise error.StanzaError("item-not-found", text="Node does not exist")
affiliation = next(
(
aff_data.affiliation
for aff_data in node.affiliations
if aff_data.entity == requestor_jid
),
None,
has_access = await self.xep_0496.has_access(
self.client, self.client.jid, node, requestor_jid
)
access_model = node.access_model or AccessModel.open
if access_model == AccessModel.open:
pass
elif access_model == AccessModel.whitelist:
if affiliation not in (
Affiliation.owner,
Affiliation.publisher,
Affiliation.member,
):
raise error.StanzaError("forbidden", text="Insufficient privileges")
else:
log.error(
f"Unknown access model {access_model!r} for node"
f" {request.nodeIdentifier!r}"
)
raise error.StanzaError(
"internal-server-error",
text=f"Unknown access model {access_model!r}",
)
if not has_access:
raise error.StanzaError("forbidden", text="Insufficient privileges")
items_data, metadata = await G.storage.get_items(
node=node,
@ -994,6 +1079,14 @@ class PubsubResource(pubsub.PubSubResource):
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
)
if not has_access:
raise error.StanzaError(
"forbidden", text="Insufficient privileges to delete this node"
)
if not G.host.memory.is_admin_jid(requestor_jid):
affiliation = next(
(
@ -1007,10 +1100,46 @@ class PubsubResource(pubsub.PubSubResource):
if affiliation != Affiliation.owner:
raise error.StanzaError("forbidden", text="Insufficient privileges")
await self._send_notifications(node, "delete", [])
await G.storage.delete_pubsub_node(
profiles=None, services=[client.jid], names=[node.name]
)
# We'll generate list of nodes to delete due to cascade deletion with
# relationships (cf. XEP-0496).
nodes_to_delete = [node.name]
linking_nodes = await self.xep_0496.get_linking_nodes_by_id(node.id)
for linking_node in linking_nodes:
nodes_to_delete.append(linking_node.name)
descendant_nodes = await self.xep_0496.get_descendant_nodes(node)
for descendant in descendant_nodes:
if descendant.name not in nodes_to_delete:
nodes_to_delete.append(descendant.name)
# Build a map of nodes to delete (reuse already fetched nodes)
nodes_map: dict[str, PubsubNode] = {node.name: node}
nodes_map.update({str(ln.name): ln for ln in linking_nodes})
nodes_map.update({str(dn.name): dn for dn in descendant_nodes})
# Fetch any missing nodes
for node_name in nodes_to_delete:
if node_name not in nodes_map:
fetched_node = await G.storage.get_pubsub_node(
client,
client.jid,
node_name,
with_affiliations=True,
with_subscriptions=True,
)
if fetched_node:
nodes_map[node_name] = fetched_node
for node_name in nodes_to_delete:
node_to_delete = nodes_map.get(node_name)
if node_to_delete:
await self._send_notifications(node_to_delete, "delete", [])
if nodes_to_delete:
await G.storage.delete_pubsub_node(
profiles=None, services=[client.jid], names=nodes_to_delete
)
def affiliationsGet(
self, request: pubsub.PubSubRequest
@ -1038,6 +1167,14 @@ class PubsubResource(pubsub.PubSubResource):
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
)
if not has_access:
raise error.StanzaError(
"forbidden", text="Insufficient privileges to retrieve affiliations"
)
if not G.host.memory.is_admin_jid(requestor_jid):
affiliation = next(
(
@ -1081,6 +1218,14 @@ class PubsubResource(pubsub.PubSubResource):
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
)
if not has_access:
raise error.StanzaError(
"forbidden", text="Insufficient privileges to manage affiliations"
)
if not G.host.memory.is_admin_jid(requestor_jid):
affiliation = next(
(