Decrypt and process messages all in one transaction.

Giddy up

app/src/main/java/org/thoughtcrime/securesms/messages/IncomingMessageProcessor.java

@@ -7,17 +7,35 @@ import androidx.annotation.NonNull;
 import androidx.annotation.Nullable;
 
 import org.signal.core.util.logging.Log;
+import org.thoughtcrime.securesms.crypto.DatabaseSessionLock;
+import org.thoughtcrime.securesms.crypto.IdentityKeyUtil;
 import org.thoughtcrime.securesms.database.DatabaseFactory;
+import org.thoughtcrime.securesms.database.GroupDatabase;
 import org.thoughtcrime.securesms.database.MessageDatabase.SyncMessageId;
 import org.thoughtcrime.securesms.database.MmsSmsDatabase;
 import org.thoughtcrime.securesms.database.PushDatabase;
 import org.thoughtcrime.securesms.dependencies.ApplicationDependencies;
+import org.thoughtcrime.securesms.groups.BadGroupIdException;
+import org.thoughtcrime.securesms.groups.GroupChangeBusyException;
+import org.thoughtcrime.securesms.groups.GroupId;
+import org.thoughtcrime.securesms.jobmanager.Job;
 import org.thoughtcrime.securesms.jobmanager.JobManager;
 import org.thoughtcrime.securesms.jobs.PushDecryptMessageJob;
+import org.thoughtcrime.securesms.jobs.PushProcessMessageJob;
+import org.thoughtcrime.securesms.messages.MessageDecryptionUtil.DecryptionResult;
 import org.thoughtcrime.securesms.recipients.Recipient;
+import org.thoughtcrime.securesms.recipients.RecipientId;
+import org.thoughtcrime.securesms.util.GroupUtil;
+import org.thoughtcrime.securesms.util.SetUtil;
+import org.thoughtcrime.securesms.util.Stopwatch;
+import org.thoughtcrime.securesms.util.TextSecurePreferences;
+import org.whispersystems.signalservice.api.SignalSessionLock;
+import org.whispersystems.signalservice.api.messages.SignalServiceContent;
 import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope;
