Add support for OMEMO call encryption

11 files changed, 605 insertions, 402 deletions

diff --git a/plugins/ice/src/transport_parameters.vala b/plugins/ice/src/transport_parameters.vala
index 52451fcf..38652952 100644
--- a/plugins/ice/src/transport_parameters.vala
+++ b/plugins/ice/src/transport_parameters.vala
@@ -77,7 +77,10 @@ public class Dino.Plugins.Ice.TransportParameters : JingleIceUdp.IceUdpTransport
                 own_setup = "actpass";
                 dtls_srtp_handler.mode = DtlsSrtp.Mode.SERVER;
                 dtls_srtp_handler.setup_dtls_connection.begin((_, res) => {
-                    this.content.encryption = dtls_srtp_handler.setup_dtls_connection.end(res) ?? this.content.encryption;
+                    var content_encryption = dtls_srtp_handler.setup_dtls_connection.end(res);
+                    if (content_encryption != null) {
+                        this.content.encryptions[content_encryption.encryption_ns] = content_encryption;
+                    }
                 });
             }
         }
@@ -157,7 +160,10 @@ public class Dino.Plugins.Ice.TransportParameters : JingleIceUdp.IceUdpTransport
                 dtls_srtp_handler.mode = DtlsSrtp.Mode.CLIENT;
                 dtls_srtp_handler.stop_dtls_connection();
                 dtls_srtp_handler.setup_dtls_connection.begin((_, res) => {
-                    this.content.encryption = dtls_srtp_handler.setup_dtls_connection.end(res) ?? this.content.encryption;
+                    var content_encryption = dtls_srtp_handler.setup_dtls_connection.end(res);
+                    if (content_encryption != null) {
+                        this.content.encryptions[content_encryption.encryption_ns] = content_encryption;
+                    }
                 });
             }
         } else {
@@ -225,7 +231,10 @@ public class Dino.Plugins.Ice.TransportParameters : JingleIceUdp.IceUdpTransport
         may_consider_ready(stream_id, component_id);
         if (incoming && dtls_srtp_handler != null && !dtls_srtp_handler.ready && is_component_ready(agent, stream_id, component_id) && dtls_srtp_handler.mode == DtlsSrtp.Mode.CLIENT) {
             dtls_srtp_handler.setup_dtls_connection.begin((_, res) => {
-                this.content.encryption = dtls_srtp_handler.setup_dtls_connection.end(res) ?? this.content.encryption;
+                Jingle.ContentEncryption? encryption = dtls_srtp_handler.setup_dtls_connection.end(res);
+                if (encryption != null) {
+                    this.content.encryptions[encryption.encryption_ns] = encryption;
+                }
             });
         }
     }
