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