Coverage for slidge/core/dispatcher/message/message.py: 83%

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 ContactMessage,

18 ContactMessageProtocol,

19 ContactSticker,

20 LinkPreview,

21 MUCMessage,

22 MUCMessageProtocol,

23 MUCSticker,

24 RecipientType,

25 Reply,

26 XMPPAttachment,

27 XMPPMessage,

28)

29from ....util.util import dict_to_named_tuple, remove_emoji_variation_selector_16

30from ... import config

31from ..util import DispatcherMixin, exceptions_to_xmpp_errors

33if TYPE_CHECKING:

34 from slidge.util.types import AnyGateway

37@dataclass

38class _IncomingAttachment:

39 attachment: XMPPAttachment

40 is_sticker: bool

41 cid: str | None = None

44class MessageContentMixin(DispatcherMixin):

45 __slots__: list[str] = []

47 def __init__(self, xmpp: "AnyGateway") -> None:

48 super().__init__(xmpp)

49 xmpp.add_event_handler("legacy_message", self.on_legacy_message)

50 xmpp.add_event_handler("message_retract", self.on_message_retract)

51 xmpp.add_event_handler("groupchat_message", self.on_groupchat_message)

52 xmpp.add_event_handler("reactions", self.on_reactions)

54 async def on_groupchat_message(self, msg: Message) -> None:

55 await self.on_legacy_message(msg)

57 @exceptions_to_xmpp_errors

58 async def on_legacy_message(self, msg: Message) -> None:

59 """

60 Meant to be called from :class:`BaseGateway` only.

62 :param msg:

63 :return:

64 """

65 # we MUST not use `if m["replace"]["id"]` because it adds the tag if not

66 # present. this is a problem for MUC echoed messages

67 if "apply_to" in msg:

68 # ignore message retraction (handled by a specific method)

69 # this is an old version of the protocol that depends on the

70 # message fastening deprecated xep.

71 return

72 if "reactions" in msg:

73 # ignore message reaction fallback.

74 # the reaction itself is handled by self.react_from_msg().

75 return

76 if "retract" in msg:

77 # ignore message retraction fallback.

78 # the retraction itself is handled by self.on_retract

79 return

80 recipient, thread = await self._get_recipient_and_thread(msg)

81 replace = await self.__get_replace(msg, recipient)

82 if msg.xml.find(".//{*}encrypted") is not None:

83 raise XMPPError(

84 "bad-request", "You cannot send encrypted messages through this gateway"

85 )

86 body, reply = await self.__get_reply(msg, recipient)

87 cid = self.__get_xhtml_sticker_cid(msg)

89 if cid:

90 legacy_msg_id = await self.__dispatch_bob(

91 msg.get_from(),

92 cid,

93 recipient,

94 reply=reply,

95 thread=thread,

96 )

97 else:

98 attachments = self.__get_attachments(msg)

99 if len(attachments) == 1 and attachments[0].is_sticker:

100 legacy_msg_id = await self.__dispatch_nonbob_sticker(

101 attachments[0],

102 recipient,

103 msg["body"],

104 reply=reply,

105 thread=thread,

106 )

107 else:

108 legacy_msg_id = await self.__dispatch_msg(

109 replace=replace,

110 body=body,

111 attachments=tuple(a.attachment for a in attachments),

112 recipient=recipient,

113 thread=thread,

114 reply=reply,

115 slix_msg=msg,

116 )

117

118 if isinstance(recipient, LegacyMUC):

119 stanza_id = await recipient.echo(msg, legacy_msg_id)

120 else:

121 stanza_id = None

122 self.__ack(msg)

123

124 if not legacy_msg_id:

125 return

126

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

128 if recipient.is_group:

129 self.xmpp.store.id_map.set_origin(

130 orm, recipient.stored.id, legacy_msg_id, msg.get_id()

131 )

132 assert stanza_id is not None

133 self.xmpp.store.id_map.set_msg(

134 orm,

135 recipient.stored.id,

136 legacy_msg_id,

137 [stanza_id],

138 True,

139 )

140 else:

141 self.xmpp.store.id_map.set_msg(

142 orm,

143 recipient.stored.id,

144 legacy_msg_id,

145 [msg.get_id()],

146 False,

147 )

148 if recipient.session.MESSAGE_IDS_ARE_THREAD_IDS and thread:

149 self.xmpp.store.id_map.set_thread(

150 orm, recipient.stored.id, thread, legacy_msg_id, recipient.is_group

151 )

