Coverage for slidge/core/dispatcher/message/message.py: 86%
274 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-10 12:26 +0000
1import base64
2import hashlib
3import logging
4from copy import copy
5from typing import TYPE_CHECKING, Any, NamedTuple
6from xml.etree import ElementTree
8from slixmpp import JID, Message
9from slixmpp.exceptions import XMPPError
10from slixmpp.plugins.xep_0511.stanza import LinkMetadata
12from ....contact.contact import LegacyContact
13from ....room.room import LegacyRoom
14from ....util.types import (
15 AnyRecipient,
16 AnyRoom,
17 ContactMessageProtocol,
18 ContactSticker,
19 LinkPreview,
20 Reply,
21 RoomMessageProtocol,
22 RoomSticker,
23 XMPPAttachment,
24 XMPPMessage,
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
34class _IncomingAttachment(NamedTuple):
35 attachment: XMPPAttachment
36 cid: str | None = None
39class MessageContentMixin(DispatcherMixin):
40 __slots__: list[str] = []
41 xmpp: "AnyGateway" # not strictly necessary, but helps ty and ty is nice
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 Room 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].attachment.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 slix_msg=msg,
112 )
114 if isinstance(recipient, LegacyRoom):
115 await recipient.echo(msg, legacy_msg_id, replace=replace)
116 else:
117 self.__ack(msg)
119 if not legacy_msg_id:
120 return
122 with self.xmpp.store.session() as orm:
123 if recipient.is_group:
124 self.xmpp.store.id_map.set_origin(
125 orm, recipient._stored.id, legacy_msg_id, msg.get_id()
126 )
127 else:
128 self.xmpp.store.id_map.set_msg(
129 orm,
130 recipient._stored.id,
131 legacy_msg_id,
132 [msg.get_id()],
133 False,
134 )
135 orm.commit()
137 def __get_xhtml_sticker_cid(self, msg: Message) -> str | None:
138 if "html" not in msg:
139 return None
140 body = ElementTree.fromstring("<body>" + msg["html"].get_body() + "</body>")
141 p = body.findall("p")
142 if p is None:
143 return None
144 if len(p) != 1:
145 return None
146 if p[0].text is not None and p[0].text.strip():
147 return None
149 images = p[0].findall("img")
150 if len(images) != 1:
151 return None
152 # no text, single img ⇒ this is a sticker
153 # other cases should be interpreted as "custom emojis" in text
154 src = images[0].get("src")
155 if src is None:
156 return None
157 if src.startswith("cid:"):
158 return src.removeprefix("cid:")
159 return None
161 async def __get_replace[T: LegacyContact | AnyRoom](
162 self, msg: Message, recipient: T
163 ) -> str | None:
164 if "replace" not in msg or "id" not in msg["replace"]:
165 return None
166 return (
167 self._xmpp_msg_id_to_legacy(msg["replace"]["id"], recipient, True) or None
168 )
170 def __get_attachments(self, msg: Message) -> list[_IncomingAttachment]:
171 is_sticker = "sticker" in msg
173 if (
174 "sfs" in msg
175 and "sources" in msg["sfs"]
176 and "url-data" in msg["sfs"]["sources"]
177 and "target" in msg["sfs"]["sources"]["url-data"]
178 ):
179 # TODO: support "attach source in later message", cf https://xmpp.org/extensions/xep-0447.html#example-5
180 # TODO: support for other sources than URL
181 # TODO: support for multiattachments in single message.
182 # What do we do if is_sticker and multiple files?
183 content_type = cid = None
184 if "file" in msg["sfs"]:
185 content_type = msg["sfs"]["file"]["media-type"] or None
186 if "hash" in msg["sfs"]["file"]:
187 algo = msg["sfs"]["file"]["hash"]["algo"]
188 h = msg["sfs"]["file"]["hash"]["value"]
189 cid = f"{algo}+{h}" if algo and h else None
190 attachment = _IncomingAttachment(
191 XMPPAttachment(
192 url=msg["sfs"]["sources"]["url-data"]["target"],
193 http=self.xmpp.http,
194 is_sticker=is_sticker,
195 content_type=content_type,
196 ),
197 cid=cid,
198 )
199 return [attachment]
201 if "oob" in msg:
202 return [
203 _IncomingAttachment(
204 XMPPAttachment(
205 url=msg["oob"]["url"],
206 http=self.xmpp.http,
207 is_sticker=is_sticker,
208 ),
209 )
210 ]
212 if (
213 "reference" in msg
214 and "sims" in msg["reference"]
215 and "sources" in msg["reference"]["sims"]
216 ):
217 for source in msg["reference"]["sims"]["sources"]["substanzas"]:
218 if source["uri"].startswith("http"):
219 attachment = _IncomingAttachment(
220 XMPPAttachment(
221 url=source["uri"],
222 http=self.xmpp.http,
223 is_sticker=is_sticker,
224 ),
225 )
226 break
227 else:
228 return []
229 if "file" in msg["reference"]["sims"]:
230 attachment.attachment.content_type = msg["media-type"] or None
231 return [attachment]
233 return []
235 async def __dispatch_msg(
236 self,
237 *,
238 replace: str | None,
239 reply: Reply | None,
240 body: str | None,
241 attachments: tuple[XMPPAttachment, ...],
242 recipient: AnyRecipient,
243 thread: str | None,
244 slix_msg: Message,
245 ) -> str | None:
246 if replace:
247 if body is not None:
248 body = body.strip()
249 if not body and not attachments and recipient.RETRACTION:
250 await recipient.on_retract(replace, thread=thread)
251 return None
252 if not recipient.CORRECTION:
253 if recipient.RETRACTION:
254 await recipient.on_retract(replace, thread=thread)
255 replace = None
256 elif body:
257 body = "Correction:\n" + body
259 if not any([attachments, body]):
260 log.debug(
261 "Ignoring msg with id '%s' because no body or attachments found",
262 slix_msg.get_id(),
263 )
264 return None
265 link_previews = (
266 parse_link_previews(slix_msg["link_metadatas"])
267 if "link_metadata" in slix_msg
268 else ()
269 )
271 for attachment in attachments:
272 if not body:
273 body = None
274 break
275 body = body.replace(attachment.url, "").strip()
277 msg = XMPPMessage[Any](
278 body=body,
279 attachments=attachments,
280 reply=reply,
281 thread=thread,
282 link_previews=link_previews,
283 replace=replace,
284 )
285 if recipient.is_group:
286 msg.mentions = tuple(await recipient.parse_mentions(body))
287 assert isinstance(msg, RoomMessageProtocol)
288 else:
289 assert isinstance(msg, ContactMessageProtocol)
290 return await recipient.on_message(msg)
292 @exceptions_to_xmpp_errors
293 async def on_message_retract(self, msg: Message) -> None:
294 recipient, thread = await self._get_recipient_and_thread(msg)
295 if not recipient.RETRACTION:
296 raise XMPPError(
297 "bad-request",
298 "This legacy service does not support message retraction.",
299 )
300 xmpp_id: str = msg["retract"]["id"]
301 legacy_id = self._xmpp_msg_id_to_legacy(xmpp_id, recipient, origin=True)
302 await recipient.on_retract(legacy_id, thread=thread)
303 if isinstance(recipient, LegacyRoom):
304 await recipient.echo(msg, None)
305 self.__ack(msg)
307 @exceptions_to_xmpp_errors
308 async def on_reactions(self, msg: Message) -> None:
309 recipient, thread = await self._get_recipient_and_thread(msg)
310 react_to: str = msg["reactions"]["id"]
312 legacy_id = self._xmpp_msg_id_to_legacy(react_to, recipient)
314 if not legacy_id:
315 log.debug("Ignored reaction from user")
316 raise XMPPError(
317 "internal-server-error",
318 "Could not convert the XMPP msg ID to a legacy ID",
319 )
321 emojis = [
322 remove_emoji_variation_selector_16(r["value"]) for r in msg["reactions"]
323 ]
324 error_msg = None
326 if recipient.REACTIONS_SINGLE_EMOJI and len(emojis) > 1:
327 error_msg = "Maximum 1 emoji/message"
329 if (
330 not error_msg
331 and (subset := await recipient.available_emojis(legacy_id))
332 and not set(emojis).issubset(subset)
333 ):
334 error_msg = (
335 f"You can only react with the following emojis: {''.join(subset)}"
336 )
338 if error_msg:
339 recipient._session.send_gateway_message(error_msg)
340 if not isinstance(recipient, LegacyRoom):
341 # no need to carbon for groups, we just don't echo the stanza
342 recipient.react(legacy_id, carbon=True)
343 await recipient.on_react(legacy_id, [], thread=thread)
344 raise XMPPError(
345 "policy-violation",
346 text=error_msg,
347 clear=False,
348 )
350 await recipient.on_react(legacy_id, emojis, thread=thread)
351 if isinstance(recipient, LegacyRoom):
352 await recipient.echo(msg, None)
353 else:
354 self.__ack(msg)
356 with self.xmpp.store.session() as orm:
357 multi = self.xmpp.store.id_map.get_xmpp(
358 orm, recipient._stored.id, legacy_id, recipient.is_group
359 )
360 multi = [m for m in multi if react_to != m]
361 if not multi:
362 return
364 if isinstance(recipient, LegacyRoom):
365 for xmpp_id in multi:
366 mc = copy(msg)
367 mc["reactions"]["id"] = xmpp_id
368 await recipient.echo(mc)
369 elif isinstance(recipient, LegacyContact):
370 for xmpp_id in multi:
371 recipient.react(legacy_id, emojis, xmpp_id=xmpp_id, carbon=True)
373 def __ack(self, msg: Message) -> None:
374 if not self.xmpp.PROPER_RECEIPTS:
375 self.xmpp.delivery_receipt.ack(msg)
377 async def __get_reply(
378 self, msg: Message, recipient: AnyRecipient
379 ) -> tuple[str, Reply | None]:
380 if "reply" not in msg:
381 return msg["body"], None
383 session = recipient._session
385 try:
386 msg_id = self._xmpp_msg_id_to_legacy(msg["reply"]["id"], recipient)
387 except XMPPError:
388 session.log.debug(
389 "Could not determine reply-to legacy msg ID, sending quote instead."
390 )
391 return redact_url(msg["body"]), None
393 if "fallback" in msg and (
394 isinstance(recipient, LegacyRoom) or recipient.REPLIES
395 ):
396 text = msg["fallback"].get_stripped_body(
397 self.xmpp.plugin["xep_0461"].namespace
398 )
399 try:
400 fallback = redact_url(msg["reply"].get_fallback_body())
401 except AttributeError:
402 fallback = None
403 else:
404 text = msg["body"]
405 fallback = None
407 jid = JID(msg["reply"]["to"])
408 reply = Reply(msg_id, fallback)
409 if recipient.is_group:
410 if not jid:
411 recipient.log.warning(
412 "Invalid replied-to JID: %s, assuming reply to the system participant.",
413 jid,
414 )
415 resource = jid.resource
416 if resource:
417 await recipient.participants.ensure_filled()
418 participant = recipient.participants.get_by_resource(resource)
419 if participant is None:
420 participant = recipient.participants.create(
421 nick=resource, temp=True
422 )
423 reply.to = participant
424 else:
425 reply.to = recipient.get_system_participant()
426 else:
427 if jid.bare == session.user_jid.bare:
428 reply.to = "self"
429 elif jid.bare == recipient.jid.bare:
430 reply.to = "contact"
431 else:
432 recipient.log.warning(
433 "Invalid replied-to JID: %r, assuming reply to the contact (%r)",
434 jid,
435 recipient.jid,
436 )
438 return text, reply
440 async def __dispatch_nonbob_sticker(
441 self,
442 attachment: _IncomingAttachment,
443 recipient: AnyRecipient,
444 fallback: str,
445 reply: Reply | None = None,
446 thread: str | None = None,
447 ) -> str | None:
448 if attachment.cid:
449 with self.xmpp.store.session() as orm:
450 sticker = self.xmpp.store.bob.get_sticker(orm, attachment.cid)
451 else:
452 sticker = None
453 if sticker is None:
454 async with attachment.attachment.get() as response:
455 response.raise_for_status()
456 data = await response.read()
457 cid = "sha256+" + hashlib.sha256(data).hexdigest()
458 with self.xmpp.store.session() as orm:
459 sticker = self.xmpp.store.bob.set_sticker(
460 orm, cid, data, attachment.attachment.content_type
461 )
462 orm.commit()
463 if sticker is None:
464 msg = XMPPMessage[Any](
465 body=f"{attachment.attachment.url}\n{fallback}",
466 thread=thread,
467 reply=reply,
468 )
469 if recipient.is_group:
470 assert isinstance(msg, RoomMessageProtocol)
471 else:
472 assert isinstance(msg, ContactMessageProtocol)
473 return await recipient.on_message(msg)
475 sticker.thread = thread
476 sticker.reply = reply
477 return await recipient.on_sticker(sticker) # type:ignore[arg-type]
479 async def __dispatch_bob(
480 self,
481 from_: JID,
482 cid: str,
483 recipient: AnyRecipient,
484 reply: Reply | None = None,
485 thread: str | None = None,
486 ) -> str | None:
487 with self.xmpp.store.session() as orm:
488 sticker = self.xmpp.store.bob.get_sticker(orm, cid)
489 if sticker is None:
490 await self.xmpp.plugin["xep_0231"].get_bob(
491 from_, cid, ifrom=self.xmpp.boundjid
492 )
493 with self.xmpp.store.session() as orm:
494 sticker = self.xmpp.store.bob.get_sticker(orm, cid)
495 assert sticker is not None
496 sticker.reply = reply
497 sticker.thread = thread
498 if recipient.is_group:
499 assert isinstance(sticker, RoomSticker)
500 else:
501 assert isinstance(sticker, ContactSticker)
502 return await recipient.on_sticker(sticker)
505def redact_url(text: str) -> str:
506 needle = config.NO_UPLOAD_URL_PREFIX or config.UPLOAD_URL_PREFIX
507 if not needle:
508 return text
509 return text.replace(needle, "")
512def parse_link_previews(link_metadatas: list[LinkMetadata]) -> tuple[LinkPreview, ...]:
513 result = []
514 for link_metadata in link_metadatas:
515 preview: LinkPreview = dict_to_named_tuple(link_metadata, LinkPreview) # type:ignore[arg-type]
516 if (
517 preview.image
518 and isinstance(preview.image, str)
519 and preview.image.startswith("data:image/jpeg;base64,")
520 ):
521 try:
522 image = base64.b64decode(
523 preview.image.removeprefix("data:image/jpeg;base64,")
524 )
525 except Exception as e: # noqa: BLE001
526 log.warning(
527 "Could not decode base64-encoded image: %s '%s'", e, preview.image
528 )
529 else:
530 preview = preview._replace(image=image)
531 result.append(preview)
532 return tuple(result)
535log = logging.getLogger(__name__)