libervia-backend: libervia/backend/plugins/plugin_xep

comparison libervia/backend/plugins/plugin_xep_0050.py @ 4184:50c919dfe61b

plugin XEP-0050: small code quality improvements + add `C.ENTITY_ADMINS` magic key to allow administrators profiles

author	Goffi <goffi@goffi.org>
date	Sun, 10 Dec 2023 15:08:08 +0100
parents	4b842c1fb686
children

comparison

equal deleted inserted replaced

-:6784d07b99c8
+:50c919dfe61b
 #!/usr/bin/env python3
-# SàT plugin for Ad-Hoc Commands (XEP-0050)
+# Libervia plugin for Ad-Hoc Commands (XEP-0050)
-# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
+# Copyright (C) 2009-2023 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.
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 from collections import namedtuple
 from uuid import uuid4
-from typing import List, Optional
+from typing import Callable
 from zope.interface import implementer
 from twisted.words.protocols.jabber import jid
-from twisted.words.protocols import jabber
+from twisted.words.protocols.jabber.error import StanzaError
 from twisted.words.protocols.jabber.xmlstream import XMPPHandler
 from twisted.words.xish import domish
 from twisted.internet import defer
 from wokkel import disco, iwokkel, data_form
 from libervia.backend.core.i18n import _, D_
 from libervia.backend.core.constants import Const as C
 from libervia.backend.core.log import getLogger
-from libervia.backend.core.xmpp import SatXMPPEntity
+from libervia.backend.core.xmpp import SatXMPPClient, SatXMPPEntity
 from libervia.backend.core import exceptions
 from libervia.backend.memory.memory import Sessions
 from libervia.backend.tools import xml_tools, utils
 from libervia.backend.tools.common import data_format
 C.PI_HANDLER: "yes",
 C.PI_DESCRIPTION: _("""Implementation of Ad-Hoc Commands"""),
 }
+Status = namedtuple("Status", ("EXECUTING", "COMPLETED", "CANCELED"))
+Action = namedtuple("Action", ("EXECUTE", "CANCEL", "NEXT", "PREV"))
+Note = namedtuple("Note", ("INFO", "WARN", "ERROR"))
+Error = namedtuple(
+"Error",
+(
+"MALFORMED_ACTION",
+"BAD_ACTION",
+"BAD_LOCALE",
+"BAD_PAYLOAD",
+"BAD_SESSIONID",
+"SESSION_EXPIRED",
+"FORBIDDEN",
+"ITEM_NOT_FOUND",
+"FEATURE_NOT_IMPLEMENTED",
+"INTERNAL",
+),
+)
 class AdHocError(Exception):
 def __init__(self, error_const):
