component pubsub: Implement subscription handling methods:
rel 476
This commit is contained in:
Goffi
parent
0a0735ec4c
commit
5038fdb8f0
1 changed files with 311 additions and 14 deletions
|
|
@ -40,7 +40,7 @@ from libervia.backend.memory.sqla_mapping import (
|
|||
Affiliation,
|
||||
PublishModel,
|
||||
PubsubAffiliation,
|
||||
PubsubNode,
|
||||
SubscriptionState,
|
||||
)
|
||||
from libervia.backend.plugins.plugin_xep_0167 import NS_AV_CONFERENCES
|
||||
from libervia.backend.plugins.plugin_xep_0060 import NodeMetadata, XEP_0060
|
||||
|
|
@ -297,20 +297,174 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
) -> defer.Deferred[pubsub.Subscription]:
|
||||
return defer.ensureDeferred(self._subscribe(request))
|
||||
|
||||
async def _subscribe(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _subscribe(self, request: pubsub.PubSubRequest) -> pubsub.Subscription:
|
||||
"""Handle a subscribe request.
|
||||
|
||||
def unsubscribe(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
@param request: The subscribe request containing node identifier, subscriber JID,
|
||||
and optional subscription options.
|
||||
@return: A Subscription object representing the resulting subscription state.
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor lacks subscribe privileges
|
||||
@raise StanzaError: with "not-authorized" if there's a pending subscription
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
requestor_jid = request.sender.userhostJID()
|
||||
|
||||
assert request.nodeIdentifier is not None
|
||||
assert request.subscriber is not None
|
||||
subscriber_jid = request.subscriber.userhostJID()
|
||||
|
||||
if subscriber_jid != requestor_jid:
|
||||
raise error.StanzaError(
|
||||
"forbidden", text="Cannot subscribe on behalf of another entity"
|
||||
)
|
||||
|
||||
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")
|
||||
|
||||
affiliation = next(
|
||||
(
|
||||
aff_data.affiliation
|
||||
for aff_data in node.affiliations
|
||||
if aff_data.entity == subscriber_jid
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
if affiliation == Affiliation.outcast:
|
||||
raise error.StanzaError("forbidden", text="You can't subscribe this node")
|
||||
|
||||
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(
|
||||
"not-authorized", text="You can't subscribe to this node"
|
||||
)
|
||||
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}",
|
||||
)
|
||||
|
||||
# Check for existing pending subscription
|
||||
existing_subs = await G.storage.get_pubsub_subscriptions(
|
||||
client,
|
||||
node=node,
|
||||
entity=subscriber_jid,
|
||||
)
|
||||
existing_sub = existing_subs[0] if existing_subs else None
|
||||
|
||||
if existing_sub and existing_sub.state == SubscriptionState.PENDING:
|
||||
raise error.StanzaError(
|
||||
"not-authorized", text="You already have a pending subscription"
|
||||
)
|
||||
|
||||
# For now, we auto-approve subscriptions
|
||||
# TODO: Implement pending subscription approval workflow
|
||||
await G.storage.set_pubsub_subscriptions(
|
||||
node,
|
||||
{subscriber_jid: SubscriptionState.SUBSCRIBED},
|
||||
)
|
||||
|
||||
return pubsub.Subscription(
|
||||
nodeIdentifier=request.nodeIdentifier,
|
||||
subscriber=request.subscriber,
|
||||
state="subscribed",
|
||||
)
|
||||
|
||||
def unsubscribe(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._unsubscribe(request))
|
||||
|
||||
async def _unsubscribe(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _unsubscribe(self, request: pubsub.PubSubRequest) -> None:
|
||||
"""Handle an unsubscribe request.
|
||||
|
||||
def subscriptions(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
@param request: The unsubscribe request
|
||||
@raise StanzaError: with "item_not_found" if the node does not exist
|
||||
@raise StanzaError: with "forbidden" if the requestor is not the subscriber
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
requestor_jid = request.sender.userhostJID()
|
||||
|
||||
assert request.nodeIdentifier is not None
|
||||
assert request.subscriber is not None
|
||||
subscriber_jid = request.subscriber.userhostJID()
|
||||
|
||||
# Check that the requestor matches the subscriber
|
||||
if subscriber_jid != requestor_jid:
|
||||
raise error.StanzaError(
|
||||
"forbidden", text="Cannot unsubscribe on behalf of another entity"
|
||||
)
|
||||
|
||||
node = await G.storage.get_pubsub_node(client, client.jid, request.nodeIdentifier)
|
||||
|
||||
if node is None:
|
||||
raise error.StanzaError("item_not_found", text="Node does not exist")
|
||||
|
||||
await G.storage.set_pubsub_subscriptions(
|
||||
node,
|
||||
{subscriber_jid: None},
|
||||
)
|
||||
|
||||
def subscriptions(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> defer.Deferred[list[pubsub.Subscription]]:
|
||||
return defer.ensureDeferred(self._subscriptions(request))
|
||||
|
||||
async def _subscriptions(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _subscriptions(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> list[pubsub.Subscription]:
|
||||
"""Retrieve the subscriptions of the requestor for one or all nodes.
|
||||
|
||||
@param request: The subscriptions request.
|
||||
If nodeIdentifier is set, return subscription for that specific node.
|
||||
If nodeIdentifier is None, return subscriptions for all nodes.
|
||||
@return: A list of Subscription objects.
|
||||
"""
|
||||
client = self.client
|
||||
assert request.sender is not None
|
||||
requestor_jid = request.sender.userhostJID()
|
||||
|
||||
subscriptions_data = await G.storage.get_pubsub_subscriptions(
|
||||
client,
|
||||
service=client.jid,
|
||||
entity=requestor_jid,
|
||||
node_name=request.nodeIdentifier,
|
||||
)
|
||||
|
||||
# Convert to Subscription objects
|
||||
subscriptions = []
|
||||
for pubsub_sub in subscriptions_data:
|
||||
state_str = (
|
||||
"subscribed"
|
||||
if pubsub_sub.state == SubscriptionState.SUBSCRIBED
|
||||
else "pending"
|
||||
)
|
||||
subscriptions.append(
|
||||
pubsub.Subscription(
|
||||
nodeIdentifier=pubsub_sub.node.name,
|
||||
subscriber=pubsub_sub.subscriber,
|
||||
state=state_str,
|
||||
)
|
||||
)
|
||||
|
||||
return subscriptions
|
||||
|
||||
def affiliations(
|
||||
self, request: pubsub.PubSubRequest
|
||||
|
|
@ -760,14 +914,157 @@ class PubsubResource(pubsub.PubSubResource):
|
|||
) -> defer.Deferred[list[pubsub.Subscription]]:
|
||||
return defer.ensureDeferred(self._subscriptionsGet(request))
|
||||
|
||||
async def _subscriptionsGet(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _subscriptionsGet(
|
||||
self, request: pubsub.PubSubRequest
|
||||
) -> list[pubsub.Subscription]:
|
||||
"""Retrieve all subscriptions for a specific node.
|
||||
|
||||
def subscriptionsSet(self, request: rsm.PubSubRequest) -> defer.Deferred[None]:
|
||||
@param request: The subscriptions get request with node identifier.
|
||||
@return: List of Subscription objects for all subscribers of the node.
|
||||
@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_subscriptions=True, 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")
|
||||
|
||||
# Convert subscriptions to Subscription objects
|
||||
subscriptions = []
|
||||
for sub in node.subscriptions:
|
||||
state_str = (
|
||||
"subscribed" if sub.state == SubscriptionState.SUBSCRIBED else "pending"
|
||||
)
|
||||
subscriptions.append(
|
||||
pubsub.Subscription(
|
||||
nodeIdentifier=request.nodeIdentifier,
|
||||
subscriber=sub.subscriber,
|
||||
state=state_str,
|
||||
)
|
||||
)
|
||||
|
||||
return subscriptions
|
||||
|
||||
def subscriptionsSet(self, request: pubsub.PubSubRequest) -> defer.Deferred[None]:
|
||||
return defer.ensureDeferred(self._subscriptionsSet(request))
|
||||
|
||||
async def _subscriptionsSet(self, request: rsm.PubSubRequest) -> None:
|
||||
return
|
||||
async def _subscriptionsSet(self, request: pubsub.PubSubRequest) -> None:
|
||||
"""Set subscriptions for entities on a specific node.
|
||||
|
||||
This allows node owners to manage subscriptions for entities.
|
||||
Owner can approve pending subscriptions or set entities as subscribed.
|
||||
|
||||
@param request: The subscriptions set request with node identifier and
|
||||
subscriptions 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 subscription 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")
|
||||
|
||||
if request.subscriptions is None:
|
||||
return
|
||||
|
||||
# Validate subscriptions from the request and convert them to the format expected
|
||||
# by "set_pubsub_subscriptions"
|
||||
subscriptions_to_set: dict[jid.JID, SubscriptionState | None] = {}
|
||||
for subscription in request.subscriptions:
|
||||
assert subscription.subscriber is not None
|
||||
assert subscription.state is not None
|
||||
entity = subscription.subscriber
|
||||
subscription_state = subscription.state
|
||||
|
||||
match subscription_state:
|
||||
case "none":
|
||||
subscriptions_to_set[entity] = None
|
||||
case "subscribed":
|
||||
entity_affiliation = next(
|
||||
(
|
||||
aff_data.affiliation
|
||||
for aff_data in node.affiliations
|
||||
if aff_data.entity == entity
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
if entity_affiliation == Affiliation.outcast:
|
||||
raise error.StanzaError(
|
||||
"not-acceptable",
|
||||
text=f"Cannot subscribe outcast entity {entity!r}",
|
||||
)
|
||||
|
||||
access_model = node.access_model or AccessModel.open
|
||||
if access_model == AccessModel.whitelist:
|
||||
if entity_affiliation not in (
|
||||
Affiliation.owner,
|
||||
Affiliation.publisher,
|
||||
Affiliation.member,
|
||||
):
|
||||
raise error.StanzaError(
|
||||
"not-acceptable",
|
||||
text=f"Entity {entity!r} is not on the whitelist",
|
||||
)
|
||||
|
||||
subscriptions_to_set[entity] = SubscriptionState.SUBSCRIBED
|
||||
case "pending":
|
||||
subscriptions_to_set[entity] = SubscriptionState.PENDING
|
||||
case "unconfigured":
|
||||
# FIXME: unconfigured is treated as subscribed for now
|
||||
subscriptions_to_set[entity] = SubscriptionState.SUBSCRIBED
|
||||
case _:
|
||||
raise error.StanzaError(
|
||||
"not-acceptable",
|
||||
text=f"Invalid subscription state: {subscription_state!r}",
|
||||
)
|
||||
|
||||
await G.storage.set_pubsub_subscriptions(
|
||||
node, subscriptions_to_set
|
||||
)
|
||||
|
||||
|
||||
class PubsubService(rsm.PubSubService):
|
||||
|
|
|
|||
Loading…
Reference in a new issue