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)