Coverage for slidge/core/mixins/message_maker.py: 84%
141 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-09-29 05:05 +0000
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
11from PIL import Image
12from slixmpp import Message
13from slixmpp.plugins.xep_0511.stanza import LinkMetadata
14from slixmpp.types import MessageTypes
16from slidge.util import strip_illegal_chars
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
31class MessageMaker(BaseSender):
32 mtype: MessageTypes = NotImplemented
33 _can_send_carbon: bool = NotImplemented
34 STRIP_SHORT_DELAY = False
35 USE_STANZA_ID = False
37 muc: AnyMUC
39 def _recipient_pk(self) -> int | None:
40 """Primary key of receiver of messages made."""
41 return None
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 if not self.xmpp.THREADS:
73 raise RuntimeError(
74 "Gateway.THREADS = False, no thread should be used here!"
75 )
76 assert isinstance(thread, str)
77 if (recipient_pk := self._recipient_pk()) is None:
78 # messages from the gateway component bare JID
79 msg["thread"] = thread
80 else:
81 with self.xmpp.store.session() as orm:
82 msg["thread"] = (
83 self.xmpp.store.id_map.get_thread(
84 orm, recipient_pk, thread, self.is_participant
85 )
86 or thread
87 )
88 if state:
89 msg["chat_state"] = state
90 for hint in hints:
91 msg.enable(hint)
92 self._set_msg_id(msg, legacy_msg_id)
93 self._add_delay(msg, when)
94 if link_previews:
95 self._add_link_previews(msg, link_previews)
96 if reply_to:
97 self._add_reply_to(msg, reply_to)
98 return msg
100 def _set_msg_id(self, msg: Message, legacy_msg_id: str | None = None) -> None:
101 if legacy_msg_id is not None:
102 msg.set_id(legacy_msg_id)
103 if self.USE_STANZA_ID:
104 msg["stanza_id"]["id"] = legacy_msg_id
105 msg["stanza_id"]["by"] = self.muc.jid
106 elif self.USE_STANZA_ID:
107 msg["stanza_id"]["id"] = str(uuid.uuid4())
108 msg["stanza_id"]["by"] = self.muc.jid
110 def _legacy_to_xmpp(self, legacy_id: str) -> list[str]:
111 """
112 Get XMPP message IDs that correspond to a given legacy msg ID.
114 This takes into account single legacy message -> multiple XMPP messages
115 (eg, multi-attachments). For user-sent 1:1 messages, this returns the
116 XMPP client-generated msg ID.
117 """
118 # In the case of messages sent by the component's bare JID itself,
119 # recipient_pk is None and this just wraps the ID in a list.
120 if (recipient_pk := self._recipient_pk()) is not None:
121 with self.xmpp.store.session() as orm:
122 ids = self.xmpp.store.id_map.get_xmpp(
123 orm,
124 recipient_pk,
125 str(legacy_id),
126 self.is_participant,
127 )
128 if ids:
129 return ids
130 return [legacy_id]
132 def _add_delay(self, msg: Message, when: datetime | None) -> None:
133 if when:
134 if when.tzinfo is None:
135 when = when.astimezone(UTC)
136 # .contacts (Roster) not being ready means we are in Contact.backfill()
137 # and in this situation we never want to strip the delay.
138 if (
139 self.session and self.session.contacts.ready.done()
140 ) and self.STRIP_SHORT_DELAY:
141 delay = (datetime.now().astimezone(UTC) - when).seconds
142 if delay < config.IGNORE_DELAY_THRESHOLD:
143 return
144 msg["delay"].set_stamp(when)
145 msg["delay"].set_from(self.xmpp.boundjid.bare)
147 def _add_reply_to(self, msg: Message, reply_to: MessageReference) -> None:
148 xmpp_id = self._legacy_to_xmpp(reply_to.legacy_id)[0]
149 msg["reply"]["id"] = xmpp_id
151 muc = getattr(self, "muc", None)
153 if entity := reply_to.author:
154 if entity == "user" or isinstance(entity, GatewayUser):
155 if isinstance(entity, GatewayUser):
156 warnings.warn(
157 "Using a GatewayUser as the author of a "
158 "MessageReference is deprecated. Use the string 'user' "
159 "instead.",
160 DeprecationWarning,
161 )
162 if muc:
163 msg["reply"]["to"] = muc.user_muc_jid
164 fallback_nick = muc.user_nick
165 elif (session := self.session) is not None:
166 msg["reply"]["to"] = session.user_jid
167 # TODO: here we should use preferably use the PEP nick of the user
168 # (but it doesn't matter much)
169 fallback_nick = session.user_jid.user
170 else:
171 if muc:
172 if hasattr(entity, "muc"):
173 # TODO: accept a Contact here and use muc.get_participant_by_legacy_id()
174 # a bit of work because right now this is a sync function
175 entity = cast(AnyParticipant, entity)
176 fallback_nick = entity.nickname
177 else:
178 warnings.warn(
179 "The author of a message reference in a MUC must be a"
180 " Participant instance, not a Contact"
181 )
182 fallback_nick = entity.name
183 else:
184 fallback_nick = entity.name
185 msg["reply"]["to"] = entity.jid
186 else:
187 fallback_nick = None
189 if fallback := reply_to.body:
190 msg["reply"].add_quoted_fallback(fallback, fallback_nick)
192 def _add_link_previews(
193 self, msg: Message, link_previews: Iterable[LinkPreview]
194 ) -> None:
195 for preview in link_previews:
196 if preview.is_empty:
197 continue
198 element = LinkMetadata()
199 for i, name in enumerate(preview._fields):
200 val = preview[i]
201 if isinstance(val, Path):
202 val = val.read_bytes()
203 if isinstance(val, bytes):
204 val = self._process_link_preview_image(val)
205 if not val:
206 continue
207 element[name] = val
208 msg.append(element)
210 @staticmethod
211 def _process_link_preview_image(data: bytes) -> str | None:
212 # this will block the main thread. if this proves to be an issue in practice,
213 # this could be rewritten to use the thread pool we use to resize avatars.
214 try:
215 image = Image.open(io.BytesIO(data))
216 except Exception:
217 log.exception("Skipping link preview image")
218 return None
220 rewrite = False
221 if image.format != "JPEG":
222 rewrite = True
224 if any(x > MAX_LINK_PREVIEW_IMAGE_SIZE for x in image.size):
225 image.thumbnail((MAX_LINK_PREVIEW_IMAGE_SIZE, MAX_LINK_PREVIEW_IMAGE_SIZE))
226 rewrite = True
228 if rewrite:
229 with io.BytesIO() as f:
230 image.save(f, format="JPEG")
231 data = f.getvalue()
233 return "data:image/jpeg;base64," + base64.b64encode(data).decode("utf-8")
236# Instead of having a hardcoded value for this, we would ideally use
237# XEP-0478: Stream Limits Advertisement to know which size is authorized.
238# However, this isn't possible until XEP-0225: Component Connections is a thing.
239# Prosody defaults to 512kb for s2s connection and 10Mb for c2s connections.
240# Some quick tests about JPEG image:
241# median size of 50 base64-encoded JPEG random RGB image
242# 128x128 pixels: 14kb ± 0.04
243# 256x256 pixels: 53kb ± 0.06 # sounds like a good tradeoff
244# 384x384 pixels: 119kb ± 0.11
245# 512x512 pixels: 211kb ± 0.13
246# 640x640 pixels: 329kb ± 0.15
247# 768x768 pixels: 473kb ± 0.21
248# 896x896 pixels: 644kb ± 0.27
249# 1024x1024 pixels: 841kb ± 0.22
250MAX_LINK_PREVIEW_IMAGE_SIZE = 256
252log = logging.getLogger(__name__)