Coverage for slidge/core/dispatcher/presence.py: 84%

1import logging 

2 

3from slixmpp import JID, Presence 

4from slixmpp.exceptions import XMPPError 

5 

6from ...util.util import merge_resources 

7from ..session import BaseSession 

8from .util import DispatcherMixin, exceptions_to_xmpp_errors 

9 

10 

11class _IsDirectedAtComponent(Exception): 

12 def __init__(self, session: BaseSession): 

13 self.session = session 

14 

15 

16class PresenceHandlerMixin(DispatcherMixin): 

17 def __init__(self, xmpp): 

18 super().__init__(xmpp) 

19 

20 xmpp.add_event_handler("presence_subscribe", self._handle_subscribe) 

21 xmpp.add_event_handler("presence_subscribed", self._handle_subscribed) 

22 xmpp.add_event_handler("presence_unsubscribe", self._handle_unsubscribe) 

23 xmpp.add_event_handler("presence_unsubscribed", self._handle_unsubscribed) 

24 xmpp.add_event_handler("presence_probe", self._handle_probe) 

25 xmpp.add_event_handler("presence", self.on_presence) 

26 

27 async def __get_contact(self, pres: Presence): 

28 sess = await self._get_session(pres) 

29 pto = pres.get_to() 

30 if pto == self.xmpp.boundjid.bare: 

31 raise _IsDirectedAtComponent(sess) 

32 await sess.contacts.ready 

33 return await sess.contacts.by_jid(pto) 

34 

35 @exceptions_to_xmpp_errors 

36 async def _handle_subscribe(self, pres: Presence): 

37 try: 

38 contact = await self.__get_contact(pres) 

39 except _IsDirectedAtComponent: 

40 pres.reply().send() 

41 return 

42 

43 if contact.is_friend: 

44 pres.reply().send() 

45 else: 

46 await contact.on_friend_request(pres["status"]) 

47 

48 @exceptions_to_xmpp_errors 

49 async def _handle_unsubscribe(self, pres: Presence): 

50 pres.reply().send() 

51 

52 try: 

53 contact = await self.__get_contact(pres) 

54 except _IsDirectedAtComponent as e: 

55 e.session.send_gateway_message("Bye bye!") 

56 await e.session.kill_by_jid(e.session.user_jid) 

57 return 

58 

59 contact.is_friend = False 

60 await contact.on_friend_delete(pres["status"]) 

61 

62 @exceptions_to_xmpp_errors 

63 async def _handle_subscribed(self, pres: Presence): 

64 try: 

65 contact = await self.__get_contact(pres) 

66 except _IsDirectedAtComponent: 

67 return 

68 

69 await contact.on_friend_accept() 

70 

71 @exceptions_to_xmpp_errors 

72 async def _handle_unsubscribed(self, pres: Presence): 

73 try: 

74 contact = await self.__get_contact(pres) 

75 except _IsDirectedAtComponent: 

76 return 

77 

78 if contact.is_friend: 

79 contact.is_friend = False 

80 await contact.on_friend_delete(pres["status"]) 

81 

82 @exceptions_to_xmpp_errors 

83 async def _handle_probe(self, pres: Presence): 

84 try: 

85 contact = await self.__get_contact(pres) 

86 except _IsDirectedAtComponent: 

87 session = await self._get_session(pres) 

88 session.send_cached_presence(pres.get_from()) 

89 return 

90 if contact.is_friend: 

91 contact.send_last_presence(force=True) 

92 else: 

93 reply = pres.reply() 

94 reply["type"] = "unsubscribed" 

95 reply.send() 

96 

97 @exceptions_to_xmpp_errors 

98 async def on_presence(self, p: Presence): 

99 if p.get_plugin("muc_join", check=True): 

100 # handled in on_groupchat_join 

101 # without this early return, since we switch from and to in this 

102 # presence stanza, on_groupchat_join ends up trying to instantiate 

103 # a MUC with the user's JID, which in turn leads to slidge sending 

104 # a (error) presence from=the user's JID, which terminates the 

105 # XML stream. 

106 return 

107 

108 session = await self._get_session(p) 

109 

110 pto = p.get_to() 

111 if pto == self.xmpp.boundjid.bare: 

112 session.log.debug("Received a presence from %s", p.get_from()) 

113 if (ptype := p.get_type()) not in _USEFUL_PRESENCES: 

114 return 

115 if not session.user.preferences.get("sync_presence", False): 

116 session.log.debug("User does not want to sync their presence") 

117 return 

118 # NB: get_type() returns either a proper presence type or 

119 # a presence show if available. Weird, weird, weird slix. 

120 resources = self.xmpp.roster[self.xmpp.boundjid.bare][ 

121 p.get_from() 

122 ].resources 

123 await session.on_presence( 

124 p.get_from().resource, 

125 ptype, # type: ignore 

126 p["status"], 

127 resources, 

128 merge_resources(resources), 

129 ) 

130 if p.get_type() == "available": 

131 await self.xmpp.pubsub.on_presence_available(p, None) 

132 return 

133 

134 if p.get_type() == "available": 

135 try: 

136 contact = await session.contacts.by_jid(pto) 

137 except XMPPError: 

138 contact = None 

139 if contact is not None: 

140 await self.xmpp.pubsub.on_presence_available(p, contact) 

141 return 

142 

143 muc = session.bookmarks.by_jid_only_if_exists(JID(pto.bare)) 

144 

145 if muc is not None and p.get_type() == "unavailable": 

146 return muc.on_presence_unavailable(p) 

147 

148 if muc is None or p.get_from().resource not in muc.get_user_resources(): 

149 return 

150 

151 if pto.resource == muc.user_nick: 

152 # Ignore presence stanzas with the valid nick. 

153 # even if joined to the group, we might receive those from clients, 

154 # when setting a status message, or going away, etc. 

155 return 

156 

157 # We can't use XMPPError here because XMPPError does not have a way to 

158 # add the <x xmlns="http://jabber.org/protocol/muc" /> element 

159 

160 error_stanza = p.error() 

161 error_stanza.set_to(p.get_from()) 

162 error_stanza.set_from(pto) 

163 error_stanza.enable("muc_join") # <x xmlns="http://jabber.org/protocol/muc" /> 

164 error_stanza.enable("error") 

165 error_stanza["error"]["type"] = "cancel" 

166 error_stanza["error"]["by"] = muc.jid 

167 error_stanza["error"]["condition"] = "not-acceptable" 

168 error_stanza["error"][ 

169 "text" 

170 ] = "Slidge does not let you change your nickname in groups." 

171 error_stanza.send() 

172 

173 

174_USEFUL_PRESENCES = {"available", "unavailable", "away", "chat", "dnd", "xa"} 

175 

176log = logging.getLogger(__name__)