Mercurial > libervia-backend
changeset 3758:b7cef1b24f83
plugins XEP-0060, XEP-0376, XEP-0465, CLI: PAM + PSS implementation:
- update psSubscriptionsGet to use serialised return value
- implement XEP-0376 Pubsub Account Management
- implement XEP-0465 Public Pubsub Subscriptions
- CLI `pubsub` commands updated accordingly, and added `--public` flags to `subscribe`,
`Subscriptions` and `node Subscriptions get`
⚠ `XEP-0465` is speculative, the XEP has been accepted by council but not published yet.
As is should be the next one, and current latest one is `XEP-0464`, `XEP-0465` has been
anticipated.
rel 365
| author | Goffi <goffi@goffi.org> |
|---|---|
| date | Fri, 13 May 2022 18:38:05 +0200 |
| parents | 5bda9d2e8b35 |
| children | c4881833cf8a |
| files | sat/plugins/plugin_xep_0060.py sat/plugins/plugin_xep_0376.py sat/plugins/plugin_xep_0465.py sat_frontends/jp/cmd_pubsub.py |
| diffstat | 4 files changed, 555 insertions(+), 67 deletions(-) [+] |
line wrap: on
line diff
--- a/sat/plugins/plugin_xep_0060.py Fri May 13 18:29:42 2022 +0200 +++ b/sat/plugins/plugin_xep_0060.py Fri May 13 18:38:05 2022 +0200 @@ -272,7 +272,7 @@ "psSubscriptionsGet", ".plugin", in_sign="sss", - out_sign="aa{ss}", + out_sign="s", method=self._subscriptions, async_=True, ) @@ -1204,36 +1204,48 @@ sender, ) - def _subscriptions(self, service, nodeIdentifier="", profile_key=C.PROF_KEY_NONE): + @utils.ensure_deferred + async def _subscriptions( + self, + service="", + nodeIdentifier="", + profile_key=C.PROF_KEY_NONE + ) -> str: client = self.host.getClient(profile_key) service = None if not service else jid.JID(service) + subs = await self.subscriptions(client, service, nodeIdentifier or None) + return data_format.serialise(subs) - def gotSubscriptions(subscriptions): - # we replace pubsub.Subscription instance by dict that we can serialize - for idx, sub in enumerate(subscriptions): - sub_dict = { - "node": sub.nodeIdentifier, - "subscriber": sub.subscriber.full(), - "state": sub.state, - } - if sub.subscriptionIdentifier is not None: - sub_dict["id"] = sub.subscriptionIdentifier - subscriptions[idx] = sub_dict - - return subscriptions - - d = self.subscriptions(client, service, nodeIdentifier or None) - d.addCallback(gotSubscriptions) - return d - - def subscriptions(self, client, service, nodeIdentifier=None): - """retrieve subscriptions from a service + async def subscriptions( + self, + client: SatXMPPEntity, + service: Optional[jid.JID] = None, + node: Optional[str] = None + ) -> List[Dict[str, Union[str, bool]]]: + """Retrieve subscriptions from a service @param service(jid.JID): PubSub service @param nodeIdentifier(unicode, None): node to check None to get all subscriptions """ - return client.pubsub_client.subscriptions(service, nodeIdentifier) + cont, ret = await self.host.trigger.asyncReturnPoint( + "XEP-0060_subscriptions", client, service, node + ) + if not cont: + return ret + subs = await client.pubsub_client.subscriptions(service, node) + ret = [] + for sub in subs: + sub_dict = { + "service": service.host if service else client.jid.host, + "node": sub.nodeIdentifier, + "subscriber": sub.subscriber.full(), + "state": sub.state, + } + if sub.subscriptionIdentifier is not None: + sub_dict["id"] = sub.subscriptionIdentifier + ret.append(sub_dict) + return ret ## misc tools ## @@ -1325,17 +1337,25 @@ # subscribe # - def _getNodeSubscriptions(self, service_s, nodeIdentifier, profile_key): + @utils.ensure_deferred + async def _getNodeSubscriptions( + self, + service: str, + node: str, + profile_key: str + ) -> Dict[str, str]: client = self.host.getClient(profile_key) - d = self.getNodeSubscriptions( - client, jid.JID(service_s) if service_s else None, nodeIdentifier + subs = await self.getNodeSubscriptions( + client, jid.JID(service) if service else None, node ) - d.addCallback( - lambda subscriptions: {j.full(): a for j, a in subscriptions.items()} - ) - return d + return {j.full(): a for j, a in subs.items()} - def getNodeSubscriptions(self, client, service, nodeIdentifier): + async def getNodeSubscriptions( + self, + client: SatXMPPEntity, + service: Optional[jid.JID], + nodeIdentifier: str + ) -> Dict[jid.JID, str]: """Retrieve subscriptions to a node @param nodeIdentifier(unicode): node to get subscriptions from @@ -1346,36 +1366,32 @@ request.recipient = service request.nodeIdentifier = nodeIdentifier - def cb(iq_elt): - try: - subscriptions_elt = next( - iq_elt.pubsub.elements((pubsub.NS_PUBSUB, "subscriptions")) - ) - except StopIteration: - raise ValueError( - _("Invalid result: missing <subscriptions> element: {}").format( - iq_elt.toXml - ) + iq_elt = await request.send(client.xmlstream) + try: + subscriptions_elt = next( + iq_elt.pubsub.elements((pubsub.NS_PUBSUB, "subscriptions")) + ) + except StopIteration: + raise ValueError( + _("Invalid result: missing <subscriptions> element: {}").format( + iq_elt.toXml ) - except AttributeError as e: - raise ValueError(_("Invalid result: {}").format(e)) - try: - return { - jid.JID(s["jid"]): s["subscription"] - for s in subscriptions_elt.elements( - (pubsub.NS_PUBSUB, "subscription") - ) - } - except KeyError: - raise ValueError( - _("Invalid result: bad <subscription> element: {}").format( - iq_elt.toXml - ) + ) + except AttributeError as e: + raise ValueError(_("Invalid result: {}").format(e)) + try: + return { + jid.JID(s["jid"]): s["subscription"] + for s in subscriptions_elt.elements( + (pubsub.NS_PUBSUB, "subscription") ) - - d = request.send(client.xmlstream) - d.addCallback(cb) - return d + } + except KeyError: + raise ValueError( + _("Invalid result: bad <subscription> element: {}").format( + iq_elt.toXml + ) + ) def _setNodeSubscriptions( self, service_s, nodeIdentifier, subscriptions, profile_key=C.PROF_KEY_NONE
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/sat/plugins/plugin_xep_0376.py Fri May 13 18:38:05 2022 +0200 @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 + +# SàT plugin for XEP-0376 +# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# 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 Dict, List, Tuple, Optional, Any +from zope.interface import implementer +from twisted.words.protocols.jabber import jid +from twisted.words.protocols.jabber.xmlstream import XMPPHandler +from wokkel import disco, iwokkel, pubsub, data_form +from sat.core.i18n import _ +from sat.core.constants import Const as C +from sat.core import exceptions +from sat.core.xmpp import SatXMPPEntity +from sat.core.log import getLogger + +log = getLogger(__name__) + +PLUGIN_INFO = { + C.PI_NAME: "Pubsub Account Management", + C.PI_IMPORT_NAME: "XEP-0376", + C.PI_TYPE: C.PLUG_TYPE_XEP, + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_PROTOCOLS: ["XEP-0376"], + C.PI_DEPENDENCIES: ["XEP-0060"], + C.PI_MAIN: "XEP_0376", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _("""Pubsub Account Management"""), +} + +NS_PAM = "urn:xmpp:pam:0" + + +class XEP_0376: + + def __init__(self, host): + log.info(_("Pubsub Account Management initialization")) + self.host = host + host.registerNamespace("pam", NS_PAM) + self._p = self.host.plugins["XEP-0060"] + host.trigger.add("XEP-0060_subscribe", self.subscribe) + host.trigger.add("XEP-0060_unsubscribe", self.unsubscribe) + host.trigger.add("XEP-0060_subscriptions", self.subscriptions) + + def getHandler(self, client): + return XEP_0376_Handler() + + async def profileConnected(self, client): + if not self.host.hasFeature(client, NS_PAM): + log.warning( + "Your server doesn't support Pubsub Account Management, this is used to " + "track all your subscriptions. You may ask your server administrator to " + "install it." + ) + + async def _subRequest( + self, + client: SatXMPPEntity, + service: jid.JID, + nodeIdentifier: str, + sub_jid: Optional[jid.JID], + options: Optional[dict], + subscribe: bool + ) -> None: + if sub_jid is None: + sub_jid = client.jid.userhostJID() + iq_elt = client.IQ() + pam_elt = iq_elt.addElement((NS_PAM, "pam")) + pam_elt["jid"] = service.full() + subscribe_elt = pam_elt.addElement( + (pubsub.NS_PUBSUB, "subscribe" if subscribe else "unsubscribe") + ) + subscribe_elt["node"] = nodeIdentifier + subscribe_elt["jid"] = sub_jid.full() + if options: + options_elt = pam_elt.addElement((pubsub.NS_PUBSUB, "options")) + options_elt["node"] = nodeIdentifier + options_elt["jid"] = sub_jid.full() + form = data_form.Form( + formType='submit', + formNamespace=pubsub.NS_PUBSUB_SUBSCRIBE_OPTIONS + ) + form.makeFields(options) + options_elt.addChild(form.toElement()) + + await iq_elt.send(client.server_jid.full()) + + async def subscribe( + self, + client: SatXMPPEntity, + service: jid.JID, + nodeIdentifier: str, + sub_jid: Optional[jid.JID] = None, + options: Optional[dict] = None + ) -> Tuple[bool, Optional[pubsub.Subscription]]: + if not self.host.hasFeature(client, NS_PAM) or client.is_component: + return True, None + + await self._subRequest(client, service, nodeIdentifier, sub_jid, options, True) + + # TODO: actual result is sent with <message> stanza, we have to get and use them + # to known the actual result. XEP-0376 returns an empty <iq> result, thus we don't + # know here is the subscription actually succeeded + + sub_id = None + sub = pubsub.Subscription(nodeIdentifier, sub_jid, "subscribed", options, sub_id) + return False, sub + + async def unsubscribe( + self, + client: SatXMPPEntity, + service: jid.JID, + nodeIdentifier: str, + sub_jid: Optional[jid.JID], + subscriptionIdentifier: Optional[str], + sender: Optional[jid.JID] = None, + ) -> bool: + if not self.host.hasFeature(client, NS_PAM) or client.is_component: + return True + await self._subRequest(client, service, nodeIdentifier, sub_jid, None, False) + return False + + async def subscriptions( + self, + client: SatXMPPEntity, + service: Optional[jid.JID], + node: str, + ) -> Tuple[bool, Optional[List[Dict[str, Any]]]]: + if not self.host.hasFeature(client, NS_PAM): + return True, None + if service is not None or node is not None: + # if we have service and/or node subscriptions, it's a regular XEP-0060 + # subscriptions request + return True, None + + iq_elt = client.IQ("get") + subscriptions_elt = iq_elt.addElement((NS_PAM, "subscriptions")) + result_elt = await iq_elt.send() + try: + subscriptions_elt = next(result_elt.elements(NS_PAM, "subscriptions")) + except StopIteration: + raise ValueError(f"invalid PAM response: {result_elt.toXml()}") + subs = [] + for subscription_elt in subscriptions_elt.elements(NS_PAM, "subscription"): + sub = {} + try: + for attr, key in ( + ("service", "service"), + ("node", "node"), + ("jid", "subscriber"), + ("subscription", "state") + ): + sub[key] = subscription_elt[attr] + except KeyError as e: + log.warning( + f"Invalid <subscription> element (missing {e.args[0]!r} attribute): " + f"{subscription_elt.toXml()}" + ) + continue + sub_id = subscription_elt.getAttribute("subid") + if sub_id: + sub["id"] = sub_id + subs.append(sub) + + return False, subs + + +@implementer(iwokkel.IDisco) +class XEP_0376_Handler(XMPPHandler): + + def getDiscoInfo(self, requestor, service, nodeIdentifier=""): + return [disco.DiscoFeature(NS_PAM)] + + def getDiscoItems(self, requestor, service, nodeIdentifier=""): + return []
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/sat/plugins/plugin_xep_0465.py Fri May 13 18:38:05 2022 +0200 @@ -0,0 +1,248 @@ +#!/usr/bin/env python3 + +# Libervia plugin for XEP-0465 +# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) + +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. + +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. + +# 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 Optional, List, Dict, Union + +from twisted.words.protocols.jabber.xmlstream import XMPPHandler +from twisted.words.protocols.jabber import jid +from twisted.words.xish import domish +from zope.interface import implementer +from wokkel import disco, iwokkel + +from sat.core.constants import Const as C +from sat.core.i18n import _ +from sat.core.log import getLogger +from sat.core import exceptions +from sat.core.core_types import SatXMPPEntity +from sat.tools import utils +from sat.tools.common import data_format + +log = getLogger(__name__) + +PLUGIN_INFO = { + C.PI_NAME: "Pubsub Public Subscriptions", + C.PI_IMPORT_NAME: "XEP-0465", + C.PI_TYPE: C.PLUG_TYPE_XEP, + C.PI_MODES: C.PLUG_MODE_BOTH, + C.PI_PROTOCOLS: ["XEP-0465"], + C.PI_DEPENDENCIES: ["XEP-0060", "XEP-0376"], + C.PI_MAIN: "XEP_0465", + C.PI_HANDLER: "yes", + C.PI_DESCRIPTION: _("""Pubsub Public Subscriptions implementation"""), +} + +NS_PPS = "urn:xmpp:pps:0" +NS_PPS_SUBSCRIPTIONS = "urn:xmpp:pps:subscriptions:0" +NS_PPS_SUBSCRIBERS = "urn:xmpp:pps:subscribers:0" +SUBSCRIBERS_NODE_PREFIX = f"{NS_PPS_SUBSCRIBERS}/" + + +class XEP_0465: + + def __init__(self, host): + log.info(_("Pubsub Public Subscriptions initialization")) + host.registerNamespace("pps", NS_PPS) + self.host = host + host.bridge.addMethod( + "psPublicSubscriptionsGet", + ".plugin", + in_sign="sss", + out_sign="s", + method=self._subscriptions, + async_=True, + ) + host.bridge.addMethod( + "psPublicSubscriptionsGet", + ".plugin", + in_sign="sss", + out_sign="s", + method=self._subscriptions, + async_=True, + ) + host.bridge.addMethod( + "psPublicNodeSubscriptionsGet", + ".plugin", + in_sign="sss", + out_sign="a{ss}", + method=self._getPublicNodeSubscriptions, + async_=True, + ) + + def getHandler(self, client): + return XEP_0465_Handler() + + @property + def subscriptions_node(self) -> str: + return NS_PPS_SUBSCRIPTIONS + + @property + def subscribers_node_prefix(self) -> str: + return SUBSCRIBERS_NODE_PREFIX + + def buildSubscriptionElt(self, node: str, service: jid.JID) -> domish.Element: + """Generate a <subscriptions> element + + This is the element that a service returns on public subscriptions request + """ + subscription_elt = domish.Element((NS_PPS, "subscription")) + subscription_elt["node"] = node + subscription_elt["service"] = service.full() + return subscription_elt + + def buildSubscriberElt(self, subscriber: jid.JID) -> domish.Element: + """Generate a <subscriber> element + + This is the element that a service returns on node public subscriptions request + """ + subscriber_elt = domish.Element((NS_PPS, "subscriber")) + subscriber_elt["jid"] = subscriber.full() + return subscriber_elt + + @utils.ensure_deferred + async def _subscriptions( + self, + service="", + nodeIdentifier="", + profile_key=C.PROF_KEY_NONE + ) -> str: + client = self.host.getClient(profile_key) + service = None if not service else jid.JID(service) + subs = await self.subscriptions(client, service, nodeIdentifier or None) + return data_format.serialise(subs) + + async def subscriptions( + self, + client: SatXMPPEntity, + service: Optional[jid.JID] = None, + node: Optional[str] = None + ) -> List[Dict[str, Union[str, bool]]]: + """Retrieve public subscriptions from a service + + @param service(jid.JID): PubSub service + @param nodeIdentifier(unicode, None): node to filter + None to get all subscriptions + """ + if service is None: + service = client.jid.userhostJID() + items, __ = await self.host.plugins["XEP-0060"].getItems( + client, service, NS_PPS_SUBSCRIPTIONS + ) + ret = [] + for item in items: + try: + subscription_elt = next(item.elements(NS_PPS, "subscription")) + except StopIteration: + log.warning(f"no <subscription> element found: {item.toXml()}") + continue + + try: + sub_dict = { + "service": subscription_elt["service"], + "node": subscription_elt["node"], + "subscriber": service.full(), + "state": subscription_elt.getAttribute("subscription", "subscribed"), + } + except KeyError: + log.warning( + f"invalid <subscription> element: {subscription_elt.toXml()}" + ) + continue + if node is not None and sub_dict["node"] != node: + # if not is specified, we filter out any other node + # FIXME: should node filtering be done by server? + continue + ret.append(sub_dict) + return ret + + @utils.ensure_deferred + async def _getPublicNodeSubscriptions( + self, + service: str, + node: str, + profile_key: str + ) -> Dict[str, str]: + client = self.host.getClient(profile_key) + subs = await self.getPublicNodeSubscriptions( + client, jid.JID(service) if service else None, node + ) + return {j.full(): a for j, a in subs.items()} + + def getPublicSubscribersNode(self, node: str) -> str: + """Return prefixed node to retrieve public subscribers""" + return f"{NS_PPS_SUBSCRIBERS}/{node}" + + async def getPublicNodeSubscriptions( + self, + client: SatXMPPEntity, + service: Optional[jid.JID], + nodeIdentifier: str + ) -> Dict[jid.JID, str]: + """Retrieve public subscriptions to a node + + @param nodeIdentifier(unicode): node to get subscriptions from + """ + if not nodeIdentifier: + raise exceptions.DataError("node identifier can't be empty") + + if service is None: + service = client.jid.userhostJID() + + subscribers_node = self.getPublicSubscribersNode(nodeIdentifier) + + items, __ = await self.host.plugins["XEP-0060"].getItems( + client, service, subscribers_node + ) + ret = {} + for item in items: + try: + subscriber_elt = next(item.elements(NS_PPS, "subscriber")) + except StopIteration: + log.warning(f"no <subscriber> element found: {item.toXml()}") + continue + + try: + ret[jid.JID(subscriber_elt["jid"])] = "subscribed" + except (KeyError, RuntimeError): + log.warning( + f"invalid <subscriber> element: {subscriber_elt.toXml()}" + ) + continue + return ret + + def setPublicOpt(self, options: Optional[dict] = None) -> dict: + """Set option to make a subscription public + + @param options: dict where the option must be set + if None, a new dict will be created + + @return: the options dict + """ + if options is None: + options = {} + options[f'{{{NS_PPS}}}public'] = True + return options + + +@implementer(iwokkel.IDisco) +class XEP_0465_Handler(XMPPHandler): + + def getDiscoInfo(self, requestor, service, nodeIdentifier=""): + return [disco.DiscoFeature(NS_PPS)] + + def getDiscoItems(self, requestor, service, nodeIdentifier=""): + return []
--- a/sat_frontends/jp/cmd_pubsub.py Fri May 13 18:29:42 2022 +0200 +++ b/sat_frontends/jp/cmd_pubsub.py Fri May 13 18:38:05 2022 +0200 @@ -497,11 +497,19 @@ ) def add_parser_options(self): - pass + self.parser.add_argument( + "--public", + action="store_true", + help=_("get public subscriptions"), + ) async def start(self): + if self.args.public: + method = self.host.bridge.psPublicNodeSubscriptionsGet + else: + method = self.host.bridge.psNodeSubscriptionsGet try: - subscriptions = await self.host.bridge.psNodeSubscriptionsGet( + subscriptions = await method( self.args.service, self.args.node, self.profile, @@ -1464,9 +1472,26 @@ ) def add_parser_options(self): - pass + self.parser.add_argument( + "--public", + action="store_true", + help=_("make the registration visible for everybody"), + ) async def start(self): + options = {} + if self.args.public: + namespaces = await self.host.bridge.namespacesGet() + try: + ns_pps = namespaces["pps"] + except KeyError: + self.disp( + "Pubsub Public Subscription plugin is not loaded, can't use --public " + "option, subscription stopped", error=True + ) + self.host.quit(C.EXIT_MISSING_FEATURE) + else: + options[f"{{{ns_pps}}}public"] = True try: sub_id = await self.host.bridge.psSubscribe( self.args.service, @@ -1528,14 +1553,25 @@ ) def add_parser_options(self): - pass + self.parser.add_argument( + "--public", + action="store_true", + help=_("get public subscriptions"), + ) async def start(self): + if self.args.public: + method = self.host.bridge.psPublicSubscriptionsGet + else: + method = self.host.bridge.psSubscriptionsGet try: - subscriptions = await self.host.bridge.psSubscriptionsGet( - self.args.service, - self.args.node, - self.profile, + subscriptions = data_format.deserialise( + await method( + self.args.service, + self.args.node, + self.profile, + ), + type_check=list ) except Exception as e: self.disp(_("can't retrieve subscriptions: {e}").format(e=e), error=True)