Mercurial > libervia-backend
view libervia/backend/plugins/plugin_xep_0115.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 | 0d7bb4df2343 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # SAT plugin for managing xep-0115 # 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 libervia.backend.core.i18n import _ from libervia.backend.core.constants import Const as C from libervia.backend.core.log import getLogger log = getLogger(__name__) from twisted.words.xish import domish from twisted.words.protocols.jabber import jid from twisted.internet import defer, error from zope.interface import implementer from wokkel import disco, iwokkel try: from twisted.words.protocols.xmlstream import XMPPHandler except ImportError: from wokkel.subprotocols import XMPPHandler PRESENCE = "/presence" NS_ENTITY_CAPABILITY = "http://jabber.org/protocol/caps" NS_CAPS_OPTIMIZE = "http://jabber.org/protocol/caps#optimize" CAPABILITY_UPDATE = PRESENCE + '/c[@xmlns="' + NS_ENTITY_CAPABILITY + '"]' PLUGIN_INFO = { C.PI_NAME: "XEP 0115 Plugin", C.PI_IMPORT_NAME: "XEP-0115", C.PI_TYPE: "XEP", C.PI_MODES: C.PLUG_MODE_BOTH, C.PI_PROTOCOLS: ["XEP-0115"], C.PI_DEPENDENCIES: [], C.PI_MAIN: "XEP_0115", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("""Implementation of entity capabilities"""), } class XEP_0115(object): cap_hash = None # capabilities hash is class variable as it is common to all profiles def __init__(self, host): log.info(_("Plugin XEP_0115 initialization")) self.host = host host.trigger.add("Presence send", self._presence_trigger) def get_handler(self, client): return XEP_0115_handler(self) @defer.inlineCallbacks def _prepare_caps(self, client): # we have to calculate hash for client # because disco infos/identities may change between clients # optimize check client._caps_optimize = yield self.host.hasFeature(client, NS_CAPS_OPTIMIZE) if client._caps_optimize: log.info(_("Caps optimisation enabled")) client._caps_sent = False else: log.warning(_("Caps optimisation not available")) # hash generation _infos = yield client.discoHandler.info(client.jid, client.jid, "") disco_infos = disco.DiscoInfo() for item in _infos: disco_infos.append(item) cap_hash = client._caps_hash = self.host.memory.disco.generate_hash(disco_infos) log.info( "Our capability hash has been generated: [{cap_hash}]".format( cap_hash=cap_hash ) ) log.debug("Generating capability domish.Element") c_elt = domish.Element((NS_ENTITY_CAPABILITY, "c")) c_elt["hash"] = "sha-1" c_elt["node"] = C.APP_URL c_elt["ver"] = cap_hash client._caps_elt = c_elt if client._caps_optimize: client._caps_sent = False if cap_hash not in self.host.memory.disco.hashes: self.host.memory.disco.hashes[cap_hash] = disco_infos self.host.memory.update_entity_data( client, client.jid, C.ENTITY_CAP_HASH, cap_hash ) def _presence_add_elt(self, client, obj): if client._caps_optimize: if client._caps_sent: return client.caps_sent = True obj.addChild(client._caps_elt) def _presence_trigger(self, client, obj, presence_d): if not hasattr(client, "_caps_optimize"): presence_d.addCallback(lambda __: self._prepare_caps(client)) presence_d.addCallback(lambda __: self._presence_add_elt(client, obj)) return True @implementer(iwokkel.IDisco) class XEP_0115_handler(XMPPHandler): def __init__(self, plugin_parent): self.plugin_parent = plugin_parent self.host = plugin_parent.host @property def client(self): return self.parent def connectionInitialized(self): self.xmlstream.addObserver(CAPABILITY_UPDATE, self.update) def getDiscoInfo(self, requestor, target, nodeIdentifier=""): return [ disco.DiscoFeature(NS_ENTITY_CAPABILITY), disco.DiscoFeature(NS_CAPS_OPTIMIZE), ] def getDiscoItems(self, requestor, target, nodeIdentifier=""): return [] def update(self, presence): """ Manage the capabilities of the entity Check if we know the version of this capabilities and get the capabilities if necessary """ from_jid = jid.JID(presence["from"]) c_elem = next(presence.elements(NS_ENTITY_CAPABILITY, "c")) try: c_ver = c_elem["ver"] c_hash = c_elem["hash"] c_node = c_elem["node"] except KeyError: log.warning(_("Received invalid capabilities tag: %s") % c_elem.toXml()) return if c_ver in self.host.memory.disco.hashes: # we already know the hash, we update the jid entity log.debug( "hash [%(hash)s] already in cache, updating entity [%(jid)s]" % {"hash": c_ver, "jid": from_jid.full()} ) self.host.memory.update_entity_data( self.client, from_jid, C.ENTITY_CAP_HASH, c_ver ) return if c_hash != "sha-1": # unknown hash method log.warning( _( "Unknown hash method for entity capabilities: [{hash_method}] " "(entity: {entity_jid}, node: {node})" ).format(hash_method=c_hash, entity_jid=from_jid, node=c_node) ) def cb(__): computed_hash = self.host.memory.get_entity_datum( self.client, from_jid, C.ENTITY_CAP_HASH ) if computed_hash != c_ver: log.warning( _( "Computed hash differ from given hash:\n" "given: [{given}]\n" "computed: [{computed}]\n" "(entity: {entity_jid}, node: {node})" ).format( given=c_ver, computed=computed_hash, entity_jid=from_jid, node=c_node, ) ) def eb(failure): if isinstance(failure.value, error.ConnectionDone): return msg = ( failure.value.condition if hasattr(failure.value, "condition") else failure.getErrorMessage() ) log.error( _("Couldn't retrieve disco info for {jid}: {error}").format( jid=from_jid.full(), error=msg ) ) d = self.host.get_disco_infos(self.parent, from_jid) d.addCallbacks(cb, eb) # TODO: me must manage the full algorithm described at XEP-0115 #5.4 part 3