Coverage for slidge/group/participant.py: 88%
371 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-10 04:45 +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 if a.http_metadata is not None:
325 metadata = self.xmpp.plugin["xep_0084"].stanza.MetaData()
326 metadata.append(a.http_metadata)
327 p.append(metadata)
328 else:
329 warnings.warn(
330 f"Private group but no 1:1 JID associated to '{self}'",
331 )
332 if self.is_user and (hash_ := self.session.user.avatar_hash):
333 p["vcard_temp_update"]["photo"] = hash_
334 p["muc"]["status_codes"] = codes
335 return p
337 @property
338 def DISCO_NAME(self) -> str:
339 return self.nickname
341 @DISCO_NAME.setter
342 def DISCO_NAME(self, _: str) -> Never:
343 raise RuntimeError
345 def __send_presence_if_needed(
346 self, stanza: Message | Presence, full_jid: JID, archive_only: bool
347 ) -> None:
348 if (
349 archive_only
350 or self.is_system
351 or self.is_user
352 or self._presence_sent
353 or stanza["subject"]
354 ):
355 return
356 if isinstance(stanza, Message):
357 if "muc" in stanza:
358 return
359 self.send_initial_presence(full_jid)
361 @property
362 def occupant_id(self) -> str:
363 return self.stored.occupant_id
365 def _send(
366 self,
367 stanza: MessageOrPresenceTypeVar,
368 full_jid: JID | None = None,
369 archive_only: bool = False,
370 legacy_msg_id: str | None = None,
371 force: bool = False,
372 **send_kwargs: Any, # noqa:ANN401
373 ) -> MessageOrPresenceTypeVar:
374 if stanza.get_from().resource:
375 stanza["occupant-id"]["id"] = self.occupant_id
376 else:
377 stanza["occupant-id"]["id"] = "room"
378 self.__add_nick_element(stanza)
379 if not self.is_user and isinstance(stanza, Presence):
380 if (
381 not force
382 and stanza["type"] == "unavailable"
383 and not self._presence_sent
384 ):
385 return stanza
386 self._presence_sent = True
387 if full_jid:
388 stanza["to"] = full_jid
389 self.__send_presence_if_needed(stanza, full_jid, archive_only)
390 if self.is_user:
391 assert stanza.stream is not None
392 stanza.stream.send(stanza, use_filters=False)
393 else:
394 stanza.send()
395 else:
396 if hasattr(self.muc, "archive") and isinstance(stanza, Message):
397 self.muc.archive.add(stanza, self, archive_only, legacy_msg_id)
398 if archive_only:
399 return stanza
400 for user_full_jid in self.muc.user_full_jids():
401 stanza = copy(stanza)
402 stanza["to"] = user_full_jid
403 self.__send_presence_if_needed(stanza, user_full_jid, archive_only)
404 stanza.send()
405 return stanza
407 def mucadmin_item(self) -> MUCAdminItem:
408 item = MUCAdminItem()
409 item["nick"] = self.nickname
410 item["affiliation"] = self.affiliation
411 item["role"] = self.role
412 if not self.muc.is_anonymous:
413 if self.is_user:
414 item["jid"] = self.user_jid.bare
415 elif self.contact:
416 item["jid"] = self.contact.jid.bare
417 else:
418 warnings.warn(
419 (
420 f"Private group but no contact JID associated to {self.jid} in"
421 f" {self}"
422 ),
423 )
424 return item
426 def __add_nick_element(self, stanza: Presence | Message) -> None:
427 if (nick := self.nickname_no_illegal) != self.jid.resource:
428 n = self.xmpp.plugin["xep_0172"].stanza.UserNick()
429 n["nick"] = nick
430 stanza.append(n)
432 def _get_last_presence(self) -> CachedPresence | None:
433 own = super()._get_last_presence()
434 if own is None and self.contact:
435 return self.contact._get_last_presence()
436 return own
438 def send_initial_presence(
439 self,
440 full_jid: JID,
441 nick_change: bool = False,
442 presence_id: str | None = None,
443 mav_until: str | None = None,
444 ) -> None:
445 """
446 Called when the user joins a MUC, as a mechanism
447 to indicate to the joining XMPP client the list of "participants".
449 Can be called this to trigger a "participant has joined the group" event.
451 :param full_jid: Set this to only send to a specific user XMPP resource.
452 :param nick_change: Used when the user joins and the MUC renames them (code 210)
453 :param presence_id: set the presence ID. used internally by slidge
454 """
455 # MUC status codes: https://xmpp.org/extensions/xep-0045.html#registrar-statuscodes
456 codes = set()
457 if nick_change:
458 codes.add(210)
460 if self.is_user:
461 # the "initial presence" of the user has to be vanilla, as it is
462 # a crucial part of the MUC join sequence for XMPP clients.
463 kwargs = {}
464 else:
465 cache = self._get_last_presence()
466 self.log.debug("Join muc, initial presence: %s", cache)
467 if cache:
468 ptype = cache.ptype
469 if ptype == "unavailable":
470 return
471 kwargs = {
472 "last_seen": cache.last_seen,
473 "pstatus": cache.pstatus,
474 "pshow": cache.pshow,
475 }
476 else:
477 kwargs = {}
478 p = self._make_presence(
479 status_codes=codes,
480 user_full_jid=full_jid,
481 **kwargs, # type:ignore
482 )
483 if presence_id:
484 p["id"] = presence_id
485 if self.is_user and mav_until is not None:
486 p["muc"]["mav"]["until"] = mav_until
487 self._send(p, full_jid)
489 def leave(self) -> None:
490 """
491 Call this when the participant leaves the room
492 """
493 self.muc.remove_participant(self)
495 def kick(self, reason: str | None = None) -> None:
496 """
497 Call this when the participant is kicked from the room
498 """
499 self.muc.remove_participant(self, kick=True, reason=reason)
501 def ban(self, reason: str | None = None) -> None:
502 """
503 Call this when the participant is banned from the room
504 """
505 self.muc.remove_participant(self, ban=True, reason=reason)
507 async def get_disco_info(
508 self, jid: OptJid = None, node: str | None = None
509 ) -> DiscoInfo:
510 if self.contact is not None:
511 return await self.contact.get_disco_info()
512 return await super().get_disco_info()
514 def moderate(self, legacy_msg_id: str, reason: str | None = None) -> None:
515 for i in self._legacy_to_xmpp(legacy_msg_id):
516 m = self.muc.get_system_participant()._make_message()
517 m["retract"]["id"] = i
518 if self.is_system:
519 m["retract"].enable("moderated")
520 else:
521 m["retract"]["moderated"]["by"] = self.jid
522 m["retract"]["moderated"]["occupant-id"]["id"] = self.occupant_id
523 if reason:
524 m["retract"]["reason"] = reason
525 self._send(m)
527 def set_room_subject(
528 self,
529 subject: str,
530 full_jid: JID | None = None,
531 when: datetime | None = None,
532 update_muc: bool = True,
533 ) -> None:
534 if update_muc:
535 self.muc._subject = subject # type: ignore
536 self.muc.subject_setter = self.nickname
537 self.muc.subject_date = when
539 msg = self._make_message()
540 if when is not None:
541 msg["delay"].set_stamp(when)
542 msg["delay"]["from"] = self.muc.jid
543 if subject:
544 msg["subject"] = subject
545 else:
546 # may be simplified if slixmpp lets it do it more easily some day
547 msg.xml.append(ET.Element(f"{{{msg.namespace}}}subject"))
548 self._send(msg, full_jid)
550 def set_thread_subject(
551 self,
552 thread: str,
553 subject: str | None,
554 when: datetime | None = None,
555 ) -> None:
556 msg = self._make_message()
557 msg["thread"] = str(thread)
558 if when is not None:
559 msg["delay"].set_stamp(when)
560 msg["delay"]["from"] = self.muc.jid
561 if subject:
562 msg["subject"] = subject
563 else:
564 # may be simplified if slixmpp lets it do it more easily some day
565 msg.xml.append(ET.Element(f"{{{msg.namespace}}}subject"))
566 self._send(msg)
568 async def on_set_affiliation(
569 self,
570 affiliation: MucAffiliation,
571 reason: str | None,
572 nickname: str | None,
573 ) -> None:
574 """
575 Triggered when the user requests changing the affiliation of a contact
576 for this group.
578 Examples: promotion them to moderator, ban (affiliation=outcast).
580 :param contact: The contact whose affiliation change is requested
581 :param affiliation: The new affiliation
582 :param reason: A reason for this affiliation change
583 :param nickname:
584 """
585 raise NotImplementedError
587 async def on_kick(self, reason: str | None) -> None:
588 """
589 Triggered when the user requests changing the role of a contact
590 to "none" for this group. Action commonly known as "kick".
592 :param contact: Contact to be kicked
593 :param reason: A reason for this kick
594 """
595 raise NotImplementedError
597 async def on_invitation(self, reason: str | None) -> None:
598 """
599 Triggered when the user invites this :term:`Contact <Legacy Contact>`
600 to a legacy MUC via :xep:`0249`.
602 The default implementation calls :meth:`LegacyMUC.on_set_affiliation`
603 with the 'member' affiliation. Override if you want to customize this
604 behaviour.
606 :param muc: The group
607 :param reason: Optionally, a reason
608 """
609 # part = await self.muc.get_participant_by_contact(self)
610 await self.on_set_affiliation("member", reason, None)
613def escape_nickname(muc_jid: JID, nickname: str) -> tuple[str, JID]:
614 nickname = nickname_no_illegal = strip_illegal_chars(nickname).replace("\n", " | ")
616 jid = JID(muc_jid)
618 try:
619 jid.resource = nickname
620 except InvalidJID:
621 nickname = nickname.encode("punycode").decode()
622 try:
623 jid.resource = nickname
624 except InvalidJID:
625 # at this point there still might be control chars
626 jid.resource = strip_non_printable(nickname)
628 return nickname_no_illegal, jid
631log = logging.getLogger(__name__)