view sat/plugins/plugin_exp_pubsub_hook.py @ 3728:b15644cae50d

component AP gateway: JID/node ⟺ AP outbox conversion: - convert a combination of JID and optional pubsub node to AP actor handle (see `getJIDAndNode` for details) and vice versa - the gateway now provides a Pubsub service - retrieve pubsub node and convert it to AP collection, AP pagination is converted to RSM - do the opposite: convert AP collection to pubsub and handle RSM request. Due to ActivityStream collection pagination limitations, some RSM request produce inefficient requests, but caching should be used most of the time in the future and avoid the problem. - set specific name to HTTP Server - new `local_only` setting (`True` by default) to indicate if the gateway can request or not XMPP Pubsub nodes from other servers - disco info now specifies important features such as Pubsub RSM, and nodes metadata ticket 363
author Goffi <goffi@goffi.org>
date Tue, 25 Jan 2022 17:54:06 +0100
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