Mercurial > libervia-backend
view libervia/backend/plugins/plugin_xep_0376.py @ 4334:111dce64dcb5
plugins XEP-0300, XEP-0446, XEP-0447, XEP0448 and others: Refactoring to use Pydantic:
Pydantic models are used more and more in Libervia, for the bridge API, and also to
convert `domish.Element` to internal representation.
Type hints have also been added in many places.
rel 453
author | Goffi <goffi@goffi.org> |
---|---|
date | Tue, 03 Dec 2024 00:12:38 +0100 |
parents | 554a87ae17a6 |
children |
line wrap: on
line source
#!/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 libervia.backend.core.i18n import _ from libervia.backend.core.constants import Const as C from libervia.backend.core import exceptions from libervia.backend.core.xmpp import SatXMPPEntity from libervia.backend.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.register_namespace("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 get_handler(self, client): return XEP_0376_Handler() async def profile_connected(self, client): if not await self.host.memory.disco.has_feature(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 _sub_request( 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 await self.host.memory.disco.has_feature(client, NS_PAM) or client.is_component ): return True, None await self._sub_request(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 await self.host.memory.disco.has_feature(client, NS_PAM) or client.is_component ): return True await self._sub_request(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 await self.host.has_feature(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 []