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

1import base64 

2import io 

3import logging 

4import uuid 

5import warnings 

6from collections.abc import Iterable 

7from datetime import UTC, datetime 

8from pathlib import Path 

9from typing import cast 

10 

11from PIL import Image 

12from slixmpp import Message 

13from slixmpp.plugins.xep_0511.stanza import LinkMetadata 

14from slixmpp.types import MessageTypes 

15 

16from slidge.util import strip_illegal_chars 

17 

18from ...db.models import GatewayUser 

19from ...util.types import ( 

20 AnyMUC, 

21 AnyParticipant, 

22 ChatState, 

23 LinkPreview, 

24 MessageReference, 

25 ProcessingHint, 

26) 

27from .. import config 

28from .base import BaseSender 

29 

30 

31class MessageMaker(BaseSender): 

32 mtype: MessageTypes = NotImplemented 

33 _can_send_carbon: bool = NotImplemented 

34 STRIP_SHORT_DELAY = False 

35 USE_STANZA_ID = False 

36 

37 muc: AnyMUC 

38 

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

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

41 return None 

42 

43 def _make_message( 

44 self, 

45 state: ChatState | None = None, 

46 hints: Iterable[ProcessingHint] = (), 

47 legacy_msg_id: str | None = None, 

48 when: datetime | None = None, 

49 reply_to: MessageReference | None = None, 

50 carbon: bool = False, 

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

52 **kwargs: object, 

53 ) -> Message: 

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

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

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

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

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

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

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

61 msg = msg_cls( 

62 sfrom=mfrom, 

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

64 sto=mto, 

65 **kwargs, 

66 ) 

67 if body: 

68 assert isinstance(body, str) 

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

70 state = "active" 

71 if thread: 

72 assert isinstance(thread, str) 

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

74 # messages from the gateway component bare JID 

75 msg["thread"] = thread 

76 else: 

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

78 msg["thread"] = ( 

79 self.xmpp.store.id_map.get_thread( 

80 orm, recipient_pk, thread, self.is_participant 

81 ) 

82 or thread 

83 ) 

84 if state: 

85 msg["chat_state"] = state 

86 for hint in hints: 

87 msg.enable(hint) 

88 self._set_msg_id(msg, legacy_msg_id) 

89 self._add_delay(msg, when) 

90 if link_previews: 

91 self._add_link_previews(msg, link_previews) 

92 if reply_to: 

93 self._add_reply_to(msg, reply_to) 

94 return msg 

95 

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

97 if legacy_msg_id is not None: 

98 msg.set_id(legacy_msg_id) 

99 if self.USE_STANZA_ID: 

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

101 msg["stanza_id"]["by"] = self.muc.jid 

102 elif self.USE_STANZA_ID: 

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

104 msg["stanza_id"]["by"] = self.muc.jid 

105 

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

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

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

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

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

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

112 orm, 

113 recipient_pk, 

114 str(legacy_id), 

115 self.is_participant, 

116 ) 

117 if ids: 

118 return ids 

119 return [legacy_id] 

120 

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

122 if when: 

123 if when.tzinfo is None: 

124 when = when.astimezone(UTC) 

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

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

127 if ( 

128 self.session and self.session.contacts.ready.done() 

129 ) and self.STRIP_SHORT_DELAY: 

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

131 if delay < config.IGNORE_DELAY_THRESHOLD: 

132 return 

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

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

135 

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

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

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

139 

140 muc = getattr(self, "muc", None) 

141 

142 if entity := reply_to.author: 

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

144 if isinstance(entity, GatewayUser): 

145 warnings.warn( 

146 "Using a GatewayUser as the author of a " 

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

148 "instead.", 

149 DeprecationWarning, 

150 ) 

151 if muc: 

152 msg["reply"]["to"] = muc.user_muc_jid 

153 fallback_nick = muc.user_nick 

154 elif (session := self.session) is not None: 

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

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

157 # (but it doesn't matter much) 

158 fallback_nick = session.user_jid.user 

159 else: 

160 if muc: 

161 if hasattr(entity, "muc"): 

162 # TODO: accept a Contact here and use muc.get_participant_by_legacy_id() 

163 # a bit of work because right now this is a sync function 

164 entity = cast(AnyParticipant, entity) 

165 fallback_nick = entity.nickname 

166 else: 

167 warnings.warn( 

168 "The author of a message reference in a MUC must be a" 

169 " Participant instance, not a Contact" 

170 ) 

171 fallback_nick = entity.name 

172 else: 

173 fallback_nick = entity.name 

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

175 else: 

176 fallback_nick = None 

177 

178 if fallback := reply_to.body: 

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

180 

181 def _add_link_previews( 

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

183 ) -> None: 

184 for preview in link_previews: 

185 if preview.is_empty: 

186 continue 

187 element = LinkMetadata() 

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

189 val = preview[i] 

190 if isinstance(val, Path): 

191 val = val.read_bytes() 

192 if isinstance(val, bytes): 

193 val = self._process_link_preview_image(val) 

194 if not val: 

195 continue 

196 element[name] = val 

197 msg.append(element) 

198 

199 @staticmethod 

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

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

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

203 try: 

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

205 except Exception: 

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

207 return None 

208 

209 rewrite = False 

210 if image.format != "JPEG": 

211 rewrite = True 

212 

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

214 image.thumbnail((MAX_LINK_PREVIEW_IMAGE_SIZE, MAX_LINK_PREVIEW_IMAGE_SIZE)) 

215 rewrite = True 

216 

217 if rewrite: 

218 with io.BytesIO() as f: 

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

220 data = f.getvalue() 

221 

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

223 

224 

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

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

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

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

229# Some quick tests about JPEG image: 

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

231# 128x128 pixels: 14kb ± 0.04 

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

233# 384x384 pixels: 119kb ± 0.11 

234# 512x512 pixels: 211kb ± 0.13 

235# 640x640 pixels: 329kb ± 0.15 

236# 768x768 pixels: 473kb ± 0.21 

237# 896x896 pixels: 644kb ± 0.27 

238# 1024x1024 pixels: 841kb ± 0.22 

239MAX_LINK_PREVIEW_IMAGE_SIZE = 256 

240 

241log = logging.getLogger(__name__)