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

plugins XEP-0060, XEP-0376, XEP-0465, CLI: PAM + PSS implementation:

Browse source 
- 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
This commit is contained in:
Goffi 2022-05-13 18:38:05 +02:00
commit d20e609a5c
4 changed files with 554 additions and 66 deletions
Download patch file Download diff file Expand all files Collapse all files

132
sat/plugins/plugin_xep_0060.py
View file

@ -272,7 +272,7 @@ class XEP_0060(object):
"psSubscriptionsGet",
".plugin",
in_sign="sss",
out_sign="aa{ss}",
out_sign="s",
method=self._subscriptions,
async_=True,
)
@ -1204,36 +1204,48 @@ class XEP_0060(object):
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 @@ class XEP_0060(object):
# 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 @@ class XEP_0060(object):
request.recipient = service
request.nodeIdentifier = nodeIdentifier
def cb(iq_elt):
try:
subscriptions_elt = next(
iq_elt.pubsub.elements((pubsub.NS_PUBSUB, "subscriptions"))
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 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 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 KeyError:
raise ValueError(
_("Invalid result: bad <subscription> element: {}").format(
iq_elt.toXml
)
d = request.send(client.xmlstream)
d.addCallback(cb)
return d
)
def _setNodeSubscriptions(
self, service_s, nodeIdentifier, subscriptions, profile_key=C.PROF_KEY_NONE

188
sat/plugins/plugin_xep_0376.py Normal file
View file

@ -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 []

248
sat/plugins/plugin_xep_0465.py Normal file
View file

@ -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 []

52
sat_frontends/jp/cmd_pubsub.py
View file

@ -497,11 +497,19 @@ class NodeSubscriptionsGet(base.CommandBase):
)
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 @@ class Subscribe(base.CommandBase):
)
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 @@ class Subscriptions(base.CommandBase):
)
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)