Coverage for slidge/command/adhoc.py: 85%
201 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-18 04:30 +0000
1import asyncio
2import inspect
3import logging
4from collections.abc import Awaitable, Callable
5from functools import partial
6from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast
8from slixmpp import JID, Iq
9from slixmpp.exceptions import XMPPError
10from slixmpp.plugins.xep_0004 import Form as SlixForm
11from slixmpp.plugins.xep_0030.stanza.items import DiscoItems
12from slixmpp.plugins.xep_0050.adhoc import CommandType
14from slidge.util.types import AnyMUC, AnyRecipient, AnySession
16from ..contact import LegacyContact
17from ..core import config
18from ..util.util import strip_leading_emoji
19from . import Command, CommandResponseType, Confirmation, Form, TableResult
20from .base import (
21 CommandResponseRecipientType,
22 CommandResponseSessionType,
23 ContactCommand,
24 FormField,
25 MUCCommand,
26)
27from .categories import CommandCategory
29if TYPE_CHECKING:
30 from ..core.gateway import BaseGateway
33AdhocSessionType = dict[str, Any]
34T = TypeVar("T")
35P = ParamSpec("P")
38class AdhocProvider:
39 """
40 A slixmpp-like plugin to handle adhoc commands, with less boilerplate and
41 untyped dict values than slixmpp.
42 """
44 FORM_TIMEOUT = 120 # seconds
45 xmpp: "BaseGateway"
47 def __init__(self, xmpp: "BaseGateway") -> None:
48 self.xmpp = xmpp
49 self._commands = dict[str, Command[AnySession]]()
50 self._categories = dict[str, list[Command[AnySession]]]()
51 xmpp.plugin["xep_0030"].set_node_handler(
52 "get_items",
53 jid=xmpp.boundjid,
54 node=self.xmpp.plugin["xep_0050"].stanza.Command.namespace,
55 handler=self.get_items,
56 )
57 self.xmpp.plugin["xep_0050"].api.register(self.__get_command, "get_command")
58 self.__timeouts: dict[str, asyncio.TimerHandle] = {}
60 async def __get_command(
61 self,
62 jid: JID | str | None = None,
63 node: str | None = None,
64 ifrom: JID | None = None,
65 args: None = None,
66 ) -> CommandType | None:
67 if node is None:
68 return None
69 if jid is None:
70 jid = self.xmpp.boundjid.bare
71 if not isinstance(jid, JID):
72 jid = JID(jid)
73 if jid == self.xmpp.boundjid.bare:
74 return self.xmpp.plugin["xep_0050"].commands.get((jid.full, node))
75 if ifrom is None:
76 raise XMPPError("undefined-condition")
77 session = self.xmpp.get_session_from_jid(ifrom)
78 if session is None:
79 raise XMPPError("subscription-required")
80 recipient = await session.get_contact_or_group_or_participant(jid)
81 if recipient is None:
82 return None
83 if recipient.is_participant:
84 if recipient.contact is None:
85 return None
86 recipient = recipient.contact
87 command = recipient.commands.get(node)
88 if command is None:
89 return None
90 name = strip_leading_emoji_if_needed(command.NAME)
91 handler = partial(self.__wrap_initial_handler, command, recipient=recipient)
92 return name, handler, None, None # type:ignore
94 async def __wrap_initial_handler(
95 self,
96 command: Command[AnySession]
97 | type[ContactCommand[LegacyContact]]
98 | type[MUCCommand[AnyMUC]],
99 iq: Iq,
100 adhoc_session: AdhocSessionType,
101 recipient: AnyRecipient | None = None,
102 ) -> AdhocSessionType:
103 ifrom = iq.get_from()
104 if recipient is None:
105 cmd = cast(Command[AnySession], command)
106 session = cmd.raise_if_not_authorized(ifrom)
107 result1: CommandResponseSessionType[Any] = await self.__wrap_handler(
108 cmd.run, session, ifrom
109 )
110 return await self.__handle_result(session, result1, adhoc_session)
111 else:
112 cmd2 = cast(
113 type[ContactCommand[LegacyContact]] | type[MUCCommand[AnyMUC]], command
114 )
115 result2: CommandResponseRecipientType[Any] = await self.__wrap_handler(
116 cmd2.run, # type:ignore[arg-type]
117 recipient,
118 )
119 return await self.__handle_result(
120 recipient.session, result2, adhoc_session, recipient
121 )
123 async def __handle_category_list(
124 self, category: CommandCategory, iq: Iq, adhoc_session: AdhocSessionType
125 ) -> AdhocSessionType:
126 try:
127 session = self.xmpp.get_session_from_stanza(iq)
128 except XMPPError:
129 session = None
130 commands: dict[str, Command[AnySession]] = {}
131 for command in self._categories[category.node]:
132 try:
133 command.raise_if_not_authorized(iq.get_from())
134 except XMPPError:
135 continue
136 commands[command.NODE] = command
137 if len(commands) == 0:
138 raise XMPPError(
139 "not-authorized", "There is no command you can run in this category"
140 )
141 return await self.__handle_result(
142 session,
143 Form(
144 category.name,
145 "",
146 [
147 FormField(
148 var="command",
149 label="Command",
150 type="list-single",
151 options=[
152 {
153 "label": strip_leading_emoji_if_needed(command.NAME),
154 "value": command.NODE,
155 }
156 for command in commands.values()
157 ],
158 )
159 ],
160 partial(self.__handle_category_choice, commands),
161 ),
162 adhoc_session,
163 )
165 async def __handle_category_choice(
166 self,
167 commands: dict[str, Command[AnySession]],
168 form_values: dict[str, str],
169 session: "AnySession | None",
170 jid: JID,
171 ) -> CommandResponseSessionType[Any]:
172 command = commands[form_values["command"]]
173 result: CommandResponseSessionType[Any] = await self.__wrap_handler(
174 command.run, session, jid
175 )
176 return result
178 async def __handle_result(
179 self,
180 session: "AnySession | None",
181 result: CommandResponseType,
182 adhoc_session: AdhocSessionType,
183 recipient: AnyRecipient | None = None,
184 ) -> AdhocSessionType:
185 if isinstance(result, str) or result is None:
186 adhoc_session["has_next"] = False
187 adhoc_session["next"] = None
188 adhoc_session["payload"] = None
189 adhoc_session["notes"] = [("info", result or "Success!")]
190 return adhoc_session
192 if isinstance(result, Form):
193 adhoc_session["next"] = partial(
194 self.__wrap_form_handler, session, result, recipient
195 )
196 adhoc_session["has_next"] = True
197 adhoc_session["payload"] = result.get_xml()
198 if result.timeout_handler is not None:
199 self.__timeouts[adhoc_session["id"]] = self.xmpp.loop.call_later(
200 self.FORM_TIMEOUT,
201 partial(
202 self.__wrap_timeout, result.timeout_handler, adhoc_session["id"]
203 ),
204 )
205 return adhoc_session
207 if isinstance(result, Confirmation):
208 adhoc_session["next"] = partial(
209 self.__wrap_confirmation, session, result, recipient
210 )
211 adhoc_session["has_next"] = True
212 adhoc_session["payload"] = result.get_form()
213 return adhoc_session
215 if isinstance(result, TableResult):
216 adhoc_session["next"] = None
217 adhoc_session["has_next"] = False
218 adhoc_session["payload"] = result.get_xml()
219 return adhoc_session
221 raise XMPPError("internal-server-error", text="OOPS!")
223 def __wrap_timeout(self, handler: Callable[[], None], session_id: str) -> None:
224 try:
225 del self.xmpp.plugin["xep_0050"].sessions[session_id]
226 except KeyError:
227 log.error("Timeout but session could not be found: %s", session_id)
228 handler()
230 @staticmethod
231 async def __wrap_handler(
232 f: Callable[P, Awaitable[T] | T],
233 *a: P.args,
234 **k: P.kwargs,
235 ) -> T:
236 try:
237 if inspect.iscoroutinefunction(f):
238 return await f(*a, **k) # type:ignore[no-any-return]
239 elif hasattr(f, "func") and inspect.iscoroutinefunction(f.func):
240 return await f(*a, **k) # type:ignore[misc,no-any-return]
241 else:
242 return f(*a, **k) # type:ignore[return-value]
243 except XMPPError:
244 raise
245 except Exception as e:
246 log.debug("Exception in %s", f, exc_info=e)
247 raise XMPPError("internal-server-error", text=str(e))
249 async def __wrap_form_handler(
250 self,
251 session: "AnySession | None",
252 result: Form,
253 recipient: AnyRecipient | None,
254 form: SlixForm,
255 adhoc_session: AdhocSessionType,
256 ) -> AdhocSessionType:
257 timer = self.__timeouts.pop(adhoc_session["id"], None)
258 if timer is not None:
259 print("canceled", adhoc_session["id"])
260 timer.cancel()
261 form_values = result.get_values(form)
262 if recipient is None:
263 new_result = await self.__wrap_handler(
264 result.handler,
265 form_values,
266 session,
267 adhoc_session["from"],
268 *result.handler_args,
269 **result.handler_kwargs,
270 )
271 else:
272 new_result = await self.__wrap_handler(
273 result.handler,
274 recipient,
275 form_values,
276 *result.handler_args,
277 **result.handler_kwargs,
278 )
279 return await self.__handle_result(session, new_result, adhoc_session, recipient)
281 async def __wrap_confirmation(
282 self,
283 session: "AnySession | None",
284 confirmation: Confirmation,
285 recipient: AnyRecipient | None,
286 form: SlixForm,
287 adhoc_session: AdhocSessionType,
288 ) -> AdhocSessionType:
289 if form.get_values().get("confirm"):
290 if recipient is None:
291 result = await self.__wrap_handler(
292 confirmation.handler,
293 session,
294 adhoc_session["from"],
295 *confirmation.handler_args,
296 **confirmation.handler_kwargs,
297 )
298 if confirmation.success:
299 result = confirmation.success
300 else:
301 result = await self.__wrap_handler(
302 confirmation.handler,
303 recipient,
304 *confirmation.handler_args,
305 **confirmation.handler_kwargs,
306 )
307 else:
308 result = "You canceled the operation"
310 return await self.__handle_result(session, result, adhoc_session, recipient)
312 def register(self, command: Command[AnySession], jid: JID | None = None) -> None:
313 """
314 Register a command as a adhoc command.
316 this does not need to be called manually, ``BaseGateway`` takes care of
317 that.
319 :param command:
320 :param jid:
321 """
322 if jid is None:
323 jid = self.xmpp.boundjid
324 elif not isinstance(jid, JID):
325 jid = JID(jid)
327 if (category := command.CATEGORY) is None:
328 if command.NODE in self._commands:
329 raise RuntimeError(
330 "There is already a command for the node '%s'", command.NODE
331 )
332 self._commands[command.NODE] = command
333 self.xmpp.plugin["xep_0050"].add_command(
334 jid=jid,
335 node=command.NODE,
336 name=strip_leading_emoji_if_needed(command.NAME),
337 handler=partial(self.__wrap_initial_handler, command),
338 )
339 else:
340 if isinstance(category, str):
341 category = CommandCategory(category, category)
342 node = category.node
343 name = category.name
344 if node not in self._categories:
345 self._categories[node] = list[Command[AnySession]]()
346 self.xmpp.plugin["xep_0050"].add_command(
347 jid=jid,
348 node=node,
349 name=strip_leading_emoji_if_needed(name),
350 handler=partial(self.__handle_category_list, category),
351 )
352 self._categories[node].append(command)
354 async def get_items(self, jid: JID, node: str, iq: Iq) -> DiscoItems:
355 """
356 Get items for a disco query
358 :param jid: the entity that should return its items
359 :param node: which command node is requested
360 :param iq: the disco query IQ
361 :return: commands accessible to the given JID will be listed
362 """
363 ifrom = iq.get_from()
364 ifrom_str = str(ifrom)
365 if (
366 not self.xmpp.jid_validator.match(ifrom_str)
367 and ifrom_str not in config.ADMINS
368 ):
369 raise XMPPError(
370 "forbidden",
371 "You are not authorized to execute adhoc commands on this gateway. "
372 "If this is unexpected, ask your administrator to verify that "
373 "'user-jid-validator' is correctly set in slidge's configuration.",
374 )
376 all_items = self.xmpp.plugin["xep_0030"].static.get_items(jid, node, None, None)
377 log.debug("Static items: %r", all_items)
378 if not all_items:
379 return DiscoItems()
381 session = self.xmpp.get_session_from_jid(ifrom)
383 filtered_items = DiscoItems()
384 filtered_items["node"] = self.xmpp.plugin["xep_0050"].stanza.Command.namespace
385 for item in all_items:
386 authorized = True
387 if item["node"] in self._categories:
388 for command in self._categories[item["node"]]:
389 try:
390 command.raise_if_not_authorized(
391 ifrom, fetch_session=False, session=session
392 )
393 except XMPPError:
394 authorized = False
395 else:
396 authorized = True
397 break
398 else:
399 try:
400 self._commands[item["node"]].raise_if_not_authorized(
401 ifrom, fetch_session=False, session=session
402 )
403 except XMPPError:
404 authorized = False
406 if authorized:
407 filtered_items.append(item)
409 return filtered_items
412def strip_leading_emoji_if_needed(text: str) -> str:
413 if config.STRIP_LEADING_EMOJI_ADHOC:
414 return strip_leading_emoji(text)
415 return text
418log = logging.getLogger(__name__)