Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_sec_autocrypt.py @ 4379:79d463e3fdeb

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
core (types): Renamed `libervia.backend.core.core_types.MessageData` to `MessageDataLegacy`: There are 2 concurrent models, this one is an historical legacy. The type to use the one from `libervia.backend.models.core`, and it will progressively be used everywhere and fully based on Pydantic.
author Goffi <goffi@goffi.org>
date Fri, 04 Jul 2025 12:28:00 +0200
parents 6a0a081485b8
children
comparison
equal deleted inserted replaced
4378:930a4ea7ab6f 4379:79d463e3fdeb
25 from twisted.internet import defer 25 from twisted.internet import defer
26 from twisted.words.protocols.jabber import jid 26 from twisted.words.protocols.jabber import jid
27 from twisted.words.xish import domish 27 from twisted.words.xish import domish
28 28
29 from libervia.backend.core.constants import Const as C 29 from libervia.backend.core.constants import Const as C
30 from libervia.backend.core.core_types import MessageData, SatXMPPEntity 30 from libervia.backend.core.core_types import MessageDataLegacy, SatXMPPEntity
31 from libervia.backend.core.i18n import _ 31 from libervia.backend.core.i18n import _
32 from libervia.backend.core.log import getLogger 32 from libervia.backend.core.log import getLogger
33 from libervia.backend.memory import persistent 33 from libervia.backend.memory import persistent
34 from libervia.backend.plugins.plugin_xep_0106 import XEP_0106 34 from libervia.backend.plugins.plugin_xep_0106 import XEP_0106
35 from libervia.backend.plugins.plugin_xep_0131 import XEP_0131 35 from libervia.backend.plugins.plugin_xep_0131 import XEP_0131
157 # Sender already handled. 157 # Sender already handled.
158 client._autocrypt_seen: set[str] = set() 158 client._autocrypt_seen: set[str] = set()
159 client._autocrypt_gpg_provider = get_gpg_provider(self.host, client) 159 client._autocrypt_gpg_provider = get_gpg_provider(self.host, client)
160 160
161 async def handle_autocrypt_data( 161 async def handle_autocrypt_data(
162 self, client: SatXMPPEntity, mess_data: MessageData, autocrypt_data_raw: str 162 self, client: SatXMPPEntity, mess_data: MessageDataLegacy, autocrypt_data_raw: str
163 ) -> None: 163 ) -> None:
164 """Process Autocrypt header from XMPP email gateway 164 """Process Autocrypt header from XMPP email gateway
165 165
166 @param client: Client session. 166 @param client: Client session.
167 @param mess_data: Message data. 167 @param mess_data: Message data.
209 "autocrypt" 209 "autocrypt"
210 ) 210 )
211 client._autocrypt_seen.add(sender_email) 211 client._autocrypt_seen.add(sender_email)
212 212
213 def _check_headers( 213 def _check_headers(
214 self, client: SatXMPPEntity, mess_data: MessageData 214 self, client: SatXMPPEntity, mess_data: MessageDataLegacy
215 ) -> MessageData: 215 ) -> MessageDataLegacy:
216 try: 216 try:
217 autocrypt_data = mess_data["extra"]["headers"]["autocrypt"] 217 autocrypt_data = mess_data["extra"]["headers"]["autocrypt"]
218 except KeyError: 218 except KeyError:
219 pass 219 pass
220 else: 220 else:
230 post_treat: defer.Deferred, 230 post_treat: defer.Deferred,
231 ) -> Literal[True]: 231 ) -> Literal[True]:
232 post_treat.addCallback(partial(self._check_headers, client)) 232 post_treat.addCallback(partial(self._check_headers, client))
233 return True 233 return True
234 234
235 async def add_autocrypt_header(self, client, mess_data: MessageData) -> MessageData: 235 async def add_autocrypt_header(self, client, mess_data: MessageDataLegacy) -> MessageDataLegacy:
236 to_jid = mess_data["to"] 236 to_jid = mess_data["to"]
237 if await self.host.memory.disco.has_identity( 237 if await self.host.memory.disco.has_identity(
238 client, "gateway", "smtp", jid.JID(to_jid.host) 238 client, "gateway", "smtp", jid.JID(to_jid.host)
239 ): 239 ):
240 gpg_provider = client._autocrypt_gpg_provider 240 gpg_provider = client._autocrypt_gpg_provider
263 def send_message_trigger( 263 def send_message_trigger(
264 self, client, mess_data, pre_xml_treatments, post_xml_treatments 264 self, client, mess_data, pre_xml_treatments, post_xml_treatments
265 ) -> Literal[True]: 265 ) -> Literal[True]:
266 """Process the XEP-0131 related data to be sent""" 266 """Process the XEP-0131 related data to be sent"""
267 267
268 def add_headers(mess_data: MessageData) -> MessageData: 268 def add_headers(mess_data: MessageDataLegacy) -> MessageDataLegacy:
269 extra = mess_data["extra"] 269 extra = mess_data["extra"]
270 self.move_keywords_to_headers(extra) 270 self.move_keywords_to_headers(extra)
271 # Now we parse headers, if any. 271 # Now we parse headers, if any.
272 if "headers" in extra: 272 if "headers" in extra:
273 headers_data = HeadersData(**extra["headers"]) 273 headers_data = HeadersData(**extra["headers"])