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

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 ....group.room import LegacyMUC

14from ....util.types import (

15 AnyRecipient,

16 ContactMessageProtocol,

17 ContactSticker,

18 LinkPreview,

19 MUCMessageProtocol,

20 MUCSticker,

21 RecipientType,

22 Reply,

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 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].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 )

113

114 if isinstance(recipient, LegacyMUC):

115 await recipient.echo(msg, legacy_msg_id)

116 else:

117 self.__ack(msg)

118

119 if not legacy_msg_id:

120 return

121

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 if (

136 self.xmpp.THREADS

137 and recipient.session.MESSAGE_IDS_ARE_THREAD_IDS

138 and thread

139 ):

140 self.xmpp.store.id_map.set_thread(

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

142 )

143 orm.commit()

144

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

146 if "html" not in msg:

147 return None

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

149 p = body.findall("p")

150 if p is None:

151 return None

152 if len(p) != 1:

153 return None

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

155 return None

156

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

158 if len(images) != 1:

159 return None

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

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

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

163 if src is None:

164 return None

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

166 return src.removeprefix("cid:")

167 return None

168

169 async def __get_replace(

170 self,

171 msg: Message,

172 recipient: RecipientType,

173 ) -> str | None:

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

175 return None

176 return (

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

178 )

179

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

181 is_sticker = "sticker" in msg

182

183 if (

184 "sfs" in msg

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

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

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

188 ):

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

190 # TODO: support for other sources than URL

191 # TODO: support for multiattachments in single message.

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

193 content_type = cid = None

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

195 content_type = msg["sfs"]["file"]["media-type"] or None

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

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

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

199 cid = f"{algo}+{h}" if algo and h else None

200 attachment = _IncomingAttachment(

201 XMPPAttachment(

202 url=msg["sfs"]["sources"]["url-data"]["target"],

203 http=self.xmpp.http,

204 is_sticker=is_sticker,

205 content_type=content_type,

206 ),

207 cid=cid,

208 )

209 return [attachment]

210

211 if "oob" in msg:

212 return [

213 _IncomingAttachment(

214 XMPPAttachment(

215 url=msg["oob"]["url"],

216 http=self.xmpp.http,

217 is_sticker=is_sticker,

218 ),

219 )

220 ]

221

222 if (

223 "reference" in msg

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

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

226 ):

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

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

229 attachment = _IncomingAttachment(

230 XMPPAttachment(

231 url=source["uri"],

232 http=self.xmpp.http,

233 is_sticker=is_sticker,

234 ),

235 )

236 break

237 else:

238 return []

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

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

241 return [attachment]

242

243 return []

244

245 async def __dispatch_msg(

246 self,

247 *,

248 replace: str | None,

249 reply: Reply | None,

250 body: str | None,

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

252 recipient: AnyRecipient,

253 thread: str | None,

254 slix_msg: Message,

255 ) -> str | None:

256 if replace:

257 if body is not None:

258 body = body.strip()

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

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

261 return None

262 if not recipient.CORRECTION:

263 if recipient.RETRACTION:

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

265 replace = None

266 elif body:

267 body = "Correction:\n" + body

268

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

270 log.debug(

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

272 slix_msg.get_id(),

273 )

274 return None

275 link_previews = (

276 parse_link_previews(slix_msg["link_metadatas"])

277 if "link_metadata" in slix_msg

278 else ()

279 )

280

281 for attachment in attachments:

282 if not body:

283 body = None

284 break

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

286

287 msg = XMPPMessage[Any](

288 body=body,

289 attachments=attachments,

290 reply=reply,

291 thread=thread,

292 link_previews=link_previews,

293 replace=replace,

294 )

295 if recipient.is_group:

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

297 assert isinstance(msg, MUCMessageProtocol)

298 else:

299 assert isinstance(msg, ContactMessageProtocol)

300 return await recipient.on_message(msg)

301

302 @exceptions_to_xmpp_errors

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

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

305 if not recipient.RETRACTION:

306 raise XMPPError(

307 "bad-request",

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

309 )

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

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

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

313 if isinstance(recipient, LegacyMUC):

314 await recipient.echo(msg, None)

315 self.__ack(msg)

316

317 @exceptions_to_xmpp_errors

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

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

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

321

322 legacy_id = self._xmpp_msg_id_to_legacy(react_to, recipient)

323

324 if not legacy_id:

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

326 raise XMPPError(

327 "internal-server-error",

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

329 )

330

331 emojis = [

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

333 ]

334 error_msg = None

