267 lines
9 KiB
Python
267 lines
9 KiB
Python
#!/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.protocols.jabber import error
|
|
from twisted.words.xish import domish
|
|
from zope.interface import implementer
|
|
from wokkel import disco, iwokkel
|
|
|
|
from libervia.backend.core.constants import Const as C
|
|
from libervia.backend.core.i18n import _
|
|
from libervia.backend.core.log import getLogger
|
|
from libervia.backend.core import exceptions
|
|
from libervia.backend.core.core_types import SatXMPPEntity
|
|
from libervia.backend.tools import utils
|
|
from libervia.backend.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}/"
|
|
NOT_IMPLEMENTED_MSG = (
|
|
"The service at {service!s} doesn't seem to support Pubsub Public Subscriptions "
|
|
"(XEP-0465), please request support from your service administrator."
|
|
)
|
|
|
|
|
|
class XEP_0465:
|
|
|
|
def __init__(self, host):
|
|
log.info(_("Pubsub Public Subscriptions initialization"))
|
|
host.register_namespace("pps", NS_PPS)
|
|
self.host = host
|
|
host.bridge.add_method(
|
|
"ps_public_subscriptions_get",
|
|
".plugin",
|
|
in_sign="sss",
|
|
out_sign="s",
|
|
method=self._subscriptions,
|
|
async_=True,
|
|
)
|
|
host.bridge.add_method(
|
|
"ps_public_subscriptions_get",
|
|
".plugin",
|
|
in_sign="sss",
|
|
out_sign="s",
|
|
method=self._subscriptions,
|
|
async_=True,
|
|
)
|
|
host.bridge.add_method(
|
|
"ps_public_node_subscriptions_get",
|
|
".plugin",
|
|
in_sign="sss",
|
|
out_sign="a{ss}",
|
|
method=self._get_public_node_subscriptions,
|
|
async_=True,
|
|
)
|
|
|
|
def get_handler(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 build_subscription_elt(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 build_subscriber_elt(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.get_client(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()
|
|
try:
|
|
items, __ = await self.host.plugins["XEP-0060"].get_items(
|
|
client, service, NS_PPS_SUBSCRIPTIONS
|
|
)
|
|
except error.StanzaError as e:
|
|
if e.condition == "forbidden":
|
|
log.warning(NOT_IMPLEMENTED_MSG.format(service=service))
|
|
return []
|
|
else:
|
|
raise e
|
|
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 _get_public_node_subscriptions(
|
|
self,
|
|
service: str,
|
|
node: str,
|
|
profile_key: str
|
|
) -> Dict[str, str]:
|
|
client = self.host.get_client(profile_key)
|
|
subs = await self.get_public_node_subscriptions(
|
|
client, jid.JID(service) if service else None, node
|
|
)
|
|
return {j.full(): a for j, a in subs.items()}
|
|
|
|
def get_public_subscribers_node(self, node: str) -> str:
|
|
"""Return prefixed node to retrieve public subscribers"""
|
|
return f"{NS_PPS_SUBSCRIBERS}/{node}"
|
|
|
|
async def get_public_node_subscriptions(
|
|
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.get_public_subscribers_node(nodeIdentifier)
|
|
|
|
try:
|
|
items, __ = await self.host.plugins["XEP-0060"].get_items(
|
|
client, service, subscribers_node
|
|
)
|
|
except error.StanzaError as e:
|
|
if e.condition == "forbidden":
|
|
log.warning(NOT_IMPLEMENTED_MSG.format(service=service))
|
|
return {}
|
|
else:
|
|
raise e
|
|
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 set_public_opt(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 []
|