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