-""" Error to be used from callback
+"""Error to be used from callback
 @param error_const: one of XEP_0050.ERROR
 """
 assert error_const in XEP_0050.ERROR
 self.callback_error = error_const
 @implementer(iwokkel.IDisco)
 class AdHocCommand(XMPPHandler):
+def __init__(
-def __init__(self, callback, label, node, features, timeout,
+self,
-allowed_jids, allowed_groups, allowed_magics, forbidden_jids,
+callback,
-forbidden_groups):
+label,
+node,
+features,
+timeout,
+allowed_jids,
+allowed_groups,
+allowed_magics,
+forbidden_jids,
+forbidden_groups,
+):
 XMPPHandler.__init__(self)
 self.callback = callback
 self.label = label
 self.node = node
 self.features = [disco.DiscoFeature(feature) for feature in features]
 def getName(self, xml_lang=None):
 return self.label
 def is_authorised(self, requestor):
-if "@ALL@" in self.allowed_magics:
+if C.ENTITY_ALL in self.allowed_magics:
 return True
 forbidden = set(self.forbidden_jids)
 for group in self.forbidden_groups:
 forbidden.update(self.client.roster.get_jids_from_group(group))
 if requestor.userhostJID() in forbidden:
 allowed = set(self.allowed_jids)
 for group in self.allowed_groups:
 try:
 allowed.update(self.client.roster.get_jids_from_group(group))
 except exceptions.UnknownGroupError:
-log.warning(_("The groups [{group}] is unknown for profile [{profile}])")
+log.warning(
-.format(group=group, profile=self.client.profile))
+_("The groups [{group}] is unknown for profile [{profile}])").format(
+group=group, profile=self.client.profile
+)
+)
 if requestor.userhostJID() in allowed:
 return True
 return False
 def getDiscoInfo(self, requestor, target, nodeIdentifier=""):
 return [disco.DiscoFeature(NS_COMMANDS)] + self.features
 def getDiscoItems(self, requestor, target, nodeIdentifier=""):
 return []
-def _sendAnswer(self, callback_data, session_id, request):
+def _sendAnswer(
-""" Send result of the command
+self,
+callback_data: tuple[
+tuple[domish.Element, None],
+Status,
+list[str] | None,
+tuple[Note, str] | None,
+],
+session_id,
+request,
+) -> None:
+"""Send result of the command
 @param callback_data: tuple (payload, status, actions, note) with:
-- payload (domish.Element, None) usualy containing data form
+- payload: usualy containing data form
-- status: current status, see XEP_0050.STATUS
+- status: current status
-- actions(list[str], None): list of allowed actions (see XEP_0050.ACTION).
+- actions: list of allowed actions (see XEP_0050.ACTION).
 First action is the default one. Default to EXECUTE
-- note(tuple[str, unicode]): optional additional note: either None or a
+- note: optional additional note: either None or a tuple with (note type,
-tuple with (note type, human readable string), "note type" being in
+human readable string)
-XEP_0050.NOTE
 @param session_id: current session id
 @param request: original request (domish.Element)
 @return: deferred
 """
 payload, status, actions, note = callback_data
 self.client.send(result)
 if status in (XEP_0050.STATUS.COMPLETED, XEP_0050.STATUS.CANCELED):
 del self.sessions[session_id]
 def _sendError(self, error_constant, session_id, request):
-""" Send error stanza
+"""Send error stanza
 @param error_constant: one of XEP_OO50.ERROR
 @param request: original request (domish.Element)
 """
 xmpp_condition, cmd_condition = error_constant
-iq_elt = jabber.error.StanzaError(xmpp_condition).toResponse(request)
+iq_elt = StanzaError(xmpp_condition).toResponse(request)
 if cmd_condition:
 error_elt = next(iq_elt.elements(None, "error"))
 error_elt.addElement(cmd_condition, NS_COMMANDS)
 self.client.send(iq_elt)
 del self.sessions[session_id]
 )
 d.addCallback(self._sendAnswer, session_id, command_elt.parent)
 d.addErrback(self._request_eb, command_elt.parent, session_id)
-class XEP_0050(object):
+class XEP_0050:
-STATUS = namedtuple("Status", ("EXECUTING", "COMPLETED", "CANCELED"))(
+STATUS = Status("executing", "completed", "canceled")
-"executing", "completed", "canceled"
+ACTION = Action("execute", "cancel", "next", "prev")
-)
+NOTE = Note("info", "warn", "error")
-ACTION = namedtuple("Action", ("EXECUTE", "CANCEL", "NEXT", "PREV"))(
+ERROR = Error(
-"execute", "cancel", "next", "prev"
-)
-NOTE = namedtuple("Note", ("INFO", "WARN", "ERROR"))("info", "warn", "error")
-ERROR = namedtuple(
-"Error",
-(
-"MALFORMED_ACTION",
-"BAD_ACTION",
-"BAD_LOCALE",
-"BAD_PAYLOAD",
-"BAD_SESSIONID",
-"SESSION_EXPIRED",
-"FORBIDDEN",
-"ITEM_NOT_FOUND",
-"FEATURE_NOT_IMPLEMENTED",
-"INTERNAL",
-),
-)(
 ("bad-request", "malformed-action"),
 ("bad-request", "bad-action"),
 ("bad-request", "bad-locale"),
 ("bad-request", "bad-payload"),
 ("bad-request", "bad-sessionid"),
 (D_("Service"), D_("Commands")),
 self._commands_menu,
 security_limit=2,
 help_string=D_("Execute ad-hoc commands"),
 )
-host.register_namespace('commands', NS_COMMANDS)
+host.register_namespace("commands", NS_COMMANDS)
 def get_handler(self, client):
 return XEP_0050_handler(self)
 def profile_connected(self, client):
 # map from node to AdHocCommand instance
 client._XEP_0050_commands = {}
 if not client.is_component:
 self.add_ad_hoc_command(client, self._status_callback, _("Status"))
-def do(self, client, entity, node, action=ACTION.EXECUTE, session_id=None,
+def do(
-form_values=None, timeout=30):
+self,
+client,
+entity,
+node,
+action=ACTION.EXECUTE,
+session_id=None,
+form_values=None,
+timeout=30,
+):
 """Do an Ad-Hoc Command
 @param entity(jid.JID): entity which will execture the command
 @param node(unicode): node of the command
 @param action(unicode): one of XEP_0050.ACTION
 @param error_type(unicode): one of XEP_0050.ERROR
 """
 raise AdHocError(error_type)
 def _items_2_xmlui(self, items, no_instructions):
-"""Convert discovery items to XMLUI dialog """
+"""Convert discovery items to XMLUI dialog"""
 # TODO: manage items on different jids
 form_ui = xml_tools.XMLUI("form", submit_id=self.__requesting_id)
 if not no_instructions:
 form_ui.addText(_("Please select a command"), "instructions")
 str(note_elt),
 )
 )
 return command_elt, data
 def _commands_answer_2_xmlui(self, iq_elt, session_id, session_data):
 """Convert command answer to an ui for frontend
 @param iq_elt: command result
 return xmlui
 form = data_form.Form.fromElement(data_elt)
 # we add any present note to the instructions
 form.instructions.extend(self._merge_notes(notes))
