component pubsub: Handle affiliations setting/retrieval:
Let retrieve affiliations for the whole service or a specific node, and set affiliations for a node. rel 476
This commit is contained in:
Goffi
parent
8ebf5360a2
commit
34d6032cf5
2 changed files with 262 additions and 37 deletions
|
|
@ -58,6 +58,7 @@ from libervia.backend.memory import migration
|
|||
from libervia.backend.memory import sqla_config
|
||||
from libervia.backend.memory.sqla_mapping import (
|
||||
AccessModel,
|
||||
Affiliation,
|
||||
Base,
|
||||
Component,
|
||||
File,
|
||||
|
|
@ -1788,9 +1789,7 @@ class Storage:
|
|||
|
||||
# item_ids with RSM doesn't make sense
|
||||
if item_ids is not None and use_rsm:
|
||||
raise exceptions.BadRequest(
|
||||
"item_ids cannot be used with RSM parameters"
|
||||
)
|
||||
raise exceptions.BadRequest("item_ids cannot be used with RSM parameters")
|
||||
|
||||
if use_rsm:
|
||||
return await self._get_items_rsm(
|
||||
|
|
@ -1810,6 +1809,107 @@ class Storage:
|
|||
result = result.scalars().all()
|
||||
return result, metadata
|
||||
|
||||
@aio
|
||||
async def get_pubsub_affiliations(
|
||||
self,
|
||||
client: SatXMPPEntity,
|
||||
service: jid.JID | None = None,
|
||||
entity: jid.JID | None = None,
|
||||
node_name: str | None = None,
|
||||
) -> list[tuple[str, jid.JID, Affiliation]]:
|
||||
"""Retrieve affiliations from the pubsub service.
|
||||
|
||||
@param client: Affiliation from this client profile will be retrieved.
|
||||
@param service: If set, filter to affiliations for this pubsub service.
|
||||
@param entity: If set, filter to affiliations for this entity.
|
||||
@param node_name: If set, filter to affiliations for this node.
|
||||
@return: List of (node_name, entity, affiliation) tuples.
|
||||
"""
|
||||
stmt = (
|
||||
select(
|
||||
PubsubNode.name, PubsubAffiliation.entity, PubsubAffiliation.affiliation
|
||||
)
|
||||
.join(PubsubNode, PubsubAffiliation.node_id == PubsubNode.id)
|
||||
.where(
|
||||
PubsubNode.profile_id == self.profiles[client.profile],
|
||||
)
|
||||
)
|
||||
|
||||
if service is not None:
|
||||
stmt = stmt.where(PubsubNode.service == service)
|
||||
if entity is not None:
|
||||
stmt = stmt.where(PubsubAffiliation.entity == entity)
|
||||
if node_name is not None:
|
||||
stmt = stmt.where(PubsubNode.name == node_name)
|
||||
|
||||
async with self.session() as session:
|
||||
result = await session.execute(stmt)
|
||||
|
||||
return [(row.name, row.entity, row.affiliation) for row in result.all()]
|
||||
|
||||
@aio
|
||||
async def set_pubsub_affiliations(
|
||||
self,
|
||||
client: SatXMPPEntity,
|
||||
service: jid.JID,
|
||||
node_name: str,
|
||||
affiliations: dict[jid.JID, Affiliation | None],
|
||||
) -> None:
|
||||
"""Set affiliations for a node.
|
||||
|
||||
Uses upsert to insert new affiliations or update existing ones.
|
||||
Affiliations with value None are removed.
|
||||
|
||||
@param client: The profile from this client will be used.
|
||||
@param service: The pubsub service JID.
|
||||
@param node_name: The node whose affiliations to set.
|
||||
@param affiliations: Dict mapping entity JIDs to their new affiliations.
|
||||
If affiliation is None, the entity's affiliation is removed.
|
||||
@raise exceptions.NotFound: Node doesn't exist.
|
||||
"""
|
||||
node = await as_future(self.get_pubsub_node(client, service, node_name))
|
||||
if node is None:
|
||||
raise exceptions.NotFound(f"Node {node_name!r} not found")
|
||||
|
||||
async with self.session() as session:
|
||||
async with session.begin():
|
||||
to_set = {
|
||||
entity: aff for entity, aff in affiliations.items() if aff is not None
|
||||
}
|
||||
to_delete = [
|
||||
entity for entity, aff in affiliations.items() if aff is None
|
||||
]
|
||||
|
||||
# Upsert affiliations
|
||||
if to_set:
|
||||
for entity, affiliation in to_set.items():
|
||||
stmt = (
|
||||
insert(PubsubAffiliation)
|
||||
.values(
|
||||
node_id=node.id,
|
||||
entity=entity,
|
||||
affiliation=affiliation,
|
||||
)
|
||||
.on_conflict_do_update(
|
||||
index_elements=(
|
||||
PubsubAffiliation.node_id,
|
||||
PubsubAffiliation.entity,
|
||||
),
|
||||
set_={PubsubAffiliation.affiliation: affiliation},
|
||||
)
|
||||
)
|
||||
await session.execute(stmt)
|
||||
|
||||
# Delete affiliations marked as None
|
||||
if to_delete:
|
||||
delete_stmt = delete(PubsubAffiliation).where(
|
||||
PubsubAffiliation.node_id == node.id,
|
||||
PubsubAffiliation.entity.in_(to_delete),
|
||||
)
|
||||
await session.execute(delete_stmt)
|
||||
|
||||
await session.commit()
|
||||
|
||||
def _get_sqlite_path(self, path: List[Union[str, int]]) -> str:
|
||||
"""generate path suitable to query JSON element with SQLite"""
|
||||
return f"${''.join(f'[{p}]' if isinstance(p, int) else f'.{p}' for p in path)}"
|
||||
|
|
|
|||
|
|
@ -17,7 +17,6 @@
|
|||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
from dataclasses import dataclass, field, fields
|
||||
from enum import StrEnum, auto
|
||||
from typing import Literal, cast
|
||||
|
||||
import shortuuid
|
||||
|
|
@ -26,7 +25,7 @@ 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 data_form, disco, itx_xmpp
|
||||
from tx_xmpp import disco, itx_xmpp
|
||||
from tx_xmpp import pubsub, rsm
|
||||
from zope.interface import implementer
|
||||
|
||||
|
|
@ -41,12 +40,11 @@ from libervia.backend.memory.sqla_mapping import (
|
|||
Affiliation,
|
||||
PublishModel,
|
||||
PubsubAffiliation,
|
||||
PubsubNode,
|
||||
)
|
||||
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.tools.config import ComponentSettings
|
||||
from libervia.backend.tools.utils import ensure_deferred
|
||||
from libervia.backend.tools.xml_tools import element_copy
|
||||
|
||||
log = getLogger(__name__)
|
||||
|
||||
|
|
@ -226,10 +224,10 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
"""
|
||||
return PubsubNodeConfig.get_field_definitions()
|
||||
|
||||
def publish(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def publish(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._publish(request))
|
||||
|
||||
async def _publish(self, request: rsm.PubSubRequest) -> None:
|
||||
async def _publish(self, request: pubsub.PubSubRequest) -> None:
|
||||
"""Publish items to a node.
|
||||
|
||||
@raise StanzaError: with "bad_request" if no items are provided
|
||||
|
|
@ -294,7 +292,9 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
await G.storage.cache_pubsub_items(client, node, items_data)
|
||||
|
||||
def subscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def subscribe(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> defer.Deferred[pubsub.Subscription]:
|
||||
return defer.ensureDeferred(self._subscribe(request))
|
||||
|
||||
async def _subscribe(self, request: rsm.PubSubRequest) -> None:
|
||||
|
|
@ -312,16 +312,38 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
async def _subscriptions(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
|
||||
def affiliations(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def affiliations(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> defer.Deferred[list[tuple[str, str]]]:
|
||||
return defer.ensureDeferred(self._affiliations(request))
|
||||
|
||||
async def _affiliations(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _affiliations(self, request: pubsub.PubSubRequest) -> list[tuple[str, str]]:
|
||||
"""Retrieve the affiliations of the requestor for one or all nodes.
|
||||
|
||||
def create(self, request: rsm.PubSubRequest) -> defer.Deferred[str]:
|
||||
@param request: The affiliations request.
|
||||
If nodeIdentifier is set, return affiliation for that specific node.
|
||||
If nodeIdentifier is None, return affiliations for all nodes.
|
||||
@return: A list of (nodeIdentifier, affiliation) tuples.
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
requestor_jid = request.sender.userhostJID()
|
||||
|
||||
affiliations = await G.storage.get_pubsub_affiliations(
|
||||
client,
|
||||
client.jid,
|
||||
entity=requestor_jid,
|
||||
node_name=request.nodeIdentifier,
|
||||
)
|
||||
|
||||
return [
|
||||
(node_name, str(affiliation)) for node_name, _, affiliation in affiliations
|
||||
]
|
||||
|
||||
def create(self, request: pubsub.PubSubRequest) -> defer.Deferred[str]:
|
||||
return defer.ensureDeferred(self._create(request))
|
||||
|
||||
async def _create(self, request: rsm.PubSubRequest) -> str:
|
||||
async def _create(self, request: pubsub.PubSubRequest) -> str:
|
||||
"""Create a new pubsub node.
|
||||
|
||||
@param request: Node creation request.
|
||||
|
|
@ -372,16 +394,16 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
return node_id
|
||||
|
||||
def default(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def default(self, request: pubsub.PubSubRequest) -> defer.Deferred[dict[str, object]]:
|
||||
return defer.ensureDeferred(self._default(request))
|
||||
|
||||
async def _default(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _default(self, request: pubsub.PubSubRequest) -> dict[str, object]:
|
||||
raise pubsub.Unsupported("retrieve-default")
|
||||
|
||||
def configureGet(self, request: rsm.PubSubRequest) -> defer.Deferred[dict]:
|
||||
def configureGet(self, request: pubsub.PubSubRequest) -> defer.Deferred[dict]:
|
||||
return defer.ensureDeferred(self._configureGet(request))
|
||||
|
||||
async def _configureGet(self, request: rsm.PubSubRequest) -> dict:
|
||||
async def _configureGet(self, request: pubsub.PubSubRequest) -> dict:
|
||||
"""Retrieve the configuration for a specific node.
|
||||
|
||||
@param request: The configuration request.
|
||||
|
|
@ -439,10 +461,10 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
return config_values
|
||||
|
||||
def configureSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def configureSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._configureSet(request))
|
||||
|
||||
async def _configureSet(self, request: rsm.PubSubRequest) -> None:
|
||||
async def _configureSet(self, request: pubsub.PubSubRequest) -> None:
|
||||
"""Set the configuration of a node.
|
||||
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
|
|
@ -526,12 +548,12 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
def items(
|
||||
self,
|
||||
request: rsm.PubSubRequest,
|
||||
request: pubsub.PubSubRequest,
|
||||
) -> defer.Deferred[tuple[list[domish.Element], rsm.RSMResponse | None]]:
|
||||
return defer.ensureDeferred(self._items(request))
|
||||
|
||||
async def _items(
|
||||
self, request: rsm.PubSubRequest
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> tuple[list[domish.Element], rsm.RSMResponse | None]:
|
||||
"""Retrieve items from a pubsub node.
|
||||
|
||||
|
|
@ -602,37 +624,140 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
|
||||
return [item.data for item in items_data], rsm_response
|
||||
|
||||
def retract(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def retract(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._retract(request))
|
||||
|
||||
async def _retract(self, request: rsm.PubSubRequest) -> None:
|
||||
async def _retract(self, request: pubsub.PubSubRequest) -> None:
|
||||
return
|
||||
|
||||
def purge(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def purge(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._purge(request))
|
||||
|
||||
async def _purge(self, request: rsm.PubSubRequest) -> None:
|
||||
async def _purge(self, request: pubsub.PubSubRequest) -> None:
|
||||
return
|
||||
|
||||
def delete(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def delete(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._delete(request))
|
||||
|
||||
async def _delete(self, request: rsm.PubSubRequest) -> None:
|
||||
async def _delete(self, request: pubsub.PubSubRequest) -> None:
|
||||
return
|
||||
|
||||
def affiliationsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
def affiliationsGet(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> defer.Deferred[dict[jid.JID, str]]:
|
||||
return defer.ensureDeferred(self._affiliationsGet(request))
|
||||
|
||||
async def _affiliationsGet(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _affiliationsGet(self, request: pubsub.PubSubRequest) -> dict[jid.JID, str]:
|
||||
"""Retrieve all affiliations for a specific node.
|
||||
|
||||
def affiliationsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
@param request: The affiliations get request with node identifier.
|
||||
@return: Mapping of entity JIDs to their affiliations.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor is not an owner or admin
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
requestor_jid = request.sender.userhostJID()
|
||||
|
||||
assert request.nodeIdentifier is not None
|
||||
|
||||
node = await G.storage.get_pubsub_node(
|
||||
client, client.jid, request.nodeIdentifier, with_affiliations=True
|
||||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
|
||||
if not G.host.memory.is_admin_jid(requestor_jid):
|
||||
affiliation = next(
|
||||
(
|
||||
aff_data.affiliation
|
||||
for aff_data in node.affiliations
|
||||
if aff_data.entity == requestor_jid
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
if affiliation != Affiliation.owner:
|
||||
raise error.StanzaError("forbidden", text="Insufficient privileges")
|
||||
|
||||
return {
|
||||
affiliation.entity: affiliation.affiliation
|
||||
for affiliation in node.affiliations
|
||||
}
|
||||
|
||||
def affiliationsSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._affiliationsSet(request))
|
||||
|
||||
async def _affiliationsSet(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _affiliationsSet(self, request: pubsub.PubSubRequest) -> None:
|
||||
"""Set affiliations for entities on a specific node.
|
||||
|
||||
def subscriptionsGet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
@param request: The affiliations set request with node identifier and affiliations
|
||||
dict.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor is not an owner or admin
|
||||
@raise StanzaError: with "not_acceptable" if affiliation values are invalid
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
requestor_jid = request.sender.userhostJID()
|
||||
|
||||
assert request.nodeIdentifier is not None
|
||||
|
||||
node = await G.storage.get_pubsub_node(
|
||||
client, client.jid, request.nodeIdentifier, with_affiliations=True
|
||||
)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
|
||||
if not G.host.memory.is_admin_jid(requestor_jid):
|
||||
affiliation = next(
|
||||
(
|
||||
aff_data.affiliation
|
||||
for aff_data in node.affiliations
|
||||
if aff_data.entity == requestor_jid
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
if affiliation != Affiliation.owner:
|
||||
raise error.StanzaError("forbidden", text="Insufficient privileges")
|
||||
|
||||
# Prevent owner from changing their own affiliation to non-owner
|
||||
if request.affiliations is not None:
|
||||
if requestor_jid in request.affiliations:
|
||||
new_affiliation = request.affiliations[requestor_jid]
|
||||
if new_affiliation != "owner":
|
||||
raise error.StanzaError(
|
||||
"forbidden", text="You can't change your own affiliation"
|
||||
)
|
||||
|
||||
# Convert affiliation strings to format expected by "set_pubsub_affiliations"
|
||||
affiliations_to_set: dict[jid.JID, Affiliation | None] = {}
|
||||
if request.affiliations is not None:
|
||||
for entity, affiliation_str in request.affiliations.items():
|
||||
if affiliation_str == "none":
|
||||
# We set the affiliation to None to mark it for removal.
|
||||
affiliations_to_set[entity] = None
|
||||
else:
|
||||
try:
|
||||
affiliation = Affiliation(affiliation_str)
|
||||
affiliations_to_set[entity] = affiliation
|
||||
except ValueError:
|
||||
raise error.StanzaError(
|
||||
"not-acceptable",
|
||||
text=f"Invalid affiliation value: {affiliation_str!r}",
|
||||
)
|
||||
|
||||
if affiliations_to_set:
|
||||
await G.storage.set_pubsub_affiliations(
|
||||
client, client.jid, request.nodeIdentifier, affiliations_to_set
|
||||
)
|
||||
|
||||
def subscriptionsGet(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> defer.Deferred[list[pubsub.Subscription]]:
|
||||
return defer.ensureDeferred(self._subscriptionsGet(request))
|
||||
|
||||
async def _subscriptionsGet(self, request: rsm.PubSubRequest) -> None:
|
||||
|
|
|
|||
Loading…
Reference in a new issue