keep processing after decoding errors during fetch

2 files changed, 145 insertions, 56 deletions

diff --git a/src/leap/mail/imap/fetch.py b/src/leap/mail/imap/fetch.py
index 40dadb3..6e12b3f 100644
--- a/src/leap/mail/imap/fetch.py
+++ b/src/leap/mail/imap/fetch.py
@@ -23,6 +23,7 @@ import threading
 import time
 import sys
 import traceback
+import warnings
 
 from email.parser import Parser
 from email.generator import Generator
@@ -32,6 +33,8 @@ from StringIO import StringIO
 from twisted.python import log
 from twisted.internet import defer
 from twisted.internet.task import LoopingCall
+from twisted.internet.task import deferLater
+from u1db import errors as u1db_errors
 from zope.proxy import sameProxiedObjects
 
 from leap.common import events as leap_events
@@ -46,7 +49,8 @@ from leap.common.mail import get_email_charset
 from leap.keymanager import errors as keymanager_errors
 from leap.keymanager.openpgp import OpenPGPKey
 from leap.mail.decorators import deferred_to_thread
-from leap.mail.utils import json_loads
+from leap.mail.imap.fields import fields
+from leap.mail.utils import json_loads, empty, first
 from leap.soledad.client import Soledad
 from leap.soledad.common.crypto import ENC_SCHEME_KEY, ENC_JSON_KEY
 
@@ -80,8 +84,6 @@ class LeapIncomingMail(object):
     """
 
     RECENT_FLAG = "\\Recent"
-
-    INCOMING_KEY = "incoming"
     CONTENT_KEY = "content"
 
     LEAP_SIGNATURE_HEADER = 'X-Leap-Signature'
@@ -130,17 +132,9 @@ class LeapIncomingMail(object):
         self._loop = None
         self._check_period = check_period
 
-        self._create_soledad_indexes()
-
         # initialize a mail parser only once
         self._parser = Parser()
 
-    def _create_soledad_indexes(self):
-        """
-        Create needed indexes on soledad.
-        """
-        self._soledad.create_index("just-mail", "incoming")
-
     @property
     def _pkey(self):
         if sameProxiedObjects(self._keymanager, None):
@@ -159,13 +153,29 @@ class LeapIncomingMail(object):
         Calls a deferred that will execute the fetch callback
         in a separate thread
         """
+        def syncSoledadCallback(result):
+            # FIXME this needs a matching change in mx!!!
+            # --> need to add ERROR_DECRYPTING_KEY = False
+            # as default.
+            try:
+                doclist = self._soledad.get_from_index(
+                    fields.JUST_MAIL_IDX, "*", "0")
+            except u1db_errors.InvalidGlobbing:
+                # It looks like we are a dealing with an outdated
+                # mx. Fallback to the version of the index
+                warnings.warn("JUST_MAIL_COMPAT_IDX will be deprecated!",
+                              DeprecationWarning)
+                doclist = self._soledad.get_from_index(
+                    fields.JUST_MAIL_COMPAT_IDX, "*")
+            self._process_doclist(doclist)
+
         logger.debug("fetching mail for: %s %s" % (
             self._soledad.uuid, self._userid))
         if not self.fetching_lock.locked():
             d1 = self._sync_soledad()
             d = defer.gatherResults([d1], consumeErrors=True)
+            d.addCallbacks(syncSoledadCallback, self._errback)
             d.addCallbacks(self._signal_fetch_to_ui, self._errback)
-            d.addCallbacks(self._signal_unread_to_ui, self._errback)
             return d
         else:
             logger.debug("Already fetching mail.")
@@ -202,46 +212,44 @@ class LeapIncomingMail(object):
     @deferred_to_thread
     def _sync_soledad(self):
         """
-        Synchronizes with remote soledad.
+        Synchronize with remote soledad.
 
         :returns: a list of LeapDocuments, or None.
         :rtype: iterable or None
         """
         with self.fetching_lock:
-            log.msg('syncing soledad...')
+            log.msg('FETCH: syncing soledad...')
             self._soledad.sync()
-            log.msg('soledad synced.')
-            doclist = self._soledad.get_from_index("just-mail", "*")
-        self._process_doclist(doclist)
-
-    def _signal_unread_to_ui(self, *args):
-        """
-        Sends unread event to ui.
-        """
-        leap_events.signal(
-            IMAP_UNREAD_MAIL, str(self._inbox.getUnseenCount()))
+            log.msg('FETCH soledad SYNCED.')
 
     def _signal_fetch_to_ui(self, doclist):
         """
-        Sends leap events to ui.
+        Send leap events to ui.
 
         :param doclist: iterable with msg documents.
         :type doclist: iterable.
         :returns: doclist
         :rtype: iterable
         """
