diff --git a/libervia/backend/plugins/plugin_comp_email_gateway/__init__.py b/libervia/backend/plugins/plugin_comp_email_gateway/__init__.py index 77ab4d63..6869c17b 100644 --- a/libervia/backend/plugins/plugin_comp_email_gateway/__init__.py +++ b/libervia/backend/plugins/plugin_comp_email_gateway/__init__.py @@ -82,6 +82,7 @@ from libervia.backend.plugins.plugin_xep_0499 import XEP_0499, ExtDiscoOptions from libervia.backend.tools.common import date_utils, regex, uri from libervia.backend.tools.utils import aio +from .ad_hoc import EmailAdHocService from .cleaning import convert_to_html_and_detect_noise from .imap import IMAPClientFactory from .models import Credentials, UserData @@ -103,6 +104,7 @@ PLUGIN_INFO = { C.PI_PROTOCOLS: [], C.PI_DEPENDENCIES: [ "XEP-0033", + "XEP-0050", "XEP-0077", "XEP-0106", "XEP-0277", @@ -123,7 +125,6 @@ PLUGIN_INFO = { ), } -CONF_SECTION = f"component {IMPORT_NAME}" PREFIX_KEY_CREDENTIALS = "CREDENTIALS_" KEY_CREDENTIALS = f"{PREFIX_KEY_CREDENTIALS}{{from_jid}}" @@ -166,6 +167,7 @@ class EmailGatewayComponent(GetDataHandler): self._ext_disco = cast(XEP_0499, host.plugins["XEP-0499"]) self._gre = cast(GRE, host.plugins["GRE"]) self._syntax = cast(TextSyntaxes, host.plugins["TEXT_SYNTAXES"]) + self.ad_hoc = EmailAdHocService(self) # TODO: For the moment, all credentials are kept in cache; we should only keep the # X latest. self.users_data: dict[jid.JID, UserData] = {} @@ -230,6 +232,9 @@ class EmailGatewayComponent(GetDataHandler): await self._init() self.initalized = True + def profile_connected(self, client) -> None: + self.ad_hoc.init(client) + def _message_received_trigger( self, client: SatXMPPEntity, @@ -955,6 +960,16 @@ class EmailGatewayComponent(GetDataHandler): await self.handle_attachment(part, user_jid) client = self.client.get_virtual_client(from_jid) + + # Message ID + message_id_list = email.get_all("message-id") + if not message_id_list: + message_id = None + else: + message_id = message_id_list[-1].strip() + if message_id.startswith("<") and message_id.endswith(">"): + message_id = message_id[1:-1] + # Now that the message is parsed, we check if it's a mailing list. list_ids = email.get_all("list-id") if list_ids: @@ -965,6 +980,7 @@ class EmailGatewayComponent(GetDataHandler): author_name, author_email, list_ids[-1], + message_id, email, user_jid, body, @@ -981,6 +997,7 @@ class EmailGatewayComponent(GetDataHandler): {"": body}, {"": subject} if subject else None, extra=extra, + uid=message_id, ) def parse_references(self, email: EmailMessage) -> list[str]: @@ -1207,6 +1224,7 @@ class EmailGatewayComponent(GetDataHandler): author_name: str, author_email: str, list_id_header: str, + message_id: str | None, email: EmailMessage, user_jid: jid.JID, body: str, @@ -1217,14 +1235,10 @@ class EmailGatewayComponent(GetDataHandler): Mailing list emails are converted to pubsub blogs. """ + if message_id is None: + raise exceptions.DataError("Missing message ID.") assert self.client is not None pubsub_service = self.client.jid - message_id_list = email.get_all("message-id") - if not message_id_list: - raise exceptions.DataError("Missing message ID.") - message_id = message_id_list[-1].strip() - if message_id.startswith("<") and message_id.endswith(">"): - message_id = message_id[1:-1] if not message_id: raise exceptions.DataError("Emtpy message ID.") diff --git a/libervia/backend/plugins/plugin_comp_email_gateway/ad_hoc.py b/libervia/backend/plugins/plugin_comp_email_gateway/ad_hoc.py new file mode 100644 index 00000000..e23fb38c --- /dev/null +++ b/libervia/backend/plugins/plugin_comp_email_gateway/ad_hoc.py @@ -0,0 +1,313 @@ +#!/usr/bin/env python3 + +# Libervia Email Gateway +# 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 . + +from typing import TYPE_CHECKING, NamedTuple, cast + +from twisted.mail.imap4 import IMAP4Exception +from twisted.words.protocols.jabber import jid +from twisted.words.xish import domish + +from libervia.backend.core.constants import Const as C +from libervia.backend.core.core_types import SatXMPPEntity +from libervia.backend.core.i18n import D_, _ +from libervia.backend.core.log import getLogger +from libervia.backend.memory.memory import SessionData +from libervia.backend.plugins.plugin_xep_0004 import ( + Boolean, + DataForm, + ListMulti, + Option, + Text, +) +from libervia.backend.plugins.plugin_xep_0050 import ( + AdHocCallbackData, + AdHocCommand, + AdHocError, + Error, + Note, + NoteType, + Status, + XEP_0050, +) + +from .imap import IMAPClient + +log = getLogger(__name__) +if TYPE_CHECKING: + from . import EmailGatewayComponent + from .imap import IMAPClient + + +class ImapInfo(NamedTuple): + client: "IMAPClient" + uid: str + form: DataForm + + +class EmailAdHocService: + """Ad-Hoc commands for AP Gateway""" + + def __init__(self, email_gw: "EmailGatewayComponent"): + self.host = email_gw.host + self.email_gw = email_gw + self._c = cast(XEP_0050, self.host.plugins["XEP-0050"]) + + def init(self, client: SatXMPPEntity) -> None: + self._c.register_ad_hoc_command( + client, + AdHocCommand( + callback=self.delete_email, + label=D_("Delete Email"), + form=DataForm( + fields=[ + Text(var="uid", label=D_("Message ID"), required=True), + Boolean( + label=D_( + "This email will be permanently deleted, please check " + "this box to confirm this operation." + ), + var="confirm", + required=True, + ), + ] + ), + allowed_magics=[C.ENTITY_ALL], + ), + ) + self._c.register_ad_hoc_command( + client, + AdHocCommand( + callback=self.move_email, + label=D_("Move Email"), + form=DataForm( + fields=[ + Text(var="uid", label=D_("Message ID"), required=True), + Text( + var="dest", + label=D_("Name of Destination Mailbox"), + required=True, + ), + ] + ), + allowed_magics=[C.ENTITY_ALL], + ), + ) + self._c.register_ad_hoc_command( + client, + AdHocCommand( + callback=self.change_flags, + label=D_("Change Flags"), + form=DataForm( + fields=[ + Text(var="uid", label=D_("Message ID"), required=True), + ] + ), + allowed_magics=[C.ENTITY_ALL], + ), + ) + + async def get_uid(self, command_elt: domish.Element) -> ImapInfo: + iq_elt = command_elt.parent + assert iq_elt is not None + requestor = jid.JID(iq_elt["from"]).userhostJID() + form = DataForm.from_element(command_elt) + data = form.get_values() + try: + message_id = cast(str, data["uid"][0]) + except (KeyError, IndexError): + log.warning('"uid" field is mandatory.') + raise AdHocError(Error.BAD_PAYLOAD) + try: + user_data = self.email_gw.users_data[requestor] + except KeyError: + log.warning(f"No user data for {requestor}, is the entity registered?") + raise AdHocError(Error.FORBIDDEN) + if user_data.imap_client is None: + log.warning(f"IMAP client is not connected for {requestor}.") + raise AdHocError(Error.FORBIDDEN) + + uid = await user_data.imap_client.get_imap_uid_from_message_id(message_id) + if uid is None: + raise AdHocError( + Error.ITEM_NOT_FOUND, f"The message {message_id!r} is not found." + ) + return ImapInfo(user_data.imap_client, uid, form) + + async def delete_email( + self, + client: SatXMPPEntity, + command_elt: domish.Element, + session_data: SessionData, + action: str, + node: str, + ) -> AdHocCallbackData: + """Delete an email from IMAP server. + + @param client: Libervia client instance. + @param command_elt: The command element from the request. + @param session_data: Session data for this command execution. + @param action: The action being performed (EXECUTE, CANCEL, etc.) + @param node: The command node identifier. + @return: Command results. + """ + imap_info = await self.get_uid(command_elt) + values = imap_info.form.get_values() + try: + confirm_str = cast(str, values["confirm"][0] or C.BOOL_FALSE) + except IndexError: + confirm_str = C.BOOL_FALSE + if not C.bool(confirm_str): + return AdHocCallbackData( + status=Status.CANCELED, + notes=[Note(text=f"User didn't confirm deletion.")], + ) + + await imap_info.client.addFlags(imap_info.uid, [r"\Deleted"], uid=True) + await imap_info.client.expunge() + + return AdHocCallbackData( + status=Status.COMPLETED, + notes=[Note(text=f"Email {imap_info.uid} successfully deleted.")], + ) + + async def move_email( + self, + client: SatXMPPEntity, + command_elt: domish.Element, + session_data: SessionData, + action: str, + node: str, + ) -> AdHocCallbackData: + """Move an email to another IMAP mailbox. + + @param client: Libervia client instance. + @param command_elt: The command element from the request. + @param session_data: Session data for this command execution. + @param action: The action being performed (EXECUTE, CANCEL, etc.) + @param node: The command node identifier. + @return: Command results. + """ + imap_info = await self.get_uid(command_elt) + values = imap_info.form.get_values() + try: + dest_mailbox = cast(str, values["dest"][0]).strip() + if not dest_mailbox: + raise IndexError + except IndexError: + raise AdHocError( + Error.BAD_PAYLOAD, D_("You must indicate a destination mailbox.") + ) + + try: + await imap_info.client.copy(imap_info.uid, dest_mailbox, uid=True) + except IMAP4Exception as e: + return AdHocCallbackData( + status=Status.CANCELED, + notes=[Note(type=NoteType.ERROR, text=f"Can't move email: {e}")], + ) + + await imap_info.client.addFlags(imap_info.uid, [r"\Deleted"], uid=True) + + await imap_info.client.expunge() + + return AdHocCallbackData( + status=Status.COMPLETED, + notes=[ + Note( + text=( + f"Email {imap_info.uid} successfully moved to mailbox " + f"{dest_mailbox!r}." + ) + ) + ], + ) + + async def change_flags( + self, + client: SatXMPPEntity, + command_elt: domish.Element, + session_data: SessionData, + action: str, + node: str, + ) -> AdHocCallbackData: + """Change flags of an email. + + @param client: Libervia client instance. + @param command_elt: The command element from the request. + @param session_data: Session data for this command execution. + @param action: The action being performed (EXECUTE, CANCEL, etc.) + @param node: The command node identifier. + @return: Command results. + """ + try: + imap_info = session_data["imap_info"] + except KeyError: + # We are on the second page of the commands, were user select flags. + imap_info = await self.get_uid(command_elt) + session_data["imap_info"] = imap_info + # FIXME: We only support INBOX for now. + mailbox = "INBOX" + mailbox_data = imap_info.client.mailboxes_data.get(mailbox, {}) + acceptable_flags = mailbox_data.get("FLAGS") + if not acceptable_flags: + return AdHocCallbackData( + status=Status.CANCELED, + notes=[ + Note( + type=NoteType.ERROR, + text=f"Can't find data on mailbox {mailbox}]", + ) + ], + ) + + flags_data = await imap_info.client.fetchFlags(imap_info.uid, uid=True) + try: + current_flags = flags_data[imap_info.uid]["FLAGS"] + except KeyError: + return AdHocCallbackData( + status=Status.CANCELED, + notes=[ + Note( + type=NoteType.ERROR, + text=f"Can't find current flags ({flags_data=}).", + ) + ], + ) + + form = DataForm( + instructions=D_("Select the flags to set on the email."), + fields=[ + ListMulti( + var="flags", + options=[Option(value=flag) for flag in acceptable_flags], + values=current_flags, + ) + ], + ) + return AdHocCallbackData(form=form, status=Status.EXECUTING) + else: + form = DataForm.from_element(command_elt) + data = form.get_values() + try: + new_flags = data["flags"] + except KeyError: + raise AdHocError(Error.BAD_PAYLOAD) + await imap_info.client.setFlags(imap_info.uid, flags=new_flags, uid=True) + return AdHocCallbackData( + status=Status.COMPLETED, notes=[Note(text=D_("Flags set successfully."))] + ) diff --git a/libervia/backend/plugins/plugin_comp_email_gateway/imap.py b/libervia/backend/plugins/plugin_comp_email_gateway/imap.py index 2969aa33..16cff9c9 100644 --- a/libervia/backend/plugins/plugin_comp_email_gateway/imap.py +++ b/libervia/backend/plugins/plugin_comp_email_gateway/imap.py @@ -17,30 +17,54 @@ # along with this program. If not, see . from collections.abc import Awaitable +from dataclasses import dataclass +from email import policy from email.message import EmailMessage from email.parser import BytesParser, Parser -from email import policy -from typing import Callable, cast -from twisted.internet import defer, protocol, reactor -from twisted.internet.base import DelayedCall +from typing import Any, Callable, cast + +from twisted.internet import defer, protocol, reactor as _reactor +from twisted.internet.base import DelayedCall, ReactorBase from twisted.mail import imap4 from twisted.python.failure import Failure from libervia.backend.core import exceptions from libervia.backend.core.i18n import _ from libervia.backend.core.log import getLogger + from .models import UserData log = getLogger(__name__) +reactor = cast(ReactorBase, _reactor) +# Number of second to wait before restarting IDLE mode when a command is received. +IDLE_DELAY = 5 + + +@dataclass +class IdleMode: + idling: defer.Deferred + timer: DelayedCall + finished: defer.Deferred | None = None class IMAPClient(imap4.IMAP4Client): - _idling = False - _idle_timer: DelayedCall | None = None + # If set, logs sent and received lines using log.debug. + verbose = False def __init__(self, connected: defer.Deferred, *args, **kwargs) -> None: super().__init__(*args, **kwargs) + self._idle_mode: IdleMode | None = None + self._delay_idle: DelayedCall | None = None self._connected = connected + self.mailboxes_data: dict[str, dict[str, Any]] = {} + + def sendLine(self, line: bytes) -> None: + if self.verbose: + command_str = line.decode("utf-8", errors="ignore") + if command_str.strip(): + log.debug(f"===> IMAP CLIENT SEND: {command_str}") + + super().sendLine(line) def serverGreeting(self, caps: dict) -> None: """Handle the server greeting and capabilities. @@ -49,12 +73,11 @@ class IMAPClient(imap4.IMAP4Client): """ defer.ensureDeferred(self.on_server_greeting(caps)) - async def on_server_greeting(self, caps: dict) -> None: + async def on_server_greeting(self, caps: dict[str, Any]) -> None: """Async method called when server greeting is received. @param caps: Server capabilities. """ - self.server_capabilities = caps try: await self.authenticate(self.factory.password.encode()) except Exception as e: @@ -66,52 +89,67 @@ class IMAPClient(imap4.IMAP4Client): log.debug("Authenticated.") self._connected.callback(None) if b"IDLE" in caps: - # We use "examine" for read-only access for now, will probably change in the - # future. - await self.examine(b"INBOX") + self.mailboxes_data["INBOX"] = await self.select(b"INBOX") log.debug("Activating IDLE mode") await self.idle() else: log.warning( f'"IDLE" mode is not supported by your server, this gateways needs a ' - "server supporting this mode." + f"server supporting this mode." ) return - async def idle(self) -> None: + async def get_imap_uid_from_message_id(self, message_id: str) -> int | None: + """Find a message by Message-ID and return its IMAP UID. + @param message_id: Message-ID as used in message header (and as XMPP message ID). + @return: IMAP UID of the message. None if the message is not found. + """ + search_query = imap4.Query(header=("Message-ID", message_id)) + uids = await self.search(search_query, uid=True) + return uids[0] if uids else None + + def idle(self) -> defer.Deferred: """Enter the IDLE mode to receive real-time updates from the server.""" - if self._idling: + if self._idle_mode: # We are already in idle state. - return - self._idling = True - self._idle_timer = reactor.callLater(29 * 60, self.on_idle_timeout) + return defer.succeed(None) + self._idle_mode = IdleMode( + idling=defer.Deferred(), + timer=reactor.callLater(29 * 60, self.on_idle_timeout), + ) log.debug("Starting IDLE mode.") - await self.sendCommand( + self._idle_mode.finished = self.sendCommand( imap4.Command( b"IDLE", continuation=lambda *a, **kw: log.debug(f"continuation: {a=} {kw=}"), ) ) + return self._idle_mode.idling - def idle_exit(self) -> None: + async def idle_exit(self) -> None: """Exit the IDLE mode.""" - assert self._idling - assert self._idle_timer is not None - if not self._idle_timer.called: - self._idle_timer.cancel() - self._idle_timer = None + assert self._idle_mode is not None + assert self._idle_mode.finished is not None + if not self._idle_mode.timer.called: + self._idle_mode.timer.cancel() # Send DONE command to exit IDLE mode. self.sendLine(b"DONE") - self._idling = False - log.debug("IDLE mode terminated.") + try: + await self._idle_mode.finished + except Exception: + log.exception("Error while exiting IDLE mode.") + else: + log.debug("IDLE mode terminated.") + finally: + self._idle_mode = None def on_idle_timeout(self): """Called when IDLE mode timeout is reached.""" - if self._idling: + if self._idle_mode: # We've reached 29 min of IDLE mode, we restart it as recommended in the # specifications. - self.idle_exit() - defer.ensureDeferred(self.idle()) + d = defer.ensureDeferred(self.idle_exit()) + d.addCallback(lambda __: self.idle()) def newMessages(self, exists: int | None, recent: int | None): """Called when new messages are received. @@ -131,10 +169,10 @@ class IMAPClient(imap4.IMAP4Client): if recent is None: log.debug("No recent messages, skipping fetch.") - defer.ensureDeferred(self.idle()) + await self.idle() return - self.idle_exit() + await self.idle_exit() # We retrieve recent messages. recent_uids = await self.search(imap4.Query(recent=True), uid=True) @@ -176,7 +214,7 @@ class IMAPClient(imap4.IMAP4Client): except Exception as e: log.error(f"Error fetching recent messages: {e}") - defer.ensureDeferred(self.idle()) + await self.idle() def connectionLost(self, reason: Failure) -> None: """Called when the connection is lost. @@ -193,19 +231,29 @@ class IMAPClient(imap4.IMAP4Client): @param line: The received line. """ - if self._idling: + if self.verbose: + response_str = line.decode("utf-8", errors="ignore") + if response_str.strip(): + log.debug(f"<=== IMAP CLIENT RECV: {response_str}") + + if self._idle_mode: + assert self._idle_mode.idling is not None if line == b"* OK Still here": - pass - elif line == b"+ idling": - pass + return + elif line.startswith(b"+ "): + if not self._idle_mode.idling.called: + self._idle_mode.idling.callback(None) + else: + log.warning(f"Received extra continuation line on IDLE: {line}") + return elif line.startswith(b"* "): # Handle unsolicited responses during IDLE - self._extraInfo([imap4.parseNestedParens(line[2:])]) + return self._extraInfo([imap4.parseNestedParens(line[2:])]) + elif b"ok idle" in line.lower(): + pass else: log.warning(f"Unexpected line received: {line!r}") - return - return super().lineReceived(line) def sendCommand(self, cmd: imap4.Command) -> defer.Deferred: @@ -216,18 +264,28 @@ class IMAPClient(imap4.IMAP4Client): @param cmd: The command to send. @return: A deferred that fires when the command is sent. """ - if self._idling and cmd.command != b"IDLE": - self.idle_exit() - d = super().sendCommand(cmd) + return defer.ensureDeferred(self.send_command(cmd)) - def restart_idle_mode(ret): - defer.ensureDeferred(self.idle()) - return ret + def _on_idle_start(self): + defer.ensureDeferred(self.idle()) + self._delay_idle = None - d.addCallback(restart_idle_mode) - return d + async def send_command(self, cmd: imap4.Command): + if self.verbose: + log.debug(f"==> sendCommand: {cmd=} {cmd.command=}]") + if self._idle_mode and cmd.command != b"IDLE": + await self.idle_exit() + ret = await super().sendCommand(cmd) + + # We restart idle mode, after 5s to avoid entering IDLE mode when there is a + # series of commands. + assert self._delay_idle is None + self._delay_idle = reactor.callLater(IDLE_DELAY, self._on_idle_start) + return ret else: - return super().sendCommand(cmd) + if self._delay_idle is not None and self._delay_idle.active(): + self._delay_idle.reset(IDLE_DELAY) + return await super().sendCommand(cmd) class IMAPClientFactory(protocol.ClientFactory): diff --git a/libervia/backend/plugins/plugin_comp_email_gateway/models.py b/libervia/backend/plugins/plugin_comp_email_gateway/models.py index 46f850a5..5097046d 100644 --- a/libervia/backend/plugins/plugin_comp_email_gateway/models.py +++ b/libervia/backend/plugins/plugin_comp_email_gateway/models.py @@ -17,9 +17,13 @@ # along with this program. If not, see . from dataclasses import dataclass +from typing import TYPE_CHECKING from pydantic import BaseModel, ConfigDict from twisted.mail import imap4 +if TYPE_CHECKING: + from .imap import IMAPClient + class Credentials(BaseModel): model_config = ConfigDict(extra="forbid") @@ -39,4 +43,4 @@ class Credentials(BaseModel): @dataclass class UserData: credentials: Credentials - imap_client: imap4.IMAP4Client | None = None + imap_client: "IMAPClient|None" = None diff --git a/libervia/backend/plugins/plugin_comp_email_gateway/pubsub_service.py b/libervia/backend/plugins/plugin_comp_email_gateway/pubsub_service.py index 4070ea6b..a505caa6 100644 --- a/libervia/backend/plugins/plugin_comp_email_gateway/pubsub_service.py +++ b/libervia/backend/plugins/plugin_comp_email_gateway/pubsub_service.py @@ -85,6 +85,9 @@ class EmailGWPubsubResource(pubsub.PubSubResource): async def get_nodes( self, requestor: jid.JID, service: jid.JID, node_id: str ) -> list[str]: + if node_id: + # we only return result for requests without node. + return [] nodes = await G.storage.get_pubsub_nodes(self.client, self.client.jid) return [self._pfs.namespace] + [cast(str, node.name) for node in nodes]