Coverage for slidge/group/participant.py: 87%
370 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 18:29 +0000
1import logging
2import string
3import uuid
4import warnings
5from copy import copy
6from datetime import datetime
7from typing import TYPE_CHECKING, Any, Generic, Literal
8from xml.etree import ElementTree as ET
10import sqlalchemy as sa
11from slixmpp import JID, InvalidJID, Message, Presence
12from slixmpp.plugins.xep_0030.stanza.info import DiscoInfo
13from slixmpp.plugins.xep_0045.stanza import MUCAdminItem
14from slixmpp.plugins.xep_0492.stanza import Never
15from slixmpp.types import MessageTypes, OptJid
16from sqlalchemy.orm.exc import DetachedInstanceError
18from ..core.mixins import ChatterDiscoMixin, MessageMixin, PresenceMixin
19from ..core.mixins.db import DBMixin
20from ..db.models import Participant
21from ..util import SubclassableOnce, strip_illegal_chars
22from ..util.types import (
23 AnyMUC,
24 CachedPresence,
25 Hat,
26 LegacyContactType,
27 MessageOrPresenceTypeVar,
28 MucAffiliation,
29 MucRole,
30)
32if TYPE_CHECKING:
33 from slidge.command.base import ContactCommand
36def strip_non_printable(nickname: str) -> str:
37 new = (
38 "".join(x for x in nickname if x in string.printable)
39 + f"-slidge-{hash(nickname)}"
40 )
41 warnings.warn(f"Could not use {nickname} as a nickname, using {new}")
42 return new
45class LegacyParticipant(
46 PresenceMixin,
47 MessageMixin,
48 ChatterDiscoMixin,
49 DBMixin,
50 SubclassableOnce,
51 Generic[LegacyContactType],
52):
53 """
54 A legacy participant of a legacy group chat.
55 """
57 is_participant: Literal[True] = True
59 mtype: MessageTypes = "groupchat"
60 _can_send_carbon = False
61 USE_STANZA_ID = True
62 STRIP_SHORT_DELAY = False
63 stored: Participant
64 contact: LegacyContactType | None
66 def __init__(
67 self,
68 muc: AnyMUC,
69 stored: Participant,
70 is_system: bool = False,
71 contact: LegacyContactType | None = None,
72 ) -> None:
73 self.muc = muc
74 self.session = muc.session
75 self.xmpp = muc.session.xmpp
76 self.is_system = is_system
78 if contact is None and stored.contact is not None:
79 contact = self.session.contacts.from_store(stored=stored.contact)
80 if contact is not None and stored.contact is None:
81 stored.contact = contact.stored
83 self.stored = stored
84 self.contact = contact
86 super().__init__()
88 if stored.resource is None:
89 self.__update_resource(stored.nickname)
91 self.log = logging.getLogger(f"{self.user_jid.bare}:{self.jid}")
93 def __eq__(self, other: object) -> bool:
94 return isinstance(other, LegacyParticipant) and self.jid == other.jid
96 @property
97 def is_user(self) -> bool:
98 try:
99 return self.stored.is_user
100 except DetachedInstanceError:
101 self.merge()
102 return self.stored.is_user
104 @is_user.setter
105 def is_user(self, is_user: bool) -> None:
106 with self.xmpp.store.session(expire_on_commit=True) as orm:
107 orm.add(self.stored)
108 self.stored.is_user = is_user
109 orm.commit()
111 @property
112 def jid(self) -> JID:
113 jid = JID(self.muc.jid)
114 if self.stored.resource:
115 jid.resource = self.stored.resource
116 return jid
118 @jid.setter
119 def jid(self, x: JID) -> None:
120 # FIXME: without this, mypy yields
121 # "Cannot override writeable attribute with read-only property"
122 # But it does not happen for LegacyContact. WTF?
123 raise RuntimeError
125 @property
126 def commands(self) -> dict[str, "type[ContactCommand[Any]]"]: # type:ignore[override]
127 if self.contact is None:
128 return {}
129 else:
130 return self.contact.commands
132 def __should_commit(self) -> bool:
133 if self.is_system:
134 return False
135 if self.muc.get_lock("fill participants"):
136 return False
137 return not self.muc.get_lock("fill history")
139 def commit(self) -> None:
140 if not self.__should_commit():
141 return
142 super().commit()
144 @property
145 def user_jid(self) -> JID:
146 return self.session.user_jid
148 def __repr__(self) -> str:
149 return f"<Participant '{self.nickname}'/'{self.jid}' of '{self.muc}'>"
151 @property
152 def _presence_sent(self) -> bool:
153 # we track if we already sent a presence for this participant.
154 # if we didn't, we send it before the first message.
155 # this way, event in plugins that don't map "user has joined" events,
156 # we send a "join"-presence from the participant before the first message
157 return self.stored.presence_sent
159 @_presence_sent.setter
160 def _presence_sent(self, val: bool) -> None:
161 if self._presence_sent == val:
162 return
163 self.stored.presence_sent = val
164 if not self.__should_commit():
165 return
166 with self.xmpp.store.session() as orm:
167 orm.execute(
168 sa.update(Participant)
169 .where(Participant.id == self.stored.id)
170 .values(presence_sent=val)
171 )
172 orm.commit()
174 @property
175 def nickname_no_illegal(self) -> str:
176 return self.stored.nickname_no_illegal
178 @property
179 def affiliation(self) -> MucAffiliation:
180 return self.stored.affiliation
182 @affiliation.setter
183 def affiliation(self, affiliation: MucAffiliation) -> None:
184 if self.affiliation == affiliation:
185 return
186 was = self.stored.affiliation
187 self.stored.affiliation = affiliation
188 if not self.muc.participants_filled:
189 return
190 self.commit()
191 if self.cached_presence is None or self.cached_presence.ptype == "unavailable":
192 self.muc.send_affiliation_change(self, was)
193 self.send_last_presence(force=True, no_cache_online=True)
195 @property
196 def role(self) -> MucRole:
197 return self.stored.role
199 @role.setter
200 def role(self, role: MucRole) -> None:
201 if self.role == role:
202 return
203 self.stored.role = role
204 if not self.muc.participants_filled:
205 return
206 self.commit()
207 if not self._presence_sent:
208 return
209 self.send_last_presence(force=True, no_cache_online=True)
211 @property
212 def hats(self) -> list[Hat]:
213 return [Hat(*h) for h in self.stored.hats] if self.stored.hats else []
215 def set_hats(self, hats: list[Hat]) -> None:
216 if self.hats == hats:
217 return
218 self.stored.hats = hats
219 if not self.muc.participants_filled:
220 return
221 self.commit()
222 if not self._presence_sent:
223 return
224 self.send_last_presence(force=True, no_cache_online=True)
226 def __update_resource(self, unescaped_nickname: str | None) -> None:
227 if not unescaped_nickname:
228 self.stored.resource = ""
229 if self.is_system:
230 self.stored.nickname_no_illegal = ""
231 else:
232 warnings.warn(
233 "Only the system participant is allowed to not have a nickname"
234 )
235 nickname = f"unnamed-{uuid.uuid4()}"
236 self.stored.resource = self.stored.nickname_no_illegal = nickname
237 return
239 self.stored.nickname_no_illegal, jid = escape_nickname(
240 self.muc.jid,
241 unescaped_nickname,
242 )
243 self.stored.resource = jid.resource
245 def send_configuration_change(self, codes: tuple[int, ...]) -> None:
246 if not self.is_system:
247 raise RuntimeError("This is only possible for the system participant")
248 msg = self._make_message()
249 msg["muc"]["status_codes"] = codes
250 self._send(msg)
252 @property
253 def nickname(self) -> str:
254 return self.stored.nickname
256 @nickname.setter
257 def nickname(self, new_nickname: str) -> None:
258 old = self.nickname
259 if new_nickname == old:
260 return
262 if self.muc.stored.id is not None:
263 with self.xmpp.store.session() as orm:
264 if not self.xmpp.store.rooms.nick_available(
265 orm, self.muc.stored.id, new_nickname
266 ):
267 if self.contact is None:
268 new_nickname = f"{new_nickname} ({self.occupant_id})"
269 else:
270 new_nickname = f"{new_nickname} ({self.contact.legacy_id})"
272 cache = getattr(self, "_last_presence", None)
273 if cache:
274 last_seen = cache.last_seen
275 kwargs = cache.presence_kwargs
276 else:
277 last_seen = None
278 kwargs = {}
280 kwargs["status_codes"] = {303}
282 p = self._make_presence(ptype="unavailable", last_seen=last_seen, **kwargs)
283 # in this order so pfrom=old resource and we actually use the escaped nick
284 # in the muc/item/nick element
285 self.__update_resource(new_nickname)
286 p["muc"]["item"]["nick"] = self.jid.resource
287 self._send(p)
289 self.stored.nickname = new_nickname
290 self.commit()
291 kwargs["status_codes"] = set()
292 p = self._make_presence(ptype="available", last_seen=last_seen, **kwargs)
293 self._send(p)
295 def _make_presence( # type:ignore[no-untyped-def]
296 self,
297 *,
298 last_seen: datetime | None = None,
299 status_codes: set[int] | None = None,
300 user_full_jid: JID | None = None,
301 **presence_kwargs, # noqa type:ignore[no-untyped-def]
302 ) -> Presence:
303 p = super()._make_presence(last_seen=last_seen, **presence_kwargs)
304 p["muc"]["affiliation"] = self.affiliation
305 p["muc"]["role"] = self.role
306 if self.hats:
307 p["hats"].add_hats(self.hats)
308 codes = status_codes or set()
309 if self.is_user:
310 codes.add(110)
311 if not self.muc.is_anonymous and not self.is_system:
312 if self.is_user:
313 if user_full_jid:
314 p["muc"]["jid"] = user_full_jid
315 else:
316 jid = JID(self.user_jid)
317 try:
318 jid.resource = next(iter(self.muc.get_user_resources()))
319 except StopIteration:
320 jid.resource = "pseudo-resource"
321 p["muc"]["jid"] = self.user_jid
322 codes.add(100)
323 elif self.contact:
324 p["muc"]["jid"] = self.contact.jid
325 if a := self.contact.get_avatar():
326 p["vcard_temp_update"]["photo"] = a.id
327 else:
328 warnings.warn(
329 f"Private group but no 1:1 JID associated to '{self}'",
330 )
331 if self.is_user and (hash_ := self.session.user.avatar_hash):
332 p["vcard_temp_update"]["photo"] = hash_
333 p["muc"]["status_codes"] = codes
334 return p
336 @property
337 def DISCO_NAME(self) -> str:
338 return self.nickname
340 @DISCO_NAME.setter
341 def DISCO_NAME(self, _: str) -> Never:
342 raise RuntimeError
344 def __send_presence_if_needed(
345 self, stanza: Message | Presence, full_jid: JID, archive_only: bool
346 ) -> None:
347 if (
348 archive_only
349 or self.is_system
350 or self.is_user
351 or self._presence_sent
352 or stanza["subject"]
353 ):
354 return
355 if isinstance(stanza, Message):
356 if "muc" in stanza:
357 return
358 self.send_initial_presence(full_jid)
360 @property
361 def occupant_id(self) -> str:
362 return self.stored.occupant_id
364 def _send(
365 self,
366 stanza: MessageOrPresenceTypeVar,
367 full_jid: JID | None = None,
368 archive_only: bool = False,
369 legacy_msg_id: str | None = None,
370 force: bool = False,
371 **send_kwargs: Any, # noqa:ANN401
372 ) -> MessageOrPresenceTypeVar:
373 if stanza.get_from().resource:
374 stanza["occupant-id"]["id"] = self.occupant_id
375 else:
376 stanza["occupant-id"]["id"] = "room"
377 self.__add_nick_element(stanza)
378 if not self.is_user and isinstance(stanza, Presence):
379 if (
380 not force
381 and stanza["type"] == "unavailable"
382 and not self._presence_sent
383 ):
384 return stanza
385 self._presence_sent = True
386 if full_jid:
387 stanza["to"] = full_jid
388 self.__send_presence_if_needed(stanza, full_jid, archive_only)
389 if self.is_user:
390 assert stanza.stream is not None
391 stanza.stream.send(stanza, use_filters=False)
392 else:
393 stanza.send()
394 else:
395 if hasattr(self.muc, "archive") and isinstance(stanza, Message):
396 self.muc.archive.add(stanza, self, archive_only, legacy_msg_id)
397 if archive_only:
398 return stanza
399 for user_full_jid in self.muc.user_full_jids():
400 stanza = copy(stanza)
401 stanza["to"] = user_full_jid
402 self.__send_presence_if_needed(stanza, user_full_jid, archive_only)
403 stanza.send()
404 return stanza
406 def mucadmin_item(self) -> MUCAdminItem:
407 item = MUCAdminItem()
408 item["nick"] = self.nickname
409 item["affiliation"] = self.affiliation
410 item["role"] = self.role
411 if not self.muc.is_anonymous:
412 if self.is_user:
413 item["jid"] = self.user_jid.bare
414 elif self.contact:
415 item["jid"] = self.contact.jid.bare
416 else:
417 warnings.warn(
418 (
419 f"Private group but no contact JID associated to {self.jid} in"
420 f" {self}"
421 ),
422 )
423 return item
425 def __add_nick_element(self, stanza: Presence | Message) -> None:
426 if (nick := self.nickname_no_illegal) != self.jid.resource:
427 n = self.xmpp.plugin["xep_0172"].stanza.UserNick()
428 n["nick"] = nick
429 stanza.append(n)
431 def _get_last_presence(self) -> CachedPresence | None:
432 own = super()._get_last_presence()
433 if own is None and self.contact:
434 return self.contact._get_last_presence()
435 return own
437 def send_initial_presence(
438 self,
439 full_jid: JID,
440 nick_change: bool = False,
441 presence_id: str | None = None,
442 mav_until: str | None = None,
443 ) -> None:
444 """
445 Called when the user joins a MUC, as a mechanism
446 to indicate to the joining XMPP client the list of "participants".
448 Can be called this to trigger a "participant has joined the group" event.
450 :param full_jid: Set this to only send to a specific user XMPP resource.
451 :param nick_change: Used when the user joins and the MUC renames them (code 210)
452 :param presence_id: set the presence ID. used internally by slidge
453 """
454 # MUC status codes: https://xmpp.org/extensions/xep-0045.html#registrar-statuscodes
455 codes = set()
456 if nick_change:
457 codes.add(210)
459 if self.is_user:
460 # the "initial presence" of the user has to be vanilla, as it is
461 # a crucial part of the MUC join sequence for XMPP clients.
462 kwargs = {}
463 else:
464 cache = self._get_last_presence()
465 self.log.debug("Join muc, initial presence: %s", cache)
466 if cache:
467 ptype = cache.ptype
468 if ptype == "unavailable":
469 return
470 kwargs = {
471 "last_seen": cache.last_seen,
472 "pstatus": cache.pstatus,
473 "pshow": cache.pshow,
474 }
475 else:
476 kwargs = {}
477 p = self._make_presence(
478 status_codes=codes,
479 user_full_jid=full_jid,
480 **kwargs, # type:ignore
481 )
482 if presence_id:
483 p["id"] = presence_id
484 if self.is_user and mav_until is not None:
485 p["muc"]["mav"]["until"] = mav_until
486 self._send(p, full_jid)
488 def leave(self) -> None:
489 """
490 Call this when the participant leaves the room
491 """
492 self.muc.remove_participant(self)
494 def kick(self, reason: str | None = None) -> None:
495 """
496 Call this when the participant is kicked from the room
497 """
498 self.muc.remove_participant(self, kick=True, reason=reason)
500 def ban(self, reason: str | None = None) -> None:
501 """
502 Call this when the participant is banned from the room
503 """
504 self.muc.remove_participant(self, ban=True, reason=reason)
506 async def get_disco_info(
507 self, jid: OptJid = None, node: str | None = None
508 ) -> DiscoInfo:
509 if self.contact is not None:
510 return await self.contact.get_disco_info()
511 return await super().get_disco_info()
513 def moderate(self, legacy_msg_id: str, reason: str | None = None) -> None:
514 for i in self._legacy_to_xmpp(legacy_msg_id):
515 m = self.muc.get_system_participant()._make_message()
516 m["retract"]["id"] = i
517 if self.is_system:
518 m["retract"].enable("moderated")
519 else:
520 m["retract"]["moderated"]["by"] = self.jid
521 m["retract"]["moderated"]["occupant-id"]["id"] = self.occupant_id
522 if reason:
523 m["retract"]["reason"] = reason
524 self._send(m)
526 def set_room_subject(
527 self,
528 subject: str,
529 full_jid: JID | None = None,
530 when: datetime | None = None,
531 update_muc: bool = True,
532 ) -> None:
533 if update_muc:
534 self.muc._subject = subject # type: ignore
535 self.muc.subject_setter = self.nickname
536 self.muc.subject_date = when
538 msg = self._make_message()
539 if when is not None:
540 msg["delay"].set_stamp(when)
541 msg["delay"]["from"] = self.muc.jid
542 if subject:
543 msg["subject"] = subject
544 else:
545 # may be simplified if slixmpp lets it do it more easily some day
546 msg.xml.append(ET.Element(f"{{{msg.namespace}}}subject"))
547 self._send(msg, full_jid)
549 def set_thread_subject(
550 self,
551 thread: str,
552 subject: str | None,
553 when: datetime | None = None,
554 ) -> None:
555 msg = self._make_message()
556 msg["thread"] = str(thread)
557 if when is not None:
558 msg["delay"].set_stamp(when)
559 msg["delay"]["from"] = self.muc.jid
560 if subject:
561 msg["subject"] = subject
562 else:
563 # may be simplified if slixmpp lets it do it more easily some day
564 msg.xml.append(ET.Element(f"{{{msg.namespace}}}subject"))
565 self._send(msg)
567 async def on_set_affiliation(
568 self,
569 affiliation: MucAffiliation,
570 reason: str | None,
571 nickname: str | None,
572 ) -> None:
573 """
574 Triggered when the user requests changing the affiliation of a contact
575 for this group.
577 Examples: promotion them to moderator, ban (affiliation=outcast).
579 :param contact: The contact whose affiliation change is requested
580 :param affiliation: The new affiliation
581 :param reason: A reason for this affiliation change
582 :param nickname:
583 """
584 raise NotImplementedError
586 async def on_kick(self, reason: str | None) -> None:
587 """
588 Triggered when the user requests changing the role of a contact
589 to "none" for this group. Action commonly known as "kick".
591 :param contact: Contact to be kicked
592 :param reason: A reason for this kick
593 """
594 raise NotImplementedError
596 async def on_invitation(self, reason: str | None) -> None:
597 """
598 Triggered when the user invites this :term:`Contact <Legacy Contact>`
599 to a legacy MUC via :xep:`0249`.
601 The default implementation calls :meth:`LegacyMUC.on_set_affiliation`
602 with the 'member' affiliation. Override if you want to customize this
603 behaviour.
605 :param muc: The group
606 :param reason: Optionally, a reason
607 """
608 # part = await self.muc.get_participant_by_contact(self)
609 await self.on_set_affiliation("member", reason, None)
612def escape_nickname(muc_jid: JID, nickname: str) -> tuple[str, JID]:
613 nickname = nickname_no_illegal = strip_illegal_chars(nickname).replace("\n", " | ")
615 jid = JID(muc_jid)
617 try:
618 jid.resource = nickname
619 except InvalidJID:
620 nickname = nickname.encode("punycode").decode()
621 try:
622 jid.resource = nickname
623 except InvalidJID:
624 # at this point there still might be control chars
625 jid.resource = strip_non_printable(nickname)
627 return nickname_no_illegal, jid
630log = logging.getLogger(__name__)