Coverage for slidge/core/dispatcher/message/message.py: 84%
281 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 18:29 +0000
1import base64
2import hashlib
3import logging
4from copy import copy
5from dataclasses import dataclass
6from typing import TYPE_CHECKING
7from xml.etree import ElementTree
9from slixmpp import JID, Message
10from slixmpp.exceptions import XMPPError
11from slixmpp.plugins.xep_0511.stanza import LinkMetadata
13from ....contact.contact import LegacyContact
14from ....group.room import LegacyMUC
15from ....util.types import (
16 AnyRecipient,
17 LinkPreview,
18 RecipientType,
19 Reply,
20 XMPPAttachment,
21 XMPPAttachmentMessage,
22 XMPPMessage,
23 XMPPTextMessage,
24)
25from ....util.util import dict_to_named_tuple, remove_emoji_variation_selector_16
26from ... import config
27from ..util import DispatcherMixin, exceptions_to_xmpp_errors
29if TYPE_CHECKING:
30 from slidge.util.types import AnyGateway
33@dataclass
34class _IncomingAttachment:
35 attachment: XMPPAttachment
36 is_sticker: bool
37 cid: str | None = None
40class MessageContentMixin(DispatcherMixin):
41 __slots__: list[str] = []
43 def __init__(self, xmpp: "AnyGateway") -> None:
44 super().__init__(xmpp)
45 xmpp.add_event_handler("legacy_message", self.on_legacy_message)
46 xmpp.add_event_handler("message_retract", self.on_message_retract)
47 xmpp.add_event_handler("groupchat_message", self.on_groupchat_message)
48 xmpp.add_event_handler("reactions", self.on_reactions)
50 async def on_groupchat_message(self, msg: Message) -> None:
51 await self.on_legacy_message(msg)
53 @exceptions_to_xmpp_errors
54 async def on_legacy_message(self, msg: Message) -> None:
55 """
56 Meant to be called from :class:`BaseGateway` only.
58 :param msg:
59 :return:
60 """
61 # we MUST not use `if m["replace"]["id"]` because it adds the tag if not
62 # present. this is a problem for MUC echoed messages
63 if "apply_to" in msg:
64 # ignore message retraction (handled by a specific method)
65 # this is an old version of the protocol that depends on the
66 # message fastening deprecated xep.
67 return
68 if "reactions" in msg:
69 # ignore message reaction fallback.
70 # the reaction itself is handled by self.react_from_msg().
71 return
72 if "retract" in msg:
73 # ignore message retraction fallback.
74 # the retraction itself is handled by self.on_retract
75 return
76 recipient, thread = await self._get_recipient_and_thread(msg)
77 replace = await self.__get_replace(msg, recipient)
78 if msg.xml.find(".//{*}encrypted") is not None:
79 raise XMPPError(
80 "bad-request", "You cannot send encrypted messages through this gateway"
81 )
82 body, reply = await self.__get_reply(msg, recipient)
83 cid = self.__get_xhtml_sticker_cid(msg)
85 if cid:
86 legacy_msg_id = await self.__dispatch_bob(
87 msg.get_from(),
88 cid,
89 recipient,
90 reply=reply,
91 thread=thread,
92 )
93 else:
94 attachments = self.__get_attachments(msg)
95 if len(attachments) == 1 and attachments[0].is_sticker:
96 legacy_msg_id = await self.__dispatch_nonbob_sticker(
97 attachments[0],
98 recipient,
99 msg["body"],
100 reply=reply,
101 thread=thread,
102 )
103 else:
104 legacy_msg_id = await self.__dispatch_msg(
105 replace=replace,
106 body=body,
107 attachments=tuple(a.attachment for a in attachments),
108 recipient=recipient,
109 thread=thread,
110 reply=reply,
111 msg=msg,
112 )
114 if isinstance(recipient, LegacyMUC):
115 stanza_id = await recipient.echo(msg, legacy_msg_id)
116 else:
117 stanza_id = None
118 self.__ack(msg)
120 if not legacy_msg_id:
121 return
123 with self.xmpp.store.session() as orm:
124 if recipient.is_group:
125 self.xmpp.store.id_map.set_origin(
126 orm, recipient.stored.id, legacy_msg_id, msg.get_id()
127 )
128 assert stanza_id is not None
129 self.xmpp.store.id_map.set_msg(
130 orm,
131 recipient.stored.id,
132 legacy_msg_id,
133 [stanza_id],
134 True,
135 )
136 else:
137 self.xmpp.store.id_map.set_msg(
138 orm,
139 recipient.stored.id,
140 legacy_msg_id,
141 [msg.get_id()],
142 False,
143 )
144 if recipient.session.MESSAGE_IDS_ARE_THREAD_IDS and thread:
145 self.xmpp.store.id_map.set_thread(
146 orm, recipient.stored.id, thread, legacy_msg_id, recipient.is_group
147 )
148 orm.commit()
150 def __get_xhtml_sticker_cid(self, msg: Message) -> str | None:
151 if "html" not in msg:
152 return None
153 body = ElementTree.fromstring("<body>" + msg["html"].get_body() + "</body>")
154 p = body.findall("p")
155 if p is None:
156 return None
157 if len(p) != 1:
158 return None
159 if p[0].text is not None and p[0].text.strip():
160 return None
162 images = p[0].findall("img")
163 if len(images) != 1:
164 return None
165 # no text, single img ⇒ this is a sticker
166 # other cases should be interpreted as "custom emojis" in text
167 src = images[0].get("src")
168 if src is None:
169 return None
170 if src.startswith("cid:"):
171 return src.removeprefix("cid:")
172 return None
174 async def __get_replace(
175 self,
176 msg: Message,
177 recipient: RecipientType,
178 ) -> str | None:
179 if "replace" not in msg or "id" not in msg["replace"]:
180 return None
181 return (
182 self._xmpp_msg_id_to_legacy(msg["replace"]["id"], recipient, True) or None
183 )
185 def __get_attachments(self, msg: Message) -> list[_IncomingAttachment]:
186 is_sticker = "sticker" in msg
188 if (
189 "sfs" in msg
190 and "sources" in msg["sfs"]
191 and "url-data" in msg["sfs"]["sources"]
192 and "target" in msg["sfs"]["sources"]["url-data"]
193 ):
194 # TODO: support "attach source in later message", cf https://xmpp.org/extensions/xep-0447.html#example-5
195 # TODO: support for other sources than URL
196 # TODO: support for multiattachments in single message.
197 # What do we do if is_sticker and multiple files?
198 attachment = _IncomingAttachment(
199 XMPPAttachment(url=msg["sfs"]["sources"]["url-data"]["target"]),
200 is_sticker=is_sticker,
201 )
202 if "file" in msg["sfs"]:
203 attachment.attachment.content_type = (
204 msg["sfs"]["file"]["media-type"] or None
205 )
206 if "hash" in msg["sfs"]["file"]:
207 algo = msg["sfs"]["file"]["hash"]["algo"]
208 h = msg["sfs"]["file"]["hash"]["value"]
209 attachment.cid = f"{algo}+{h}" if algo and h else None
210 return [attachment]
212 if "oob" in msg:
213 return [
214 _IncomingAttachment(
215 XMPPAttachment(url=msg["oob"]["url"]), is_sticker=is_sticker
216 )
217 ]
219 if (
220 "reference" in msg
221 and "sims" in msg["reference"]
222 and "sources" in msg["reference"]["sims"]
223 ):
224 for source in msg["reference"]["sims"]["sources"]["substanzas"]:
225 if source["uri"].startswith("http"):
226 attachment = _IncomingAttachment(
227 XMPPAttachment(url=source["uri"]), is_sticker=is_sticker
228 )
229 break
230 else:
231 return []
232 if "file" in msg["reference"]["sims"]:
233 attachment.attachment.content_type = msg["media-type"] or None
234 return [attachment]
236 return []
238 async def __dispatch_msg(
239 self,
240 *,
241 replace: str | None,
242 reply: Reply | None,
243 body: str | None,
244 attachments: tuple[XMPPAttachment, ...],
245 recipient: AnyRecipient,
246 thread: str | None,
247 msg: Message,
248 ) -> str | None:
249 if replace:
250 if body is not None:
251 body = body.strip()
252 if not body and not attachments and recipient.RETRACTION:
253 await recipient.on_retract(replace, thread=thread)
254 return None
255 if not recipient.CORRECTION:
256 if recipient.RETRACTION:
257 await recipient.on_retract(replace, thread=thread)
258 replace = None
259 elif body:
260 body = "Correction:\n" + body
262 if not any([attachments, body]):
263 log.debug(
264 "Ignoring msg with id '%s' because no body or attachments found",
265 msg.get_id(),
266 )
267 return None
268 link_previews = (
269 parse_link_previews(msg["link_metadatas"]) if "link_metadata" in msg else ()
270 )
272 for attachment in attachments:
273 if not body:
274 body = None
275 break
276 body = body.replace(attachment.url, "").strip()
278 if recipient.is_group:
279 mentions = tuple(await recipient.parse_mentions(body))
280 else:
281 mentions = ()
283 if attachments:
284 xmpp_msg: XMPPMessage = XMPPAttachmentMessage(
285 body=body,
286 attachments=attachments,
287 reply=reply,
288 thread=thread,
289 link_previews=link_previews,
290 mentions=mentions,
291 replace=replace,
292 )
293 else:
294 xmpp_msg = XMPPTextMessage(
295 body=body,
296 attachments=attachments,
297 reply=reply,
298 thread=thread,
299 link_previews=link_previews,
300 mentions=mentions,
301 replace=replace,
302 )
304 return await recipient.on_message(xmpp_msg)
306 @exceptions_to_xmpp_errors
307 async def on_message_retract(self, msg: Message) -> None:
308 recipient, thread = await self._get_recipient_and_thread(msg)
309 if not recipient.RETRACTION:
310 raise XMPPError(
311 "bad-request",
312 "This legacy service does not support message retraction.",
313 )
314 xmpp_id: str = msg["retract"]["id"]
315 legacy_id = self._xmpp_msg_id_to_legacy(xmpp_id, recipient, origin=True)
316 await recipient.on_retract(legacy_id, thread=thread)
317 if isinstance(recipient, LegacyMUC):
318 await recipient.echo(msg, None)
319 self.__ack(msg)
321 @exceptions_to_xmpp_errors
322 async def on_reactions(self, msg: Message) -> None:
323 recipient, thread = await self._get_recipient_and_thread(msg)
324 react_to: str = msg["reactions"]["id"]
326 special_msg = recipient.session.SPECIAL_MSG_ID_PREFIX and react_to.startswith(
327 recipient.session.SPECIAL_MSG_ID_PREFIX
328 )
330 if special_msg:
331 legacy_id = react_to
332 else:
333 legacy_id = self._xmpp_msg_id_to_legacy(react_to, recipient)
335 if not legacy_id:
336 log.debug("Ignored reaction from user")
337 raise XMPPError(
338 "internal-server-error",
339 "Could not convert the XMPP msg ID to a legacy ID",
340 )
342 emojis = [
343 remove_emoji_variation_selector_16(r["value"]) for r in msg["reactions"]
344 ]
345 error_msg = None
347 if not special_msg:
348 if recipient.REACTIONS_SINGLE_EMOJI and len(emojis) > 1:
349 error_msg = "Maximum 1 emoji/message"
351 if (
352 not error_msg
353 and (subset := await recipient.available_emojis(legacy_id))
354 and not set(emojis).issubset(subset)
355 ):
356 error_msg = (
357 f"You can only react with the following emojis: {''.join(subset)}"
358 )
360 if error_msg:
361 recipient.session.send_gateway_message(error_msg)
362 if not isinstance(recipient, LegacyMUC):
363 # no need to carbon for groups, we just don't echo the stanza
364 recipient.react(legacy_id, carbon=True)
365 await recipient.on_react(legacy_id, [], thread=thread)
366 raise XMPPError(
367 "policy-violation",
368 text=error_msg,
369 clear=False,
370 )
372 await recipient.on_react(legacy_id, emojis, thread=thread)
373 if isinstance(recipient, LegacyMUC):
374 await recipient.echo(msg, None)
375 else:
376 self.__ack(msg)
378 with self.xmpp.store.session() as orm:
379 multi = self.xmpp.store.id_map.get_xmpp(
380 orm, recipient.stored.id, legacy_id, recipient.is_group
381 )
382 if not multi:
383 return
384 multi = [m for m in multi if react_to != m]
386 if isinstance(recipient, LegacyMUC):
387 for xmpp_id in multi:
388 mc = copy(msg)
389 mc["reactions"]["id"] = xmpp_id
390 await recipient.echo(mc)
391 elif isinstance(recipient, LegacyContact):
392 for xmpp_id in multi:
393 recipient.react(legacy_id, emojis, xmpp_id=xmpp_id, carbon=True)
395 def __ack(self, msg: Message) -> None:
396 if not self.xmpp.PROPER_RECEIPTS:
397 self.xmpp.delivery_receipt.ack(msg)
399 async def __get_reply(
400 self, msg: Message, recipient: AnyRecipient
401 ) -> tuple[str, Reply | None]:
402 if "reply" not in msg:
403 return msg["body"], None
405 session = recipient.session
407 try:
408 reply_to_msg_id = self._xmpp_msg_id_to_legacy(msg["reply"]["id"], recipient)
409 except XMPPError:
410 session.log.debug(
411 "Could not determine reply-to legacy msg ID, sending quote instead."
412 )
413 return redact_url(msg["body"]), None
415 reply_to_jid = JID(msg["reply"]["to"])
416 reply_to = None
417 if msg["type"] == "chat":
418 if reply_to_jid.bare != session.user_jid.bare:
419 try:
420 reply_to = await session.contacts.by_jid(reply_to_jid)
421 except XMPPError:
422 session.log.exception("Could not instantiate replied-to contact")
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 session.log.exception("Could not instantiate replied-to participant")
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=f"{attachment.attachment.url}\n{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: # noqa: BLE001
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__)