-return xml_tools.data_form_2_xmlui(form, self.__requesting_id, session_id=session_id)
+return xml_tools.data_form_2_xmlui(
+form, self.__requesting_id, session_id=session_id
+)
 def _requesting_entity(self, data, profile):
 def serialise(ret_data):
 if "xmlui" in ret_data:
 ret_data["xmlui"] = ret_data["xmlui"].toXml()
 # remote_id is the XEP_0050 sessionid used by answering command
 # while session_id is our own session id used with the frontend
 remote_id = session_data.get("remote_id")
 # we request execute node's command
-d = self.do(client, entity, session_data["node"], action=XEP_0050.ACTION.EXECUTE,
+d = self.do(
-session_id=remote_id, form_values=data_form_values)
+client,
+entity,
+session_data["node"],
+action=XEP_0050.ACTION.EXECUTE,
+session_id=remote_id,
+form_values=data_form_values,
+)
 d.addCallback(self._commands_answer_2_xmlui, session_id, session_data)
 d.addCallback(lambda xmlui: {"xmlui": xmlui} if xmlui is not None else {})
 return d
 xmlui = cb_data["xmlui"]
 xmlui.session_id = session_id
 return xmlui
-def list(self, client, to_jid):
+def list_commands(self, client, to_jid):
 """Request available commands
 @param to_jid(jid.JID, None): the entity answering the commands
 None to use profile's server
 @return D(disco.DiscoItems): found commands
 @param to_jid(jid.JID, None): the entity answering the commands
 None to use profile's server
 @param no_instructions(bool): if True, don't add instructions widget
 @return D(xml_tools.XMLUI): UI with the commands
 """
-d = self.list(client, to_jid)
+d = self.list_commands(client, to_jid)
 d.addCallback(self._items_2_xmlui, no_instructions)
 return d
 def _sequence(self, sequence, node, service_jid_s="", profile_key=C.PROF_KEY_NONE):
 sequence = data_format.deserialise(sequence, type_check=list)
 return d
 async def sequence(
 self,
 client: SatXMPPEntity,
-sequence: List[dict],
+sequence: list[dict],
 node: str,
-service_jid: Optional[jid.JID] = None,
+service_jid: jid.JID | None = None,
 ) -> dict:
 """Send a series of data to an ad-hoc service
 @param sequence: list of values to send
 value are specified by a dict mapping var name to value.
 @param node: node of the ad-hoc commnad
 @param service_jid: jid of the ad-hoc service
 None to use profile's server
 @return: data received in final answer
 """
+assert sequence
+answer_data = None
 if service_jid is None:
 service_jid = jid.JID(client.jid.host)
 session_id = None
 form_values=data_to_send,
 )
 __, answer_data = self.parse_command_answer(iq_result_elt)
 session_id = answer_data.pop("session_id")
