#!/usr/bin/env python3 # Libervia implementation of XEP-0045 (Multi-User Chat) # Copyright (C) 2009-2026 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . from dataclasses import dataclass import difflib from enum import StrEnum import time from typing import TYPE_CHECKING, cast from collections.abc import Sequence from twisted.python.failure import Failure import uuid from zope.interface import implementer from pydantic import Field import shortuuid from twisted.internet import defer from twisted.python import failure from twisted.words.protocols.jabber import jid from twisted.words.protocols.jabber.error import StanzaError from tx_xmpp import disco, itx_xmpp, mam, muc, rsm from libervia.backend import G from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import D_, _ from libervia.backend.core.log import getLogger from libervia.backend.memory import memory from libervia.backend.plugins import plugin_misc_jid_search from libervia.backend.plugins.plugin_xep_0313 import XEP_0313 from libervia.backend.plugins.plugin_xep_0359 import XEP_0359 from libervia.backend.tools import utils from libervia.backend.tools import xml_tools from libervia.backend.tools.config import PluginSettings if TYPE_CHECKING: from libervia.backend.core.main import LiberviaBackend log = getLogger(__name__) IMPORT_NAME = "XEP-0045" PLUGIN_INFO = { C.PI_NAME: "XEP-0045 Plugin", C.PI_IMPORT_NAME: IMPORT_NAME, C.PI_TYPE: "XEP", C.PI_MODES: C.PLUG_MODE_BOTH, C.PI_PROTOCOLS: ["XEP-0045"], C.PI_DEPENDENCIES: ["XEP-0359", "JID_SEARCH"], C.PI_RECOMMENDATIONS: [C.TEXT_CMDS, "XEP-0313"], C.PI_MAIN: "XEP_0045", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("Implementation of Multi-User Chat"), } NS_MUC = "http://jabber.org/protocol/muc" ROOM_USER_JOINED = "ROOM_USER_JOINED" ROOM_USER_LEFT = "ROOM_USER_LEFT" OCCUPANT_KEYS = ("nick", "entity", "affiliation", "role") DEFAULT_HISTORY_SYNC_LIMIT = 50 @dataclass class MUCClientData: _muc_client: "LiberviaMUCClient | None" = None muc_service: jid.JID | None = None @property def muc_client(self) -> "LiberviaMUCClient": """Get the MUC client, asserting it exists.""" assert self._muc_client is not None return self._muc_client class Affiliation(StrEnum): OWNER = "owner" ADMIN = "admin" MEMBER = "member" NONE = "none" OUTCAST = "outcast" @dataclass class OccupantInfo: nick: str entity: str affiliation: str role: str class PluginsMucSettings(PluginSettings, plugin_info=PLUGIN_INFO): default_muc: str = Field(default="libervia@chat.jabberfr.org") class AlreadyJoined(exceptions.ConflictError): """Raised when attempting to join an already joined room.""" def __init__(self, room: muc.Room) -> None: super().__init__() self.room = room class XEP_0045: # TODO: handle invitations def __init__(self, host: "LiberviaBackend") -> None: log.info(_("Plugin XEP_0045 initialization")) self.host = host self.settings = G.config.get_settings(PluginsMucSettings) self._sessions = memory.Sessions() self._jid_search = cast( plugin_misc_jid_search.JidSearch, host.plugins["JID_SEARCH"] ) # return same arguments as muc_room_joined + a boolean set to True if the room was # already joined (first argument) host.bridge.add_method( "muc_join", ".plugin", in_sign="ssa{ss}s", out_sign="(bsa{sa{ss}}ssass)", method=self._join, async_=True, ) host.bridge.add_method( "muc_nick", ".plugin", in_sign="sss", out_sign="", method=self._nick ) host.bridge.add_method( "muc_nick_get", ".plugin", in_sign="ss", out_sign="s", method=self._get_room_nick, ) host.bridge.add_method( "muc_leave", ".plugin", in_sign="ss", out_sign="", method=self._leave, async_=True, ) host.bridge.add_method( "muc_occupants_get", ".plugin", in_sign="ss", out_sign="a{sa{ss}}", method=self._get_room_occupants, ) host.bridge.add_method( "muc_subject", ".plugin", in_sign="sss", out_sign="", method=self._subject ) host.bridge.add_method( "muc_get_rooms_joined", ".plugin", in_sign="s", out_sign="a(sa{sa{ss}}ssas)", method=self._get_rooms_joined, ) host.bridge.add_method( "muc_get_unique_room_name", ".plugin", in_sign="ss", out_sign="s", method=self._get_unique_name, ) host.bridge.add_method( "muc_configure_room", ".plugin", in_sign="ss", out_sign="s", method=self._configure_room, async_=True, ) host.bridge.add_method( "muc_get_default_service", ".plugin", in_sign="", out_sign="s", method=self.get_default_muc, ) host.bridge.add_method( "muc_get_service", ".plugin", in_sign="ss", out_sign="s", method=self._get_muc_service, async_=True, ) # args: room_jid, occupants, user_nick, subject, profile host.bridge.add_signal("muc_room_joined", ".plugin", signature="sa{sa{ss}}ssass") # args: room_jid, profile host.bridge.add_signal("muc_room_left", ".plugin", signature="ss") # args: room_jid, old_nick, new_nick, profile host.bridge.add_signal("muc_room_user_changed_nick", ".plugin", signature="ssss") # args: room_jid, subject, profile host.bridge.add_signal("muc_room_new_subject", ".plugin", signature="sss") self.__submit_conf_id = host.register_callback( self._submit_configuration, with_data=True ) self._room_join_id = host.register_callback(self._ui_room_join_cb, with_data=True) host.import_menu( (D_("MUC"), D_("configure")), self._configure_room_menu, security_limit=0, help_string=D_("Configure Multi-User Chat room"), type_=C.MENU_ROOM, ) try: self.text_cmds = self.host.plugins[C.TEXT_CMDS] except KeyError: log.info(_("Text commands not available")) else: self.text_cmds.register_text_commands(self) self.text_cmds.add_who_is_cb(self._whois, 100) self._mam = cast(XEP_0313, self.host.plugins.get("XEP-0313")) self._si = cast(XEP_0359, self.host.plugins["XEP-0359"]) host.trigger.add("presence_available", self.presence_trigger) host.trigger.add("presence_received", self.presence_received_trigger) host.trigger.add( "message_received", self.message_received_trigger, priority=1000000 ) host.trigger.add("message_parse", self._message_parse_trigger) host.trigger.add( "JID_SEARCH_perform_search", self.jid_search_perform_search_trigger ) async def profile_connected(self, client: SatXMPPEntity) -> None: """Called when a profile connects.""" muc_data = client.get_plugin_data(IMPORT_NAME, MUCClientData) muc_data.muc_service = await self.get_muc_service(client) def _message_parse_trigger( self, client: SatXMPPEntity, message_elt, data: dict ) -> bool: """Add stanza-id from the room if present.""" # stanza_id will not be filled by parse_message because the emitter # is the room and not our server, so we have to parse it here if message_elt.getAttribute("type") == C.MESS_TYPE_GROUPCHAT: room_jid = data["from"].userhostJID() stanza_id = self._si.get_stanza_id(message_elt, room_jid) if stanza_id: data["extra"]["stanza_id"] = stanza_id return True def message_received_trigger( self, client: SatXMPPEntity, message_elt, post_treat: bool ) -> bool: """Handle incoming groupchat messages.""" if message_elt.getAttribute("type") != C.MESS_TYPE_GROUPCHAT: return True from_jid = jid.JID(message_elt["from"]) room_jid = from_jid.userhostJID() if ( room_jid not in client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_client.joined_rooms ): log.warning( "Received groupchat message for unjoined room, ignoring: %s", message_elt.toXml(), ) return False return True async def jid_search_perform_search_trigger( self, client: SatXMPPEntity, search_term: str, options: plugin_misc_jid_search.Options, sequence_matcher: difflib.SequenceMatcher, matches: plugin_misc_jid_search.SearchItems, ) -> bool: """Search MUC rooms if groupchat option is enabled.""" # FIXME: RSM is not handled, we need proper RSM handling as number of # rooms may be large. muc_data = client.get_plugin_data(IMPORT_NAME, MUCClientData) if not options.groupchat or muc_data.muc_service is None: return True disco_items = await self.host.memory.disco.get_items(client, muc_data.muc_service) for disco_item in disco_items: search_item = plugin_misc_jid_search.RoomSearchItem( entity=disco_item.entity, name=disco_item.name or "", local=True ) self._jid_search.process_matching( search_term, sequence_matcher, matches, search_item ) return True def get_room(self, client: SatXMPPEntity, room_jid: jid.JID) -> muc.Room: """Retrieve Room instance from its jid. @param client: The client instance. @param room_jid: JID of the room. @return: The joined room. """ try: return client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_client.joined_rooms[room_jid] except KeyError: raise exceptions.NotFound(_("This room has not been joined")) def check_room_joined(self, client: SatXMPPEntity, room_jid: jid.JID) -> None: """Check that given room has been joined in current session. @param client: The client instance. @param room_jid: Room JID. """ if ( room_jid not in client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_client.joined_rooms ): raise exceptions.NotFound(_("This room has not been joined")) def is_joined_room(self, client: SatXMPPEntity, room_jid: jid.JID) -> bool: """Check if a JID is a known and joined room. @param client: The client instance. @param room_jid: JID of the room. @return: True if the room is joined. """ try: self.check_room_joined(client, room_jid) except exceptions.NotFound: return False return True def is_room(self, client: SatXMPPEntity, entity_jid: jid.JID) -> bool: """Check if a JID is a joined MUC. @param client: The client instance. @param entity_jid: JID of the entity to check. @return: True if the JID is a joined MUC. """ try: self.check_room_joined(client, entity_jid.userhostJID()) except exceptions.NotFound: return False return True def get_bare_or_full(self, client: SatXMPPEntity, peer_jid: jid.JID) -> jid.JID: """Use full JID if peer is an occupant of a room, bare JID else. @param client: The client instance. @param peer_jid: JID of the entity to test. @return: Bare or full JID. """ if peer_jid.resource: if not self.is_room(client, peer_jid): return peer_jid.userhostJID() return peer_jid def get_room_user_jid(self, client: SatXMPPEntity, room_jid: jid.JID) -> jid.JID: """Get the nick used in the room. @param client: The client instance. @param room_jid: Room JID (bare, no resource). @return: Occupant JID with nick as resource. """ # FIXME: doesn't check if room is anonymous or not # TODO: check if room is (semi)anonymous if room_jid.resource: raise exceptions.InternalError(f"{room_jid} should not have a resource") room = self.get_room(client, room_jid) return jid.JID(tuple=(room_jid.user, room_jid.host, room.nick)) def _get_room_joined_bridge_args( self, client: SatXMPPEntity, room: muc.Room ) -> Sequence[str | dict | list[str]]: """Build arguments for muc_room_joined bridge signal. @param client: The client instance. @param room: The joined room. @return: Sequence of arguments for the bridge signal. """ occupants = self.get_room_occupants(client, room.roomJID) return [ room.roomJID.userhost(), occupants, room.nick, # getattr because subject arrives after self-presence (XEP-0045 §7.1) getattr(room, "subject", ""), [s.name for s in getattr(room, "statuses", [])], client.profile, ] def _ui_room_join_cb(self, data: dict, profile: str) -> dict: """Handle UI room join callback. @param data: UI data dict. @param profile: Profile key. @return: Result dict. """ room_jid = jid.JID(data["index"]) client = self.host.get_client(profile) defer.ensureDeferred(self.join(client, room_jid)) return {} def _password_ui_cb( self, data: dict, client: SatXMPPEntity, room_jid: jid.JID, nick: str, ) -> defer.Deferred: """Handle password UI callback. @param data: UI data dict. @param client: The client instance. @param room_jid: Room JID. @param nick: Nickname. @return: Deferred. """ if C.bool(data.get(C.XMLUI_DATA_CANCELLED, "false")): log.info("Room join for %s cancelled", room_jid.userhost()) return defer.fail( failure.Failure(exceptions.CancelError("Room joining cancelled")) ) password = data[xml_tools.form_escape("password")] return ( client.get_plugin_data(IMPORT_NAME, MUCClientData) .muc_client.join(room_jid, nick, password) .addCallbacks( lambda room: room, lambda f: self._join_eb(f, client, room_jid, nick, password), ) ) def _show_list_ui(self, items: list, client: SatXMPPEntity, service: jid.JID) -> None: """Show room list UI.""" xmlui = xml_tools.XMLUI(title=f"Rooms in {service.full()}") adv_list = xmlui.change_container( "advanced_list", columns=1, selectable="single", callback_id=self._room_join_id, ) items = sorted(items, key=lambda i: i.name.lower()) for item in items: adv_list.set_row_index(item.entity.full()) xmlui.addText(item.name) adv_list.end() self.host.action_new({"xmlui": xmlui.toXml()}, profile=client.profile) def _join_eb( self, failure_: failure.Failure, client: SatXMPPEntity, room_jid: jid.JID, nick: str, password: str | None, ) -> defer.Deferred: """Handle join errors. @param failure_: The failure object. @param client: The client instance. @param room_jid: Room JID. @param nick: Nickname. @param password: Password or None. @return: Deferred. """ # TODO: error management + signal in bridge try: condition = failure_.value.condition except AttributeError: msg_suffix = f": {failure_}" else: if condition == "conflict": # we have a nickname conflict, we try again with "_" suffixed to current nickname nick += "_" return ( client.get_plugin_data(IMPORT_NAME, MUCClientData) .muc_client.join(room_jid, nick, password) .addCallbacks( lambda room: room, lambda f: self._join_eb(f, client, room_jid, nick, password), ) ) elif condition == "not-allowed": # room is restricted, we need a password password_ui = xml_tools.XMLUI( "form", title=f"Room {room_jid.userhost()} is restricted", submit_id="", ) password_ui.addText("This room is restricted, please enter the password") password_ui.addPassword("password") d = xml_tools.defer_xmlui(self.host, password_ui, profile=client.profile) d.addCallback( lambda data: self._password_ui_cb(data, client, room_jid, nick) ) return d msg_suffix = f' with condition "{condition}"' mess = f"Error while joining room {room_jid.userhost()}{msg_suffix}" log.warning(mess) xmlui = xml_tools.note(mess, "Group chat error", level=C.XMLUI_DATA_LVL_ERROR) self.host.action_new({"xmlui": xmlui.toXml()}, profile=client.profile) return defer.fail(failure_) def _get_room_occupants( self, room_jid_s: str, profile_key: str ) -> dict[str, dict[str, str]]: """Get room occupants.""" client = self.host.get_client(profile_key) room_jid = jid.JID(room_jid_s) return self.get_room_occupants(client, room_jid) def get_room_occupants( self, client: SatXMPPEntity, room_jid: jid.JID ) -> dict[str, dict[str, str]]: """Get room occupants as dict suitable for bridge.""" room = self.get_room(client, room_jid) return { u.nick: {k: str(getattr(u, k, "")) for k in OCCUPANT_KEYS} for u in room.roster.values() } def _get_rooms_joined(self, profile_key: str = C.PROF_KEY_NONE) -> list[tuple]: """Get joined rooms. @param profile_key: Profile key. @return: List of joined room tuples. """ client = self.host.get_client(profile_key) return self.get_rooms_joined(client) def get_rooms_joined(self, client: SatXMPPEntity) -> list[tuple]: """Return rooms where user is. @param client: The client instance. @return: List of joined room tuples. """ result = [] for room in client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_client.joined_rooms.values(): occupants = { u.nick: {k: str(getattr(u, k, "")) for k in OCCUPANT_KEYS} for u in room.roster.values() } result.append( ( room.roomJID.userhost(), occupants, room.nick, # getattr because subject arrives after self-presence (XEP-0045 §7.1) getattr(room, "subject", ""), [s.name for s in getattr(room, "statuses", [])], ) ) return result def _get_room_nick(self, room_jid_s: str, profile_key: str = C.PROF_KEY_NONE) -> str: """Get room nick. @param room_jid_s: Room JID string. @param profile_key: Profile key. @return: Room nick string. """ client = self.host.get_client(profile_key) return self.get_room_nick(client, jid.JID(room_jid_s)) def get_room_nick(self, client: SatXMPPEntity, room_jid: jid.JID) -> str: """Return nick used in room by user. @param client: The client instance. @param room_jid: Room JID. @return: Nick or empty string in case of error. """ self.check_room_joined(client, room_jid) return ( client.get_plugin_data(IMPORT_NAME, MUCClientData) .muc_client.joined_rooms[room_jid] .nick ) def _configure_room( self, room_jid_s: str, profile_key: str = C.PROF_KEY_NONE ) -> defer.Deferred: """Configure room. @param room_jid_s: Room JID string. @param profile_key: Profile key. @return: Deferred with XMLUI form string. """ client = self.host.get_client(profile_key) d = self.configure_room(client, jid.JID(room_jid_s)) d.addCallback(lambda xmlui: xmlui.toXml() if xmlui else "") return d def _configure_room_menu(self, menu_data: dict, profile: str) -> defer.Deferred: """Return room configuration form. @param menu_data: Menu data dict. @param profile: Profile key. @return: Deferred with XMLUI form string. """ client = self.host.get_client(profile) try: room_jid = jid.JID(menu_data["room_jid"]) except KeyError: log.error("room_jid key is not present!") return defer.fail(exceptions.DataError()) def xmlui_received(xmlui: xml_tools.XMLUI | None) -> dict: if not xmlui: msg = D_("No configuration available for this room") return {"xmlui": xml_tools.note(msg).toXml()} return {"xmlui": xmlui.toXml()} return self.configure_room(client, room_jid).addCallback(xmlui_received) def configure_room(self, client: SatXMPPEntity, room_jid: jid.JID) -> defer.Deferred: """Return the room configuration form. @param client: The client instance. @param room_jid: Room JID to configure. @return: Deferred firing with configuration form as XMLUI. """ self.check_room_joined(client, room_jid) def config_2_xmlui(result) -> xml_tools.XMLUI | None: if not result: return None session_id, session_data = self._sessions.new_session(profile=client.profile) session_data["room_jid"] = room_jid xmlui = xml_tools.data_form_2_xmlui(result, submit_id=self.__submit_conf_id) xmlui.session_id = session_id return xmlui d = client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_client.getConfiguration(room_jid) d.addCallback(config_2_xmlui) return d def _submit_configuration(self, raw_data: dict, profile: str) -> defer.Deferred: """Submit room configuration.""" cancelled = C.bool(raw_data.get("cancelled", C.BOOL_FALSE)) if cancelled: return defer.succeed({}) client = self.host.get_client(profile) try: session_data = self._sessions.profile_get(raw_data["session_id"], profile) except KeyError: log.warning("Session ID doesn't exist, session has probably expired.") dialog = xml_tools.XMLUI("popup", title=D_("Room configuration failed")) dialog.addText(D_("Session ID doesn't exist, session has probably expired.")) return defer.succeed({"xmlui": dialog.toXml()}) data = xml_tools.xmlui_result_2_data_form_result(raw_data) d = client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client.configure( session_data["room_jid"], data ) def on_success(_) -> dict: dialog = xml_tools.XMLUI("popup", title=D_("Room configuration succeed")) dialog.addText(D_("The new settings have been saved.")) return {"xmlui": dialog.toXml()} d.addCallback(on_success) del self._sessions[raw_data["session_id"]] return d def is_nick_in_room( self, client: SatXMPPEntity, room_jid: jid.JID, nick: str ) -> bool: """Tell if a nick is currently present in a room.""" self.check_room_joined(client, room_jid) return ( client.get_plugin_data(IMPORT_NAME, MUCClientData) .muc_client.joined_rooms[room_jid] .inRoster(muc.User(nick)) ) # Text commands def _get_muc_service( self, jid_: str | None = None, profile: str = C.PROF_KEY_NONE ) -> defer.Deferred: """Get MUC service JID. @param jid_: JID string or None. @param profile: Profile key. @return: Deferred with MUC service JID string. """ client = self.host.get_client(profile) d = defer.ensureDeferred( self.get_muc_service(client, jid_.userhostJID() if jid_ else None) ) d.addCallback( lambda service_jid: service_jid.full() if service_jid is not None else "" ) return d async def get_muc_service( self, client: SatXMPPEntity, jid_: jid.JID | None = None ) -> jid.JID | None: """Return first found MUC service of an entity. @param client: The client instance. @param jid_: Entity which may have a MUC service, or None for our own server. @return: Found service JID or None. """ if jid_ is None: try: muc_service = client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_service except AttributeError: pass else: if muc_service is not None: return muc_service services = await self.host.find_service_entities( client, "conference", "text", jid_ ) for service in services: if ".irc." not in service.userhost(): # FIXME: # This ugly hack is here to avoid an issue with openfire: the IRC gateway # use "conference/text" identity (instead of "conference/irc") return service return None def _get_unique_name( self, muc_service: str = "", profile_key: str = C.PROF_KEY_NONE ) -> str: """Get unique room name. @param muc_service: MUC service JID string. @param profile_key: Profile key. @return: Unique room name JID string. """ client = self.host.get_client(profile_key) return self.get_unique_name(client, muc_service or None).full() def get_unique_name( self, client: SatXMPPEntity, muc_service: jid.JID | None = None, prefix: str = "", ) -> jid.JID: """Return unique name for a room, avoiding collision. @param client: The client instance. @param muc_service: Leave empty to use the default service. @param prefix: Optional prefix for room name. @return: Unique room bare JID. """ room_name = f"{prefix}{shortuuid.uuid()}" if muc_service is None: try: muc_service = client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_service except AttributeError: raise exceptions.NotReady( "Main server MUC service has not been checked yet" ) if muc_service is None: log.warning("No MUC service found on main server") raise exceptions.FeatureNotFound return jid.JID(f"{room_name}@{muc_service.userhost()}") def get_default_muc(self) -> str: """Return the default MUC. @return: Default MUC JID string. """ return self.settings.default_muc @utils.ensure_deferred async def _join( self, room_jid_s: str, nick: str, options: dict, profile_key: str = C.PROF_KEY_NONE, ) -> list[str | dict | list[str] | bool]: """Join method used by bridge. @param room_jid_s: Room JID string (bare or full). @param nick: Nickname to use. @param options: Join options dict. @param profile_key: Profile identifier. @return: Tuple of (already_joined, room_joined_args). """ client = self.host.get_client(profile_key) if room_jid_s: muc_service = client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_service try: room_jid = jid.JID(room_jid_s) except (RuntimeError, jid.InvalidFormat, AttributeError): raise jid.InvalidFormat( _( "Invalid room identifier: {room_id}. " "Use 'room' or 'room@{muc_service}'" ).format(room_id=room_jid_s, muc_service=str(muc_service)) ) if not room_jid.user: assert muc_service is not None room_jid.user, room_jid.host = room_jid.host, str(muc_service) else: room_jid = self.get_unique_name(client) # TODO: error management + signal in bridge try: room = await self.join(client, room_jid, nick, options or None) except AlreadyJoined as e: return [True] + list(self._get_room_joined_bridge_args(client, e.room)) else: return [False] + list(self._get_room_joined_bridge_args(client, room)) async def join( self, client: SatXMPPEntity, room_jid: jid.JID, nick: str | None = None, options: dict | None = None, ) -> muc.Room: """Join a MUC room. History sync will run in background to join as quick as possible. @param client: The client instance. @param room_jid: Room JID to join. @param nick: Nickname to use (defaults to username). @param options: Optional dict with 'password' key. @return: The joined room. @raise AlreadyJoined: The room is already joined by this profile. """ if nick is None: nick = client.jid.user if options is None: options = {} muc_data = client.get_plugin_data(IMPORT_NAME, MUCClientData) muc_client = muc_data.muc_client if room_jid in muc_client.joined_rooms: room = muc_client.joined_rooms[room_jid] log.info( "{profile} is already in room {room_jid}", profile=client.profile, room_jid=room_jid.userhost(), ) raise AlreadyJoined(room) log.info( "[{profile}] joining room {room} as {nick}", profile=client.profile, room=room_jid.userhost(), nick=nick, ) password = options.get("password") try: room = await muc_client.join( room_jid, nick, muc.HistoryOptions(maxStanzas=0), password ) except Exception as e: room = await self._join_eb( failure.Failure(e), client, room_jid, nick, password ) else: if room.locked: log.debug("Room locked, configuring it") try: await muc_client.configure(room.roomJID, {}) except Exception as e: log.exception(f"Can't configure MUC room {room_jid}") if self._mam is not None: # We use `ensureDeferred` to run the archive sync in background defer.ensureDeferred(self._sync_mam_history(client, room)) return room async def _sync_mam_history(self, client: SatXMPPEntity, room: muc.Room) -> None: """Retrieve MAM history and send messages to bridge progressively. @param client: The client instance. @param room: The joined room. """ if self._mam is None: return room_jid = room.roomJID total_messages = 0 self._mam.sync_started(room_jid, client.profile) try: last_mess = await self.host.memory.history_get( room_jid, None, limit=1, between=False, filters={"types": C.MESS_TYPE_GROUPCHAT, "last_stanza_id": True}, profile=client.profile, ) if last_mess: stanza_id = last_mess[0][-1].get("stanza_id") rsm_req = rsm.RSMRequest(max_=100, after=stanza_id) else: log.info( "We have no MAM archive for room {room_jid}.".format( room_jid=room_jid ) ) stanza_id = None rsm_req = rsm.RSMRequest(max_=DEFAULT_HISTORY_SYNC_LIMIT, before="") mam_req = mam.MAMRequest(rsm_=rsm_req) while True: try: mam_data = await self._mam.get_archives( client, mam_req, service=room_jid ) except StanzaError as e: if last_mess and e.condition == "item-not-found": log.warning( f"Requested item (with stanza_id {stanza_id!r}) can't be " f"found in history of {room_jid}, history has probably been " "purged on server. Falling back to last N items." ) # Fall back to retrieving last N items like for a new room rsm_req = rsm.RSMRequest( max_=DEFAULT_HISTORY_SYNC_LIMIT, before="" ) mam_req = mam.MAMRequest(rsm_=rsm_req) # Clear last_mess to prevent infinite loop if server keeps failing last_mess = None continue else: raise elt_list, rsm_response, mam_response = mam_data if not elt_list: break for mess_elt in elt_list: try: fwd_message_elt = self._mam.get_message_from_result( client, mess_elt, mam_req, service=room_jid ) except exceptions.DataError: log.exception("Invalid element received") continue if to_ := fwd_message_elt.getAttribute("to"): log.warning( f'Forwarded message element has a "to" attribute ({to_!r})' " while it is forbidden by specifications" ) fwd_message_elt["to"] = client.jid.full() client.messageProt.onMessage(fwd_message_elt) total_messages += 1 if mam_response.get("complete", False): break if rsm_response and rsm_response.last: # Create new MAMRequest with updated rsm (rsm is a property) rsm_req = rsm.RSMRequest(max_=100, after=rsm_response.last) mam_req = mam.MAMRequest(rsm_=rsm_req) else: break if total_messages == 0: log.info( _("No message received while offline in {room_jid}").format( room_jid=room_jid ) ) else: log.info( f"Retrieved {total_messages} message(s) from {room_jid} while" " offline.", ) self._mam.sync_completed(room_jid, client.profile, total_messages) except Exception as e: log.exception(f"MAM history sync failed for {room_jid}") self._mam.sync_error(room_jid, client.profile, str(e)) finally: client.messageProt.flush_archive_batch() def pop_rooms(self, client: SatXMPPEntity) -> list[tuple]: """Remove rooms and return data needed to re-join them. This method is to be called before a hot reconnection. @param client: The client instance. @return: List of tuples (client, room_jid, nick) for re-joining. """ args_list = [] muc_client = client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client for room in list(muc_client.joined_rooms.values()): muc_client._removeRoom(room.roomJID) args_list.append((client, room.roomJID, room.nick)) return args_list def _nick( self, room_jid_s: str, nick: str, profile_key: str = C.PROF_KEY_NONE ) -> None: """Change nickname.""" client = self.host.get_client(profile_key) self.nick(client, jid.JID(room_jid_s), nick) def nick(self, client: SatXMPPEntity, room_jid: jid.JID, nick: str) -> None: """Change nickname in a room. @param client: The client instance. @param room_jid: Room JID. @param nick: New nickname. """ self.check_room_joined(client, room_jid) client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client.nick(room_jid, nick) def _leave(self, room_jid: str, profile_key: str) -> defer.Deferred: """Leave room. @param room_jid: Room JID string. @param profile_key: Profile key. @return: Deferred. """ client = self.host.get_client(profile_key) return self.leave(client, jid.JID(room_jid)) def leave(self, client: SatXMPPEntity, room_jid: jid.JID) -> defer.Deferred: """Leave a room. @param client: The client instance. @param room_jid: Room JID. @return: Deferred firing when leave is complete. """ self.check_room_joined(client, room_jid) return client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client.leave( room_jid ) def _subject(self, room_jid_s: str, new_subject: str, profile_key: str) -> None: """Set room subject.""" client = self.host.get_client(profile_key) self.subject(client, jid.JID(room_jid_s), new_subject) def subject(self, client: SatXMPPEntity, room_jid: jid.JID, subject: str) -> None: """Set room subject. @param client: The client instance. @param room_jid: Room JID. @param subject: New subject. """ self.check_room_joined(client, room_jid) client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client.subject( room_jid, subject ) def get_handler(self, client: SatXMPPEntity) -> "LiberviaMUCClient": """Create a MUC client and associate it with profile's session. @param client: The client instance. @return: The MUC client. """ # create a MUC client and associate it with profile' session muc_data = client.get_plugin_data(IMPORT_NAME, MUCClientData) muc_client = muc_data._muc_client = LiberviaMUCClient(self) return muc_client def kick( self, client: SatXMPPEntity, nick: str, room_jid: jid.JID, options: dict | None = None, ) -> defer.Deferred: """Kick a participant from the room. @param client: The client instance. @param nick: Nick of the user to kick. @param room_jid: Room JID. @param options: Dict with 'reason' key. @return: Deferred firing when kick is complete. """ if options is None: options = {} self.check_room_joined(client, room_jid) return client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client.kick( room_jid, nick, reason=options.get("reason", None) ) def ban( self, client: SatXMPPEntity, entity_jid: jid.JID, room_jid: jid.JID, options: dict | None = None, ) -> defer.Deferred: """Ban an entity from the room. @param client: The client instance. @param entity_jid: Bare JID of the entity to be banned. @param room_jid: Room JID. @param options: Dict with 'reason' key. @return: Deferred firing when ban is complete. """ self.check_room_joined(client, room_jid) if options is None: options = {} assert not entity_jid.resource assert not room_jid.resource return client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client.ban( room_jid, entity_jid, reason=options.get("reason", None) ) def affiliate( self, client: SatXMPPEntity, entity_jid: jid.JID, room_jid: jid.JID, options: dict, ) -> defer.Deferred: """Change the affiliation of an entity. @param client: The client instance. @param entity_jid: Bare JID of the entity. @param room_jid: Room JID. @param options: Dict with 'affiliation' key. @return: Deferred firing when affiliation change is complete. """ self.check_room_joined(client, room_jid) assert not entity_jid.resource assert not room_jid.resource assert "affiliation" in options # TODO: handles reason and nick return client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_client.modifyAffiliationList(room_jid, [entity_jid], options["affiliation"]) def cmd_nick(self, client: SatXMPPEntity, mess_data: dict) -> bool: """Change nickname. @command (group): new_nick - new_nick: new nick to use """ nick = mess_data["unparsed"].strip() if nick: room = mess_data["to"] self.nick(client, room, nick) return False def cmd_join(self, client: SatXMPPEntity, mess_data: dict) -> bool: """Join a new room. @command (all): JID - JID: room to join (on the same service if full jid is not specified) """ room_raw = mess_data["unparsed"].strip() if not room_raw: return False if self.is_joined_room(client, mess_data["to"]): # we use the same service as the one from the room where the command has # been entered if full jid is not entered muc_service = mess_data["to"].host nick = self.get_room_nick(client, mess_data["to"]) or client.jid.user else: # the command has been entered in a one2one conversation, so we use # our server MUC service as default service muc_service = ( client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_service or "" ) nick = client.jid.user room_jid = self.text_cmds.get_room_jid(room_raw, muc_service) defer.ensureDeferred(self.join(client, room_jid, nick, {})) return False def cmd_leave(self, client: SatXMPPEntity, mess_data: dict) -> bool: """Quit a room. @command (group): [ROOM_JID] - ROOM_JID: jid of the room to leave (current room if not specified) """ room_raw = mess_data["unparsed"].strip() if room_raw: room = self.text_cmds.get_room_jid(room_raw, mess_data["to"].host) else: room = mess_data["to"] self.leave(client, room) return False def cmd_part(self, client: SatXMPPEntity, mess_data: dict) -> bool: """Synonym of /leave.""" return self.cmd_leave(client, mess_data) def cmd_kick(self, client: SatXMPPEntity, mess_data: dict) -> bool | defer.Deferred: """Kick a room member. @command (group): ROOM_NICK [reason] - ROOM_NICK: the nick of the person to kick - reason: reason for the kick """ options = mess_data["unparsed"].strip().split() try: nick = options[0] assert self.is_nick_in_room(client, mess_data["to"], nick) except (IndexError, AssertionError): feedback = _("You must provide a member's nick to kick.") self.text_cmds.feed_back(client, feedback, mess_data) return False reason = " ".join(options[1:]) if len(options) > 1 else None d = self.kick(client, nick, mess_data["to"], {"reason": reason}) def cb(_) -> bool: feedback_msg = _("You have kicked {}").format(nick) if reason is not None: feedback_msg += _(" for the following reason: {reason}").format( reason=reason ) self.text_cmds.feed_back(client, feedback_msg, mess_data) return True d.addCallback(cb) return d def cmd_ban(self, client: SatXMPPEntity, mess_data: dict) -> bool | defer.Deferred: """Ban an entity from the room. @command (group): (JID) [reason] - JID: the JID of the entity to ban - reason: the reason why this entity is being banned """ options = mess_data["unparsed"].strip().split() try: jid_s = options[0] entity_jid = jid.JID(jid_s).userhostJID() assert entity_jid.user assert entity_jid.host except ( RuntimeError, jid.InvalidFormat, AttributeError, IndexError, AssertionError, ): feedback = _( "You must provide a valid JID to ban, like in '/ban contact@example.net'" ) self.text_cmds.feed_back(client, feedback, mess_data) return False reason = " ".join(options[1:]) if len(options) > 1 else None d = self.ban(client, entity_jid, mess_data["to"], {"reason": reason}) def cb(_) -> bool: feedback_msg = _("You have banned {}").format(entity_jid) if reason is not None: feedback_msg += _(" for the following reason: {reason}").format( reason=reason ) self.text_cmds.feed_back(client, feedback_msg, mess_data) return True d.addCallback(cb) return d def cmd_affiliate( self, client: SatXMPPEntity, mess_data: dict ) -> bool | defer.Deferred: """Affiliate an entity to the room. @command (group): (JID) [owner|admin|member|none|outcast] - JID: the JID of the entity to affiliate - owner: grant owner privileges - admin: grant admin privileges - member: grant member privileges - none: reset entity privileges - outcast: ban entity """ options = mess_data["unparsed"].strip().split() try: jid_s = options[0] entity_jid = jid.JID(jid_s).userhostJID() assert entity_jid.user assert entity_jid.host except ( RuntimeError, jid.InvalidFormat, AttributeError, IndexError, AssertionError, ): feedback = _( "You must provide a valid JID to affiliate, like in '/affiliate" " contact@example.net member'" ) self.text_cmds.feed_back(client, feedback, mess_data) return False affiliation = options[1] if len(options) > 1 else "none" if affiliation not in Affiliation: feedback = _("You must provide a valid affiliation: {}").format( " ".join(a.value for a in Affiliation) ) self.text_cmds.feed_back(client, feedback, mess_data) return False d = self.affiliate( client, entity_jid, mess_data["to"], {"affiliation": affiliation} ) def cb(_) -> bool: feedback_msg = _("New affiliation for {entity}: {affiliation}").format( entity=entity_jid, affiliation=affiliation ) self.text_cmds.feed_back(client, feedback_msg, mess_data) return True d.addCallback(cb) return d def cmd_title(self, client: SatXMPPEntity, mess_data: dict) -> bool: """Change room's subject. @command (group): title - title: new room subject """ subject = mess_data["unparsed"].strip() if subject: room = mess_data["to"] self.subject(client, room, subject) return False def cmd_topic(self, client: SatXMPPEntity, mess_data: dict) -> bool: """Synonym of /title.""" return self.cmd_title(client, mess_data) def cmd_list(self, client: SatXMPPEntity, mess_data: dict) -> bool | defer.Deferred: """List available rooms in a MUC server. @command (all): [MUC_SERVICE] - MUC_SERVICE: service to request empty value will request room's service for a room, or user's server default MUC service in a one2one chat """ unparsed = mess_data["unparsed"].strip() try: service = jid.JID(unparsed) except RuntimeError: muc_data = client.get_plugin_data(IMPORT_NAME, MUCClientData) if mess_data["type"] == C.MESS_TYPE_GROUPCHAT: room_jid = mess_data["to"] service = jid.JID(room_jid.host) elif muc_data.muc_service is not None: service = muc_data.muc_service else: msg = D_("No known default MUC service") self.text_cmds.feed_back(client, msg, mess_data) return False except jid.InvalidFormat: msg = D_("{} is not a valid JID!").format(unparsed) self.text_cmds.feed_back(client, msg, mess_data) return False d = self.host.getDiscoItems(client, service) d.addCallback(self._show_list_ui, client, service) return d def _whois( self, client: SatXMPPEntity, whois_msg: list, mess_data: dict, target_jid: jid.JID, ) -> None: """Add MUC user information to whois.""" if mess_data["type"] != "groupchat": return if ( target_jid.userhostJID() not in client.get_plugin_data( IMPORT_NAME, MUCClientData ).muc_client.joined_rooms ): log.warning("This room has not been joined") return if not target_jid.resource: return muc_data = client.get_plugin_data(IMPORT_NAME, MUCClientData) user = muc_data.muc_client.joined_rooms[target_jid.userhostJID()].getUser( target_jid.resource ) whois_msg.append(_("Nickname: {}").format(user.nick)) if user.entity: whois_msg.append(_("Entity: {}").format(user.entity)) if user.affiliation != "none": whois_msg.append(_("Affiliation: {}").format(user.affiliation)) if user.role != "none": whois_msg.append(_("Role: {}").format(user.role)) if user.status: whois_msg.append(_("Status: {}").format(user.status)) if user.show: whois_msg.append(_("Show: {}").format(user.show)) def presence_trigger(self, presence_elt, client: SatXMPPEntity) -> bool: """Broadcast presence to all joined rooms.""" # FIXME: should we add a privacy parameters in settings to activate before # broadcasting the presence to all MUC rooms ? muc_client = client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client for room_jid, room in muc_client.joined_rooms.items(): elt = xml_tools.element_copy(presence_elt) elt["to"] = room_jid.userhost() + "/" + room.nick client.presence.send(elt) return True def presence_received_trigger( self, client: SatXMPPEntity, entity: jid.JID, show: str, priority: int, statuses: dict, ) -> bool: """Handle presence received trigger.""" # presence is already handled in (un)availableReceived entity_bare = entity.userhostJID() muc_client = client.get_plugin_data(IMPORT_NAME, MUCClientData).muc_client if entity_bare in muc_client.joined_rooms: return False return True @implementer(itx_xmpp.IDisco) class LiberviaMUCClient(muc.MUCClient): ## presence/roster ## def __init__(self, plugin_parent: XEP_0045) -> None: self.plugin_parent = plugin_parent super().__init__() @property def joined_rooms(self) -> dict: """Get joined rooms.""" return self._rooms @property def host(self) -> "LiberviaBackend": """Get host.""" return self.plugin_parent.host @property def client(self) -> SatXMPPEntity: """Get client.""" assert self.parent is not None return self.parent def availableReceived(self, presence) -> None: """Handle available presence.""" room, user = self._getRoomUser(presence) if room is None or user is None: return # we want to keep statuses with room so they can be sent to the frontend via the # muc_room_joined bridge signal # XXX: after the first received, subsequent broadcasts won't include # status codes like 110 (SELF_PRESENCE) or 100 (REALJID_PUBLIC), so we only # keep the statuses from the initial presence (identified by SELF_PRESENCE); # thus we check if the attribute already exists. if ( not hasattr(room, "statuses") and muc.STATUS_CODE.SELF_PRESENCE in presence.mucStatuses ): room.statuses = presence.mucStatuses super().availableReceived(presence) def unavailableReceived(self, presence) -> None: """Handle unavailable presence.""" # XXX: we override this method to manage nickname change room, user = self._getRoomUser(presence) if room is None or user is None: return room.removeUser(user) if muc.STATUS_CODE.NEW_NICK in presence.mucStatuses: self.user_changed_nick(room, user, presence.nick) else: # we left the room self.userLeftRoom(room, user) def userJoinedRoom(self, room: muc.Room, user: muc.User) -> None: """Handle user joined room event.""" if user.nick == room.nick: # we have received our own nick, this mean that the full room roster was # received log.debug( "room {room} joined with nick {nick}", room=room.occupantJID.userhost(), nick=user.nick, ) args = self.plugin_parent._get_room_joined_bridge_args(self.client, room) self.plugin_parent.host.bridge.muc_room_joined(*args) # we set type so we don't have to use a deferred with disco to check entity # type self.host.memory.update_entity_data( self.client, room.roomJID, C.ENTITY_TYPE, C.ENTITY_TYPE_MUC ) return log.debug( "user {nick} has joined room {room_id}", nick=user.nick, room_id=room.occupantJID.userhost(), ) if not self.host.trigger.point( "MUC user joined", room, user, self.client.profile ): return extra = { "info_type": ROOM_USER_JOINED, "user_affiliation": user.affiliation, "user_role": user.role, "user_nick": user.nick, } if user.entity is not None: extra["user_entity"] = user.entity.full() mess_data = { "from": room.roomJID, "to": self.client.jid, "uid": str(uuid.uuid4()), "message": {"": _("=> {} has joined the room").format(user.nick)}, "subject": {}, "type": C.MESS_TYPE_INFO, "extra": extra, "timestamp": time.time(), } self.client.message_send_to_bridge(mess_data) def userLeftRoom(self, room: muc.Room, user: muc.User) -> None: """Handle user left room event.""" if not self.host.trigger.point("MUC user left", room, user, self.client.profile): return if user.nick == room.nick: room_jid_s = room.roomJID.userhost() log.info( "Room ({room}) left ({profile})", room=room_jid_s, profile=self.client.profile, ) self.host.memory.del_entity_cache( room.roomJID, profile_key=self.client.profile ) self.host.bridge.muc_room_left(room_jid_s, self.client.profile) return log.debug( "user {nick} left room {room_id}", nick=user.nick, room_id=room.occupantJID.userhost(), ) extra = { "info_type": ROOM_USER_LEFT, "user_affiliation": user.affiliation, "user_role": user.role, "user_nick": user.nick, } if user.entity is not None: extra["user_entity"] = user.entity.full() mess_data = { "from": room.roomJID, "to": self.client.jid, "uid": str(uuid.uuid4()), "message": {"": _("<= {} has left the room").format(user.nick)}, "subject": {}, "type": C.MESS_TYPE_INFO, "extra": extra, "timestamp": time.time(), } self.client.message_send_to_bridge(mess_data) def user_changed_nick(self, room: muc.Room, user: muc.User, new_nick: str) -> None: """Handle user nick change.""" self.host.bridge.muc_room_user_changed_nick( room.roomJID.userhost(), user.nick, new_nick, self.client.profile ) def userUpdatedStatus( self, room: muc.Room, user: muc.User, show: str, status: str ) -> None: """Handle user status update.""" entity = jid.JID(tuple=(room.roomJID.user, room.roomJID.host, user.nick)) statuses = {C.PRESENCE_STATUSES_DEFAULT: status or ""} self.host.bridge.presence_update( entity.full(), show or "", 0, statuses, self.client.profile ) def receivedSubject( self, room: muc.Room, user: muc.User | None, subject: str ) -> None: """Handle room subject.""" # We override this method to fix subject handling (empty strings were discarded) # FIXME: remove this once fixed upstream room.subject = subject if user is not None and user.nick == room.nick: return log.debug( "New subject for room ({room_id}): {subject}", room_id=room.occupantJID.full(), subject=subject, ) # the subject has been changed self.host.bridge.muc_room_new_subject( room.roomJID.userhost(), subject, self.client.profile ) ## disco ## def getDiscoInfo(self, requestor, target: jid.JID, nodeIdentifier: str = "") -> list: """Get disco info.""" return [disco.DiscoFeature(NS_MUC)] def getDiscoItems(self, requestor, target: jid.JID, nodeIdentifier: str = "") -> list: """Get disco items.""" # TODO: manage room queries ? Bad for privacy, must be disabled by default # see XEP-0045 § 6.7 return []