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