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