335

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

337 error_msg = "Maximum 1 emoji/message"

338

339 if (

340 not error_msg

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

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

343 ):

344 error_msg = (

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

346 )

347

348 if error_msg:

349 recipient.session.send_gateway_message(error_msg)

350 if not isinstance(recipient, LegacyMUC):

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

352 recipient.react(legacy_id, carbon=True)

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

354 raise XMPPError(

355 "policy-violation",

356 text=error_msg,

357 clear=False,

358 )

359

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

361 if isinstance(recipient, LegacyMUC):

362 await recipient.echo(msg, None)

363 else:

364 self.__ack(msg)

365

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

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

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

369 )

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

371 if not multi:

372 return

373

374 if isinstance(recipient, LegacyMUC):

375 for xmpp_id in multi:

376 mc = copy(msg)

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

378 await recipient.echo(mc)

379 elif isinstance(recipient, LegacyContact):

380 for xmpp_id in multi:

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

382

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

384 if not self.xmpp.PROPER_RECEIPTS:

385 self.xmpp.delivery_receipt.ack(msg)

386

387 async def __get_reply(

388 self, msg: Message, recipient: AnyRecipient

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

390 if "reply" not in msg:

391 return msg["body"], None

392

393 session = recipient.session

394

395 try:

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

397 except XMPPError:

398 session.log.debug(

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

400 )

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

402

403 if "fallback" in msg and (

404 isinstance(recipient, LegacyMUC) or recipient.REPLIES

405 ):

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

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

408 )

409 try:

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

411 except AttributeError:

412 fallback = None

413 else:

414 text = msg["body"]

415 fallback = None

416

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

418 reply = Reply(msg_id, fallback)

419 if recipient.is_group:

420 if not jid:

421 recipient.log.warning(

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

423 jid,

424 )

425 nick = jid.resource

426 if nick == recipient.user_nick:

427 reply.to = await recipient.get_user_participant()

428 elif not nick:

429 reply.to = recipient.get_system_participant()

430 else:

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

432 else:

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

434 reply.to = "self"

435 elif jid.bare == recipient.jid.bare:

436 reply.to = "contact"

437 else:

438 recipient.log.warning(

439 "Invalid replied-to JID: %r, assuming reply to the contact (%r)",

440 jid,

441 recipient.jid,

442 )

443

444 return text, reply

445

446 async def __dispatch_nonbob_sticker(

447 self,

448 attachment: _IncomingAttachment,

449 recipient: AnyRecipient,

450 fallback: str,

451 reply: Reply | None = None,

452 thread: str | None = None,

453 ) -> str | None:

454 if attachment.cid:

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

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

457 else:

458 sticker = None

459 if sticker is None:

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

461 response.raise_for_status()

462 data = await response.read()

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

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

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

466 orm, cid, data, attachment.attachment.content_type

467 )

468 orm.commit()

469 if sticker is None:

470 msg = XMPPMessage[Any](

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

472 thread=thread,

473 reply=reply,

474 )

475 if recipient.is_group:

476 assert isinstance(msg, MUCMessageProtocol)

477 else:

478 assert isinstance(msg, ContactMessageProtocol)

479 return await recipient.on_message(msg)

480

481 sticker.thread = thread

482 sticker.reply = reply

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

484

485 async def __dispatch_bob(

486 self,

487 from_: JID,

488 cid: str,

489 recipient: AnyRecipient,

490 reply: Reply | None = None,

491 thread: str | None = None,

492 ) -> str | None:

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

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

495 if sticker is None:

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

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

498 )

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

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

501 assert sticker is not None

502 sticker.reply = reply

503 sticker.thread = thread

504 if recipient.is_group:

505 assert isinstance(sticker, MUCSticker)

506 else:

507 assert isinstance(sticker, ContactSticker)

508 return await recipient.on_sticker(sticker)

509

510

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

512 needle = config.NO_UPLOAD_URL_PREFIX or config.UPLOAD_URL_PREFIX

513 if not needle:

514 return text

515 return text.replace(needle, "")

516

517

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

519 result = []

520 for link_metadata in link_metadatas:

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

522 if (

523 preview.image

524 and isinstance(preview.image, str)

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

526 ):

527 try:

528 image = base64.b64decode(

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

530 )

531 except Exception as e: # noqa: BLE001

532 log.warning(

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

534 )

535 else:

536 preview = preview._replace(image=image)

537 result.append(preview)

538 return tuple(result)

539

540

541log = logging.getLogger(__name__)