Coverage for slidge/core/session.py: 83%
229 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 18:29 +0000
1import abc
2import asyncio
3import logging
4from asyncio.tasks import Task
5from collections.abc import Coroutine
6from typing import TYPE_CHECKING, Any, Generic, NamedTuple, Self, cast
8import aiohttp
9import sqlalchemy as sa
10from slixmpp import JID, Iq, Message, Presence
11from slixmpp.exceptions import XMPPError
12from slixmpp.types import PresenceShows, ResourceDict
14from slidge.db.meta import JSONSerializable
16from ..command import SearchResult
17from ..contact import LegacyContact
18from ..db.models import Contact, GatewayUser
19from ..util import SubclassableOnce
20from ..util.lock import NamedLockMixin
21from ..util.types import (
22 AnyBookmarks,
23 AnyMUC,
24 AnyParticipant,
25 AnyRoster,
26 AnySession,
27 LegacyContactType,
28 PseudoPresenceShow,
29)
30from ..util.util import noop_coro
32if TYPE_CHECKING:
33 from .gateway import BaseGateway
36class CachedPresence(NamedTuple):
37 status: str | None
38 show: str | None
39 kwargs: dict[str, Any]
42class BaseSession(
43 NamedLockMixin, SubclassableOnce, abc.ABC, Generic[LegacyContactType]
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: "BaseGateway[Self]"
64 """
65 The gateway instance singleton. Use it for low-level XMPP calls or custom methods that are not
66 session-specific.
67 """
69 MESSAGE_IDS_ARE_THREAD_IDS = False
70 """
71 Set this to True if the legacy service uses message IDs as thread IDs,
72 eg Mattermost, where you can only 'create a thread' by replying to the message,
73 in which case the message ID is also a thread ID (and all messages are potential
74 threads).
75 """
76 SPECIAL_MSG_ID_PREFIX: str | None = None
77 """
78 If you set this, XMPP message IDs starting with this won't be converted to legacy ID,
79 but passed as is to :meth:`LegacyContact.on_react`, and usual checks for emoji restriction won't be
80 applied.
81 This can be used to implement voting in polls in a hacky way.
82 """
84 _roster_cls: type[AnyRoster]
85 _bookmarks_cls: type[AnyBookmarks]
87 def __init__(self, user: GatewayUser) -> None:
88 super().__init__()
89 self.user = user
90 """
91 The :term:`slidge user <User>`.
92 """
93 self.log = logging.getLogger(user.jid.bare)
95 self.ignore_messages = set[str]()
97 self.contacts: AnyRoster = self._roster_cls(self)
98 self.is_logging_in = False
99 self._logged = False
100 self.__reset_ready()
102 self.bookmarks = self._bookmarks_cls(self)
104 self.thread_creation_lock = asyncio.Lock()
106 self.__cached_presence: CachedPresence | None = None
108 self.__tasks = set[asyncio.Task[Any]]()
110 @property
111 def user_jid(self) -> JID:
112 return self.user.jid
114 @property
115 def user_pk(self) -> int:
116 return self.user.id
118 @property
119 def http(self) -> aiohttp.ClientSession:
120 return self.xmpp.http
122 def __remove_task(self, fut: Task[Any]) -> None:
123 self.log.debug("Removing fut %s", fut)
124 self.__tasks.remove(fut)
126 def create_task(
127 self, coro: Coroutine[Any, Any, Any], name: str | None = None
128 ) -> asyncio.Task[Any]:
129 task = self.xmpp.loop.create_task(coro, name=name)
130 self.__tasks.add(task)
131 self.log.debug("Creating task %s", task)
132 task.add_done_callback(lambda _: self.__remove_task(task))
133 return task
135 def cancel_all_tasks(self) -> None:
136 for task in self.__tasks:
137 task.cancel()
139 @abc.abstractmethod
140 async def login(self) -> str | None:
141 """
142 Logs in the gateway user to the legacy network.
144 Triggered when the gateway start and on user registration.
145 It is recommended that this function returns once the user is logged in,
146 so if you need to await forever (for instance to listen to incoming events),
147 it's a good idea to wrap your listener in an asyncio.Task.
149 :return: Optionally, a text to use as the gateway status, e.g., "Connected as 'dude@legacy.network'"
150 """
151 raise NotImplementedError
153 async def logout(self) -> None:
154 """
155 Logs out the gateway user from the legacy network.
157 Called on gateway shutdown.
158 """
159 raise NotImplementedError
161 async def on_presence(
162 self,
163 resource: str,
164 show: PseudoPresenceShow,
165 status: str,
166 resources: dict[str, ResourceDict],
167 merged_resource: ResourceDict | None,
168 ) -> None:
169 """
170 Called when the gateway component receives a presence, ie, when
171 one of the user's clients goes online of offline, or changes its
172 status.
174 :param resource: The XMPP client identifier, arbitrary string.
175 :param show: The presence ``<show>``, if available. If the resource is
176 just 'available' without any ``<show>`` element, this is an empty
177 str.
178 :param status: A status message, like a deeply profound quote, eg,
179 "Roses are red, violets are blue, [INSERT JOKE]".
180 :param resources: A summary of all the resources for this user.
181 :param merged_resource: A global presence for the user account,
182 following rules described in :meth:`merge_resources`
183 """
184 raise NotImplementedError
186 async def on_search(self, form_values: dict[str, str]) -> SearchResult | None:
187 """
188 Triggered when the user uses Jabber Search (:xep:`0055`) on the component
190 Form values is a dict in which keys are defined in :attr:`.BaseGateway.SEARCH_FIELDS`
192 :param form_values: search query, defined for a specific plugin by overriding
193 in :attr:`.BaseGateway.SEARCH_FIELDS`
194 :return:
195 """
196 raise NotImplementedError
198 async def on_avatar(
199 self,
200 bytes_: bytes | None,
201 hash_: str | None,
202 type_: str | None,
203 width: int | None,
204 height: int | None,
205 ) -> None:
206 """
207 Triggered when the user uses modifies their avatar via :xep:`0084`.
209 :param bytes_: The data of the avatar. According to the spec, this
210 should always be a PNG, but some implementations do not respect
211 that. If `None` it means the user has unpublished their avatar.
212 :param hash_: The SHA1 hash of the avatar data. This is an identifier of
213 the avatar.
214 :param type_: The MIME type of the avatar.
215 :param width: The width of the avatar image.
216 :param height: The height of the avatar image.
217 """
218 raise NotImplementedError
220 async def on_create_group(
221 self, name: str, contacts: list[LegacyContactType]
222 ) -> str:
223 """
224 Triggered when the user request the creation of a group via the
225 dedicated :term:`Command`.
227 :param name: Name of the group
228 :param contacts: list of contacts that should be members of the group
229 """
230 raise NotImplementedError
232 async def on_leave_space(self, space_legacy_id: str) -> None:
233 """
234 Triggered when the user sends a request to leave a :xep:`0503` space.
236 :param space_legacy_id: The legacy ID of the space to leave
237 """
238 raise NotImplementedError
240 async def on_preferences(
241 self, previous: dict[str, Any], new: dict[str, Any]
242 ) -> None:
243 """
244 This is called when the user updates their preferences.
246 Override this if you need set custom preferences field and need to trigger
247 something when a preference has changed.
248 """
249 raise NotImplementedError
251 def __reset_ready(self) -> None:
252 self.ready = self.xmpp.loop.create_future()
254 @property
255 def logged(self) -> bool:
256 return self._logged
258 @logged.setter
259 def logged(self, v: bool) -> None:
260 self.is_logging_in = False
261 self._logged = v
262 if self.ready.done():
263 if v:
264 return
265 self.__reset_ready()
266 self.shutdown(logout=False)
267 with self.xmpp.store.session() as orm:
268 self.xmpp.store.mam.reset_source(orm)
269 self.xmpp.store.rooms.reset_updated(orm)
270 self.xmpp.store.contacts.reset_updated(orm)
271 orm.commit()
272 else:
273 if v:
274 self.ready.set_result(True)
276 def __repr__(self) -> str:
277 return f"<Session of {self.user_jid}>"
279 def shutdown(self, logout: bool = True) -> asyncio.Task[None]:
280 for m in self.bookmarks:
281 m.shutdown()
282 with self.xmpp.store.session() as orm:
283 for localpart in orm.execute(
284 sa.select(Contact.jid_localpart).filter_by(
285 user=self.user, is_friend=True
286 )
287 ).scalars():
288 pres = self.xmpp.make_presence(
289 pfrom=f"{localpart}@{self.xmpp.boundjid.bare}",
290 pto=self.user_jid,
291 ptype="unavailable",
292 pstatus="Gateway has shut down.",
293 )
294 pres.send()
295 if logout:
296 return self.xmpp.loop.create_task(self.__logout())
297 else:
298 return self.xmpp.loop.create_task(noop_coro())
300 async def __logout(self) -> None:
301 try:
302 await self.logout()
303 except NotImplementedError:
304 pass
305 except KeyboardInterrupt:
306 pass
308 def raise_if_not_logged(self) -> None:
309 if not self.logged:
310 raise XMPPError(
311 "internal-server-error",
312 text="You are not logged to the legacy network",
313 )
315 @classmethod
316 def _from_user_or_none(cls, user: GatewayUser | None) -> Self:
317 if user is None:
318 log.debug("user not found")
319 raise XMPPError(text="User not found", condition="subscription-required")
321 session = _sessions.get(user.jid.bare)
322 if session is None:
323 _sessions[user.jid.bare] = session = cls(user)
324 assert isinstance(session, cls)
325 return session
327 @classmethod
328 def from_user(cls, user: GatewayUser) -> Self:
329 return cls._from_user_or_none(user)
331 @classmethod
332 def from_stanza(cls, s: Message | Iq | Presence) -> Self:
333 # """
334 # Get a user's :class:`.LegacySession` using the "from" field of a stanza
335 #
336 # Meant to be called from :class:`BaseGateway` only.
337 #
338 # :param s:
339 # :return:
340 # """
341 return cls.from_jid(s.get_from())
343 @classmethod
344 def from_jid(cls, jid: JID) -> Self:
345 # """
346 # Get a user's :class:`.LegacySession` using its jid
347 #
348 # Meant to be called from :class:`BaseGateway` only.
349 #
350 # :param jid:
351 # :return:
352 # """
353 session = _sessions.get(jid.bare)
354 if session is not None:
355 assert isinstance(session, cls)
356 return session
357 with cls.xmpp.store.session() as orm:
358 user = orm.query(GatewayUser).filter_by(jid=jid.bare).one_or_none()
359 return cls._from_user_or_none(user)
361 @classmethod
362 async def kill_by_jid(cls, jid: JID) -> None:
363 # """
364 # Terminate a user session.
365 #
366 # Meant to be called from :class:`BaseGateway` only.
367 #
368 # :param jid:
369 # :return:
370 # """
371 log.debug("Killing session of %s", jid)
372 for user_jid, session in _sessions.items():
373 if user_jid == jid.bare:
374 break
375 else:
376 log.debug("Did not find a session for %s", jid)
377 return
378 for c in session.contacts:
379 c.unsubscribe()
380 for m in session.bookmarks:
381 m.shutdown()
383 try:
384 session = _sessions.pop(jid.bare)
385 except KeyError:
386 log.warning("User not found during unregistration")
387 return
389 session.cancel_all_tasks()
391 await cls.xmpp.unregister(cast(Self, session))
392 with cls.xmpp.store.session() as orm:
393 orm.delete(session.user)
394 orm.commit()
396 def __ack(self, msg: Message) -> None:
397 if not self.xmpp.PROPER_RECEIPTS:
398 self.xmpp.delivery_receipt.ack(msg)
400 def send_gateway_status(
401 self,
402 status: str | None = None,
403 show: PresenceShows | None = None,
404 **kwargs: Any, # noqa
405 ) -> None:
406 """
407 Send a presence from the gateway to the user.
409 Can be used to indicate the user session status, ie "SMS code required", "connected", …
411 :param status: A status message
412 :param show: Presence stanza 'show' element. I suggest using "dnd" to show
413 that the gateway is not fully functional
414 """
415 self.__cached_presence = CachedPresence(status, show, kwargs)
416 self.xmpp.send_presence(
417 pto=self.user_jid.bare, pstatus=status, pshow=show, **kwargs
418 )
420 def send_cached_presence(self, to: JID) -> None:
421 if not self.__cached_presence:
422 self.xmpp.send_presence(pto=to, ptype="unavailable")
423 return
424 self.xmpp.send_presence(
425 pto=to,
426 pstatus=self.__cached_presence.status,
427 pshow=self.__cached_presence.show,
428 **self.__cached_presence.kwargs,
429 )
431 def send_gateway_message(
432 self,
433 text: str,
434 **msg_kwargs: Any, # noqa
435 ) -> None:
436 """
437 Send a message from the gateway component to the user.
439 Can be used to indicate the user session status, ie "SMS code required", "connected", …
441 :param text: A text
442 """
443 self.xmpp.send_text(text, mto=self.user_jid, **msg_kwargs)
445 def send_gateway_invite(
446 self,
447 muc: AnyMUC,
448 reason: str | None = None,
449 password: str | None = None,
450 ) -> None:
451 """
452 Send an invitation to join a MUC, emanating from the gateway component.
454 :param muc:
455 :param reason:
456 :param password:
457 """
458 self.xmpp.invite_to(muc, reason=reason, password=password, mto=self.user_jid)
460 async def input(self, text: str, **msg_kwargs: Any) -> str: # noqa
461 """
462 Request user input via direct messages from the gateway component.
464 Wraps call to :meth:`.BaseSession.input`
466 :param text: The prompt to send to the user
467 :param msg_kwargs: Extra attributes
468 :return:
469 """
470 return await self.xmpp.input(self.user_jid, text, **msg_kwargs)
472 async def send_qr(self, text: str) -> None:
473 """
474 Sends a QR code generated from 'text' via HTTP Upload and send the URL to
475 ``self.user``
477 :param text: Text to encode as a QR code
478 """
479 await self.xmpp.send_qr(text, mto=self.user_jid)
481 async def get_contact_or_group_or_participant(
482 self, jid: JID, create: bool = True
483 ) -> "LegacyContact | AnyMUC | AnyParticipant | None":
484 if (contact := self.contacts.by_jid_only_if_exists(jid)) is not None:
485 return contact # type:ignore[no-any-return]
486 if (muc := self.bookmarks.by_jid_only_if_exists(JID(jid.bare))) is not None:
487 return await self.__get_muc_or_participant(muc, jid)
488 else:
489 muc = None
491 if not create:
492 return None
494 try:
495 return await self.contacts.by_jid(jid) # type:ignore[no-any-return]
496 except XMPPError:
497 if muc is None:
498 try:
499 muc = await self.bookmarks.by_jid(jid)
500 except XMPPError:
501 return None
502 return await self.__get_muc_or_participant(muc, jid)
504 @staticmethod
505 async def __get_muc_or_participant(
506 muc: AnyMUC, jid: JID
507 ) -> "AnyMUC | AnyParticipant | None":
508 if nick := jid.resource:
509 return await muc.get_participant(nick, create=False, fill_first=True)
510 return muc
512 async def wait_for_ready(self, timeout: float | None = 10) -> None:
513 # """
514 # Wait until session, contacts and bookmarks are ready
515 #
516 # (slidge internal use)
517 #
518 # :param timeout:
519 # :return:
520 # """
521 try:
522 await asyncio.wait_for(asyncio.shield(self.ready), timeout)
523 await asyncio.wait_for(asyncio.shield(self.contacts.ready), timeout)
524 await asyncio.wait_for(asyncio.shield(self.bookmarks.ready), timeout)
525 except TimeoutError:
526 raise XMPPError(
527 "recipient-unavailable",
528 "Legacy session is not fully initialized, retry later",
529 )
531 def legacy_module_data_update(self, data: JSONSerializable) -> None:
532 user = self.user
533 user.legacy_module_data.update(data)
534 self.xmpp.store.users.update(user)
536 def legacy_module_data_set(self, data: JSONSerializable) -> None:
537 user = self.user
538 user.legacy_module_data = data
539 self.xmpp.store.users.update(user)
541 def legacy_module_data_clear(self) -> None:
542 user = self.user
543 user.legacy_module_data.clear()
544 self.xmpp.store.users.update(user)
547# keys = user.jid.bare
548_sessions: dict[str, AnySession] = {}
549log = logging.getLogger(__name__)