Coverage for slidge/core/session.py: 83%
240 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-10 04:45 +0000
1import abc
2import asyncio
3import contextlib
4import logging
5import warnings
6from asyncio.tasks import Task
7from collections.abc import Coroutine
8from typing import Any, Final, Generic, NamedTuple, Self
10import aiohttp
11import sqlalchemy as sa
12from slixmpp import JID, Iq, Message, Presence
13from slixmpp.exceptions import XMPPError
14from slixmpp.types import PresenceShows, ResourceDict
16from slidge.db.meta import JSONSerializable
18from ..command import SearchResult
19from ..contact import LegacyContact, LegacyRoster
20from ..db.models import Contact, GatewayUser
21from ..group import LegacyBookmarks
22from ..util.lock import NamedLockMixin
23from ..util.types import (
24 AnyGateway,
25 AnyMUC,
26 AnyParticipant,
27 AnySession,
28 LegacyBookmarksType_co,
29 LegacyRosterType_co,
30 PseudoPresenceShow,
31)
32from ..util.util import derive_wired_class, noop_coro
35class CachedPresence(NamedTuple):
36 status: str | None
37 show: str | None
38 kwargs: dict[str, Any]
41class BaseSession(
42 NamedLockMixin,
43 abc.ABC,
44 Generic[LegacyRosterType_co, LegacyBookmarksType_co], # noqa: UP046 (mypy cannot infer variance with PEP 695 syntax)
45):
46 """
47 The session of a registered :term:`User`.
49 Represents a gateway user logged in to the legacy network and performing actions.
51 Will be instantiated automatically on slidge startup for each registered user,
52 or upon registration for new (validated) users.
54 Must be subclassed for a functional :term:`Legacy Module`.
55 """
57 """
58 Since we cannot set the XMPP ID of messages sent by XMPP clients, we need to keep a mapping
59 between XMPP IDs and legacy message IDs if we want to further refer to a message that was sent
60 by the user. This also applies to 'carboned' messages, ie, messages sent by the user from
61 the official client of a legacy network.
62 """
64 xmpp: AnyGateway
65 """
66 The gateway instance singleton. Use it for low-level XMPP calls or custom methods that are not
67 session-specific.
69 It is set on the session class by the gateway on startup, before any
70 session is instantiated. Plugins may redeclare it with their own gateway
71 class for typed access, e.g. ``xmpp: "Gateway"``.
72 """
74 MESSAGE_IDS_ARE_THREAD_IDS = False
75 """
76 Set this to True if the legacy service uses message IDs as thread IDs,
77 eg Mattermost, where you can only 'create a thread' by replying to the message,
78 in which case the message ID is also a thread ID (and all messages are potential
79 threads).
80 """
81 SPECIAL_MSG_ID_PREFIX: str | None = None
82 """
83 If you set this, XMPP message IDs starting with this won't be converted to legacy ID,
84 but passed as is to :meth:`LegacyContact.on_react`, and usual checks for emoji restriction won't be
85 applied.
86 This can be used to implement voting in polls in a hacky way.
87 """
89 roster_cls: type[LegacyRosterType_co]
90 """
91 The :class:`.LegacyRoster` subclass to use for this session's contacts.
93 Derived automatically from the first generic parameter, e.g.,
94 ``class Session(BaseSession[Roster, Bookmarks])``, which also types
95 :attr:`.contacts`.
96 """
97 bookmarks_cls: type[LegacyBookmarksType_co]
98 """
99 The :class:`.LegacyBookmarks` subclass to use for this session's groups.
101 Derived automatically from the second generic parameter, e.g.,
102 ``class Session(BaseSession[Roster, Bookmarks])``, which also types
103 :attr:`.bookmarks`.
104 """
106 def __init_subclass__(cls, **kwargs: object) -> None:
107 super().__init_subclass__(**kwargs)
108 derive_wired_class(cls, BaseSession, "roster_cls", "bookmarks_cls")
110 def __init__(self, user: GatewayUser) -> None:
111 super().__init__()
112 self.user = user
113 """
114 The :term:`slidge user <User>`.
115 """
116 self.log = logging.getLogger(user.jid.bare)
118 self.ignore_messages = set[str]()
120 self.contacts: Final[LegacyRosterType_co] = self.roster_cls(self)
121 """This session's roster, an instance of :attr:`.roster_cls`."""
123 self.is_logging_in = False
124 self._logged = False
125 self.__reset_ready()
127 self.bookmarks: Final[LegacyBookmarksType_co] = self.bookmarks_cls(self)
128 """This session's groups, an instance of :attr:`.bookmarks_cls`."""
130 self.thread_creation_lock = asyncio.Lock()
132 self.__cached_presence: CachedPresence | None = None
134 self.__tasks = set[asyncio.Task[Any]]()
136 @property
137 def user_jid(self) -> JID:
138 return self.user.jid
140 @property
141 def user_pk(self) -> int:
142 return self.user.id
144 @property
145 def http(self) -> aiohttp.ClientSession:
146 return self.xmpp.http
148 def __remove_task(self, fut: Task[Any]) -> None:
149 self.log.debug("Removing fut %s", fut)
150 self.__tasks.remove(fut)
152 def create_task(
153 self, coro: Coroutine[Any, Any, Any], name: str | None = None
154 ) -> asyncio.Task[Any]:
155 if name is None:
156 warnings.warn(
157 "Calling Session.create_task without a 'name' argument will "
158 "be deprecated in slidge >= 0.6",
159 DeprecationWarning,
160 )
161 task = self.xmpp.loop.create_task(coro, name=name)
162 self.__tasks.add(task)
163 self.log.debug("Creating task %s", task)
164 task.add_done_callback(lambda _: self.__remove_task(task))
165 return task
167 def cancel_all_tasks(self) -> None:
168 for task in self.__tasks:
169 task.cancel()
171 @abc.abstractmethod
172 async def login(self) -> str | None:
173 """
174 Logs in the gateway user to the legacy network.
176 Triggered when the gateway start and on user registration.
177 It is recommended that this function returns once the user is logged in,
178 so if you need to await forever (for instance to listen to incoming events),
179 it's a good idea to wrap your listener in an asyncio.Task.
181 :return: Optionally, a text to use as the gateway status, e.g., "Connected as 'dude@legacy.network'"
182 """
183 raise NotImplementedError
185 async def logout(self) -> None:
186 """
187 Logs out the gateway user from the legacy network.
189 Called on gateway shutdown.
190 """
191 raise NotImplementedError
193 async def on_unregister(self) -> None:
194 """Called when the user unregisters from the gateway, after their
195 session has been terminated but before their persistent data is
196 deleted.
198 Optionally override this if you need to clean up additional stuff; by
199 default it just calls :meth:`.logout`.
200 """
201 with contextlib.suppress(NotImplementedError):
202 await self.logout()
204 async def on_presence(
205 self,
206 resource: str,
207 show: PseudoPresenceShow,
208 status: str,
209 resources: dict[str, ResourceDict],
210 merged_resource: ResourceDict | None,
211 ) -> None:
212 """
213 Called when the gateway component receives a presence, ie, when
214 one of the user's clients goes online of offline, or changes its
215 status.
217 :param resource: The XMPP client identifier, arbitrary string.
218 :param show: The presence ``<show>``, if available. If the resource is
219 just 'available' without any ``<show>`` element, this is an empty
220 str.
221 :param status: A status message, like a deeply profound quote, eg,
222 "Roses are red, violets are blue, [INSERT JOKE]".
223 :param resources: A summary of all the resources for this user.
224 :param merged_resource: A global presence for the user account,
225 following rules described in :meth:`merge_resources`
226 """
227 raise NotImplementedError
229 async def on_search(self, form_values: dict[str, str]) -> SearchResult | None:
230 """
231 Triggered when the user uses Jabber Search (:xep:`0055`) on the component
233 Form values is a dict in which keys are defined in :attr:`.BaseGateway.SEARCH_FIELDS`
235 :param form_values: search query, defined for a specific plugin by overriding
236 in :attr:`.BaseGateway.SEARCH_FIELDS`
237 :return:
238 """
239 raise NotImplementedError
241 async def on_avatar(
242 self,
243 bytes_: bytes | None,
244 hash_: str | None,
245 type_: str | None,
246 width: int | None,
247 height: int | None,
248 ) -> None:
249 """
250 Triggered when the user uses modifies their avatar via :xep:`0084`.
252 :param bytes_: The data of the avatar. According to the spec, this
253 should always be a PNG, but some implementations do not respect
254 that. If `None` it means the user has unpublished their avatar.
255 :param hash_: The SHA1 hash of the avatar data. This is an identifier of
256 the avatar.
257 :param type_: The MIME type of the avatar.
258 :param width: The width of the avatar image.
259 :param height: The height of the avatar image.
260 """
261 raise NotImplementedError
263 async def on_leave_space(self, space_legacy_id: str) -> None:
264 """
265 Triggered when the user sends a request to leave a :xep:`0503` space.
267 :param space_legacy_id: The legacy ID of the space to leave
268 """
269 raise NotImplementedError
271 async def on_preferences(
272 self, previous: dict[str, Any], new: dict[str, Any]
273 ) -> None:
274 """
275 This is called when the user updates their preferences.
277 Override this if you need set custom preferences field and need to trigger
278 something when a preference has changed.
279 """
280 raise NotImplementedError
282 def __reset_ready(self) -> None:
283 self.ready = self.xmpp.loop.create_future()
285 @property
286 def logged(self) -> bool:
287 return self._logged
289 @logged.setter
290 def logged(self, v: bool) -> None:
291 self.is_logging_in = False
292 self._logged = v
293 if self.ready.done():
294 if v:
295 return
296 self.__reset_ready()
297 self.shutdown(logout=False)
298 with self.xmpp.store.session() as orm:
299 self.xmpp.store.mam.reset_source(orm)
300 self.xmpp.store.rooms.reset_updated(orm)
301 self.xmpp.store.contacts.reset_updated(orm)
302 orm.commit()
303 else:
304 if v:
305 self.ready.set_result(True)
307 def __repr__(self) -> str:
308 return f"<Session of {self.user_jid}>"
310 def shutdown(self, logout: bool = True) -> asyncio.Task[None]:
311 for m in self.bookmarks:
312 m.shutdown()
313 with self.xmpp.store.session() as orm:
314 for localpart in orm.execute(
315 sa.select(Contact.jid_localpart).filter_by(
316 user=self.user, is_friend=True
317 )
318 ).scalars():
319 pres = self.xmpp.make_presence(
320 pfrom=f"{localpart}@{self.xmpp.boundjid.bare}",
321 pto=self.user_jid,
322 ptype="unavailable",
323 pstatus="Gateway has shut down.",
324 )
325 pres.send()
326 if logout:
327 return self.xmpp.loop.create_task(
328 self.__logout(), name=f"logout of {self.user}"
329 )
330 else:
331 return self.xmpp.loop.create_task(noop_coro(), name="noop")
333 async def __logout(self) -> None:
334 try:
335 await self.logout()
336 except NotImplementedError:
337 pass
338 except KeyboardInterrupt:
339 pass
341 def raise_if_not_logged(self) -> None:
342 if not self.logged:
343 raise XMPPError(
344 "internal-server-error",
345 text="You are not logged to the legacy network",
346 )
348 @classmethod
349 def _from_user_or_none(cls, user: GatewayUser | None) -> Self:
350 if user is None:
351 log.debug("user not found")
352 raise XMPPError(text="User not found", condition="subscription-required")
354 session = _sessions.get(user.jid.bare)
355 if session is None:
356 _sessions[user.jid.bare] = session = cls(user)
357 assert isinstance(session, cls)
358 return session
360 @classmethod
361 def from_user(cls, user: GatewayUser) -> Self:
362 return cls._from_user_or_none(user)
364 @classmethod
365 def from_stanza(cls, s: Message | Iq | Presence) -> Self:
366 # """
367 # Get a user's :class:`.LegacySession` using the "from" field of a stanza
368 #
369 # Meant to be called from :class:`BaseGateway` only.
370 #
371 # :param s:
372 # :return:
373 # """
374 return cls.from_jid(s.get_from())
376 @classmethod
377 def from_jid(cls, jid: JID) -> Self:
378 # """
379 # Get a user's :class:`.LegacySession` using its jid
380 #
381 # Meant to be called from :class:`BaseGateway` only.
382 #
383 # :param jid:
384 # :return:
385 # """
386 session = _sessions.get(jid.bare)
387 if session is not None:
388 assert isinstance(session, cls)
389 return session
390 with cls.xmpp.store.session() as orm:
391 user = orm.query(GatewayUser).filter_by(jid=jid.bare).one_or_none()
392 return cls._from_user_or_none(user)
394 @classmethod
395 async def kill_by_jid(cls, jid: JID) -> None:
396 # """
397 # Terminate a user session.
398 #
399 # Meant to be called from :class:`BaseGateway` only.
400 #
401 # :param jid:
402 # :return:
403 # """
404 log.debug("Killing session of %s", jid)
405 for user_jid, session in _sessions.items():
406 if user_jid == jid.bare:
407 break
408 else:
409 log.debug("Did not find a session for %s", jid)
410 return
411 for c in session.contacts:
412 c.unsubscribe()
413 for m in session.bookmarks:
414 m.shutdown()
416 try:
417 session = _sessions.pop(jid.bare)
418 except KeyError:
419 log.warning("User not found during unregistration")
420 return
422 session.cancel_all_tasks()
424 await session.on_unregister()
425 with cls.xmpp.store.session() as orm:
426 orm.delete(session.user)
427 orm.commit()
429 def __ack(self, msg: Message) -> None:
430 if not self.xmpp.PROPER_RECEIPTS:
431 self.xmpp.delivery_receipt.ack(msg)
433 def send_gateway_status(
434 self,
435 status: str | None = None,
436 show: PresenceShows | None = None,
437 **kwargs: Any, # noqa
438 ) -> None:
439 """
440 Send a presence from the gateway to the user.
442 Can be used to indicate the user session status, ie "SMS code required", "connected", …
444 :param status: A status message
445 :param show: Presence stanza 'show' element. I suggest using "dnd" to show
446 that the gateway is not fully functional
447 """
448 self.__cached_presence = CachedPresence(status, show, kwargs)
449 self.xmpp.send_presence(
450 pto=self.user_jid.bare, pstatus=status, pshow=show, **kwargs
451 )
453 def send_cached_presence(self, to: JID) -> None:
454 if not self.__cached_presence:
455 self.xmpp.send_presence(pto=to, ptype="unavailable")
456 return
457 self.xmpp.send_presence(
458 pto=to,
459 pstatus=self.__cached_presence.status,
460 pshow=self.__cached_presence.show,
461 **self.__cached_presence.kwargs,
462 )
464 def send_gateway_message(
465 self,
466 text: str,
467 **msg_kwargs: Any, # noqa
468 ) -> None:
469 """
470 Send a message from the gateway component to the user.
472 Can be used to indicate the user session status, ie "SMS code required", "connected", …
474 :param text: A text
475 """
476 self.xmpp.send_text(text, mto=self.user_jid, **msg_kwargs)
478 def send_gateway_invite(
479 self,
480 muc: AnyMUC | JID | str,
481 reason: str | None = None,
482 password: str | None = None,
483 ) -> None:
484 """
485 Send an invitation to join a MUC, emanating from the gateway component.
487 :param muc:
488 :param reason:
489 :param password:
490 """
491 self.xmpp.invite_to(muc, reason=reason, password=password, mto=self.user_jid)
493 async def input(self, text: str, **msg_kwargs: Any) -> str: # noqa
494 """
495 Request user input via direct messages from the gateway component.
497 Wraps call to :meth:`.BaseSession.input`
499 :param text: The prompt to send to the user
500 :param msg_kwargs: Extra attributes
501 :return:
502 """
503 return await self.xmpp.input(self.user_jid, text, **msg_kwargs)
505 async def send_qr(self, text: str) -> None:
506 """
507 Sends a QR code generated from 'text' via HTTP Upload and send the URL to
508 ``self.user``
510 :param text: Text to encode as a QR code
511 """
512 await self.xmpp.send_qr(text, mto=self.user_jid)
514 async def get_contact_or_group_or_participant(
515 self, jid: JID, create: bool = True
516 ) -> "LegacyContact | AnyMUC | AnyParticipant | None":
517 contact: LegacyContact | None = self.contacts.by_jid_only_if_exists(jid)
518 if contact is not None:
519 return contact
520 if (muc := self.bookmarks.by_jid_only_if_exists(JID(jid.bare))) is not None:
521 return await self.__get_muc_or_participant(muc, jid)
522 else:
523 muc = None
525 if not create:
526 return None
528 try:
529 contact = await self.contacts.by_jid(jid)
530 except XMPPError:
531 if muc is None:
532 try:
533 muc = await self.bookmarks.by_jid(jid)
534 except XMPPError:
535 return None
536 return await self.__get_muc_or_participant(muc, jid)
537 return contact
539 @staticmethod
540 async def __get_muc_or_participant(
541 muc: AnyMUC, jid: JID
542 ) -> "AnyMUC | AnyParticipant | None":
543 if nick := jid.resource:
544 return await muc.get_participant(nick, create=False, fill_first=True)
545 return muc
547 async def wait_for_ready(self, timeout: float | None = 10) -> None:
548 # """
549 # Wait until session, contacts and bookmarks are ready
550 #
551 # (slidge internal use)
552 #
553 # :param timeout:
554 # :return:
555 # """
556 try:
557 await asyncio.wait_for(asyncio.shield(self.ready), timeout)
558 await asyncio.wait_for(asyncio.shield(self.contacts.ready), timeout)
559 await asyncio.wait_for(asyncio.shield(self.bookmarks.ready), timeout)
560 except TimeoutError:
561 raise XMPPError(
562 "recipient-unavailable",
563 "Legacy session is not fully initialized, retry later",
564 )
566 def legacy_module_data_update(self, data: JSONSerializable) -> None:
567 user = self.user
568 user.legacy_module_data.update(data)
569 self.xmpp.store.users.update(user)
571 def legacy_module_data_set(self, data: JSONSerializable) -> None:
572 user = self.user
573 user.legacy_module_data = data
574 self.xmpp.store.users.update(user)
576 def legacy_module_data_clear(self) -> None:
577 user = self.user
578 user.legacy_module_data.clear()
579 self.xmpp.store.users.update(user)
582# References to `BaseSession` need for this to be defined.
583# Does not satisfy `roster_cls = type[Roster]` for subclasses.
584BaseSession.roster_cls = LegacyRoster # type:ignore[misc]
585BaseSession.bookmarks_cls = LegacyBookmarks # type:ignore[misc]
587# keys = user.jid.bare
588_sessions: dict[str, AnySession] = {}
589log = logging.getLogger(__name__)