Add 2FA code capture with libnotify plugin

.gitignore

@@ -10,3 +10,4 @@ Mailnag/plugins/messagingmenuplugin.py
 *.egg-info/
 .python-version
 
+*~

AUTHORS

@@ -17,6 +17,7 @@ Popper was written by Ralf Hersel <ralf.hersel@gmx.net>.
 Code, docs and packaging contributors:
 ======================================
 
+André Auzi <aauzi@free.fr>
 Amin Bandali <me@aminb.org>
 Andreas Angerer
 Balló György <ballogyor@gmail.com>

Mailnag/backends/base.py

@@ -1,3 +1,4 @@
+# Copyright 2025 André Auzi <aauzi@free.fr>
 # Copyright 2020 Patrick Ulbrich <zulu99@gmx.net>
 # Copyright 2016, 2024 Timo Kankare <timo.kankare@iki.fi>
 #
@@ -54,9 +55,9 @@ def is_open(self) -> bool:
 		raise NotImplementedError
 
 	@abstractmethod
-	def list_messages(self) -> Iterator[tuple[str, Message, dict[str, Any]]]:
+	def list_messages(self) -> Iterator[tuple[str, Message, str | None, dict[str, Any]]]:
 		"""Lists unseen messages from the mailbox for this account.
-		Yields tuples (folder, message, flags) for every message.
+		Yields tuples (folder, message, uid, flags) for every message.
 		"""
 		raise NotImplementedError
 
@@ -78,7 +79,7 @@ def mark_as_seen(self, mails: list[Mail]):
 		This may raise an exception if mailbox does not support this action.
 		"""
 		raise NotImplementedError
-		
+
 	def supports_notifications(self) -> bool:
 		"""Returns True if mailbox supports notifications."""
 		# Default implementation
@@ -105,3 +106,9 @@ def cancel_notifications(self) -> None:
 		"""
 		raise NotImplementedError
 
+	@abstractmethod
+	def fetch_text(self, mail: Mail) -> str | None:
+		"""Fetches the text body of the message identified by an uid.
+		This should raise an exception if uid is None.
+		"""
+		raise NotImplementedError

Mailnag/backends/imap.py

@@ -1,3 +1,4 @@
+# Copyright 2025 André Auzi <aauzi@free.fr>
 # Copyright 2011 - 2021 Patrick Ulbrich <zulu99@gmx.net>
 # Copyright 2020 Andreas Angerer
 # Copyright 2016, 2024 Timo Kankare <timo.kankare@iki.fi>
@@ -26,6 +27,7 @@
 import email
 import logging
 import re
+import time
 from collections.abc import Callable
 from email.message import Message
 from typing import Any, Iterator, Optional
@@ -35,12 +37,16 @@
 from Mailnag.common.imaplib2 import AUTH
 from Mailnag.common.exceptions import InvalidOperationException
 from Mailnag.common.mutf7 import encode_mutf7, decode_mutf7
-from Mailnag.daemon.mails import Mail
+from Mailnag.daemon.mails import Mail, message_text
+from Mailnag.common.utils import dbgindent, get_goa_account_id, refresh_goa_token
 
+_LOGGER = logging.getLogger(__name__)
 
 class IMAPMailboxBackend(MailboxBackend):
 	"""Implementation of IMAP mail boxes."""
 
