From a32b8bf9e84170d9cabca2992bed4f82dfa0bd69 Mon Sep 17 00:00:00 2001 From: Goffi Date: Wed, 1 Oct 2025 17:57:47 +0200 Subject: [PATCH] plugin XEP-0050: use Enums and add type hints: - Replaces namedtuple with enums which are more appropriate here. - Add docstrings and type hints to `AdHocCommand` rel 465 --- libervia/backend/plugins/plugin_adhoc_dbus.py | 36 +-- .../plugins/plugin_adhoc_registration.py | 30 +- .../plugins/plugin_comp_ap_gateway/ad_hoc.py | 10 +- .../plugin_comp_file_sharing_management.py | 45 +-- libervia/backend/plugins/plugin_xep_0050.py | 279 +++++++++++------- 5 files changed, 242 insertions(+), 158 deletions(-) diff --git a/libervia/backend/plugins/plugin_adhoc_dbus.py b/libervia/backend/plugins/plugin_adhoc_dbus.py index 027bb03a..05aa14fd 100644 --- a/libervia/backend/plugins/plugin_adhoc_dbus.py +++ b/libervia/backend/plugins/plugin_adhoc_dbus.py @@ -19,6 +19,7 @@ from collections import OrderedDict import os.path +from typing import cast import uuid from twisted.internet import defer @@ -29,6 +30,7 @@ from wokkel import data_form from libervia.backend.core.constants import Const as C from libervia.backend.core.i18n import D_, _ from libervia.backend.core.log import getLogger +from libervia.backend.plugins.plugin_xep_0050 import XEP_0050, Action, Error, Note, Status log = getLogger(__name__) @@ -124,7 +126,7 @@ class AdHocDBus: method=self._ad_hoc_remotes_get, async_=True, ) - self._c = host.plugins["XEP-0050"] + self._c = cast(XEP_0050, host.plugins["XEP-0050"]) host.register_namespace("mediaplayer", NS_MEDIA_PLAYER) self.session_con = None @@ -295,7 +297,7 @@ class AdHocDBus: if len(actions) == 1: # it's our first request, we ask the desired new status - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form("form", title=_("Command selection")) options = [] for path, iface, command in methods: @@ -319,10 +321,10 @@ class AdHocDBus: answer_form = data_form.Form.fromElement(x_elt) command = answer_form["command"] except (KeyError, StopIteration): - raise self._c.AdHocError(self._c.ERROR.BAD_PAYLOAD) + raise self._c.ad_hoc_error(Error.BAD_PAYLOAD) if command not in names_map: - raise self._c.AdHocError(self._c.ERROR.BAD_PAYLOAD) + raise self._c.ad_hoc_error(Error.BAD_PAYLOAD) path, iface, command = names_map[command] proxy = await self.session_con.getRemoteObject(bus_name, path) @@ -337,15 +339,15 @@ class AdHocDBus: del actions[:] names_map.clear() return await d_bus_callback( - client, None, session_data, self._c.ACTION.EXECUTE, node + client, None, session_data, Action.EXECUTE, node ) form = data_form.Form("form", title=_("Updated")) form.addField(data_form.Field("fixed", "Command sent")) - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED payload = None - note = (self._c.NOTE.INFO, _("Command sent")) + note = (Note.INFO, _("Command sent")) else: - raise self._c.AdHocError(self._c.ERROR.INTERNAL) + raise self._c.ad_hoc_error(Error.INTERNAL) return (payload, status, None, note) @@ -408,7 +410,7 @@ class AdHocDBus: client, device_jid, NS_MEDIA_PLAYER, - action=self._c.ACTION.CANCEL, + action=Action.CANCEL, session_id=session_id, ) @@ -458,10 +460,10 @@ class AdHocDBus: bus_names = await self._dbus_list_names() bus_names = [b for b in bus_names if b.startswith(MPRIS_PREFIX)] if len(bus_names) == 0: - note = (self._c.NOTE.INFO, D_("No media player found.")) - return (None, self._c.STATUS.COMPLETED, None, note) + note = (Note.INFO, D_("No media player found.")) + return (None, Status.COMPLETED, None, note) options = [] - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form( "form", title=D_("Media Player Selection"), formNamespace=NS_MEDIA_PLAYER ) @@ -491,8 +493,8 @@ class AdHocDBus: "Hack attempt? Refused bus: {bus_name}" ).format(bus_name=bus_name) ) - note = (self._c.NOTE.ERROR, D_("Invalid player name.")) - return (None, self._c.STATUS.COMPLETED, None, note) + note = (Note.ERROR, D_("Invalid player name.")) + return (None, Status.COMPLETED, None, note) try: proxy = await self.session_con.getRemoteObject( @@ -500,8 +502,8 @@ class AdHocDBus: ) except Exception as e: log.warning(_("Can't get D-Bus proxy: {reason}").format(reason=e)) - note = (self._c.NOTE.ERROR, D_("Media player is not available anymore")) - return (None, self._c.STATUS.COMPLETED, None, note) + note = (Note.ERROR, D_("Media player is not available anymore")) + return (None, Status.COMPLETED, None, note) try: command = command_form["command"] except KeyError: @@ -543,5 +545,5 @@ class AdHocDBus: ) payload = form.toElement() - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING return (payload, status, None, None) diff --git a/libervia/backend/plugins/plugin_adhoc_registration.py b/libervia/backend/plugins/plugin_adhoc_registration.py index 43162cea..bb6d54ea 100644 --- a/libervia/backend/plugins/plugin_adhoc_registration.py +++ b/libervia/backend/plugins/plugin_adhoc_registration.py @@ -18,6 +18,7 @@ import time +from typing import cast from uuid import uuid4 from twisted.internet import defer @@ -30,6 +31,7 @@ from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import D_, _ from libervia.backend.core.log import getLogger from libervia.backend.memory import persistent +from libervia.backend.plugins.plugin_xep_0050 import XEP_0050, Error, Note, Status from libervia.backend.tools.common import data_format, date_utils log = getLogger(__name__) @@ -51,7 +53,7 @@ class AdHocRegistration: def __init__(self, host): log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") self.host = host - self._c = host.plugins["XEP-0050"] + self._c = cast(XEP_0050, host.plugins["XEP-0050"]) self.ad_hoc_registration_data = persistent.LazyPersistentBinaryDict( "registration_links" ) @@ -134,7 +136,7 @@ class AdHocRegistration: if len(actions) == 1: # First request, presenting the form to the user - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form("form", title=D_("Create Registration Link")) form.addField( @@ -216,18 +218,18 @@ class AdHocRegistration: await self.ad_hoc_registration_data.aset(link_id, link_data) - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED payload = None note = ( - self._c.NOTE.INFO, + Note.INFO, D_("Registration link created successfully: {link_id}").format( link_id=link_id ), ) except (KeyError, StopIteration, ValueError): - raise self._c.AdHocError(self._c.ERROR.BAD_PAYLOAD) + raise self._c.ad_hoc_error(Error.BAD_PAYLOAD) else: - raise self._c.AdHocError(self._c.ERROR.INTERNAL) + raise self._c.ad_hoc_error(Error.INTERNAL) return (payload, status, None, note) @@ -256,7 +258,7 @@ class AdHocRegistration: if len(actions) == 1: all_links = await self.ad_hoc_registration_data.all() - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form("form", title=D_("Registered Links")) for link_id, link_data in all_links.items(): @@ -272,10 +274,10 @@ class AdHocRegistration: payload = form.toElement() note = None - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED else: - raise self._c.AdHocError(self._c.ERROR.INTERNAL) + raise self._c.ad_hoc_error(Error.INTERNAL) return (payload, status, None, note) @@ -305,7 +307,7 @@ class AdHocRegistration: if len(actions) == 1: all_links = await self.ad_hoc_registration_data.all() - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form("form", title=D_("Delete Registration Link")) link_options = [data_form.Option(link_id, link_id) for link_id in all_links] @@ -330,17 +332,17 @@ class AdHocRegistration: await self.ad_hoc_registration_data.adel(link_to_delete) - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED payload = None note = ( - self._c.NOTE.INFO, + Note.INFO, D_("Registration link {link_id} deleted successfully.").format( link_id=link_to_delete ), ) except (KeyError, StopIteration, ValueError): - raise self._c.AdHocError(self._c.ERROR.BAD_PAYLOAD) + raise self._c.ad_hoc_error(Error.BAD_PAYLOAD) else: - raise self._c.AdHocError(self._c.ERROR.INTERNAL) + raise self._c.ad_hoc_error(Error.INTERNAL) return (payload, status, None, note) diff --git a/libervia/backend/plugins/plugin_comp_ap_gateway/ad_hoc.py b/libervia/backend/plugins/plugin_comp_ap_gateway/ad_hoc.py index 67290cdc..3a9df8c2 100644 --- a/libervia/backend/plugins/plugin_comp_ap_gateway/ad_hoc.py +++ b/libervia/backend/plugins/plugin_comp_ap_gateway/ad_hoc.py @@ -16,6 +16,7 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . +from typing import cast from twisted.words.protocols.jabber import jid from twisted.words.xish import domish from wokkel import data_form @@ -24,6 +25,7 @@ from libervia.backend.core.constants import Const as C from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger +from libervia.backend.plugins.plugin_xep_0050 import XEP_0050, Note, Status log = getLogger(__name__) @@ -36,7 +38,7 @@ class APAdHocService: def __init__(self, apg): self.host = apg.host self.apg = apg - self._c = self.host.plugins["XEP-0050"] + self._c = cast(XEP_0050, self.host.plugins["XEP-0050"]) def init(self, client: SatXMPPEntity) -> None: self._c.add_ad_hoc_command( @@ -62,7 +64,7 @@ class APAdHocService: command_form = None if command_form is None or len(command_form.fields) == 0: # root request - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form( "form", title="XMPP JID/node to AP actor conversion", @@ -81,7 +83,7 @@ class APAdHocService: xmpp_jid = jid.JID(command_form["jid"]) xmpp_node = command_form.get("node") actor = await self.apg.get_ap_account_from_jid_and_node(xmpp_jid, xmpp_node) - note = (self._c.NOTE.INFO, actor) - status = self._c.STATUS.COMPLETED + note = (Note.INFO, actor) + status = Status.COMPLETED payload = None return (payload, status, None, note) diff --git a/libervia/backend/plugins/plugin_comp_file_sharing_management.py b/libervia/backend/plugins/plugin_comp_file_sharing_management.py index a5fa6c5b..ab464afc 100644 --- a/libervia/backend/plugins/plugin_comp_file_sharing_management.py +++ b/libervia/backend/plugins/plugin_comp_file_sharing_management.py @@ -18,6 +18,7 @@ import os.path from functools import partial +from typing import cast from wokkel import data_form from twisted.internet import defer from twisted.words.protocols.jabber import jid @@ -25,6 +26,8 @@ from libervia.backend.core.i18n import _, D_ from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.log import getLogger +from libervia.backend.plugins.plugin_xep_0050 import XEP_0050, Note, Status, Error +from libervia.backend.plugins.plugin_xep_0264 import XEP_0264 from libervia.backend.tools.common import utils log = getLogger(__name__) @@ -71,8 +74,8 @@ class FileSharingManagement(object): def __init__(self, host): log.info(_("File Sharing Management plugin initialization")) self.host = host - self._c = host.plugins["XEP-0050"] - self._t = host.plugins["XEP-0264"] + self._c = cast(XEP_0050, host.plugins["XEP-0050"]) + self._t = cast(XEP_0264, host.plugins["XEP-0264"]) self.files_path = host.get_local_path(None, C.FILES_DIR) host.bridge.add_method( "file_sharing_delete", @@ -132,9 +135,9 @@ class FileSharingManagement(object): @param reason(unicode): reason of the error @return (tuple): value to return """ - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED payload = None - note = (self._c.NOTE.ERROR, reason) + note = (Note.ERROR, reason) return payload, status, None, note def _get_root_args(self): @@ -142,7 +145,7 @@ class FileSharingManagement(object): @return (tuple): value to return """ - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form( "form", title="File Management", formNamespace=NS_FILE_MANAGEMENT ) @@ -169,10 +172,10 @@ class FileSharingManagement(object): path = fields["path"].value.strip() namespace = fields["namespace"].value or None except KeyError: - self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) + self._c.ad_hoc_error(Error.BAD_PAYLOAD) if not path: - self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) + self._c.ad_hoc_error(Error.BAD_PAYLOAD) requestor = session_data["requestor"] requestor_bare = requestor.userhostJID() @@ -302,7 +305,7 @@ class FileSharingManagement(object): ) form.addField(field) - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING payload = form.toElement() return (payload, status, None, None) @@ -311,7 +314,7 @@ class FileSharingManagement(object): try: read_allowed = command_form.fields["read_allowed"] except KeyError: - self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) + self._c.ad_hoc_error(Error.BAD_PAYLOAD) if read_allowed.value == "PUBLIC": allowed_jids = "PUBLIC" @@ -326,7 +329,7 @@ class FileSharingManagement(object): log.warning( _("Can't use read_allowed values: {reason}").format(reason=e) ) - self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) + self._c.ad_hoc_error(Error.BAD_PAYLOAD) if found_file["type"] == C.FILE_TYPE_FILE: await self.host.memory.file_update( @@ -338,7 +341,7 @@ class FileSharingManagement(object): try: recursive = command_form.fields["recursive"] except KeyError: - self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) + self._c.ad_hoc_error(Error.BAD_PAYLOAD) await self.host.memory.file_update( found_file["id"], "access", @@ -354,9 +357,9 @@ class FileSharingManagement(object): ) # job done, we can end the session - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED payload = None - note = (self._c.NOTE.INFO, _("management session done")) + note = (Note.INFO, _("management session done")) return (payload, status, None, note) async def _on_delete_file(self, client, command_elt, session_data, action, node): @@ -403,7 +406,7 @@ class FileSharingManagement(object): desc="check this box to confirm", ) form.addField(field) - status = self._c.STATUS.EXECUTING + status = Status.EXECUTING payload = form.toElement() return (payload, status, None, None) @@ -412,7 +415,7 @@ class FileSharingManagement(object): try: confirmed = C.bool(command_form.fields["confirm"].value) except KeyError: - self._c.ad_hoc_error(self._c.ERROR.BAD_PAYLOAD) + self._c.ad_hoc_error(Error.BAD_PAYLOAD) if not confirmed: note = None else: @@ -420,8 +423,8 @@ class FileSharingManagement(object): await self.host.memory.file_delete( client, requestor_bare, found_file["id"], recursive ) - note = (self._c.NOTE.INFO, _("file deleted")) - status = self._c.STATUS.COMPLETED + note = (Note.INFO, _("file deleted")) + status = Status.COMPLETED payload = None return (payload, status, None, note) @@ -501,9 +504,9 @@ class FileSharingManagement(object): await self._gen_thumbs(client, requestor, found_file["namespace"], found_file) # job done, we can end the session - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED payload = None - note = (self._c.NOTE.INFO, _("thumbnails generated")) + note = (Note.INFO, _("thumbnails generated")) return (payload, status, None, note) async def _on_quota(self, client, command_elt, session_data, action, node): @@ -513,12 +516,12 @@ class FileSharingManagement(object): size_used = await self.host.memory.file_get_used_space(client, requestor) except exceptions.PermissionError: raise WorkflowError(self._err(_("forbidden"))) - status = self._c.STATUS.COMPLETED + status = Status.COMPLETED form = data_form.Form("result") form.makeFields({"quota": quota, "user": size_used}) payload = form.toElement() note = ( - self._c.NOTE.INFO, + Note.INFO, _("You are currently using {size_used} on {size_quota}").format( size_used=utils.get_human_size(size_used), size_quota=( diff --git a/libervia/backend/plugins/plugin_xep_0050.py b/libervia/backend/plugins/plugin_xep_0050.py index b08f3da1..c767df6a 100644 --- a/libervia/backend/plugins/plugin_xep_0050.py +++ b/libervia/backend/plugins/plugin_xep_0050.py @@ -17,24 +17,25 @@ # along with this program. If not, see . -from collections import namedtuple -from uuid import uuid4 +from enum import Enum, StrEnum, auto from typing import Callable +from uuid import uuid4 -from zope.interface import implementer +from twisted.internet import defer from twisted.words.protocols.jabber import jid 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.core_types import SatXMPPEntity +from wokkel import data_form, disco, iwokkel +from zope.interface import implementer + from libervia.backend.core import exceptions +from libervia.backend.core.constants import Const as C +from libervia.backend.core.core_types import SatXMPPEntity +from libervia.backend.core.i18n import D_, _ +from libervia.backend.core.log import getLogger from libervia.backend.memory.memory import Sessions -from libervia.backend.tools import xml_tools, utils +from libervia.backend.tools import utils, xml_tools from libervia.backend.tools.common import data_format @@ -68,50 +69,83 @@ PLUGIN_INFO = { } -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 Status(StrEnum): + EXECUTING = auto() + COMPLETED = auto() + CANCELED = auto() + + +class Action(StrEnum): + EXECUTE = auto() + CANCEL = auto() + NEXT = auto() + PREV = auto() + + +class Note(StrEnum): + INFO = auto() + WARN = auto() + ERROR = auto() + + +class Error(tuple[str, str | None], Enum): + """Errors from XEP-0050 §4.4 Table 5""" + MALFORMED_ACTION = ("bad-request", "malformed-action") + BAD_ACTION = ("bad-request", "bad-action") + BAD_LOCALE = ("bad-request", "bad-locale") + BAD_PAYLOAD = ("bad-request", "bad-payload") + BAD_SESSIONID = ("bad-request", "bad-sessionid") + SESSION_EXPIRED = ("not-allowed", "session-expired") + FORBIDDEN = ("forbidden", None) + ITEM_NOT_FOUND = ("item-not-found", None) + FEATURE_NOT_IMPLEMENTED = ("feature-not-implemented", None) + INTERNAL = ("internal-server-error", None) class AdHocError(Exception): def __init__(self, error_const): """Error to be used from callback - @param error_const: one of XEP_0050.ERROR + @param error_const: one of Error """ - assert error_const in XEP_0050.ERROR + assert error_const in Error self.callback_error = error_const @implementer(iwokkel.IDisco) class AdHocCommand(XMPPHandler): + """Implementation of an Ad-Hoc Command handler for XMPP. + + This class handles the execution of ad-hoc commands as defined in XEP-0050. + It manages command sessions, authorization checks, and communication with clients. + """ + def __init__( self, - callback, - label, - node, - features, - timeout, - allowed_jids, - allowed_groups, - allowed_magics, - forbidden_jids, - forbidden_groups, - ): + callback: Callable, + label: str, + node: str, + features: list[str], + timeout: int, + allowed_jids: list[jid.JID], + allowed_groups: list[str], + allowed_magics: list[str], + forbidden_jids: list[jid.JID], + forbidden_groups: list[str], + ) -> None: + """Initialize the AdHocCommand handler. + + @param callback: The callback function to execute when command is invoked + @param label: Label for the command displayed in UI + @param node: Unique identifier for this command + @param features: List of features supported by this command + @param timeout: Session timeout in seconds + @param allowed_jids: List of JIDs that are allowed to use this command + @param allowed_groups: List of roster groups that are allowed to use this command + @param allowed_magics: List of magic identifiers for special access rules + @param forbidden_jids: List of JIDs that are forbidden from using this command + @param forbidden_groups: List of roster groups that are forbidden from using this + command + """ XMPPHandler.__init__(self) self.callback = callback self.label = label @@ -125,13 +159,28 @@ class AdHocCommand(XMPPHandler): self.sessions = Sessions(timeout=timeout) @property - def client(self): + def client(self) -> SatXMPPEntity: + """Get the client associated with this command handler. + + @return: The XMPP client instance + """ + assert self.parent is not None return self.parent - def getName(self, xml_lang=None): + def getName(self, xml_lang: str | None = None) -> str: + """Get the name of this command for display purposes. + + @param xml_lang: Language code for localization (unused) + @return: The label of this command + """ return self.label - def is_authorised(self, requestor): + def is_authorised(self, requestor: jid.JID) -> bool: + """Check if a requestor is authorized to use this command. + + @param requestor: JID of the entity requesting access + @return: True if authorized, False otherwise + """ if C.ENTITY_ALL in self.allowed_magics: return True forbidden = set(self.forbidden_jids) @@ -153,7 +202,16 @@ class AdHocCommand(XMPPHandler): return True return False - def getDiscoInfo(self, requestor, target, nodeIdentifier=""): + def getDiscoInfo( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoFeature]: + """Get disco info for this command. + + @param requestor: JID of the entity requesting information + @param target: Target JID + @param nodeIdentifier: Node identifier (unused) + @return: List of disco features supported by this command + """ if ( nodeIdentifier != NS_COMMANDS ): # FIXME: we should manage other disco nodes here @@ -161,38 +219,46 @@ class AdHocCommand(XMPPHandler): # identities = [ID_CMD_LIST if self.node == NS_COMMANDS else ID_CMD_NODE] # FIXME return [disco.DiscoFeature(NS_COMMANDS)] + self.features - def getDiscoItems(self, requestor, target, nodeIdentifier=""): + def getDiscoItems( + self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" + ) -> list[disco.DiscoItem]: + """Get disco items for this command. + + @param requestor: JID of the entity requesting information + @param target: Target JID + @param nodeIdentifier: Node identifier (unused) + @return: Empty list as no items are supported + """ return [] def _sendAnswer( self, callback_data: tuple[ - tuple[domish.Element, None], + domish.Element | None, Status, list[str] | None, tuple[Note, str] | None, ], - session_id, - request, + session_id: str, + request: domish.Element, ) -> None: - """Send result of the command + """Send result of the command. @param callback_data: tuple (payload, status, actions, note) with: - - payload: usualy containing data form + - payload: usually containing data form - status: current status - - actions: list of allowed actions (see XEP_0050.ACTION). + - actions: list of allowed actions (see Action). First action is the default one. Default to EXECUTE - note: optional additional note: either None or a tuple with (note type, human readable string) @param session_id: current session id @param request: original request (domish.Element) - @return: deferred """ payload, status, actions, note = callback_data assert isinstance(payload, domish.Element) or payload is None - assert status in XEP_0050.STATUS + assert status in Status if not actions: - actions = [XEP_0050.ACTION.EXECUTE] + actions = [Action.EXECUTE] result = domish.Element((None, "iq")) result["type"] = "result" result["id"] = request["id"] @@ -202,8 +268,8 @@ class AdHocCommand(XMPPHandler): command_elt["node"] = self.node command_elt["status"] = status - if status != XEP_0050.STATUS.CANCELED: - if status != XEP_0050.STATUS.COMPLETED: + if status != Status.CANCELED: + if status != Status.COMPLETED: actions_elt = command_elt.addElement("actions") actions_elt["execute"] = actions[0] for action in actions: @@ -218,13 +284,19 @@ class AdHocCommand(XMPPHandler): command_elt.addChild(payload) self.client.send(result) - if status in (XEP_0050.STATUS.COMPLETED, XEP_0050.STATUS.CANCELED): + if status in (Status.COMPLETED, Status.CANCELED): del self.sessions[session_id] - def _sendError(self, error_constant, session_id, request): - """Send error stanza + def _sendError( + self, + error_constant: tuple[str, str | None], + session_id: str, + request: domish.Element, + ) -> None: + """Send error stanza. - @param error_constant: one of XEP_OO50.ERROR + @param error_constant: one of Error + @param session_id: Session ID @param request: original request (domish.Element) """ xmpp_condition, cmd_condition = error_constant @@ -235,36 +307,54 @@ class AdHocCommand(XMPPHandler): self.client.send(iq_elt) del self.sessions[session_id] - def _request_eb(self, failure_, request, session_id): + def _request_eb( + self, failure_: defer.Failure, request: domish.Element, session_id: str + ) -> None: + """Error handler for command requests. + + @param failure_: The error that occurred + @param request: Original request (domish.Element) + @param session_id: Session ID + """ if failure_.check(AdHocError): error_constant = failure_.value.callback_error else: log.error(f"unexpected error while handling request: {failure_}") - error_constant = XEP_0050.ERROR.INTERNAL + error_constant = Error.INTERNAL self._sendError(error_constant, session_id, request) - def on_request(self, command_elt, requestor, action, session_id): + def on_request( + self, + command_elt: domish.Element, + requestor: jid.JID, + action: str, + session_id: str | None, + ) -> None: + """Handle an incoming command request. + + @param command_elt: The command element from the request + @param requestor: JID of the entity making the request + @param action: The action to perform (execute, cancel, etc.) + @param session_id: Session ID if continuing an existing session + @return: Deferred for handling the response + """ if not self.is_authorised(requestor): - return self._sendError( - XEP_0050.ERROR.FORBIDDEN, session_id, command_elt.parent - ) + return self._sendError(Error.FORBIDDEN, session_id, command_elt.parent) if session_id: try: session_data = self.sessions[session_id] except KeyError: return self._sendError( - XEP_0050.ERROR.SESSION_EXPIRED, session_id, command_elt.parent + Error.SESSION_EXPIRED, session_id, command_elt.parent ) if session_data["requestor"] != requestor: - return self._sendError( - XEP_0050.ERROR.FORBIDDEN, session_id, command_elt.parent - ) + return self._sendError(Error.FORBIDDEN, session_id, command_elt.parent) else: session_id, session_data = self.sessions.new_session() session_data["requestor"] = requestor - if action == XEP_0050.ACTION.CANCEL: - d = defer.succeed((None, XEP_0050.STATUS.CANCELED, None, None)) + if action == Action.CANCEL: + d = defer.succeed((None, Status.CANCELED, None, None)) else: d = utils.as_deferred( self.callback, @@ -279,21 +369,6 @@ class AdHocCommand(XMPPHandler): class XEP_0050: - STATUS = Status("executing", "completed", "canceled") - ACTION = Action("execute", "cancel", "next", "prev") - NOTE = Note("info", "warn", "error") - ERROR = Error( - ("bad-request", "malformed-action"), - ("bad-request", "bad-action"), - ("bad-request", "bad-locale"), - ("bad-request", "bad-payload"), - ("bad-request", "bad-sessionid"), - ("not-allowed", "session-expired"), - ("forbidden", None), - ("item-not-found", None), - ("feature-not-implemented", None), - ("internal-server-error", None), - ) # XEP-0050 §4.4 Table 5 def __init__(self, host): log.info(_("plugin XEP-0050 initialization")) @@ -357,7 +432,7 @@ class XEP_0050: @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 action(unicode): one of Action @param session_id(unicode, None): id of the ad-hoc session None if no session is involved @param form_values(dict, None): values to use to create command form @@ -386,10 +461,10 @@ class XEP_0050: except StopIteration: raise exceptions.NotFound(_("Missing command element")) - def ad_hoc_error(self, error_type): + def ad_hoc_error(self, error_type: Error): """Shortcut to raise an AdHocError - @param error_type(unicode): one of XEP_0050.ERROR + @param error_type(unicode): one of Error """ raise AdHocError(error_type) @@ -436,7 +511,7 @@ class XEP_0050: def parse_command_answer(self, iq_elt): command_elt = self.get_command_elt(iq_elt) data = {} - data["status"] = command_elt.getAttribute("status", XEP_0050.STATUS.EXECUTING) + data["status"] = command_elt.getAttribute("status", Status.EXECUTING) data["session_id"] = command_elt.getAttribute("sessionid") data["notes"] = notes = [] for note_elt in command_elt.elements(NS_COMMANDS, "note"): @@ -458,10 +533,10 @@ class XEP_0050: """ command_elt, answer_data = self.parse_command_answer(iq_elt) status = answer_data["status"] - if status in [XEP_0050.STATUS.COMPLETED, XEP_0050.STATUS.CANCELED]: + if status in [Status.COMPLETED, Status.CANCELED]: # the command session is finished, we purge our session del self.requesting[session_id] - if status == XEP_0050.STATUS.COMPLETED: + if status == Status.COMPLETED: session_id = None else: return None @@ -474,7 +549,7 @@ class XEP_0050: break else: # no matching data element found - if status != XEP_0050.STATUS.COMPLETED: + if status != Status.COMPLETED: log.warning( _("No known payload found in ad-hoc command result, aborting") ) @@ -586,7 +661,7 @@ class XEP_0050: client, entity, session_data["node"], - action=XEP_0050.ACTION.EXECUTE, + action=Action.EXECUTE, session_id=remote_id, form_values=data_form_values, ) @@ -614,7 +689,7 @@ class XEP_0050: if len(actions) == 1: # it's our first request, we ask the desired new status - status = XEP_0050.STATUS.EXECUTING + status = Status.EXECUTING form = data_form.Form("form", title=_("status selection")) show_options = [ data_form.Option(name, label) for name, label in list(SHOWS.items()) @@ -634,20 +709,20 @@ class XEP_0050: answer_form = data_form.Form.fromElement(x_elt) show = answer_form["show"] except (KeyError, StopIteration): - self.ad_hoc_error(XEP_0050.ERROR.BAD_PAYLOAD) + self.ad_hoc_error(Error.BAD_PAYLOAD) if show not in SHOWS: - self.ad_hoc_error(XEP_0050.ERROR.BAD_PAYLOAD) + self.ad_hoc_error(Error.BAD_PAYLOAD) if show == "disconnect": self.host.disconnect(client.profile) else: self.host.presence_set(show=show, profile_key=client.profile) # job done, we can end the session - status = XEP_0050.STATUS.COMPLETED + status = Status.COMPLETED payload = None - note = (self.NOTE.INFO, _("Status updated")) + note = (Note.INFO, _("Status updated")) else: - self.ad_hoc_error(XEP_0050.ERROR.INTERNAL) + self.ad_hoc_error(Error.INTERNAL) return (payload, status, None, note)