Mercurial > libervia-backend
view libervia/backend/plugins/plugin_sec_gre_formatter_mime.py @ 4351:6a0a081485b8
plugin autocrypt: Autocrypt protocol implementation:
Implementation of autocrypt: `autocrypt` header is checked, and if present and no public
key is known for the peer, the key is imported.
`autocrypt` header is also added to outgoing message (only if an email gateway is
detected).
For the moment, the JID is use as identifier, but the real email used by gateway should be
used in the future.
rel 456
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 28 Feb 2025 09:23:35 +0100 |
parents | 07e87adb2f65 |
children |
line wrap: on
line source
#!/usr/bin/env python3 # Libervia plugin # Copyright (C) 2009-2025 Jérôme Poisson (goffi@goffi.org) # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. from typing import Final, TYPE_CHECKING, cast from twisted.words.protocols.jabber import jid from twisted.words.protocols.jabber.xmlstream import XMPPHandler from twisted.words.xish import domish from wokkel import data_form, disco, iwokkel from zope.interface import implementer from libervia.backend.core import exceptions from libervia.backend.core.constants import Const as C from libervia.backend.core.core_types import SatXMPPEntity from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger from libervia.backend.plugins.plugin_xep_0106 import XEP_0106 from .plugin_exp_gre import Formatter from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText if TYPE_CHECKING: from libervia.backend.core.main import LiberviaBackend log = getLogger(__name__) PLUGIN_INFO = { C.PI_NAME: "GRE Formatter: MIME", C.PI_IMPORT_NAME: "GRE-MIME", C.PI_TYPE: "XEP", C.PI_MODES: C.PLUG_MODE_BOTH, C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: [ "GRE", ], C.PI_RECOMMENDATIONS: [], C.PI_MAIN: "GREFormatterMime", C.PI_HANDLER: "yes", C.PI_DESCRIPTION: _("Handle MIME formatting for Gateway Relayed Encryption."), } NS_GRE_MIME: Final = "urn:xmpp:gre:formatter:mime:0" class GREFormatterMime(Formatter): name = "mime" namespace = NS_GRE_MIME def __init__(self, host: "LiberviaBackend") -> None: log.info(f"plugin {PLUGIN_INFO[C.PI_NAME]!r} initialization") super().__init__(host) host.register_namespace("gre-mime", NS_GRE_MIME) def get_handler(self, client: SatXMPPEntity) -> XMPPHandler: return GREMIMEHandler(self) async def format( self, client: SatXMPPEntity, recipient_id: str, message_elt: domish.Element, encryption_data_form: data_form.Form, ) -> bytes: """Format the sent stanza as multipart MIME payload. @param client: Client session. @param message_elt: <message> element being sent. @return: MIME formatted payload. """ if message_elt.name != "message": raise exceptions.InternalError("Only <message> stanza should be received.") try: body = str(next(message_elt.elements(None, "body"))) except StopIteration as e: msg = f"Cancelling message sending due to missing body: {message_elt.toXml()}" log.warning(msg) raise exceptions.CancelError(msg) from e msg = MIMEMultipart() # "sender_id" is the real email address used by the gateway for this user. msg["From"] = encryption_data_form["sender_id"] msg["To"] = recipient_id subject_elt = next(message_elt.elements(None, "subject"), None) if subject_elt: msg["Subject"] = str(subject_elt) msg.attach(MIMEText(body)) return msg.as_bytes() @implementer(iwokkel.IDisco) class GREMIMEHandler(XMPPHandler): def __init__(self, plugin_parent): self.plugin_parent = plugin_parent def getDiscoInfo( self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" ) -> list[disco.DiscoFeature]: return [ disco.DiscoFeature(NS_GRE_MIME), ] def getDiscoItems( self, requestor: jid.JID, target: jid.JID, nodeIdentifier: str = "" ) -> list[disco.DiscoItems]: return []