152 orm.commit()

153

154 def __get_xhtml_sticker_cid(self, msg: Message) -> str | None:

155 if "html" not in msg:

156 return None

157 body = ElementTree.fromstring("<body>" + msg["html"].get_body() + "</body>")

158 p = body.findall("p")

159 if p is None:

160 return None

161 if len(p) != 1:

162 return None

163 if p[0].text is not None and p[0].text.strip():

164 return None

165

166 images = p[0].findall("img")

167 if len(images) != 1:

168 return None

169 # no text, single img ⇒ this is a sticker

170 # other cases should be interpreted as "custom emojis" in text

171 src = images[0].get("src")

172 if src is None:

173 return None

174 if src.startswith("cid:"):

175 return src.removeprefix("cid:")

176 return None

177

178 async def __get_replace(

179 self,

180 msg: Message,

181 recipient: RecipientType,

182 ) -> str | None:

183 if "replace" not in msg or "id" not in msg["replace"]:

184 return None

185 return (

186 self._xmpp_msg_id_to_legacy(msg["replace"]["id"], recipient, True) or None

187 )

188

189 def __get_attachments(self, msg: Message) -> list[_IncomingAttachment]:

190 is_sticker = "sticker" in msg

191

192 if (

193 "sfs" in msg

194 and "sources" in msg["sfs"]

195 and "url-data" in msg["sfs"]["sources"]

196 and "target" in msg["sfs"]["sources"]["url-data"]

197 ):

198 # TODO: support "attach source in later message", cf https://xmpp.org/extensions/xep-0447.html#example-5

199 # TODO: support for other sources than URL

200 # TODO: support for multiattachments in single message.

201 # What do we do if is_sticker and multiple files?

202 attachment = _IncomingAttachment(

203 XMPPAttachment(url=msg["sfs"]["sources"]["url-data"]["target"]),

204 is_sticker=is_sticker,

205 )

206 if "file" in msg["sfs"]:

207 attachment.attachment.content_type = (

208 msg["sfs"]["file"]["media-type"] or None

209 )

210 if "hash" in msg["sfs"]["file"]:

211 algo = msg["sfs"]["file"]["hash"]["algo"]

212 h = msg["sfs"]["file"]["hash"]["value"]

213 attachment.cid = f"{algo}+{h}" if algo and h else None

214 return [attachment]

215

216 if "oob" in msg:

217 return [

218 _IncomingAttachment(

219 XMPPAttachment(url=msg["oob"]["url"]), is_sticker=is_sticker

220 )

221 ]

222

223 if (

224 "reference" in msg

225 and "sims" in msg["reference"]

226 and "sources" in msg["reference"]["sims"]

227 ):

228 for source in msg["reference"]["sims"]["sources"]["substanzas"]:

229 if source["uri"].startswith("http"):

230 attachment = _IncomingAttachment(

231 XMPPAttachment(url=source["uri"]), is_sticker=is_sticker

232 )

233 break

234 else:

235 return []

236 if "file" in msg["reference"]["sims"]:

237 attachment.attachment.content_type = msg["media-type"] or None

238 return [attachment]

239

240 return []

241

242 async def __dispatch_msg(

243 self,

244 *,

245 replace: str | None,

246 reply: Reply | None,

247 body: str | None,

248 attachments: tuple[XMPPAttachment, ...],

249 recipient: AnyRecipient,

250 thread: str | None,

251 slix_msg: Message,

252 ) -> str | None:

253 if replace:

254 if body is not None:

255 body = body.strip()

256 if not body and not attachments and recipient.RETRACTION:

257 await recipient.on_retract(replace, thread=thread)

258 return None

259 if not recipient.CORRECTION:

260 if recipient.RETRACTION:

261 await recipient.on_retract(replace, thread=thread)

262 replace = None

263 elif body:

264 body = "Correction:\n" + body

265

266 if not any([attachments, body]):

267 log.debug(

268 "Ignoring msg with id '%s' because no body or attachments found",

269 slix_msg.get_id(),

270 )

271 return None

272 link_previews = (

273 parse_link_previews(slix_msg["link_metadatas"])

274 if "link_metadata" in slix_msg

275 else ()

276 )

277

278 for attachment in attachments:

279 if not body:

280 body = None

281 break

282 body = body.replace(attachment.url, "").strip()

283

284 msg = XMPPMessage(

285 body=body,

286 attachments=attachments,

287 reply=reply,

288 thread=thread,

289 link_previews=link_previews,

290 replace=replace,

291 )

