Coverage for slidge/core/mixins/message_maker.py: 85%

1import base64 

2import io 

3import logging 

4import uuid 

5import warnings 

6from collections.abc import Iterable 

7from datetime import UTC, datetime 

8from pathlib import Path 

9 

10from PIL import Image 

11from slixmpp import Message 

12from slixmpp.plugins.xep_0511.stanza import LinkMetadata 

13from slixmpp.types import MessageTypes 

14 

15from slidge.util import strip_illegal_chars 

16 

17from ...db.models import GatewayUser 

18from ...util.types import ( 

19 AnyRoom, 

20 ChatState, 

21 LinkPreview, 

22 MessageReference, 

23 ProcessingHint, 

24) 

25from .. import config 

26from .base import BaseSender 

27 

28 

29class MessageMaker(BaseSender): 

30 _mtype: MessageTypes = NotImplemented 

31 _can_send_carbon: bool = NotImplemented 

32 _STRIP_SHORT_DELAY = False 

33 _USE_STANZA_ID = False 

34 

35 _room: AnyRoom 

36 

37 def _recipient_pk(self) -> int | None: 

38 """Primary key of receiver of messages made.""" 

39 return None 

40 

41 def _make_message( 

42 self, 

43 state: ChatState | None = None, 

44 hints: Iterable[ProcessingHint] = (), 

45 legacy_msg_id: str | None = None, 

46 when: datetime | None = None, 

47 reply_to: MessageReference | None = None, 

48 carbon: bool = False, 

49 link_previews: Iterable[LinkPreview] | None = None, 

50 **kwargs: object, 

51 ) -> Message: 

52 body = kwargs.pop("mbody", None) 

53 mfrom = kwargs.pop("mfrom", self.jid) 

54 mto = kwargs.pop("mto", None) 

55 thread = kwargs.pop("thread", None) 

56 # the msg needs to have jabber:client as xmlns, so 

57 # we don't want to associate with the XML stream 

58 msg_cls = Message if carbon and self._can_send_carbon else self.xmpp.Message 

59 msg = msg_cls( 

60 sfrom=mfrom, 

61 stype=kwargs.pop("mtype", None) or self._mtype, 

62 sto=mto, 

63 **kwargs, 

64 ) 

65 if body: 

66 assert isinstance(body, str) 

67 msg["body"] = strip_illegal_chars(body, "�") 

68 state = "active" 

69 if thread: 

70 if not self.xmpp.THREADS: 

71 raise RuntimeError( 

72 "Gateway.THREADS = False, no thread should be used here!" 

73 ) 

74 assert isinstance(thread, str) 

75 if (recipient_pk := self._recipient_pk()) is None: 

76 # messages from the gateway component bare JID 

77 msg["thread"] = thread 

78 else: 

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

80 msg["thread"] = ( 

81 self.xmpp.store.id_map.get_thread( 

82 orm, recipient_pk, thread, self.is_participant 

83 ) 

84 or thread 

85 ) 

86 if state: 

87 msg["chat_state"] = state 

88 for hint in hints: 

89 msg.enable(hint) 

90 self._set_msg_id(msg, legacy_msg_id) 

91 self._add_delay(msg, when) 

92 if link_previews: 

93 self._add_link_previews(msg, link_previews) 

94 if reply_to: 

95 self._add_reply_to(msg, reply_to) 

96 return msg 

97 

98 def _set_msg_id(self, msg: Message, legacy_msg_id: str | None = None) -> None: 

99 if legacy_msg_id is not None: 

100 msg.set_id(legacy_msg_id) 

101 if self._USE_STANZA_ID: 

102 msg["stanza_id"]["id"] = legacy_msg_id 

103 msg["stanza_id"]["by"] = self._room.jid 

104 elif self._USE_STANZA_ID: 

105 msg["stanza_id"]["id"] = str(uuid.uuid4()) 

106 msg["stanza_id"]["by"] = self._room.jid 

107 

108 def _legacy_to_xmpp(self, legacy_id: str) -> list[str]: 

109 """ 

110 Get XMPP message IDs that correspond to a given legacy msg ID. 

111 

112 This takes into account single legacy message -> multiple XMPP messages 

113 (eg, multi-attachments). For user-sent 1:1 messages, this returns the 

114 XMPP client-generated msg ID. 

115 """ 

116 # In the case of messages sent by the component's bare JID itself, 

117 # recipient_pk is None and this just wraps the ID in a list. 

118 if (recipient_pk := self._recipient_pk()) is not None: 

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

120 ids = self.xmpp.store.id_map.get_xmpp( 

121 orm, 

122 recipient_pk, 

123 str(legacy_id), 

124 self.is_participant, 

125 ) 

126 if ids: 

127 return ids 

128 return [legacy_id] 

129 

