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