Coverage for slidge/core/mixins/message.py: 87%
105 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 18:29 +0000
1import logging
2import uuid
3import warnings
5from slixmpp import Iq, Message
6from slixmpp.plugins.xep_0004.stanza.form import Form
8from ...util.types import AnyMUC, ChatState, Marker
9from .attachment import AttachmentMixin
10from .message_maker import MessageMaker
11from .message_text import TextMessageMixin
13# this is for MDS
14PUBLISH_OPTIONS = Form()
15PUBLISH_OPTIONS["type"] = "submit"
16PUBLISH_OPTIONS.add_field(
17 "FORM_TYPE", "hidden", value="http://jabber.org/protocol/pubsub#publish-options"
18)
19PUBLISH_OPTIONS.add_field("pubsub#persist_items", value="true")
20PUBLISH_OPTIONS.add_field("pubsub#max_items", value="max")
21PUBLISH_OPTIONS.add_field("pubsub#send_last_published_item", value="never")
22PUBLISH_OPTIONS.add_field("pubsub#access_model", value="whitelist")
25class ChatStateMixin(MessageMaker):
26 def __init__(self) -> None:
27 super().__init__()
28 self.__last_chat_state: ChatState | None = None
30 def _chat_state(
31 self, state: ChatState, forced: bool = False, **kwargs: object
32 ) -> None:
33 carbon = kwargs.get("carbon", False)
34 if carbon or (state == self.__last_chat_state and not forced):
35 return
36 self.__last_chat_state = state
37 msg = self._make_message(state=state, hints={"no-store"})
38 self._send(msg, **kwargs)
40 def active(self, **kwargs: object) -> None:
41 """
42 Send an "active" chat state (:xep:`0085`) from this
43 :term:`XMPP Entity`.
44 """
45 self._chat_state("active", forced=False, **kwargs)
47 def composing(self, **kwargs: object) -> None:
48 """
49 Send a "composing" (ie "typing notification") chat state (:xep:`0085`)
50 from this :term:`XMPP Entity`.
51 """
52 self._chat_state("composing", forced=True, **kwargs)
54 def paused(self, **kwargs: object) -> None:
55 """
56 Send a "paused" (ie "typing paused notification") chat state
57 (:xep:`0085`) from this :term:`XMPP Entity`.
58 """
59 self._chat_state("paused", forced=False, **kwargs)
61 def inactive(self, **kwargs: object) -> None:
62 """
63 Send an "inactive" (ie "contact has not interacted with the chat session
64 interface for an intermediate period of time") chat state (:xep:`0085`)
65 from this :term:`XMPP Entity`.
66 """
67 self._chat_state("inactive", forced=False, **kwargs)
69 def gone(self, **kwargs: object) -> None:
70 """
71 Send a "gone" (ie "contact has not interacted with the chat session interface,
72 system, or device for a relatively long period of time") chat state
73 (:xep:`0085`) from this :term:`XMPP Entity`.
74 """
75 self._chat_state("gone", forced=False, **kwargs)
78class MarkerMixin(MessageMaker):
79 def _make_marker(
80 self, legacy_msg_id: str, marker: Marker, carbon: bool = False
81 ) -> Message:
82 msg = self._make_message(carbon=carbon)
83 msg[marker]["id"] = self._legacy_to_xmpp(legacy_msg_id)[-1]
84 return msg
86 def ack(self, legacy_msg_id: str, **kwargs: object) -> None:
87 """
88 Send an "acknowledged" message marker (:xep:`0333`) from this :term:`XMPP Entity`.
90 :param legacy_msg_id: The message this marker refers to
91 """
92 self._send(
93 self._make_marker(
94 legacy_msg_id, "acknowledged", carbon=bool(kwargs.get("carbon"))
95 ),
96 **kwargs,
97 )
99 def received(self, legacy_msg_id: str, **kwargs: object) -> None:
100 """
101 Send a "received" message marker (:xep:`0333`) from this :term:`XMPP Entity`.
102 If called on a :class:`LegacyContact`, also send a delivery receipt
103 marker (:xep:`0184`).
105 :param legacy_msg_id: The message this marker refers to
106 """
107 carbon = bool(kwargs.get("carbon"))
108 if self.mtype == "chat":
109 for msg_id in self._legacy_to_xmpp(legacy_msg_id):
110 self._send(
111 self.xmpp.delivery_receipt.make_ack(
112 msg_id,
113 mfrom=self.jid,
114 mto=self.user_jid,
115 )
116 )
117 self._send(
118 self._make_marker(legacy_msg_id, "received", carbon=carbon), **kwargs
119 )
121 def displayed(self, legacy_msg_id: str, **kwargs: object) -> None:
122 """
123 Send a "displayed" message marker (:xep:`0333`) from this :term:`XMPP Entity`.
125 :param legacy_msg_id: The message this marker refers to
126 """
127 if (
128 self.xmpp.MARK_ALL_MESSAGES
129 and (muc := getattr(self, "muc", None))
130 and getattr(self, "is_user", False)
131 ):
132 with self.xmpp.store.session() as orm:
133 if self.xmpp.store.mam.is_displayed_by_user(
134 orm, muc.jid.local, str(legacy_msg_id)
135 ):
136 self.session.log.debug(
137 "Ignoring carbon marker for message already displayed by user."
138 )
139 return
140 else:
141 muc.pop_unread_xmpp_ids_up_to(
142 self._legacy_to_xmpp(legacy_msg_id)[-1]
143 )
145 self._send(
146 self._make_marker(
147 legacy_msg_id, "displayed", carbon=bool(kwargs.get("carbon"))
148 ),
149 **kwargs,
150 )
151 if getattr(self, "is_user", False):
152 self.session.create_task(self.__send_mds(legacy_msg_id))
154 async def __send_mds(self, legacy_msg_id: str) -> None:
155 # Send a MDS displayed marker on behalf of the user for a group chat
156 if muc := getattr(self, "muc", None):
157 muc_jid = muc.jid.bare
158 else:
159 # This is not implemented for 1:1 chat because it would rely on
160 # storing the XMPP-server injected stanza-id, which we don't track
161 # ATM.
162 # In practice, MDS should mostly be useful for public group chats,
163 # so it should not be an issue.
164 # We'll see if we need to implement that later
165 return
166 xmpp_msg_id = self._legacy_to_xmpp(legacy_msg_id)[-1]
167 iq = Iq(sto=self.user_jid.bare, sfrom=self.user_jid.bare, stype="set")
168 iq["pubsub"]["publish"]["node"] = self.xmpp.plugin["xep_0490"].stanza.NS
169 iq["pubsub"]["publish"]["item"]["id"] = muc_jid
170 displayed = self.xmpp.plugin["xep_0490"].stanza.Displayed()
171 displayed["stanza_id"]["id"] = xmpp_msg_id
172 displayed["stanza_id"]["by"] = muc_jid
173 iq["pubsub"]["publish"]["item"]["payload"] = displayed
174 iq["pubsub"]["publish_options"] = PUBLISH_OPTIONS
175 try:
176 await self.xmpp.plugin["xep_0356"].send_privileged_iq(iq)
177 except Exception as e:
178 self.session.log.debug("Could not MDS mark", exc_info=e)
181class ContentMessageMixin(AttachmentMixin, TextMessageMixin):
182 pass
185class CarbonMessageMixin(ContentMessageMixin, MarkerMixin):
186 def _privileged_send(self, msg: Message) -> None:
187 i = msg.get_id()
188 if i:
189 self.session.ignore_messages.add(i)
190 else:
191 i = "slidge-carbon-" + str(uuid.uuid4())
192 msg.set_id(i)
193 msg.del_origin_id()
194 try:
195 self.xmpp.plugin["xep_0356"].send_privileged_message(msg)
196 except PermissionError:
197 warnings.warn(
198 f"Slidge does not have the privilege (XEP-0356) to send messages on behalf of {self.user_jid}. "
199 "If this is a local user, consider configuring your XMPP server for that."
200 )
203class InviteMixin(MessageMaker):
204 def invite_to(
205 self,
206 muc: AnyMUC,
207 reason: str | None = None,
208 password: str | None = None,
209 **send_kwargs: object,
210 ) -> None:
211 """
212 Send an invitation to join a group (:xep:`0249`) from this :term:`XMPP Entity`.
214 :param muc: the muc the user is invited to
215 :param reason: a text explaining why the user should join this muc
216 :param password: maybe this will make sense later? not sure
217 :param send_kwargs: additional kwargs to be passed to _send()
218 (internal use by slidge)
219 """
220 msg = self._make_message(mtype="normal")
221 msg["groupchat_invite"]["jid"] = muc.jid
222 if reason:
223 msg["groupchat_invite"]["reason"] = reason
224 if password:
225 msg["groupchat_invite"]["password"] = password
226 self._send(msg, **send_kwargs)
229class MessageMixin(InviteMixin, ChatStateMixin, MarkerMixin, ContentMessageMixin):
230 pass
233class MessageCarbonMixin(InviteMixin, ChatStateMixin, CarbonMessageMixin):
234 pass
237log = logging.getLogger(__name__)