Coverage for slidge/core/session.py: 84%
257 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-10 12:26 +0000
1from __future__ import annotations
3import abc
4import asyncio
5import contextlib
6import functools
7import logging
8import warnings
9from asyncio.tasks import Task
10from collections.abc import Coroutine
11from typing import Any, ClassVar, Final, NamedTuple, Self, cast
13import aiohttp
14import sqlalchemy as sa
15from slixmpp import JID, Iq, Message, Presence
16from slixmpp.exceptions import XMPPError
17from slixmpp.types import PresenceShows, ResourceDict
19from ..command import SearchResult
20from ..contact import LegacyContact, LegacyRoster
21from ..db.models import Contact, GatewayUser
22from ..db.track import TrackedDict
23from ..room import LegacyBookmarks
24from ..space import LegacySpaceNodes
25from ..util.lock import NamedLockMixin
26from ..util.types import (
27 AnyBookmarks,
28 AnyGateway,
29 AnyParticipant,
30 AnyRoom,
31 PseudoPresenceShow,
32)
33from ..util.util import derive_wired_class, noop_coro
36class CachedPresence(NamedTuple):
37 status: str | None
38 show: str | None
39 kwargs: dict[str, Any]
42class BaseSession[
43 RosterT: LegacyRoster = LegacyRoster,
44 BookmarksT: AnyBookmarks = LegacyBookmarks,
45 SpaceNodesT: LegacySpaceNodes = LegacySpaceNodes,
46 ModuleDataT: Any = Any,
47 PrefsT: Any = Any,
48](NamedLockMixin, abc.ABC):
49 """
50 The session of a registered :term:`User`.
52 Represents a gateway user logged in to the legacy network and performing actions.
54 Will be instantiated automatically on slidge startup for each registered user,
55 or upon registration for new (validated) users.
57 Must be subclassed for a functional :term:`Legacy Module`.
58 """
60 """
61 Since we cannot set the XMPP ID of messages sent by XMPP clients, we need to keep a mapping
62 between XMPP IDs and legacy message IDs if we want to further refer to a message that was sent
63 by the user. This also applies to 'carboned' messages, ie, messages sent by the user from
64 the official client of a legacy network.
65 """
67 xmpp: AnyGateway
68 """
69 The gateway instance singleton. Use it for low-level XMPP calls or custom methods that are not
70 session-specific.
72 It is set on the session class by the gateway on startup, before any
73 session is instantiated. Plugins may redeclare it with their own gateway
74 class for typed access, e.g. ``xmpp: "Gateway"``.
75 """
77 legacy_module_data: ModuleDataT
79 # References to `BaseSession` need for _roster_cls and _bookmarks_cls to be defined.
80 # Does not satisfy `roster_cls = type[Roster]` for subclasses.
82 _roster_cls: ClassVar[type[LegacyRoster]] = LegacyRoster
83 """
84 The :class:`.LegacyRoster` subclass to use for this session's contacts.
86 Derived automatically from the first generic parameter, e.g.,
87 ``class Session(BaseSession[Roster, Bookmarks])``, which also types
88 :attr:`.contacts`.
89 """
91 _bookmarks_cls: ClassVar[type[LegacyBookmarks]] = LegacyBookmarks
92 """
93 The :class:`.LegacyBookmarks` subclass to use for this session's rooms.
95 Derived automatically from the second generic parameter, e.g.,
96 ``class Session(BaseSession[Roster, Bookmarks])``, which also types
97 :attr:`.rooms`.
98 """
100 _space_nodes_cls: ClassVar[type[LegacySpaceNodes]] = LegacySpaceNodes
102 def __init_subclass__(cls, **kwargs: object) -> None:
103 super().__init_subclass__(**kwargs)
104 derive_wired_class(
105 cls, BaseSession, "_roster_cls", "_bookmarks_cls", "_space_nodes_cls"
106 )
108 def __init__(self, user: GatewayUser) -> None:
109 super().__init__()
110 self.user = user
111 """
112 The :term:`slidge user <User>`.
113 """
114 self.log = logging.getLogger(user.jid.bare)
116 self.ignore_messages = set[str]()
118 self.contacts: Final[RosterT] = cast(RosterT, self._roster_cls(self))
119 """This session's roster, an instance of :attr:`.roster_cls`."""
121 self.is_logging_in = False
122 self._logged = False
123 self.__reset_ready()
125 self.rooms: Final[BookmarksT] = cast(BookmarksT, self._bookmarks_cls(self))
126 """This session's rooms, an instance of :attr:`.bookmarks_cls`."""
128 self.spaces: Final[SpaceNodesT] = cast(SpaceNodesT, self._space_nodes_cls(self))
130 self.thread_creation_lock = asyncio.Lock()
132 self.__cached_presence: CachedPresence | None = None
134 self.__tasks = dict[str, asyncio.Task[Any]]()
136 self._pep_broadcasted_resources = set[str]()
138 self.legacy_module_data = TrackedDict( # type:ignore[assignment] # ty:ignore
139 self.user.legacy_module_data, self._on_legacy_module_data_change
140 )
142 def _on_legacy_module_data_change(self, new_data: ModuleDataT) -> None:
143 self.user.legacy_module_data = new_data
144 self.xmpp.store.users.update(self.user)
146 @property
147 def user_jid(self) -> JID:
148 return self.user.jid
150 @property
151 def user_pk(self) -> int:
152 return self.user.id
154 @property
155 def http(self) -> aiohttp.ClientSession:
156 return self.xmpp.http
158 def __remove_task(self, name: str, task: Task[Any]) -> None:
159 # Only remove the entry if it still points at this task;
160 # it may have been replaced by a newer task with the same name.
161 if self.__tasks.get(name) is task:
162 del self.__tasks[name]
164 async def __run_after[T](
165 self, prev: asyncio.Task[Any] | None, coro: Coroutine[Any, Any, T]
166 ) -> T:
167 try:
168 if prev is not None and not prev.done():
169 self.log.debug("Waiting for previous task named %s to complete…", prev)
170 await asyncio.wait({prev})
171 except asyncio.CancelledError:
172 # Never started, so close the coroutine to avoid a
173 # "coroutine was never awaited" warning.
174 coro.close()
175 raise
176 return await coro
178 def create_task(
179 self, coro: Coroutine[Any, Any, Any], name: str
180 ) -> asyncio.Task[Any]:
181 self.log.debug("Creating task %s", name)
182 prev = self.__tasks.get(name)
183 task = self.xmpp.create_task(self.__run_after(prev, coro), name=name)
184 self.__tasks[name] = task
185 task.add_done_callback(functools.partial(self.__remove_task, name))
186 return task
188 def cancel_all_tasks(self) -> None:
189 for task in self.__tasks.values():
190 task.cancel()
192 def get_task(self, name: str) -> asyncio.Task[Any] | None:
193 return self.__tasks.get(name)
195 @abc.abstractmethod
196 async def login(self) -> str | None:
197 """
198 Logs in the gateway user to the legacy network.
200 Triggered when the gateway start and on user registration.
201 It is recommended that this function returns once the user is logged in,
202 so if you need to await forever (for instance to listen to incoming events),
203 it's a good idea to wrap your listener in an asyncio.Task.
205 :return: Optionally, a text to use as the gateway status, e.g., "Connected as 'dude@legacy.network'"
206 """
207 raise NotImplementedError
209 async def logout(self) -> None:
210 """
211 Logs out the gateway user from the legacy network.
213 Called on gateway shutdown.
214 """
215 raise NotImplementedError
217 async def on_unregister(self) -> None:
218 """Called when the user unregisters from the gateway, after their
219 session has been terminated but before their persistent data is
220 deleted.
222 Optionally override this if you need to clean up additional stuff; by
223 default it just calls :meth:`.logout`.
224 """
225 with contextlib.suppress(NotImplementedError):
226 await self.logout()
228 async def on_presence(
229 self,
230 resource: str,
231 show: PseudoPresenceShow,
232 status: str,
233 resources: dict[str, ResourceDict],
234 merged_resource: ResourceDict | None,
235 ) -> None:
236 """
237 Called when the gateway component receives a presence, ie, when
238 one of the user's clients goes online of offline, or changes its
239 status.
241 :param resource: The XMPP client identifier, arbitrary string.
242 :param show: The presence ``<show>``, if available. If the resource is
243 just 'available' without any ``<show>`` element, this is an empty
244 str.
245 :param status: A status message, like a deeply profound quote, eg,
246 "Roses are red, violets are blue, [INSERT JOKE]".
247 :param resources: A summary of all the resources for this user.
248 :param merged_resource: A global presence for the user account,
249 following rules described in :meth:`merge_resources`
250 """
251 raise NotImplementedError
253 async def on_search(self, form_values: dict[str, str]) -> SearchResult | None:
254 """
255 Triggered when the user uses Jabber Search (:xep:`0055`) on the component
257 Form values is a dict in which keys are defined in :attr:`.BaseGateway.SEARCH_FIELDS`
259 :param form_values: search query, defined for a specific plugin by overriding
260 in :attr:`.BaseGateway.SEARCH_FIELDS`
261 :return:
262 """
263 raise NotImplementedError
265 async def on_avatar(
266 self,
267 bytes_: bytes | None,
268 hash_: str | None,
269 type_: str | None,
270 width: int | None,
271 height: int | None,
272 ) -> None:
273 """
274 Triggered when the user uses modifies their avatar via :xep:`0084`.
276 :param bytes_: The data of the avatar. According to the spec, this
277 should always be a PNG, but some implementations do not respect
278 that. If `None` it means the user has unpublished their avatar.
279 :param hash_: The SHA1 hash of the avatar data. This is an identifier of
280 the avatar.
281 :param type_: The MIME type of the avatar.
282 :param width: The width of the avatar image.
283 :param height: The height of the avatar image.
284 """
285 raise NotImplementedError
287 async def on_leave_space(self, space_legacy_id: str) -> None:
288 """
289 Triggered when the user sends a request to leave a :xep:`0503` space.
291 :param space_legacy_id: The legacy ID of the space to leave
292 """
293 raise NotImplementedError
295 # type hints should really be `PrefsT`, but it changes inferred variance from
296 # covariant to invariant, which breaks a lot down the line. With `Any`, we
297 # don't enforce `PrefsT`, but at least it makes it possible for concrete
298 # subclasses to narrow down with the to legacy module-specific preferences
299 # `TypedDict`.
300 async def on_preferences(self, previous: Any, new: Any) -> None: # noqa: ANN401
301 """
302 This is called when the user updates their preferences.
304 Override this if you need set custom preferences field and need to trigger
305 something when a preference has changed.
306 """
307 raise NotImplementedError
309 @property
310 def preferences(self) -> PrefsT:
311 return cast(PrefsT, self.user.preferences)
313 def __reset_ready(self) -> None:
314 self.ready = self.xmpp.loop.create_future()
316 @property
317 def logged(self) -> bool:
318 return self._logged
320 @logged.setter
321 def logged(self, v: bool) -> None:
322 self.is_logging_in = False
323 self._logged = v
324 if self.ready.done():
325 if v:
326 return
327 self.__reset_ready()
328 self.shutdown(logout=False)
329 with self.xmpp.store.session() as orm:
330 self.xmpp.store.mam.reset_source(orm)
332 orm.commit()
333 else:
334 if v:
335 self.ready.set_result(True)
337 def __repr__(self) -> str:
338 return f"<Session of {self.user_jid}>"
340 def shutdown(self, logout: bool = True) -> asyncio.Task[None]:
341 with self.xmpp.store.session() as orm:
342 self.rooms._shutdown(orm)
343 for localpart in orm.execute(
344 sa.select(Contact.jid_localpart).filter_by(
345 user=self.user, is_friend=True
346 )
347 ).scalars():
348 pres = self.xmpp.make_presence(
349 pfrom=f"{localpart}@{self.xmpp.boundjid.bare}",
350 pto=self.user_jid,
351 ptype="unavailable",
352 pstatus="Gateway has shut down.",
353 )
354 pres.send()
355 if logout:
356 return self.xmpp.loop.create_task(
357 self.__logout(), name=f"logout of {self.user}"
358 )
359 else:
360 return self.xmpp.loop.create_task(noop_coro(), name="noop")
362 async def __logout(self) -> None:
363 try:
364 await self.logout()
365 except NotImplementedError:
366 pass
367 except KeyboardInterrupt:
368 pass
370 def raise_if_not_logged(self) -> None:
371 if not self.logged:
372 raise XMPPError(
373 "internal-server-error",
374 text="You are not logged to the legacy network",
375 )
377 @classmethod
378 def _from_user_or_none(cls, user: GatewayUser | None) -> Self:
379 if user is None:
380 log.debug("user not found")
381 raise XMPPError(text="User not found", condition="subscription-required")
383 session = _sessions.get(user.jid.bare)
384 if session is None:
385 _sessions[user.jid.bare] = session = cls(user)
386 assert isinstance(session, cls)
387 return session
389 @classmethod
390 def from_user(cls, user: GatewayUser) -> Self:
391 return cls._from_user_or_none(user)
393 @classmethod
394 def from_stanza(cls, s: Message | Iq | Presence) -> BaseSession:
395 # """
396 # Get a user's :class:`.LegacySession` using the "from" field of a stanza
397 #
398 # Meant to be called from :class:`BaseGateway` only.
399 #
400 # :param s:
401 # :return:
402 # """
403 return cls.from_jid(s.get_from())
405 @classmethod
406 def from_jid(cls, jid: JID) -> Self:
407 # """
408 # Get a user's :class:`.LegacySession` using its jid
409 #
410 # Meant to be called from :class:`BaseGateway` only.
411 #
412 # :param jid:
413 # :return:
414 # """
415 session = _sessions.get(jid.bare)
416 if session is not None:
417 assert isinstance(session, cls)
418 return session
419 with cls.xmpp.store.session() as orm:
420 user = orm.query(GatewayUser).filter_by(jid=jid.bare).one_or_none()
421 return cls._from_user_or_none(user)
423 @classmethod
424 async def kill_by_jid(cls, jid: JID) -> None:
425 # """
426 # Terminate a user session.
427 #
428 # Meant to be called from :class:`BaseGateway` only.
429 #
430 # :param jid:
431 # :return:
432 # """
433 log.debug("Killing session of %s", jid)
434 for user_jid, session in _sessions.items():
435 if user_jid == jid.bare:
436 break
437 else:
438 log.debug("Did not find a session for %s", jid)
439 return
440 for c in session.contacts:
441 c.unsubscribe()
442 with cls.xmpp.store.session() as orm:
443 session.rooms._shutdown(orm)
445 try:
446 session = _sessions.pop(jid.bare)
447 except KeyError:
448 log.warning("User not found during unregistration")
449 return
451 session.cancel_all_tasks()
453 await session.on_unregister()
454 with cls.xmpp.store.session() as orm:
455 orm.delete(session.user)
456 orm.commit()
458 def __ack(self, msg: Message) -> None:
459 if not self.xmpp.PROPER_RECEIPTS:
460 self.xmpp.delivery_receipt.ack(msg)
462 def send_gateway_status(
463 self,
464 status: str | None = None,
465 show: PresenceShows | None = None,
466 **kwargs: Any, # noqa
467 ) -> None:
468 """
469 Send a presence from the gateway to the user.
471 Can be used to indicate the user session status, ie "SMS code required", "connected", …
473 :param status: A status message
474 :param show: Presence stanza 'show' element. I suggest using "dnd" to show
475 that the gateway is not fully functional
476 """
477 self.__cached_presence = CachedPresence(status, show, kwargs)
478 self.xmpp.send_presence(
479 pto=self.user_jid.bare, pstatus=status, pshow=show, **kwargs
480 )
482 def send_cached_presence(self, to: JID) -> None:
483 if not self.__cached_presence:
484 self.xmpp.send_presence(pto=to, ptype="unavailable")
485 return
486 self.xmpp.send_presence(
487 pto=to,
488 pstatus=self.__cached_presence.status,
489 pshow=self.__cached_presence.show,
490 **self.__cached_presence.kwargs,
491 )
493 def send_gateway_message(
494 self,
495 text: str,
496 **msg_kwargs: Any, # noqa
497 ) -> None:
498 """
499 Send a message from the gateway component to the user.
501 Can be used to indicate the user session status, ie "SMS code required", "connected", …
503 :param text: A text
504 """
505 self.xmpp.send_text(text, mto=self.user_jid, **msg_kwargs)
507 def send_gateway_invite(
508 self,
509 muc: AnyRoom | JID | str,
510 reason: str | None = None,
511 password: str | None = None,
512 ) -> None:
513 """
514 Send an invitation to join a Room, emanating from the gateway component.
516 The send is deferred until the session is fully initialised
517 (``bookmarks.ready``), so that clients can actually join the room when
518 they receive the invitation.
520 :param muc:
521 :param reason:
522 :param password:
523 """
524 self.xmpp.invite_to(
525 muc,
526 reason=reason,
527 password=password,
528 session=self,
529 mto=self.user_jid,
530 )
532 async def input(self, text: str, **msg_kwargs: Any) -> str: # noqa
533 """
534 Request user input via direct messages from the gateway component.
536 Wraps call to :meth:`.BaseSession.input`
538 :param text: The prompt to send to the user
539 :param msg_kwargs: Extra attributes
540 :return:
541 """
542 return await self.xmpp.input(self.user_jid, text, **msg_kwargs)
544 async def send_qr(self, text: str) -> None:
545 """
546 Sends a QR code generated from 'text' via HTTP Upload and send the URL to
547 ``self.user``
549 :param text: Text to encode as a QR code
550 """
551 await self.xmpp.send_qr(text, mto=self.user_jid)
553 async def get_contact_or_group_or_participant(
554 self, jid: JID, create: bool = True
555 ) -> LegacyContact | AnyRoom | AnyParticipant | None:
556 type_, _legacy_id = self.xmpp.localpart_to_legacy_id(jid.local)
557 match type_:
558 case "contact":
559 if create:
560 return await self.contacts.get_or_create_by_jid(jid)
561 else:
562 return self.contacts.get_by_jid(jid)
563 case "room":
564 if create:
565 muc = await self.rooms.get_or_create_by_jid(jid)
566 else:
567 muc = self.rooms.get_by_jid(JID(jid.bare))
568 if muc is None:
569 return None
570 return await self.__get_muc_or_participant(muc, jid)
571 case _:
572 raise XMPPError(
573 "internal-server-error",
574 "Session.jid_localpart_type returned an invalid value",
575 )
577 @staticmethod
578 async def __get_muc_or_participant(
579 muc: AnyRoom, jid: JID
580 ) -> AnyRoom | AnyParticipant | None:
581 if resource := jid.resource:
582 await muc.participants.ensure_filled()
583 return muc.participants.get_by_resource(resource=resource)
584 return muc
586 async def wait_for_ready(self, timeout: float | None = 10) -> None:
587 # """
588 # Wait until session, contacts and bookmarks are ready
589 #
590 # (slidge internal use)
591 #
592 # :param timeout:
593 # :return:
594 # """
595 try:
596 await asyncio.wait_for(asyncio.shield(self.ready), timeout)
597 await asyncio.wait_for(asyncio.shield(self.contacts.ready), timeout)
598 await asyncio.wait_for(asyncio.shield(self.rooms.ready), timeout)
599 except TimeoutError:
600 raise XMPPError(
601 "recipient-unavailable",
602 "Legacy session is not fully initialized, retry later",
603 )
605 def legacy_module_data_update(self, data: dict[str, object]) -> None:
606 warnings.warn(
607 "Session.legacy_module_data_update() is deprecated. "
608 "Simply use Session.legacy_module_data as if it were a normal dict.",
609 FutureWarning,
610 stacklevel=2,
611 )
612 self.legacy_module_data.update(data)
614 def legacy_module_data_set(self, data: dict[str, object]) -> None:
615 warnings.warn(
616 "Session.legacy_module_data_set() is deprecated. "
617 "Simply use Session.legacy_module_data as if it were a normal dict.",
618 FutureWarning,
619 stacklevel=2,
620 )
621 self.legacy_module_data.clear()
622 self.legacy_module_data.update(data)
624 def legacy_module_data_clear(self) -> None:
625 warnings.warn(
626 "Session.legacy_module_data_clear() is deprecated. "
627 "Simply use Session.legacy_module_data as if it were a normal dict.",
628 FutureWarning,
629 stacklevel=2,
630 )
631 self.legacy_module_data.clear()
634# keys = user.jid.bare
635_sessions: dict[str, BaseSession[Any]] = {}
636log = logging.getLogger(__name__)