Coverage for slidge/core/dispatcher/util.py: 95%

1import logging 

2from collections.abc import Awaitable, Callable 

3from functools import wraps 

4from typing import TYPE_CHECKING, TypeVar 

5 

6from slixmpp import JID, Iq, Message, Presence 

7from slixmpp.exceptions import XMPPError 

8from slixmpp.xmlstream import StanzaBase 

9 

10from ...contact.roster import ContactIsUser 

11from ...util.types import ( 

12 AnyMUC, 

13 AnyRecipient, 

14 AnySession, 

15 RecipientType, 

16) 

17 

18if TYPE_CHECKING: 

19 from slidge.util.types import AnyGateway 

20 

21 

22class Ignore(BaseException): 

23 pass 

24 

25 

26class DispatcherMixin: 

27 __slots__: list[str] = [] 

28 xmpp: "AnyGateway" 

29 

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

31 self.xmpp = xmpp # type:ignore[misc] 

32 

33 async def _get_session( 

34 self, 

35 stanza: Message | Presence | Iq, 

36 timeout: int | None = 10, 

37 wait_for_ready: bool = True, 

38 logged: bool = False, 

39 ) -> AnySession: 

40 xmpp = self.xmpp 

41 if stanza.get_from().server == xmpp.boundjid.bare: 

42 log.debug("Ignoring echo") 

43 raise Ignore 

44 if ( 

45 isinstance(stanza, Message) 

46 and stanza.get_type() == "chat" 

47 and stanza.get_to() == xmpp.boundjid.bare 

48 ): 

49 log.debug("Ignoring message to component") 

50 raise Ignore 

51 session = await self._get_session_from_jid( 

52 stanza.get_from(), timeout, wait_for_ready, logged 

53 ) 

54 if isinstance(stanza, Message) and _ignore(session, stanza): 

55 raise Ignore 

56 return session 

57 

58 async def _get_session_from_jid( 

59 self, 

60 jid: JID, 

61 timeout: int | None = 10, 

62 wait_for_ready: bool = True, 

63 logged: bool = False, 

64 ) -> AnySession: 

65 session = self.xmpp.get_session_from_jid(jid) 

66 if session is None: 

67 raise XMPPError("registration-required") 

68 if logged: 

69 session.raise_if_not_logged() 

70 if wait_for_ready: 

71 await session.wait_for_ready(timeout) 

72 return session 

73 

74 async def get_muc_from_stanza(self, iq: Iq | Message | Presence) -> AnyMUC: 

75 ito = iq.get_to() 

76 if ito == self.xmpp.boundjid.bare: 

77 raise XMPPError("bad-request", text="This is only handled for MUCs") 

78 

79 session = await self._get_session(iq, logged=True) 

80 muc = await session.bookmarks.by_jid(ito) 

81 return muc # type:ignore[no-any-return] 

82 

83 def _xmpp_msg_id_to_legacy( 

84 self, 

85 xmpp_id: str, 

86 recipient: AnyRecipient, 

87 origin: bool = False, 

88 ) -> str: 

89 with self.xmpp.store.session() as orm: 

90 sent = self.xmpp.store.id_map.get_legacy( 

91 orm, recipient.stored.id, xmpp_id, recipient.is_group, origin 

92 ) 

93 if sent is not None: 

94 return sent 

95 

96 return xmpp_id 

97 

98 async def _get_recipient_and_thread( 

99 self, msg: Message 

100 ) -> tuple[AnyRecipient, str | None]: 

101 session = await self._get_session(msg) 

102 e: AnyRecipient = await get_recipient(session, msg) 

103 legacy_thread = await self._xmpp_to_legacy_thread(session, msg, e) 

104 return e, legacy_thread 

105 

106 async def _xmpp_to_legacy_thread( 

107 self, session: AnySession, msg: Message, recipient: RecipientType 

108 ) -> str | None: 

109 if not self.xmpp.THREADS: 

110 return None 

111 

112 xmpp_thread = msg["thread"] 

113 if not xmpp_thread: 

