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

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 orm.commit()

136

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

148

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

160

161 async def __get_replace(

162 self,

163 msg: Message,

164 recipient: RecipientType,

165 ) -> str | None:

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

167 return None

168 return (

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

170 )

171

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

173 is_sticker = "sticker" in msg

174

175 if (

176 "sfs" in msg

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

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

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

180 ):

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

182 # TODO: support for other sources than URL

183 # TODO: support for multiattachments in single message.

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

185 content_type = cid = None

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

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

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

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

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

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

192 attachment = _IncomingAttachment(

193 XMPPAttachment(

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

195 http=self.xmpp.http,

196 is_sticker=is_sticker,

197 content_type=content_type,

198 ),

199 cid=cid,

200 )

201 return [attachment]

202

203 if "oob" in msg:

204 return [

205 _IncomingAttachment(

206 XMPPAttachment(

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

208 http=self.xmpp.http,

209 is_sticker=is_sticker,

210 ),

211 )

212 ]

213

214 if (

215 "reference" in msg

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

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

218 ):

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

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

221 attachment = _IncomingAttachment(

222 XMPPAttachment(

223 url=source["uri"],

224 http=self.xmpp.http,

225 is_sticker=is_sticker,

226 ),

227 )

228 break

229 else:

230 return []

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

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

233 return [attachment]

234

235 return []

236

237 async def __dispatch_msg(

238 self,

239 *,

240 replace: str | None,

241 reply: Reply | None,

242 body: str | None,

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

244 recipient: AnyRecipient,

245 thread: str | None,

246 slix_msg: Message,

247 ) -> str | None:

248 if replace:

249 if body is not None:

250 body = body.strip()

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

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

253 return None

254 if not recipient.CORRECTION:

255 if recipient.RETRACTION:

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

257 replace = None

258 elif body:

259 body = "Correction:\n" + body

260

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

262 log.debug(

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

264 slix_msg.get_id(),

265 )

266 return None

267 link_previews = (

268 parse_link_previews(slix_msg["link_metadatas"])

269 if "link_metadata" in slix_msg

270 else ()

271 )

272

273 for attachment in attachments:

274 if not body:

275 body = None

276 break

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

278

279 msg = XMPPMessage[Any](

280 body=body,

281 attachments=attachments,

282 reply=reply,

283 thread=thread,

284 link_previews=link_previews,

285 replace=replace,

286 )

287 if recipient.is_group:

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

289 assert isinstance(msg, MUCMessageProtocol)

290 else:

291 assert isinstance(msg, ContactMessageProtocol)

292 return await recipient.on_message(msg)

293

294 @exceptions_to_xmpp_errors

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

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

297 if not recipient.RETRACTION:

298 raise XMPPError(

299 "bad-request",

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

301 )

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

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

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

305 if isinstance(recipient, LegacyMUC):

306 await recipient.echo(msg, None)

307 self.__ack(msg)

308

309 @exceptions_to_xmpp_errors

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

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

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

313

314 legacy_id = self._xmpp_msg_id_to_legacy(react_to, recipient)

315

316 if not legacy_id:

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

318 raise XMPPError(

319 "internal-server-error",

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

321 )

322

323 emojis = [

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

325 ]

326 error_msg = None

327

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

329 error_msg = "Maximum 1 emoji/message"

330

331 if (

332 not error_msg

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

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

335 ):

336 error_msg = (

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

338 )

339

340 if error_msg:

341 recipient.session.send_gateway_message(error_msg)

342 if not isinstance(recipient, LegacyMUC):

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

344 recipient.react(legacy_id, carbon=True)

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

346 raise XMPPError(

347 "policy-violation",

348 text=error_msg,

349 clear=False,

350 )

351

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

353 if isinstance(recipient, LegacyMUC):

354 await recipient.echo(msg, None)

355 else:

356 self.__ack(msg)

357

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

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

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

361 )

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

363 if not multi:

364 return

365

366 if isinstance(recipient, LegacyMUC):

367 for xmpp_id in multi:

368 mc = copy(msg)

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

370 await recipient.echo(mc)

371 elif isinstance(recipient, LegacyContact):

372 for xmpp_id in multi:

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

374

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

376 if not self.xmpp.PROPER_RECEIPTS:

377 self.xmpp.delivery_receipt.ack(msg)

378

379 async def __get_reply(

380 self, msg: Message, recipient: AnyRecipient

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

382 if "reply" not in msg:

383 return msg["body"], None

384

385 session = recipient.session

386

387 try:

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

389 except XMPPError:

390 session.log.debug(

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

392 )

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

394

395 if "fallback" in msg and (

396 isinstance(recipient, LegacyMUC) or recipient.REPLIES

397 ):

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

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

400 )

401 try:

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

403 except AttributeError:

404 fallback = None

405 else:

406 text = msg["body"]

407 fallback = None

408

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

410 reply = Reply(msg_id, fallback)

411 if recipient.is_group:

412 if not jid:

413 recipient.log.warning(

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

415 jid,

416 )

417 nick = jid.resource

418 if nick == recipient.user_nick:

419 reply.to = await recipient.get_user_participant()

420 elif not nick:

421 reply.to = recipient.get_system_participant()

422 else:

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

424 else:

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

426 reply.to = "self"

427 elif jid.bare == recipient.jid.bare:

428 reply.to = "contact"

429 else:

430 recipient.log.warning(

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

432 jid,

433 recipient.jid,

434 )

435

436 return text, reply

437

438 async def __dispatch_nonbob_sticker(

439 self,

440 attachment: _IncomingAttachment,

441 recipient: AnyRecipient,

442 fallback: str,

443 reply: Reply | None = None,

444 thread: str | None = None,

445 ) -> str | None:

446 if attachment.cid:

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

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

449 else:

450 sticker = None

451 if sticker is None:

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

453 response.raise_for_status()

454 data = await response.read()

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

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

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

458 orm, cid, data, attachment.attachment.content_type

459 )

460 orm.commit()

461 if sticker is None:

462 msg = XMPPMessage[Any](

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

464 thread=thread,

465 reply=reply,

466 )

467 if recipient.is_group:

468 assert isinstance(msg, MUCMessageProtocol)

469 else:

470 assert isinstance(msg, ContactMessageProtocol)

471 return await recipient.on_message(msg)

472

473 sticker.thread = thread

474 sticker.reply = reply

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

476

477 async def __dispatch_bob(

478 self,

479 from_: JID,

480 cid: str,

481 recipient: AnyRecipient,

482 reply: Reply | None = None,

483 thread: str | None = None,

484 ) -> str | None:

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

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

487 if sticker is None:

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

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

490 )

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

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

493 assert sticker is not None

494 sticker.reply = reply

495 sticker.thread = thread

496 if recipient.is_group:

497 assert isinstance(sticker, MUCSticker)

498 else:

499 assert isinstance(sticker, ContactSticker)

500 return await recipient.on_sticker(sticker)

501

502

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

504 needle = config.NO_UPLOAD_URL_PREFIX or config.UPLOAD_URL_PREFIX

505 if not needle:

506 return text

507 return text.replace(needle, "")

508

509

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

511 result = []

512 for link_metadata in link_metadatas:

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

514 if (

515 preview.image

516 and isinstance(preview.image, str)

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

518 ):

519 try:

520 image = base64.b64decode(

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

522 )

523 except Exception as e: # noqa: BLE001

524 log.warning(

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

526 )

527 else:

528 preview = preview._replace(image=image)

529 result.append(preview)

530 return tuple(result)

531

532

533log = logging.getLogger(__name__)