292 if recipient.is_group:

293 msg.mentions = tuple(await recipient.parse_mentions(body))

294 assert isinstance(msg, MUCMessageProtocol)

295 else:

296 assert isinstance(msg, ContactMessageProtocol)

297 return await recipient.on_message(msg)

298

299 @exceptions_to_xmpp_errors

300 async def on_message_retract(self, msg: Message) -> None:

301 recipient, thread = await self._get_recipient_and_thread(msg)

302 if not recipient.RETRACTION:

303 raise XMPPError(

304 "bad-request",

305 "This legacy service does not support message retraction.",

306 )

307 xmpp_id: str = msg["retract"]["id"]

308 legacy_id = self._xmpp_msg_id_to_legacy(xmpp_id, recipient, origin=True)

309 await recipient.on_retract(legacy_id, thread=thread)

310 if isinstance(recipient, LegacyMUC):

311 await recipient.echo(msg, None)

312 self.__ack(msg)

313

314 @exceptions_to_xmpp_errors

315 async def on_reactions(self, msg: Message) -> None:

316 recipient, thread = await self._get_recipient_and_thread(msg)

317 react_to: str = msg["reactions"]["id"]

318

319 special_msg = recipient.session.SPECIAL_MSG_ID_PREFIX and react_to.startswith(

320 recipient.session.SPECIAL_MSG_ID_PREFIX

321 )

322

323 if special_msg:

324 legacy_id = react_to

325 else:

326 legacy_id = self._xmpp_msg_id_to_legacy(react_to, recipient)

327

328 if not legacy_id:

329 log.debug("Ignored reaction from user")

330 raise XMPPError(

331 "internal-server-error",

332 "Could not convert the XMPP msg ID to a legacy ID",

333 )

334

335 emojis = [

336 remove_emoji_variation_selector_16(r["value"]) for r in msg["reactions"]

337 ]

338 error_msg = None

339

340 if not special_msg:

341 if recipient.REACTIONS_SINGLE_EMOJI and len(emojis) > 1:

342 error_msg = "Maximum 1 emoji/message"

343

344 if (

345 not error_msg

346 and (subset := await recipient.available_emojis(legacy_id))

347 and not set(emojis).issubset(subset)

348 ):

349 error_msg = (

350 f"You can only react with the following emojis: {''.join(subset)}"

351 )

352

353 if error_msg:

354 recipient.session.send_gateway_message(error_msg)

355 if not isinstance(recipient, LegacyMUC):

356 # no need to carbon for groups, we just don't echo the stanza

357 recipient.react(legacy_id, carbon=True)

358 await recipient.on_react(legacy_id, [], thread=thread)

359 raise XMPPError(

360 "policy-violation",

361 text=error_msg,

362 clear=False,

363 )

364

365 await recipient.on_react(legacy_id, emojis, thread=thread)

366 if isinstance(recipient, LegacyMUC):

367 await recipient.echo(msg, None)

368 else:

369 self.__ack(msg)

370

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

372 multi = self.xmpp.store.id_map.get_xmpp(

373 orm, recipient.stored.id, legacy_id, recipient.is_group

374 )

375 if not multi:

376 return

377 multi = [m for m in multi if react_to != m]

378

379 if isinstance(recipient, LegacyMUC):

380 for xmpp_id in multi:

381 mc = copy(msg)

382 mc["reactions"]["id"] = xmpp_id

383 await recipient.echo(mc)

384 elif isinstance(recipient, LegacyContact):

385 for xmpp_id in multi:

386 recipient.react(legacy_id, emojis, xmpp_id=xmpp_id, carbon=True)

387

388 def __ack(self, msg: Message) -> None:

389 if not self.xmpp.PROPER_RECEIPTS:

390 self.xmpp.delivery_receipt.ack(msg)

391

392 async def __get_reply(

393 self, msg: Message, recipient: AnyRecipient

394 ) -> tuple[str, Reply | None]:

395 if "reply" not in msg:

396 return msg["body"], None

397

398 session = recipient.session

399

400 try:

401 msg_id = self._xmpp_msg_id_to_legacy(msg["reply"]["id"], recipient)

402 except XMPPError:

403 session.log.debug(

404 "Could not determine reply-to legacy msg ID, sending quote instead."

405 )

406 return redact_url(msg["body"]), None

407

408 if "fallback" in msg and (

409 isinstance(recipient, LegacyMUC) or recipient.REPLIES

410 ):