-        doclist = doclist[0]  # gatherResults pass us a list
-        fetched_ts = time.mktime(time.gmtime())
-        num_mails = len(doclist) if doclist is not None else 0
-        if num_mails != 0:
-            log.msg("there are %s mails" % (num_mails,))
+        doclist = first(doclist)  # gatherResults pass us a list
+        if doclist:
+            fetched_ts = time.mktime(time.gmtime())
+            num_mails = len(doclist) if doclist is not None else 0
+            if num_mails != 0:
+                log.msg("there are %s mails" % (num_mails,))
+            leap_events.signal(
+                IMAP_FETCHED_INCOMING, str(num_mails), str(fetched_ts))
+            return doclist
+
+    def _signal_unread_to_ui(self, *args):
+        """
+        Sends unread event to ui.
+        """
         leap_events.signal(
-            IMAP_FETCHED_INCOMING, str(num_mails), str(fetched_ts))
-        return doclist
+            IMAP_UNREAD_MAIL, str(self._inbox.getUnseenCount()))
 
     # process incoming mail.
 
-    @defer.inlineCallbacks
     def _process_doclist(self, doclist):
         """
         Iterates through the doclist, checks if each doc
@@ -262,22 +270,36 @@ class LeapIncomingMail(object):
             logger.debug("processing doc %d of %d" % (index + 1, num_mails))
             leap_events.signal(
                 IMAP_MSG_PROCESSING, str(index), str(num_mails))
+
             keys = doc.content.keys()
-            if self._is_msg(keys):
-                # Ok, this looks like a legit msg.
+
+            # TODO Compatibility check with the index in pre-0.6 mx
+            # that does not write the ERROR_DECRYPTING_KEY
+            # This should be removed in 0.7
+
+            has_errors = doc.content.get(fields.ERROR_DECRYPTING_KEY, None)
+            if has_errors is None:
+                warnings.warn("JUST_MAIL_COMPAT_IDX will be deprecated!",
+                              DeprecationWarning)
+            if has_errors:
+                logger.debug("skipping msg with decrypting errors...")
+
+            if self._is_msg(keys) and not has_errors:
+                # Evaluating to bool of has_errors is intentional here.
+                # We don't mind at this point if it's None or False.
+
+                # Ok, this looks like a legit msg, and with no errors.
                 # Let's process it!
-                decrypted = list(self._decrypt_doc(doc))[0]
-                res = self._add_message_locally(decrypted)
-                yield res
 
-            else:
-                # Ooops, this does not.
-                logger.debug('This does not look like a proper msg.')
+                d1 = self._decrypt_doc(doc)
+                d = defer.gatherResults([d1], consumeErrors=True)
+                d.addCallbacks(self._add_message_locally, self._errback)
 
     #
     # operations on individual messages
     #
 
+    @deferred_to_thread
     def _decrypt_doc(self, doc):
         """
         Decrypt the contents of a document.
@@ -302,8 +324,8 @@ class LeapIncomingMail(object):
             decrdata = ""
         leap_events.signal(IMAP_MSG_DECRYPTED, "1" if success else "0")
 
-        data = list(self._process_decrypted_doc((doc, decrdata)))
-        yield (doc, data)
+        data = self._process_decrypted_doc((doc, decrdata))
+        return (doc, data)
 
     def _process_decrypted_doc(self, msgtuple):
         """
@@ -319,11 +341,35 @@ class LeapIncomingMail(object):
         """
         log.msg('processing decrypted doc')
         doc, data = msgtuple
-        msg = json_loads(data)
+
+        from twisted.internet import reactor
+
+        # XXX turn this into an errBack for each one of
+        # the deferreds that would process an individual document
+        try:
+            msg = json_loads(data)
+        except UnicodeError as exc:
+            logger.error("Error while decrypting %s" % (doc.doc_id,))
+            logger.exception(exc)
+
+            # we flag the message as "with decrypting errors",
+            # to avoid further decryption attempts during sync
+            # cycles until we're prepared to deal with that.
+            # What is the same, when Ivan deals with it...
+            # A new decrypting attempt event could be triggered by a
+            # future a library upgrade, or a cli flag to the client,
+            # we just `defer` that for now... :)
+            doc.content[fields.ERROR_DECRYPTING_KEY] = True
+            deferLater(reactor, 0, self._update_incoming_message, doc)
+
+            # FIXME this is just a dirty hack to delay the proper
+            # deferred organization here...
+            # and remember, boys, do not do this at home.
+            return []
 
         if not isinstance(msg, dict):
             defer.returnValue(False)