diff --git a/plugins/omemo/CMakeLists.txt b/plugins/omemo/CMakeLists.txt
index c7a45069..944fc649 100644
--- a/plugins/omemo/CMakeLists.txt
+++ b/plugins/omemo/CMakeLists.txt
@@ -29,6 +29,7 @@ compile_gresources(
 
 vala_precompile(OMEMO_VALA_C
 SOURCES
+    src/dtls_srtp_verification_draft.vala
     src/plugin.vala
     src/register_plugin.vala
     src/trust_level.vala
@@ -39,7 +40,8 @@ SOURCES
     src/jingle/jet_omemo.vala
 
     src/logic/database.vala
-    src/logic/encrypt_state.vala
+    src/logic/decrypt.vala
+    src/logic/encrypt.vala
     src/logic/manager.vala
     src/logic/pre_key_store.vala
     src/logic/session_store.vala
diff --git a/plugins/omemo/src/dtls_srtp_verification_draft.vala b/plugins/omemo/src/dtls_srtp_verification_draft.vala
new file mode 100644
index 00000000..e2441670
--- /dev/null
+++ b/plugins/omemo/src/dtls_srtp_verification_draft.vala
@@ -0,0 +1,195 @@
+using Signal;
+using Gee;
+using Xmpp;
+
+namespace Dino.Plugins.Omemo.DtlsSrtpVerificationDraft {
+    public const string NS_URI = "http://gultsch.de/xmpp/drafts/omemo/dlts-srtp-verification";
+
+    public class StreamModule : XmppStreamModule {
+
+        public static Xmpp.ModuleIdentity<StreamModule> IDENTITY = new Xmpp.ModuleIdentity<StreamModule>(NS_URI, "dtls_srtp_omemo_verification_draft");
+
+        private VerificationSendListener send_listener = new VerificationSendListener();
+        private HashMap<string, int> device_id_by_jingle_sid = new HashMap<string, int>();
+        private HashMap<string, Gee.List<string>> content_names_by_jingle_sid = new HashMap<string, Gee.List<string>>();
+
+        private void on_preprocess_incoming_iq_set_get(XmppStream stream, Xmpp.Iq.Stanza iq) {
+            if (iq.type_ != Iq.Stanza.TYPE_SET) return;
+
+            Gee.List<StanzaNode> content_nodes = iq.stanza.get_deep_subnodes(Xep.Jingle.NS_URI + ":jingle", Xep.Jingle.NS_URI + ":content");
+            if (content_nodes.size == 0) return;
+
+            string? jingle_sid = iq.stanza.get_deep_attribute(Xep.Jingle.NS_URI + ":jingle", "sid");
+            if (jingle_sid == null) return;
+
+            Xep.Omemo.OmemoDecryptor decryptor = stream.get_module(Xep.Omemo.OmemoDecryptor.IDENTITY);
+
+            foreach (StanzaNode content_node in content_nodes) {
+                string? content_name = content_node.get_attribute("name");
+                if (content_name == null) continue;
+                StanzaNode? transport_node = content_node.get_subnode("transport", Xep.JingleIceUdp.NS_URI);
+                if (transport_node == null) continue;
+                StanzaNode? fingerprint_node = transport_node.get_subnode("fingerprint", NS_URI);
+                if (fingerprint_node == null) continue;
+                StanzaNode? encrypted_node = fingerprint_node.get_subnode("encrypted", Omemo.NS_URI);
+                if (encrypted_node == null) continue;
+
+                Xep.Omemo.ParsedData? parsed_data = decryptor.parse_node(encrypted_node);
+                if (parsed_data == null || parsed_data.ciphertext == null) continue;
+
+                if (device_id_by_jingle_sid.has_key(jingle_sid) && device_id_by_jingle_sid[jingle_sid] != parsed_data.sid) {
+                    warning("Expected DTLS fingerprint to be OMEMO encrypted from %s %d, but it was from %d", iq.from.to_string(), device_id_by_jingle_sid[jingle_sid], parsed_data.sid);
+                }
+
+                foreach (Bytes encr_key in parsed_data.our_potential_encrypted_keys.keys) {
+                    parsed_data.is_prekey = parsed_data.our_potential_encrypted_keys[encr_key];
+                    parsed_data.encrypted_key = encr_key.get_data();
+
+                    try {
+                        uint8[] key = decryptor.decrypt_key(parsed_data, iq.from.bare_jid);
+                        string cleartext = decryptor.decrypt(parsed_data.ciphertext, key, parsed_data.iv);
+
+                        StanzaNode new_fingerprint_node = new StanzaNode.build("fingerprint", Xep.JingleIceUdp.DTLS_NS_URI).add_self_xmlns()
+                                .put_node(new StanzaNode.text(cleartext));
+                        string? hash_attr = fingerprint_node.get_attribute("hash", NS_URI);
+                        string? setup_attr = fingerprint_node.get_attribute("setup", NS_URI);
+                        if (hash_attr != null) new_fingerprint_node.put_attribute("hash", hash_attr);
+                        if (setup_attr != null) new_fingerprint_node.put_attribute("setup", setup_attr);
+                        transport_node.put_node(new_fingerprint_node);
+
+                        device_id_by_jingle_sid[jingle_sid] = parsed_data.sid;
+                        if (!content_names_by_jingle_sid.has_key(content_name)) {
+                            content_names_by_jingle_sid[content_name] = new ArrayList<string>();
+                        }
+                        content_names_by_jingle_sid[content_name].add(content_name);
+
+                        stream.get_flag(Xep.Jingle.Flag.IDENTITY).get_session.begin(jingle_sid, (_, res) => {
+                            Xep.Jingle.Session? session = stream.get_flag(Xep.Jingle.Flag.IDENTITY).get_session.end(res);
+                            if (session != null) print(@"$(session.contents_map.has_key(content_name))\n");
+                            if (session == null || !session.contents_map.has_key(content_name)) return;
+                            var encryption = new OmemoContentEncryption() { encryption_ns=NS_URI, encryption_name="OMEMO", our_key=new uint8[0], peer_key=new uint8[0], peer_device_id=device_id_by_jingle_sid[jingle_sid] };
+                            session.contents_map[content_name].encryptions[NS_URI] = encryption;
+
+                            if (iq.stanza.get_deep_attribute(Xep.Jingle.NS_URI + ":jingle", "action") == "session-accept") {
+                                session.additional_content_add_incoming.connect(on_content_add_received);
+                            }
+                        });
+
+                        break;
+                    } catch (Error e) {
+                        debug("Decrypting message from %s/%d failed: %s", iq.from.bare_jid.to_string(), parsed_data.sid, e.message);
+                    }
+                }
+            }
+        }
+
+        private void on_preprocess_outgoing_iq_set_get(XmppStream stream, Xmpp.Iq.Stanza iq) {
+            if (iq.type_ != Iq.Stanza.TYPE_SET) return;
+
+            StanzaNode? jingle_node = iq.stanza.get_subnode("jingle", Xep.Jingle.NS_URI);
+            if (jingle_node == null) return;
+
+            string? sid = jingle_node.get_attribute("sid", Xep.Jingle.NS_URI);
+            if (sid == null || !device_id_by_jingle_sid.has_key(sid)) return;
+
+            Gee.List<StanzaNode> content_nodes = jingle_node.get_subnodes("content", Xep.Jingle.NS_URI);
+            if (content_nodes.size == 0) return;
+
+            foreach (StanzaNode content_node in content_nodes) {
+                StanzaNode? transport_node = content_node.get_subnode("transport", Xep.JingleIceUdp.NS_URI);
+                if (transport_node == null) continue;
+                StanzaNode? fingerprint_node = transport_node.get_subnode("fingerprint", Xep.JingleIceUdp.DTLS_NS_URI);
+                if (fingerprint_node == null) continue;
+                string fingerprint = fingerprint_node.get_deep_string_content();
+
+                Xep.Omemo.OmemoEncryptor encryptor = stream.get_module(Xep.Omemo.OmemoEncryptor.IDENTITY);
+                Xep.Omemo.EncryptionData enc_data = encryptor.encrypt_plaintext(fingerprint);
+                encryptor.encrypt_key(enc_data, iq.to.bare_jid, device_id_by_jingle_sid[sid]);
+
+                StanzaNode new_fingerprint_node = new StanzaNode.build("fingerprint", NS_URI).add_self_xmlns().put_node(enc_data.get_encrypted_node());
+                string? hash_attr = fingerprint_node.get_attribute("hash", Xep.JingleIceUdp.DTLS_NS_URI);
+                string? setup_attr = fingerprint_node.get_attribute("setup", Xep.JingleIceUdp.DTLS_NS_URI);
+                if (hash_attr != null) new_fingerprint_node.put_attribute("hash", hash_attr);
+                if (setup_attr != null) new_fingerprint_node.put_attribute("setup", setup_attr);
+                transport_node.put_node(new_fingerprint_node);
+
+                transport_node.sub_nodes.remove(fingerprint_node);
+            }
+        }
+
+        private void on_message_received(XmppStream stream, Xmpp.MessageStanza message) {
+            StanzaNode? proceed_node = message.stanza.get_subnode("proceed", Xep.JingleMessageInitiation.NS_URI);
+            if (proceed_node == null) return;
+
+            string? jingle_sid = proceed_node.get_attribute("id");
+            if (jingle_sid == null) return;
+
+            StanzaNode? device_node = proceed_node.get_subnode("device", NS_URI);
+            if (device_node == null) return;
+
+            int device_id = device_node.get_attribute_int("id", -1);
+            if (device_id == -1) return;
+
+            device_id_by_jingle_sid[jingle_sid] = device_id;
+        }
+
+        private void on_session_initiate_received(XmppStream stream, Xep.Jingle.Session session) {
+            if (device_id_by_jingle_sid.has_key(session.sid)) {
+                foreach (Xep.Jingle.Content content in session.contents) {
+                    on_content_add_received(stream, content);
+                }
+            }
+            session.additional_content_add_incoming.connect(on_content_add_received);
+        }
+
+        private void on_content_add_received(XmppStream stream, Xep.Jingle.Content content) {
+            if (!content_names_by_jingle_sid.has_key(content.session.sid) || content_names_by_jingle_sid[content.session.sid].contains(content.content_name)) {
+                var encryption = new OmemoContentEncryption() { encryption_ns=NS_URI, encryption_name="OMEMO", our_key=new uint8[0], peer_key=new uint8[0], peer_device_id=device_id_by_jingle_sid[content.session.sid] };
+                content.encryptions[encryption.encryption_ns] = encryption;
+            }
+        }
+
+        public override void attach(XmppStream stream) {
+            stream.get_module(Xmpp.MessageModule.IDENTITY).received_message.connect(on_message_received);
+            stream.get_module(Xmpp.MessageModule.IDENTITY).send_pipeline.connect(send_listener);
+            stream.get_module(Xmpp.Iq.Module.IDENTITY).preprocess_incoming_iq_set_get.connect(on_preprocess_incoming_iq_set_get);
+            stream.get_module(Xmpp.Iq.Module.IDENTITY).preprocess_outgoing_iq_set_get.connect(on_preprocess_outgoing_iq_set_get);
+            stream.get_module(Xep.Jingle.Module.IDENTITY).session_initiate_received.connect(on_session_initiate_received);
+        }
+
+        public override void detach(XmppStream stream) {
+            stream.get_module(Xmpp.MessageModule.IDENTITY).received_message.disconnect(on_message_received);
+            stream.get_module(Xmpp.MessageModule.IDENTITY).send_pipeline.disconnect(send_listener);
+            stream.get_module(Xmpp.Iq.Module.IDENTITY).preprocess_incoming_iq_set_get.disconnect(on_preprocess_incoming_iq_set_get);
+            stream.get_module(Xmpp.Iq.Module.IDENTITY).preprocess_outgoing_iq_set_get.disconnect(on_preprocess_outgoing_iq_set_get);
+            stream.get_module(Xep.Jingle.Module.IDENTITY).session_initiate_received.disconnect(on_session_initiate_received);
+        }
+
+        public override string get_ns() { return NS_URI; }
+
+        public override string get_id() { return IDENTITY.id; }
+    }
+
+    public class VerificationSendListener : StanzaListener<MessageStanza> {
+
+        private const string[] after_actions_const = {};
+
+        public override string action_group { get { return "REWRITE_NODES"; } }
+        public override string[] after_actions { get { return after_actions_const; } }
+
+        public override async bool run(XmppStream stream, MessageStanza message) {
+            StanzaNode? proceed_node = message.stanza.get_subnode("proceed", Xep.JingleMessageInitiation.NS_URI);
+            if (proceed_node == null) return false;
+
+            StanzaNode device_node = new StanzaNode.build("device", NS_URI).add_self_xmlns()
+                    .put_attribute("id", stream.get_module(Omemo.StreamModule.IDENTITY).store.local_registration_id.to_string());
+            proceed_node.put_node(device_node);
+            return false;
+        }
+    }
+
+    public class OmemoContentEncryption : Xep.Jingle.ContentEncryption {
+        public int peer_device_id { get; set; }
+    }
+}
+
diff --git a/plugins/omemo/src/jingle/jet_omemo.vala b/plugins/omemo/src/jingle/jet_omemo.vala
index 14307be2..afcdfcd6 100644
--- a/plugins/omemo/src/jingle/jet_omemo.vala
+++ b/plugins/omemo/src/jingle/jet_omemo.vala
@@ -7,18 +7,15 @@ using Xmpp;
 using Xmpp.Xep;
 
 namespace Dino.Plugins.JetOmemo {
+
 private const string NS_URI = "urn:xmpp:jingle:jet-omemo:0";
 private const string AES_128_GCM_URI = "urn:xmpp:ciphers:aes-128-gcm-nopadding";
+
 public class Module : XmppStreamModule, Jet.EnvelopEncoding {
     public static Xmpp.ModuleIdentity<Module> IDENTITY = new Xmpp.ModuleIdentity<Module>(NS_URI, "0396_jet_omemo");
-    private Omemo.Plugin plugin;
     const uint KEY_SIZE = 16;
     const uint IV_SIZE = 12;
 
-    public Module(Omemo.Plugin plugin) {
-        this.plugin = plugin;
-    }
-
     public override void attach(XmppStream stream) {
         if (stream.get_module(Jet.Module.IDENTITY) != null) {
             stream.get_module(ServiceDiscovery.Module.IDENTITY).add_feature(stream, NS_URI);
@@ -44,71 +41,38 @@ public class Module : XmppStreamModule, Jet.EnvelopEncoding {
     }
 
     public Jet.TransportSecret decode_envolop(XmppStream stream, Jid local_full_jid, Jid peer_full_jid, StanzaNode security) throws Jingle.IqError {
-        Store store = stream.get_module(Omemo.StreamModule.IDENTITY).store;
         StanzaNode? encrypted = security.get_subnode("encrypted", Omemo.NS_URI);
         if (encrypted == null) throw new Jingle.IqError.BAD_REQUEST("Invalid JET-OMEMO envelop: missing encrypted element");
-        StanzaNode? header = encrypted.get_subnode("header", Omemo.NS_URI);
-        if (header == null) throw new Jingle.IqError.BAD_REQUEST("Invalid JET-OMEMO envelop: missing header element");
-        string? iv_node = header.get_deep_string_content("iv");
-        if (header == null) throw new Jingle.IqError.BAD_REQUEST("Invalid JET-OMEMO envelop: missing iv element");
-        uint8[] iv = Base64.decode((!)iv_node);
-        foreach (StanzaNode key_node in header.get_subnodes("key")) {
-            if (key_node.get_attribute_int("rid") == store.local_registration_id) {
-                string? key_node_content = key_node.get_string_content();
-
-                uint8[] key;
-                Address address = new Address(peer_full_jid.bare_jid.to_string(), header.get_attribute_int("sid"));
-                if (key_node.get_attribute_bool("prekey")) {
-                    PreKeySignalMessage msg = Omemo.Plugin.get_context().deserialize_pre_key_signal_message(Base64.decode((!)key_node_content));
-                    SessionCipher cipher = store.create_session_cipher(address);
-                    key = cipher.decrypt_pre_key_signal_message(msg);
-                } else {
-                    SignalMessage msg = Omemo.Plugin.get_context().deserialize_signal_message(Base64.decode((!)key_node_content));
-                    SessionCipher cipher = store.create_session_cipher(address);
-                    key = cipher.decrypt_signal_message(msg);
-                }
-                address.device_id = 0; // TODO: Hack to have address obj live longer
-
-                uint8[] authtag = null;
-                if (key.length >= 32) {
-                    int authtaglength = key.length - 16;
-                    authtag = new uint8[authtaglength];
-                    uint8[] new_key = new uint8[16];
-                    Memory.copy(authtag, (uint8*)key + 16, 16);
-                    Memory.copy(new_key, key, 16);
-                    key = new_key;
-                }
-                // TODO: authtag?
-                return new Jet.TransportSecret(key, iv);
+
+        Xep.Omemo.OmemoDecryptor decryptor = stream.get_module(Xep.Omemo.OmemoDecryptor.IDENTITY);
+
+        Xmpp.Xep.Omemo.ParsedData? data = decryptor.parse_node(encrypted);
+        if (data == null)  throw new Jingle.IqError.BAD_REQUEST("Invalid JET-OMEMO envelop: bad encrypted element");
+
+        foreach (Bytes encr_key in data.our_potential_encrypted_keys.keys) {
+            data.is_prekey = data.our_potential_encrypted_keys[encr_key];
+            data.encrypted_key = encr_key.get_data();
+
+            try {
+                uint8[] key = decryptor.decrypt_key(data, peer_full_jid.bare_jid);
+                return new Jet.TransportSecret(key, data.iv);
+            } catch (GLib.Error e) {
+                debug("Decrypting JET key from %s/%d failed: %s", peer_full_jid.bare_jid.to_string(), data.sid, e.message);
             }
         }
         throw new Jingle.IqError.NOT_ACCEPTABLE("Not encrypted for targeted device");
     }
 
     public void encode_envelop(XmppStream stream, Jid local_full_jid, Jid peer_full_jid, Jet.SecurityParameters security_params, StanzaNode security) {
-        ArrayList<Account> accounts = plugin.app.stream_interactor.get_accounts();
         Store store = stream.get_module(Omemo.StreamModule.IDENTITY).store;
-        Account? account = null;
-        foreach (Account compare in accounts) {
-            if (compare.bare_jid.equals_bare(local_full_jid)) {
-                account = compare;
-                break;
-            }
-        }
-        if (account == null) {
-            // TODO
-            critical("Sending from offline account %s", local_full_jid.to_string());
-        }
 
-        StanzaNode header_node;
-        StanzaNode encrypted_node = new StanzaNode.build("encrypted", Omemo.NS_URI).add_self_xmlns()
-                .put_node(header_node = new StanzaNode.build("header", Omemo.NS_URI)
-                    .put_attribute("sid", store.local_registration_id.to_string())
-                    .put_node(new StanzaNode.build("iv", Omemo.NS_URI)
-                        .put_node(new StanzaNode.text(Base64.encode(security_params.secret.initialization_vector)))));
+        var encryption_data = new Xep.Omemo.EncryptionData(store.local_registration_id);
+        encryption_data.iv = security_params.secret.initialization_vector;
+        encryption_data.keytag = security_params.secret.transport_key;
+        Xep.Omemo.OmemoEncryptor encryptor = stream.get_module(Xep.Omemo.OmemoEncryptor.IDENTITY);
+        encryptor.encrypt_key_to_recipient(stream, encryption_data, peer_full_jid.bare_jid);
 
-        plugin.trust_manager.encrypt_key(header_node, security_params.secret.transport_key, local_full_jid.bare_jid, new ArrayList<Jid>.wrap(new Jid[] {peer_full_jid.bare_jid}), stream, account);
-        security.put_node(encrypted_node);
+        security.put_node(encryption_data.get_encrypted_node());
     }
 
     public override string get_ns() { return NS_URI; }
diff --git a/plugins/omemo/src/logic/decrypt.vala b/plugins/omemo/src/logic/decrypt.vala
new file mode 100644
index 00000000..3cdacbf7
--- /dev/null
+++ b/plugins/omemo/src/logic/decrypt.vala
@@ -0,0 +1,210 @@
+using Dino.Entities;
+using Qlite;
+using Gee;
+using Signal;
+using Xmpp;
+
+namespace Dino.Plugins.Omemo {
+
+    public class OmemoDecryptor : Xep.Omemo.OmemoDecryptor {
+
+        private Account account;
+        private Store store;
+        private Database db;
+        private StreamInteractor stream_interactor;
+        private TrustManager trust_manager;
+
+        public override uint32 own_device_id { get { return store.local_registration_id; }}
+
+        public OmemoDecryptor(Account account, StreamInteractor stream_interactor, TrustManager trust_manager, Database db, Store store) {
+            this.account = account;
+            this.stream_interactor = stream_interactor;
+            this.trust_manager = trust_manager;
+            this.db = db;
+            this.store = store;
+        }
+
+        public bool decrypt_message(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+            StanzaNode? encrypted_node = stanza.stanza.get_subnode("encrypted", NS_URI);
+            if (encrypted_node == null || MessageFlag.get_flag(stanza) != null || stanza.from == null) return false;
+
+            if (message.body == null && Xep.ExplicitEncryption.get_encryption_tag(stanza) == NS_URI) {
+                message.body = "[This message is OMEMO encrypted]"; // TODO temporary
+            }
+            if (!Plugin.ensure_context()) return false;
+            int identity_id = db.identity.get_id(conversation.account.id);
+
+            MessageFlag flag = new MessageFlag();
+            stanza.add_flag(flag);
+
+            Xep.Omemo.ParsedData? data = parse_node(encrypted_node);
+            if (data == null || data.ciphertext == null) return false;
+
+
+            foreach (Bytes encr_key in data.our_potential_encrypted_keys.keys) {
+                data.is_prekey = data.our_potential_encrypted_keys[encr_key];
+                data.encrypted_key = encr_key.get_data();
+                Gee.List<Jid> possible_jids = get_potential_message_jids(message, data, identity_id);
+                if (possible_jids.size == 0) {
+                    debug("Received message from unknown entity with device id %d", data.sid);
+                }
+
+                foreach (Jid possible_jid in possible_jids) {
+                    try {
+                        uint8[] key = decrypt_key(data, possible_jid);
+                        string cleartext = arr_to_str(aes_decrypt(Cipher.AES_GCM_NOPADDING, key, data.iv, data.ciphertext));
+
+                        // If we figured out which real jid a message comes from due to decryption working, save it
+                        if (conversation.type_ == Conversation.Type.GROUPCHAT && message.real_jid == null) {
+                            message.real_jid = possible_jid;
+                        }
+
+                        trust_manager.message_device_id_map[message] = data.sid;
+                        message.body = cleartext;
+                        message.encryption = Encryption.OMEMO;
+                        return true;
+                    } catch (Error e) {
+                        debug("Decrypting message from %s/%d failed: %s", possible_jid.to_string(), data.sid, e.message);
+                    }
+                }
+            }
+
+            if (
+                encrypted_node.get_deep_string_content("payload") != null && // Ratchet forwarding doesn't contain payload and might not include us, which is ok
+                data.our_potential_encrypted_keys.size == 0 && // The message was not encrypted to us
+                stream_interactor.module_manager.get_module(message.account, StreamModule.IDENTITY).store.local_registration_id != data.sid // Message from this device. Never encrypted to itself.
+            ) {
+                db.identity_meta.update_last_message_undecryptable(identity_id, data.sid, message.time);
+                trust_manager.bad_message_state_updated(conversation.account, message.from, data.sid);
+            }
+
+            debug("Received OMEMO encryped message that could not be decrypted.");
+            return false;
+        }
+
+        public Gee.List<Jid> get_potential_message_jids(Entities.Message message, Xmpp.Xep.Omemo.ParsedData data, int identity_id) {
+            Gee.List<Jid> possible_jids = new ArrayList<Jid>();
+            if (message.type_ == Message.Type.CHAT) {
+                possible_jids.add(message.from.bare_jid);
+            } else {
+                if (message.real_jid != null) {
+                    possible_jids.add(message.real_jid.bare_jid);
+                } else if (data.is_prekey) {
+                    // pre key messages do store the identity key, so we can use that to find the real jid
+                    PreKeySignalMessage msg = Plugin.get_context().deserialize_pre_key_signal_message(data.encrypted_key);
+                    string identity_key = Base64.encode(msg.identity_key.serialize());
+                    foreach (Row row in db.identity_meta.get_with_device_id(identity_id, data.sid).with(db.identity_meta.identity_key_public_base64, "=", identity_key)) {
+                        try {
+                            possible_jids.add(new Jid(row[db.identity_meta.address_name]));
+                        } catch (InvalidJidError e) {
+                            warning("Ignoring invalid jid from database: %s", e.message);
+                        }
+                    }
+                } else {
+                    // If we don't know the device name (MUC history w/o MAM), test decryption with all keys with fitting device id
+                    foreach (Row row in db.identity_meta.get_with_device_id(identity_id, data.sid)) {
+                        try {
+                            possible_jids.add(new Jid(row[db.identity_meta.address_name]));
+                        } catch (InvalidJidError e) {
+                            warning("Ignoring invalid jid from database: %s", e.message);
+                        }
+                    }
+                }
+            }
+            return possible_jids;
+        }
+
+        public override uint8[] decrypt_key(Xmpp.Xep.Omemo.ParsedData data, Jid from_jid) throws GLib.Error {
+            int sid = data.sid;
+            uint8[] ciphertext = data.ciphertext;
+            uint8[] encrypted_key = data.encrypted_key;
+
+            Address address = new Address(from_jid.to_string(), sid);
+            uint8[] key;
+
+            if (data.is_prekey) {
+                int identity_id = db.identity.get_id(account.id);
+                PreKeySignalMessage msg = Plugin.get_context().deserialize_pre_key_signal_message(encrypted_key);
+                string identity_key = Base64.encode(msg.identity_key.serialize());
+
+                bool ok = update_db_for_prekey(identity_id, identity_key, from_jid, sid);
+                if (!ok) return null;
+
+                debug("Starting new session for decryption with device from %s/%d", from_jid.to_string(), sid);
+                SessionCipher cipher = store.create_session_cipher(address);
+                key = cipher.decrypt_pre_key_signal_message(msg);
+                // TODO: Finish session
+                } else {
+                debug("Continuing session for decryption with device from %s/%d", from_jid.to_string(), sid);
+                SignalMessage msg = Plugin.get_context().deserialize_signal_message(encrypted_key);
+                SessionCipher cipher = store.create_session_cipher(address);
+                key = cipher.decrypt_signal_message(msg);
+            }
+
+            if (key.length >= 32) {
+                int authtaglength = key.length - 16;
+                uint8[] new_ciphertext = new uint8[ciphertext.length + authtaglength];
+                uint8[] new_key = new uint8[16];
+                Memory.copy(new_ciphertext, ciphertext, ciphertext.length);
+                Memory.copy((uint8*)new_ciphertext + ciphertext.length, (uint8*)key + 16, authtaglength);
+                Memory.copy(new_key, key, 16);
+                data.ciphertext = new_ciphertext;
+                key = new_key;
+            }
+
+            return key;
+        }
+
+        public override string decrypt(uint8[] ciphertext, uint8[] key, uint8[] iv) throws GLib.Error {
+            return arr_to_str(aes_decrypt(Cipher.AES_GCM_NOPADDING, key, iv, ciphertext));
+        }
+
+        private bool update_db_for_prekey(int identity_id, string identity_key, Jid from_jid, int sid) {
+            Row? device = db.identity_meta.get_device(identity_id, from_jid.to_string(), sid);
+            if (device != null && device[db.identity_meta.identity_key_public_base64] != null) {
+                if (device[db.identity_meta.identity_key_public_base64] != identity_key) {
+                    critical("Tried to use a different identity key for a known device id.");
+                    return false;
+                }
+            } else {
+                debug("Learn new device from incoming message from %s/%d", from_jid.to_string(), sid);
+                bool blind_trust = db.trust.get_blind_trust(identity_id, from_jid.to_string(), true);
+                if (db.identity_meta.insert_device_session(identity_id, from_jid.to_string(), sid, identity_key, blind_trust ? TrustLevel.TRUSTED : TrustLevel.UNKNOWN) < 0) {
+                    critical("Failed learning a device.");
+                    return false;
+                }
+
+                XmppStream? stream = stream_interactor.get_stream(account);
+                if (device == null && stream != null) {
+                    stream.get_module(StreamModule.IDENTITY).request_user_devicelist.begin(stream, from_jid);
+                }
+            }
+            return true;
+        }
+
+        private string arr_to_str(uint8[] arr) {
+            // null-terminate the array
+            uint8[] rarr = new uint8[arr.length+1];
+            Memory.copy(rarr, arr, arr.length);
+            return (string)rarr;
+        }
+    }
+
+    public class DecryptMessageListener : MessageListener {
+        public string[] after_actions_const = new string[]{ };
+        public override string action_group { get { return "DECRYPT"; } }
+        public override string[] after_actions { get { return after_actions_const; } }
+
+        private HashMap<Account, OmemoDecryptor> decryptors;
+
+        public DecryptMessageListener(HashMap<Account, OmemoDecryptor> decryptors) {
+            this.decryptors = decryptors;
+        }
+
+        public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
+            decryptors[message.account].decrypt_message(message, stanza, conversation);
+            return false;
+        }
+    }
+}
+
diff --git a/plugins/omemo/src/logic/encrypt.vala b/plugins/omemo/src/logic/encrypt.vala
new file mode 100644
index 00000000..cd994c3a
--- /dev/null
+++ b/plugins/omemo/src/logic/encrypt.vala
@@ -0,0 +1,131 @@
+using Gee;
+using Signal;
+using Dino.Entities;
+using Xmpp;
+using Xmpp.Xep.Omemo;
+
+namespace Dino.Plugins.Omemo {
+
+    public class OmemoEncryptor : Xep.Omemo.OmemoEncryptor {
+
+        private Account account;
+        private Store store;
+        private TrustManager trust_manager;
+
+        public override uint32 own_device_id { get { return store.local_registration_id; }}
+
+        public OmemoEncryptor(Account account, TrustManager trust_manager, Store store) {
+            this.account = account;
+            this.trust_manager = trust_manager;
+            this.store = store;
+        }
+
+        public override Xep.Omemo.EncryptionData encrypt_plaintext(string plaintext) throws GLib.Error {
+            const uint KEY_SIZE = 16;
+            const uint IV_SIZE = 12;
+
+            //Create a key and use it to encrypt the message
+            uint8[] key = new uint8[KEY_SIZE];
+            Plugin.get_context().randomize(key);
+            uint8[] iv = new uint8[IV_SIZE];
+            Plugin.get_context().randomize(iv);
+
+            uint8[] aes_encrypt_result = aes_encrypt(Cipher.AES_GCM_NOPADDING, key, iv, plaintext.data);
+            uint8[] ciphertext = aes_encrypt_result[0:aes_encrypt_result.length - 16];
+            uint8[] tag = aes_encrypt_result[aes_encrypt_result.length - 16:aes_encrypt_result.length];
+            uint8[] keytag = new uint8[key.length + tag.length];
+            Memory.copy(keytag, key, key.length);
+            Memory.copy((uint8*)keytag + key.length, tag, tag.length);
+
+            var ret = new Xep.Omemo.EncryptionData(own_device_id);
+            ret.ciphertext = ciphertext;
+            ret.keytag = keytag;
+            ret.iv = iv;
+            return ret;
+        }
+
+        public EncryptState encrypt(MessageStanza message, Jid self_jid, Gee.List<Jid> recipients, XmppStream stream) {
+
+            EncryptState status = new EncryptState();
+            if (!Plugin.ensure_context()) return status;
+            if (message.to == null) return status;
+
+            try {
+                EncryptionData enc_data = encrypt_plaintext(message.body);
+                status = encrypt_key_to_recipients(enc_data, self_jid, recipients, stream);
+
+                message.stanza.put_node(enc_data.get_encrypted_node());
+                Xep.ExplicitEncryption.add_encryption_tag_to_message(message, NS_URI, "OMEMO");
+                message.body = "[This message is OMEMO encrypted]";
+                status.encrypted = true;
+            } catch (Error e) {
+                warning(@"Signal error while encrypting message: $(e.message)\n");
+                message.body = "[OMEMO encryption failed]";
+                status.encrypted = false;
+            }
+            return status;
+        }
+
+        internal EncryptState encrypt_key_to_recipients(EncryptionData enc_data, Jid self_jid, Gee.List<Jid> recipients, XmppStream stream) throws Error {
+            EncryptState status = new EncryptState();
+
+            //Check we have the bundles and device lists needed to send the message
+            if (!trust_manager.is_known_address(account, self_jid)) return status;
+            status.own_list = true;
+            status.own_devices = trust_manager.get_trusted_devices(account, self_jid).size;
+            status.other_waiting_lists = 0;
+            status.other_devices = 0;
+            foreach (Jid recipient in recipients) {
+                if (!trust_manager.is_known_address(account, recipient)) {
+                    status.other_waiting_lists++;
+                }
+                if (status.other_waiting_lists > 0) return status;
+                status.other_devices += trust_manager.get_trusted_devices(account, recipient).size;
+            }
+            if (status.own_devices == 0 || status.other_devices == 0) return status;
+
+
+            //Encrypt the key for each recipient's device individually
+            foreach (Jid recipient in recipients) {
+                EncryptionResult enc_res = encrypt_key_to_recipient(stream, enc_data, recipient);
+                status.add_result(enc_res, false);
+            }
+
+            // Encrypt the key for each own device
+            EncryptionResult enc_res = encrypt_key_to_recipient(stream, enc_data, self_jid);
+            status.add_result(enc_res, true);
+
+            return status;
+        }
+
+        public override EncryptionResult encrypt_key_to_recipient(XmppStream stream, Xep.Omemo.EncryptionData enc_data, Jid recipient) throws GLib.Error {
+            var result = new EncryptionResult();
+            StreamModule module = stream.get_module(StreamModule.IDENTITY);
+
+            foreach(int32 device_id in trust_manager.get_trusted_devices(account, recipient)) {
+                if (module.is_ignored_device(recipient, device_id)) {
+                    result.lost++;
+                    continue;
+                }
+                try {
+                    encrypt_key(enc_data, recipient, device_id);
+                    result.success++;
+                } catch (Error e) {
+                    if (e.code == ErrorCode.UNKNOWN) result.unknown++;
+                    else result.failure++;
+                }
+            }
+            return result;
+        }
+
+        public override void encrypt_key(Xep.Omemo.EncryptionData encryption_data, Jid jid, int32 device_id) throws GLib.Error {
+            Address address = new Address(jid.to_string(), device_id);
+            SessionCipher cipher = store.create_session_cipher(address);
+            CiphertextMessage device_key = cipher.encrypt(encryption_data.keytag);
+            address.device_id = 0;
+            debug("Created encrypted key for %s/%d", jid.to_string(), device_id);
+
+            encryption_data.add_device_key(device_id, device_key.serialized, device_key.type == CiphertextType.PREKEY);
+        }
+    }
+}
\ No newline at end of file
diff --git a/plugins/omemo/src/logic/encrypt_state.vala b/plugins/omemo/src/logic/encrypt_state.vala
deleted file mode 100644
index fd72faf4..00000000
--- a/plugins/omemo/src/logic/encrypt_state.vala
+++ /dev/null
@@ -1,24 +0,0 @@
-namespace Dino.Plugins.Omemo {
-
-public class EncryptState {
-    public bool encrypted { get; internal set; }
-    public int other_devices { get; internal set; }
-    public int other_success { get; internal set; }
-    public int other_lost { get; internal set; }
-    public int other_unknown { get; internal set; }
-    public int other_failure { get; internal set; }
-    public int other_waiting_lists { get; internal set; }
-
-    public int own_devices { get; internal set; }
-    public int own_success { get; internal set; }
-    public int own_lost { get; internal set; }
-    public int own_unknown { get; internal set; }
-    public int own_failure { get; internal set; }
-    public bool own_list { get; internal set; }
-
-    public string to_string() {
-        return @"EncryptState (encrypted=$encrypted, other=(devices=$other_devices, success=$other_success, lost=$other_lost, unknown=$other_unknown, failure=$other_failure, waiting_lists=$other_waiting_lists, own=(devices=$own_devices, success=$own_success, lost=$own_lost, unknown=$own_unknown, failure=$own_failure, list=$own_list))";
-    }
-}
-
-}
diff --git a/plugins/omemo/src/logic/manager.vala b/plugins/omemo/src/logic/manager.vala
index 64b117c7..5552e212 100644
--- a/plugins/omemo/src/logic/manager.vala
+++ b/plugins/omemo/src/logic/manager.vala
@@ -13,11 +13,12 @@ public class Manager : StreamInteractionModule, Object {
     private StreamInteractor stream_interactor;
     private Database db;
     private TrustManager trust_manager;
+    private HashMap<Account, OmemoEncryptor> encryptors;
     private Map<Entities.Message, MessageState> message_states = new HashMap<Entities.Message, MessageState>(Entities.Message.hash_func, Entities.Message.equals_func);
 
     private class MessageState {
         public Entities.Message msg { get; private set; }
-        public EncryptState last_try { get; private set; }
+        public Xep.Omemo.EncryptState last_try { get; private set; }
         public int waiting_other_sessions { get; set; }
         public int waiting_own_sessions { get; set; }
         public bool waiting_own_devicelist { get; set; }
@@ -26,11 +27,11 @@ public class Manager : StreamInteractionModule, Object {
         public bool will_send_now { get; private set; }
         public bool active_send_attempt { get; set; }
 
-        public MessageState(Entities.Message msg, EncryptState last_try) {
+        public MessageState(Entities.Message msg, Xep.Omemo.EncryptState last_try) {
             update_from_encrypt_status(msg, last_try);
         }
 
-        public void update_from_encrypt_status(Entities.Message msg, EncryptState new_try) {
+        public void update_from_encrypt_status(Entities.Message msg, Xep.Omemo.EncryptState new_try) {
             this.msg = msg;
             this.last_try = new_try;
             this.waiting_other_sessions = new_try.other_unknown;
@@ -59,10 +60,11 @@ public class Manager : StreamInteractionModule, Object {
         }
     }
 
-    private Manager(StreamInteractor stream_interactor, Database db, TrustManager trust_manager) {
+    private Manager(StreamInteractor stream_interactor, Database db, TrustManager trust_manager, HashMap<Account, OmemoEncryptor> encryptors) {
         this.stream_interactor = stream_interactor;
         this.db = db;
         this.trust_manager = trust_manager;
+        this.encryptors = encryptors;
 
         stream_interactor.stream_negotiated.connect(on_stream_negotiated);
         stream_interactor.get_module(MessageProcessor.IDENTITY).pre_message_send.connect(on_pre_message_send);
@@ -125,7 +127,7 @@ public class Manager : StreamInteractionModule, Object {
             }
 
             //Attempt to encrypt the message
-            EncryptState enc_state = trust_manager.encrypt(message_stanza, conversation.account.bare_jid, recipients, stream, conversation.account);
+            Xep.Omemo.EncryptState enc_state = encryptors[conversation.account].encrypt(message_stanza, conversation.account.bare_jid, recipients, stream);
             MessageState state;
             lock (message_states) {
                 if (message_states.has_key(message)) {
@@ -411,8 +413,8 @@ public class Manager : StreamInteractionModule, Object {
         return true; // TODO wait for stream?
     }
 
-    public static void start(StreamInteractor stream_interactor, Database db, TrustManager trust_manager) {
-        Manager m = new Manager(stream_interactor, db, trust_manager);
+    public static void start(StreamInteractor stream_interactor, Database db, TrustManager trust_manager, HashMap<Account, OmemoEncryptor> encryptors) {
+        Manager m = new Manager(stream_interactor, db, trust_manager, encryptors);
         stream_interactor.add_module(m);
     }
 }
diff --git a/plugins/omemo/src/logic/trust_manager.vala b/plugins/omemo/src/logic/trust_manager.vala
index 1e61b201..20076a43 100644
--- a/plugins/omemo/src/logic/trust_manager.vala
+++ b/plugins/omemo/src/logic/trust_manager.vala
@@ -12,18 +12,15 @@ public class TrustManager {
 
     private StreamInteractor stream_interactor;
     private Database db;
-    private DecryptMessageListener decrypt_message_listener;
     private TagMessageListener tag_message_listener;
 
-    private HashMap<Message, int> message_device_id_map = new HashMap<Message, int>(Message.hash_func, Message.equals_func);
+    public HashMap<Message, int> message_device_id_map = new HashMap<Message, int>(Message.hash_func, Message.equals_func);
 
     public TrustManager(StreamInteractor stream_interactor, Database db) {
         this.stream_interactor = stream_interactor;
         this.db = db;
 
-        decrypt_message_listener = new DecryptMessageListener(stream_interactor, this, db, message_device_id_map);
         tag_message_listener = new TagMessageListener(stream_interactor, this, db, message_device_id_map);
-        stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(decrypt_message_listener);
         stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(tag_message_listener);
     }
 
@@ -69,127 +66,6 @@ public class TrustManager {
         }
     }
 
-    private StanzaNode create_encrypted_key_node(uint8[] key, Address address, Store store) throws GLib.Error {
-        SessionCipher cipher = store.create_session_cipher(address);
-        CiphertextMessage device_key = cipher.encrypt(key);
-        debug("Created encrypted key for %s/%d", address.name, address.device_id);
-        StanzaNode key_node = new StanzaNode.build("key", NS_URI)
-            .put_attribute("rid", address.device_id.to_string())
-            .put_node(new StanzaNode.text(Base64.encode(device_key.serialized)));
-        if (device_key.type == CiphertextType.PREKEY) key_node.put_attribute("prekey", "true");
-        return key_node;
-    }
-
-    internal EncryptState encrypt_key(StanzaNode header_node, uint8[] keytag, Jid self_jid, Gee.List<Jid> recipients, XmppStream stream, Account account) throws Error {
-        EncryptState status = new EncryptState();
-        StreamModule module = stream.get_module(StreamModule.IDENTITY);
-
-        //Check we have the bundles and device lists needed to send the message
-        if (!is_known_address(account, self_jid)) return status;
-        status.own_list = true;
-        status.own_devices = get_trusted_devices(account, self_jid).size;
-        status.other_waiting_lists = 0;
-        status.other_devices = 0;
-        foreach (Jid recipient in recipients) {
-            if (!is_known_address(account, recipient)) {
-                status.other_waiting_lists++;
-            }
-            if (status.other_waiting_lists > 0) return status;
-            status.other_devices += get_trusted_devices(account, recipient).size;
-        }
-        if (status.own_devices == 0 || status.other_devices == 0) return status;
-
-
-        //Encrypt the key for each recipient's device individually
-        Address address = new Address("", 0);
-        foreach (Jid recipient in recipients) {
-            foreach(int32 device_id in get_trusted_devices(account, recipient)) {
-                if (module.is_ignored_device(recipient, device_id)) {
-                    status.other_lost++;
-                    continue;
-                }
-                try {
-                    address.name = recipient.bare_jid.to_string();
-                    address.device_id = (int) device_id;
-                    StanzaNode key_node = create_encrypted_key_node(keytag, address, module.store);
-                    header_node.put_node(key_node);
-                    status.other_success++;
-                } catch (Error e) {
-                    if (e.code == ErrorCode.UNKNOWN) status.other_unknown++;
-                    else status.other_failure++;
-                }
-            }
-        }
-
-        // Encrypt the key for each own device
-        address.name = self_jid.bare_jid.to_string();
-        foreach(int32 device_id in get_trusted_devices(account, self_jid)) {
-            if (module.is_ignored_device(self_jid, device_id)) {
-                status.own_lost++;
-                continue;
-            }
-            if (device_id != module.store.local_registration_id) {
-                address.device_id = (int) device_id;
-                try {
-                    StanzaNode key_node = create_encrypted_key_node(keytag, address, module.store);
-                    header_node.put_node(key_node);
-                    status.own_success++;
-                } catch (Error e) {
-                    if (e.code == ErrorCode.UNKNOWN) status.own_unknown++;
-                    else status.own_failure++;
-                }
-            }
-        }
-
-        return status;
-    }
-
-    public EncryptState encrypt(MessageStanza message, Jid self_jid, Gee.List<Jid> recipients, XmppStream stream, Account account) {
-        const uint KEY_SIZE = 16;
-        const uint IV_SIZE = 12;
-        EncryptState status = new EncryptState();
-        if (!Plugin.ensure_context()) return status;
-        if (message.to == null) return status;
-
-        StreamModule module = stream.get_module(StreamModule.IDENTITY);
-
-        try {
-            //Create a key and use it to encrypt the message
-            uint8[] key = new uint8[KEY_SIZE];
-            Plugin.get_context().randomize(key);
-            uint8[] iv = new uint8[IV_SIZE];
-            Plugin.get_context().randomize(iv);
-
-            uint8[] aes_encrypt_result = aes_encrypt(Cipher.AES_GCM_NOPADDING, key, iv, message.body.data);
-            uint8[] ciphertext = aes_encrypt_result[0:aes_encrypt_result.length-16];
-            uint8[] tag = aes_encrypt_result[aes_encrypt_result.length-16:aes_encrypt_result.length];
-            uint8[] keytag = new uint8[key.length + tag.length];
-            Memory.copy(keytag, key, key.length);
-            Memory.copy((uint8*)keytag + key.length, tag, tag.length);
-
-            StanzaNode header_node;
-            StanzaNode encrypted_node = new StanzaNode.build("encrypted", NS_URI).add_self_xmlns()
-                    .put_node(header_node = new StanzaNode.build("header", NS_URI)
-                        .put_attribute("sid", module.store.local_registration_id.to_string())
-                        .put_node(new StanzaNode.build("iv", NS_URI)
-                            .put_node(new StanzaNode.text(Base64.encode(iv)))))
-                    .put_node(new StanzaNode.build("payload", NS_URI)
-                        .put_node(new StanzaNode.text(Base64.encode(ciphertext))));
-
-            status = encrypt_key(header_node, keytag, self_jid, recipients, stream, account);
-
-            message.stanza.put_node(encrypted_node);
-            Xep.ExplicitEncryption.add_encryption_tag_to_message(message, NS_URI, "OMEMO");
-            message.body = "[This message is OMEMO encrypted]";
-            status.encrypted = true;
-        } catch (Error e) {
-            warning(@"Signal error while encrypting message: $(e.message)\n");
-            message.body = "[OMEMO encryption failed]";
-            status.encrypted = false;
-        }
-        return status;
-    }
-
     public bool is_known_address(Account account, Jid jid) {
         int identity_id = db.identity.get_id(account.id);
         if (identity_id < 0) return false;
@@ -260,182 +136,6 @@ public class TrustManager {
             return false;
         }
     }
-
-    private class DecryptMessageListener : MessageListener {
-        public string[] after_actions_const = new string[]{ };
-        public override string action_group { get { return "DECRYPT"; } }
-        public override string[] after_actions { get { return after_actions_const; } }
-
-        private StreamInteractor stream_interactor;
-        private TrustManager trust_manager;
-        private Database db;
-        private HashMap<Message, int> message_device_id_map;
-
-        public DecryptMessageListener(StreamInteractor stream_interactor, TrustManager trust_manager, Database db, HashMap<Message, int> message_device_id_map) {
-            this.stream_interactor = stream_interactor;
-            this.trust_manager = trust_manager;
-            this.db = db;
-            this.message_device_id_map = message_device_id_map;
-        }
-
-        public override async bool run(Entities.Message message, Xmpp.MessageStanza stanza, Conversation conversation) {
-            StreamModule module = stream_interactor.module_manager.get_module(conversation.account, StreamModule.IDENTITY);
-            Store store = module.store;
-
-            StanzaNode? _encrypted = stanza.stanza.get_subnode("encrypted", NS_URI);
-            if (_encrypted == null || MessageFlag.get_flag(stanza) != null || stanza.from == null) return false;
-            StanzaNode encrypted = (!)_encrypted;
-            if (message.body == null && Xep.ExplicitEncryption.get_encryption_tag(stanza) == NS_URI) {
-                message.body = "[This message is OMEMO encrypted]"; // TODO temporary
-            };
-            if (!Plugin.ensure_context()) return false;
-            int identity_id = db.identity.get_id(conversation.account.id);
-            MessageFlag flag = new MessageFlag();
-            stanza.add_flag(flag);
-            StanzaNode? _header = encrypted.get_subnode("header");
-            if (_header == null) return false;
-            StanzaNode header = (!)_header;
-            int sid = header.get_attribute_int("sid");
-            if (sid <= 0) return false;
-
-            var our_nodes = new ArrayList<StanzaNode>();
-            foreach (StanzaNode key_node in header.get_subnodes("key")) {
-                debug("Is ours? %d =? %u", key_node.get_attribute_int("rid"), store.local_registration_id);
-                if (key_node.get_attribute_int("rid") == store.local_registration_id) {
-                    our_nodes.add(key_node);
-                }
-            }
-
-            string? payload = encrypted.get_deep_string_content("payload");
-            string? iv_node = header.get_deep_string_content("iv");
-
-            foreach (StanzaNode key_node in our_nodes) {
-                string? key_node_content = key_node.get_string_content();
-                if (payload == null || iv_node == null || key_node_content == null) continue;
-                uint8[] key;
-                uint8[] ciphertext = Base64.decode((!)payload);
-                uint8[] iv = Base64.decode((!)iv_node);
-                Gee.List<Jid> possible_jids = new ArrayList<Jid>();
-                if (conversation.type_ == Conversation.Type.CHAT) {
-                    possible_jids.add(stanza.from.bare_jid);
-                } else {
-                    Jid? real_jid = message.real_jid;
-                    if (real_jid != null) {
-                        possible_jids.add(real_jid.bare_jid);
-                    } else if (key_node.get_attribute_bool("prekey")) {
-                        // pre key messages do store the identity key, so we can use that to find the real jid
-                        PreKeySignalMessage msg = Plugin.get_context().deserialize_pre_key_signal_message(Base64.decode((!)key_node_content));
-                        string identity_key = Base64.encode(msg.identity_key.serialize());
-                        foreach (Row row in db.identity_meta.get_with_device_id(identity_id, sid).with(db.identity_meta.identity_key_public_base64, "=", identity_key)) {
-                            try {
-                                possible_jids.add(new Jid(row[db.identity_meta.address_name]));
-                            } catch (InvalidJidError e) {
-                                warning("Ignoring invalid jid from database: %s", e.message);
-                            }
-                        }
-                        if (possible_jids.size != 1) {
-                            continue;
-                        }
-                    } else {
-                        // If we don't know the device name (MUC history w/o MAM), test decryption with all keys with fitting device id
-                        foreach (Row row in db.identity_meta.get_with_device_id(identity_id, sid)) {
-                            try {
-                                possible_jids.add(new Jid(row[db.identity_meta.address_name]));
-                            } catch (InvalidJidError e) {
-                                warning("Ignoring invalid jid from database: %s", e.message);
-                            }
-                        }
-                    }
-                }
-
-                if (possible_jids.size == 0) {
-                    debug("Received message from unknown entity with device id %d", sid);
-                }
-
-                foreach (Jid possible_jid in possible_jids) {
-                    try {
-                        Address address = new Address(possible_jid.to_string(), sid);
-                        if (key_node.get_attribute_bool("prekey")) {
-                            Row? device = db.identity_meta.get_device(identity_id, possible_jid.to_string(), sid);
-                            PreKeySignalMessage msg = Plugin.get_context().deserialize_pre_key_signal_message(Base64.decode((!)key_node_content));
-                            string identity_key = Base64.encode(msg.identity_key.serialize());
-                            if (device != null && device[db.identity_meta.identity_key_public_base64] != null) {
-                                if (device[db.identity_meta.identity_key_public_base64] != identity_key) {
-                                    critical("Tried to use a different identity key for a known device id.");
-                                    continue;
-                                }
-                            } else {
-                                debug("Learn new device from incoming message from %s/%d", possible_jid.to_string(), sid);
-                                bool blind_trust = db.trust.get_blind_trust(identity_id, possible_jid.to_string(), true);
-                                if (db.identity_meta.insert_device_session(identity_id, possible_jid.to_string(), sid, identity_key, blind_trust ? TrustLevel.TRUSTED : TrustLevel.UNKNOWN) < 0) {
-                                    critical("Failed learning a device.");
-                                    continue;
-                                }
-                                XmppStream? stream = stream_interactor.get_stream(conversation.account);
-                                if (device == null && stream != null) {
-                                    module.request_user_devicelist.begin(stream, possible_jid);
-                                }
-                            }
-                            debug("Starting new session for decryption with device from %s/%d", possible_jid.to_string(), sid);
-                            SessionCipher cipher = store.create_session_cipher(address);
-                            key = cipher.decrypt_pre_key_signal_message(msg);
-                            // TODO: Finish session
-                        } else {
-                            debug("Continuing session for decryption with device from %s/%d", possible_jid.to_string(), sid);
-                            SignalMessage msg = Plugin.get_context().deserialize_signal_message(Base64.decode((!)key_node_content));
-                            SessionCipher cipher = store.create_session_cipher(address);
-                            key = cipher.decrypt_signal_message(msg);
-                        }
-                        //address.device_id = 0; // TODO: Hack to have address obj live longer
-
-                        if (key.length >= 32) {
-                            int authtaglength = key.length - 16;
-                            uint8[] new_ciphertext = new uint8[ciphertext.length + authtaglength];
-                            uint8[] new_key = new uint8[16];
-                            Memory.copy(new_ciphertext, ciphertext, ciphertext.length);
-                            Memory.copy((uint8*)new_ciphertext + ciphertext.length, (uint8*)key + 16, authtaglength);
-                            Memory.copy(new_key, key, 16);
-                            ciphertext = new_ciphertext;
-                            key = new_key;
-                        }
-
-                        message.body = arr_to_str(aes_decrypt(Cipher.AES_GCM_NOPADDING, key, iv, ciphertext));
-                        message_device_id_map[message] = address.device_id;
-                        message.encryption = Encryption.OMEMO;
-                        flag.decrypted = true;
-                    } catch (Error e) {
-                        debug("Decrypting message from %s/%d failed: %s", possible_jid.to_string(), sid, e.message);
-                        continue;
-                    }
-
-                    // If we figured out which real jid a message comes from due to decryption working, save it
-                    if (conversation.type_ == Conversation.Type.GROUPCHAT && message.real_jid == null) {
-                        message.real_jid = possible_jid;
-                    }
-                    return false;
-                }
-            }
-
-            if (
-                payload != null && // Ratchet forwarding doesn't contain payload and might not include us, which is ok
-                our_nodes.size == 0 && // The message was not encrypted to us
-                module.store.local_registration_id != sid // Message from this device. Never encrypted to itself.
-            ) {
-                db.identity_meta.update_last_message_undecryptable(identity_id, sid, message.time);
-                trust_manager.bad_message_state_updated(conversation.account, message.from, sid);
-            }
-
-            debug("Received OMEMO encryped message that could not be decrypted.");
-            return false;
-        }
-
-        private string arr_to_str(uint8[] arr) {
-            // null-terminate the array
-            uint8[] rarr = new uint8[arr.length+1];
-            Memory.copy(rarr, arr, arr.length);
-            return (string)rarr;
-        }
-    }
 }
 
 }
diff --git a/plugins/omemo/src/plugin.vala b/plugins/omemo/src/plugin.vala
index e739fc4d..7a0304d1 100644
--- a/plugins/omemo/src/plugin.vala
+++ b/plugins/omemo/src/plugin.vala
@@ -1,3 +1,4 @@
+using Gee;
 using Dino.Entities;
 
 extern const string GETTEXT_PACKAGE;
@@ -20,6 +21,7 @@ public class Plugin : RootInterface, Object {
                 }
                 return true;
             } catch (Error e) {
+                warning("Error initializing Signal Context %s", e.message);
                 return false;
             }
         }
@@ -33,6 +35,9 @@ public class Plugin : RootInterface, Object {
     public DeviceNotificationPopulator device_notification_populator;
     public OwnNotifications own_notifications;
     public TrustManager trust_manager;
+    public DecryptMessageListener decrypt_message_listener;
+    public HashMap<Account, OmemoDecryptor> decryptors = new HashMap<Account, OmemoDecryptor>(Account.hash_func, Account.equals_func);
+    public HashMap<Account, OmemoEncryptor> encryptors = new HashMap<Account, OmemoEncryptor>(Account.hash_func, Account.equals_func);
 
     public void registered(Dino.Application app) {
         ensure_context();
@@ -43,22 +48,33 @@ public class Plugin : RootInterface, Object {
         this.contact_details_provider = new ContactDetailsProvider(this);
         this.device_notification_populator = new DeviceNotificationPopulator(this, this.app.stream_interactor);
         this.trust_manager = new TrustManager(this.app.stream_interactor, this.db);
+
         this.app.plugin_registry.register_encryption_list_entry(list_entry);
         this.app.plugin_registry.register_account_settings_entry(settings_entry);
         this.app.plugin_registry.register_contact_details_entry(contact_details_provider);
         this.app.plugin_registry.register_notification_populator(device_notification_populator);
         this.app.plugin_registry.register_conversation_addition_populator(new BadMessagesPopulator(this.app.stream_interactor, this));
+
         this.app.stream_interactor.module_manager.initialize_account_modules.connect((account, list) => {
-            list.add(new StreamModule());
-            list.add(new JetOmemo.Module(this));
+            Signal.Store signal_store = Plugin.get_context().create_store();
+            list.add(new StreamModule(signal_store));
+            decryptors[account] = new OmemoDecryptor(account, app.stream_interactor, trust_manager, db, signal_store);
+            list.add(decryptors[account]);
+            encryptors[account] = new OmemoEncryptor(account, trust_manager,signal_store);
+            list.add(encryptors[account]);
+            list.add(new JetOmemo.Module());
+            list.add(new DtlsSrtpVerificationDraft.StreamModule());
             this.own_notifications = new OwnNotifications(this, this.app.stream_interactor, account);
         });
 
+        decrypt_message_listener = new DecryptMessageListener(decryptors);
+        app.stream_interactor.get_module(MessageProcessor.IDENTITY).received_pipeline.connect(decrypt_message_listener);
+
         app.stream_interactor.get_module(FileManager.IDENTITY).add_file_decryptor(new OmemoFileDecryptor());
         app.stream_interactor.get_module(FileManager.IDENTITY).add_file_encryptor(new OmemoFileEncryptor());
         JingleFileHelperRegistry.instance.add_encryption_helper(Encryption.OMEMO, new JetOmemo.EncryptionHelper(app.stream_interactor));
 
-        Manager.start(this.app.stream_interactor, db, trust_manager);
+        Manager.start(this.app.stream_interactor, db, trust_manager, encryptors);
 
         SimpleAction own_keys_action = new SimpleAction("own-keys", VariantType.INT32);
         own_keys_action.activate.connect((variant) => {
diff --git a/plugins/omemo/src/protocol/stream_module.vala b/plugins/omemo/src/protocol/stream_module.vala
index e4a2733c..39d9c448 100644
--- a/plugins/omemo/src/protocol/stream_module.vala
+++ b/plugins/omemo/src/protocol/stream_module.vala
@@ -25,10 +25,8 @@ public class StreamModule : XmppStreamModule {
     public signal void bundle_fetched(Jid jid, int device_id, Bundle bundle);
     public signal void bundle_fetch_failed(Jid jid, int device_id);
 
-    public StreamModule() {
-        if (Plugin.ensure_context()) {
-            this.store = Plugin.get_context().create_store();
-        }
+    public StreamModule(Store store) {
+        this.store = store;
     }
 
     public override void attach(XmppStream stream) {