+	THROTTLE_TIME=0.25
+
 	def __init__(
 		self,
 		name: str = '',
@@ -64,15 +70,19 @@ def __init__(
 		self.folders = [encode_mutf7(folder) for folder in folders]
 		self.idle = idle
 		self._conn: Optional[imaplib.IMAP4] = None
+		self._last_access = None
 
+		self.goa_account_id = None
+		if self.oauth2string:
+			self.goa_account_id = get_goa_account_id(name, user)
 
 	def open(self) -> None:
 		if self._conn != None:
 			raise InvalidOperationException("Account is aready open")
-		
+
 		self._conn = self._connect()
 
-		
+
 	def close(self) -> None:
 		# if conn has already been closed, don't try to close it again
 		if self._conn is not None:
@@ -84,12 +94,12 @@ def close(self) -> None:
 	def is_open(self) -> bool:
 		return self._conn != None
 
-
-	def list_messages(self) -> Iterator[tuple[str, Message, dict[str, Any]]]:
+	
+	def list_messages(self) -> Iterator[tuple[str, Message, str | None, dict[str, Any]]]:
 		self._ensure_open()
 		assert self._conn is not None
 		conn = self._conn
-		
+
 		if len(self.folders) == 0:
 			folder_list = ['INBOX']
 		else:
@@ -101,27 +111,61 @@ def list_messages(self) -> Iterator[tuple[str, Message, dict[str, Any]]]:
 			try:
 				status, data = conn.uid('SEARCH', None, '(UNSEEN)') # ALL or UNSEEN
 			except:
-				logging.warning('Folder %s does not exist.', folder)
+				_LOGGER.warning('Folder %s does not exist.', folder)
 				continue
 
 			if status != 'OK' or None in [d for d in data]:
-				logging.debug('Folder %s in status %s | Data: %s', (folder, status, data))
+				_LOGGER.debug('Folder %s in status %s | Data: %s', folder, status, data)
 				continue # Bugfix LP-735071
+
 			for num in data[0].split():
-				typ, msg_data = conn.uid('FETCH', num, '(BODY.PEEK[HEADER])') # header only (without setting READ flag)
+				typ, msg_data = conn.uid('FETCH', num, '(BODY.PEEK[HEADER])') # header (without setting READ flag)
+				_LOGGER.debug("Msg data (length=%d):\n%s", len(msg_data),
+					      dbgindent(msg_data))
+				header = None
 				for response_part in msg_data:
 					if isinstance(response_part, tuple):
-						try:
-							msg = email.message_from_bytes(response_part[1])
-						except:
-							logging.debug("Couldn't get IMAP message.")
-							continue
-						yield (folder, msg, {'uid' : num.decode("utf-8"), 'folder' : folder})
+						if b'BODY[HEADER]' in response_part[0]:
+							header = email.message_from_bytes(response_part[1])
+				if header:
+					_LOGGER.debug("Msg header:\n%s", dbgindent(header))
+					yield (folder, header, num.decode("utf-8"), { 'folder' : folder })
+
+	def fetch_text(self, mail: Mail) -> str | None:
+		text = self._fetch_text(mail)
+
+		# NOTE: Sometimes a server does not return the text immediately
+		if text is not None:
+			return text
+
+		_LOGGER.warning('Retry fetch_text.')
+		return self._fetch_text(mail)
+
+	def _fetch_text(self, mail: Mail) -> str | None:
+		self._ensure_open()
+		assert self._conn is not None
+		conn = self._conn
+
+		#AAU#typ, msg_data = conn.uid('FETCH', mail.uid.encode('utf-8'), '(RFC822)') # body text (setting READ flag)
+		typ, msg_data = conn.uid('FETCH', mail.uid.encode('utf-8'), '(BODY.PEEK[])') # body text (without setting READ flag)
+		_LOGGER.debug("Msg data (length=%d):\n%s", len(msg_data),
+			      dbgindent(msg_data))
+
+		bbb = b''
+		for response_part in msg_data:
+			if isinstance(response_part, tuple):
+				bbb += response_part[1]
+
+		msg = email.message_from_bytes(bbb)
+		if msg is not None:
+			return message_text(msg)
+
+		return None
 
 
 	def request_folders(self) -> list[str]:
 		lst = []
-		
+
 		# Always create a new connection as an existing one may
 		# be used for IMAP IDLE.
 		conn = self._connect()
@@ -130,16 +174,16 @@ def request_folders(self) -> list[str]:
 			status, data = conn.list()
 		finally:
 			self._disconnect(conn)
-		
+
 		for d in data:
 			match = re.match(r'.+\s+("."|"?NIL"?)\s+"?([^"]+)"?$', d.decode('utf-8'))
 
 			if match is None:
-				logging.warning("Folder format not supported.")
+				_LOGGER.warning("Folder format not supported.")
 			else:
 				folder = match.group(2)
 				lst.append(decode_mutf7(folder))
-		
+
 		return lst
 
 
@@ -151,11 +195,11 @@ def mark_as_seen(self, mails: list[Mail]) -> None:
 		# Always create a new connection as an existing one may
 		# be used for IMAP IDLE.
 		conn = self._connect()
-		
+
 		try:
 			sorted_mails = sorted(mails, key = lambda m : m.flags['folder'] if 'folder' in m.flags else '')
 			last_folder = ''
-			
+
 			for m in sorted_mails:
 				if ('uid' in m.flags) and ('folder' in m.flags):
 					try:
@@ -165,12 +209,13 @@ def mark_as_seen(self, mails: list[Mail]) -> None:
 							last_folder = folder
 						status, data = conn.uid("STORE", m.flags['uid'], "+FLAGS", r"(\Seen)")
 					except:
-						logging.warning("Failed to set mail with uid %s to seen on server (account: '%s').", m.flags['uid'], self.name)
+						_LOGGER.warning("Failed to set mail with uid %s to seen on server (account: '%s').",
+                                                                m.flags['uid'], self.name)
 
 		finally:
 			self._disconnect(conn)
-	
-	
+
+
 	def supports_notifications(self) -> bool:
 		"""Returns True if mailbox supports notifications.
 		IMAP mailbox supports notifications if idle parameter is True"""
@@ -184,7 +229,7 @@ def notify_next_change(
 	) -> None:
 		self._ensure_open()
 		assert self._conn is not None
-		
+
 		# register idle callback that is called whenever an idle event
 		# arrives (new mail / mail deleted).
 		# the callback is called after <idle_timeout> minutes at the latest.
@@ -197,7 +242,7 @@ def _idle_callback(args: tuple[Any, Any, tuple[str, int]]) -> None:
 			# call actual callback
 			if callback is not None:
 				callback(error)
-		
+
 		self._conn.idle(callback = _idle_callback, timeout = timeout)
 
 
@@ -206,19 +251,47 @@ def cancel_notifications(self) -> None:
 		# Analogous to close().
 		# (Otherwise cleanup code like in Idler._idle() will fail)
 		# self._ensure_open()
-		
+
 		try:
 			if self._conn is not None:
 				# Exit possible active idle state.
 				# (also calls idle_callback)
 				self._conn.noop()
 		except:
 			pass
-
+
+
+	def _throttle(self, reset=False):
+		if not reset and self._last_access is not None:
+			duration = time.time() - self._last_access
+			if duration < self.THROTTLE_TIME:
+				time.sleep(self.THROTTLE_TIME - duration)
+		self._last_access = time.time()
+
+
+	def _refresh_token(self):
+		_LOGGER.debug("Refresh token...")
+		if not self.goa_account_id:
+			return True
+
+		token = refresh_goa_token(self.goa_account_id)
+		if token is None:
+			_LOGGER.debug("Refresh GOA token did not return token.")
+			return False
+
+		oauth2string = 'user=%s\1auth=Bearer %s\1\1' % (self.user, token[0])
+		if oauth2string == self.oauth2string:
+			_LOGGER.debug("OAuth2string did not change.")
+			return True
+
+		self.oauth2string = oauth2string
+		_LOGGER.debug("Token refreshed")
+		return True
+
 
 	def _connect(self) -> imaplib.IMAP4:
 		conn: Optional[imaplib.IMAP4] = None
-		
+
 		try:
 			if self.ssl:
 				if self.port == '':
@@ -230,51 +303,54 @@ def _connect(self) -> imaplib.IMAP4:
 					conn = imaplib.IMAP4(self.server)
 				else:
 					conn = imaplib.IMAP4(self.server, int(self.port))
-				
+
 				if 'STARTTLS' in conn.capabilities:
 					conn.starttls()
 				else:
-					logging.warning("Using unencrypted connection for account '%s'" % self.name)
-				
+					_LOGGER.warning("Using unencrypted connection for account '%s'", self.name)
+
 			if self.oauth2string != '':
+				self._refresh_token()
 				conn.authenticate('XOAUTH2', lambda x: self.oauth2string)
 			elif 'AUTH=CRAM-MD5' in conn.capabilities:
 				# use CRAM-MD5 auth if available
 				conn.login_cram_md5(self.user, self.password)
 			else:
 				conn.login(self.user, self.password)
+
+			self._throttle(reset=True)
 		except:
 			try:
 				if conn is not None:
 					# conn.close() # allowed in SELECTED state only
 					conn.logout()
 			except: pass
 			raise # re-throw exception
-		
+
 		# notify_next_change() (IMAP IDLE) requires a selected folder
 		if conn.state == AUTH:
 			self._select_single_folder(conn)
-		
+
 		return conn
 
 
 	def _disconnect(self, conn: imaplib.IMAP4) -> None:
 		try:
 			conn.close()
-		finally:		
-			# Closing the connection may fail (e.g. wrong state), 
+		finally:
+			# Closing the connection may fail (e.g. wrong state),
 			# but resources need to be freed anyway.
 			conn.logout()
-	
-	
+
+
 	def _select_single_folder(self, conn: imaplib.IMAP4) -> None:
 		if len(self.folders) == 1:
 			folder = self.folders[0]
 		else:
 			folder = "INBOX"
 		conn.select(f'"{folder}"', readonly = True)
-	
+
 	def _ensure_open(self) -> None:
 		if not self.is_open():
 			raise InvalidOperationException("Account is not open")
-
+		self._throttle()