Coverage for slidge/core/gateway.py: 57%
561 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-22 03:37 +0000
1"""
2This module extends slixmpp.ComponentXMPP to make writing new LegacyClients easier
3"""
5from __future__ import annotations
7import abc
8import asyncio
9import logging
10import re
11import sys
12import tempfile
13from collections.abc import (
14 Awaitable,
15 Callable,
16 Coroutine,
17 Iterable,
18 Sequence,
19)
20from copy import copy
21from datetime import UTC, datetime, timedelta
22from pathlib import Path
23from typing import (
24 Any,
25 ClassVar,
26 Concatenate,
27 Generic,
28 ParamSpec,
29 TypeVar,
30)
32import aiohttp
33from slixmpp import JID, ComponentXMPP, Iq
34from slixmpp.exceptions import IqError, IqTimeout, XMPPError
35from slixmpp.plugins.xep_0004.stanza.form import Form
36from slixmpp.plugins.xep_0060.stanza import OwnerAffiliation
37from slixmpp.plugins.xep_0356.privilege import PrivilegedIqError
38from slixmpp.types import MessageTypes
39from slixmpp.xmlstream.xmlstream import NotConnectedError
40from sqlalchemy.orm import Session as OrmSession
42import slidge.command.categories
43from slidge.command import BUILTIN_COMMANDS
44from slidge.command.adhoc import AdhocProvider
45from slidge.command.admin import Exec
46from slidge.command.base import (
47 Command,
48 CommandBase,
49 ContactCommand,
50 FormField,
51 MUCCommand,
52)
53from slidge.command.chat_command import ChatCommandProvider
54from slidge.command.register import AltRegistrationFlow, Register, RegistrationType
55from slidge.contact import LegacyContact
56from slidge.core import config
57from slidge.core.attachment_upload import AttachmentUploader
58from slidge.core.dispatcher.session_dispatcher import SessionDispatcher
59from slidge.core.mixins.avatar import convert_avatar
60from slidge.core.mixins.message import ContentMessageMixin, InviteMixin
61from slidge.core.pubsub import PubSubComponent
62from slidge.db import GatewayUser, SlidgeStore
63from slidge.db.avatar import CachedAvatar, avatar_cache
64from slidge.db.meta import JSONSerializable
65from slidge.db.models import Attachment
66from slidge.group import LegacyMUC
67from slidge.slixfix.delivery_receipt import DeliveryReceipt
68from slidge.slixfix.roster import RosterBackend
69from slidge.util.jid_escaping import escape_node, unescape_node
70from slidge.util.types import (
71 AnySession,
72 Avatar,
73 LocalpartType,
74 MessageOrPresenceTypeVar,
75 RegistrationValidationCoroutine,
76 SessionType_co,
77)
78from slidge.util.util import derive_wired_class
80T = TypeVar("T")
81P = ParamSpec("P")
84class BaseGateway(
85 ComponentXMPP,
86 InviteMixin,
87 ContentMessageMixin,
88 abc.ABC,
89 Generic[SessionType_co], # noqa: UP046 (mypy cannot infer variance with PEP 695 syntax)
90):
91 """
92 The gateway component, handling registrations and un-registrations.
94 On slidge launch, a singleton is instantiated, and it will be made available
95 to public classes such :class:`.LegacyContact` or :class:`.BaseSession` as the
96 ``.xmpp`` attribute.
98 Must be subclassed by a legacy module to set up various aspects of the XMPP
99 component behaviour, such as its display name or welcome message, via
100 class attributes :attr:`.COMPONENT_NAME` :attr:`.WELCOME_MESSAGE`.
102 Abstract methods related to the registration process must be overriden
103 for a functional :term:`Legacy Module`:
105 - :meth:`.validate`
106 - :meth:`.validate_two_factor_code`
107 - :meth:`.get_qr_text`
108 - :meth:`.confirm_qr`
110 NB: Not all of these must be overridden, it depends on the
111 :attr:`REGISTRATION_TYPE`.
113 The other methods, such as :meth:`.send_text` or :meth:`.react` are the same
114 as those of :class:`.LegacyContact` and :class:`.LegacyParticipant`, because
115 the component itself is also a "messaging actor", ie, an :term:`XMPP Entity`.
116 For these methods, you need to specify the JID of the recipient with the
117 `mto` parameter.
119 Since it inherits from :class:`slixmpp.componentxmpp.ComponentXMPP`,you also
120 have a hand on low-level XMPP interactions via slixmpp methods, e.g.:
122 .. code-block:: python
124 self.send_presence(
125 pfrom="somebody@component.example.com",
126 pto="someonwelse@anotherexample.com",
127 )
129 However, you should not need to do so often since the classes of the plugin
130 API provides higher level abstractions around most commonly needed use-cases, such
131 as sending messages, or displaying a custom status.
133 """
135 COMPONENT_NAME: str = NotImplemented
136 """Name of the component, as seen in service discovery by XMPP clients"""
137 COMPONENT_TYPE: str = ""
138 """Type of the gateway, should follow https://xmpp.org/registrar/disco-categories.html"""
139 COMPONENT_AVATAR: Avatar | Path | str | None = None
140 """
141 Path, bytes or URL used by the component as an avatar.
142 """
144 REGISTRATION_FIELDS: ClassVar[Sequence[FormField]] = [
145 FormField(var="username", label="User name", required=True),
146 FormField(var="password", label="Password", required=True, private=True),
147 ]
148 """
149 Iterable of fields presented to the gateway user when registering using :xep:`0077`
150 `extended <https://xmpp.org/extensions/xep-0077.html#extensibility>`_ by :xep:`0004`.
151 """
152 REGISTRATION_INSTRUCTIONS: str = "Enter your credentials"
153 """
154 The text presented to a user who wants to register (or modify) their
155 :term:`legacy <Legacy>` account configuration.
156 """
157 REGISTRATION_TYPE: RegistrationType = RegistrationType.SINGLE_STEP_FORM
158 """
159 This attribute determines how users register to the gateway, ie, how they
160 login to the :term:`legacy network <Legacy Network>`.
161 The credentials are then stored persistently, so this process should happen
162 once per user (unless they unregister).
164 The registration process always start with a basic data form (:xep:`0004`)
165 presented to the user.
166 But the legacy login flow might require something more sophisticated, see
167 :class:`.RegistrationType` for more details.
168 """
170 REGISTRATION_2FA_TITLE = "Enter your 2FA code"
171 REGISTRATION_2FA_INSTRUCTIONS = (
172 "You should have received something via email or SMS, or something"
173 )
174 REGISTRATION_QR_INSTRUCTIONS = "Flash this code or follow this link"
176 ALTERNATIVE_REGISTRATION_FLOWS: ClassVar[list[AltRegistrationFlow]] = []
178 PREFERENCES: ClassVar[list[FormField]] = [
179 FormField(
180 var="sync_presence",
181 label="Propagate your XMPP presence to the legacy network.",
182 value="true",
183 required=True,
184 type="boolean",
185 ),
186 FormField(
187 var="sync_avatar",
188 label="Propagate your XMPP avatar to the legacy network.",
189 value="true",
190 required=True,
191 type="boolean",
192 ),
193 FormField(
194 var="always_invite_when_adding_bookmarks",
195 label="Always send invitations to join MUCs.",
196 value="false`",
197 required=True,
198 type="boolean",
199 ),
200 FormField(
201 var="last_seen_fallback",
202 label="Use contact presence status message to show when they were last seen.",
203 value="true",
204 required=True,
205 type="boolean",
206 ),
207 FormField(
208 var="roster_push",
209 label="Add contacts to your roster.",
210 value="true",
211 required=True,
212 type="boolean",
213 ),
214 FormField(
215 var="reaction_fallback",
216 label="Receive fallback messages for reactions (for legacy XMPP clients)",
217 value="false",
218 required=True,
219 type="boolean",
220 ),
221 ]
223 ROSTER_GROUP: str = "slidge"
224 """
225 Name of the group assigned to a :class:`.LegacyContact` automagically
226 added to the :term:`User`'s roster with :meth:`.LegacyContact.add_to_roster`.
227 """
228 WELCOME_MESSAGE = (
229 "Thank you for registering. Type 'help' to list the available commands, "
230 "or just start messaging away!"
231 )
232 """
233 A welcome message displayed to users on registration.
234 This is useful notably for clients that don't consider component JIDs as a
235 valid recipient in their UI, yet still open a functional chat window on
236 incoming messages from components.
237 """
239 SEARCH_FIELDS: ClassVar[Sequence[FormField]] = [
240 FormField(var="first", label="First name", required=True),
241 FormField(var="last", label="Last name", required=True),
242 FormField(var="phone", label="Phone number", required=False),
243 ]
244 """
245 Fields used for searching items via the component, through :xep:`0055` (jabber search).
246 A common use case is to allow users to search for legacy contacts by something else than
247 their usernames, eg their phone number.
249 Plugins should implement search by overriding :meth:`.BaseSession.search`
250 (restricted to registered users).
252 If there is only one field, it can also be used via the ``jabber:iq:gateway`` protocol
253 described in :xep:`0100`. Limitation: this only works if the search request returns
254 one result item, and if this item has a 'jid' var.
255 """
256 SEARCH_TITLE: str = "Search for legacy contacts"
257 """
258 Title of the search form.
259 """
260 SEARCH_INSTRUCTIONS: str = ""
261 """
262 Instructions of the search form.
263 """
265 MARK_ALL_MESSAGES = False
266 """
267 Set this to True for :term:`legacy networks <Legacy Network>` that expects
268 read marks for *all* messages and not just the latest one that was read
269 (as most XMPP clients will only send a read mark for the latest msg).
270 """
272 PROPER_RECEIPTS = False
273 """
274 Set this to True if the legacy service provides a real equivalent of message delivery receipts
275 (:xep:`0184`), meaning that there is an event thrown when the actual device of a contact receives
276 a message. Make sure to call Contact.received() adequately if this is set to True.
277 """
279 GROUPS = False
280 """
281 This must be set to True if this gateway supports groups.
282 """
283 SPACES = False
284 """
285 This must be set to True if this gateway supports spaces, cf :xep:`0503`.
286 """
287 THREADS: bool = True
288 """
289 This must be set to True if this gateway supports threads, to enable
290 storing an XMPP/legacy thread ID mapping when necessary.
291 """
293 mtype: MessageTypes = "chat"
294 is_group = False
295 _can_send_carbon = False
296 store: SlidgeStore
298 session_cls: type[SessionType_co]
299 """Concrete :class:`.BaseSession` subclass of this legacy module.
301 Derived automatically from the generic parameter, e.g.,
302 ``class Gateway(BaseGateway[Session])``.
303 """
305 def __init_subclass__(cls, **kwargs: object) -> None:
306 super().__init_subclass__(**kwargs)
307 derive_wired_class(cls, BaseGateway, "session_cls")
309 COMMANDS: ClassVar[tuple[type[CommandBase], ...]] = BUILTIN_COMMANDS
310 """Commands (adhoc and chatbot) this gateway provides.
312 Subclasses may override this with additional custom commands.
313 E.g.: ``COMMANDS = BaseGateway.COMMANDS + commands_from_module(command)``.
314 """
316 http: aiohttp.ClientSession
317 avatar: CachedAvatar | None = None
319 def __init__(self) -> None:
320 if getattr(self, "session_cls", None) is None:
321 raise RuntimeError(
322 f"{type(self).__name__}.session_cls is not set. Your gateway"
323 " class must be parameterized with its BaseSession subclass,"
324 " e.g. 'class Gateway(BaseGateway[Session])'."
325 )
326 if config.COMPONENT_NAME:
327 self.COMPONENT_NAME = config.COMPONENT_NAME
328 if config.WELCOME_MESSAGE:
329 self.WELCOME_MESSAGE = config.WELCOME_MESSAGE
330 self.log = log
331 self.datetime_started = datetime.now(tz=UTC)
332 # FIXME: ugly hack to work with the BaseSender mixin :/
333 self.xmpp = self
334 self.default_ns = "jabber:component:accept"
335 super().__init__(
336 config.JID,
337 config.SECRET,
338 config.SERVER,
339 config.PORT,
340 plugin_whitelist=SLIXMPP_PLUGINS,
341 plugin_config={
342 "xep_0077": {
343 "form_fields": None,
344 "form_instructions": self.REGISTRATION_INSTRUCTIONS,
345 "enable_subscription": self.REGISTRATION_TYPE
346 == RegistrationType.SINGLE_STEP_FORM,
347 },
348 "xep_0100": {
349 "component_name": self.COMPONENT_NAME,
350 "type": self.COMPONENT_TYPE,
351 },
352 "xep_0184": {
353 "auto_ack": False,
354 "auto_request": False,
355 },
356 "xep_0363": {
357 "upload_service": config.UPLOAD_SERVICE,
358 },
359 },
360 fix_error_ns=True,
361 )
362 self.loop.set_exception_handler(self.__exception_handler)
363 self.create_task(self.__set_http(), name="set http client")
364 self.has_crashed: bool = False
365 self.use_origin_id = False
367 if config.USER_JID_VALIDATOR is None:
368 config.USER_JID_VALIDATOR = f".*@{self.infer_real_domain()}"
369 log.info(
370 "No USER_JID_VALIDATOR was set, using '%s'.",
371 config.USER_JID_VALIDATOR,
372 )
373 self.jid_validator: re.Pattern[str] = re.compile(config.USER_JID_VALIDATOR)
374 self.qr_pending_registrations = dict[
375 str, asyncio.Future[JSONSerializable | None]
376 ]()
378 self.register_plugins()
379 self.__setup_session_cls()
381 self.get_session_from_stanza = self.session_cls.from_stanza
382 self.get_session_from_user = self.session_cls.from_user
384 self.__register_slixmpp_events()
385 self.__register_slixmpp_api()
386 self.roster.set_backend(RosterBackend(self))
388 self.register_plugin("pubsub", {"component_name": self.COMPONENT_NAME})
389 self.pubsub: PubSubComponent = self.plugin["pubsub"] # type:ignore[typeddict-item]
390 self.delivery_receipt = DeliveryReceipt(self)
392 # with this we receive user avatar updates
393 self.plugin["xep_0030"].add_feature("urn:xmpp:avatar:metadata+notify")
395 self.plugin["xep_0030"].add_feature("urn:xmpp:chat-markers:0")
397 if self.GROUPS:
398 self.plugin["xep_0030"].add_feature("http://jabber.org/protocol/muc")
399 self.plugin["xep_0030"].add_feature(self.plugin["xep_0463"].stanza.NS)
400 self.plugin["xep_0030"].add_feature("urn:xmpp:mam:2")
401 self.plugin["xep_0030"].add_feature("urn:xmpp:mam:2#extended")
402 self.plugin["xep_0030"].add_feature(self.plugin["xep_0421"].namespace)
403 self.plugin["xep_0030"].add_feature(self.plugin["xep_0317"].stanza.NS)
404 self.plugin["xep_0030"].add_identity(
405 category="conference",
406 name=self.COMPONENT_NAME,
407 itype="text",
408 jid=self.boundjid,
409 )
410 if self.SPACES:
411 self.plugin["xep_0030"].add_feature("urn:xmpp:spaces:0")
413 self.__adhoc_handler = AdhocProvider(self)
414 self.__chat_commands_handler = ChatCommandProvider(self)
416 self.__dispatcher = SessionDispatcher(self)
417 self.get_caps_ver = self.__dispatcher.get_caps_ver
419 self.__register_commands()
421 def __setup_session_cls(self) -> None:
422 contact_cls = self.session_cls.roster_cls.contact_cls
424 if contact_cls.REACTIONS_SINGLE_EMOJI:
425 form = Form()
426 form["type"] = "result"
427 form.add_field(
428 "FORM_TYPE", "hidden", value="urn:xmpp:reactions:0:restrictions"
429 )
430 form.add_field("max_reactions_per_user", value="1", type="text-single")
431 form.add_field("scope", value="domain")
432 self.plugin["xep_0128"].add_extended_info(data=form)
434 self.session_cls.xmpp = self
436 @property
437 def _uploader(self) -> AttachmentUploader:
438 # the component itself is not bound to any user session
439 return AttachmentUploader(self, None)
441 async def kill_session(self, jid: JID) -> None:
442 await self.session_cls.kill_by_jid(jid)
444 async def __set_http(self) -> None:
445 self.http = aiohttp.ClientSession()
446 if getattr(self, "_test_mode", False):
447 return
448 avatar_cache.http = self.http
450 def __register_commands(self) -> None:
451 for cls in self.COMMANDS:
452 if any(x is NotImplemented for x in [cls.CHAT_COMMAND, cls.NODE, cls.NAME]):
453 raise TypeError(
454 f"{cls} is listed in {type(self).__name__}.COMMANDS but"
455 " does not set NAME, NODE and CHAT_COMMAND."
456 )
457 if issubclass(cls, ContactCommand):
458 LegacyContact.commands[cls.NODE] = cls
459 LegacyContact.commands_chat[cls.CHAT_COMMAND] = cls
460 continue
461 if issubclass(cls, MUCCommand):
462 LegacyMUC.commands[cls.NODE] = cls
463 LegacyMUC.commands_chat[cls.CHAT_COMMAND] = cls
464 continue
465 if not issubclass(cls, Command):
466 raise TypeError(
467 f"{cls} is listed in {type(self).__name__}.COMMANDS but is"
468 " not a Command, ContactCommand or MUCCommand subclass."
469 )
470 if cls is Exec:
471 if config.DEV_MODE:
472 log.warning(r"/!\ DEV MODE ENABLED /!\\")
473 else:
474 continue
475 if cls.CATEGORY == slidge.command.categories.GROUPS and not self.GROUPS:
476 continue
477 if cls.CATEGORY == slidge.command.categories.SPACES and not self.SPACES:
478 continue
479 c = cls(self)
480 log.debug("Registering %s", cls)
481 self.__adhoc_handler.register(c)
482 self.__chat_commands_handler.register(c)
484 for i, alt in enumerate(self.ALTERNATIVE_REGISTRATION_FLOWS):
486 class AltRegister(Register):
487 NAME = alt.title
488 NODE = f"https://slidge.im/register/alt{i}"
489 CHAT_COMMAND = f"register-{i}"
490 _instructions = alt.instructions
491 _type = alt.type
492 _fields = alt.fields
493 _validate = staticmethod(alt.validate) if alt.validate else None
495 inst = AltRegister(self)
496 log.debug("Registering alternative registration flow: %s", alt.title)
497 self.__adhoc_handler.register(inst)
498 self.__chat_commands_handler.register(inst)
500 def __exception_handler(
501 self, loop: asyncio.AbstractEventLoop, context: dict[Any, Any]
502 ) -> None:
503 """
504 Called when the event loop encounters an exception that was not
505 handled anywhere else.
507 Tasks created via :meth:`create_task`, are handled separately and do not reach
508 this handler.
509 """
510 log.debug("Context in the exception handler: %s", context)
511 exc = context.get("exception")
512 if exc is None:
513 log.debug("No exception in this context: %s", context)
514 elif isinstance(exc, SystemExit):
515 log.debug("SystemExit called in an asyncio task")
516 elif context.get("future") is not None:
517 log.warning(
518 "Exception in a third-party task was never retrieved: %s",
519 context,
520 exc_info=exc,
521 )
522 else:
523 log.error("Crash in an asyncio task: %s", context)
524 log.exception("Crash in task", exc_info=exc)
525 self.has_crashed = True
526 loop.stop()
528 def create_task(
529 self,
530 coro: Coroutine[Any, Any, Any],
531 *,
532 name: str | None = None,
533 ) -> asyncio.Task[Any]:
534 """Create a task that crashes the gateway if it fails."""
535 task = self.loop.create_task(coro, name=name)
536 task.add_done_callback(self.__handle_task_result)
537 return task
539 def __handle_task_result(self, task: asyncio.Task[Any]) -> None:
540 if task.cancelled():
541 return
542 exc = task.exception()
543 if exc is not None:
544 log.error(
545 "Crash in task %s",
546 task.get_name(),
547 exc_info=(type(exc), exc, exc.__traceback__),
548 )
549 self.has_crashed = True
550 self.loop.stop()
552 def __register_slixmpp_events(self) -> None:
553 self.del_event_handler("presence_subscribe", self._handle_subscribe)
554 self.del_event_handler("presence_unsubscribe", self._handle_unsubscribe)
555 self.del_event_handler("presence_subscribed", self._handle_subscribed)
556 self.del_event_handler("presence_unsubscribed", self._handle_unsubscribed)
557 self.del_event_handler(
558 "roster_subscription_request", self._handle_new_subscription
559 )
560 self.del_event_handler("presence_probe", self._handle_probe)
561 self.add_event_handler("session_start", self.__on_session_start)
563 def __register_slixmpp_api(self) -> None:
564 def with_session(
565 func: Callable[Concatenate[OrmSession, P], T], commit: bool = True
566 ) -> Callable[P, T]:
567 def wrapped(*a: P.args, **kw: P.kwargs) -> T:
568 with self.store.session() as orm:
569 res = func(orm, *a, **kw)
570 if commit:
571 orm.commit()
572 return res
574 return wrapped
576 self.plugin["xep_0231"].api.register(
577 with_session(self.store.bob.get_bob, False), "get_bob"
578 )
579 self.plugin["xep_0231"].api.register(
580 with_session(self.store.bob.set_bob), "set_bob"
581 )
582 self.plugin["xep_0231"].api.register(
583 with_session(self.store.bob.del_bob), "del_bob"
584 )
586 @property # type:ignore[override]
587 def jid(self) -> JID:
588 # Override to avoid slixmpp deprecation warnings.
589 return self.boundjid
591 @jid.setter
592 def jid(self, jid: JID) -> None:
593 raise RuntimeError
595 async def __on_session_start(self, event: object) -> None:
596 log.debug("Gateway session start: %s", event)
598 await self.__setup_attachments()
600 # prevents XMPP clients from considering the gateway as an HTTP upload
601 disco = self.plugin["xep_0030"]
602 await disco.del_feature(feature="urn:xmpp:http:upload:0", jid=self.boundjid)
603 await self.plugin["xep_0115"].update_caps(jid=self.boundjid)
605 if self.COMPONENT_AVATAR is not None:
606 log.debug("Setting gateway avatar…")
607 avatar = convert_avatar(self.COMPONENT_AVATAR, "!!---slidge---special---")
608 assert avatar is not None
609 try:
610 cached_avatar = await avatar_cache.get(avatar)
611 except Exception as e:
612 log.exception("Could not set the component avatar.", exc_info=e)
613 cached_avatar = None
614 else:
615 assert cached_avatar is not None
616 self.avatar = cached_avatar
617 else:
618 cached_avatar = None
620 with self.store.session() as orm:
621 users = orm.query(GatewayUser).all()
622 for user in users:
623 # TODO: before this, we should check if the user has removed us from their roster
624 # while we were offline and trigger unregister from there. Presence probe does not seem
625 # to work in this case, there must be another way. privileged entity could be used
626 # as last resort.
627 try:
628 await self["xep_0100"].add_component_to_roster(user.jid)
629 await self.__add_component_to_mds_whitelist(user.jid)
630 except (IqError, IqTimeout) as e:
631 # TODO: remove the user when this happens? or at least
632 # this can happen when the user has unsubscribed from the XMPP server
633 log.warning(
634 "Error with user %s, not logging them automatically",
635 user,
636 exc_info=e,
637 )
638 continue
639 session = self.session_cls.from_user(user)
640 session.create_task(self.login_wrap(session), name=f"login wrap of {user}")
641 if cached_avatar is not None:
642 await self.pubsub.broadcast_avatar(
643 self.boundjid.bare, session.user_jid, cached_avatar
644 )
646 log.info("Slidge has successfully started")
648 async def __setup_attachments(self) -> None:
649 expire_coro = self.__expire_attachments_upload
650 if config.NO_UPLOAD_PATH:
651 if config.NO_UPLOAD_URL_PREFIX is None:
652 raise RuntimeError(
653 "If you set NO_UPLOAD_PATH you must set NO_UPLOAD_URL_PREFIX too."
654 )
655 expire_coro = self.__expire_attachments_no_upload
656 elif not config.UPLOAD_SERVICE:
657 try:
658 info_iq = await self.xmpp.plugin["xep_0363"].find_upload_service(
659 self.infer_real_domain()
660 )
661 except XMPPError:
662 info_iq = None
663 log.exception(
664 "The upload service could not be automatically determine. "
665 "Attachments to XMPP will not work. "
666 "Either specify 'upload-service' or 'no-upload-path' to fix that."
667 )
668 if info_iq is None:
669 if self.REGISTRATION_TYPE == RegistrationType.QRCODE:
670 log.warning(
671 "No method was configured for attachment and slidge "
672 "could not automatically determine the JID of a usable upload service. "
673 "Users likely won't be able to register since this network uses a "
674 "QR-code based registration flow."
675 )
676 if not config.USE_ATTACHMENT_ORIGINAL_URLS:
677 log.warning(
678 "Setting USE_ATTACHMENT_ORIGINAL_URLS to True since no method was configured "
679 "for attachments and no upload service was found. NB: this does not work for all "
680 "networks, especially for the E2EE attachments."
681 )
682 config.USE_ATTACHMENT_ORIGINAL_URLS = True
683 else:
684 log.info("Auto-discovered upload service: %s", info_iq["from"])
685 config.UPLOAD_SERVICE = info_iq["from"]
687 self.__expire_attachments_task = self.create_task(_loop(expire_coro, 3600 * 24))
689 async def __expire_attachments_no_upload(self) -> None:
690 with self.store.session() as orm:
691 attachments = self.store.attachments.get_all(orm)
692 self.__expire_attachments_in_store(
693 await asyncio.to_thread(self.__rm_local_attachments, attachments)
694 )
696 def __rm_local_attachments(self, attachments: list[Attachment]) -> list[int]:
697 to_remove = []
698 cutoff = datetime.now(tz=UTC) - timedelta(
699 days=config.NO_UPLOAD_MAX_DAYS or config.MAM_MAX_DAYS
700 )
702 for attachment in attachments:
703 path = attachment.local_path
704 try:
705 created = datetime.fromtimestamp(path.stat().st_ctime, tz=UTC)
706 except OSError:
707 self.log.debug(
708 "mtime of %s could not be determined, clearing up the row",
709 attachment,
710 )
711 to_remove.append(attachment.id)
712 else:
713 if created > cutoff:
714 continue
715 to_remove.append(attachment.id)
716 self.log.debug("%s is too old", attachment)
717 _safe_rm_parent(path)
718 to_remove.append(attachment.id)
720 return to_remove
722 async def __expire_attachments_upload(self) -> None:
723 with self.store.session() as orm:
724 attachments = self.store.attachments.get_all(orm)
725 to_remove = []
726 for attachment in attachments:
727 async with self.http.head(attachment.url) as resp:
728 if not resp.ok:
729 self.log.debug("%s is not reachable anymore", attachment)
730 to_remove.append(attachment.id)
731 self.__expire_attachments_in_store(to_remove)
733 def __expire_attachments_in_store(self, to_remove: list[int]) -> None:
734 with self.store.session() as orm:
735 self.store.attachments.remove(orm, to_remove)
736 orm.commit()
738 def infer_real_domain(self) -> JID:
739 return JID(re.sub(r"^.*?\.", "", self.xmpp.boundjid.bare))
741 async def __add_component_to_mds_whitelist(self, user_jid: JID) -> None:
742 # Uses privileged entity to add ourselves to the whitelist of the PEP
743 # MDS node so we receive MDS events
744 iq_creation = Iq(sto=user_jid.bare, sfrom=user_jid, stype="set")
745 iq_creation["pubsub"]["create"]["node"] = self.plugin["xep_0490"].stanza.NS
747 try:
748 await self.plugin["xep_0356"].send_privileged_iq(iq_creation)
749 except PermissionError:
750 log.warning(
751 "IQ privileges not granted for pubsub namespace, we cannot "
752 "create the MDS node of %s",
753 user_jid,
754 )
755 except PrivilegedIqError as exc:
756 nested = exc.nested_error()
757 # conflict this means the node already exists, we can ignore that
758 if nested is not None and nested.condition != "conflict":
759 log.exception(
760 "Could not create the MDS node of %s", user_jid, exc_info=exc
761 )
762 except Exception as e:
763 log.exception(
764 "Error while trying to create to the MDS node of %s",
765 user_jid,
766 exc_info=e,
767 )
769 iq_affiliation = Iq(sto=user_jid.bare, sfrom=user_jid, stype="set")
770 iq_affiliation["pubsub_owner"]["affiliations"]["node"] = self.plugin[
771 "xep_0490"
772 ].stanza.NS
774 aff = OwnerAffiliation()
775 aff["jid"] = self.boundjid.bare
776 aff["affiliation"] = "member"
777 iq_affiliation["pubsub_owner"]["affiliations"].append(aff)
779 try:
780 await self.plugin["xep_0356"].send_privileged_iq(iq_affiliation)
781 except PermissionError:
782 log.warning(
783 "IQ privileges not granted for pubsub#owner namespace, we cannot "
784 "listen to the MDS events of %s",
785 user_jid,
786 )
787 except Exception as e:
788 log.exception(
789 "Error while trying to subscribe to the MDS node of %s",
790 user_jid,
791 exc_info=e,
792 )
794 async def login_wrap(self, session: AnySession) -> str:
795 session.send_gateway_status("Logging in…", show="dnd")
796 session.is_logging_in = True
797 try:
798 status = await session.login()
799 except Exception as e:
800 log.warning("Login problem for %s", session.user_jid, exc_info=e)
801 session.send_gateway_status(f"Could not login: {e}", show="dnd")
802 msg = (
803 "You are not connected to this gateway! "
804 f"Maybe this message will tell you why: {e}"
805 )
806 session.send_gateway_message(msg)
807 session.logged = False
808 session.send_gateway_status("Login failed", show="dnd")
809 return msg
811 log.info("Login success for %s", session.user_jid)
812 session.logged = True
813 session.send_gateway_status("Syncing contacts…", show="dnd")
814 with self.store.session() as orm:
815 await session.contacts._fill(orm)
816 if not (r := session.contacts.ready).done():
817 r.set_result(True)
818 if self.GROUPS:
819 session.send_gateway_status("Syncing groups…", show="dnd")
820 await session.bookmarks.fill()
821 if not (r := session.bookmarks.ready).done():
822 r.set_result(True)
823 self.send_presence(pto=session.user.jid.bare, ptype="probe")
824 if status is None:
825 status = "Logged in"
826 session.send_gateway_status(status, show="chat")
828 if session.user.preferences.get("sync_avatar", False):
829 session.create_task(
830 self.fetch_user_avatar(session), name=f"fetch avatar of {session.user}"
831 )
832 else:
833 with self.store.session(expire_on_commit=False) as orm:
834 session.user.avatar_hash = None
835 orm.add(session.user)
836 orm.commit()
837 return status
839 async def fetch_user_avatar(self, session: AnySession) -> None:
840 try:
841 iq = await self.xmpp.plugin["xep_0060"].get_items(
842 session.user_jid.bare,
843 self.xmpp.plugin["xep_0084"].stanza.MetaData.namespace,
844 ifrom=self.boundjid.bare,
845 )
846 except IqTimeout:
847 self.log.warning("Iq timeout trying to fetch user avatar")
848 return
849 except IqError as e:
850 self.log.debug("Iq error when trying to fetch user avatar: %s", e)
851 if e.condition == "item-not-found":
852 try:
853 await session.on_avatar(None, None, None, None, None)
854 except NotImplementedError:
855 pass
856 else:
857 with self.store.session(expire_on_commit=False) as orm:
858 session.user.avatar_hash = None
859 orm.add(session.user)
860 orm.commit()
861 return
862 await self.__dispatcher.on_avatar_metadata_info(
863 session, iq["pubsub"]["items"]["item"]["avatar_metadata"]["info"]
864 )
866 def _send(
867 self,
868 stanza: MessageOrPresenceTypeVar,
869 **send_kwargs: Any, # noqa:ANN401
870 ) -> MessageOrPresenceTypeVar:
871 stanza.set_from(self.boundjid.bare)
872 if mto := send_kwargs.get("mto"):
873 stanza.set_to(mto)
874 stanza.send()
875 return stanza
877 def raise_if_not_allowed_jid(self, jid: JID) -> None:
878 if not self.jid_validator.match(jid.bare):
879 raise XMPPError(
880 condition="not-allowed",
881 text="Your account is not allowed to use this gateway. "
882 "The admin controls that with the USER_JID_VALIDATOR option.",
883 )
885 def send_raw(self, data: str | bytes) -> None:
886 # overridden from XMLStream to strip base64-encoded data from the logs
887 # to make them more readable.
888 if log.isEnabledFor(level=logging.DEBUG):
889 stripped = copy(data) if isinstance(data, str) else data.decode("utf-8")
890 # there is probably a way to do that in a single RE,
891 # but since it's only for debugging, the perf penalty
892 # does not matter much
893 for el in LOG_STRIP_ELEMENTS:
894 stripped = re.sub(
895 f"(<{el}.*?>)(.*)(</{el}>)",
896 "\1[STRIPPED]\3",
897 stripped,
898 flags=re.DOTALL | re.IGNORECASE,
899 )
900 log.debug("SEND: %s", stripped)
901 if not self.transport:
902 raise NotConnectedError()
903 if isinstance(data, str):
904 data = data.encode("utf-8")
905 self.transport.write(data)
907 def get_session_from_jid(self, j: JID) -> AnySession | None:
908 try:
909 return self.session_cls.from_jid(j)
910 except XMPPError:
911 return None
913 def exception(self, exception: Exception) -> None:
914 # """
915 # Called when a task created by slixmpp's internal (eg, on slix events) raises an Exception.
916 #
917 # Stop the event loop and exit on unhandled exception.
918 #
919 # The default :class:`slixmpp.basexmpp.BaseXMPP` behaviour is just to
920 # log the exception, but we want to avoid undefined behaviour.
921 #
922 # :param exception: An unhandled :class:`Exception` object.
923 # """
924 if isinstance(exception, IqError):
925 iq = exception.iq
926 log.error("%s: %s", iq["error"]["condition"], iq["error"]["text"])
927 log.warning("You should catch IqError exceptions")
928 elif isinstance(exception, IqTimeout):
929 iq = exception.iq
930 log.error("Request timed out: %s", iq)
931 log.warning("You should catch IqTimeout exceptions")
932 elif isinstance(exception, SyntaxError):
933 # Hide stream parsing errors that occur when the
934 # stream is disconnected (they've been handled, we
935 # don't need to make a mess in the logs).
936 pass
937 else:
938 if exception:
939 log.exception(exception)
940 self.loop.stop()
941 sys.exit(1)
943 async def make_registration_form(
944 self, _jid: JID, _node: str, _ifrom: JID, iq: Iq
945 ) -> Iq:
946 self.raise_if_not_allowed_jid(iq.get_from())
947 reg = iq["register"]
948 with self.store.session() as orm:
949 user = (
950 orm.query(GatewayUser).filter_by(jid=iq.get_from().bare).one_or_none()
951 )
952 log.debug("User found: %s", user)
954 form = reg["form"]
955 form.add_field(
956 "FORM_TYPE",
957 ftype="hidden",
958 value="jabber:iq:register",
959 )
960 form["title"] = f"Registration to '{self.COMPONENT_NAME}'"
961 form["instructions"] = self.REGISTRATION_INSTRUCTIONS
963 if user is not None:
964 reg["registered"] = False
965 form.add_field(
966 "remove",
967 label="Remove my registration",
968 required=True,
969 ftype="boolean",
970 value=False,
971 )
973 for field in self.REGISTRATION_FIELDS:
974 if field.var in reg.interfaces:
975 val = None if user is None else user.get(field.var)
976 if val is None:
977 reg.add_field(field.var)
978 else:
979 reg[field.var] = val
981 reg["instructions"] = self.REGISTRATION_INSTRUCTIONS
983 for field in self.REGISTRATION_FIELDS:
984 form.add_field(
985 field.var,
986 label=field.label,
987 required=field.required,
988 ftype=field.type,
989 options=field.options,
990 value=field.value if user is None else user.get(field.var, field.value),
991 )
993 reply = iq.reply()
994 reply.set_payload(reg)
995 return reply # type:ignore[no-any-return]
997 async def user_prevalidate(
998 self,
999 ifrom: JID,
1000 form_dict: JSONSerializable,
1001 fields: Iterable[FormField] | None = None,
1002 validate: RegistrationValidationCoroutine | None = None,
1003 ) -> JSONSerializable | None:
1004 # Pre validate a registration form using the content of self.REGISTRATION_FIELDS
1005 # before passing it to the plugin custom validation logic
1006 if fields is None:
1007 fields = self.REGISTRATION_FIELDS
1008 for field in fields:
1009 if field.required and not form_dict.get(field.var):
1010 raise ValueError(f"Missing field: '{field.label}'")
1011 if validate:
1012 return await validate(ifrom, form_dict)
1013 else:
1014 return await self.validate(ifrom, form_dict)
1016 @abc.abstractmethod
1017 async def validate(
1018 self, user_jid: JID, registration_form: JSONSerializable
1019 ) -> JSONSerializable | None:
1020 """
1021 Validate a user's initial registration form.
1023 Should raise the appropriate :class:`slixmpp.exceptions.XMPPError`
1024 if the registration does not allow to continue the registration process.
1026 If :py:attr:`REGISTRATION_TYPE` is a
1027 :attr:`.RegistrationType.SINGLE_STEP_FORM`,
1028 this method should raise something if it wasn't possible to successfully
1029 log in to the legacy service with the registration form content.
1031 It is also used for other types of :py:attr:`REGISTRATION_TYPE` too, since
1032 the first step is always a form. If :attr:`.REGISTRATION_FIELDS` is an
1033 empty list (ie, it declares no :class:`.FormField`), the "form" is
1034 effectively a confirmation dialog displaying
1035 :attr:`.REGISTRATION_INSTRUCTIONS`.
1037 :param user_jid: JID of the user that has just registered
1038 :param registration_form: A dict where keys are the :attr:`.FormField.var` attributes
1039 of the :attr:`.BaseGateway.REGISTRATION_FIELDS` iterable.
1040 This dict can be modified and will be accessible as the ``legacy_module_data``
1041 of the
1043 :return : A dict that will be stored as the persistent "legacy_module_data"
1044 for this user. If you don't return anything here, the whole registration_form
1045 content will be stored.
1046 """
1047 raise NotImplementedError
1049 async def validate_two_factor_code(
1050 self, user: GatewayUser, code: str
1051 ) -> JSONSerializable | None:
1052 """
1053 Called when the user enters their 2FA code.
1055 Should raise the appropriate :class:`slixmpp.exceptions.XMPPError`
1056 if the login fails, and return successfully otherwise.
1058 Only used when :attr:`REGISTRATION_TYPE` is
1059 :attr:`.RegistrationType.TWO_FACTOR_CODE`.
1061 :param user: The :class:`.GatewayUser` whose registration is pending
1062 Use their :attr:`.GatewayUser.bare_jid` and/or
1063 :attr:`.registration_form` attributes to get what you need.
1064 :param code: The code they entered, either via "chatbot" message or
1065 adhoc command
1067 :return : A dict which keys and values will be added to the persistent "legacy_module_data"
1068 for this user.
1069 """
1070 raise NotImplementedError
1072 async def get_qr_text(self, user: GatewayUser) -> str:
1073 """
1074 This is where slidge gets the QR code content for the QR-based
1075 registration process. It will turn it into a QR code image and send it
1076 to the not-yet-fully-registered :class:`.GatewayUser`.
1078 Only used in when :attr:`BaseGateway.REGISTRATION_TYPE` is
1079 :attr:`.RegistrationType.QRCODE`.
1081 :param user: The :class:`.GatewayUser` whose registration is pending
1082 Use their :attr:`.GatewayUser.bare_jid` and/or
1083 :attr:`.registration_form` attributes to get what you need.
1084 """
1085 raise NotImplementedError
1087 async def confirm_qr(
1088 self,
1089 user_bare_jid: str,
1090 exception: Exception | None = None,
1091 legacy_data: JSONSerializable | None = None,
1092 ) -> None:
1093 """
1094 This method is meant to be called to finalize QR code-based registration
1095 flows, once the legacy service confirms the QR flashing.
1097 Only used in when :attr:`BaseGateway.REGISTRATION_TYPE` is
1098 :attr:`.RegistrationType.QRCODE`.
1100 :param user_bare_jid: The bare JID of the almost-registered
1101 :class:`GatewayUser` instance
1102 :param exception: Optionally, an XMPPError to be raised to **not** confirm
1103 QR code flashing.
1104 :param legacy_data: dict which keys and values will be added to the persistent
1105 "legacy_module_data" for this user.
1106 """
1107 fut = self.qr_pending_registrations[user_bare_jid]
1108 if exception is None:
1109 fut.set_result(legacy_data)
1110 else:
1111 fut.set_exception(exception)
1113 async def unregister_user(
1114 self, user: GatewayUser, msg: str = "You unregistered from this gateway."
1115 ) -> None:
1116 self.send_presence(pshow="dnd", pstatus=msg, pto=user.jid)
1117 await self.xmpp.plugin["xep_0077"].api["user_remove"](None, None, user.jid) # type:ignore[call-arg]
1118 await self.xmpp.session_cls.kill_by_jid(user.jid)
1120 async def input(
1121 self,
1122 jid: JID,
1123 text: str | None = None,
1124 mtype: MessageTypes = "chat",
1125 **input_kwargs: Any, # noqa:ANN401
1126 ) -> str:
1127 """
1128 Request arbitrary user input using a simple chat message, and await the result.
1130 You shouldn't need to call this directly bust instead use
1131 :meth:`.BaseSession.input` to directly target a user.
1133 :param jid: The JID we want input from
1134 :param text: A prompt to display for the user
1135 :param mtype: Message type
1136 :return: The user's reply
1137 """
1138 return await self.__chat_commands_handler.input(
1139 jid, text, mtype=mtype, **input_kwargs
1140 )
1142 async def send_qr(
1143 self,
1144 text: str,
1145 **msg_kwargs: Any, # noqa:ANN401
1146 ) -> str | None:
1147 """
1148 Sends a QR Code to a JID
1150 You shouldn't need to call directly bust instead use
1151 :meth:`.BaseSession.send_qr` to directly target a user.
1153 :param text: The text that will be converted to a QR Code
1154 :param msg_kwargs: Optional additional arguments to pass to
1155 :meth:`.BaseGateway.send_file`, such as the recipient of the QR,
1156 code
1157 """
1158 try:
1159 import qrcode
1160 except ImportError:
1161 log.error("Slidge needs the [qr] extra to be able to generate QR codes")
1162 raise
1163 qr = qrcode.make(text)
1164 with tempfile.NamedTemporaryFile(suffix=".png") as f:
1165 qr.save(f.name)
1166 url, _msgs = await self.send_file(Path(f.name), **msg_kwargs)
1167 return url
1169 def shutdown(self) -> list[asyncio.Task[None]]:
1170 # """
1171 # Called by the slidge entrypoint on normal exit.
1172 #
1173 # Sends offline presences from all contacts of all user sessions and from
1174 # the gateway component itself.
1175 # No need to call this manually, :func:`slidge.__main__.main` should take care of it.
1176 # """
1177 log.debug("Shutting down")
1178 tasks = []
1179 with self.store.session() as orm:
1180 for user in orm.query(GatewayUser).all():
1181 tasks.append(self.session_cls.from_jid(user.jid).shutdown())
1182 self.send_presence(ptype="unavailable", pto=user.jid)
1183 return tasks
1185 def localpart_to_legacy_id(self, localpart: str) -> tuple[LocalpartType, str]:
1186 """
1187 Converts a JID local part to a legacy ID, specifying whether this is a
1188 group or a contact.
1190 The default implementation mandates the "group-" and "contact-"
1191 prefixes and applies JID (un)escaping (:xep:`0106`).
1193 Should be overridden if you want to customize how legacy IDs map
1194 to JIDs. This must be overridden if the default conversion is not
1195 enough, e.g., if legacy IDs are case-sensitive.
1197 No network calls are expected, but if a legacy ID is not of the right
1198 type (e.g., it contains letters when it should only be numbers), this
1199 should raise an ``XMPPError``.
1201 Implementations must ensure consistency with
1202 :meth:`.legacy_id_to_localpart`.
1204 :param localpart: The local part of a JID (aka "user part" or
1205 sometimes "node part").
1206 :return: (legacy entity type, legacy ID).
1207 """
1208 if localpart.startswith("group-"):
1209 return "group", unescape_node(localpart.removeprefix("group-"))
1210 elif localpart.startswith("contact-"):
1211 return "contact", unescape_node(localpart.removeprefix("contact-"))
1212 else:
1213 raise XMPPError("bad-request")
1215 def legacy_id_to_localpart(self, type_: LocalpartType, legacy_id: str) -> str:
1216 """
1217 Converts a legacy ID into a JID local part.
1219 The default implementation adds the "group-" and "contact-"
1220 prefixes and applies JID escaping (:xep:`0106`).
1222 Should be overridden if you want to customize how legacy IDs map to
1223 JIDs. This must be overridden if the default conversion is not enough,
1224 e.g., if legacy IDs are case-sensitive.
1226 Implementations must ensure consistency with
1227 :meth:`.localpart_to_legacy_id`.
1229 :param type_: Whether this is a "contact" or "group" legacy ID.
1230 :param legacy_id: The legacy ID.
1232 :return: A valid JID local part.
1233 """
1234 match type_:
1235 case "group":
1236 return f"group-{escape_node(legacy_id)}"
1237 case "contact":
1238 return f"contact-{escape_node(legacy_id)}"
1239 case _:
1240 raise XMPPError("bad-request")
1243SLIXMPP_PLUGINS = [
1244 "xep_0030", # Service discovery
1245 "xep_0045", # Multi-User Chat
1246 "xep_0050", # Adhoc commands
1247 "xep_0054", # VCard-temp (for MUC avatars)
1248 "xep_0055", # Jabber search
1249 "xep_0059", # Result Set Management
1250 "xep_0066", # Out of Band Data
1251 "xep_0071", # XHTML-IM (for stickers and custom emojis maybe later)
1252 "xep_0077", # In-band registration
1253 "xep_0084", # User Avatar
1254 "xep_0085", # Chat state notifications
1255 "xep_0100", # Gateway interaction
1256 "xep_0106", # JID Escaping
1257 "xep_0115", # Entity capabilities
1258 "xep_0122", # Data Forms Validation
1259 "xep_0128", # Service Discovery Extensions
1260 "xep_0153", # vCard-Based Avatars (for MUC avatars)
1261 "xep_0172", # User nickname
1262 "xep_0184", # Message Delivery Receipts
1263 "xep_0199", # XMPP Ping
1264 "xep_0221", # Data Forms Media Element
1265 "xep_0231", # Bits of Binary (for stickers and custom emojis maybe later)
1266 "xep_0249", # Direct MUC Invitations
1267 "xep_0264", # Jingle Content Thumbnails
1268 "xep_0280", # Carbons
1269 "xep_0292_provider", # VCard4
1270 "xep_0308", # Last message correction
1271 "xep_0313", # Message Archive Management
1272 "xep_0317", # Hats
1273 "xep_0319", # Last User Interaction in Presence
1274 "xep_0333", # Chat markers
1275 "xep_0334", # Message Processing Hints
1276 "xep_0356", # Privileged Entity
1277 "xep_0363", # HTTP file upload
1278 "xep_0385", # Stateless in-line media sharing
1279 "xep_0402", # PEP Native Bookmarks
1280 "xep_0421", # Anonymous unique occupant identifiers for MUCs
1281 "xep_0424", # Message retraction
1282 "xep_0425", # Message moderation
1283 "xep_0444", # Message reactions
1284 "xep_0447", # Stateless File Sharing
1285 "xep_0449", # Stickers
1286 "xep_0461", # Message replies
1287 "xep_0462", # Pubsub Type Filtering
1288 "xep_0463", # MUC Affiliation Versioning
1289 "xep_0469", # Bookmark Pinning
1290 "xep_0490", # Message Displayed Synchronization
1291 "xep_0492", # Chat Notification Settings
1292 # "xep_0503", # Server-side spaces
1293 "xep_0511", # Link Metadata
1294]
1297async def _loop(coro: Callable[[], Awaitable[None]], sleep: int) -> None:
1298 while True:
1299 await coro()
1300 await asyncio.sleep(sleep)
1303def _safe_rm_parent(path: Path) -> None:
1304 try:
1305 path.unlink()
1306 path.parent.rmdir()
1307 except OSError as exc:
1308 log.warning("%s couldn't be cleanly removed: !r", exc)
1311LOG_STRIP_ELEMENTS = ["data", "binval"]
1313log = logging.getLogger(__name__)