diff options
Diffstat (limited to 'mail')
| -rw-r--r-- | mail/changes/feature_3423_refactor_imap_fetch | 1 | ||||
| -rw-r--r-- | mail/src/leap/mail/imap/fetch.py | 217 |
2 files changed, 153 insertions, 65 deletions
diff --git a/mail/changes/feature_3423_refactor_imap_fetch b/mail/changes/feature_3423_refactor_imap_fetch new file mode 100644 index 0000000..cacceef --- /dev/null +++ b/mail/changes/feature_3423_refactor_imap_fetch @@ -0,0 +1 @@ + o Refactor imap fetch code for better defer handling. Closes: #3423 diff --git a/mail/src/leap/mail/imap/fetch.py b/mail/src/leap/mail/imap/fetch.py index 2b25d82..8b29c5e 100644 --- a/mail/src/leap/mail/imap/fetch.py +++ b/mail/src/leap/mail/imap/fetch.py @@ -20,10 +20,10 @@ Incoming mail fetcher. import logging import json import ssl +import threading import time from twisted.python import log -from twisted.internet import defer from twisted.internet.task import LoopingCall from twisted.internet.threads import deferToThread @@ -53,6 +53,8 @@ class LeapIncomingMail(object): INCOMING_KEY = "incoming" CONTENT_KEY = "content" + fetching_lock = threading.Lock() + def __init__(self, keymanager, soledad, imap_account, check_period): @@ -95,6 +97,10 @@ class LeapIncomingMail(object): """ self._soledad.create_index("just-mail", "incoming") + # + # Public API: fetch, start_loop, stop. + # + def fetch(self): """ Fetch incoming mail, to be called periodically. @@ -102,9 +108,13 @@ class LeapIncomingMail(object): Calls a deferred that will execute the fetch callback in a separate thread """ - d = deferToThread(self._sync_soledad) - d.addCallbacks(self._process_doclist, self._sync_soledad_err) - return d + if not self.fetching_lock.locked(): + d = deferToThread(self._sync_soledad) + d.addCallbacks(self._signal_fetch_to_ui, self._sync_soledad_error) + d.addCallbacks(self._process_doclist, self._sync_soledad_error) + return d + else: + logger.debug("Already fetching mail.") def start_loop(self): """ @@ -117,52 +127,113 @@ class LeapIncomingMail(object): """ Stops the loop that fetches mail. """ + # XXX should cancel ongoing fetches too. if self._loop and self._loop.running is True: self._loop.stop() + # + # Private methods. + # + + # synchronize incoming mail + def _sync_soledad(self): - log.msg('syncing soledad...') + """ + Synchronizes with remote soledad. - try: + :returns: a list of LeapDocuments, or None. + :rtype: iterable or None + """ + with self.fetching_lock: + log.msg('syncing soledad...') self._soledad.sync() - fetched_ts = time.mktime(time.gmtime()) doclist = self._soledad.get_from_index("just-mail", "*") - num_mails = len(doclist) - log.msg("there are %s mails" % (num_mails,)) - leap_events.signal( - IMAP_FETCHED_INCOMING, str(num_mails), str(fetched_ts)) - leap_events.signal( - IMAP_UNREAD_MAIL, str(self._inbox.getUnseenCount())) - return doclist - except ssl.SSLError as exc: - logger.warning('SSL Error while syncing soledad: %r' % (exc,)) - except Exception as exc: - logger.warning('Error while syncing soledad: %r' % (exc,)) + return doclist - def _sync_soledad_err(self, f): - log.err("error syncing soledad: %s" % (f.value,)) - return f + def _signal_fetch_to_ui(self, doclist): + """ + Sends leap events to ui. + + :param doclist: iterable with msg documents. + :type doclist: iterable. + :returns: doclist + :rtype: iterable + """ + fetched_ts = time.mktime(time.gmtime()) + num_mails = len(doclist) + log.msg("there are %s mails" % (num_mails,)) + leap_events.signal( + IMAP_FETCHED_INCOMING, str(num_mails), str(fetched_ts)) + leap_events.signal( + IMAP_UNREAD_MAIL, str(self._inbox.getUnseenCount())) + return doclist + + def _sync_soledad_error(self, failure): + """ + Errback for sync errors. + """ + # XXX should signal unrecoverable maybe. + err = failure.value + logger.error("error syncing soledad: %s" % (err,)) + if failure.check(ssl.SSLError): + logger.warning('SSL Error while ' + 'syncing soledad: %r' % (err,)) + elif failure.check(Exception): + logger.warning('Unknown error while ' + 'syncing soledad: %r' % (err,)) def _process_doclist(self, doclist): + """ + Iterates through the doclist, checks if each doc + looks like a message, and yields a deferred that will decrypt and + process the message. + + :param doclist: iterable with msg documents. + :type doclist: iterable. + :returns: a list of deferreds for individual messages. + """ log.msg('processing doclist') if not doclist: logger.debug("no docs found") return num_mails = len(doclist) + + docs_cb = [] for index, doc in enumerate(doclist): logger.debug("processing doc %d of %d: %s" % ( index, num_mails, doc)) leap_events.signal( IMAP_MSG_PROCESSING, str(index), str(num_mails)) keys = doc.content.keys() - if ENC_SCHEME_KEY in keys and ENC_JSON_KEY in keys: - - # XXX should check for _enc_scheme == "pubkey" || "none" - # that is what incoming mail uses. + if self._is_msg(keys): + # Ok, this looks like a legit msg. + # Let's process it! encdata = doc.content[ENC_JSON_KEY] - defer.Deferred(self._decrypt_msg(doc, encdata)) + + # Deferred chain for individual messages + d = deferToThread(self._decrypt_msg, doc, encdata) + d.addCallback(self._process_decrypted) + d.addCallback(self._add_message_locally) + docs_cb.append(d) else: + # Ooops, this does not. logger.debug('This does not look like a proper msg.') + return docs_cb + + # + # operations on individual messages + # + + def _is_msg(self, keys): + """ + Checks if the keys of a dictionary match the signature + of the document type we use for messages. + + :param keys: iterable containing the strings to match. + :type keys: iterable of strings. + :rtype: bool + """ + return ENC_SCHEME_KEY in keys and ENC_JSON_KEY in keys def _decrypt_msg(self, doc, encdata): log.msg('decrypting msg') @@ -170,64 +241,80 @@ class LeapIncomingMail(object): try: decrdata = (self._keymanager.decrypt( encdata, key, - # XXX get from public method instead - passphrase=self._soledad._passphrase)) + passphrase=self._soledad.passphrase)) ok = True except Exception as exc: + # XXX move this to errback !!! logger.warning("Error while decrypting msg: %r" % (exc,)) decrdata = "" ok = False leap_events.signal(IMAP_MSG_DECRYPTED, "1" if ok else "0") - # XXX TODO: defer this properly - return self._process_decrypted(doc, decrdata) + return doc, decrdata - def _process_decrypted(self, doc, data): + def _process_decrypted(self, msgtuple): """ Process a successfully decrypted message. - :param doc: a SoledadDocument instance containing the incoming message - :type doc: SoledadDocument - - :param data: the json-encoded, decrypted content of the incoming - message - :type data: str - - :param inbox: a open SoledadMailbox instance where this message is - to be saved - :type inbox: SoledadMailbox + :param msgtuple: a tuple consisting of a SoledadDocument + instance containing the incoming message + and data, the json-encoded, decrypted content of the + incoming message + :type msgtuple: (SoledadDocument, str) + :returns: a SoledadDocument and the processed data. + :rtype: (doc, data) """ - log.msg("processing incoming message!") + doc, data = msgtuple msg = json.loads(data) if not isinstance(msg, dict): return False if not msg.get(self.INCOMING_KEY, False): return False + # ok, this is an incoming message rawmsg = msg.get(self.CONTENT_KEY, None) if not rawmsg: return False logger.debug('got incoming message: %s' % (rawmsg,)) + data = self._maybe_decrypt_gpg_msg(rawmsg) + return doc, data - # XXX factor out gpg bits. - try: - pgp_beg = "-----BEGIN PGP MESSAGE-----" - pgp_end = "-----END PGP MESSAGE-----" - if pgp_beg in rawmsg: - first = rawmsg.find(pgp_beg) - last = rawmsg.rfind(pgp_end) - pgp_message = rawmsg[first:first+last] - - decrdata = (self._keymanager.decrypt( - pgp_message, self._pkey, - # XXX get from public method instead - passphrase=self._soledad._passphrase)) - rawmsg = rawmsg.replace(pgp_message, decrdata) - # add to inbox and delete from soledad - self._inbox.addMessage(rawmsg, (self.RECENT_FLAG,)) - leap_events.signal(IMAP_MSG_SAVED_LOCALLY) - doc_id = doc.doc_id - self._soledad.delete_doc(doc) - log.msg("deleted doc %s from incoming" % doc_id) - leap_events.signal(IMAP_MSG_DELETED_INCOMING) - except Exception as e: - logger.error("Problem processing incoming mail: %r" % (e,)) + def _maybe_decrypt_gpg_msg(self, data): + """ + Tries to decrypt a gpg message if data looks like one. + + :param data: the text to be decrypted. + :type data: str + :return: data, possibly descrypted. + :rtype: str + """ + PGP_BEGIN = "-----BEGIN PGP MESSAGE-----" + PGP_END = "-----END PGP MESSAGE-----" + if PGP_BEGIN in data: + begin = data.find(PGP_BEGIN) + end = data.rfind(PGP_END) + pgp_message = data[begin:begin+end] + + decrdata = (self._keymanager.decrypt( + pgp_message, self._pkey, + passphrase=self._soledad.passphrase)) + data = data.replace(pgp_message, decrdata) + return data + + def _add_message_locally(self, msgtuple): + """ + Adds a message to local inbox and delete it from the incoming db + in soledad. + + :param msgtuple: a tuple consisting of a SoledadDocument + instance containing the incoming message + and data, the json-encoded, decrypted content of the + incoming message + :type msgtuple: (SoledadDocument, str) + """ + doc, data = msgtuple + self._inbox.addMessage(data, (self.RECENT_FLAG,)) + leap_events.signal(IMAP_MSG_SAVED_LOCALLY) + doc_id = doc.doc_id + self._soledad.delete_doc(doc) + log.msg("deleted doc %s from incoming" % doc_id) + leap_events.signal(IMAP_MSG_DELETED_INCOMING) |