114 return None 

115 

116 with session.xmpp.store.session() as orm: 

117 legacy_thread_str = session.xmpp.store.id_map.get_thread( 

118 orm, recipient.stored.id, xmpp_thread, recipient.is_group 

119 ) 

120 if legacy_thread_str is not None: 

121 return legacy_thread_str 

122 async with session.thread_creation_lock: 

123 if "reply" in msg and "id" in msg["reply"]: 

124 root_legacy_id = self._xmpp_msg_id_to_legacy( 

125 msg["reply"]["id"], recipient 

126 ) 

127 else: 

128 root_legacy_id = self._xmpp_msg_id_to_legacy( 

129 xmpp_thread, 

130 recipient, 

131 ) 

132 legacy_thread = await recipient.create_thread( 

133 xmpp_thread, 

134 root_legacy_id, 

135 ) 

136 with session.xmpp.store.session() as orm: 

137 session.xmpp.store.id_map.set_thread( 

138 orm, 

139 recipient.stored.id, 

140 str(legacy_thread), 

141 xmpp_thread, 

142 recipient.is_group, 

143 ) 

144 orm.commit() 

145 return legacy_thread 

146 

147 

148def _ignore(session: AnySession, msg: Message) -> bool: 

149 i = msg.get_id() 

150 if i.startswith("slidge-carbon-"): 

151 return True 

152 if i not in session.ignore_messages: 

153 return False 

154 session.log.debug("Ignored sent carbon: %s", i) 

155 session.ignore_messages.remove(i) 

156 return True 

157 

158 

159async def get_recipient(session: AnySession, m: Message) -> AnyRecipient: 

160 session.raise_if_not_logged() 

161 if m.get_type() == "groupchat": 

162 muc = await session.bookmarks.by_jid(m.get_to()) 

163 r = m.get_from().resource 

164 if r not in muc.get_user_resources(): 

165 session.create_task(muc.kick_resource(r), name=f"kick of {r} from {muc}") 

166 raise XMPPError("not-acceptable", "You are not connected to this chat") 

167 return muc # type:ignore[no-any-return] 

168 else: 

169 return await session.contacts.by_jid(m.get_to()) # type:ignore[no-any-return] 

170 

171 

172SelfType = TypeVar("SelfType") 

173StanzaType = TypeVar("StanzaType", bound=StanzaBase) 

174HandlerType = Callable[[SelfType, StanzaType], Awaitable[None]] 

175 

176 

177def exceptions_to_xmpp_errors[SelfType, StanzaType: StanzaBase]( 

178 cb: HandlerType[SelfType, StanzaType], 

179) -> HandlerType[SelfType, StanzaType]: 

180 @wraps(cb) 

181 async def wrapped(self: SelfType, stanza: StanzaType) -> None: 

182 try: 

183 await cb(self, stanza) 

184 except Ignore: 

185 pass 

186 except XMPPError: 

187 raise 

188 except NotImplementedError as e: 

189 log.debug("NotImplementedError raised in %s", cb) 

190 tb = e.__traceback__ 

191 assert tb is not None 

192 while tb.tb_next is not None: 

193 tb = tb.tb_next 

194 frame = tb.tb_frame 

195 method_name = frame.f_code.co_name 

196 self_obj = frame.f_locals.get("self") 

197 if self_obj is not None: 

198 method_name = f"{self_obj.__class__.__name__}.{method_name}" 

199 raise XMPPError( 

200 "feature-not-implemented", 

201 f"{method_name} is not implemented by the legacy module", 

202 clear=False, 

203 ) 

204 except ContactIsUser: 

205 raise XMPPError( 

206 "bad-request", "Actions with your bridged self are not allowed." 

207 ) 

208 except Exception as e: 

209 log.error( 

210 "Failed to handle incoming stanza: %s - %s", self, stanza, exc_info=e 

211 ) 

212 raise XMPPError("internal-server-error", str(e)) 

213 

214 return wrapped 

215 

216 

217log = logging.getLogger(__name__)