implement closed group v2 handling and configuration message handling in refactored message receiving pipeline

app/src/main/java/org/thoughtcrime/securesms/database/Storage.kt

@@ -376,6 +376,22 @@ class Storage(context: Context, helper: SQLCipherOpenHelper) : Database(context,
         return DatabaseFactory.getSSKDatabase(context).getAllClosedGroupPublicKeys()
     }
 
+    override fun addClosedGroupPublicKey(groupPublicKey: String) {
+        DatabaseFactory.getLokiAPIDatabase(context).addClosedGroupPublicKey(groupPublicKey)
+    }
+
+    override fun removeClosedGroupPublicKey(groupPublicKey: String) {
+        DatabaseFactory.getLokiAPIDatabase(context).removeClosedGroupPublicKey(groupPublicKey)
+    }
+
+    override fun addClosedGroupEncryptionKeyPair(encryptionKeyPair: ECKeyPair, groupPublicKey: String) {
+        DatabaseFactory.getLokiAPIDatabase(context).addClosedGroupEncryptionKeyPair(encryptionKeyPair, groupPublicKey)
+    }
+
+    override fun removeAllClosedGroupEncryptionKeyPairs(groupPublicKey: String) {
+        DatabaseFactory.getLokiAPIDatabase(context).removeAllClosedGroupEncryptionKeyPairs(groupPublicKey)
+    }
+
     override fun getAllOpenGroups(): Map<Long, PublicChat> {
         return DatabaseFactory.getLokiThreadDatabase(context).getAllPublicChats()
     }

libsession/src/main/java/org/session/libsession/messaging/StorageProtocol.kt

