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

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.core.gateway import BaseGateway 

19 

20 

21class MucMiscMixin(DispatcherMixin): 

22 __slots__: list[str] = [] 

23 

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

25 super().__init__(xmpp) 

26 xmpp.register_handler( 

27 CoroutineCallback( 

28 "ibr_remove", StanzaPath("/iq/register"), self.on_ibr_remove 

29 ) 

30 ) 

31 

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

33 xmpp.add_event_handler( 

34 "groupchat_direct_invite", self.on_groupchat_direct_invite 

35 ) 

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

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

38 xmpp.register_handler( 

39 CoroutineCallback( 

40 "muc_thread_subject", 

41 MatchXMLMask( 

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

43 ), 

44 self.on_thread_subject, 

45 ) 

46 ) 

47 

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

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

50 if condition not in KICKABLE_ERRORS: 

51 return 

52 

53 try: 

54 muc = await self.get_muc_from_stanza(msg) 

55 except XMPPError as e: 

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

57 return 

58 mfrom = msg.get_from() 

59 resource = mfrom.resource 

60 try: 

61 muc.remove_user_resource(resource) 

62 except KeyError: 

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

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

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

66 else: 

67 log.debug( 

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

69 ) 

70 

71 @exceptions_to_xmpp_errors 

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

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

74 return 

75 

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

77 muc = await self.get_muc_from_stanza(iq) 

78 await muc.session.on_leave_group(muc.legacy_id) 

79 iq.reply().send() 

80 await muc.session.bookmarks.remove( 

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

82 ) 

83 return 

84 

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

86 

87 @exceptions_to_xmpp_errors 

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

89 if not self.xmpp.GROUPS: 

90 raise XMPPError( 

91 "feature-not-implemented", 

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

93 ) 

94 muc = await self.get_muc_from_stanza(p) 

95 await muc.join(p) 

96 

97 @exceptions_to_xmpp_errors 

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

99 invite = msg["groupchat_invite"] 

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

101 

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

103 raise XMPPError( 

104 "bad-request", 

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

106 ) 

107 

108 if invite["password"]: 

109 raise XMPPError( 

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

111 ) 

112 

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

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

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

116 

117 await session.on_invitation(contact, muc, invite["reason"] or None) 

118 

119 @exceptions_to_xmpp_errors 

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

121 muc = await self.get_muc_from_stanza(msg) 

122 if not muc.HAS_SUBJECT: 

123 raise XMPPError( 

124 "bad-request", 

125 "There are no room subject in here. " 

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

127 ) 

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

129 

130 @exceptions_to_xmpp_errors 

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

132 if msg["body"]: 

133 return 

134 _session, muc, thread = await self._get_session_recipient_thread(msg) 

135 assert thread is not None 

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

137 

138 

139KICKABLE_ERRORS = { 

140 "gone", 

141 "internal-server-error", 

142 "item-not-found", 

143 "jid-malformed", 

144 "recipient-unavailable", 

145 "redirect", 

146 "remote-server-not-found", 

147 "remote-server-timeout", 

148 "service-unavailable", 

149 "malformed error", 

150} 

151 

152log = logging.getLogger(__name__)