message sender closed group + convenience

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

@@ -17,6 +17,8 @@ interface StorageProtocol {
     fun getUserProfileKey(): ByteArray?
     fun getUserProfilePictureURL(): String?
 
+    fun getProfileKeyForRecipient(recipientPublicKey: String): ByteArray?
+
     // Signal Protocol
 
     fun getOrGenerateRegistrationID(): Int //TODO needs impl
@@ -65,6 +67,10 @@ interface StorageProtocol {
     fun updateTitle(groupID: String, newValue: String)
     fun updateProfilePicture(groupID: String, newValue: ByteArray)
 
+    // Message Handling
+    fun getReceivedMessageTimestamps(): Set<Long>
+    fun addReceivedMessageTimestamp(timestamp: Long)
+
 
 
 

libsession/src/main/java/org/session/libsession/messaging/jobs/MessageSendJob.kt

@@ -1,6 +1,9 @@
 package org.session.libsession.messaging.jobs
 
-class MessageSendJob : Job {
+import org.session.libsession.messaging.messages.Destination
+import org.session.libsession.messaging.messages.Message
+
+class MessageSendJob(val message: Message, val destination: Destination) : Job {
     override var delegate: JobDelegate? = null
     override var id: String? = null
     override var failureCount: Int = 0

libsession/src/main/java/org/session/libsession/messaging/messages/Destination.kt

@@ -8,5 +8,8 @@ sealed class Destination {
 
     companion object {
         //TODO need to implement the equivalent to TSThread and then implement from(...)
+        fun from(threadID: String): Destination {
+            return Contact(threadID) // Fake for dev
+        }
     }
 }

libsession/src/main/java/org/session/libsession/messaging/messages/Message.kt

@@ -2,7 +2,7 @@ package org.session.libsession.messaging.messages
 
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 
-abstract class Message<T: com.google.protobuf.MessageOrBuilder?> {
+abstract class Message {
 
     var id: String? = null
     var threadID: String? = null
@@ -21,6 +21,6 @@ abstract class Message<T: com.google.protobuf.MessageOrBuilder?> {
         return sender != null && recipient != null
     }
 
-    abstract fun toProto(): T
+    abstract fun toProto(): SignalServiceProtos.Content?
 
 }

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

@@ -1,7 +1,6 @@
 package org.session.libsession.messaging.messages.control
 
 import org.session.libsession.messaging.messages.Message
-import org.session.libsignal.service.internal.push.SignalServiceProtos
 
-abstract class ControlMessage : Message<SignalServiceProtos.Content?>() {
+abstract class ControlMessage : Message() {
 }

libsession/src/main/java/org/session/libsession/messaging/messages/visible/Attachment.kt

@@ -5,7 +5,7 @@ import android.webkit.MimeTypeMap
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 import java.io.File
 
-class Attachment : VisibleMessageProto<SignalServiceProtos.AttachmentPointer?>() {
+class Attachment {
 
     var fileName: String? = null
     var contentType: String? = null
@@ -32,7 +32,7 @@ class Attachment : VisibleMessageProto<SignalServiceProtos.AttachmentPointer?>()
             result.digest = proto.digest.toByteArray()
             val kind: Kind
             if (proto.hasFlags() && (proto.flags and SignalServiceProtos.AttachmentPointer.Flags.VOICE_MESSAGE_VALUE) > 0) { //TODO validate that 'and' operator = swift '&'
-                kind = Kind.VOICEMESSAGE
+                kind = Kind.VOICE_MESSAGE
             } else {
                 kind = Kind.GENERIC
             }
@@ -52,18 +52,17 @@ class Attachment : VisibleMessageProto<SignalServiceProtos.AttachmentPointer?>()
     }
 
     enum class Kind {
-        VOICEMESSAGE,
+        VOICE_MESSAGE,
         GENERIC
     }
 
     // validation
-    override fun isValid(): Boolean {
+    fun isValid(): Boolean {
-        if (!super.isValid()) return false
         // key and digest can be nil for open group attachments
         return (contentType != null && kind != null && size != null && sizeInBytes != null && url != null)
     }
 
-    override fun toProto(): SignalServiceProtos.AttachmentPointer? {
+    fun toProto(): SignalServiceProtos.AttachmentPointer? {
         TODO("Not implemented")
     }
 }

libsession/src/main/java/org/session/libsession/messaging/messages/visible/Contact.kt

@@ -2,7 +2,7 @@ package org.session.libsession.messaging.messages.visible
 
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 
-class Contact : VisibleMessageProto<SignalServiceProtos.DataMessage.Contact?>() {
+class Contact() {
 
     companion object {
         fun fromProto(proto: SignalServiceProtos.Content): Contact? {
@@ -10,7 +10,7 @@ class Contact : VisibleMessageProto<SignalServiceProtos.DataMessage.Contact?>()
         }
     }
 
-    override fun toProto(): SignalServiceProtos.DataMessage.Contact? {
+    fun toProto(): SignalServiceProtos.DataMessage.Contact? {
         TODO("Not yet implemented")
     }
 }

libsession/src/main/java/org/session/libsession/messaging/messages/visible/LinkPreview.kt

@@ -3,7 +3,7 @@ package org.session.libsession.messaging.messages.visible
 import org.session.libsignal.libsignal.logging.Log
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 
-class LinkPreview() : VisibleMessageProto<SignalServiceProtos.DataMessage.Preview?>(){
+class LinkPreview() {
 
     var title: String? = null
     var url: String? = null
@@ -28,12 +28,11 @@ class LinkPreview() : VisibleMessageProto<SignalServiceProtos.DataMessage.Previe
 
 
     // validation
-    override fun isValid(): Boolean {
+    fun isValid(): Boolean {
-        if (!super.isValid()) return false
         return (title != null && url != null && attachmentID != null)
     }
 
-    override fun toProto(): SignalServiceProtos.DataMessage.Preview? {
+    fun toProto(): SignalServiceProtos.DataMessage.Preview? {
         val url = url
         if (url == null) {
             Log.w(TAG, "Couldn't construct link preview proto from: $this")

libsession/src/main/java/org/session/libsession/messaging/messages/visible/Profile.kt

@@ -4,7 +4,7 @@ import com.google.protobuf.ByteString
 import org.session.libsignal.libsignal.logging.Log
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 
-class Profile() : VisibleMessageProto<SignalServiceProtos.DataMessage?>() {
+class Profile() {
 
     var displayName: String? = null
     var profileKey: ByteArray? = null
@@ -19,7 +19,6 @@ class Profile() : VisibleMessageProto<SignalServiceProtos.DataMessage?>() {
             val profileKey = proto.profileKey
             val profilePictureURL = profileProto.profilePictureURL
             profileKey?.let {
-                val profilePictureURL = profilePictureURL
                 profilePictureURL?.let {
                     return Profile(displayName = displayName, profileKey = profileKey.toByteArray(), profilePictureURL = profilePictureURL)
                 }
@@ -35,7 +34,7 @@ class Profile() : VisibleMessageProto<SignalServiceProtos.DataMessage?>() {
         this.profilePictureURL = profilePictureURL
     }
 
-    override fun toProto(): SignalServiceProtos.DataMessage? {
+    fun toProto(): SignalServiceProtos.DataMessage? {
         val displayName = displayName
         if (displayName == null) {
             Log.w(TAG, "Couldn't construct link preview proto from: $this")

libsession/src/main/java/org/session/libsession/messaging/messages/visible/Quote.kt

@@ -3,7 +3,7 @@ package org.session.libsession.messaging.messages.visible
 import org.session.libsignal.libsignal.logging.Log
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 
-class Quote() : VisibleMessageProto<SignalServiceProtos.DataMessage.Quote?>() {
+class Quote() {
 
     var timestamp: Long? = 0
     var publicKey: String? = null
@@ -31,12 +31,11 @@ class Quote() : VisibleMessageProto<SignalServiceProtos.DataMessage.Quote?>() {
 
 
     // validation
-    override fun isValid(): Boolean {
+    fun isValid(): Boolean {
-        if (!super.isValid()) return false
         return (timestamp != null && publicKey != null)
     }
 
-    override fun toProto(): SignalServiceProtos.DataMessage.Quote? {
+    fun toProto(): SignalServiceProtos.DataMessage.Quote? {
         val timestamp = timestamp
         val publicKey = publicKey
         if (timestamp == null || publicKey == null) {

libsession/src/main/java/org/session/libsession/messaging/messages/visible/VisibleMessage.kt

@@ -1,9 +1,10 @@
 package org.session.libsession.messaging.messages.visible
 
+import org.session.libsession.messaging.messages.Message
 import org.session.libsignal.libsignal.logging.Log
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 
-class VisibleMessage() : VisibleMessageProto<SignalServiceProtos.Content?>() {
+class VisibleMessage : Message()  {
 
     var text: String? = null
     var attachmentIDs = ArrayList<String>()

libsession/src/main/java/org/session/libsession/messaging/messages/visible/VisibleMessageProto.kt

@@ -1,6 +0,0 @@
-package org.session.libsession.messaging.messages.visible
-
-import org.session.libsession.messaging.messages.Message
-
-abstract class VisibleMessageProto<T: com.google.protobuf.MessageOrBuilder?> : Message<T>() {
-}

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

@@ -1,7 +1,18 @@
 package org.session.libsession.messaging.sending_receiving
 
+import org.session.libsession.messaging.Configuration
+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.visible.VisibleMessage
+
+import org.session.libsignal.service.internal.push.SignalServiceProtos
+
 object MessageReceiver {
     internal sealed class Error(val description: String) : Exception() {
+        object DuplicateMessage: Error("Duplicate message.")
         object InvalidMessage: Error("Invalid message.")
         object UnknownMessage: Error("Unknown message type.")
         object UnknownEnvelopeType: Error("Unknown envelope type.")
@@ -26,4 +37,59 @@ object MessageReceiver {
             else -> true
         }
     }
+
+    internal fun parse(data: ByteArray, openGroupServerID: Long?): Pair<Message, SignalServiceProtos.Content> {
+        val storage = Configuration.shared.storage
+        val userPublicKey = storage.getUserPublicKey()
+        val isOpenGroupMessage = openGroupServerID != null
+        // Parse the envelope
+        val envelope = SignalServiceProtos.Envelope.parseFrom(data)
+        if (storage.getReceivedMessageTimestamps().contains(envelope.timestamp)) throw Error.DuplicateMessage
+        storage.addReceivedMessageTimestamp(envelope.timestamp)
+        // Decrypt the contents
+        val plaintext: ByteArray
+        val sender: String
+        var groupPublicKey: String? = null
+        if (isOpenGroupMessage) {
+            plaintext = envelope.content.toByteArray()
+            sender = envelope.source
+        } else {
+            when (envelope.type) {
+                SignalServiceProtos.Envelope.Type.UNIDENTIFIED_SENDER -> {
+                    val decryptionResult = MessageReceiverDecryption.decryptWithSignalProtocol(envelope)
+                    plaintext = decryptionResult.first()
+                    sender = decryptionResult.second()
+                }
+                SignalServiceProtos.Envelope.Type.CLOSED_GROUP_CIPHERTEXT -> {
+                    val decryptionResult = MessageReceiverDecryption.decryptWithSharedSenderKeys(envelope)
+                    plaintext = decryptionResult.first()
+                    sender = decryptionResult.second()
+                }
+                else -> throw Error.UnknownEnvelopeType
+            }
+        }
+        // Don't process the envelope any further if the sender is blocked
+        if (isBlock(sender)) throw Error.SenderBlocked
+        // Ignore self sends
+        if (sender == userPublicKey) throw Error.SelfSend
+        // Parse the proto
+        val proto = SignalServiceProtos.Content.parseFrom(plaintext)
+        // Parse the message
+        val message: Message = ReadReceipt.fromProto(proto) ?:
+                               TypingIndicator.fromProto(proto) ?:
+                               ClosedGroupUpdate.fromProto(proto) ?:
+                               ExpirationTimerUpdate.fromProto(proto) ?:
+                               VisibleMessage.fromProto(proto) ?: throw Error.UnknownMessage
+        if (isOpenGroupMessage && message !is VisibleMessage) throw Error.InvalidMessage
+        message.sender = sender
+        message.recipient = userPublicKey
+        message.sentTimestamp = envelope.timestamp
+        message.receivedTimestamp = System.currentTimeMillis()
+        message.groupPublicKey = groupPublicKey
+        message.openGroupServerMessageID = openGroupServerID
+        var isValid = message.isValid()
+        if (message is VisibleMessage && !isValid && proto.dataMessage.attachmentsCount == 0) { isValid = true }
+        if (!isValid) { throw Error.InvalidMessage }
+        return Pair(message, proto)
+    }
 }

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

@@ -1,35 +1,40 @@
 package org.session.libsession.messaging.sending_receiving
 
 import org.session.libsession.messaging.Configuration
-import org.session.libsignal.service.api.push.SignalServiceAddress
-import org.session.libsignal.service.loki.crypto.LokiServiceCipher
 import org.session.libsession.messaging.sending_receiving.MessageReceiver.Error
 import org.session.libsession.utilities.AESGCM
+
 import org.whispersystems.curve25519.Curve25519
+
 import org.session.libsignal.libsignal.loki.ClosedGroupCiphertextMessage
 import org.session.libsignal.libsignal.util.Pair
+import org.session.libsignal.service.api.crypto.SignalServiceCipher
 import org.session.libsignal.service.api.messages.SignalServiceEnvelope
+import org.session.libsignal.service.api.push.SignalServiceAddress
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 import org.session.libsignal.service.loki.protocol.closedgroups.SharedSenderKeysImplementation
 import org.session.libsignal.service.loki.utilities.toHexString
+
 import javax.crypto.Mac
 import javax.crypto.spec.SecretKeySpec
 
 object MessageReceiverDecryption {
 
-    internal fun decryptWithSignalProtocol(envelope: SignalServiceEnvelope): Pair<ByteArray, String> {
+    internal fun decryptWithSignalProtocol(envelope: SignalServiceProtos.Envelope): Pair<ByteArray, String> {
         val storage = Configuration.shared.signalStorage
-        val certificateValidator = Configuration.shared.certificateValidator
         val sskDatabase = Configuration.shared.sskDatabase
         val sessionResetImp = Configuration.shared.sessionResetImp
+        val certificateValidator = Configuration.shared.certificateValidator
         val data = envelope.content
         if (data.count() == 0) { throw Error.NoData }
         val userPublicKey = Configuration.shared.storage.getUserPublicKey() ?: throw Error.NoUserPublicKey
-        val cipher = LokiServiceCipher(SignalServiceAddress(userPublicKey), storage, sskDatabase, sessionResetImp, certificateValidator)
+        val localAddress = SignalServiceAddress(userPublicKey)
-        val result = cipher.decrypt(envelope)
+        val cipher = SignalServiceCipher(localAddress, storage, sskDatabase, sessionResetImp, certificateValidator)
+        val result = cipher.decrypt(SignalServiceEnvelope(envelope))
+        return Pair(result, result.sender)
     }
 
-    internal fun decryptWithSharedSenderKeys(envelope: SignalServiceEnvelope): Pair<ByteArray, String> {
+    internal fun decryptWithSharedSenderKeys(envelope: SignalServiceProtos.Envelope): Pair<ByteArray, String> {
         // 1. ) Check preconditions
         val groupPublicKey = envelope.source
         if (!Configuration.shared.storage.isClosedGroup(groupPublicKey)) { throw Error.InvalidGroupPublicKey }

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

@@ -1,4 +0,0 @@
-package org.session.libsession.messaging.sending_receiving
-
-interface MessageReceiverDelegate {
-}

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

@@ -0,0 +1,158 @@
+package org.session.libsession.messaging.sending_receiving
+
+import org.session.libsession.messaging.Configuration
+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.visible.VisibleMessage
+import org.session.libsession.messaging.sending_receiving.notifications.PushNotificationAPI
+import org.session.libsession.utilities.LKGroupUtilities
+import org.session.libsignal.libsignal.util.Hex
+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.toHexString
+import java.util.*
+
+internal fun MessageReceiver.isBlock(publicKey: String): Boolean {
+    // TODO: move isBlocked from Recipient to BlockManager
+    return false
+}
+
+fun MessageReceiver.handle(message: Message, proto: SignalServiceProtos.Content, openGroupID: String?) {
+    when (message) {
+        is ReadReceipt -> handleReadReceipt(message)
+        is TypingIndicator -> handleTypingIndicator(message)
+        is ClosedGroupUpdate -> handleClosedGroupUpdate(message)
+        is ExpirationTimerUpdate -> handleExpirationTimerUpdate(message)
+        is VisibleMessage -> handleVisibleMessage(message, proto, openGroupID)
+    }
+}
+
+private fun MessageReceiver.handleReadReceipt(message: ReadReceipt) {
+
+}
+
+private fun MessageReceiver.handleTypingIndicator(message: TypingIndicator) {
+    when (message.kind!!) {
+        TypingIndicator.Kind.STARTED -> showTypingIndicatorIfNeeded(message.sender!!)
+        TypingIndicator.Kind.STOPPED -> hideTypingIndicatorIfNeeded(message.sender!!)
+    }
+}
+
+fun MessageReceiver.showTypingIndicatorIfNeeded(senderPublicKey: String) {
+
+}
+
+fun MessageReceiver.hideTypingIndicatorIfNeeded(senderPublicKey: String) {
+
+}
+
+fun MessageReceiver.cancelTypingIndicatorsIfNeeded(senderPublicKey: String) {
+
+}
+
+private fun MessageReceiver.handleExpirationTimerUpdate(message: ExpirationTimerUpdate) {
+    if (message.duration!! > 0) {
+        setExpirationTimer(message.duration!!, message.sender!!, message.groupPublicKey)
+    } else {
+        disableExpirationTimer(message.sender!!, message.groupPublicKey)
+    }
+}
+
+fun MessageReceiver.setExpirationTimer(duration: Int, senderPublicKey: String, groupPublicKey: String?) {
+
+}
+
+fun MessageReceiver.disableExpirationTimer(senderPublicKey: String, groupPublicKey: String?) {
+
+}
+
+fun MessageReceiver.handleVisibleMessage(message: VisibleMessage, proto: SignalServiceProtos.Content, openGroupID: String?) {
+
+}
+
+private fun MessageReceiver.handleClosedGroupUpdate(message: ClosedGroupUpdate) {
+    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)
+    }
+}
+
+private fun MessageReceiver.handleNewGroup(message: ClosedGroupUpdate) {
+    val storage = Configuration.shared.storage
+    val sskDatabase = Configuration.shared.sskDatabase
+    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 = LKGroupUtilities.getEncodedClosedGroupIDAsData(groupPublicKey)
+    val groupDB = DatabaseFactory.getGroupDatabase(context)
+    if (groupDB.getGroup(groupID).orNull() != null) {
+        // Update the group
+        groupDB.updateTitle(groupID, name)
+        groupDB.updateMembers(groupID, members.map { Address.fromSerialized(it) })
+    } else {
+        groupDB.create(groupID, name, LinkedList<Address>(members.map { Address.fromSerialized(it) }),
+                null, null, LinkedList<Address>(admins.map { Address.fromSerialized(it) }))
+    }
+    DatabaseFactory.getRecipientDatabase(context).setProfileSharing(Recipient.from(context, Address.fromSerialized(groupID), false), 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(context, ClosedGroupOperation.Subscribe, groupPublicKey, userPublicKey)
+    // Notify the user
+    insertIncomingInfoMessage(context, senderPublicKey, groupID, SignalServiceProtos.GroupContext.Type.UPDATE, SignalServiceGroup.Type.UPDATE, name, members, admins)
+    // Establish sessions if needed
+    establishSessionsWithMembersIfNeeded(context, members)
+
+
+}
+
+private fun MessageReceiver.handleGroupUpdate(message: ClosedGroupUpdate) {
+
+}
+
+private fun MessageReceiver.handleSenderKeyRequest(message: ClosedGroupUpdate) {
+
+}
+
+private fun MessageReceiver.handleSenderKey(message: ClosedGroupUpdate) {
+
+}

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

@@ -1,5 +1,6 @@
 package org.session.libsession.messaging.sending_receiving
 
+import com.google.protobuf.MessageOrBuilder
 import nl.komponents.kovenant.Promise
 import nl.komponents.kovenant.deferred
 
@@ -17,6 +18,7 @@ import org.session.libsession.snode.SnodeAPI
 import org.session.libsession.snode.SnodeMessage
 
 import org.session.libsignal.libsignal.logging.Log
+import org.session.libsignal.service.api.messages.SignalServiceAttachment
 import org.session.libsignal.service.internal.push.SignalServiceProtos
 import org.session.libsignal.service.internal.util.Base64
 import org.session.libsignal.service.loki.api.crypto.ProofOfWork
@@ -45,6 +47,11 @@ object MessageSender {
         }
     }
 
+    // Preparation
+    fun prep(signalAttachments: List<SignalServiceAttachment>, message: VisibleMessage) {
+        // TODO: Deal with attachments
+    }
+
     // Convenience
     fun send(message: Message, destination: Destination): Promise<Unit, Exception> {
         if (destination is Destination.OpenGroup) {

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

@@ -0,0 +1,209 @@
+@file:Suppress("NAME_SHADOWING")
+
+package org.session.libsession.messaging.sending_receiving
+
+import android.content.Context
+import android.util.Log
+import nl.komponents.kovenant.Promise
+import nl.komponents.kovenant.deferred
+
+import org.session.libsession.messaging.Configuration
+import org.session.libsession.messaging.messages.Destination
+import org.session.libsession.messaging.messages.control.ClosedGroupUpdate
+import org.session.libsession.messaging.sending_receiving.notifications.PushNotificationAPI
+import org.session.libsession.utilities.LKGroupUtilities
+
+import org.session.libsignal.libsignal.ecc.Curve
+import org.session.libsignal.libsignal.util.Hex
+import org.session.libsignal.service.internal.push.SignalServiceProtos
+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.hexEncodedPrivateKey
+import org.session.libsignal.service.loki.utilities.hexEncodedPublicKey
+import java.util.*
+
+fun MessageSender.createClosedGroup(name: String, members: Collection<String>): Promise<String, Exception> {
+    val deferred = deferred<String, Exception>()
+    // Prepare
+    val members = members
+    val userPublicKey = Configuration.shared.storage.getUserPublicKey()!!
+    // Generate a key pair for the group
+    val groupKeyPair = Curve.generateKeyPair()
+    val groupPublicKey = groupKeyPair.hexEncodedPublicKey // Includes the "05" prefix
+    members.plus(userPublicKey)
+    val membersAsData = members.map { Hex.fromStringCondensed(it) }
+    // Create ratchets for all members
+    val senderKeys: List<ClosedGroupSenderKey> = members.map { publicKey ->
+        val ratchet = SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, publicKey)
+        ClosedGroupSenderKey(Hex.fromStringCondensed(ratchet.chainKey), ratchet.keyIndex, Hex.fromStringCondensed(publicKey))
+    }
+    // Create the group
+    val admins = setOf( userPublicKey )
+    val adminsAsData = admins.map { Hex.fromStringCondensed(it) }
+    val groupID = LKGroupUtilities.getEncodedClosedGroupIDAsData(groupPublicKey)
+    /* TODO:
+    DatabaseFactory.getGroupDatabase(context).create(groupID, name, LinkedList<Address>(members.map { Address.fromSerialized(it) }),
+             null, null, LinkedList<Address>(admins.map { Address.fromSerialized(it) }))
+    DatabaseFactory.getRecipientDatabase(context).setProfileSharing(Recipient.from(context, Address.fromSerialized(groupID), false), true)
+     */
+    // Send a closed group update message to all members using established channels
+    val promises = mutableListOf<Promise<Unit, Exception>>()
+    for (member in members) {
+        if (member == userPublicKey) { continue }
+        val closedGroupUpdateKind = ClosedGroupUpdate.Kind.New(Hex.fromStringCondensed(groupPublicKey), name, groupKeyPair.privateKey.serialize(),
+                senderKeys, membersAsData, adminsAsData)
+        val closedGroupUpdate = ClosedGroupUpdate()
+        closedGroupUpdate.kind = closedGroupUpdateKind
+        val promise = MessageSender.sendNonDurably(closedGroupUpdate, threadID)
+        promises.add(promise)
+    }
+    // Add the group to the user's set of public keys to poll for
+    Configuration.shared.sskDatabase.setClosedGroupPrivateKey(groupPublicKey, groupKeyPair.hexEncodedPrivateKey)
+    // Notify the PN server
+    PushNotificationAPI.performOperation(PushNotificationAPI.ClosedGroupOperation.Subscribe, groupPublicKey, userPublicKey)
+    // Notify the user
+    /* TODO
+    val threadID = DatabaseFactory.getThreadDatabase(context).getOrCreateThreadIdFor(Recipient.from(context, Address.fromSerialized(groupID), false))
+    insertOutgoingInfoMessage(context, groupID, GroupContext.Type.UPDATE, name, members, admins, threadID)
+     */
+    // Fulfill the promise
+    deferred.resolve(groupPublicKey)
+    // Return
+    return deferred.promise
+}
+
+fun MessageSender.update(groupPublicKey: String, members: Collection<String>, name: String): Promise<Unit, Exception> {
+    val deferred = deferred<Unit, Exception>()
+    val userPublicKey = Configuration.shared.storage.getUserPublicKey()!!
+    val sskDatabase = Configuration.shared.sskDatabase
+    val groupDB = DatabaseFactory.getGroupDatabase(context)
+    val groupID = LKGroupUtilities.getEncodedClosedGroupIDAsData(groupPublicKey)
+    val group = groupDB.getGroup(groupID).orNull()
+    if (group == null) {
+        Log.d("Loki", "Can't update nonexistent closed group.")
+        return deferred.reject(Error.NoThread)
+    }
+    val oldMembers = group.members.map { it.serialize() }.toSet()
+    val newMembers = members.minus(oldMembers)
+    val membersAsData = members.map { Hex.fromStringCondensed(it) }
+    val admins = group.admins.map { it.serialize() }
+    val adminsAsData = admins.map { Hex.fromStringCondensed(it) }
+    val groupPrivateKey = DatabaseFactory.getSSKDatabase(context).getClosedGroupPrivateKey(groupPublicKey)
+    if (groupPrivateKey == null) {
+        Log.d("Loki", "Couldn't get private key for closed group.")
+        return@Thread deferred.reject(Error.NoPrivateKey)
+    }
+    val wasAnyUserRemoved = members.toSet().intersect(oldMembers) != oldMembers.toSet()
+    val removedMembers = oldMembers.minus(members)
+    val isUserLeaving = removedMembers.contains(userPublicKey)
+    var newSenderKeys = listOf<ClosedGroupSenderKey>()
+    if (wasAnyUserRemoved) {
+        if (isUserLeaving && removedMembers.count() != 1) {
+            Log.d("Loki", "Can't remove self and others simultaneously.")
+            return@Thread deferred.reject(Error.InvalidUpdate)
+        }
+        // Establish sessions if needed
+        establishSessionsWithMembersIfNeeded(context, members)
+        // Send the update to the existing members using established channels (don't include new ratchets as everyone should regenerate new ratchets individually)
+        for (member in oldMembers) {
+            @Suppress("NAME_SHADOWING")
+            val closedGroupUpdateKind = ClosedGroupUpdateMessageSendJob.Kind.Info(Hex.fromStringCondensed(groupPublicKey),
+                    name, setOf(), membersAsData, adminsAsData)
+            @Suppress("NAME_SHADOWING")
+            val job = ClosedGroupUpdateMessageSendJob(member, closedGroupUpdateKind)
+            job.setContext(context)
+            job.onRun() // Run the job immediately
+        }
+        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)
+        }
+        // Delete all ratchets (it's important that this happens * after * sending out the update)
+        sskDatabase.removeAllClosedGroupRatchets(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
+        // Remove the group from the user's set of public keys to poll for if the user is leaving. Otherwise generate a new ratchet and
+        // send it out to all members (minus the removed ones) using established channels.
+        if (isUserLeaving) {
+            sskDatabase.removeClosedGroupPrivateKey(groupPublicKey)
+            groupDB.setActive(groupID, false)
+            groupDB.removeMember(groupID, Address.fromSerialized(userPublicKey))
+            // Notify the PN server
+            LokiPushNotificationManager.performOperation(context, ClosedGroupOperation.Unsubscribe, groupPublicKey, userPublicKey)
+        } else {
+            // Send closed group update messages to any new members using established channels
+            for (member in newMembers) {
+                @Suppress("NAME_SHADOWING")
+                val closedGroupUpdateKind = ClosedGroupUpdateMessageSendJob.Kind.New(Hex.fromStringCondensed(groupPublicKey), name,
+                        Hex.fromStringCondensed(groupPrivateKey), listOf(), membersAsData, adminsAsData)
+                @Suppress("NAME_SHADOWING")
+                val job = ClosedGroupUpdateMessageSendJob(member, closedGroupUpdateKind)
+                ApplicationContext.getInstance(context).jobManager.add(job)
+            }
+            // Send out the user's new ratchet to all members (minus the removed ones) using established channels
+            val userRatchet = SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, userPublicKey)
+            val userSenderKey = ClosedGroupSenderKey(Hex.fromStringCondensed(userRatchet.chainKey), userRatchet.keyIndex, Hex.fromStringCondensed(userPublicKey))
+            for (member in members) {
+                if (member == userPublicKey) { continue }
+                @Suppress("NAME_SHADOWING")
+                val closedGroupUpdateKind = ClosedGroupUpdateMessageSendJob.Kind.SenderKey(Hex.fromStringCondensed(groupPublicKey), userSenderKey)
+                @Suppress("NAME_SHADOWING")
+                val job = ClosedGroupUpdateMessageSendJob(member, closedGroupUpdateKind)
+                ApplicationContext.getInstance(context).jobManager.add(job)
+            }
+        }
+    } else if (newMembers.isNotEmpty()) {
+        // Generate ratchets for any new members
+        newSenderKeys = newMembers.map { publicKey ->
+            val ratchet = SharedSenderKeysImplementation.shared.generateRatchet(groupPublicKey, publicKey)
+            ClosedGroupSenderKey(Hex.fromStringCondensed(ratchet.chainKey), ratchet.keyIndex, Hex.fromStringCondensed(publicKey))
+        }
+        // Send a closed group update message to the existing members with the new members' ratchets (this message is aimed at the group)
+        val closedGroupUpdateKind = ClosedGroupUpdateMessageSendJob.Kind.Info(Hex.fromStringCondensed(groupPublicKey), name,
+                newSenderKeys, membersAsData, adminsAsData)
+        val job = ClosedGroupUpdateMessageSendJob(groupPublicKey, closedGroupUpdateKind)
+        ApplicationContext.getInstance(context).jobManager.add(job)
+        // Establish sessions if needed
+        establishSessionsWithMembersIfNeeded(context, newMembers)
+        // Send closed group update messages to the new members using established channels
+        var allSenderKeys = sskDatabase.getAllClosedGroupSenderKeys(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
+        allSenderKeys = allSenderKeys.union(newSenderKeys)
+        for (member in newMembers) {
+            @Suppress("NAME_SHADOWING")
+            val closedGroupUpdateKind = ClosedGroupUpdateMessageSendJob.Kind.New(Hex.fromStringCondensed(groupPublicKey), name,
+                    Hex.fromStringCondensed(groupPrivateKey), allSenderKeys, membersAsData, adminsAsData)
+            @Suppress("NAME_SHADOWING")
+            val job = ClosedGroupUpdateMessageSendJob(member, closedGroupUpdateKind)
+            ApplicationContext.getInstance(context).jobManager.add(job)
+        }
+    } else {
+        val allSenderKeys = sskDatabase.getAllClosedGroupSenderKeys(groupPublicKey, ClosedGroupRatchetCollectionType.Current)
+        val closedGroupUpdateKind = ClosedGroupUpdateMessageSendJob.Kind.Info(Hex.fromStringCondensed(groupPublicKey), name,
+                allSenderKeys, membersAsData, adminsAsData)
+        val job = ClosedGroupUpdateMessageSendJob(groupPublicKey, closedGroupUpdateKind)
+        ApplicationContext.getInstance(context).jobManager.add(job)
+    }
+    // Update the group
+    groupDB.updateTitle(groupID, name)
+    if (!isUserLeaving) {
+        // The call below sets isActive to true, so if the user is leaving we have to use groupDB.remove(...) instead
+        groupDB.updateMembers(groupID, members.map { Address.fromSerialized(it) })
+    }
+    // Notify the user
+    val infoType = if (isUserLeaving) SignalServiceProtos.GroupContext.Type.QUIT else SignalServiceProtos.GroupContext.Type.UPDATE
+    val threadID = DatabaseFactory.getThreadDatabase(context).getOrCreateThreadIdFor(Recipient.from(context, Address.fromSerialized(groupID), false))
+    insertOutgoingInfoMessage(context, groupID, infoType, name, members, admins, threadID)
+    deferred.resolve(Unit)
+    return deferred.promise
+}
+
+fun MessageSender.requestSenderKey(groupPublicKey: String, senderPublicKey: String) {
+    Log.d("Loki", "Requesting sender key for group public key: $groupPublicKey, sender public key: $senderPublicKey.")
+    // Send the request
+    val closedGroupUpdateKind = ClosedGroupUpdate.Kind.SenderKeyRequest(Hex.fromStringCondensed(groupPublicKey))
+    val closedGroupUpdate = ClosedGroupUpdate()
+    closedGroupUpdate.kind = closedGroupUpdateKind
+    MessageSender.send(closedGroupUpdate, Destination.ClosedGroup(groupPublicKey))
+}

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

@@ -0,0 +1,35 @@
+package org.session.libsession.messaging.sending_receiving
+
+import nl.komponents.kovenant.Promise
+
+import org.session.libsession.messaging.jobs.JobQueue
+import org.session.libsession.messaging.jobs.MessageSendJob
+import org.session.libsession.messaging.messages.Destination
+import org.session.libsession.messaging.messages.Message
+import org.session.libsession.messaging.messages.visible.VisibleMessage
+
+import org.session.libsignal.service.api.messages.SignalServiceAttachment
+
+fun MessageSender.send(message: VisibleMessage, attachments: List<SignalServiceAttachment>, threadID: String) {
+    prep(attachments, message)
+    send(message, threadID)
+}
+
+fun MessageSender.send(message: Message, threadID: String) {
+    message.threadID = threadID
+    val destination = Destination.from(threadID)
+    val job = MessageSendJob(message, destination)
+    JobQueue.shared.add(job)
+}
+
+fun MessageSender.sendNonDurably(message: VisibleMessage, attachments: List<SignalServiceAttachment>, threadID: String): Promise<Unit, Exception> {
+    prep(attachments, message)
+    // TODO: Deal with attachments
+    return sendNonDurably(message, threadID)
+}
+
+fun MessageSender.sendNonDurably(message: Message, threadID: String): Promise<Unit, Exception> {
+    message.threadID = threadID
+    val destination = Destination.from(threadID)
+    return MessageSender.send(message, destination)
+}

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

@@ -1,4 +0,0 @@
-package org.session.libsession.messaging.sending_receiving
-
-interface MessageSenderDelegate {
-}

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

@@ -3,6 +3,7 @@ package org.session.libsession.messaging.sending_receiving.notifications
 import android.content.Context
 import nl.komponents.kovenant.functional.map
 import okhttp3.*
+import org.session.libsession.messaging.Configuration
 import org.session.libsignal.libsignal.logging.Log
 import org.session.libsignal.service.internal.util.JsonUtil
 import org.session.libsignal.service.loki.api.onionrequests.OnionRequestAPI
@@ -27,7 +28,7 @@ object PushNotificationAPI {
             }
     }
 
-    fun unregister(token: String, context: Context) {
+    fun unregister(token: String) {
         val parameters = mapOf( "token" to token )
         val url = "$server/unregister"
         val body = RequestBody.create(MediaType.get("application/json"), JsonUtil.toJson(parameters))
@@ -45,14 +46,14 @@ object PushNotificationAPI {
             }
         }
         // Unsubscribe from all closed groups
-        val allClosedGroupPublicKeys = DatabaseFactory.getSSKDatabase(context).getAllClosedGroupPublicKeys()
+        val allClosedGroupPublicKeys = Configuration.shared.sskDatabase.getAllClosedGroupPublicKeys()
-        val userPublicKey = TextSecurePreferences.getLocalNumber(context)
+        val userPublicKey = Configuration.shared.storage.getUserPublicKey()!!
         allClosedGroupPublicKeys.forEach { closedGroup ->
-            performOperation(context, ClosedGroupOperation.Unsubscribe, closedGroup, userPublicKey)
+            performOperation(ClosedGroupOperation.Unsubscribe, closedGroup, userPublicKey)
         }
     }
 
-    fun register(token: String, publicKey: String, context: Context, force: Boolean) {
+    fun register(token: String, publicKey: String, force: Boolean) {
         val oldToken = TextSecurePreferences.getFCMToken(context)
         val lastUploadDate = TextSecurePreferences.getLastFCMUploadTime(context)
         if (!force && token == oldToken && System.currentTimeMillis() - lastUploadDate < tokenExpirationInterval) { return }
@@ -75,13 +76,13 @@ object PushNotificationAPI {
             }
         }
         // Subscribe to all closed groups
-        val allClosedGroupPublicKeys = DatabaseFactory.getSSKDatabase(context).getAllClosedGroupPublicKeys()
+        val allClosedGroupPublicKeys = Configuration.shared.sskDatabase.getAllClosedGroupPublicKeys()
         allClosedGroupPublicKeys.forEach { closedGroup ->
-            performOperation(context, ClosedGroupOperation.Subscribe, closedGroup, publicKey)
+            performOperation(ClosedGroupOperation.Subscribe, closedGroup, publicKey)
         }
     }
 
-    fun performOperation(context: Context, operation: ClosedGroupOperation, closedGroupPublicKey: String, publicKey: String) {
+    fun performOperation(operation: ClosedGroupOperation, closedGroupPublicKey: String, publicKey: String) {
         if (!TextSecurePreferences.isUsingFCM(context)) { return }
         val parameters = mapOf( "closedGroupPublicKey" to closedGroupPublicKey, "pubKey" to publicKey )
         val url = "$server/${operation.rawValue}"

libsession/src/main/java/org/session/libsession/messaging/utilities/MessageWrapper.kt

@@ -71,11 +71,11 @@ object MessageWrapper {
     /**
      * `data` shouldn't be base 64 encoded.
      */
-    fun unwrap(data: ByteArray): Envelope {
+    fun unwrap(data: ByteArray): ByteArray {
         try {
             val webSocketMessage = WebSocketMessage.parseFrom(data)
             val envelopeAsData = webSocketMessage.request.body
-            return Envelope.parseFrom(envelopeAsData)
+            return envelopeAsData.toByteArray()
         } catch (e: Exception) {
             Log.d("Loki", "Failed to unwrap data: ${e.message}.")
             throw Error.FailedToUnwrapData

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

@@ -0,0 +1,47 @@
+package org.session.libsession.utilities
+
+object LKGroupUtilities {
+    const val CLOSED_GROUP_PREFIX = "__textsecure_group__!"
+    const val MMS_GROUP_PREFIX = "__signal_mms_group__!"
+    const val OPEN_GROUP_PREFIX = "__loki_public_chat_group__!"
+
+    fun getEncodedOpenGroupID(groupID: String): String {
+        return OPEN_GROUP_PREFIX + groupID
+    }
+
+    fun getEncodedOpenGroupIDAsData(groupID: String): ByteArray {
+        return (OPEN_GROUP_PREFIX + groupID).toByteArray()
+    }
+
+    fun getEncodedClosedGroupID(groupID: String): String {
+        return CLOSED_GROUP_PREFIX + groupID
+    }
+
+    fun getEncodedClosedGroupIDAsData(groupID: String): ByteArray {
+        return (CLOSED_GROUP_PREFIX + groupID).toByteArray()
+    }
+
+    fun getEncodedMMSGroupID(groupID: String): String {
+        return MMS_GROUP_PREFIX + groupID
+    }
+
+    fun getEncodedMMSGroupIDAsData(groupID: String): ByteArray {
+        return (MMS_GROUP_PREFIX + groupID).toByteArray()
+    }
+
+    fun getEncodedGroupID(groupID: ByteArray): String {
+        return groupID.toString()
+    }
+
+    fun getDecodedGroupID(groupID: ByteArray): String {
+        val encodedGroupID = groupID.toString()
+        if (encodedGroupID.split("!").count() > 1) {
+            return encodedGroupID.split("!")[1]
+        }
+        return encodedGroupID.split("!")[0]
+    }
+
+    fun getDecodedGroupIDAsData(groupID: ByteArray): ByteArray {
+        return getDecodedGroupID(groupID).toByteArray()
+    }
+}