@@ -57,6 +57,7 @@ interface StorageProtocol {
     // Open Groups
     fun getOpenGroup(threadID: String): OpenGroup?
     fun getThreadID(openGroupID: String): String?
+    fun getAllOpenGroups(): Map<Long, PublicChat>
 
     // Open Group Public Keys
     fun getOpenGroupPublicKey(server: String): String?
@@ -66,6 +67,13 @@ interface StorageProtocol {
     fun setOpenGroupDisplayName(publicKey: String, channel: Long, server: String, displayName: String)
     fun getOpenGroupDisplayName(publicKey: String, channel: Long, server: String): String?
 
+    // Open Group Metadata
+    fun setUserCount(group: Long, server: String, newValue: Int)
+    fun setOpenGroupProfilePictureURL(group: Long, server: String, newValue: String)
+    fun getOpenGroupProfilePictureURL(group: Long, server: String): String?
+    fun updateTitle(groupID: String, newValue: String)
+    fun updateProfilePicture(groupID: String, newValue: ByteArray)
+
     // Last Message Server ID
     fun getLastMessageServerID(group: Long, server: String): Long?
     fun setLastMessageServerID(group: Long, server: String, newValue: Long)
@@ -76,13 +84,6 @@ interface StorageProtocol {
     fun setLastDeletionServerID(group: Long, server: String, newValue: Long)
     fun removeLastDeletionServerID(group: Long, server: String)
 
-    // Open Group Metadata
-    fun setUserCount(group: Long, server: String, newValue: Int)
-    fun setOpenGroupProfilePictureURL(group: Long, server: String, newValue: String)
-    fun getOpenGroupProfilePictureURL(group: Long, server: String): String?
-    fun updateTitle(groupID: String, newValue: String)
-    fun updateProfilePicture(groupID: String, newValue: ByteArray)
-
     // Message Handling
     fun getReceivedMessageTimestamps(): Set<Long>
     fun addReceivedMessageTimestamp(timestamp: Long)
@@ -102,6 +103,11 @@ interface StorageProtocol {
     fun removeMember(groupID: String, member: Address)
     fun updateMembers(groupID: String, members: List<Address>)
     // Closed Group
+    fun getAllClosedGroupPublicKeys(): Set<String>
+    fun addClosedGroupPublicKey(groupPublicKey: String)
+    fun removeClosedGroupPublicKey(groupPublicKey: String)
+    fun addClosedGroupEncryptionKeyPair(encryptionKeyPair: ECKeyPair, groupPublicKey: String)
+    fun removeAllClosedGroupEncryptionKeyPairs(groupPublicKey: String)
     fun insertIncomingInfoMessage(context: Context, senderPublicKey: String, groupID: String, type0: SignalServiceProtos.GroupContext.Type, type1: SignalServiceGroup.Type,
                                   name: String, members: Collection<String>, admins: Collection<String>)
     fun insertOutgoingInfoMessage(context: Context, groupID: String, type: SignalServiceProtos.GroupContext.Type, name: String,
@@ -109,9 +115,8 @@ interface StorageProtocol {
     fun isClosedGroup(publicKey: String): Boolean
     fun getClosedGroupEncryptionKeyPairs(groupPublicKey: String): MutableList<ECKeyPair>
     fun getLatestClosedGroupEncryptionKeyPair(groupPublicKey: String): ECKeyPair?
+
     // Groups
-    fun getAllClosedGroupPublicKeys(): Set<String>
-    fun getAllOpenGroups(): Map<Long, PublicChat>
     fun getAllGroups(): List<GroupRecord>
 
     // Settings

libsession/src/main/java/org/session/libsession/messaging/messages/control/ClosedGroupControlMessage.kt

@@ -6,6 +6,8 @@ import org.session.libsignal.libsignal.ecc.DjbECPublicKey
 import org.session.libsignal.libsignal.ecc.ECKeyPair
 import org.session.libsignal.utilities.logging.Log
 import org.session.libsignal.service.internal.push.SignalServiceProtos
+import org.session.libsignal.service.loki.utilities.toHexString
+import org.session.libsignal.utilities.Hex
 
 class ClosedGroupControlMessage() : ControlMessage() {
 
@@ -33,11 +35,23 @@ class ClosedGroupControlMessage() : ControlMessage() {
         class NameChange(val name: String) : Kind()
         class MembersAdded(val members: List<ByteString>) : Kind()
         class MembersRemoved( val members: List<ByteString>) : Kind()
-        class MemberLeft() : Kind()
+        object MemberLeft : Kind()
+
+        val description: String = run {
+            when(this) {
+                is New -> "new"
+                is Update -> "update"
+                is EncryptionKeyPair -> "encryptionKeyPair"
+                is NameChange -> "nameChange"
+                is MembersAdded -> "membersAdded"
+                is MembersRemoved -> "membersRemoved"
+                MemberLeft -> "memberLeft"
+            }
+        }
     }
 
     companion object {
-        const val TAG = "ClosedGroupUpdateV2"
+        const val TAG = "ClosedGroupControlMessage"
 
         fun fromProto(proto: SignalServiceProtos.Content): ClosedGroupControlMessage? {
             val closedGroupUpdateProto = proto.dataMessage?.closedGroupUpdateV2 ?: return null
@@ -75,7 +89,7 @@ class ClosedGroupControlMessage() : ControlMessage() {
                     kind = Kind.MembersRemoved(closedGroupUpdateProto.membersList)
                 }
                 SignalServiceProtos.ClosedGroupUpdateV2.Type.MEMBER_LEFT -> {
-                    kind = Kind.MemberLeft()
+                    kind = Kind.MemberLeft
                 }
             }
             return ClosedGroupControlMessage(kind)
@@ -168,10 +182,15 @@ class ClosedGroupControlMessage() : ControlMessage() {
         }
     }
 
-    final class KeyPairWrapper(val publicKey: String?, private val encryptedKeyPair: ByteString?) {
+    class KeyPairWrapper(val publicKey: String?, val encryptedKeyPair: ByteString?) {
+
+        val isValid: Boolean = run {
+            this.publicKey != null && this.encryptedKeyPair != null
+        }
+
         companion object {
             fun fromProto(proto: SignalServiceProtos.ClosedGroupUpdateV2.KeyPairWrapper): KeyPairWrapper {
-                return KeyPairWrapper(proto.publicKey.toString(), proto.encryptedKeyPair)
+                return KeyPairWrapper(proto.publicKey.toByteArray().toHexString(), proto.encryptedKeyPair)
             }
         }
 
@@ -179,7 +198,7 @@ class ClosedGroupControlMessage() : ControlMessage() {
             val publicKey = publicKey ?: return null
             val encryptedKeyPair = encryptedKeyPair ?: return null
             val result = SignalServiceProtos.ClosedGroupUpdateV2.KeyPairWrapper.newBuilder()
-            result.publicKey = ByteString.copyFrom(publicKey.toByteArray())
+            result.publicKey = ByteString.copyFrom(Hex.fromStringCondensed(publicKey))
             result.encryptedKeyPair = encryptedKeyPair
 
             return try {

libsession/src/main/java/org/session/libsession/messaging/messages/control/ConfigurationMessage.kt

@@ -62,7 +62,7 @@ class ConfigurationMessage(val closedGroups: List<ClosedGroup>, val openGroups:
             for (groupRecord in groups) {
                 if (groupRecord.isClosedGroup) {
                     if (!groupRecord.members.contains(Address.fromSerialized(storage.getUserPublicKey()!!))) continue
-                    val groupPublicKey = GroupUtil.getDecodedGroupIDAsData(GroupUtil.getDecodedGroupID(groupRecord.encodedId)).toHexString() // Double decoded
+                    val groupPublicKey = GroupUtil.getDecodedGroupIDAsData(groupRecord.encodedId).toHexString()
                     if (!storage.isClosedGroup(groupPublicKey)) continue
                     val encryptionKeyPair = storage.getLatestClosedGroupEncryptionKeyPair(groupPublicKey) ?: continue
                     val closedGroup = ClosedGroup(groupPublicKey, groupRecord.title, encryptionKeyPair, groupRecord.members.map { it.serialize() }, groupRecord.admins.map { it.serialize() })

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/MessageReceiverHandler.kt

@@ -4,12 +4,8 @@ import android.text.TextUtils
 import org.session.libsession.messaging.MessagingConfiguration
 import org.session.libsession.messaging.jobs.AttachmentDownloadJob
 import org.session.libsession.messaging.jobs.JobQueue
-import org.session.libsession.messaging.messages.Destination
 import org.session.libsession.messaging.messages.Message
-import org.session.libsession.messaging.messages.control.ClosedGroupUpdate
-import org.session.libsession.messaging.messages.control.ExpirationTimerUpdate
-import org.session.libsession.messaging.messages.control.ReadReceipt
-import org.session.libsession.messaging.messages.control.TypingIndicator
+import org.session.libsession.messaging.messages.control.*
 import org.session.libsession.messaging.messages.visible.Attachment
 import org.session.libsession.messaging.messages.visible.VisibleMessage
 import org.session.libsession.messaging.sending_receiving.attachments.PointerAttachment
@@ -17,19 +13,19 @@ import org.session.libsession.messaging.sending_receiving.linkpreview.LinkPrevie
 import org.session.libsession.messaging.sending_receiving.notifications.PushNotificationAPI
 import org.session.libsession.messaging.sending_receiving.quotes.QuoteModel
 import org.session.libsession.messaging.threads.Address
+import org.session.libsession.messaging.threads.GroupRecord
 import org.session.libsession.messaging.threads.recipients.Recipient
 import org.session.libsession.utilities.GroupUtil
-import org.session.libsignal.utilities.Hex
 import org.session.libsession.utilities.SSKEnvironment
 import org.session.libsession.utilities.TextSecurePreferences
+import org.session.libsignal.libsignal.ecc.DjbECPrivateKey
+import org.session.libsignal.libsignal.ecc.DjbECPublicKey
+import org.session.libsignal.libsignal.ecc.ECKeyPair
 import org.session.libsignal.utilities.logging.Log
 import org.session.libsignal.libsignal.util.guava.Optional
 import org.session.libsignal.service.api.messages.SignalServiceGroup
 import org.session.libsignal.service.internal.push.SignalServiceProtos
-import org.session.libsignal.service.loki.protocol.closedgroups.ClosedGroupRatchet
-import org.session.libsignal.service.loki.protocol.closedgroups.ClosedGroupRatchetCollectionType
-import org.session.libsignal.service.loki.protocol.closedgroups.ClosedGroupSenderKey
-import org.session.libsignal.service.loki.protocol.closedgroups.SharedSenderKeysImplementation
+import org.session.libsignal.service.loki.utilities.removing05PrefixIfNeeded
 import org.session.libsignal.service.loki.utilities.toHexString
 import java.security.MessageDigest
 import java.util.*
@@ -45,8 +41,9 @@ fun MessageReceiver.handle(message: Message, proto: SignalServiceProtos.Content,
     when (message) {
         is ReadReceipt -> handleReadReceipt(message)
         is TypingIndicator -> handleTypingIndicator(message)
-        is ClosedGroupUpdate -> handleClosedGroupUpdate(message)
+        is ClosedGroupControlMessage -> handleClosedGroupControlMessage(message)
         is ExpirationTimerUpdate -> handleExpirationTimerUpdate(message, proto)
+        is ConfigurationMessage -> handleConfigurationMessage(message)
         is VisibleMessage -> handleVisibleMessage(message, proto, openGroupID)
     }
 }
@@ -105,6 +102,21 @@ fun MessageReceiver.disableExpirationTimer(message: ExpirationTimerUpdate, proto
     SSKEnvironment.shared.messageExpirationManager.disableExpirationTimer(id, senderPublicKey, proto)
 }
 
+private fun MessageReceiver.handleConfigurationMessage(message: ConfigurationMessage) {
+    val storage = MessagingConfiguration.shared.storage
+    if (message.sender != storage.getUserPublicKey()) return
+    val allClosedGroupPublicKeys = storage.getAllClosedGroupPublicKeys()
+    for (closeGroup in message.closedGroups) {
+        if (allClosedGroupPublicKeys.contains(closeGroup.publicKey)) continue
+        handleNewClosedGroup(message.sender!!, closeGroup.publicKey, closeGroup.name, closeGroup.encryptionKeyPair, closeGroup.members, closeGroup.admins)
+    }
+    val allOpenGroups = storage.getAllOpenGroups().map { it.value.server }
+    for (openGroup in message.openGroups) {
+        if (allOpenGroups.contains(openGroup)) continue
+        // TODO
+    }
+}
+
 fun MessageReceiver.handleVisibleMessage(message: VisibleMessage, proto: SignalServiceProtos.Content, openGroupID: String?) {
     val storage = MessagingConfiguration.shared.storage
     val context = MessagingConfiguration.shared.context
@@ -188,173 +200,293 @@ fun MessageReceiver.handleVisibleMessage(message: VisibleMessage, proto: SignalS
     SSKEnvironment.shared.notificationManager.updateNotification(context, threadID)
 }
 
-private fun MessageReceiver.handleClosedGroupUpdate(message: ClosedGroupUpdate) {
+private fun MessageReceiver.handleClosedGroupControlMessage(message: ClosedGroupControlMessage) {
     when (message.kind!!) {
-        is ClosedGroupUpdate.Kind.New -> handleNewGroup(message)
-        is ClosedGroupUpdate.Kind.Info -> handleGroupUpdate(message)
-        is ClosedGroupUpdate.Kind.SenderKeyRequest -> handleSenderKeyRequest(message)
-        is ClosedGroupUpdate.Kind.SenderKey -> handleSenderKey(message)
+        is ClosedGroupControlMessage.Kind.New -> handleNewClosedGroup(message)
+        is ClosedGroupControlMessage.Kind.Update -> handleClosedGroupUpdated(message)
+        is ClosedGroupControlMessage.Kind.EncryptionKeyPair -> handleClosedGroupEncryptionKeyPair(message)
+        is ClosedGroupControlMessage.Kind.NameChange -> handleClosedGroupNameChanged(message)
+        is ClosedGroupControlMessage.Kind.MembersAdded -> handleClosedGroupMembersAdded(message)
+        is ClosedGroupControlMessage.Kind.MembersRemoved -> handleClosedGroupMembersRemoved(message)
+        ClosedGroupControlMessage.Kind.MemberLeft -> handleClosedGroupMemberLeft(message)
     }
 }
 
-private fun MessageReceiver.handleNewGroup(message: ClosedGroupUpdate) {
+private fun MessageReceiver.handleNewClosedGroup(message: ClosedGroupControlMessage) {
+    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.New ?: return
+    val groupPublicKey = kind.publicKey.toByteArray().toHexString()
+    val members = kind.members.map { it.toByteArray().toHexString() }
+    val admins = kind.admins.map { it.toByteArray().toHexString() }
+    handleNewClosedGroup(message.sender!!, groupPublicKey, kind.name, kind.encryptionKeyPair, members, admins)
+}
+
+// Parameter @sender:String is just for inserting incoming info message
+private fun handleNewClosedGroup(sender: String, groupPublicKey: String, name: String, encryptionKeyPair: ECKeyPair, members: List<String>, admins: List<String>) {
     val context = MessagingConfiguration.shared.context
     val storage = MessagingConfiguration.shared.storage
-    val sskDatabase = MessagingConfiguration.shared.sskDatabase
-    if (message.kind !is ClosedGroupUpdate.Kind.New) { return }
-    val kind = message.kind!! as ClosedGroupUpdate.Kind.New
-    val groupPublicKey = kind.groupPublicKey.toHexString()
-    val name = kind.name
-    val groupPrivateKey = kind.groupPrivateKey
-    val senderKeys = kind.senderKeys
-    val members = kind.members.map { it.toHexString() }
-    val admins = kind.admins.map { it.toHexString() }
-    // Persist the ratchets
-    senderKeys.forEach { senderKey ->
-        if (!members.contains(senderKey.publicKey.toHexString())) { return@forEach }
-        val ratchet = ClosedGroupRatchet(senderKey.chainKey.toHexString(), senderKey.keyIndex, listOf())
-        sskDatabase.setClosedGroupRatchet(groupPublicKey, senderKey.publicKey.toHexString(), ratchet, ClosedGroupRatchetCollectionType.Current)
-    }
-    // Sort out any discrepancies between the provided sender keys and what's required
-    val missingSenderKeys = members.toSet().subtract(senderKeys.map { Hex.toStringCondensed(it.publicKey) })
-    val userPublicKey = storage.getUserPublicKey()!!
-    if (missingSenderKeys.contains(userPublicKey)) {
-        val userRatchet = SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, userPublicKey)
-        val userSenderKey = ClosedGroupSenderKey(Hex.fromStringCondensed(userRatchet.chainKey), userRatchet.keyIndex, Hex.fromStringCondensed(userPublicKey))
-        members.forEach { member ->
-            if (member == userPublicKey) return@forEach
-            val closedGroupUpdateKind = ClosedGroupUpdate.Kind.SenderKey(groupPublicKey.toByteArray(), userSenderKey)
-            val closedGroupUpdate = ClosedGroupUpdate()
-            closedGroupUpdate.kind = closedGroupUpdateKind
-            MessageSender.send(closedGroupUpdate, Destination.ClosedGroup(groupPublicKey))
-        }
-    }
-    missingSenderKeys.minus(userPublicKey).forEach { publicKey ->
-        MessageSender.requestSenderKey(groupPublicKey, publicKey)
-    }
     // Create the group
-    val groupID = GroupUtil.getEncodedClosedGroupID(GroupUtil.getEncodedClosedGroupID(Hex.fromStringCondensed(groupPublicKey)).toByteArray()) //double encoded
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
     if (storage.getGroup(groupID) != null) {
         // Update the group
         storage.updateTitle(groupID, name)
         storage.updateMembers(groupID, members.map { Address.fromSerialized(it) })
     } else {
         storage.createGroup(groupID, name, LinkedList(members.map { Address.fromSerialized(it) }),
-                null, null, LinkedList(admins.map { Address.fromSerialized(it) }))
+                            null, null, LinkedList(admins.map { Address.fromSerialized(it) }))
     }
     storage.setProfileSharing(Address.fromSerialized(groupID), true)
     // Add the group to the user's set of public keys to poll for
-    sskDatabase.setClosedGroupPrivateKey(groupPublicKey, groupPrivateKey.toHexString())
-    // Notify the PN server
-    PushNotificationAPI.performOperation(PushNotificationAPI.ClosedGroupOperation.Subscribe, groupPublicKey, userPublicKey)
+    storage.addClosedGroupPublicKey(groupPublicKey)
+    // Store the encryption key pair
+    storage.addClosedGroupEncryptionKeyPair(encryptionKeyPair, groupPublicKey)
     // Notify the user
-    storage.insertIncomingInfoMessage(context, message.sender!!, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
+    storage.insertIncomingInfoMessage(context, sender, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
+    // Notify the PN server
+    PushNotificationAPI.performOperation(PushNotificationAPI.ClosedGroupOperation.Subscribe, groupPublicKey, storage.getUserPublicKey()!!)
 }
 
-private fun MessageReceiver.handleGroupUpdate(message: ClosedGroupUpdate) {
+private fun MessageReceiver.handleClosedGroupUpdated(message: ClosedGroupControlMessage) {
+    // Prepare
     val context = MessagingConfiguration.shared.context
     val storage = MessagingConfiguration.shared.storage
-    val sskDatabase = MessagingConfiguration.shared.sskDatabase
-    if (message.kind !is ClosedGroupUpdate.Kind.Info) { return }
-    val kind = message.kind!! as ClosedGroupUpdate.Kind.Info
-    val groupPublicKey = kind.groupPublicKey.toHexString()
-    val name = kind.name
-    val senderKeys = kind.senderKeys
-    val members = kind.members.map { it.toHexString() }
-    val admins = kind.admins.map { it.toHexString() }
-    // Get the group
-    val groupID = GroupUtil.getEncodedClosedGroupID(GroupUtil.getEncodedClosedGroupID(Hex.fromStringCondensed(groupPublicKey)).toByteArray()) //double encoded
-    val group = storage.getGroup(groupID) ?: return Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
-    // Check that the sender is a member of the group (before the update)
-    if (!group.members.contains(Address.fromSerialized(message.sender!!))) { return Log.d("Loki", "Ignoring closed group info message from non-member.") }
-    // Store the ratchets for any new members (it's important that this happens before the code below)
-    senderKeys.forEach { senderKey ->
-        val ratchet = ClosedGroupRatchet(senderKey.chainKey.toHexString(), senderKey.keyIndex, listOf())
-        sskDatabase.setClosedGroupRatchet(groupPublicKey, senderKey.publicKey.toHexString(), ratchet, ClosedGroupRatchetCollectionType.Current)
-    }
-    // Delete all ratchets and either:
-    // • Send out the user's new ratchet using established channels if other members of the group left or were removed
-    // • Remove the group from the user's set of public keys to poll for if the current user was among the members that were removed
-    val oldMembers = group.members.map { it.serialize() }.toSet()
+    val senderPublicKey = message.sender ?: return
+    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.Update ?: return
+    val groupPublicKey = message.groupPublicKey ?: return
     val userPublicKey = storage.getUserPublicKey()!!
-    val wasUserRemoved = !members.contains(userPublicKey)
-    val wasSenderRemoved = !members.contains(message.sender!!)
-    if (members.toSet().intersect(oldMembers) != oldMembers.toSet()) {
-        val allOldRatchets = sskDatabase.getAllClosedGroupRatchets(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
-        for (pair in allOldRatchets) {
-            val senderPublicKey = pair.first
-            val ratchet = pair.second
-            val collection = ClosedGroupRatchetCollectionType.Old
-            sskDatabase.setClosedGroupRatchet(groupPublicKey, senderPublicKey, ratchet, collection)
-        }
-        sskDatabase.removeAllClosedGroupRatchets(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
-        if (wasUserRemoved) {
-            sskDatabase.removeClosedGroupPrivateKey(groupPublicKey)
-            storage.setActive(groupID, false)
-            storage.removeMember(groupID, Address.fromSerialized(userPublicKey))
-            // Notify the PN server
-            PushNotificationAPI.performOperation(PushNotificationAPI.ClosedGroupOperation.Unsubscribe, groupPublicKey, userPublicKey)
-        } else {
-            val userRatchet = SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, userPublicKey)
-            val userSenderKey = ClosedGroupSenderKey(Hex.fromStringCondensed(userRatchet.chainKey), userRatchet.keyIndex, Hex.fromStringCondensed(userPublicKey))
-            members.forEach { member ->
-                if (member == userPublicKey) return@forEach
-                val address = Address.fromSerialized(member)
-                val closedGroupUpdateKind = ClosedGroupUpdate.Kind.SenderKey(Hex.fromStringCondensed(groupPublicKey), userSenderKey)
-                val closedGroupUpdate = ClosedGroupUpdate()
-                closedGroupUpdate.kind = closedGroupUpdateKind
-                MessageSender.send(closedGroupUpdate, address)
-            }
-        }
+    // Unwrap the message
+    val name = kind.name
+    val members = kind.members.map { it.toByteArray().toHexString() }
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
+        return
+    }
+    val oldMembers = group.members.map { it.serialize() }
+    // Check common group update logic
+    if (!isValidGroupUpdate(group, message.sentTimestamp!!, senderPublicKey)) {
+        return
+    }
+    // Check that the admin wasn't removed unless the group was destroyed entirely
+    if (!members.contains(group.admins.first().toString()) && members.isNotEmpty()) {
+        android.util.Log.d("Loki", "Ignoring invalid closed group update message.")
+        return
+    }
+    // Remove the group from the user's set of public keys to poll for if the current user was removed
+    val wasCurrentUserRemoved = !members.contains(userPublicKey)
+    if (wasCurrentUserRemoved) {
+        disableLocalGroupAndUnsubscribe(groupPublicKey, groupID, userPublicKey)
+    }
+    // Generate and distribute a new encryption key pair if needed
+    val wasAnyUserRemoved = (members.toSet().intersect(oldMembers) != oldMembers.toSet())
+    val isCurrentUserAdmin = group.admins.map { it.toString() }.contains(userPublicKey)
+    if (wasAnyUserRemoved && isCurrentUserAdmin) {
+        MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey, members)
     }
     // Update the group
     storage.updateTitle(groupID, name)
-    storage.updateMembers(groupID, members.map { Address.fromSerialized(it) })
-    // Notify the user if needed
+    if (!wasCurrentUserRemoved) {
+        // The call below sets isActive to true, so if the user is leaving we have to use groupDB.remove(...) instead
+        storage.updateMembers(groupID, members.map { Address.fromSerialized(it) })
+    }
+    // Notify the user
+    val wasSenderRemoved = !members.contains(senderPublicKey)
     val type0 = if (wasSenderRemoved) SignalServiceProtos.GroupContext.Type.QUIT else SignalServiceProtos.GroupContext.Type.UPDATE
     val type1 = if (wasSenderRemoved) SignalServiceGroup.Type.QUIT else SignalServiceGroup.Type.UPDATE
-    storage.insertIncomingInfoMessage(context, message.sender!!, groupID, type0, type1, name, members, admins)
+    storage.insertIncomingInfoMessage(context, senderPublicKey, groupID, type0, type1, name, members, group.admins.map { it.toString() })
 }
 
-private fun MessageReceiver.handleSenderKeyRequest(message: ClosedGroupUpdate) {
-    if (message.kind !is ClosedGroupUpdate.Kind.SenderKeyRequest) { return }
-    val kind = message.kind!! as ClosedGroupUpdate.Kind.SenderKeyRequest
+private fun MessageReceiver.handleClosedGroupEncryptionKeyPair(message: ClosedGroupControlMessage) {
+    // Prepare
     val storage = MessagingConfiguration.shared.storage
-    val sskDatabase = MessagingConfiguration.shared.sskDatabase
+    val senderPublicKey = message.sender ?: return
+    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.EncryptionKeyPair ?: return
+    val groupPublicKey = message.groupPublicKey ?: return
     val userPublicKey = storage.getUserPublicKey()!!
-    val groupPublicKey = kind.groupPublicKey.toHexString()
-    val groupID = GroupUtil.getEncodedClosedGroupID(GroupUtil.getEncodedClosedGroupID(Hex.fromStringCondensed(groupPublicKey)).toByteArray()) //double encoded
-    val group = storage.getGroup(groupID)
-    if (group == null) {
-        Log.d("Loki", "Ignoring closed group sender key request for nonexistent group.")
+    val userKeyPair = storage.getUserX25519KeyPair()
+    // Unwrap the message
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
         return
     }
-    // Check that the requesting user is a member of the group
-    if (!group.members.map { it.serialize() }.contains(message.sender!!)) {
-        Log.d("Loki", "Ignoring closed group sender key request from non-member.")
+    if (!group.admins.map { it.toString() }.contains(senderPublicKey)) {
+        android.util.Log.d("Loki", "Ignoring closed group encryption key pair from non-admin.")
         return
     }
-    // Respond to the request
-    Log.d("Loki", "Responding to sender key request from: ${message.sender!!}.")
-    val userRatchet = sskDatabase.getClosedGroupRatchet(groupPublicKey, userPublicKey, ClosedGroupRatchetCollectionType.Current)
-            ?: SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, userPublicKey)
-    val userSenderKey = ClosedGroupSenderKey(Hex.fromStringCondensed(userRatchet.chainKey), userRatchet.keyIndex, Hex.fromStringCondensed(userPublicKey))
-    val closedGroupUpdateKind = ClosedGroupUpdate.Kind.SenderKey(Hex.fromStringCondensed(groupPublicKey), userSenderKey)
-    val closedGroupUpdate = ClosedGroupUpdate()
-    closedGroupUpdate.kind = closedGroupUpdateKind
-    MessageSender.send(closedGroupUpdate, Address.fromSerialized(groupID))
+    // Find our wrapper and decrypt it if possible
+    val wrapper = kind.wrappers.firstOrNull { it.publicKey!!.toByteArray().toHexString() == userPublicKey } ?: return
+    val encryptedKeyPair = wrapper.encryptedKeyPair!!.toByteArray()
+    val plaintext = MessageReceiverDecryption.decryptWithSessionProtocol(encryptedKeyPair, userKeyPair).first
+    // Parse it
+    val proto = SignalServiceProtos.KeyPair.parseFrom(plaintext)
+    val keyPair = ECKeyPair(DjbECPublicKey(proto.publicKey.toByteArray().removing05PrefixIfNeeded()), DjbECPrivateKey(proto.privateKey.toByteArray()))
+    // Store it
+    storage.addClosedGroupEncryptionKeyPair(keyPair, groupPublicKey)
+    Log.d("Loki", "Received a new closed group encryption key pair")
 }
 
-private fun MessageReceiver.handleSenderKey(message: ClosedGroupUpdate) {
-    if (message.kind !is ClosedGroupUpdate.Kind.SenderKey) { return }
-    val kind = message.kind!! as ClosedGroupUpdate.Kind.SenderKey
-    val groupPublicKey = kind.groupPublicKey.toHexString()
-    val senderKey = kind.senderKey
-    if (senderKey.publicKey.toHexString() != message.sender!!) {
-        Log.d("Loki", "Ignoring invalid closed group sender key.")
+private fun MessageReceiver.handleClosedGroupNameChanged(message: ClosedGroupControlMessage) {
+    val context = MessagingConfiguration.shared.context
+    val storage = MessagingConfiguration.shared.storage
+    val senderPublicKey = message.sender ?: return
+    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.NameChange ?: return
+    val groupPublicKey = message.groupPublicKey ?: return
+    // Check that the sender is a member of the group (before the update)
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
         return
     }
-    Log.d("Loki", "Received a sender key from: ${message.sender!!}.")
-    val ratchet = ClosedGroupRatchet(senderKey.chainKey.toHexString(), senderKey.keyIndex, listOf())
-    MessagingConfiguration.shared.sskDatabase.setClosedGroupRatchet(groupPublicKey, senderKey.publicKey.toHexString(), ratchet, ClosedGroupRatchetCollectionType.Current)
+    // Check common group update logic
+    if (!isValidGroupUpdate(group, message.sentTimestamp!!, senderPublicKey)) {
+        return
+    }
+    val members = group.members.map { it.serialize() }
+    val admins = group.admins.map { it.serialize() }
+    val name = kind.name
+    storage.updateTitle(groupID, name)
+
+    storage.insertIncomingInfoMessage(context, senderPublicKey, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
+}
+
+private fun MessageReceiver.handleClosedGroupMembersAdded(message: ClosedGroupControlMessage) {
+    val context = MessagingConfiguration.shared.context
+    val storage = MessagingConfiguration.shared.storage
+    val senderPublicKey = message.sender ?: return
+    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.MembersAdded ?: return
+    val groupPublicKey = message.groupPublicKey ?: return
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
+        return
+    }
+    if (!isValidGroupUpdate(group, message.sentTimestamp!!, senderPublicKey)) {
+        return
+    }
+    val name = group.title
+    // Check common group update logic
+    val members = group.members.map { it.serialize() }
+    val admins = group.admins.map { it.serialize() }
+
+    // Users that are part of this remove update
+    val updateMembers = kind.members.map { it.toByteArray().toHexString() }
+    // newMembers to save is old members minus removed members
+    val newMembers = members + updateMembers
+    storage.updateMembers(groupID, newMembers.map { Address.fromSerialized(it) })
+
+    storage.insertIncomingInfoMessage(context, senderPublicKey, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
+}
+
+private fun MessageReceiver.handleClosedGroupMembersRemoved(message: ClosedGroupControlMessage) {
+    val context = MessagingConfiguration.shared.context
+    val storage = MessagingConfiguration.shared.storage
+    val userPublicKey = storage.getUserPublicKey()!!
+    val senderPublicKey = message.sender ?: return
+    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.MembersRemoved ?: return
+    val groupPublicKey = message.groupPublicKey ?: return
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
+        return
+    }
+    val name = group.title
+    // Check common group update logic
+    val members = group.members.map { it.serialize() }
+    val admins = group.admins.map { it.toString() }
+
+    // Users that are part of this remove update
+    val updateMembers = kind.members.map { it.toByteArray().toHexString() }
+
+    if (!isValidGroupUpdate(group, message.sentTimestamp!!, senderPublicKey)) { return }
+    // If admin leaves the group is disbanded
+    val didAdminLeave = admins.any { it in updateMembers }
+    // newMembers to save is old members minus removed members
+    val newMembers = members - updateMembers
+    // user should be posting MEMBERS_LEFT so this should not be encountered
+    val senderLeft = senderPublicKey in updateMembers
+    if (senderLeft) {
+        android.util.Log.d("Loki", "Received a MEMBERS_REMOVED instead of a MEMBERS_LEFT from sender $senderPublicKey")
+    }
+    val wasCurrentUserRemoved = userPublicKey in updateMembers
+
+    // admin should send a MEMBERS_LEFT message but handled here in case
+    if (didAdminLeave || wasCurrentUserRemoved) {
+        disableLocalGroupAndUnsubscribe(groupPublicKey, groupID, userPublicKey)
+    } else {
+        val isCurrentUserAdmin = admins.contains(userPublicKey)
+        storage.updateMembers(groupID, newMembers.map { Address.fromSerialized(it) })
+        if (isCurrentUserAdmin) {
+            MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey, newMembers)
+        }
+    }
+    val (contextType, signalType) =
+            if (senderLeft) SignalServiceProtos.GroupContext.Type.QUIT to SignalServiceGroup.Type.QUIT
+            else SignalServiceProtos.GroupContext.Type.UPDATE to SignalServiceGroup.Type.UPDATE
+
+    storage.insertIncomingInfoMessage(context, senderPublicKey, groupID, contextType, signalType, name, members, admins)
+}
+
+private fun MessageReceiver.handleClosedGroupMemberLeft(message: ClosedGroupControlMessage) {
+    val context = MessagingConfiguration.shared.context
+    val storage = MessagingConfiguration.shared.storage
+    val senderPublicKey = message.sender ?: return
+    val userPublicKey = storage.getUserPublicKey()!!
+    if (senderPublicKey == userPublicKey) { return } // Check the user leaving isn't us, will already be handled
+    val kind = message.kind!! as? ClosedGroupControlMessage.Kind.MembersAdded ?: return
+    val groupPublicKey = message.groupPublicKey ?: return
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Ignoring closed group info message for nonexistent group.")
+        return
+    }
+    val name = group.title
+    // Check common group update logic
+    val members = group.members.map { it.serialize() }
+    val admins = group.admins.map { it.toString() }
+    if (!isValidGroupUpdate(group, message.sentTimestamp!!, senderPublicKey)) {
+        return
+    }
+    // If admin leaves the group is disbanded
+    val didAdminLeave = admins.contains(senderPublicKey)
+    val updatedMemberList = members - senderPublicKey
+
+    if (didAdminLeave) {
+        disableLocalGroupAndUnsubscribe(groupPublicKey, groupID, userPublicKey)
+    } else {
+        val isCurrentUserAdmin = admins.contains(userPublicKey)
+        storage.updateMembers(groupID, updatedMemberList.map { Address.fromSerialized(it) })
+        if (isCurrentUserAdmin) {
+            MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey, updatedMemberList)
+        }
+    }
+    storage.insertIncomingInfoMessage(context, senderPublicKey, groupID, SignalServiceProtos.GroupContext.Type.QUIT, SignalServiceGroup.Type.QUIT, name, members, admins)
+}
+
+private fun isValidGroupUpdate(group: GroupRecord,
+                               sentTimestamp: Long,
+                               senderPublicKey: String): Boolean  {
+    val oldMembers = group.members.map { it.serialize() }
+    // Check that the message isn't from before the group was created
+    if (group.createdAt > sentTimestamp) {
+        android.util.Log.d("Loki", "Ignoring closed group update from before thread was created.")
+        return false
+    }
+    // Check that the sender is a member of the group (before the update)
+    if (senderPublicKey !in oldMembers) {
+        android.util.Log.d("Loki", "Ignoring closed group info message from non-member.")
+        return false
+    }
+    return true
+}
+
+private fun disableLocalGroupAndUnsubscribe(groupPublicKey: String, groupID: String, userPublicKey: String) {
+    val storage = MessagingConfiguration.shared.storage
+    storage.removeClosedGroupPublicKey(groupPublicKey)
+    // Remove the key pairs
+    storage.removeAllClosedGroupEncryptionKeyPairs(groupPublicKey)
+    // Mark the group as inactive
+    storage.setActive(groupID, false)
+    storage.removeMember(groupID, Address.fromSerialized(userPublicKey))
+    // Notify the PN server
+    PushNotificationAPI.performOperation(PushNotificationAPI.ClosedGroupOperation.Unsubscribe, groupPublicKey, userPublicKey)
 }

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/MessageSenderClosedGroup.kt

@@ -3,10 +3,12 @@
 package org.session.libsession.messaging.sending_receiving
 
 import android.util.Log
+import com.google.protobuf.ByteString
 import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.deferred
 
 import org.session.libsession.messaging.MessagingConfiguration
+import org.session.libsession.messaging.messages.control.ClosedGroupControlMessage
 import org.session.libsession.messaging.messages.control.ClosedGroupUpdate
 import org.session.libsession.messaging.sending_receiving.notifications.PushNotificationAPI
 import org.session.libsession.messaging.sending_receiving.MessageSender.Error
@@ -21,6 +23,7 @@ import org.session.libsignal.service.loki.protocol.closedgroups.ClosedGroupSende
 import org.session.libsignal.service.loki.protocol.closedgroups.SharedSenderKeysImplementation
 import org.session.libsignal.service.loki.utilities.hexEncodedPrivateKey
 import org.session.libsignal.service.loki.utilities.hexEncodedPublicKey
+import org.session.libsignal.service.loki.utilities.removing05PrefixIfNeeded
 import java.util.*
 
 fun MessageSender.createClosedGroup(name: String, members: Collection<String>): Promise<String, Exception> {
@@ -216,11 +219,34 @@ fun MessageSender.leave(groupPublicKey: String) {
     return update(groupPublicKey, newMembers, name).get()
 }
 
-fun MessageSender.requestSenderKey(groupPublicKey: String, senderPublicKey: String) {
-    Log.d("Loki", "Requesting sender key for group public key: $groupPublicKey, sender public key: $senderPublicKey.")
-    val address = Address.fromSerialized(senderPublicKey)
-    val closedGroupUpdateKind = ClosedGroupUpdate.Kind.SenderKeyRequest(Hex.fromStringCondensed(groupPublicKey))
-    val closedGroupUpdate = ClosedGroupUpdate()
-    closedGroupUpdate.kind = closedGroupUpdateKind
-    MessageSender.send(closedGroupUpdate, address)
+fun MessageSender.generateAndSendNewEncryptionKeyPair(groupPublicKey: String, targetMembers: Collection<String>) {
+    // Prepare
+    val storage = MessagingConfiguration.shared.storage
+    val userPublicKey = storage.getUserPublicKey()!!
+    val groupID = GroupUtil.doubleEncodeGroupID(groupPublicKey)
+    val group = storage.getGroup(groupID) ?: run {
+        Log.d("Loki", "Can't update nonexistent closed group.")
+        throw Error.NoThread
+    }
+    if (!group.admins.map { it.toString() }.contains(userPublicKey)) {
+        Log.d("Loki", "Can't distribute new encryption key pair as non-admin.")
+        throw Error.InvalidClosedGroupUpdate
+    }
+    // Generate the new encryption key pair
+    val newKeyPair = Curve.generateKeyPair()
+    // Distribute it
+    val proto = SignalServiceProtos.KeyPair.newBuilder()
+    proto.publicKey = ByteString.copyFrom(newKeyPair.publicKey.serialize().removing05PrefixIfNeeded())
+    proto.privateKey = ByteString.copyFrom(newKeyPair.privateKey.serialize())
+    val plaintext = proto.build().toByteArray()
+    val wrappers = targetMembers.map { publicKey ->
+        val ciphertext = MessageSenderEncryption.encryptWithSessionProtocol(plaintext, publicKey)
+        ClosedGroupControlMessage.KeyPairWrapper(publicKey, ByteString.copyFrom(ciphertext))
+    }
+    val kind = ClosedGroupControlMessage.Kind.EncryptionKeyPair(wrappers)
+    val closedGroupControlMessage = ClosedGroupControlMessage(kind)
+    sendNonDurably(closedGroupControlMessage, Address.fromSerialized(groupID)).success {
+        // Store it * after * having sent out the message to the group
+        storage.addClosedGroupEncryptionKeyPair(newKeyPair, groupPublicKey)
+    }
 }

libsession/src/main/java/org/session/libsession/messaging/sending_receiving/notifications/PushNotificationAPI.kt

@@ -1,5 +1,6 @@
 package org.session.libsession.messaging.sending_receiving.notifications
 
+import android.annotation.SuppressLint
 import nl.komponents.kovenant.functional.map
 import okhttp3.*
 import org.session.libsession.messaging.MessagingConfiguration

libsession/src/main/java/org/session/libsession/utilities/GroupUtil.kt

@@ -2,6 +2,8 @@ package org.session.libsession.utilities
 
 import org.session.libsignal.service.api.messages.SignalServiceGroup
 import org.session.libsignal.utilities.Hex
+import java.io.IOException
+import kotlin.jvm.Throws
 
 object GroupUtil {
     const val CLOSED_GROUP_PREFIX = "__textsecure_group__!"
@@ -65,4 +67,18 @@ object GroupUtil {
     fun isClosedGroup(groupId: String): Boolean {
         return groupId.startsWith(CLOSED_GROUP_PREFIX)
     }
+
+    // NOTE: Signal group ID handling is weird. The ID is double encoded in the database, but not in a `GroupContext`.
+
+    @JvmStatic
+    @Throws(IOException::class)
+    fun doubleEncodeGroupID(groupPublicKey: String): String {
+        return getEncodedClosedGroupID(getEncodedClosedGroupID(Hex.fromStringCondensed(groupPublicKey)).toByteArray())
+    }
+
+    @JvmStatic
+    @Throws(IOException::class)
+    fun doubleDecodeGroupID(groupID: String): ByteArray {
+        return getDecodedGroupIDAsData(getDecodedGroupID(groupID))
+    }
 }