Coverage for slidge/core/dispatcher/message/message.py: 85%
281 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-09 03:59 +0000
1import base64
2import contextlib
3import hashlib
4import logging
5from copy import copy
6from dataclasses import dataclass
7from typing import TYPE_CHECKING
8from xml.etree import ElementTree
10from slixmpp import JID, Message
11from slixmpp.exceptions import XMPPError
12from slixmpp.plugins.xep_0511.stanza import LinkMetadata
14from ....contact.contact import LegacyContact
15from ....group.room import LegacyMUC
16from ....util.types import (
17 AnyRecipient,
18 LinkPreview,
19 RecipientType,
20 Reply,
21 XMPPAttachment,
22 XMPPAttachmentMessage,
23 XMPPMessage,
24 XMPPTextMessage,
25)
26from ....util.util import dict_to_named_tuple, remove_emoji_variation_selector_16
27from ... import config
28from ..util import DispatcherMixin, exceptions_to_xmpp_errors
30if TYPE_CHECKING:
31 from slidge.util.types import AnyGateway
34@dataclass
35class _IncomingAttachment:
36 attachment: XMPPAttachment
37 is_sticker: bool
38 cid: str | None = None
41class MessageContentMixin(DispatcherMixin):
42 __slots__: list[str] = []
44 def __init__(self, xmpp: "AnyGateway") -> None:
45 super().__init__(xmpp)
46 xmpp.add_event_handler("legacy_message", self.on_legacy_message)
47 xmpp.add_event_handler("message_retract", self.on_message_retract)
48 xmpp.add_event_handler("groupchat_message", self.on_groupchat_message)
49 xmpp.add_event_handler("reactions", self.on_reactions)
51 async def on_groupchat_message(self, msg: Message) -> None:
52 await self.on_legacy_message(msg)
54 @exceptions_to_xmpp_errors
55 async def on_legacy_message(self, msg: Message) -> None:
56 """
57 Meant to be called from :class:`BaseGateway` only.
59 :param msg:
60 :return:
61 """
62 # we MUST not use `if m["replace"]["id"]` because it adds the tag if not
63 # present. this is a problem for MUC echoed messages
64 if "apply_to" in msg:
65 # ignore message retraction (handled by a specific method)
66 # this is an old version of the protocol that depends on the
67 # message fastening deprecated xep.
68 return
69 if "reactions" in msg:
70 # ignore message reaction fallback.
71 # the reaction itself is handled by self.react_from_msg().
72 return
73 if "retract" in msg:
74 # ignore message retraction fallback.
75 # the retraction itself is handled by self.on_retract
76 return
77 recipient, thread = await self._get_recipient_and_thread(msg)
78 replace = await self.__get_replace(msg, recipient)
79 if msg.xml.find(".//{*}encrypted") is not None:
80 raise XMPPError(
81 "bad-request", "You cannot send encrypted messages through this gateway"
82 )
83 body, reply = await self.__get_reply(msg, recipient)
84 cid = self.__get_xhtml_sticker_cid(msg)
86 if cid:
87 legacy_msg_id = await self.__dispatch_bob(
88 msg.get_from(),
89 cid,
90 recipient,
91 reply=reply,
92 thread=thread,
93 )
94 else:
95 attachments = self.__get_attachments(msg)
96 if len(attachments) == 1 and attachments[0].is_sticker:
97 legacy_msg_id = await self.__dispatch_nonbob_sticker(
98 attachments[0],
99 recipient,
100 msg["body"],
101 reply=reply,
102 thread=thread,
103 )
104 else:
105 legacy_msg_id = await self.__dispatch_msg(
106 replace=replace,
107 body=body,
108 attachments=tuple(a.attachment for a in attachments),
109 recipient=recipient,
110 thread=thread,
111 reply=reply,
112 msg=msg,
113 )
115 if isinstance(recipient, LegacyMUC):
116 stanza_id = await recipient.echo(msg, legacy_msg_id)
117 else:
118 stanza_id = None
119 self.__ack(msg)
121 if not legacy_msg_id:
122 return
124 with self.xmpp.store.session() as orm:
125 if recipient.is_group:
126 self.xmpp.store.id_map.set_origin(
127 orm, recipient.stored.id, legacy_msg_id, msg.get_id()
128 )
129 assert stanza_id is not None
130 self.xmpp.store.id_map.set_msg(
131 orm,
132 recipient.stored.id,
133 legacy_msg_id,
134 [stanza_id],
135 True,
136 )
137 else:
138 self.xmpp.store.id_map.set_msg(
139 orm,
140 recipient.stored.id,
141 legacy_msg_id,
142 [msg.get_id()],
143 False,
144 )
145 if recipient.session.MESSAGE_IDS_ARE_THREAD_IDS and thread:
146 self.xmpp.store.id_map.set_thread(
147 orm, recipient.stored.id, thread, legacy_msg_id, recipient.is_group
148 )
149 orm.commit()
151 def __get_xhtml_sticker_cid(self, msg: Message) -> str | None:
152 if "html" not in msg:
153 return None
154 body = ElementTree.fromstring("<body>" + msg["html"].get_body() + "</body>")
155 p = body.findall("p")
156 if p is None:
157 return None
158 if len(p) != 1:
159 return None
160 if p[0].text is not None and p[0].text.strip():
161 return None
163 images = p[0].findall("img")
164 if len(images) != 1:
165 return None
166 # no text, single img ⇒ this is a sticker
167 # other cases should be interpreted as "custom emojis" in text
168 src = images[0].get("src")
169 if src is None:
170 return None
171 if src.startswith("cid:"):
172 return src.removeprefix("cid:")
173 return None
175 async def __get_replace(
176 self,
177 msg: Message,
178 recipient: RecipientType,
179 ) -> str | None:
180 if "replace" not in msg or "id" not in msg["replace"]:
181 return None
182 return (
183 self._xmpp_msg_id_to_legacy(msg["replace"]["id"], recipient, True) or None
184 )
186 def __get_attachments(self, msg: Message) -> list[_IncomingAttachment]:
187 is_sticker = "sticker" in msg
189 if (
190 "sfs" in msg
191 and "sources" in msg["sfs"]
192 and "url-data" in msg["sfs"]["sources"]
193 and "target" in msg["sfs"]["sources"]["url-data"]
194 ):
195 # TODO: support "attach source in later message", cf https://xmpp.org/extensions/xep-0447.html#example-5
196 # TODO: support for other sources than URL
197 # TODO: support for multiattachments in single message.
198 # What do we do if is_sticker and multiple files?
199 attachment = _IncomingAttachment(
200 XMPPAttachment(url=msg["sfs"]["sources"]["url-data"]["target"]),
201 is_sticker=is_sticker,
202 )
203 if "file" in msg["sfs"]:
204 attachment.attachment.content_type = (
205 msg["sfs"]["file"]["media-type"] or None
206 )
207 if "hash" in msg["sfs"]["file"]:
208 algo = msg["sfs"]["file"]["hash"]["algo"]
209 h = msg["sfs"]["file"]["hash"]["value"]
210 attachment.cid = f"{algo}+{h}" if algo and h else None
211 return [attachment]
213 if "oob" in msg:
214 return [
215 _IncomingAttachment(
216 XMPPAttachment(url=msg["oob"]["url"]), is_sticker=is_sticker
217 )
218 ]
220 if (
221 "reference" in msg
222 and "sims" in msg["reference"]
223 and "sources" in msg["reference"]["sims"]
224 ):
225 for source in msg["reference"]["sims"]["sources"]["substanzas"]:
226 if source["uri"].startswith("http"):
227 attachment = _IncomingAttachment(
228 XMPPAttachment(url=source["uri"]), is_sticker=is_sticker
229 )
230 break
231 else:
232 return []
233 if "file" in msg["reference"]["sims"]:
234 attachment.attachment.content_type = msg["media-type"] or None
235 return [attachment]
237 return []
239 async def __dispatch_msg(
240 self,
241 *,
242 replace: str | None,
243 reply: Reply | None,
244 body: str | None,
245 attachments: tuple[XMPPAttachment, ...],
246 recipient: AnyRecipient,
247 thread: str | None,
248 msg: Message,
249 ) -> str | None:
250 if replace:
251 if body is not None:
252 body = body.strip()
253 if not body and not attachments and recipient.RETRACTION:
254 await recipient.on_retract(replace, thread=thread)
255 return None
256 if not recipient.CORRECTION:
257 if recipient.RETRACTION:
258 await recipient.on_retract(replace, thread=thread)
259 replace = None
260 elif body:
261 body = "Correction:\n" + body
263 if not any([attachments, body]):
264 log.debug(
265 "Ignoring msg with id '%s' because no body or attachments found",
266 msg.get_id(),
267 )
268 return None
269 link_previews = (
270 parse_link_previews(msg["link_metadatas"]) if "link_metadata" in msg else ()
271 )
273 for attachment in attachments:
274 if not body:
275 body = None
276 break
277 body = body.replace(attachment.url, "").strip()
279 if recipient.is_group:
280 mentions = tuple(await recipient.parse_mentions(body))
281 else:
282 mentions = ()
284 if attachments:
285 xmpp_msg: XMPPMessage = XMPPAttachmentMessage(
286 body=body,
287 attachments=attachments,
288 reply=reply,
289 thread=thread,
290 link_previews=link_previews,
291 mentions=mentions,
292 replace=replace,
293 )
294 else:
295 xmpp_msg = XMPPTextMessage(
296 body=body,
297 attachments=attachments,
298 reply=reply,
299 thread=thread,
300 link_previews=link_previews,
301 mentions=mentions,
302 replace=replace,
303 )
305 return await recipient.on_message(xmpp_msg)
307 @exceptions_to_xmpp_errors
308 async def on_message_retract(self, msg: Message) -> None:
309 recipient, thread = await self._get_recipient_and_thread(msg)
310 if not recipient.RETRACTION:
311 raise XMPPError(
312 "bad-request",
313 "This legacy service does not support message retraction.",
314 )
315 xmpp_id: str = msg["retract"]["id"]
316 legacy_id = self._xmpp_msg_id_to_legacy(xmpp_id, recipient, origin=True)
317 await recipient.on_retract(legacy_id, thread=thread)
318 if isinstance(recipient, LegacyMUC):
319 await recipient.echo(msg, None)
320 self.__ack(msg)
322 @exceptions_to_xmpp_errors
323 async def on_reactions(self, msg: Message) -> None:
324 recipient, thread = await self._get_recipient_and_thread(msg)
325 react_to: str = msg["reactions"]["id"]
327 special_msg = recipient.session.SPECIAL_MSG_ID_PREFIX and react_to.startswith(
328 recipient.session.SPECIAL_MSG_ID_PREFIX
329 )
331 if special_msg:
332 legacy_id = react_to
333 else:
334 legacy_id = self._xmpp_msg_id_to_legacy(react_to, recipient)
336 if not legacy_id:
337 log.debug("Ignored reaction from user")
338 raise XMPPError(
339 "internal-server-error",
340 "Could not convert the XMPP msg ID to a legacy ID",
341 )
343 emojis = [
344 remove_emoji_variation_selector_16(r["value"]) for r in msg["reactions"]
345 ]
346 error_msg = None
347 recipient = recipient
349 if not special_msg:
350 if recipient.REACTIONS_SINGLE_EMOJI and len(emojis) > 1:
351 error_msg = "Maximum 1 emoji/message"
353 if (
354 not error_msg
355 and (subset := await recipient.available_emojis(legacy_id))
356 and not set(emojis).issubset(subset)
357 ):
358 error_msg = (
359 f"You can only react with the following emojis: {''.join(subset)}"
360 )
362 if error_msg:
363 recipient.session.send_gateway_message(error_msg)
364 if not isinstance(recipient, LegacyMUC):
365 # no need to carbon for groups, we just don't echo the stanza
366 recipient.react(legacy_id, carbon=True)
367 await recipient.on_react(legacy_id, [], thread=thread)
368 raise XMPPError(
369 "policy-violation",
370 text=error_msg,
371 clear=False,
372 )
374 await recipient.on_react(legacy_id, emojis, thread=thread)
375 if isinstance(recipient, LegacyMUC):
376 await recipient.echo(msg, None)
377 else:
378 self.__ack(msg)
380 with self.xmpp.store.session() as orm:
381 multi = self.xmpp.store.id_map.get_xmpp(
382 orm, recipient.stored.id, legacy_id, recipient.is_group
383 )
384 if not multi:
385 return
386 multi = [m for m in multi if react_to != m]
388 if isinstance(recipient, LegacyMUC):
389 for xmpp_id in multi:
390 mc = copy(msg)
391 mc["reactions"]["id"] = xmpp_id
392 await recipient.echo(mc)
393 elif isinstance(recipient, LegacyContact):
394 for xmpp_id in multi:
395 recipient.react(legacy_id, emojis, xmpp_id=xmpp_id, carbon=True)
397 def __ack(self, msg: Message) -> None:
398 if not self.xmpp.PROPER_RECEIPTS:
399 self.xmpp.delivery_receipt.ack(msg)
401 async def __get_reply(
402 self, msg: Message, recipient: AnyRecipient
403 ) -> tuple[str, Reply | None]:
404 if "reply" not in msg:
405 return msg["body"], None
407 session = recipient.session
409 try:
410 reply_to_msg_id = self._xmpp_msg_id_to_legacy(msg["reply"]["id"], recipient)
411 except XMPPError:
412 session.log.debug(
413 "Could not determine reply-to legacy msg ID, sending quote instead."
414 )
415 return redact_url(msg["body"]), None
417 reply_to_jid = JID(msg["reply"]["to"])
418 reply_to = None
419 if msg["type"] == "chat":
420 if reply_to_jid.bare != session.user_jid.bare:
421 with contextlib.suppress(XMPPError):
422 reply_to = await session.contacts.by_jid(reply_to_jid)
423 elif msg["type"] == "groupchat":
424 nick = reply_to_jid.resource
425 try:
426 muc = await session.bookmarks.by_jid(reply_to_jid)
427 except XMPPError:
428 pass
429 else:
430 if nick == muc.user_nick:
431 reply_to = await muc.get_user_participant()
432 elif not nick:
433 reply_to = muc.get_system_participant()
434 else:
435 reply_to = await muc.get_participant(nick, store=False)
437 if "fallback" in msg and (
438 isinstance(recipient, LegacyMUC) or recipient.REPLIES
439 ):
440 text = msg["fallback"].get_stripped_body(
441 self.xmpp.plugin["xep_0461"].namespace
442 )
443 try:
444 reply_fallback = redact_url(msg["reply"].get_fallback_body())
445 except AttributeError:
446 reply_fallback = None
447 else:
448 text = msg["body"]
449 reply_fallback = None
451 return text, Reply(reply_to_msg_id, reply_to, reply_fallback)
453 async def __dispatch_nonbob_sticker(
454 self,
455 attachment: _IncomingAttachment,
456 recipient: AnyRecipient,
457 fallback: str,
458 reply: Reply | None = None,
459 thread: str | None = None,
460 ) -> str | None:
461 if attachment.cid:
462 with self.xmpp.store.session() as orm:
463 sticker = self.xmpp.store.bob.get_sticker(orm, attachment.cid)
464 else:
465 sticker = None
466 if sticker is None:
467 async with attachment.attachment.get() as response:
468 response.raise_for_status()
469 data = await response.read()
470 cid = "sha256+" + hashlib.sha256(data).hexdigest()
471 with self.xmpp.store.session() as orm:
472 sticker = self.xmpp.store.bob.set_sticker(
473 orm, cid, data, attachment.attachment.content_type
474 )
475 orm.commit()
476 if sticker is None:
477 return await recipient.on_message(
478 XMPPTextMessage(
479 body="\n".join([attachment.attachment.url, fallback]),
480 thread=thread,
481 reply=reply,
482 )
483 )
484 sticker.reply = reply
485 sticker.thread = thread
486 return await recipient.on_sticker(sticker)
488 async def __dispatch_bob(
489 self,
490 from_: JID,
491 cid: str,
492 recipient: AnyRecipient,
493 reply: Reply | None = None,
494 thread: str | None = None,
495 ) -> str | None:
496 with self.xmpp.store.session() as orm:
497 sticker = self.xmpp.store.bob.get_sticker(orm, cid)
498 if sticker is None:
499 await self.xmpp.plugin["xep_0231"].get_bob(
500 from_, cid, ifrom=self.xmpp.boundjid
501 )
502 with self.xmpp.store.session() as orm:
503 sticker = self.xmpp.store.bob.get_sticker(orm, cid)
504 assert sticker is not None
505 sticker.reply = reply
506 sticker.thread = thread
507 return await recipient.on_sticker(sticker)
510def redact_url(text: str) -> str:
511 needle = config.NO_UPLOAD_URL_PREFIX or config.UPLOAD_URL_PREFIX
512 if not needle:
513 return text
514 return text.replace(needle, "")
517def parse_link_previews(link_metadatas: list[LinkMetadata]) -> tuple[LinkPreview, ...]:
518 result = []
519 for link_metadata in link_metadatas:
520 preview: LinkPreview = dict_to_named_tuple(link_metadata, LinkPreview) # type:ignore[arg-type]
521 if (
522 preview.image
523 and isinstance(preview.image, str)
524 and preview.image.startswith("data:image/jpeg;base64,")
525 ):
526 try:
527 image = base64.b64decode(
528 preview.image.removeprefix("data:image/jpeg;base64,")
529 )
530 except Exception as e:
531 log.warning(
532 "Could not decode base64-encoded image: %s '%s'", e, preview.image
533 )
534 else:
535 preview = preview._replace(image=image)
536 result.append(preview)
537 return tuple(result)
540log = logging.getLogger(__name__)