130 def _add_delay(self, msg: Message, when: datetime | None) -> None: 

131 if when: 

132 if when.tzinfo is None: 

133 when = when.astimezone(UTC) 

134 # .contacts (Roster) not being ready means we are in Contact.backfill() 

135 # and in this situation we never want to strip the delay. 

136 if ( 

137 self._session and self._session.contacts.ready.done() 

138 ) and self._STRIP_SHORT_DELAY: 

139 delay = (datetime.now().astimezone(UTC) - when).seconds 

140 if delay < config.IGNORE_DELAY_THRESHOLD: 

141 return 

142 msg["delay"].set_stamp(when) 

143 msg["delay"].set_from(self.xmpp.boundjid.bare) 

144 

145 def _add_reply_to(self, msg: Message, reply_to: MessageReference) -> None: 

146 xmpp_id = self._legacy_to_xmpp(reply_to.legacy_id)[0] 

147 msg["reply"]["id"] = xmpp_id 

148 

149 if entity := reply_to.author: 

150 if entity == "user" or isinstance(entity, GatewayUser): 

151 if isinstance(entity, GatewayUser): 

152 warnings.warn( 

153 "Using a GatewayUser as the author of a " 

154 "MessageReference is deprecated. Use the string 'user' " 

155 "instead.", 

156 DeprecationWarning, 

157 ) 

158 if self.is_participant: 

159 msg["reply"]["to"] = self._room.user_muc_jid 

160 fallback_nick: str | None = self._room.user_nick 

161 elif (session := self._session) is not None: 

162 msg["reply"]["to"] = session.user_jid 

163 # TODO: here we should use preferably use the PEP nick of the user 

164 # (but it doesn't matter much) 

165 fallback_nick = session.user_jid.user 

166 else: 

167 if self.is_participant: 

168 fallback_nick = entity.nickname # type:ignore[union-attr] 

169 msg["reply"]["to"] = entity.jid 

170 else: 

171 fallback_nick = entity.name 

172 else: 

173 fallback_nick = None 

174 

175 if fallback := reply_to.body: 

176 msg["reply"].add_quoted_fallback(fallback, fallback_nick) 

177 

178 def _add_link_previews( 

179 self, msg: Message, link_previews: Iterable[LinkPreview] 

180 ) -> None: 

181 for preview in link_previews: 

182 if preview.is_empty: 

183 continue 

184 element = LinkMetadata() 

185 for i, name in enumerate(preview._fields): 

186 val = preview[i] 

187 if isinstance(val, Path): 

188 val = val.read_bytes() 

189 if isinstance(val, bytes): 

190 val = self._process_link_preview_image(val) 

191 if not val: 

192 continue 

193 element[name] = val 

194 msg.append(element) 

195 

196 @staticmethod 

197 def _process_link_preview_image(data: bytes) -> str | None: 

198 # this will block the main thread. if this proves to be an issue in practice, 

199 # this could be rewritten to use the thread pool we use to resize avatars. 

200 try: 

201 image = Image.open(io.BytesIO(data)) 

202 except Exception: 

203 log.exception("Skipping link preview image") 

204 return None 

205 

206 rewrite = False 

207 if image.format != "JPEG": 

208 rewrite = True 

209 

210 if any(x > MAX_LINK_PREVIEW_IMAGE_SIZE for x in image.size): 

211 image.thumbnail((MAX_LINK_PREVIEW_IMAGE_SIZE, MAX_LINK_PREVIEW_IMAGE_SIZE)) 

212 rewrite = True 

213 

214 if rewrite: 

215 with io.BytesIO() as f: 

216 image.save(f, format="JPEG") 

217 data = f.getvalue() 

218 

219 return "data:image/jpeg;base64," + base64.b64encode(data).decode("utf-8") 

220 

221 

222# Instead of having a hardcoded value for this, we would ideally use 

223# XEP-0478: Stream Limits Advertisement to know which size is authorized. 

224# However, this isn't possible until XEP-0225: Component Connections is a thing. 

225# Prosody defaults to 512kb for s2s connection and 10Mb for c2s connections. 

226# Some quick tests about JPEG image: 

227# median size of 50 base64-encoded JPEG random RGB image 

228# 128x128 pixels: 14kb ± 0.04 

229# 256x256 pixels: 53kb ± 0.06 # sounds like a good tradeoff 

230# 384x384 pixels: 119kb ± 0.11 

231# 512x512 pixels: 211kb ± 0.13 

232# 640x640 pixels: 329kb ± 0.15 

233# 768x768 pixels: 473kb ± 0.21 

234# 896x896 pixels: 644kb ± 0.27 

235# 1024x1024 pixels: 841kb ± 0.22 

236MAX_LINK_PREVIEW_IMAGE_SIZE = 256 

237 

238log = logging.getLogger(__name__)