+assert answer_data is not None
 return answer_data
-def add_ad_hoc_command(self, client, callback, label, node=None, features=None,
+def add_ad_hoc_command(
-timeout=600, allowed_jids=None, allowed_groups=None,
+self,
-allowed_magics=None, forbidden_jids=None, forbidden_groups=None,
+client: SatXMPPClient,
-):
+callback: Callable,
+label: str,
+node: str | None = None,
+features: list[str] | None = None,
+timeout: int = 600,
+allowed_jids: list[jid.JID] | None = None,
+allowed_groups: list[str] | None = None,
+allowed_magics: list[str] | None = None,
+forbidden_jids: list[jid.JID] | None = None,
+forbidden_groups: list[str] | None = None,
+) -> str:
 """Add an ad-hoc command for the current profile
 @param callback: method associated with this ad-hoc command which return the
-payload data (see AdHocCommand._sendAnswer), can return a
+payload data (see AdHocCommand._sendAnswer), can return a deferred
-deferred
 @param label: label associated with this command on the main menu
 @param node: disco item node associated with this command. None to use
 autogenerated node
 @param features: features associated with the payload (list of strings), usualy
 data form
 @param timeout: delay between two requests before canceling the session (in
 seconds)
 @param allowed_jids: list of allowed entities
 @param allowed_groups: list of allowed roster groups
 @param allowed_magics: list of allowed magic keys, can be:
-@ALL@: allow everybody
+C.ENTITY_ALL: allow everybody
-@PROFILE_BAREJID@: allow only the jid of the profile
+C.ENTITY_PROFILE_BARE: allow only the jid of the profile
+C.ENTITY_ADMINS: any administrator user
 @param forbidden_jids: black list of entities which can't access this command
 @param forbidden_groups: black list of groups which can't access this command
 @return: node of the added command, useful to remove the command later
 """
-# FIXME: "@ALL@" for profile_key seems useless and dangerous
+commands = client._XEP_0050_commands
 if node is None:
-node = "%s_%s" % ("COMMANDS", uuid4())
+node = label.lower().replace(" ", "_")
+if not node:
+node = f"COMMANDS_{uuid4()}"
+elif node in commands:
+node = f"{node}_{uuid4()}"
 if features is None:
 features = [data_form.NS_X_DATA]
 if allowed_jids is None:
 allowed_jids = []
+else:
+# we don't want to modify the initial list
+allowed_jids = allowed_jids.copy()
 if allowed_groups is None:
 allowed_groups = []
 if allowed_magics is None:
-allowed_magics = ["@PROFILE_BAREJID@"]
+allowed_magics = [C.ENTITY_PROFILE_BARE]
 if forbidden_jids is None:
 forbidden_jids = []
 if forbidden_groups is None:
 forbidden_groups = []
 # TODO: manage newly created/removed profiles
-_allowed_jids = (
+if C.ENTITY_PROFILE_BARE in allowed_magics:
-(allowed_jids + [client.jid.userhostJID()])
+allowed_jids += [client.jid.userhostJID()]
-if "@PROFILE_BAREJID@" in allowed_magics
+# TODO: manage dynamic addition/removal of admin status once possible
-else allowed_jids
+if C.ENTITY_ADMINS in allowed_magics:
-)
+allowed_jids += list(self.host.memory.admin_jids)
 ad_hoc_command = AdHocCommand(
 callback,
 label,
 node,
 features,
 timeout,
-_allowed_jids,
+allowed_jids,
 allowed_groups,
 allowed_magics,
 forbidden_jids,
 forbidden_groups,
 )
 ad_hoc_command.setHandlerParent(client)
-commands = client._XEP_0050_commands
 commands[node] = ad_hoc_command
+return node
 def on_cmd_request(self, request, client):
 request.handled = True
 requestor = jid.JID(request["from"])
 command_elt = next(request.elements(NS_COMMANDS, "command"))
 command.on_request(command_elt, requestor, action, sessionid)
 @implementer(iwokkel.IDisco)
 class XEP_0050_handler(XMPPHandler):
 def __init__(self, plugin_parent):
 self.plugin_parent = plugin_parent
 @property
 def client(self):

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0050.py @ 4184:50c919dfe61b