Coverage for slidge/contact/contact.py: 89%
272 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-29 05:05 +0000
1import datetime
2import logging
3import warnings
4from collections.abc import Iterable, Iterator, Sequence
5from datetime import date
6from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self
8import sqlalchemy as sa
9from slixmpp import JID, Message, Presence
10from slixmpp.exceptions import IqError, IqTimeout
11from slixmpp.plugins.xep_0292.stanza import VCard4
12from slixmpp.types import MessageTypes
14from slidge.db.avatar import CachedAvatar
16from ..core.mixins import AvatarMixin, FullCarbonMixin
17from ..core.mixins.disco import ContactAccountDiscoMixin
18from ..core.mixins.recipient import RecipientMixin
19from ..db.models import Contact, ContactSent
20from ..util.types import (
21 AnySession,
22 ClientType,
23 ContactMessage,
24 ContactSticker,
25 HoleBound,
26 MessageOrPresenceTypeVar,
27)
29if TYPE_CHECKING:
30 from ..command.base import ContactCommand
31 from ..group.participant import LegacyParticipant
34class LegacyContact(
35 AvatarMixin,
36 ContactAccountDiscoMixin,
37 FullCarbonMixin,
38 RecipientMixin,
39):
40 """
41 This class centralizes actions in relation to a specific legacy contact.
43 You shouldn't create instances of contacts manually, but rather rely on
44 :meth:`.LegacyRoster.by_legacy_id` to ensure that contact instances are
45 singletons. The :class:`.LegacyRoster` instance of a session is accessible
46 through the :attr:`.BaseSession.contacts` attribute.
48 Typically, your plugin should have methods hook to the legacy events and
49 call appropriate methods here to transmit the "legacy action" to the xmpp
50 user. This should look like this:
52 .. code-block:python
54 class Session(BaseSession):
55 ...
57 async def on_cool_chat_network_new_text_message(self, legacy_msg_event):
58 contact = self.contacts.by_legacy_id(legacy_msg_event.from)
59 contact.send_text(legacy_msg_event.text)
61 async def on_cool_chat_network_new_typing_event(self, legacy_typing_event):
62 contact = self.contacts.by_legacy_id(legacy_msg_event.from)
63 contact.composing()
64 ...
66 Use ``carbon=True`` as a keyword arg for methods to represent an action FROM
67 the user TO the contact, typically when the user uses an official client to
68 do an action such as sending a message or marking as message as read.
69 This will use :xep:`0363` to impersonate the XMPP user in order.
70 """
72 RESOURCE: str = "slidge"
73 """
74 A full JID, including a resource part is required for chat states (and maybe other stuff)
75 to work properly. This is the name of the resource the contacts will use.
76 """
77 PROPAGATE_PRESENCE_TO_GROUPS = True
79 mtype: MessageTypes = "chat"
80 _can_send_carbon = True
81 is_participant: Literal[False] = False
82 is_group: Literal[False] = False
84 _ONLY_SEND_PRESENCE_CHANGES = True
86 STRIP_SHORT_DELAY = True
87 _NON_FRIEND_PRESENCES_FILTER: ClassVar[set[str]] = {"subscribe", "unsubscribed"}
89 INVITATION_RECIPIENT = True
91 commands: ClassVar[dict[str, "type[ContactCommand[LegacyContact]]"]] = {}
92 commands_chat: ClassVar[dict[str, "type[ContactCommand[LegacyContact]]"]] = {}
94 stored: Contact
95 model: Contact
97 def __init__(self, session: AnySession, stored: Contact) -> None:
98 self.session = session
99 self.xmpp = session.xmpp
100 self.stored = stored
101 self._set_logger()
102 super().__init__()
104 def _recipient_pk(self) -> int:
105 return self.stored.id
107 async def on_message(self, message: ContactMessage) -> str | None:
108 """
109 Triggered when the user sends a message to this :term:`Contact`.
111 :return: A message ID of that can be used later to further reference
112 this message (reactions, read marks, etc.).
113 """
114 raise NotImplementedError
116 async def on_sticker(self, sticker: ContactSticker) -> str | None:
117 """
118 Triggered when the user sends a sticker to this :term:`Contact`.
120 :param sticker: The sticker sent by the user.
122 :return: A message ID of that can be used later to further reference
123 this message (reactions, read marks, etc.).
124 """
125 raise NotImplementedError
127 @property
128 def jid(self) -> JID:
129 jid = JID(self.stored.jid)
130 jid.resource = self.RESOURCE
131 return jid
133 @jid.setter
134 def jid(self, _jid: JID) -> None:
135 raise RuntimeError
137 @property
138 def legacy_id(self) -> str:
139 return self.stored.legacy_id
141 async def get_vcard(self, fetch: bool = True) -> VCard4:
142 if fetch and not self.stored.vcard_fetched:
143 await self.fetch_vcard()
144 with self.xmpp.store.session() as orm:
145 orm.add(self.stored)
146 return self.stored.vcard(self.xmpp.boundjid.bare)
148 @property
149 def is_friend(self) -> bool:
150 return self.stored.is_friend
152 @is_friend.setter
153 def is_friend(self, value: bool) -> None:
154 if value == self.is_friend:
155 return
156 self.update_stored_attribute(is_friend=value)
158 @property
159 def added_to_roster(self) -> bool:
160 return self.stored.added_to_roster
162 @added_to_roster.setter
163 def added_to_roster(self, value: bool) -> None:
164 if value == self.added_to_roster:
165 return
166 self.update_stored_attribute(added_to_roster=value)
168 @property
169 def participants(self) -> Iterator["LegacyParticipant[Self]"]:
170 with self.xmpp.store.session() as orm:
171 self.stored = orm.merge(self.stored)
172 participants = self.stored.participants
173 for p in participants:
174 with self.xmpp.store.session() as orm:
175 p = orm.merge(p)
176 muc = self.session.bookmarks.from_store(p.room)
177 part = muc.participant_from_store(p, contact=self)
178 yield part
180 @property # type:ignore
181 def DISCO_TYPE(self) -> ClientType:
182 return self.client_type
184 @DISCO_TYPE.setter
185 def DISCO_TYPE(self, value: ClientType) -> None:
186 self.client_type = value
188 @property
189 def client_type(self) -> ClientType:
190 """
191 The client type of this contact, cf https://xmpp.org/registrar/disco-categories.html#client
193 Default is "pc".
194 """
195 return self.stored.client_type
197 @client_type.setter
198 def client_type(self, value: ClientType) -> None:
199 if self.stored.client_type == value:
200 return
201 self.update_stored_attribute(client_type=value)
203 def _set_logger(self) -> None:
204 self.log = logging.getLogger(f"{self.user_jid.bare}:contact:{self}")
206 def __repr__(self) -> str:
207 return f"<Contact #{self.stored.id} '{self.name}' ({self.legacy_id} - {self.jid.user})'>"
209 def __get_subscription_string(self) -> str:
210 if self.is_friend:
211 return "both"
212 return "none"
214 def __propagate_to_participants(self, stanza: Presence) -> None:
215 if not self.PROPAGATE_PRESENCE_TO_GROUPS:
216 return
218 ptype = stanza["type"]
219 if ptype in ("available", "chat"):
220 func_name = "online"
221 elif ptype in ("xa", "unavailable"):
222 # we map unavailable to extended_away, because offline is
223 # "participant leaves the MUC"
224 # TODO: improve this with a clear distinction between participant
225 # and member list
226 func_name = "extended_away"
227 elif ptype == "busy":
228 func_name = "busy"
229 elif ptype == "away":
230 func_name = "away"
231 else:
232 return
234 last_seen: datetime.datetime | None = (
235 stanza["idle"]["since"] if "idle" in stanza else None
236 )
238 kw = {"status": stanza["status"], "last_seen": last_seen}
240 for part in self.participants:
241 func = getattr(part, func_name)
242 func(**kw)
244 def _send(
245 self,
246 stanza: MessageOrPresenceTypeVar,
247 carbon: bool = False,
248 nick: bool = False,
249 **send_kwargs: Any, # noqa:ANN401
250 ) -> MessageOrPresenceTypeVar:
251 if carbon and isinstance(stanza, Message):
252 stanza["to"] = self.jid.bare
253 stanza["from"] = self.user_jid
254 self._privileged_send(stanza)
255 return stanza
257 if isinstance(stanza, Presence):
258 if not self._updating_info:
259 self.__propagate_to_participants(stanza)
260 if (
261 not self.is_friend
262 and stanza["type"] not in self._NON_FRIEND_PRESENCES_FILTER
263 ):
264 return stanza
265 if self.name and (nick or not self.is_friend):
266 n = self.xmpp.plugin["xep_0172"].stanza.UserNick()
267 n["nick"] = self.name
268 stanza.append(n)
269 if (
270 not self._updating_info
271 and self.xmpp.MARK_ALL_MESSAGES
272 and is_markable(stanza)
273 ):
274 with self.xmpp.store.session(expire_on_commit=False) as orm:
275 self.stored = orm.merge(self.stored)
276 exists = (
277 orm.query(ContactSent)
278 .filter_by(contact_id=self.stored.id, msg_id=stanza["id"])
279 .first()
280 )
281 if exists:
282 self.log.warning(
283 "Contact has already sent message %s", stanza["id"]
284 )
285 else:
286 new = ContactSent(contact=self.stored, msg_id=stanza["id"])
287 orm.add(new)
288 self.stored.sent_order.append(new)
289 orm.commit()
290 stanza["to"] = self.user_jid
291 stanza.send()
292 return stanza
294 def _store_last_sent_msg(
295 self, legacy_id: str, when: datetime.datetime | None
296 ) -> None:
297 with self.xmpp.store.session(expire_on_commit=False) as orm:
298 orm.execute(
299 sa.update(Contact)
300 .where(Contact.id == self._recipient_pk())
301 .values(
302 last_sent_msg_legacy_id=legacy_id,
303 last_sent_msg_date=when or datetime.datetime.now(tz=datetime.UTC),
304 )
305 )
306 orm.commit()
308 def pop_unread_xmpp_ids_up_to(self, horizon_xmpp_id: str) -> list[str]:
309 """
310 Return XMPP msg ids sent by this contact up to a given XMPP msg id.
312 Legacy modules have no reason to use this, but it is used by slidge core
313 for legacy networks that need to mark all messages as read (most XMPP
314 clients only send a read marker for the latest message).
316 This has side effects, if the horizon XMPP id is found, messages up to
317 this horizon are cleared, to avoid sending the same read mark twice.
319 :param horizon_xmpp_id: The latest message
320 :return: A list of XMPP ids up to horizon_xmpp_id, included
321 """
322 with self.xmpp.store.session() as orm:
323 assert self.stored.id is not None
324 ids = self.xmpp.store.contacts.pop_sent_up_to(
325 orm, self.stored.id, horizon_xmpp_id
326 )
327 orm.commit()
328 return ids
330 @property
331 def name(self) -> str:
332 """
333 Friendly name of the contact, as it should appear in the user's roster
334 """
335 return self.stored.nick or ""
337 @name.setter
338 def name(self, n: str | None) -> None:
339 if self.stored.nick == n:
340 return
341 self.update_stored_attribute(nick=n)
342 self._set_logger()
343 if self.is_friend and self.added_to_roster:
344 self.xmpp.pubsub.broadcast_nick(
345 user_jid=self.user_jid, jid=self.jid.bare, nick=n
346 )
347 for p in self.participants:
348 p.nickname = n or str(self.legacy_id)
350 def _post_avatar_update(self, cached_avatar: CachedAvatar | None) -> None:
351 if self.is_friend and self.added_to_roster:
352 self.session.create_task(
353 self.session.xmpp.pubsub.broadcast_avatar(
354 self.jid.bare, self.session.user_jid, cached_avatar
355 ),
356 name=f"Post avatar update of {self}",
357 )
358 for p in self.participants:
359 self.log.debug("Propagating new avatar to %s", p.muc)
360 p.send_last_presence(force=True, no_cache_online=True)
362 def set_vcard(
363 self,
364 /,
365 full_name: str | None = None,
366 given: str | None = None,
367 surname: str | None = None,
368 birthday: date | None = None,
369 phone: str | None = None,
370 phones: Iterable[str] = (),
371 note: str | None = None,
372 url: str | None = None,
373 email: str | None = None,
374 country: str | None = None,
375 locality: str | None = None,
376 pronouns: str | None = None,
377 ) -> None:
378 """
379 Update xep:`0292` data for this contact.
381 Use this for additional metadata about this contact to be available to XMPP
382 clients. The "note" argument is a text of arbitrary size and can be useful when
383 no other field is a good fit.
384 """
385 if given:
386 warnings.warn(
387 "Contact.set_vcard() 'given' arg is deprecated",
388 FutureWarning,
389 stacklevel=2,
390 )
391 if surname:
392 warnings.warn(
393 "Contact.set_vcard() 'surname' arg is deprecated",
394 FutureWarning,
395 stacklevel=2,
396 )
397 with self.xmpp.store.session(expire_on_commit=False) as orm:
398 orm.add(self.stored)
399 self.stored.full_name = full_name
400 self.stored.birthday = birthday
401 self.stored.note = note
402 self.stored.url = url
403 self.stored.email = email
404 self.stored.phones = [x for x in [phone, *phones] if x]
405 self.stored.locality = locality
406 self.stored.pronouns = pronouns
407 self.stored.country = country
408 self.stored.vcard_fetched = True
409 orm.commit()
411 self.session.create_task(
412 self.xmpp.pubsub.broadcast_vcard_event(
413 self.jid, self.user_jid, self.stored.vcard(self.xmpp.boundjid.bare)
414 ),
415 name=f"Broadcast vcard of {self}",
416 )
418 def get_roster_item(self) -> dict[str, dict[str, str | Sequence[str]]]:
419 item = {
420 "subscription": self.__get_subscription_string(),
421 "groups": [self.xmpp.ROSTER_GROUP],
422 }
423 if (n := self.name) is not None:
424 item["name"] = n
425 return {self.jid.bare: item}
427 async def add_to_roster(self, force: bool = False) -> None:
428 """
429 Add this contact to the user roster using :xep:`0356`
431 :param force: add even if the contact was already added successfully
432 """
433 if self.added_to_roster and not force:
434 return
435 if not self.session.user.preferences.get("roster_push", True):
436 log.debug("Roster push request by plugin ignored (--no-roster-push)")
437 return
438 try:
439 await self.xmpp.plugin["xep_0356"].set_roster(
440 jid=self.user_jid, roster_items=self.get_roster_item()
441 )
442 except PermissionError:
443 warnings.warn(
444 f"Slidge does not have the privilege (XEP-0356) to manage the roster of {self.user_jid}. "
445 "If this is a local user, consider configuring your XMPP server for that."
446 )
447 self.send_friend_request(
448 f"I'm already your friend on {self.xmpp.COMPONENT_TYPE}, but "
449 "slidge is not allowed to manage your roster."
450 )
451 return
452 except (IqError, IqTimeout) as e:
453 self.log.warning("Could not add to roster", exc_info=e)
454 else:
455 # we only broadcast pubsub events for contacts added to the roster
456 # so if something was set before, we need to push it now
457 self.added_to_roster = True
458 self.send_last_presence(force=True)
460 async def __broadcast_pubsub_items(self) -> None:
461 if not self.is_friend:
462 return
463 if not self.added_to_roster:
464 return
465 cached_avatar = self.get_cached_avatar()
466 if cached_avatar is not None:
467 await self.xmpp.pubsub.broadcast_avatar(
468 self.jid.bare, self.session.user_jid, cached_avatar
469 )
470 nick = self.name
472 if nick is not None:
473 self.xmpp.pubsub.broadcast_nick(
474 self.session.user_jid,
475 self.jid.bare,
476 nick,
477 )
479 def send_friend_request(self, text: str | None = None) -> None:
480 presence = self._make_presence(ptype="subscribe", pstatus=text, bare=True)
481 self._send(presence, nick=True)
483 async def accept_friend_request(self, text: str | None = None) -> None:
484 """
485 Call this to signify that this Contact has accepted to be a friend
486 of the user.
488 :param text: Optional message from the friend to the user
489 """
490 self.is_friend = True
491 self.added_to_roster = True
492 self.log.debug("Accepting friend request")
493 presence = self._make_presence(ptype="subscribed", pstatus=text, bare=True)
494 self._send(presence, nick=True)
495 self.send_last_presence()
496 await self.__broadcast_pubsub_items()
497 self.log.debug("Accepted friend request")
499 def reject_friend_request(self, text: str | None = None) -> None:
500 """
501 Call this to signify that this Contact has refused to be a contact
502 of the user (or that they don't want to be friends anymore)
504 :param text: Optional message from the non-friend to the user
505 """
506 presence = self._make_presence(ptype="unsubscribed", pstatus=text, bare=True)
507 self.offline()
508 self._send(presence, nick=True)
509 self.is_friend = False
511 async def on_friend_request(self, text: str = "") -> None:
512 """
513 Called when receiving a "subscribe" presence, ie, "I would like to add
514 you to my contacts/friends", from the user to this contact.
516 In XMPP terms: "I would like to receive your presence updates"
518 This is only called if self.is_friend = False. If self.is_friend = True,
519 slidge will automatically "accept the friend request", ie, reply with
520 a "subscribed" presence.
522 When called, a 'friend request event' should be sent to the legacy
523 service, and when the contact responds, you should either call
524 self.accept_subscription() or self.reject_subscription()
525 """
527 async def on_friend_delete(self, text: str = "") -> None:
528 """
529 Called when receiving an "unsubscribed" presence, ie, "I would like to
530 remove you to my contacts/friends" or "I refuse your friend request"
531 from the user to this contact.
533 In XMPP terms: "You won't receive my presence updates anymore (or you
534 never have)".
535 """
537 async def on_friend_accept(self) -> None:
538 """
539 Called when receiving a "subscribed" presence, ie, "I accept to be
540 your/confirm that you are my friend" from the user to this contact.
542 In XMPP terms: "You will receive my presence updates".
543 """
545 def unsubscribe(self) -> None:
546 """
547 (internal use by slidge)
549 Send an "unsubscribe", "unsubscribed", "unavailable" presence sequence
550 from this contact to the user, ie, "this contact has removed you from
551 their 'friends'".
552 """
553 for ptype in "unsubscribe", "unsubscribed", "unavailable":
554 self.xmpp.send_presence(pfrom=self.jid, pto=self.user_jid.bare, ptype=ptype)
556 async def update_info(self) -> None:
557 """
558 Fetch information about this contact from the legacy network
560 This is awaited on Contact instantiation, and should be overridden to
561 update the nickname, avatar, vcard [...] of this contact, by making
562 "legacy API calls".
564 To take advantage of the slidge avatar cache, you can check the .avatar
565 property to retrieve the "legacy file ID" of the cached avatar. If there
566 is no change, you should not call
567 :py:meth:`slidge.core.mixins.avatar.AvatarMixin.set_avatar` or attempt
568 to modify the ``.avatar`` property.
570 :raises XMPPError: MUST be raised when the legacy contact does not exist.
571 """
573 async def fetch_vcard(self) -> None:
574 """
575 It the legacy network doesn't like that you fetch too many profiles on startup,
576 it's also possible to fetch it here, which will be called when XMPP clients
577 of the user request the vcard, if it hasn't been fetched before
578 :return:
579 """
581 def _make_presence(
582 self,
583 *,
584 last_seen: datetime.datetime | None = None,
585 **presence_kwargs: Any, # noqa:ANN401
586 ) -> Presence:
587 p = super()._make_presence(last_seen=last_seen, **presence_kwargs)
588 caps = self.xmpp.plugin["xep_0115"]
590 if p.get_from().resource and (ver := self.xmpp.get_caps_ver(self.client_type)):
591 p["caps"]["node"] = caps.caps_node
592 p["caps"]["hash"] = caps.hash
593 p["caps"]["ver"] = ver
594 return p
596 async def backfill(self, after: HoleBound | None) -> None:
597 """
598 This method can be overridden to implement history fetching for this
599 contact.
601 Since the message archive (:xep:`0313`) is managed by the XMPP server
602 of the user, there are several caveats.
604 - We cannot prevent the XMPP server from injecting a :xep:`0203`
605 timestamp at the time it receives the messages. Most XMPP clients
606 will use that timestamp instead of the one we inject and thus message
607 ordering may be messed up under certain circumstances.
608 - We cannot know if a message is already in this archive, so legacy
609 modules should ensure that this does not send message that have
610 already sent before or there will be duplicates.
611 - Related to the previous point, if a legacy network APIs returns a
612 message while iterating over a "fetch history" call and also pass
613 it as a live message, you may end up with duplicates.
615 For these reasons, we recommend sending only recent messages, mostly
616 messages that could have been missed when slidge was down.
618 NB: if the legacy client receives "message while it was down" as live
619 messages on startup, this is pretty much useless.
621 :param after: Last message to or from this contact that slidge saw
622 passing through. If ``None``, it means slidge never saw any message
623 from this contact.
624 """
625 raise NotImplementedError
628def is_markable(stanza: Message | Presence) -> bool:
629 if isinstance(stanza, Presence):
630 return False
631 return bool(stanza["body"])
634log = logging.getLogger(__name__)