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

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 __slots__: list[str] = [] 

11 

12 def __init__(self, xmpp) -> None: 

13 super().__init__(xmpp) 

14 xmpp.register_handler( 

15 CoroutineCallback( 

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

17 ) 

18 ) 

19 

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

21 xmpp.add_event_handler( 

22 "groupchat_direct_invite", self.on_groupchat_direct_invite 

23 ) 

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

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

26 

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

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

29 if condition not in KICKABLE_ERRORS: 

30 return 

31 

32 try: 

33 muc = await self.get_muc_from_stanza(msg) 

34 except XMPPError as e: 

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

36 return 

37 mfrom = msg.get_from() 

38 resource = mfrom.resource 

39 try: 

40 muc.remove_user_resource(resource) 

41 except KeyError: 

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

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

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

45 else: 

46 log.info( 

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

48 ) 

49 

50 @exceptions_to_xmpp_errors 

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

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

53 return 

54 

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

56 muc = await self.get_muc_from_stanza(iq) 

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

58 iq.reply().send() 

59 await muc.session.bookmarks.remove( 

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

61 ) 

62 return 

63 

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

65 

66 @exceptions_to_xmpp_errors 

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

68 if not self.xmpp.GROUPS: 

69 raise XMPPError( 

70 "feature-not-implemented", 

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

72 ) 

73 muc = await self.get_muc_from_stanza(p) 

74 await muc.join(p) 

75 

76 @exceptions_to_xmpp_errors 

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

78 invite = msg["groupchat_invite"] 

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

80 

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

82 raise XMPPError( 

83 "bad-request", 

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

85 ) 

86 

87 if invite["password"]: 

88 raise XMPPError( 

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

90 ) 

91 

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

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

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

95 

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

97 

98 @exceptions_to_xmpp_errors 

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

100 muc = await self.get_muc_from_stanza(msg) 

101 if not muc.HAS_SUBJECT: 

102 raise XMPPError( 

103 "bad-request", 

104 "There are no room subject in here. " 

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

106 ) 

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

108 

109 

110KICKABLE_ERRORS = { 

111 "gone", 

112 "internal-server-error", 

113 "item-not-found", 

114 "jid-malformed", 

115 "recipient-unavailable", 

116 "redirect", 

117 "remote-server-not-found", 

118 "remote-server-timeout", 

119 "service-unavailable", 

120 "malformed error", 

121} 

122 

123log = logging.getLogger(__name__)