Coverage for slidge/core/dispatcher/muc/misc.py: 92%

1import logging 

2import typing 

3 

4from slixmpp import ( 

5 JID, 

6 CoroutineCallback, 

7 Iq, 

8 MatchXMLMask, 

9 Message, 

10 Presence, 

11 StanzaPath, 

12) 

13from slixmpp.exceptions import XMPPError 

14 

15from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

16 

17if typing.TYPE_CHECKING: 

18 from slidge.util.types import AnyGateway 

19 

20 

21class MucMiscMixin(DispatcherMixin): 

22 __slots__: list[str] = [] 

23 

24 def __init__(self, xmpp: "AnyGateway") -> None: 

25 super().__init__(xmpp) 

26 xmpp.register_handler( 

27 CoroutineCallback( 

28 "ibr_remove", 

29 StanzaPath("/iq/register"), 

30 self.on_ibr_remove, # type:ignore[arg-type] 

31 ) 

32 ) 

33 

34 xmpp.add_event_handler("groupchat_join", self.on_groupchat_join) 

35 xmpp.add_event_handler( 

36 "groupchat_direct_invite", self.on_groupchat_direct_invite 

37 ) 

38 xmpp.add_event_handler("groupchat_subject", self.on_groupchat_subject) 

39 xmpp.add_event_handler("groupchat_message_error", self.__on_group_chat_error) 

40 xmpp.register_handler( 

41 CoroutineCallback( 

42 "muc_thread_subject", 

43 MatchXMLMask( 

44 "<message xmlns='jabber:component:accept' type='groupchat'><subject/><thread/></message>" 

45 ), 

46 self.on_thread_subject, # type:ignore[arg-type] 

47 ) 

48 ) 

49 

50 async def __on_group_chat_error(self, msg: Message) -> None: 

51 condition = msg["error"].get_condition() 

52 if condition not in KICKABLE_ERRORS: 

53 return 

54 

55 try: 

56 muc = await self.get_muc_from_stanza(msg) 

57 except XMPPError as e: 

58 log.debug("Not removing resource", exc_info=e) 

59 return 

60 mfrom = msg.get_from() 

61 resource = mfrom.resource 

62 try: 

63 muc.remove_user_resource(resource) 

64 except KeyError: 

65 # this actually happens quite frequently on for both beagle and monal 

66 # (not sure why?), but is of no consequence 

67 log.debug("%s was not in the resources of %s", resource, muc) 

68 else: 

69 log.debug( 

70 "Removed %s from the resources of %s because of error", resource, muc 

71 ) 

72 

73 @exceptions_to_xmpp_errors 

74 async def on_ibr_remove(self, iq: Iq) -> None: 

75 if iq.get_to() == self.xmpp.boundjid.bare: 

76 return 

77 

78 if iq["type"] == "set" and iq["register"]["remove"]: 

79 muc = await self.get_muc_from_stanza(iq) 

80 await muc.on_leave() 

81 iq.reply().send() 

82 await muc.session.bookmarks.remove( 

83 muc, "You left this chat from an XMPP client." 

84 ) 

85 return 

86 

87 raise XMPPError("feature-not-implemented") 

88 

89 @exceptions_to_xmpp_errors 

90 async def on_groupchat_join(self, p: Presence) -> None: 

91 if not self.xmpp.GROUPS: 

92 raise XMPPError( 

93 "feature-not-implemented", 

94 "This gateway does not implement multi-user chats.", 

95 ) 

96 muc = await self.get_muc_from_stanza(p) 

97 await muc.join(p) 

98 

99 @exceptions_to_xmpp_errors 

100 async def on_groupchat_direct_invite(self, msg: Message) -> None: 

101 invite = msg["groupchat_invite"] 

102 jid = JID(invite["jid"]) 

103 

104 if jid.domain != self.xmpp.boundjid.bare: 

105 raise XMPPError( 

106 "bad-request", 

107 "Legacy contacts can only be invited to legacy groups, not standard XMPP MUCs.", 

108 ) 

109 

110 if invite["password"]: 

111 raise XMPPError( 

112 "bad-request", "Password-protected groups are not supported" 

113 ) 

114 

115 session = await self._get_session(msg, logged=True) 

116 contact = await session.contacts.by_jid(msg.get_to()) 

117 muc = await session.bookmarks.by_jid(jid) 

118 part = await muc.get_participant_by_contact(contact) 

119 await part.on_invitation(invite["reason"] or None) 

120 

121 @exceptions_to_xmpp_errors 

122 async def on_groupchat_subject(self, msg: Message) -> None: 

123 muc = await self.get_muc_from_stanza(msg) 

124 if not muc.HAS_SUBJECT: 

125 raise XMPPError( 

126 "bad-request", 

127 "There are no room subject in here. " 

128 "Use the room configuration to update its name or description", 

129 ) 

130 await muc.on_set_subject(msg["subject"]) 

131 

132 @exceptions_to_xmpp_errors 

133 async def on_thread_subject(self, msg: Message) -> None: 

134 if msg["body"]: 

135 return 

136 if not self.xmpp.THREADS: 

137 raise XMPPError("bad-request", "This gateway does not support threads") 

138 muc, thread = await self._get_recipient_and_thread(msg) 

139 assert thread is not None 

140 await muc.on_set_thread_subject(thread, msg["subject"]) # type:ignore[union-attr] 

141 

142 

143KICKABLE_ERRORS = { 

144 "gone", 

145 "internal-server-error", 

146 "item-not-found", 

147 "jid-malformed", 

148 "recipient-unavailable", 

149 "redirect", 

150 "remote-server-not-found", 

151 "remote-server-timeout", 

152 "service-unavailable", 

153 "malformed error", 

154} 

155 

156log = logging.getLogger(__name__)