Coverage for slidge/core/session.py: 82%
233 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-22 03:37 +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 roster_cls: type[LegacyRosterType_co]
75 """
76 The :class:`.LegacyRoster` subclass to use for this session's contacts.
78 Derived automatically from the first generic parameter, e.g.,
79 ``class Session(BaseSession[Roster, Bookmarks])``, which also types
80 :attr:`.contacts`.
81 """
82 bookmarks_cls: type[LegacyBookmarksType_co]
83 """
84 The :class:`.LegacyBookmarks` subclass to use for this session's groups.
86 Derived automatically from the second generic parameter, e.g.,
87 ``class Session(BaseSession[Roster, Bookmarks])``, which also types
88 :attr:`.bookmarks`.
89 """
91 def __init_subclass__(cls, **kwargs: object) -> None:
92 super().__init_subclass__(**kwargs)
93 derive_wired_class(cls, BaseSession, "roster_cls", "bookmarks_cls")
95 def __init__(self, user: GatewayUser) -> None:
96 super().__init__()
97 self.user = user
98 """
99 The :term:`slidge user <User>`.
100 """
101 self.log = logging.getLogger(user.jid.bare)
103 self.ignore_messages = set[str]()
105 self.contacts: Final[LegacyRosterType_co] = self.roster_cls(self)
106 """This session's roster, an instance of :attr:`.roster_cls`."""
108 self.is_logging_in = False
109 self._logged = False
110 self.__reset_ready()
112 self.bookmarks: Final[LegacyBookmarksType_co] = self.bookmarks_cls(self)
113 """This session's groups, an instance of :attr:`.bookmarks_cls`."""
115 self.thread_creation_lock = asyncio.Lock()
117 self.__cached_presence: CachedPresence | None = None
119 self.__tasks = set[asyncio.Task[Any]]()
121 @property
122 def user_jid(self) -> JID:
123 return self.user.jid
125 @property
126 def user_pk(self) -> int:
127 return self.user.id
129 @property
130 def http(self) -> aiohttp.ClientSession:
131 return self.xmpp.http
133 def __remove_task(self, fut: Task[Any]) -> None:
134 self.log.debug("Removing fut %s", fut)
135 self.__tasks.remove(fut)
137 def create_task(
138 self, coro: Coroutine[Any, Any, Any], name: str | None = None
139 ) -> asyncio.Task[Any]:
140 if name is None:
141 warnings.warn(
142 "Calling Session.create_task without a 'name' argument will "
143 "be deprecated in slidge >= 0.6",
144 DeprecationWarning,
145 )
146 task = self.xmpp.create_task(coro, name=name)
147 self.__tasks.add(task)
148 self.log.debug("Creating task %s", task)
149 task.add_done_callback(lambda _: self.__remove_task(task))
150 return task
152 def cancel_all_tasks(self) -> None:
153 for task in self.__tasks:
154 task.cancel()
156 @abc.abstractmethod
157 async def login(self) -> str | None:
158 """
159 Logs in the gateway user to the legacy network.
161 Triggered when the gateway start and on user registration.
162 It is recommended that this function returns once the user is logged in,
163 so if you need to await forever (for instance to listen to incoming events),
164 it's a good idea to wrap your listener in an asyncio.Task.
166 :return: Optionally, a text to use as the gateway status, e.g., "Connected as 'dude@legacy.network'"
167 """
168 raise NotImplementedError
170 async def logout(self) -> None:
171 """
172 Logs out the gateway user from the legacy network.
174 Called on gateway shutdown.
175 """
176 raise NotImplementedError
178 async def on_unregister(self) -> None:
179 """Called when the user unregisters from the gateway, after their
180 session has been terminated but before their persistent data is
181 deleted.
183 Optionally override this if you need to clean up additional stuff; by
184 default it just calls :meth:`.logout`.
185 """
186 with contextlib.suppress(NotImplementedError):
187 await self.logout()
189 async def on_presence(
190 self,
191 resource: str,
192 show: PseudoPresenceShow,
193 status: str,
194 resources: dict[str, ResourceDict],
195 merged_resource: ResourceDict | None,
196 ) -> None:
197 """
198 Called when the gateway component receives a presence, ie, when
199 one of the user's clients goes online of offline, or changes its
200 status.
202 :param resource: The XMPP client identifier, arbitrary string.
203 :param show: The presence ``<show>``, if available. If the resource is
204 just 'available' without any ``<show>`` element, this is an empty
205 str.
206 :param status: A status message, like a deeply profound quote, eg,
207 "Roses are red, violets are blue, [INSERT JOKE]".
208 :param resources: A summary of all the resources for this user.
209 :param merged_resource: A global presence for the user account,
210 following rules described in :meth:`merge_resources`
211 """
212 raise NotImplementedError
214 async def on_search(self, form_values: dict[str, str]) -> SearchResult | None:
215 """
216 Triggered when the user uses Jabber Search (:xep:`0055`) on the component
218 Form values is a dict in which keys are defined in :attr:`.BaseGateway.SEARCH_FIELDS`
220 :param form_values: search query, defined for a specific plugin by overriding
221 in :attr:`.BaseGateway.SEARCH_FIELDS`
222 :return:
223 """
224 raise NotImplementedError
226 async def on_avatar(
227 self,
228 bytes_: bytes | None,
229 hash_: str | None,
230 type_: str | None,
231 width: int | None,
232 height: int | None,
233 ) -> None:
234 """
235 Triggered when the user uses modifies their avatar via :xep:`0084`.
237 :param bytes_: The data of the avatar. According to the spec, this
238 should always be a PNG, but some implementations do not respect
239 that. If `None` it means the user has unpublished their avatar.
240 :param hash_: The SHA1 hash of the avatar data. This is an identifier of
241 the avatar.
242 :param type_: The MIME type of the avatar.
243 :param width: The width of the avatar image.
244 :param height: The height of the avatar image.
245 """
246 raise NotImplementedError
248 async def on_leave_space(self, space_legacy_id: str) -> None:
249 """
250 Triggered when the user sends a request to leave a :xep:`0503` space.
252 :param space_legacy_id: The legacy ID of the space to leave
253 """
254 raise NotImplementedError
256 async def on_preferences(
257 self, previous: dict[str, Any], new: dict[str, Any]
258 ) -> None:
259 """
260 This is called when the user updates their preferences.
262 Override this if you need set custom preferences field and need to trigger
263 something when a preference has changed.
264 """
265 raise NotImplementedError
267 def __reset_ready(self) -> None:
268 self.ready = self.xmpp.loop.create_future()
270 @property
271 def logged(self) -> bool:
272 return self._logged
274 @logged.setter
275 def logged(self, v: bool) -> None:
276 self.is_logging_in = False
277 self._logged = v
278 if self.ready.done():
279 if v:
280 return
281 self.__reset_ready()
282 self.shutdown(logout=False)
283 with self.xmpp.store.session() as orm:
284 self.xmpp.store.mam.reset_source(orm)
285 self.xmpp.store.rooms.reset_updated(orm)
286 self.xmpp.store.contacts.reset_updated(orm)
287 orm.commit()
288 else:
289 if v:
290 self.ready.set_result(True)
292 def __repr__(self) -> str:
293 return f"<Session of {self.user_jid}>"
295 def shutdown(self, logout: bool = True) -> asyncio.Task[None]:
296 for m in self.bookmarks:
297 m.shutdown()
298 with self.xmpp.store.session() as orm:
299 for localpart in orm.execute(
300 sa.select(Contact.jid_localpart).filter_by(
301 user=self.user, is_friend=True
302 )
303 ).scalars():
304 pres = self.xmpp.make_presence(
305 pfrom=f"{localpart}@{self.xmpp.boundjid.bare}",
306 pto=self.user_jid,
307 ptype="unavailable",
308 pstatus="Gateway has shut down.",
309 )
310 pres.send()
311 if logout:
312 return self.xmpp.loop.create_task(
313 self.__logout(), name=f"logout of {self.user}"
314 )
315 else:
316 return self.xmpp.loop.create_task(noop_coro(), name="noop")
318 async def __logout(self) -> None:
319 try:
320 await self.logout()
321 except NotImplementedError:
322 pass
323 except KeyboardInterrupt:
324 pass
326 def raise_if_not_logged(self) -> None:
327 if not self.logged:
328 raise XMPPError(
329 "internal-server-error",
330 text="You are not logged to the legacy network",
331 )
333 @classmethod
334 def _from_user_or_none(cls, user: GatewayUser | None) -> Self:
335 if user is None:
336 log.debug("user not found")
337 raise XMPPError(text="User not found", condition="subscription-required")
339 session = _sessions.get(user.jid.bare)
340 if session is None:
341 _sessions[user.jid.bare] = session = cls(user)
342 assert isinstance(session, cls)
343 return session
345 @classmethod
346 def from_user(cls, user: GatewayUser) -> Self:
347 return cls._from_user_or_none(user)
349 @classmethod
350 def from_stanza(cls, s: Message | Iq | Presence) -> Self:
351 # """
352 # Get a user's :class:`.LegacySession` using the "from" field of a stanza
353 #
354 # Meant to be called from :class:`BaseGateway` only.
355 #
356 # :param s:
357 # :return:
358 # """
359 return cls.from_jid(s.get_from())
361 @classmethod
362 def from_jid(cls, jid: JID) -> Self:
363 # """
364 # Get a user's :class:`.LegacySession` using its jid
365 #
366 # Meant to be called from :class:`BaseGateway` only.
367 #
368 # :param jid:
369 # :return:
370 # """
371 session = _sessions.get(jid.bare)
372 if session is not None:
373 assert isinstance(session, cls)
374 return session
375 with cls.xmpp.store.session() as orm:
376 user = orm.query(GatewayUser).filter_by(jid=jid.bare).one_or_none()
377 return cls._from_user_or_none(user)
379 @classmethod
380 async def kill_by_jid(cls, jid: JID) -> None:
381 # """
382 # Terminate a user session.
383 #
384 # Meant to be called from :class:`BaseGateway` only.
385 #
386 # :param jid:
387 # :return:
388 # """
389 log.debug("Killing session of %s", jid)
390 for user_jid, session in _sessions.items():
391 if user_jid == jid.bare:
392 break
393 else:
394 log.debug("Did not find a session for %s", jid)
395 return
396 for c in session.contacts:
397 c.unsubscribe()
398 for m in session.bookmarks:
399 m.shutdown()
401 try:
402 session = _sessions.pop(jid.bare)
403 except KeyError:
404 log.warning("User not found during unregistration")
405 return
407 session.cancel_all_tasks()
409 await session.on_unregister()
410 with cls.xmpp.store.session() as orm:
411 orm.delete(session.user)
412 orm.commit()
414 def __ack(self, msg: Message) -> None:
415 if not self.xmpp.PROPER_RECEIPTS:
416 self.xmpp.delivery_receipt.ack(msg)
418 def send_gateway_status(
419 self,
420 status: str | None = None,
421 show: PresenceShows | None = None,
422 **kwargs: Any, # noqa
423 ) -> None:
424 """
425 Send a presence from the gateway to the user.
427 Can be used to indicate the user session status, ie "SMS code required", "connected", …
429 :param status: A status message
430 :param show: Presence stanza 'show' element. I suggest using "dnd" to show
431 that the gateway is not fully functional
432 """
433 self.__cached_presence = CachedPresence(status, show, kwargs)
434 self.xmpp.send_presence(
435 pto=self.user_jid.bare, pstatus=status, pshow=show, **kwargs
436 )
438 def send_cached_presence(self, to: JID) -> None:
439 if not self.__cached_presence:
440 self.xmpp.send_presence(pto=to, ptype="unavailable")
441 return
442 self.xmpp.send_presence(
443 pto=to,
444 pstatus=self.__cached_presence.status,
445 pshow=self.__cached_presence.show,
446 **self.__cached_presence.kwargs,
447 )
449 def send_gateway_message(
450 self,
451 text: str,
452 **msg_kwargs: Any, # noqa
453 ) -> None:
454 """
455 Send a message from the gateway component to the user.
457 Can be used to indicate the user session status, ie "SMS code required", "connected", …
459 :param text: A text
460 """
461 self.xmpp.send_text(text, mto=self.user_jid, **msg_kwargs)
463 def send_gateway_invite(
464 self,
465 muc: AnyMUC | JID | str,
466 reason: str | None = None,
467 password: str | None = None,
468 ) -> None:
469 """
470 Send an invitation to join a MUC, emanating from the gateway component.
472 The send is deferred until the session is fully initialised
473 (``bookmarks.ready``), so that clients can actually join the room when
474 they receive the invitation.
476 :param muc:
477 :param reason:
478 :param password:
479 """
480 self.xmpp.invite_to(
481 muc,
482 reason=reason,
483 password=password,
484 session=self,
485 mto=self.user_jid,
486 )
488 async def input(self, text: str, **msg_kwargs: Any) -> str: # noqa
489 """
490 Request user input via direct messages from the gateway component.
492 Wraps call to :meth:`.BaseSession.input`
494 :param text: The prompt to send to the user
495 :param msg_kwargs: Extra attributes
496 :return:
497 """
498 return await self.xmpp.input(self.user_jid, text, **msg_kwargs)
500 async def send_qr(self, text: str) -> None:
501 """
502 Sends a QR code generated from 'text' via HTTP Upload and send the URL to
503 ``self.user``
505 :param text: Text to encode as a QR code
506 """
507 await self.xmpp.send_qr(text, mto=self.user_jid)
509 async def get_contact_or_group_or_participant(
510 self, jid: JID, create: bool = True
511 ) -> "LegacyContact | AnyMUC | AnyParticipant | None":
512 type_, _legacy_id = self.xmpp.localpart_to_legacy_id(jid.local)
513 match type_:
514 case "contact":
515 if create:
516 # for some invariance thing, the return value type cannot
517 # be inferred by mypy
518 return await self.contacts.by_jid(jid) # type:ignore[no-any-return]
519 else:
520 return self.contacts.by_jid_only_if_exists(jid)
521 case "group":
522 if create:
523 muc = await self.bookmarks.by_jid(jid)
524 else:
525 muc = self.bookmarks.by_jid_only_if_exists(JID(jid.bare))
526 if muc is None:
527 return None
528 return await self.__get_muc_or_participant(muc, jid)
529 case _:
530 raise XMPPError(
531 "internal-server-error",
532 "Session.jid_localpart_type returned an invalid value",
533 )
535 @staticmethod
536 async def __get_muc_or_participant(
537 muc: AnyMUC, jid: JID
538 ) -> "AnyMUC | AnyParticipant | None":
539 if nick := jid.resource:
540 return await muc.get_participant(nick, create=False, fill_first=True)
541 return muc
543 async def wait_for_ready(self, timeout: float | None = 10) -> None:
544 # """
545 # Wait until session, contacts and bookmarks are ready
546 #
547 # (slidge internal use)
548 #
549 # :param timeout:
550 # :return:
551 # """
552 try:
553 await asyncio.wait_for(asyncio.shield(self.ready), timeout)
554 await asyncio.wait_for(asyncio.shield(self.contacts.ready), timeout)
555 await asyncio.wait_for(asyncio.shield(self.bookmarks.ready), timeout)
556 except TimeoutError:
557 raise XMPPError(
558 "recipient-unavailable",
559 "Legacy session is not fully initialized, retry later",
560 )
562 def legacy_module_data_update(self, data: JSONSerializable) -> None:
563 user = self.user
564 user.legacy_module_data.update(data)
565 self.xmpp.store.users.update(user)
567 def legacy_module_data_set(self, data: JSONSerializable) -> None:
568 user = self.user
569 user.legacy_module_data = data
570 self.xmpp.store.users.update(user)
572 def legacy_module_data_clear(self) -> None:
573 user = self.user
574 user.legacy_module_data.clear()
575 self.xmpp.store.users.update(user)
578# References to `BaseSession` need for this to be defined.
579# Does not satisfy `roster_cls = type[Roster]` for subclasses.
580BaseSession.roster_cls = LegacyRoster # type:ignore[misc]
581BaseSession.bookmarks_cls = LegacyBookmarks # type:ignore[misc]
583# keys = user.jid.bare
584_sessions: dict[str, AnySession] = {}
585log = logging.getLogger(__name__)