-        if not msg.get(self.INCOMING_KEY, False):
+        if not msg.get(fields.INCOMING_KEY, False):
             defer.returnValue(False)
 
         # ok, this is an incoming message
@@ -332,6 +378,27 @@ class LeapIncomingMail(object):
             return False
         return self._maybe_decrypt_msg(rawmsg)
 
+    @deferred_to_thread
+    def _update_incoming_message(self, doc):
+        """
+        Do a put for a soledad document. This probably has been called only
+        in the case that we've needed to update the ERROR_DECRYPTING_KEY
+        flag in an incoming message, to get it out of the decrypting queue.
+
+        :param doc: the SoledadDocument to update
+        :type doc: SoledadDocument
+        """
+        log.msg("Updating SoledadDoc %s" % (doc.doc_id))
+        self._soledad.put_doc(doc)
+
+    @deferred_to_thread
+    def _delete_incoming_message(self, doc):
+        """
+        Delete document.
+        """
+        log.msg("Deleting SoledadDoc %s" % (doc.doc_id))
+        self._soledad.delete_doc(doc)
+
     def _maybe_decrypt_msg(self, data):
         """
         Tries to decrypt a gpg message if data looks like one.
@@ -384,7 +451,7 @@ class LeapIncomingMail(object):
                 self.LEAP_SIGNATURE_INVALID,
                 pubkey=senderPubkey.key_id)
 
-        yield decrmsg.as_string()
+        return decrmsg.as_string()
 
     def _decrypt_multipart_encrypted_msg(self, msg, encoding, senderPubkey):
         """
@@ -505,32 +572,39 @@ class LeapIncomingMail(object):
                 data, self._pkey)
         return (decrdata, valid_sig)
 
-    def _add_message_locally(self, msgtuple):
+    def _add_message_locally(self, result):
         """
         Adds a message to local inbox and delete it from the incoming db
         in soledad.
 
+        # XXX this comes from a gatherresult...
         :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)
         """
-        log.msg('adding message to local db')
+        from twisted.internet import reactor
+        msgtuple = first(result)
+
         doc, data = msgtuple
+        log.msg('adding message %s to local db' % (doc.doc_id,))
 
         if isinstance(data, list):
+            if empty(data):
+                return False
             data = data[0]
 
-        self._inbox.addMessage(data, flags=(self.RECENT_FLAG,))
+        def msgSavedCallback(result):
+            if not empty(result):
+                leap_events.signal(IMAP_MSG_SAVED_LOCALLY)
+                deferLater(reactor, 0, self._delete_incoming_message, result)
+                leap_events.signal(IMAP_MSG_DELETED_INCOMING)
+                deferLater(reactor, 1, self._signal_unread_to_ui)
 
-        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)
-        self._signal_unread_to_ui()
-        return True
+        # XXX should pass a notify_on_disk=True along...
+        d = self._inbox.addMessage(data, flags=(self.RECENT_FLAG,))
+        d.addCallbacks(msgSavedCallback, self._errback)
 
     #
     # helpers
diff --git a/src/leap/mail/imap/fields.py b/src/leap/mail/imap/fields.py
index 886ee63..4576939 100644
--- a/src/leap/mail/imap/fields.py
+++ b/src/leap/mail/imap/fields.py
@@ -108,6 +108,14 @@ class WithMsgFields(object):
     # correct since the recent flag is volatile.
     TYPE_MBOX_RECT_SEEN_IDX = 'by-type-and-mbox-and-recent-and-seen'
 
+    # Soledad index for incoming mail, without decrypting errors.
+    JUST_MAIL_IDX = "just-mail"
+    # XXX the backward-compatible index, will be deprecated at 0.7
+    JUST_MAIL_COMPAT_IDX = "just-mail-compat"
+
+    INCOMING_KEY = "incoming"
+    ERROR_DECRYPTING_KEY = "errdecr"
+
     KTYPE = TYPE_KEY
     MBOX_VAL = TYPE_MBOX_VAL
     CHASH_VAL = CONTENT_HASH_KEY
@@ -140,6 +148,13 @@ class WithMsgFields(object):
         TYPE_MBOX_DEL_IDX: [KTYPE, MBOX_VAL, 'bool(deleted)'],
         TYPE_MBOX_RECT_SEEN_IDX: [KTYPE, MBOX_VAL,
                                   'bool(recent)', 'bool(seen)'],
+
+        # incoming queue
+        JUST_MAIL_IDX: [INCOMING_KEY,
+                        "bool(%s)" % (ERROR_DECRYPTING_KEY,)],
+
+        # the backward-compatible index, will be deprecated at 0.7
+        JUST_MAIL_COMPAT_IDX: [INCOMING_KEY],
     }
 
     MBOX_KEY = MBOX_VAL