Coverage for slidge/command/base.py: 94%
209 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-18 04:30 +0000
1from abc import ABC, abstractmethod
2from collections.abc import Awaitable, Callable, Iterable, Sequence
3from dataclasses import dataclass, field
4from enum import Enum
5from typing import (
6 TYPE_CHECKING,
7 Any,
8 TypedDict,
9 TypeVar,
10 Union,
11)
13from slixmpp import JID
14from slixmpp.exceptions import XMPPError
15from slixmpp.plugins.xep_0004 import Form as SlixForm
16from slixmpp.plugins.xep_0004.stanza.field import FormField as SlixFormField
17from slixmpp.types import JidStr
19from slidge.contact import LegacyContact
20from slidge.group import LegacyMUC
22from ..core import config
23from ..util.types import (
24 AnyMUC,
25 AnySession,
26 FieldType,
27 LegacyContactType,
28 LegacyMUCType,
29 SessionType,
30)
32NODE_PREFIX = "https://slidge.im/command/core/"
34if TYPE_CHECKING:
35 from ..util.types import AnyGateway
36 from .categories import CommandCategory
39HandlerType = (
40 Callable[[AnySession, JID], "CommandResponseType"]
41 | Callable[[AnySession, JID], Awaitable["CommandResponseType"]]
42)
44FormValues = dict[str, str | JID | bool | list[str] | list[JID]]
47@dataclass
48class TableResult:
49 """
50 Structured data as the result of a command
51 """
53 fields: Sequence["FormField"]
54 """
55 The 'columns names' of the table.
56 """
57 items: Sequence[dict[str, str | JID]]
58 """
59 The rows of the table. Each row is a dict where keys are the fields ``var``
60 attribute.
61 """
62 description: str
63 """
64 A description of the content of the table.
65 """
67 jids_are_mucs: bool = False
69 def get_xml(self) -> SlixForm:
70 """
71 Get a slixmpp "form" (with <reported> header)to represent the data
73 :return: some XML
74 """
75 form = SlixForm()
76 form["type"] = "result"
77 form["title"] = self.description
78 for f in self.fields:
79 form.add_reported(f.var, label=f.label, type=f.type)
80 for item in self.items:
81 form.add_item({k: str(v) for k, v in item.items()})
82 return form
85@dataclass
86class SearchResult(TableResult):
87 """
88 Results of the search command (search for contacts via Jabber Search)
90 Return type of :meth:`BaseSession.search`.
91 """
93 description: str = "Contact search results"
96@dataclass
97class Confirmation:
98 """
99 A confirmation 'dialog'
100 """
102 prompt: str
103 """
104 The text presented to the command triggering user
105 """
106 handler: Any
107 """
108 An async function that should return a ResponseType
109 """
110 success: str | None = None
111 """
112 Text in case of success, used if handler does not return anything
113 """
114 handler_args: Iterable[Any] = field(default_factory=list)
115 """
116 arguments passed to the handler
117 """
118 handler_kwargs: dict[str, Any] = field(default_factory=dict)
119 """
120 keyword arguments passed to the handler
121 """
123 def get_form(self) -> SlixForm:
124 """
125 Get the slixmpp form
127 :return: some xml
128 """
129 form = SlixForm()
130 form["type"] = "form"
131 form["title"] = self.prompt
132 form.append(
133 FormField(
134 "confirm", type="boolean", value="true", label="Confirm"
135 ).get_xml()
136 )
137 return form
140@dataclass
141class ConfirmationSession[SessionType: AnySession](Confirmation):
142 handler: Callable[
143 [SessionType | None, JID],
144 Awaitable["CommandResponseSessionType[SessionType]"],
145 ]
148RecipientType = TypeVar("RecipientType", bound=LegacyContact | LegacyMUC[Any])
151@dataclass
152class ConfirmationRecipient[RecipientType: LegacyContact | LegacyMUC[Any]](
153 Confirmation
154):
155 handler: Callable[
156 [RecipientType],
157 Awaitable["CommandResponseRecipientType[RecipientType]"],
158 ]
161@dataclass
162class Form:
163 """
164 A form, to request user input
165 """
167 title: str
168 instructions: str
169 fields: Sequence["FormField"]
170 handler: Any
171 handler_args: Iterable[Any] = field(default_factory=list)
172 handler_kwargs: dict[str, Any] = field(default_factory=dict)
173 timeout_handler: Callable[[], None] | None = None
175 def get_values(
176 self, slix_form: SlixForm
177 ) -> dict[str, list[str] | list[JID] | str | JID | bool | None]:
178 """
179 Parse form submission
181 :param slix_form: the xml received as the submission of a form
182 :return: A dict where keys=field.var and values are either strings
183 or JIDs (if field.type=jid-single)
184 """
185 str_values: dict[str, str] = slix_form.get_values()
186 values = {}
187 for f in self.fields:
188 values[f.var] = f.validate(str_values.get(f.var))
189 return values
191 def get_xml(self) -> SlixForm:
192 """
193 Get the slixmpp "form"
195 :return: some XML
196 """
197 form = SlixForm()
198 form["type"] = "form"
199 form["title"] = self.title
200 form["instructions"] = self.instructions
201 for fi in self.fields:
202 form.append(fi.get_xml())
203 return form
206class FormSession[SessionType: AnySession](Form):
207 handler: Callable[
208 [FormValues, SessionType | None, JID],
209 Awaitable["CommandResponseSessionType[SessionType]"],
210 ]
213@dataclass
214class FormRecipient[RecipientType: LegacyContact | LegacyMUC[Any]](Form):
215 handler: Callable[
216 [RecipientType, FormValues],
217 Awaitable["CommandResponseRecipientType[RecipientType]"],
218 ]
221class CommandAccess(int, Enum):
222 """
223 Defines who can access a given Command
224 """
226 ADMIN_ONLY = 0
227 USER = 1
228 USER_LOGGED = 2
229 USER_NON_LOGGED = 3
230 NON_USER = 4
231 ANY = 5
234class Option(TypedDict):
235 """
236 Options to be used for ``FormField``s of type ``list-*``
237 """
239 label: str
240 value: str
243# TODO: support forms validation XEP-0122
244@dataclass
245class FormField:
246 """
247 Represents a field of the form that a user will see when registering to the gateway
248 via their XMPP client.
249 """
251 var: str = ""
252 """
253 Internal name of the field, will be used to retrieve via :py:attr:`slidge.GatewayUser.registration_form`
254 """
255 label: str | None = None
256 """Description of the field that the user will see"""
257 required: bool = False
258 """Whether this field is mandatory or not"""
259 private: bool = False
260 """
261 For sensitive info that should not be displayed on screen while the user types.
262 Forces field_type to "text-private"
263 """
264 type: FieldType = "text-single"
265 """Type of the field, see `XEP-0004 <https://xmpp.org/extensions/xep-0004.html#protocol-fieldtypes>`_"""
266 value: str = ""
267 """Pre-filled value. Will be automatically pre-filled if a registered user modifies their subscription"""
268 options: list[Option] | None = None
270 image_url: str | None = None
271 """An image associated to this field, eg, a QR code"""
273 def __post_init__(self) -> None:
274 if self.private:
275 self.type = "text-private"
277 def __acceptable_options(self) -> list[str]:
278 if self.options is None:
279 raise RuntimeError
280 return [x["value"] for x in self.options]
282 def validate(
283 self, value: str | list[str] | None
284 ) -> list[str] | list[JID] | str | JID | bool | None:
285 """
286 Raise appropriate XMPPError if a given value is valid for this field
288 :param value: The value to test
289 :return: The same value OR a JID if ``self.type=jid-single``
290 """
291 if isinstance(value, list) and not self.type.endswith("multi"):
292 raise XMPPError("not-acceptable", "A single value was expected")
294 if self.type in ("list-multi", "jid-multi", "text-multi"):
295 if not value:
296 value = []
297 if isinstance(value, list):
298 if self.type == "text-multi":
299 return value
300 return self.__validate_list_multi(value)
301 else:
302 raise XMPPError("not-acceptable", "Multiple values was expected")
304 assert isinstance(value, (str, bool, JID)) or value is None
306 if self.required and value is None:
307 raise XMPPError("not-acceptable", f"Missing field: '{self.label}'")
309 if value is None:
310 return None
312 if self.type == "jid-single":
313 try:
314 return JID(value)
315 except ValueError:
316 raise XMPPError("not-acceptable", f"Not a valid JID: '{value}'")
318 elif self.type == "list-single":
319 if value not in self.__acceptable_options():
320 raise XMPPError("not-acceptable", f"Not a valid option: '{value}'")
322 elif self.type == "boolean":
323 return value.lower() in ("1", "true") if isinstance(value, str) else value
325 return value
327 def __validate_list_multi(self, value: list[str]) -> list[str] | list[JID]:
328 for v in value:
329 if v not in self.__acceptable_options():
330 raise XMPPError("not-acceptable", f"Not a valid option: '{v}'")
331 if self.type == "list-multi":
332 return value
333 return [JID(v) for v in value]
335 def get_xml(self) -> SlixFormField:
336 """
337 Get the field in slixmpp format
339 :return: some XML
340 """
341 f = SlixFormField()
342 f["var"] = self.var
343 f["label"] = self.label
344 f["required"] = self.required
345 f["type"] = self.type
346 if self.options:
347 for o in self.options:
348 f.add_option(**o)
349 f["value"] = self.value
350 if self.image_url:
351 f["media"].add_uri(self.image_url, itype="image/png")
352 return f
355CommandResponseType = TableResult | Confirmation | Form | str | None
357CommandResponseSessionType = (
358 TableResult
359 | ConfirmationSession[SessionType]
360 | FormSession[SessionType]
361 | str
362 | None
363)
365CommandResponseRecipientType = (
366 TableResult
367 | ConfirmationRecipient[RecipientType]
368 | FormRecipient[RecipientType]
369 | str
370 | None
371)
374class CommandBase(ABC):
375 NAME: str = NotImplemented
376 """
377 Friendly name of the command, eg: "do something with stuff"
378 """
379 HELP: str = NotImplemented
380 """
381 Long description of what the command does
382 """
383 NODE: str = NotImplemented
384 """
385 Name of the node used for ad-hoc commands
386 """
387 CHAT_COMMAND: str = NotImplemented
388 """
389 Text to send to the gateway to trigger the command via a message
390 """
393class Command[SessionType: AnySession](CommandBase):
394 """
395 Abstract base class to implement gateway commands (chatbot and ad-hoc)
397 For a command to be available to users, it must be listed in
398 :attr:`.BaseGateway.COMMANDS`.
399 """
401 ACCESS: "CommandAccess" = NotImplemented
402 """
403 Who can use this command
404 """
406 CATEGORY: Union[str, "CommandCategory"] | None = None
407 """
408 If used, the command will be under this top-level category.
409 Use the same string for several commands to group them.
410 This hierarchy only used for the adhoc interface, not the chat command
411 interface.
412 """
414 def __init__(self, xmpp: "AnyGateway") -> None:
415 self.xmpp = xmpp
417 async def run(
418 self,
419 session: SessionType | None,
420 ifrom: JID,
421 *args: str,
422 ) -> CommandResponseSessionType[SessionType]:
423 """
424 Entry point of the command
426 :param session: If triggered by a registered user, its slidge Session
427 :param ifrom: JID of the command-triggering entity
428 :param args: When triggered via chatbot type message, additional words
429 after the CHAT_COMMAND string was passed
431 :return: Either a TableResult, a Form, a Confirmation, a text, or None
432 """
433 raise XMPPError("feature-not-implemented")
435 def _get_session(self, jid: JID) -> SessionType | None:
436 return self.xmpp.get_session_from_jid(jid) # type:ignore
438 def __can_use_command(self, jid: JID) -> bool:
439 j = jid.bare
440 return bool(self.xmpp.jid_validator.match(j) or j in config.ADMINS)
442 def raise_if_not_authorized(
443 self,
444 jid: JID,
445 fetch_session: bool = True,
446 session: SessionType | None = None,
447 ) -> SessionType | None:
448 """
449 Raise an appropriate error is jid is not authorized to use the command
451 :param jid: jid of the entity trying to access the command
452 :param fetch_session:
453 :param session:
455 :return:session of JID if it exists
456 """
457 if not self.__can_use_command(jid):
458 raise XMPPError(
459 "bad-request", "Your JID is not allowed to use this gateway."
460 )
461 if fetch_session:
462 session = self._get_session(jid)
464 if self.ACCESS == CommandAccess.ADMIN_ONLY and not is_admin(jid):
465 raise XMPPError("not-authorized")
466 elif self.ACCESS == CommandAccess.NON_USER and session is not None:
467 raise XMPPError(
468 "bad-request", "This is only available for non-users. Unregister first."
469 )
470 elif self.ACCESS == CommandAccess.USER and session is None:
471 raise XMPPError(
472 "forbidden",
473 "This is only available for users that are registered to this gateway",
474 )
475 elif self.ACCESS == CommandAccess.USER_NON_LOGGED:
476 if session is None or session.logged:
477 raise XMPPError(
478 "forbidden",
479 (
480 "This is only available for users that are not logged to the"
481 " legacy service"
482 ),
483 )
484 elif self.ACCESS == CommandAccess.USER_LOGGED and (
485 session is None or not session.logged
486 ):
487 raise XMPPError(
488 "forbidden",
489 ("This is only available when you are logged in to the legacy service"),
490 )
491 return session
494T = TypeVar("T", bound="LegacyContact | AnyMUC")
497class _RecipientCommand[T: "LegacyContact | AnyMUC"](CommandBase):
498 @staticmethod
499 @abstractmethod
500 async def run(recipient: T, *args: str) -> CommandResponseRecipientType[T]:
501 """
502 Entrypoint for a recipient-specific command.
504 The first argument is a :class:`LegacyContact` or :class:`LegacyMUC`
505 instance. ``*args`` are extra args passed when using the chatbot.
506 """
507 raise NotImplementedError
510class ContactCommand(_RecipientCommand[LegacyContactType]):
511 """
512 A command that will be avaible on a contact.
514 It implicitly requires the user to be registered and logged.
515 It is never instantiated, so all methods must be static methods.
516 Its entrypoint is the ``run()`` static method.
518 For the command to be available to users, it must be listed in
519 :attr:`.BaseGateway.COMMANDS`.
520 """
523class MUCCommand(_RecipientCommand[LegacyMUCType]):
524 """
525 A command that will be avaible on a MUC.
527 It implicitly requires the user to be registered and logged.
528 It is never instantiated, so all methods must be static methods.
529 Its entrypoint is the ``run()`` static method.
531 For the command to be available to users, it must be listed in
532 :attr:`.BaseGateway.COMMANDS`.
533 """
536def is_admin(jid: JidStr) -> bool:
537 return JID(jid).bare in config.ADMINS