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

1import logging 

2 

3from slixmpp import JID, CoroutineCallback, Iq, Message, Presence, StanzaPath 

4from slixmpp.exceptions import XMPPError 

5 

6from ..util import DispatcherMixin, exceptions_to_xmpp_errors 

7 

8 

9class MucMiscMixin(DispatcherMixin): 

10 def __init__(self, xmpp): 

11 super().__init__(xmpp) 

12 xmpp.register_handler( 

13 CoroutineCallback( 

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

15 ) 

16 ) 

17 

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

19 xmpp.add_event_handler( 

20 "groupchat_direct_invite", self.on_groupchat_direct_invite 

21 ) 

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

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

24 

25 async def __on_group_chat_error(self, msg: Message): 

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

27 if condition not in KICKABLE_ERRORS: 

28 return 

29 

30 try: 

31 muc = await self.get_muc_from_stanza(msg) 

32 except XMPPError as e: 

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

34 return 

35 mfrom = msg.get_from() 

36 resource = mfrom.resource 

37 try: 

38 muc.remove_user_resource(resource) 

39 except KeyError: 

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

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

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

43 else: 

44 log.info( 

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

46 ) 

47 

48 @exceptions_to_xmpp_errors 

49 async def on_ibr_remove(self, iq: Iq): 

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

51 return 

52 

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

54 muc = await self.get_muc_from_stanza(iq) 

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

56 iq.reply().send() 

57 await muc.session.bookmarks.remove( 

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

59 ) 

60 return 

61 

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

63 

64 @exceptions_to_xmpp_errors 

65 async def on_groupchat_join(self, p: Presence): 

66 if not self.xmpp.GROUPS: 

67 raise XMPPError( 

68 "feature-not-implemented", 

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

70 ) 

71 muc = await self.get_muc_from_stanza(p) 

72 await muc.join(p) 

73 

74 @exceptions_to_xmpp_errors 

75 async def on_groupchat_direct_invite(self, msg: Message): 

76 invite = msg["groupchat_invite"] 

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

78 

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

80 raise XMPPError( 

81 "bad-request", 

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

83 ) 

84 

85 if invite["password"]: 

86 raise XMPPError( 

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

88 ) 

89 

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

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

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

93 

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

95 

96 @exceptions_to_xmpp_errors 

97 async def on_groupchat_subject(self, msg: Message): 

98 muc = await self.get_muc_from_stanza(msg) 

99 if not muc.HAS_SUBJECT: 

100 raise XMPPError( 

101 "bad-request", 

102 "There are no room subject in here. " 

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

104 ) 

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

106 

107 

108KICKABLE_ERRORS = { 

109 "gone", 

110 "internal-server-error", 

111 "item-not-found", 

112 "jid-malformed", 

113 "recipient-unavailable", 

114 "redirect", 

115 "remote-server-not-found", 

116 "remote-server-timeout", 

117 "service-unavailable", 

118 "malformed error", 

119} 

120 

121log = logging.getLogger(__name__)