""" Internal Python-level IMAP handling used by the testplugin and for cleaning up inbox/mvbox for each test function run. """ import io import ssl import pathlib from contextlib import contextmanager from imap_tools import MailBox, MailBoxTls, errors, AND, Header, MailMessageFlags, MailMessage import imaplib from deltachat import const, Account from typing import List FLAGS = b'FLAGS' FETCH = b'FETCH' ALL = "1:*" class DirectImap: def __init__(self, account: Account) -> None: self.account = account self.logid = account.get_config("displayname") or id(account) self._idling = False self.connect() def connect(self): host = self.account.get_config("configured_mail_server") port = int(self.account.get_config("configured_mail_port")) security = int(self.account.get_config("configured_mail_security")) user = self.account.get_config("addr") pw = self.account.get_config("mail_pw") if security == const.DC_SOCKET_PLAIN: ssl_context = None else: ssl_context = ssl.create_default_context() # don't check if certificate hostname doesn't match target hostname ssl_context.check_hostname = False # don't check if the certificate is trusted by a certificate authority ssl_context.verify_mode = ssl.CERT_NONE if security == const.DC_SOCKET_STARTTLS: self.conn = MailBoxTls(host, port, ssl_context=ssl_context) elif security == const.DC_SOCKET_PLAIN or security == const.DC_SOCKET_SSL: self.conn = MailBox(host, port, ssl_context=ssl_context) self.conn.login(user, pw) self.select_folder("INBOX") def shutdown(self): try: self.conn.logout() except (OSError, imaplib.IMAP4.abort): print("Could not logout direct_imap conn") def create_folder(self, foldername): try: self.conn.folder.create(foldername) except errors.MailboxFolderCreateError as e: print("Can't create", foldername, "probably it already exists:", str(e)) def select_folder(self, foldername: str) -> tuple: assert not self._idling return self.conn.folder.set(foldername) def select_config_folder(self, config_name: str): """ Return info about selected folder if it is configured, otherwise None. """ if "_" not in config_name: config_name = "configured_{}_folder".format(config_name) foldername = self.account.get_config(config_name) if foldername: return self.select_folder(foldername) def list_folders(self) -> List[str]: """ return list of all existing folder names""" assert not self._idling return [folder.name for folder in self.conn.folder.list()] def delete(self, uid_list: str, expunge=True): """ delete a range of messages (imap-syntax). If expunge is true, perform the expunge-operation to make sure the messages are really gone and not just flagged as deleted. """ self.conn.client.uid('STORE', uid_list, '+FLAGS', r'(\Deleted)') if expunge: self.conn.expunge() def get_all_messages(self) -> List[MailMessage]: assert not self._idling return [mail for mail in self.conn.fetch()] def get_unread_messages(self) -> List[str]: assert not self._idling return [msg.uid for msg in self.conn.fetch(AND(seen=False))] def mark_all_read(self): messages = self.get_unread_messages() if messages: res = self.conn.flag(messages, MailMessageFlags.SEEN, True) print("marked seen:", messages, res) def get_unread_cnt(self) -> int: return len(self.get_unread_messages()) def dump_imap_structures(self, dir, logfile): assert not self._idling stream = io.StringIO() def log(*args, **kwargs): kwargs["file"] = stream print(*args, **kwargs) empty_folders = [] for imapfolder in self.list_folders(): self.select_folder(imapfolder) messages = list(self.get_all_messages()) if not messages: empty_folders.append(imapfolder) continue log("---------", imapfolder, len(messages), "messages ---------") # get message content without auto-marking it as seen # fetching 'RFC822' would mark it as seen. for msg in self.conn.fetch(mark_seen=False): body = getattr(msg.obj, "text", None) if not body: body = getattr(msg.obj, "html", None) if not body: log("Message", msg.uid, "has empty body") continue path = pathlib.Path(str(dir)).joinpath("IMAP", self.logid, imapfolder) path.mkdir(parents=True, exist_ok=True) fn = path.joinpath(str(msg.uid)) fn.write_bytes(body) log("Message", msg.uid, fn) log("Message", msg.uid, msg.flags, "Message-Id:", msg.obj.get("Message-Id")) if empty_folders: log("--------- EMPTY FOLDERS:", empty_folders) print(stream.getvalue(), file=logfile) @contextmanager def idle(self): """ return Idle ContextManager. """ idle_manager = IdleManager(self) try: yield idle_manager finally: idle_manager.done() def append(self, folder: str, msg: str): """Upload a message to *folder*. Trailing whitespace or a linebreak at the beginning will be removed automatically. """ if msg.startswith("\n"): msg = msg[1:] msg = '\n'.join([s.lstrip() for s in msg.splitlines()]) self.conn.append(bytes(msg, encoding='ascii'), folder) def get_uid_by_message_id(self, message_id) -> str: msgs = [msg.uid for msg in self.conn.fetch(AND(header=Header('MESSAGE-ID', message_id)))] if len(msgs) == 0: raise Exception("Did not find message " + message_id + ", maybe you forgot to select the correct folder?") return msgs[0] class IdleManager: def __init__(self, direct_imap): self.direct_imap = direct_imap self.log = direct_imap.account.log # fetch latest messages before starting idle so that it only # returns messages that arrive anew self.direct_imap.conn.fetch("1:*") self.direct_imap.conn.idle.start() def check(self, timeout=None) -> List[bytes]: """ (blocking) wait for next idle message from server. """ self.log("imap-direct: calling idle_check") res = self.direct_imap.conn.idle.poll(timeout=timeout) self.log("imap-direct: idle_check returned {!r}".format(res)) return res def wait_for_new_message(self, timeout=None) -> bytes: while 1: for item in self.check(timeout=timeout): if b'EXISTS' in item or b'RECENT' in item: return item def wait_for_seen(self, timeout=None) -> int: """ Return first message with SEEN flag from a running idle-stream. """ while 1: for item in self.check(timeout=timeout): if FETCH in item: self.log(str(item)) if FLAGS in item and rb'\Seen' in item: return int(item.split(b' ')[1]) def done(self): """ send idle-done to server if we are currently in idle mode. """ res = self.direct_imap.conn.idle.stop() return res