Mercurial > libervia-backend
view sat/plugins/plugin_exp_pubsub_hook.py @ 3913:944f51f9c2b4
core (xmpp): make `send` a blocking method, fix `sendMessageData` calls:
original `send` method is blocking, and it is used as such by Wokkel and thus can't be
changed to an async method easily. However, an Async method is necessary to have an async
trigger at the very end of the send workflow for end-to-end encryption. To workaround
that, `send` is an async method which call `a_send`, an async method which actually does
the sending. This way legacy method can still call `send` while `a_send` can be await
otherwise.
Fix calls to `sendMessageData`: the method now being an `async` one, `ensureDeferred` had
to be used in some calls.
author | Goffi <goffi@goffi.org> |
---|---|
date | Sat, 24 Sep 2022 16:31:39 +0200 |
parents | be6d91572633 |
children | 524856bd7b19 |
line wrap: on
line source
#!/usr/bin/env python3 # SAT plugin for Pubsub Hooks # 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 sat.core.i18n import _ from sat.core.constants import Const as C from sat.core import exceptions from sat.core.log import getLogger from sat.memory import persistent from twisted.words.protocols.jabber import jid from twisted.internet import defer log = getLogger(__name__) NS_PUBSUB_HOOK = "PUBSUB_HOOK" PLUGIN_INFO = { C.PI_NAME: "PubSub Hook", C.PI_IMPORT_NAME: NS_PUBSUB_HOOK, C.PI_TYPE: "EXP", C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: ["XEP-0060"], C.PI_MAIN: "PubsubHook", C.PI_HANDLER: "no", C.PI_DESCRIPTION: _( """Experimental plugin to launch on action on Pubsub notifications""" ), } # python module HOOK_TYPE_PYTHON = "python" # python file path HOOK_TYPE_PYTHON_FILE = "python_file" # python code directly HOOK_TYPE_PYTHON_CODE = "python_code" HOOK_TYPES = (HOOK_TYPE_PYTHON, HOOK_TYPE_PYTHON_FILE, HOOK_TYPE_PYTHON_CODE) class PubsubHook(object): def __init__(self, host): log.info(_("PubSub Hook initialization")) self.host = host self.node_hooks = {} # keep track of the number of hooks per node (for all profiles) host.bridge.addMethod( "psHookAdd", ".plugin", in_sign="ssssbs", out_sign="", method=self._addHook ) host.bridge.addMethod( "psHookRemove", ".plugin", in_sign="sssss", out_sign="i", method=self._removeHook, ) host.bridge.addMethod( "psHookList", ".plugin", in_sign="s", out_sign="aa{ss}", method=self._listHooks, ) @defer.inlineCallbacks def profileConnected(self, client): hooks = client._hooks = persistent.PersistentBinaryDict( NS_PUBSUB_HOOK, client.profile ) client._hooks_temporary = {} yield hooks.load() for node in hooks: self._installNodeManager(client, node) def profileDisconnected(self, client): for node in client._hooks: self._removeNodeManager(client, node) def _installNodeManager(self, client, node): if node in self.node_hooks: log.debug(_("node manager already set for {node}").format(node=node)) self.node_hooks[node] += 1 else: # first hook on this node self.host.plugins["XEP-0060"].addManagedNode( node, items_cb=self._itemsReceived ) self.node_hooks[node] = 0 log.info(_("node manager installed on {node}").format(node=node)) def _removeNodeManager(self, client, node): try: self.node_hooks[node] -= 1 except KeyError: log.error(_("trying to remove a {node} without hook").format(node=node)) else: if self.node_hooks[node] == 0: del self.node_hooks[node] self.host.plugins["XEP-0060"].removeManagedNode(node, self._itemsReceived) log.debug(_("hook removed")) else: log.debug(_("node still needed for an other hook")) def installHook(self, client, service, node, hook_type, hook_arg, persistent): if hook_type not in HOOK_TYPES: raise exceptions.DataError( _("{hook_type} is not handled").format(hook_type=hook_type) ) if hook_type != HOOK_TYPE_PYTHON_FILE: raise NotImplementedError( _("{hook_type} hook type not implemented yet").format( hook_type=hook_type ) ) self._installNodeManager(client, node) hook_data = {"service": service, "type": hook_type, "arg": hook_arg} if persistent: hooks_list = client._hooks.setdefault(node, []) hooks_list.append(hook_data) client._hooks.force(node) else: hooks_list = client._hooks_temporary.setdefault(node, []) hooks_list.append(hook_data) log.info( _("{persistent} hook installed on {node} for {profile}").format( persistent=_("persistent") if persistent else _("temporary"), node=node, profile=client.profile, ) ) def _itemsReceived(self, client, itemsEvent): node = itemsEvent.nodeIdentifier for hooks in (client._hooks, client._hooks_temporary): if node not in hooks: continue hooks_list = hooks[node] for hook_data in hooks_list[:]: if hook_data["service"] != itemsEvent.sender.userhostJID(): continue try: callback = hook_data["callback"] except KeyError: # first time we get this hook, we create the callback hook_type = hook_data["type"] try: if hook_type == HOOK_TYPE_PYTHON_FILE: hook_globals = {} exec(compile(open(hook_data["arg"], "rb").read(), hook_data["arg"], 'exec'), hook_globals) callback = hook_globals["hook"] else: raise NotImplementedError( _("{hook_type} hook type not implemented yet").format( hook_type=hook_type ) ) except Exception as e: log.warning( _( "Can't load Pubsub hook at node {node}, it will be removed: {reason}" ).format(node=node, reason=e) ) hooks_list.remove(hook_data) continue for item in itemsEvent.items: try: callback(self.host, client, item) except Exception as e: log.warning( _( "Error while running Pubsub hook for node {node}: {msg}" ).format(node=node, msg=e) ) def _addHook(self, service, node, hook_type, hook_arg, persistent, profile): client = self.host.getClient(profile) service = jid.JID(service) if service else client.jid.userhostJID() return self.addHook( client, service, str(node), str(hook_type), str(hook_arg), persistent, ) def addHook(self, client, service, node, hook_type, hook_arg, persistent): r"""Add a hook which will be triggered on a pubsub notification @param service(jid.JID): service of the node @param node(unicode): Pubsub node @param hook_type(unicode): type of the hook, one of: - HOOK_TYPE_PYTHON: a python module (must be in path) module must have a "hook" method which will be called - HOOK_TYPE_PYTHON_FILE: a python file file must have a "hook" method which will be called - HOOK_TYPE_PYTHON_CODE: direct python code /!\ Python hooks will be executed in SàT context, with host, client and item as arguments, it means that: - they can do whatever they wants, so don't run untrusted hooks - they MUST NOT BLOCK, they are run in Twisted async environment and blocking would block whole SàT process - item are domish.Element @param hook_arg(unicode): argument of the hook, depending on the hook_type can be a module path, file path, python code """ assert service is not None return self.installHook(client, service, node, hook_type, hook_arg, persistent) def _removeHook(self, service, node, hook_type, hook_arg, profile): client = self.host.getClient(profile) service = jid.JID(service) if service else client.jid.userhostJID() return self.removeHook(client, service, node, hook_type or None, hook_arg or None) def removeHook(self, client, service, node, hook_type=None, hook_arg=None): """Remove a persistent or temporaty root @param service(jid.JID): service of the node @param node(unicode): Pubsub node @param hook_type(unicode, None): same as for [addHook] match all if None @param hook_arg(unicode, None): same as for [addHook] match all if None @return(int): number of hooks removed """ removed = 0 for hooks in (client._hooks, client._hooks_temporary): if node in hooks: for hook_data in hooks[node]: if ( service != hook_data["service"] or hook_type is not None and hook_type != hook_data["type"] or hook_arg is not None and hook_arg != hook_data["arg"] ): continue hooks[node].remove(hook_data) removed += 1 if not hooks[node]: # no more hooks, we can remove the node del hooks[node] self._removeNodeManager(client, node) else: if hooks == client._hooks: hooks.force(node) return removed def _listHooks(self, profile): hooks_list = self.listHooks(self.host.getClient(profile)) for hook in hooks_list: hook["service"] = hook["service"].full() hook["persistent"] = C.boolConst(hook["persistent"]) return hooks_list def listHooks(self, client): """return list of registered hooks""" hooks_list = [] for hooks in (client._hooks, client._hooks_temporary): persistent = hooks is client._hooks for node, hooks_data in hooks.items(): for hook_data in hooks_data: hooks_list.append( { "service": hook_data["service"], "node": node, "type": hook_data["type"], "arg": hook_data["arg"], "persistent": persistent, } ) return hooks_list