+import org.whispersystems.signalservice.api.messages.SignalServiceGroupContext;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.util.concurrent.locks.ReentrantLock;
 
 /**
@@ -52,13 +70,11 @@ public class IncomingMessageProcessor {
   public class Processor implements Closeable {
 
     private final Context           context;
-    private final PushDatabase      pushDatabase;
     private final MmsSmsDatabase    mmsSmsDatabase;
     private final JobManager        jobManager;
 
     private Processor(@NonNull Context context) {
       this.context           = context;
-      this.pushDatabase      = DatabaseFactory.getPushDatabase(context);
       this.mmsSmsDatabase    = DatabaseFactory.getMmsSmsDatabase(context);
       this.jobManager        = ApplicationDependencies.getJobManager();
     }
@@ -84,20 +100,51 @@ public class IncomingMessageProcessor {
     }
 
     private @Nullable String processMessage(@NonNull SignalServiceEnvelope envelope) {
-      Log.i(TAG, "Received message " + envelope.getTimestamp() + ". Inserting in PushDatabase.");
+      Log.i(TAG, "Received message " + envelope.getTimestamp() + ".");
 
-      long id  = pushDatabase.insert(envelope);
-
-      if (id > 0) {
-        PushDecryptMessageJob job = new PushDecryptMessageJob(context, id);
+      Stopwatch stopwatch = new Stopwatch("message");
 
+      if (needsToEnqueueDecryption()) {
+        Log.d(TAG, "Need to enqueue decryption.");
+        PushDecryptMessageJob job = new PushDecryptMessageJob(context, envelope);
         jobManager.add(job);
-
         return job.getId();
-      } else {
-        Log.w(TAG, "The envelope was already present in the PushDatabase.");
-        return null;
       }
+
+      stopwatch.split("queue-check");
+
+      try (SignalSessionLock.Lock unused = DatabaseSessionLock.INSTANCE.acquire()) {
+        DecryptionResult result = MessageDecryptionUtil.decrypt(context, envelope);
+        stopwatch.split("decrypt");
+
+        for (Job job : result.getJobs()) {
+          jobManager.add(job);
+        }
+
+        stopwatch.split("jobs");
+
+        if (needsToEnqueueProcessing(result)) {
+          Log.d(TAG, "Need to enqueue processing.");
+          jobManager.add(new PushProcessMessageJob(result.getState(), result.getContent(), result.getException(), -1, envelope.getTimestamp()));
+          return null;
+        }
+
+        stopwatch.split("group-check");
+
+        try {
+          MessageContentProcessor processor = new MessageContentProcessor(context);
+          processor.process(result.getState(), result.getContent(), result.getException(), envelope.getTimestamp(), -1);
+          return null;
+        } catch (IOException | GroupChangeBusyException e) {
+          Log.w(TAG, "Exception during message processing.", e);
+          jobManager.add(new PushProcessMessageJob(result.getState(), result.getContent(), result.getException(), -1, envelope.getTimestamp()));
+        }
+      } finally {
+        stopwatch.split("process");
+        stopwatch.stop(TAG);
+      }
+
+      return null;
     }
 
     private void processReceipt(@NonNull SignalServiceEnvelope envelope) {
@@ -106,6 +153,48 @@ public class IncomingMessageProcessor {
                                                    System.currentTimeMillis());
     }
 
+    private boolean needsToEnqueueDecryption() {
+      return !jobManager.areQueuesEmpty(SetUtil.newHashSet(Job.Parameters.MIGRATION_QUEUE_KEY, PushDecryptMessageJob.QUEUE)) ||
+             !IdentityKeyUtil.hasIdentityKey(context)                                                                        ||
+             TextSecurePreferences.getNeedsSqlCipherMigration(context);
+    }
+
+    private boolean needsToEnqueueProcessing(@NonNull DecryptionResult result) {
+      SignalServiceGroupContext groupContext = GroupUtil.getGroupContextIfPresent(result.getContent());
+
+      if (groupContext != null) {
+        try {
+          GroupId groupId = GroupUtil.idFromGroupContext(groupContext);
+
+          if (groupId.isV2()) {
+            String        queueName     = PushProcessMessageJob.getQueueName(Recipient.externalPossiblyMigratedGroup(context, groupId).getId());
+            GroupDatabase groupDatabase = DatabaseFactory.getGroupDatabase(context);
+
+            return !jobManager.isQueueEmpty(queueName)                                                                   ||
+                   groupContext.getGroupV2().get().getRevision() > groupDatabase.getGroupV2Revision(groupId.requireV2()) ||
+                   groupDatabase.getGroupV1ByExpectedV2(groupId.requireV2()).isPresent();
+          } else {
+            return false;
+          }
+        } catch (BadGroupIdException e) {
+          Log.w(TAG, "Bad group ID!");
+          return false;
+        }
+      } else if (result.getContent() != null) {
+        RecipientId recipientId = RecipientId.fromHighTrust(result.getContent().getSender());
+        String      queueKey    = PushProcessMessageJob.getQueueName(recipientId);
+
+        return !jobManager.isQueueEmpty(queueKey);
+      } else if (result.getException() != null) {
+        RecipientId recipientId = Recipient.external(context, result.getException().getSender()).getId();
+        String      queueKey    = PushProcessMessageJob.getQueueName(recipientId);
+
+        return !jobManager.isQueueEmpty(queueKey);
+      } else {
+        return false;
+      }
+    }
+
     @Override
     public void close() {
       release();

app/src/main/java/org/thoughtcrime/securesms/messages/MessageDecryptionUtil.java

@@ -0,0 +1,184 @@
+package org.thoughtcrime.securesms.messages;
+
+import android.content.Context;
+
+import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
+
+import org.signal.core.util.logging.Log;
+import org.signal.libsignal.metadata.InvalidMetadataMessageException;
+import org.signal.libsignal.metadata.InvalidMetadataVersionException;
+import org.signal.libsignal.metadata.ProtocolDuplicateMessageException;
+import org.signal.libsignal.metadata.ProtocolException;
+import org.signal.libsignal.metadata.ProtocolInvalidKeyException;
+import org.signal.libsignal.metadata.ProtocolInvalidKeyIdException;
+import org.signal.libsignal.metadata.ProtocolInvalidMessageException;
+import org.signal.libsignal.metadata.ProtocolInvalidVersionException;
+import org.signal.libsignal.metadata.ProtocolLegacyMessageException;
+import org.signal.libsignal.metadata.ProtocolNoSessionException;
+import org.signal.libsignal.metadata.ProtocolUntrustedIdentityException;
+import org.signal.libsignal.metadata.SelfSendException;
+import org.thoughtcrime.securesms.crypto.DatabaseSessionLock;
+import org.thoughtcrime.securesms.crypto.UnidentifiedAccessUtil;
+import org.thoughtcrime.securesms.crypto.storage.SignalProtocolStoreImpl;
+import org.thoughtcrime.securesms.groups.BadGroupIdException;
+import org.thoughtcrime.securesms.groups.GroupId;
+import org.thoughtcrime.securesms.jobmanager.Job;
+import org.thoughtcrime.securesms.jobs.AutomaticSessionResetJob;
+import org.thoughtcrime.securesms.jobs.RefreshPreKeysJob;
+import org.thoughtcrime.securesms.messages.MessageContentProcessor.ExceptionMetadata;
+import org.thoughtcrime.securesms.messages.MessageContentProcessor.MessageState;
+import org.thoughtcrime.securesms.recipients.Recipient;
+import org.thoughtcrime.securesms.util.GroupUtil;
+import org.thoughtcrime.securesms.util.TextSecurePreferences;
+import org.whispersystems.libsignal.state.SignalProtocolStore;
+import org.whispersystems.libsignal.util.guava.Optional;
+import org.whispersystems.signalservice.api.crypto.SignalServiceCipher;
+import org.whispersystems.signalservice.api.messages.SignalServiceContent;
+import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope;
+import org.whispersystems.signalservice.api.push.SignalServiceAddress;
+import org.whispersystems.signalservice.internal.push.UnsupportedDataMessageException;
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Handles taking an encrypted {@link SignalServiceEnvelope} and turning it into a plaintext model.
+ */
+public final class MessageDecryptionUtil {
+
+  private static final String TAG = Log.tag(MessageDecryptionUtil.class);
+
+  private MessageDecryptionUtil() {}
+
+  /**
+   * Takes a {@link SignalServiceEnvelope} and returns a {@link DecryptionResult}, which has either
+   * a plaintext {@link SignalServiceContent} or information about an error that happened.
+   *
+   * Excluding the data updated in our protocol stores that results from decrypting a message, this
+   * method is side-effect free, preferring to return the decryption results to be handled by the
+   * caller.
+   */
+  public static @NonNull DecryptionResult decrypt(@NonNull Context context, @NonNull SignalServiceEnvelope envelope) {
+    SignalProtocolStore  axolotlStore = new SignalProtocolStoreImpl(context);
+    SignalServiceAddress localAddress = new SignalServiceAddress(Optional.of(TextSecurePreferences.getLocalUuid(context)), Optional.of(TextSecurePreferences.getLocalNumber(context)));
+    SignalServiceCipher  cipher       = new SignalServiceCipher(localAddress, axolotlStore, DatabaseSessionLock.INSTANCE, UnidentifiedAccessUtil.getCertificateValidator());
+    List<Job>            jobs         = new LinkedList<>();
+
+    if (envelope.isPreKeySignalMessage()) {
+      jobs.add(new RefreshPreKeysJob());
+    }
+
+    try {
+      try {
+        return DecryptionResult.forSuccess(cipher.decrypt(envelope), jobs);
+      } catch (ProtocolInvalidVersionException e) {
+        Log.w(TAG, String.valueOf(envelope.getTimestamp()), e);
+        return DecryptionResult.forError(MessageState.INVALID_VERSION, toExceptionMetadata(e), jobs);
+
+      } catch (ProtocolInvalidMessageException | ProtocolInvalidKeyIdException | ProtocolInvalidKeyException | ProtocolUntrustedIdentityException | ProtocolNoSessionException e) {
+        Log.w(TAG, String.valueOf(envelope.getTimestamp()), e);
+        jobs.add(new AutomaticSessionResetJob(Recipient.external(context, e.getSender()).getId(), e.getSenderDevice(), envelope.getTimestamp()));
+        return DecryptionResult.forNoop(jobs);
+      } catch (ProtocolLegacyMessageException e) {
+        Log.w(TAG, String.valueOf(envelope.getTimestamp()), e);
+        return DecryptionResult.forError(MessageState.LEGACY_MESSAGE, toExceptionMetadata(e), jobs);
+      } catch (ProtocolDuplicateMessageException e) {
+        Log.w(TAG, String.valueOf(envelope.getTimestamp()), e);
+        return DecryptionResult.forError(MessageState.DUPLICATE_MESSAGE, toExceptionMetadata(e), jobs);
+      } catch (InvalidMetadataVersionException | InvalidMetadataMessageException e) {
+        Log.w(TAG, String.valueOf(envelope.getTimestamp()), e);
+        return DecryptionResult.forNoop(jobs);
+      } catch (SelfSendException e) {
+        Log.i(TAG, "Dropping UD message from self.");
+        return DecryptionResult.forNoop(jobs);
+      } catch (UnsupportedDataMessageException e) {
+        Log.w(TAG, String.valueOf(envelope.getTimestamp()), e);
+        return DecryptionResult.forError(MessageState.UNSUPPORTED_DATA_MESSAGE, toExceptionMetadata(e), jobs);
+      }
+    } catch (NoSenderException e) {
+      Log.w(TAG, "Invalid message, but no sender info!");
+      return DecryptionResult.forNoop(jobs);
+    }
+  }
+
+  private static ExceptionMetadata toExceptionMetadata(@NonNull UnsupportedDataMessageException e)
+      throws NoSenderException
+  {
+    String sender = e.getSender();
+
+    if (sender == null) throw new NoSenderException();
+
+    GroupId groupId = null;
+
+    if (e.getGroup().isPresent()) {
+      try {
+        groupId = GroupUtil.idFromGroupContext(e.getGroup().get());
+      } catch (BadGroupIdException ex) {
+        Log.w(TAG, "Bad group id found in unsupported data message", ex);
+      }
+    }
+
+    return new ExceptionMetadata(sender, e.getSenderDevice(), groupId);
+  }
+
+  private static ExceptionMetadata toExceptionMetadata(@NonNull ProtocolException e) throws NoSenderException {
+    String sender = e.getSender();
+
+    if (sender == null) throw new NoSenderException();
+
+    return new ExceptionMetadata(sender, e.getSenderDevice());
+  }
+
+  private static class NoSenderException extends Exception {}
+
+  public static class DecryptionResult {
+    private final @NonNull  MessageState         state;
+    private final @Nullable SignalServiceContent content;
+    private final @Nullable ExceptionMetadata    exception;
+    private final @NonNull  List<Job>            jobs;
+
+    static @NonNull DecryptionResult forSuccess(@NonNull SignalServiceContent content, @NonNull List<Job> jobs) {
+      return new DecryptionResult(MessageState.DECRYPTED_OK, content, null, jobs);
+    }
+
+    static @NonNull DecryptionResult forError(@NonNull MessageState messageState,
+                                              @NonNull ExceptionMetadata exception,
+                                              @NonNull List<Job> jobs)
+    {
+      return new DecryptionResult(messageState, null, exception, jobs);
+    }
+
+    static @NonNull DecryptionResult forNoop(@NonNull List<Job> jobs) {
+      return new DecryptionResult(MessageState.NOOP, null, null, jobs);
+    }
+
+    private DecryptionResult(@NonNull MessageState state,
+                             @Nullable SignalServiceContent content,
+                             @Nullable ExceptionMetadata exception,
+                             @NonNull List<Job> jobs)
+    {
+      this.state     = state;
+      this.content   = content;
+      this.exception = exception;
+      this.jobs      = jobs;
+    }
+
+    public @NonNull MessageState getState() {
+      return state;
+    }
+
+    public @Nullable SignalServiceContent getContent() {
+      return content;
+    }
+
+    public @Nullable ExceptionMetadata getException() {
+      return exception;
+    }
+
+    public @NonNull List<Job> getJobs() {
+      return jobs;
+    }
+  }
+}