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