411 text = msg["fallback"].get_stripped_body(

412 self.xmpp.plugin["xep_0461"].namespace

413 )

414 try:

415 fallback = redact_url(msg["reply"].get_fallback_body())

416 except AttributeError:

417 fallback = None

418 else:

419 text = msg["body"]

420 fallback = None

421

422 jid = JID(msg["reply"]["to"])

423 reply = Reply(msg_id, fallback)

424 if recipient.is_group:

425 if not jid:

426 recipient.log.warning(

427 "Invalid replied-to JID: %s, assuming reply to the system participant.",

428 jid,

429 )

430 nick = jid.resource

431 if nick == recipient.user_nick:

432 reply.to = await recipient.get_user_participant()

433 elif not nick:

434 reply.to = recipient.get_system_participant()

435 else:

436 reply.to = await recipient.get_participant(nick, store=False)

437 else:

438 if jid.bare == session.user_jid.bare:

439 reply.to = "self"

440 elif jid.bare == recipient.jid:

441 reply.to = "contact"

442 else:

443 recipient.log.warning(

444 "Invalid replied-to JID: %s, assuming reply to the contact.",

445 jid,

446 )

447

448 return text, reply

449

450 async def __dispatch_nonbob_sticker(

451 self,

452 attachment: _IncomingAttachment,

453 recipient: AnyRecipient,

454 fallback: str,

455 reply: Reply | None = None,

456 thread: str | None = None,

457 ) -> str | None:

458 if attachment.cid:

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

460 sticker = self.xmpp.store.bob.get_sticker(orm, attachment.cid)

461 else:

462 sticker = None

463 if sticker is None:

464 async with attachment.attachment.get() as response:

465 response.raise_for_status()

466 data = await response.read()

467 cid = "sha256+" + hashlib.sha256(data).hexdigest()

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

469 sticker = self.xmpp.store.bob.set_sticker(

470 orm, cid, data, attachment.attachment.content_type

471 )

472 orm.commit()

473 if sticker is None:

474 msg = XMPPMessage(

475 body=f"{attachment.attachment.url}\n{fallback}",

476 thread=thread,

477 reply=reply,

478 )

479 if recipient.is_group:

480 assert isinstance(msg, MUCMessage)

481 else:

482 assert isinstance(msg, ContactMessage)

483 return await recipient.on_message(msg)

484

485 sticker.thread = thread

486 sticker.reply = reply

487 return await recipient.on_sticker(sticker) # type:ignore[arg-type]

488

489 async def __dispatch_bob(

490 self,

491 from_: JID,

492 cid: str,

493 recipient: AnyRecipient,

494 reply: Reply | None = None,

495 thread: str | None = None,

496 ) -> str | None:

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

498 sticker = self.xmpp.store.bob.get_sticker(orm, cid)

499 if sticker is None:

500 await self.xmpp.plugin["xep_0231"].get_bob(

501 from_, cid, ifrom=self.xmpp.boundjid

502 )

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

504 sticker = self.xmpp.store.bob.get_sticker(orm, cid)

505 assert sticker is not None

506 sticker.reply = reply

507 sticker.thread = thread

508 if recipient.is_group:

509 assert isinstance(sticker, MUCSticker)

510 else:

511 assert isinstance(sticker, ContactSticker)

512 return await recipient.on_sticker(sticker)

513

514

515def redact_url(text: str) -> str:

516 needle = config.NO_UPLOAD_URL_PREFIX or config.UPLOAD_URL_PREFIX

517 if not needle:

518 return text

519 return text.replace(needle, "")

520

521

522def parse_link_previews(link_metadatas: list[LinkMetadata]) -> tuple[LinkPreview, ...]:

523 result = []

524 for link_metadata in link_metadatas:

525 preview: LinkPreview = dict_to_named_tuple(link_metadata, LinkPreview) # type:ignore[arg-type]

526 if (

527 preview.image

528 and isinstance(preview.image, str)

529 and preview.image.startswith("data:image/jpeg;base64,")

530 ):

531 try:

532 image = base64.b64decode(

533 preview.image.removeprefix("data:image/jpeg;base64,")

534 )

535 except Exception as e: # noqa: BLE001

536 log.warning(

537 "Could not decode base64-encoded image: %s '%s'", e, preview.image

538 )

539 else:

540 preview = preview._replace(image=image)

541 result.append(preview)

542 return tuple(result)

543

544

545log = logging.getLogger(__name__)