signalapp/Signal-iOS
signalapp/Signal-iOS
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 4 Wiki Activity

Break envelope processing into two parts (#4340)

Browse Source 
The purpose of this change is to alleviate the
main source of CPU-grinding after sending a
message to a large group: about 20% of the total
time is spent decoding and re-encoding messages as
they are fetched from the DB and then updated with
info about which recipients have acknowledged
delivery.

Prior to this commit, we decrypted the payload and
then processed the message in a single rather deep
call tree that began in `processNextBatch`. There
are two completely different paths through this
tree that lead to handling a delivery receipt
because server-generated delivery receipts have
the necessary info in plaintext in the envelope
and client-generated delivery receipts have it in
the ciphertext.

The idea here is to break incoming envelope
handling into two parts.

In the first part, each envelope is decrypted and
other common activities (like handling sender key
distribution messages) are performed. A processing
request is created for each. Next, processing
requests are handled. It's easy to examine a
processing request to determine if it is a
delivery receipt, and if so, for what outgoing
message. Doing so allows the new
DeliveryReceiptContext class to cache message
fetches and combine updates to the same message.

Processing requests are grouped together so that
sequential delivery receipt requests enjoy caching
of messages and coalescing of updates into a
single fetch/decode/encode/commit. Other kinds of
envelopes are handled immediately to avoid
increasing memory pressure that caching multiple
messages could cause.
This commit is contained in:
george-signal 2022-06-09 16:40:34 -07:00 committed by GitHub
parent d4f290954e
commit ec66f3b21e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1317 additions and 490 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Pods

@ -1 +1 @@
Subproject commit c1cd4f10255e0d4265cdc342981c4baf858d6441
Subproject commit 830d90c416c66d20d9e65d812784cb29e3d85bf6

2
Signal/src/ViewControllers/DebugUI/DebugUIMessages.m
View File

@ -4282,6 +4282,7 @@ typedef OWSContact * (^OWSContactBlock)(SDSAnyWriteTransaction *transaction);
[message updateWithDeliveredRecipient:address
recipientDeviceId:0
deliveryTimestamp:timestamp
context:[[PassthroughDeliveryReceiptContext alloc] init]
transaction:transaction];
[message updateWithReadRecipient:address
recipientDeviceId:0
@ -4584,6 +4585,7 @@ typedef OWSContact * (^OWSContactBlock)(SDSAnyWriteTransaction *transaction);
[message updateWithDeliveredRecipient:address
recipientDeviceId:0
deliveryTimestamp:@([NSDate ows_millisecondTimeStamp])
context:[[PassthroughDeliveryReceiptContext alloc] init]
transaction:transaction];
}
}

143
SignalServiceKit/src/Messages/DeliveryReceiptContext.swift Normal file
View File

@ -0,0 +1,143 @@
//
// Copyright (c) 2022 Open Whisper Systems. All rights reserved.
//
import Foundation
@objc
protocol DeliveryReceiptContext: AnyObject {
@objc(addUpdateForMessage:transaction:update:)
func addUpdate(message: TSOutgoingMessage,
transaction: SDSAnyWriteTransaction,
update: @escaping (TSOutgoingMessage) -> ())
@objc(messagesWithTimestamp:transaction:)
func messages(_ timestamps: UInt64,
transaction: SDSAnyReadTransaction) -> [TSOutgoingMessage]
}
fileprivate struct Update {
let message: TSOutgoingMessage
let update: (TSOutgoingMessage) -> ()
}
fileprivate extension TSOutgoingMessage {
static func fetch(_ timestamp: UInt64,
transaction: SDSAnyReadTransaction) -> [TSOutgoingMessage] {
var messages = [TSOutgoingMessage]()
do {
let fetched = try InteractionFinder.interactions(withTimestamp: timestamp,
filter: { _ in true },
transaction: transaction).compactMap { $0 as? TSOutgoingMessage }
messages.append(contentsOf: fetched)
} catch {
owsFailDebug("Error loading interactions: \(error)")
}
return messages
}
}
@objc
public class PassthroughDeliveryReceiptContext: NSObject, DeliveryReceiptContext {
@objc(addUpdateForMessage:transaction:update:)
func addUpdate(message: TSOutgoingMessage,
transaction: SDSAnyWriteTransaction,
update: @escaping (TSOutgoingMessage) -> ()) {
let deferredUpdate = Update(message: message, update: update)
message.anyUpdateOutgoingMessage(transaction: transaction) { message in
deferredUpdate.update(message)
}
}
@objc(messagesWithTimestamp:transaction:)
func messages(_ timestamp: UInt64,
transaction: SDSAnyReadTransaction) -> [TSOutgoingMessage] {
return TSOutgoingMessage.fetch(timestamp, transaction: transaction)
}
}
public class BatchingDeliveryReceiptContext: NSObject, DeliveryReceiptContext {
private var messages = [UInt64: [TSOutgoingMessage]]()
private var deferredUpdates: [Update] = []
#if TESTABLE_BUILD
static var didRunDeferredUpdates: ((Int, SDSAnyWriteTransaction) -> ())? = nil
#endif
private override init() {
super.init()
}
static func withDeferredUpdates(transaction: SDSAnyWriteTransaction, _ closure: (DeliveryReceiptContext) -> ()) {
let instance = BatchingDeliveryReceiptContext()
closure(instance)
instance.runDeferredUpdates(transaction: transaction)
}
// Adds a closure to run that mutates a message. Note that it will be run twice - once for the
// in-memory instance and a second time for the most up-to-date copy in the database.
@objc(addUpdateForMessage:transaction:update:)
func addUpdate(message: TSOutgoingMessage,
transaction: SDSAnyWriteTransaction,
update: @escaping (TSOutgoingMessage) -> ()) {
deferredUpdates.append(Update(message: message, update: update))
}
@objc(messagesWithTimestamp:transaction:)
func messages(_ timestamp: UInt64, transaction: SDSAnyReadTransaction) -> [TSOutgoingMessage] {
if let result = messages[timestamp] {
return result
}
let fetched = TSOutgoingMessage.fetch(timestamp, transaction: transaction)
messages[timestamp] = fetched
return fetched
}
private struct UpdateCollection {
private var message: TSOutgoingMessage? = nil
private var closures = [(TSOutgoingMessage) -> ()]()
mutating func addOrExecute(update: Update,
transaction: SDSAnyWriteTransaction) {
if message?.grdbId != update.message.grdbId {
execute(transaction: transaction)
message = update.message
}
owsAssertDebug(message != nil)
closures.append(update.update)
}
mutating func execute(transaction: SDSAnyWriteTransaction) {
guard let message = message else {
owsAssertDebug(closures.isEmpty)
return
}
message.anyUpdateOutgoingMessage(transaction: transaction) { _ in
for closure in closures {
closure(message)
}
}
self.message = nil
closures = []
}
}
private func runDeferredUpdates(transaction: SDSAnyWriteTransaction) {
var updateCollection = UpdateCollection()
#if TESTABLE_BUILD
let count = deferredUpdates.count
#endif
while let update = deferredUpdates.first {
deferredUpdates.removeFirst()
updateCollection.addOrExecute(update: update, transaction: transaction)
}
updateCollection.execute(transaction: transaction)
#if TESTABLE_BUILD
let closure = Self.didRunDeferredUpdates
Self.didRunDeferredUpdates = nil
closure?(count, transaction)
#endif
}
}

1
SignalServiceKit/src/Messages/EarlyMessageManager.swift
View File

@ -388,6 +388,7 @@ public class EarlyMessageManager: NSObject {
withDeliveredRecipient: sender,
recipientDeviceId: deviceId,
deliveryTimestamp: NSNumber(value: timestamp),
context: PassthroughDeliveryReceiptContext(),
transaction: transaction
)
case .messageReadOnLinkedDevice(let timestamp):

3
SignalServiceKit/src/Messages/Interactions/TSOutgoingMessage.h
View File

@ -57,6 +57,8 @@ typedef NS_ENUM(NSInteger, TSGroupMetaMessage) {
TSGroupMetaMessageRequestInfo,
};
@protocol DeliveryReceiptContext;
@class SDSAnyWriteTransaction;
@class SSKProtoAttachmentPointer;
@class SSKProtoContentBuilder;
@ -318,6 +320,7 @@ NS_DESIGNATED_INITIALIZER NS_SWIFT_NAME(init(grdbId:uniqueId:receivedAtTimestamp
- (void)updateWithDeliveredRecipient:(SignalServiceAddress *)recipientAddress
recipientDeviceId:(uint32_t)deviceId
deliveryTimestamp:(NSNumber *_Nullable)deliveryTimestamp
context:(id<DeliveryReceiptContext>)deliveryReceiptContext
transaction:(SDSAnyWriteTransaction *)transaction;
- (void)updateWithWasSentFromLinkedDeviceWithUDRecipientAddresses:

34
SignalServiceKit/src/Messages/Interactions/TSOutgoingMessage.m
View File

@ -756,6 +756,7 @@ NSUInteger const TSOutgoingMessageSchemaVersion = 1;
- (void)updateWithDeliveredRecipient:(SignalServiceAddress *)recipientAddress
recipientDeviceId:(uint32_t)deviceId
deliveryTimestamp:(NSNumber *_Nullable)deliveryTimestamp
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction
{
OWSAssertDebug(recipientAddress.isValid);
@ -781,22 +782,23 @@ NSUInteger const TSOutgoingMessageSchemaVersion = 1;
deviceId:deviceId
transaction:transaction];
[self anyUpdateOutgoingMessageWithTransaction:transaction
block:^(TSOutgoingMessage *message) {
TSOutgoingMessageRecipientState *_Nullable recipientState
= message.recipientAddressStates[recipientAddress];
if (!recipientState) {
OWSFailDebug(@"Missing recipient state for delivered recipient: %@",
recipientAddress);
return;
}
if (recipientState.state != OWSOutgoingMessageRecipientStateSent) {
OWSLogWarn(@"marking unsent message as delivered.");
}
recipientState.state = OWSOutgoingMessageRecipientStateSent;
recipientState.deliveryTimestamp = deliveryTimestamp;
recipientState.errorCode = nil;
}];
[context
addUpdateForMessage:self
transaction:transaction
update:^(TSOutgoingMessage *message) {
TSOutgoingMessageRecipientState *_Nullable recipientState
= message.recipientAddressStates[recipientAddress];
if (!recipientState) {
OWSFailDebug(@"Missing recipient state for delivered recipient: %@", recipientAddress);
return;
}
if (recipientState.state != OWSOutgoingMessageRecipientStateSent) {
OWSLogWarn(@"marking unsent message as delivered.");
}
recipientState.state = OWSOutgoingMessageRecipientStateSent;
recipientState.deliveryTimestamp = deliveryTimestamp;
recipientState.errorCode = nil;
}];
}
- (void)updateWithReadRecipient:(SignalServiceAddress *)recipientAddress

514
SignalServiceKit/src/Messages/MessageProcessor.swift
View File

@ -130,40 +130,40 @@ public class MessageProcessor: NSObject {
}
public func processEncryptedEnvelopeData(
_ encryptedEnvelopeData: Data,
_ envelopeData: Data,
serverDeliveryTimestamp: UInt64,
envelopeSource: EnvelopeSource,
completion: @escaping (Error?) -> Void
) {
guard !encryptedEnvelopeData.isEmpty else {
guard !envelopeData.isEmpty else {
completion(OWSAssertionError("Empty envelope, envelopeSource: \(envelopeSource)."))
return
}
// Drop any too-large messages on the floor. Well behaving clients should never send them.
guard encryptedEnvelopeData.count <= Self.maxEnvelopeByteCount else {
guard envelopeData.count <= Self.maxEnvelopeByteCount else {
completion(OWSAssertionError("Oversize envelope, envelopeSource: \(envelopeSource)."))
return
}
// Take note of any messages larger than we expect, but still process them.
// This likely indicates a misbehaving sending client.
if encryptedEnvelopeData.count > Self.largeEnvelopeWarningByteCount {
Logger.verbose("encryptedEnvelopeData: \(encryptedEnvelopeData.count) > : \(Self.largeEnvelopeWarningByteCount)")
if envelopeData.count > Self.largeEnvelopeWarningByteCount {
Logger.verbose("encryptedEnvelopeData: \(envelopeData.count) > : \(Self.largeEnvelopeWarningByteCount)")
owsFailDebug("Unexpectedly large envelope, envelopeSource: \(envelopeSource).")
}
let encryptedEnvelopeProto: SSKProtoEnvelope
let protoEnvelope: SSKProtoEnvelope
do {
encryptedEnvelopeProto = try SSKProtoEnvelope(serializedData: encryptedEnvelopeData)
protoEnvelope = try SSKProtoEnvelope(serializedData: envelopeData)
} catch {
owsFailDebug("Failed to parse encrypted envelope \(error), envelopeSource: \(envelopeSource)")
completion(error)
return
}
processEncryptedEnvelope(EncryptedEnvelope(encryptedEnvelopeData: encryptedEnvelopeData,
encryptedEnvelope: encryptedEnvelopeProto,
processEncryptedEnvelope(EncryptedEnvelope(encryptedEnvelopeData: envelopeData,
encryptedEnvelope: protoEnvelope,
serverDeliveryTimestamp: serverDeliveryTimestamp,
completion: completion),
envelopeSource: envelopeSource)
@ -274,14 +274,17 @@ public class MessageProcessor: NSObject {
var processedEnvelopes: [PendingEnvelope] = []
SDSDatabaseStorage.shared.write { transaction in
for envelope in batchEnvelopes {
if messagePipelineSupervisor.isMessageProcessingPermitted {
self.processEnvelope(envelope, transaction: transaction)
processedEnvelopes.append(envelope)
} else {
// If we're skipping one message, we have to skip them all to preserve ordering
// Next time around we can process the skipped messages in order
break
var remainingEnvelopes = batchEnvelopes
while messagePipelineSupervisor.isMessageProcessingPermitted && !remainingEnvelopes.isEmpty {
autoreleasepool {
let combinedRequest = buildNextCombinedRequest(envelopes: &remainingEnvelopes,
transaction: transaction)
let processed = handle(combinedRequest: combinedRequest,
transaction: transaction)
if processed {
let envelopes = combinedRequest.processingRequests.lazy.map { $0.pendingEnvelope }
processedEnvelopes.append(contentsOf: envelopes)
}
}
}
}
@ -292,138 +295,92 @@ public class MessageProcessor: NSObject {
}
}
private func processEnvelope(_ pendingEnvelope: PendingEnvelope, transaction: SDSAnyWriteTransaction) {
assertOnQueue(serialQueue)
switch pendingEnvelope.decrypt(transaction: transaction) {
case .success(let result):
// NOTE: We use the envelope from the decrypt result, not the pending envelope,
// since the envelope may be altered by the decryption process in the UD case.
guard let sourceAddress = result.envelope.sourceAddress, sourceAddress.isValid else {
owsFailDebug("Successful decryption with no source address; discarding message")
transaction.addAsyncCompletionOffMain {
pendingEnvelope.completion(OWSAssertionError("successful decryption with no source address"))
}
return
}
// Pre-processing happens during the same transaction that performed decryption
messageManager.preprocessEnvelope(envelope: result.envelope,
plaintext: result.plaintextData,
transaction: transaction)
// If the sender is in the block list, we can skip scheduling any additional processing.
if blockingManager.isAddressBlocked(sourceAddress, transaction: transaction) {
Logger.info("Skipping processing for blocked envelope: \(sourceAddress)")
let error = MessageProcessingError.blockedSender
transaction.addAsyncCompletionOffMain {
pendingEnvelope.completion(error)
}
return
}
if result.identity == .pni {
identityManager.setShouldSharePhoneNumber(with: sourceAddress, transaction: transaction)
}
enum ProcessingStep {
case discard
case enqueueForGroupProcessing
case processNow(shouldDiscardVisibleMessages: Bool)
}
let processingStep = { () -> ProcessingStep in
guard let plaintextData = result.plaintextData,
let groupContextV2 =
GroupsV2MessageProcessor.groupContextV2(fromPlaintextData: plaintextData) else {
// Non-v2-group messages can be processed immediately.
return .processNow(shouldDiscardVisibleMessages: false)
}
guard GroupsV2MessageProcessor.canContextBeProcessedImmediately(
groupContext: groupContextV2,
transaction: transaction
) else {
// Some v2 group messages required group state to be
// updated before they can be processed.
return .enqueueForGroupProcessing
}
let discardMode = GroupsMessageProcessor.discardMode(forMessageFrom: sourceAddress,
groupContext: groupContextV2,
transaction: transaction)
if discardMode == .discard {
// Some v2 group messages should be discarded and not processed.
Logger.verbose("Discarding job.")
return .discard
}
// Some v2 group messages should be processed, but
// discarding any "visible" messages, e.g. text messages
// or calls.
return .processNow(shouldDiscardVisibleMessages: discardMode == .discardVisibleMessages)
}()
switch processingStep {
case .discard:
// Do nothing.
Logger.verbose("Discarding job.")
case .enqueueForGroupProcessing:
// If we can't process the message immediately, we enqueue it for
// for processing in the same transaction within which it was decrypted
// to prevent data loss.
let envelopeData: Data
if let existingEnvelopeData = result.envelopeData {
envelopeData = existingEnvelopeData
} else {
do {
envelopeData = try result.envelope.serializedData()
} catch {
owsFailDebug("failed to reserialize envelope: \(error)")
transaction.addAsyncCompletionOffMain {
pendingEnvelope.completion(error)
}
return
}
}
Self.groupsV2MessageProcessor.enqueue(
envelopeData: envelopeData,
// All GV2 messages have plaintext data (because that's where the group context lives)
plaintextData: result.plaintextData!,
wasReceivedByUD: result.wasReceivedByUD,
serverDeliveryTimestamp: result.serverDeliveryTimestamp,
transaction: transaction
)
case .processNow(let shouldDiscardVisibleMessages):
// Envelopes can be processed immediately if they're:
// 1. Not a GV2 message.
// 2. A GV2 message that doesn't require updating the group.
//
// The advantage to processing the message immediately is that
// we can full process the message in the same transaction that
// we used to decrypt it. This results in a significant perf
// benefit verse queueing the message and waiting for that queue
// to open new transactions and process messages. The downside is
// that if we *fail* to process this message (e.g. the app crashed
// or was killed), we'll have to re-decrypt again before we process.
// This is safe, since the decrypt operation would also be rolled
// back (since the transaction didn't finalize) and should be rare.
Self.messageManager.processEnvelope(
result.envelope,
plaintextData: result.plaintextData,
wasReceivedByUD: result.wasReceivedByUD,
serverDeliveryTimestamp: result.serverDeliveryTimestamp,
shouldDiscardVisibleMessages: shouldDiscardVisibleMessages,
transaction: transaction
)
}
transaction.addAsyncCompletionOffMain {
pendingEnvelope.completion(nil)
}
case .failure(let error):
transaction.addAsyncCompletionOffMain {
pendingEnvelope.completion(error)
// If envelopes is not empty, this will emit a single request for a non-delivery receipt or one or more requests
// all for delivery receipts.
private func buildNextCombinedRequest(envelopes: inout [PendingEnvelope],
transaction: SDSAnyWriteTransaction) -> RelatedProcessingRequests {
let result = RelatedProcessingRequests()
while let envelope = envelopes.first {
envelopes.removeFirst()
let request = processingRequest(for: envelope,
messageManager: Self.messageManager,
blockingManager: Self.blockingManager,
transaction: transaction)
result.add(request)
if request.deliveryReceiptMessageTimestamps == nil {
// If we hit a non-delivery receipt envelope, handle it immediately to avoid
// keeping potentially large decrypted envelopes in memory.
break
}
}
return result
}
private func handle(combinedRequest: RelatedProcessingRequests,
transaction: SDSAnyWriteTransaction) -> Bool {
guard messagePipelineSupervisor.isMessageProcessingPermitted else {
// If we're skipping one message, we have to skip them all to preserve ordering
// Next time around we can process the skipped messages in order
return false
}
Logger.info("Process \(combinedRequest.processingRequests.count) related requests")
// Efficiently handle delivery receipts for the same message by fetching the sent message only
// once and only using one transaction to update the message with new recipient state.
// If `relation` doesn't contain a bunch of delivery receipts, handle the envelopes individually.
BatchingDeliveryReceiptContext.withDeferredUpdates(transaction: transaction) { context in
for request in combinedRequest.processingRequests {
if handleProcessingRequest(request, context: context, transaction: transaction),
let protoEnvelope = request.protoEnvelope {
messageManager.finishProcessingEnvelope(protoEnvelope,
transaction: transaction)
}
}
}
return true
}
private func reallyHandleProcessingRequest(_ request: ProcessingRequest,
context: DeliveryReceiptContext,
transaction: SDSAnyWriteTransaction) -> (Bool, Error?) {
switch request.state {
case .completed(error: let error):
Logger.info("Envelope completed early with error \(String(describing: error))")
return (false, error)
case .enqueueForGroup(decryptedEnvelope: let decryptedEnvelope, envelopeData: let envelopeData):
Logger.info("Enqueue group envelope")
Self.groupsV2MessageProcessor.enqueue(
envelopeData: envelopeData,
// All GV2 messages have plaintext data (because that's where the group context lives)
plaintextData: decryptedEnvelope.plaintextData!,
wasReceivedByUD: decryptedEnvelope.wasReceivedByUD,
serverDeliveryTimestamp: decryptedEnvelope.serverDeliveryTimestamp,
transaction: transaction)
return (false, nil)
case .messageManagerRequest(let messageManagerRequest):
Logger.info("Pass envelope to message manager")
messageManager.handle(messageManagerRequest, context: context, transaction: transaction)
return (true, nil)
case .plaintextReceipt(let protoEnvelope):
Logger.info("Handle server-generated receipt")
messageManager.handleDeliveryReceipt(protoEnvelope,
context: context,
transaction: transaction)
return (true, nil)
case .clearPlaceholdersOnly:
return (true, nil)
}
}
private func handleProcessingRequest(_ request: ProcessingRequest,
context: DeliveryReceiptContext,
transaction: SDSAnyWriteTransaction) -> Bool {
let (shouldFinishProcessing, maybeError) = reallyHandleProcessingRequest(request,
context: context,
transaction: transaction)
transaction.addAsyncCompletionOffMain {
request.pendingEnvelope.completion(maybeError)
}
return shouldFinishProcessing
}
@objc
@ -461,6 +418,265 @@ public class MessageProcessor: NSObject {
}
}
struct ProcessingRequest {
enum State {
case completed(error: Error?)
case enqueueForGroup(decryptedEnvelope: DecryptedEnvelope,
envelopeData: Data)
case messageManagerRequest(MessageManagerRequest)
case plaintextReceipt(SSKProtoEnvelope)
// Message decrypted but had an invalid protobuf.
case clearPlaceholdersOnly(SSKProtoEnvelope)
}
let state: State
let pendingEnvelope: PendingEnvelope
var protoEnvelope: SSKProtoEnvelope? {
switch state {
case .completed, .enqueueForGroup:
return nil
case .messageManagerRequest(let messageManagerRequest):
return messageManagerRequest.envelope
case .plaintextReceipt(let protoEnvelope), .clearPlaceholdersOnly(let protoEnvelope):
return protoEnvelope
}
}
// If this request is for a delivery receipt, return the timestamps for the sent-messages it
// corresponds to.
var deliveryReceiptMessageTimestamps: [UInt64]? {
switch state {
case .completed, .enqueueForGroup, .clearPlaceholdersOnly:
return nil
case .plaintextReceipt(let protoEnvelope):
return [protoEnvelope.timestamp]
case .messageManagerRequest(let request):
switch request.messageType {
case .receiptMessage:
guard let receiptMessage = request.protoContent?.receiptMessage,
receiptMessage.unwrappedType == .delivery else {
return nil
}
return receiptMessage.timestamp
case .syncMessage, .dataMessage, .callMessage, .typingMessage, .nullMessage,
.decryptionErrorMessage, .storyMessage, .hasSenderKeyDistributionMessage, .unknown:
return nil
}
}
}
init(_ pendingEnvelope: PendingEnvelope,
state: State) {
self.pendingEnvelope = pendingEnvelope
self.state = state
}
}
class RelatedProcessingRequests {
private(set) var processingRequests = [ProcessingRequest]()
var deliveryReceiptMessageTimestamps: [UInt64]? {
guard let request = processingRequests.first else {
return nil
}
return request.deliveryReceiptMessageTimestamps
}
func add(_ processingRequest: ProcessingRequest) {
processingRequests.append(processingRequest)
}
}
struct ProcessingRequestBuilder {
let pendingEnvelope: PendingEnvelope
let messageManager: OWSMessageManager
let blockingManager: BlockingManager
init(_ pendingEnvelope: PendingEnvelope,
messageManager: OWSMessageManager,
blockingManager: BlockingManager) {
self.pendingEnvelope = pendingEnvelope
self.messageManager = messageManager
self.blockingManager = blockingManager
}
func build(transaction: SDSAnyWriteTransaction) -> ProcessingRequest {
switch pendingEnvelope.decrypt(transaction: transaction) {
case .success(let result):
return decryptionSucceeded(result, transaction: transaction)
case .failure(let error):
return decryptionFailed(error)
}
}
private enum ProcessingStep {
case discard
case enqueueForGroupProcessing
case processNow(shouldDiscardVisibleMessages: Bool)
}
private func processingStep(_ result: DecryptedEnvelope,
sourceAddress: SignalServiceAddress,
transaction: SDSAnyWriteTransaction) -> ProcessingStep {
guard let plaintextData = result.plaintextData,
let groupContextV2 =
GroupsV2MessageProcessor.groupContextV2(fromPlaintextData: plaintextData) else {
// Non-v2-group messages can be processed immediately.
return .processNow(shouldDiscardVisibleMessages: false)
}
guard GroupsV2MessageProcessor.canContextBeProcessedImmediately(
groupContext: groupContextV2,
transaction: transaction
) else {
// Some v2 group messages required group state to be
// updated before they can be processed.
return .enqueueForGroupProcessing
}
let discardMode = GroupsMessageProcessor.discardMode(forMessageFrom: sourceAddress,
groupContext: groupContextV2,
transaction: transaction)
if discardMode == .discard {
// Some v2 group messages should be discarded and not processed.
Logger.verbose("Discarding job.")
return .discard
}
// Some v2 group messages should be processed, but
// discarding any "visible" messages, e.g. text messages
// or calls.
return .processNow(shouldDiscardVisibleMessages: discardMode == .discardVisibleMessages)
}
private func decryptionSucceeded(_ result: DecryptedEnvelope,
transaction: SDSAnyWriteTransaction) -> ProcessingRequest {
// NOTE: We use the envelope from the decrypt result, not the pending envelope,
// since the envelope may be altered by the decryption process in the UD case.
guard let sourceAddress = result.envelope.sourceAddress, sourceAddress.isValid else {
owsFailDebug("Successful decryption with no source address; discarding message")
let error = OWSAssertionError("successful decryption with no source address")
return ProcessingRequest(pendingEnvelope,
state: .completed(error: error))
}
// Pre-processing has to happen during the same transaction that performed decryption
messageManager.preprocessEnvelope(envelope: result.envelope,
plaintext: result.plaintextData,
transaction: transaction)
// If the sender is in the block list, we can skip scheduling any additional processing.
if blockingManager.isAddressBlocked(sourceAddress, transaction: transaction) {
Logger.info("Skipping processing for blocked envelope: \(sourceAddress)")
return ProcessingRequest(pendingEnvelope,
state: .completed(error: MessageProcessingError.blockedSender))
}
if result.identity == .pni {
NSObject.identityManager.setShouldSharePhoneNumber(with: sourceAddress, transaction: transaction)
}
switch processingStep(result, sourceAddress: sourceAddress, transaction: transaction) {
case .discard:
// Do nothing.
Logger.verbose("Discarding job.")
return ProcessingRequest(pendingEnvelope,
state: .completed(error: nil))
case .enqueueForGroupProcessing:
// If we can't process the message immediately, we enqueue it for
// for processing in the same transaction within which it was decrypted
// to prevent data loss.
let envelopeData: Data
if let existingEnvelopeData = result.envelopeData {
envelopeData = existingEnvelopeData
} else {
do {
envelopeData = try result.envelope.serializedData()
} catch {
owsFailDebug("failed to reserialize envelope: \(error)")
return ProcessingRequest(pendingEnvelope,
state: .completed(error: error))
}
}
return ProcessingRequest(pendingEnvelope,
state: .enqueueForGroup(decryptedEnvelope: result, envelopeData: envelopeData))
case .processNow(let shouldDiscardVisibleMessages):
// Envelopes can be processed immediately if they're:
// 1. Not a GV2 message.
// 2. A GV2 message that doesn't require updating the group.
//
// The advantage to processing the message immediately is that
// we can full process the message in the same transaction that
// we used to decrypt it. This results in a significant perf
// benefit verse queueing the message and waiting for that queue
// to open new transactions and process messages. The downside is
// that if we *fail* to process this message (e.g. the app crashed
// or was killed), we'll have to re-decrypt again before we process.
// This is safe, since the decrypt operation would also be rolled
// back (since the transaction didn't finalize) and should be rare.
if !messageManager.canProcessEnvelope(result.envelope, transaction: transaction) {
return ProcessingRequest(pendingEnvelope,
state: .completed(error: nil))
}
messageManager.checkForUnknownLinkedDevice(inEnvelope: result.envelope, transaction: transaction)
switch result.envelope.unwrappedType {
case .ciphertext, .prekeyBundle, .unidentifiedSender, .senderkeyMessage, .plaintextContent:
if result.plaintextData == nil {
owsFailDebug("Missing plaintextData")
}
guard let plaintextData = result.plaintextData,
let messageManagerRequest = messageManager.request(for: result.envelope,
plaintextData: plaintextData,
wasReceivedByUD: result.wasReceivedByUD,
serverDeliveryTimestamp: result.serverDeliveryTimestamp,
shouldDiscardVisibleMessages: shouldDiscardVisibleMessages,
transaction: transaction) else {
return ProcessingRequest(pendingEnvelope,
state: .completed(error: nil))
}
if messageManagerRequest.protoContent == nil {
messageManager.logUnactionablePayload(result.envelope)
return ProcessingRequest(pendingEnvelope,
state: .clearPlaceholdersOnly(result.envelope))
}
return ProcessingRequest(pendingEnvelope,
state: .messageManagerRequest(messageManagerRequest))
case .receipt:
return ProcessingRequest(pendingEnvelope,
state: .plaintextReceipt(result.envelope))
case .keyExchange:
Logger.warn("Received Key Exchange Message, not supported")
return ProcessingRequest(pendingEnvelope,
state: .completed(error: nil))
case .unknown:
Logger.warn("Received an unknown message type")
return ProcessingRequest(pendingEnvelope,
state: .completed(error: nil))
default:
Logger.warn("Received unhandled envelope type: \(result.envelope.unwrappedType)")
return ProcessingRequest(pendingEnvelope,
state: .completed(error: nil))
}
}
}
private func decryptionFailed(_ error: Error) -> ProcessingRequest {
return ProcessingRequest(pendingEnvelope,
state: .completed(error: error))
}
}
extension MessageProcessor {
func processingRequest(for envelope: PendingEnvelope,
messageManager: OWSMessageManager,
blockingManager: BlockingManager,
transaction: SDSAnyWriteTransaction) -> ProcessingRequest {
assertOnQueue(serialQueue)
let builder = ProcessingRequestBuilder(envelope,
messageManager: messageManager,
blockingManager: blockingManager)
return builder.build(transaction: transaction)
}
}
// MARK: -
extension MessageProcessor: MessageProcessingPipelineStage {
@ -471,7 +687,7 @@ extension MessageProcessor: MessageProcessingPipelineStage {
// MARK: -
private protocol PendingEnvelope {
protocol PendingEnvelope {
var completion: (Error?) -> Void { get }
var wasReceivedByUD: Bool { get }
func decrypt(transaction: SDSAnyWriteTransaction) -> Swift.Result<DecryptedEnvelope, Error>
@ -546,7 +762,7 @@ private struct EncryptedEnvelope: PendingEnvelope, Dependencies {
// MARK: -
private struct DecryptedEnvelope: PendingEnvelope {
struct DecryptedEnvelope: PendingEnvelope {
let envelope: SSKProtoEnvelope
let envelopeData: Data?
let plaintextData: Data?

2
SignalServiceKit/src/Messages/OWSMessageDecrypter.swift
View File

@ -98,7 +98,7 @@ public class OWSMessageDecrypter: OWSMessageHandler {
envelopeData: Data?,
transaction: SDSAnyWriteTransaction) -> Result<OWSMessageDecryptResult, Error> {
owsAssertDebug(tsAccountManager.isRegistered)
OWSMessageHandler.logInvalidEnvelope(envelope)
Logger.info("decrypting envelope: \(description(for: envelope))")
guard envelope.hasType else {

3
SignalServiceKit/src/Messages/OWSMessageHandler.h
View File

@ -1,5 +1,5 @@
//
// Copyright (c) 2020 Open Whisper Systems. All rights reserved.
// Copyright (c) 2022 Open Whisper Systems. All rights reserved.
//
NS_ASSUME_NONNULL_BEGIN
@ -19,6 +19,7 @@ NSString *envelopeAddress(SSKProtoEnvelope *envelope);
- (NSString *)descriptionForContent:(SSKProtoContent *)content;
- (NSString *)descriptionForDataMessage:(SSKProtoDataMessage *)dataMessage;
+ (void)logInvalidEnvelope:(SSKProtoEnvelope *)envelope;
@end

215
SignalServiceKit/src/Messages/OWSMessageHandler.m
View File

@ -10,11 +10,52 @@ NS_ASSUME_NONNULL_BEGIN
// used in log formatting
NSString *envelopeAddress(SSKProtoEnvelope *envelope)
{
return [NSString stringWithFormat:@"%@.%d", envelope.sourceAddress, (unsigned int)envelope.sourceDevice];
return [envelope formattedAddress];
}
@implementation OWSMessageHandler
+ (void)logInvalidEnvelope:(SSKProtoEnvelope *)envelope
{
OWSAssertDebug(envelope != nil);
if (!envelope.hasType) {
NSString *name = [OWSAnalyticsEvents messageManagerErrorEnvelopeTypeUnknown];
OWSFailDebug(@"%@", name);
return;
}
switch (envelope.unwrappedType) {
case SSKProtoEnvelopeTypeCiphertext:
case SSKProtoEnvelopeTypePrekeyBundle:
case SSKProtoEnvelopeTypeReceipt:
case SSKProtoEnvelopeTypeUnidentifiedSender:
case SSKProtoEnvelopeTypeSenderkeyMessage:
case SSKProtoEnvelopeTypePlaintextContent:
break;
case SSKProtoEnvelopeTypeUnknown: {
// Shouldn't happen
NSString *name = [OWSAnalyticsEvents messageManagerErrorEnvelopeTypeUnknown];
OWSFailDebug(@"%@", name);
return;
}
case SSKProtoEnvelopeTypeKeyExchange: {
// Unsupported
NSString *name = [OWSAnalyticsEvents messageManagerErrorEnvelopeTypeKeyExchange];
OWSFailDebug(@"%@", name);
return;
}
default: {
// Shouldn't happen
NSString *name = [OWSAnalyticsEvents messageManagerErrorEnvelopeTypeOther];
OWSFailDebug(@"%@", name);
return;
}
}
}
+ (NSString *)descriptionForEnvelopeType:(SSKProtoEnvelope *)envelope
{
OWSAssertDebug(envelope != nil);
@ -26,13 +67,11 @@ NSString *envelopeAddress(SSKProtoEnvelope *envelope)
switch (envelope.unwrappedType) {
case SSKProtoEnvelopeTypeUnknown:
// Shouldn't happen
OWSProdFail([OWSAnalyticsEvents messageManagerErrorEnvelopeTypeUnknown]);
return @"Unknown";
case SSKProtoEnvelopeTypeCiphertext:
return @"SignalEncryptedMessage";
case SSKProtoEnvelopeTypeKeyExchange:
// Unsupported
OWSProdFail([OWSAnalyticsEvents messageManagerErrorEnvelopeTypeKeyExchange]);
return @"KeyExchange";
case SSKProtoEnvelopeTypePrekeyBundle:
return @"PreKeyEncryptedMessage";
@ -46,7 +85,6 @@ NSString *envelopeAddress(SSKProtoEnvelope *envelope)
return @"PlaintextContent";
default:
// Shouldn't happen
OWSProdFail([OWSAnalyticsEvents messageManagerErrorEnvelopeTypeOther]);
return @"Other";
}
}
@ -77,92 +115,17 @@ NSString *envelopeAddress(SSKProtoEnvelope *envelope)
*/
- (NSString *)descriptionForContent:(SSKProtoContent *)content
{
NSMutableString *message = [[NSMutableString alloc] init];
if (content.syncMessage) {
[message appendFormat:@"<SyncMessage: %@ />", [self descriptionForSyncMessage:content.syncMessage]];
} else if (content.dataMessage) {
[message appendFormat:@"<DataMessage: %@ />", [self descriptionForDataMessage:content.dataMessage]];
} else if (content.callMessage) {
[message appendFormat:@"<CallMessage %@ />", [self descriptionForCallMessage:content.callMessage]];
} else if (content.nullMessage) {
[message appendFormat:@"<NullMessage: %@ />", content.nullMessage];
} else if (content.receiptMessage) {
[message appendFormat:@"<ReceiptMessage: %@ />", content.receiptMessage];
} else if (content.typingMessage) {
[message appendFormat:@"<TypingMessage: %@ />", content.typingMessage];
} else if (content.decryptionErrorMessage) {
[message appendFormat:@"<DecryptionErrorMessage: %@ />", content.decryptionErrorMessage];
} else if (content.storyMessage) {
[message appendFormat:@"<StoryMessage: %@ />", content.storyMessage];
}
// SKDM's are not mutually exclusive with other content types
if (content.hasSenderKeyDistributionMessage) {
if (message.length > 0) {
[message appendString:@" + "];
}
[message appendString:@"SenderKeyDistributionMessage"];
}
if (message.length > 0) {
return message;
} else {
// Don't fire an analytics event; if we ever add a new content type, we'd generate a ton of
// analytics traffic.
OWSFailDebug(@"Unknown content type.");
return @"UnknownContent";
}
return [content contentDescription];
}
- (NSString *)descriptionForCallMessage:(SSKProtoCallMessage *)callMessage
{
NSString *messageType;
UInt64 callId;
if (callMessage.offer) {
messageType = @"Offer";
callId = callMessage.offer.id;
} else if (callMessage.busy) {
messageType = @"Busy";
callId = callMessage.busy.id;
} else if (callMessage.answer) {
messageType = @"Answer";
callId = callMessage.answer.id;
} else if (callMessage.legacyHangup) {
messageType = @"legacyHangup";
callId = callMessage.legacyHangup.id;
} else if (callMessage.hangup) {
messageType = @"Hangup";
callId = callMessage.hangup.id;
} else if (callMessage.iceUpdate.count > 0) {
messageType = [NSString stringWithFormat:@"Ice Updates (%lu)", (unsigned long)callMessage.iceUpdate.count];
callId = callMessage.iceUpdate.firstObject.id;
} else if (callMessage.opaque) {
messageType = @"Opaque";
if (callMessage.opaque.hasUrgency) {
messageType = [NSString stringWithFormat:@"Opaque (%@)",
[self descriptionForCallMessageOpaqueUrgency:callMessage.opaque.unwrappedUrgency]];
}
callId = 0;
} else {
OWSFailDebug(@"failure: unexpected call message type: %@", callMessage);
messageType = @"Unknown";
callId = 0;
}
return [NSString stringWithFormat:@"type: %@, id: %llu", messageType, callId];
return [callMessage contentDescription];
}
- (NSString *)descriptionForCallMessageOpaqueUrgency:(SSKProtoCallMessageOpaqueUrgency)urgency
{
switch (urgency) {
case SSKProtoCallMessageOpaqueUrgencyDroppable:
return @"Droppable";
case SSKProtoCallMessageOpaqueUrgencyHandleImmediately:
return @"HandleImmediately";
default:
return @"Unknown";
}
return [SSKProtoCallMessageOpaque urgencyDescription:urgency];
}
/**
@ -170,25 +133,7 @@ NSString *envelopeAddress(SSKProtoEnvelope *envelope)
*/
- (NSString *)descriptionForDataMessage:(SSKProtoDataMessage *)dataMessage
{
NSMutableString *description = [NSMutableString new];
if (dataMessage.group) {
[description appendString:@"(Group:YES) "];
}
if ((dataMessage.flags & SSKProtoDataMessageFlagsEndSession) != 0) {
[description appendString:@"EndSession"];
} else if ((dataMessage.flags & SSKProtoDataMessageFlagsExpirationTimerUpdate) != 0) {
[description appendString:@"ExpirationTimerUpdate"];
} else if ((dataMessage.flags & SSKProtoDataMessageFlagsProfileKeyUpdate) != 0) {
[description appendString:@"ProfileKey"];
} else if (dataMessage.attachments.count > 0) {
[description appendString:@"MessageWithAttachment"];
} else {
[description appendString:@"Plain"];
}
return [NSString stringWithFormat:@"<%@ />", description];
return [dataMessage contentDescription];
}
/**
@ -196,73 +141,7 @@ NSString *envelopeAddress(SSKProtoEnvelope *envelope)
*/
- (NSString *)descriptionForSyncMessage:(SSKProtoSyncMessage *)syncMessage
{
if (syncMessage.sent) {
return @"SentTranscript";
} else if (syncMessage.request) {
if (!syncMessage.request.hasType) {
return @"Unknown sync request.";
}
if (syncMessage.request.unwrappedType == SSKProtoSyncMessageRequestTypeContacts) {
return @"ContactRequest";
} else if (syncMessage.request.unwrappedType == SSKProtoSyncMessageRequestTypeGroups) {
return @"GroupRequest";
} else if (syncMessage.request.unwrappedType == SSKProtoSyncMessageRequestTypeBlocked) {
return @"BlockedRequest";
} else if (syncMessage.request.unwrappedType == SSKProtoSyncMessageRequestTypeConfiguration) {
return @"ConfigurationRequest";
} else if (syncMessage.request.unwrappedType == SSKProtoSyncMessageRequestTypeKeys) {
return @"KeysRequest";
} else if (syncMessage.request.unwrappedType == SSKProtoSyncMessageRequestTypePniIdentity) {
return @"PniIdentityRequest";
} else {
OWSFailDebug(@"Unknown sync message request type");
return @"UnknownRequest";
}
} else if (syncMessage.blocked) {
return @"Blocked";
} else if (syncMessage.pniIdentity) {
return @"PniIdentity";
} else if (syncMessage.read.count > 0) {
return @"ReadReceipt";
} else if (syncMessage.verified) {
return [NSString stringWithFormat:@"Verification for: %@", syncMessage.verified.destinationAddress];
} else if (syncMessage.stickerPackOperation) {
NSMutableString *description = [NSMutableString new];
NSMutableArray<NSString *> *operationTypes = [NSMutableArray new];
for (SSKProtoSyncMessageStickerPackOperation *packOperationProto in syncMessage.stickerPackOperation) {
if (!packOperationProto.hasType) {
[operationTypes addObject:@"unknown"];
continue;
}
switch (packOperationProto.unwrappedType) {
case SSKProtoSyncMessageStickerPackOperationTypeInstall:
[operationTypes addObject:@"install"];
break;
case SSKProtoSyncMessageStickerPackOperationTypeRemove:
[operationTypes addObject:@"remove"];
break;
default:
[operationTypes addObject:@"unknown"];
break;
}
}
[description appendFormat:@"StickerPackOperation: %@", [operationTypes componentsJoinedByString:@", "]];
return description;
} else if (syncMessage.fetchLatest) {
switch (syncMessage.fetchLatest.unwrappedType) {
case SSKProtoSyncMessageFetchLatestTypeUnknown:
return @"FetchLatest_Unknown";
case SSKProtoSyncMessageFetchLatestTypeLocalProfile:
return @"FetchLatest_LocalProfile";
case SSKProtoSyncMessageFetchLatestTypeStorageManifest:
return @"FetchLatest_StorageManifest";
case SSKProtoSyncMessageFetchLatestTypeSubscriptionStatus:
return @"FetchLatest_SubscriptionStatus";
}
} else {
OWSFailDebug(@"Unknown sync message type");
return @"Unknown";
}
return [syncMessage contentDescription];
}
@end

44
SignalServiceKit/src/Messages/OWSMessageManager.h
View File

@ -1,14 +1,31 @@
//
// Copyright (c) 2021 Open Whisper Systems. All rights reserved.
// Copyright (c) 2022 Open Whisper Systems. All rights reserved.
//
#import <SignalServiceKit/OWSMessageHandler.h>
NS_ASSUME_NONNULL_BEGIN
@protocol DeliveryReceiptContext;
@class MessageManagerRequest;
@class SDSAnyWriteTransaction;
@class SSKProtoEnvelope;
typedef NS_CLOSED_ENUM(NSUInteger, OWSMessageManagerMessageType)
{
OWSMessageManagerMessageTypeSyncMessage,
OWSMessageManagerMessageTypeDataMessage,
OWSMessageManagerMessageTypeCallMessage,
OWSMessageManagerMessageTypeTypingMessage,
OWSMessageManagerMessageTypeNullMessage,
OWSMessageManagerMessageTypeReceiptMessage,
OWSMessageManagerMessageTypeDecryptionErrorMessage,
OWSMessageManagerMessageTypeStoryMessage,
OWSMessageManagerMessageTypeHasSenderKeyDistributionMessage,
OWSMessageManagerMessageTypeUnknown
};
@interface OWSMessageManager : OWSMessageHandler
// processEnvelope: can be called from any thread.
@ -21,6 +38,31 @@ NS_ASSUME_NONNULL_BEGIN
shouldDiscardVisibleMessages:(BOOL)shouldDiscardVisibleMessages
transaction:(SDSAnyWriteTransaction *)transaction;
- (BOOL)handleRequest:(MessageManagerRequest *)request
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction;
- (BOOL)canProcessEnvelope:(SSKProtoEnvelope *)envelope transaction:(SDSAnyWriteTransaction *)transaction;
- (void)checkForUnknownLinkedDevice:(SSKProtoEnvelope *)envelope
transaction:(SDSAnyWriteTransaction *)transaction
NS_SWIFT_NAME(checkForUnknownLinkedDevice(inEnvelope:transaction:));
- (void)finishProcessingEnvelope:(SSKProtoEnvelope *)envelope transaction:(SDSAnyWriteTransaction *)transaction;
- (MessageManagerRequest *_Nullable)requestForEnvelope:(SSKProtoEnvelope *)envelope
plaintextData:(NSData *)plaintextData
wasReceivedByUD:(BOOL)wasReceivedByUD
serverDeliveryTimestamp:(uint64_t)serverDeliveryTimestamp
shouldDiscardVisibleMessages:(BOOL)shouldDiscardVisibleMessages
transaction:(SDSAnyWriteTransaction *)transaction;
- (void)logUnactionablePayload:(SSKProtoEnvelope *)envelope;
- (void)handleDeliveryReceipt:(SSKProtoEnvelope *)envelope
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction;
@end
NS_ASSUME_NONNULL_END

304
SignalServiceKit/src/Messages/OWSMessageManager.m
View File

@ -181,6 +181,49 @@ NS_ASSUME_NONNULL_BEGIN
}
}
- (BOOL)canProcessEnvelope:(SSKProtoEnvelope *)envelope transaction:(SDSAnyWriteTransaction *)transaction
{
if (!envelope) {
OWSFailDebug(@"Missing envelope.");
return NO;
}
if (!transaction) {
OWSFail(@"Missing transaction.");
return NO;
}
if (!self.tsAccountManager.isRegistered) {
OWSFailDebug(@"Not registered.");
return NO;
}
if (!CurrentAppContext().shouldProcessIncomingMessages) {
OWSFail(@"Should not process messages.");
return NO;
}
OWSLogInfo(@"handling decrypted envelope: %@", [self descriptionForEnvelope:envelope]);
if (!envelope.hasValidSource) {
OWSFailDebug(@"incoming envelope has invalid source");
return NO;
}
if (!envelope.hasSourceDevice || envelope.sourceDevice < 1) {
OWSFailDebug(@"incoming envelope has invalid source device");
return NO;
}
if (!envelope.hasType) {
OWSFailDebug(@"incoming envelope is missing type.");
return NO;
}
if ([self isEnvelopeSenderBlocked:envelope transaction:transaction]) {
OWSLogInfo(@"incoming envelope sender is blocked.");
return NO;
}
return YES;
}
- (void)throws_processEnvelope:(SSKProtoEnvelope *)envelope
plaintextData:(NSData *_Nullable)plaintextData
wasReceivedByUD:(BOOL)wasReceivedByUD
@ -188,40 +231,7 @@ NS_ASSUME_NONNULL_BEGIN
shouldDiscardVisibleMessages:(BOOL)shouldDiscardVisibleMessages
transaction:(SDSAnyWriteTransaction *)transaction
{
if (!envelope) {
OWSFailDebug(@"Missing envelope.");
return;
}
if (!transaction) {
OWSFail(@"Missing transaction.");
return;
}
if (!self.tsAccountManager.isRegistered) {
OWSFailDebug(@"Not registered.");
return;
}
if (!CurrentAppContext().shouldProcessIncomingMessages) {
OWSFail(@"Should not process messages.");
return;
}
OWSLogInfo(@"handling decrypted envelope: %@", [self descriptionForEnvelope:envelope]);
if (!envelope.hasValidSource) {
OWSFailDebug(@"incoming envelope has invalid source");
return;
}
if (!envelope.hasSourceDevice || envelope.sourceDevice < 1) {
OWSFailDebug(@"incoming envelope has invalid source device");
return;
}
if (!envelope.hasType) {
OWSFailDebug(@"incoming envelope is missing type.");
return;
}
if ([self isEnvelopeSenderBlocked:envelope transaction:transaction]) {
OWSLogInfo(@"incoming envelope sender is blocked.");
if (![self canProcessEnvelope:envelope transaction:transaction]) {
return;
}
@ -246,7 +256,9 @@ NS_ASSUME_NONNULL_BEGIN
break;
case SSKProtoEnvelopeTypeReceipt:
OWSAssertDebug(!plaintextData);
[self handleDeliveryReceipt:envelope transaction:transaction];
[self handleDeliveryReceipt:envelope
context:[[PassthroughDeliveryReceiptContext alloc] init]
transaction:transaction];
break;
// Other messages are just dismissed for now.
case SSKProtoEnvelopeTypeKeyExchange:
@ -260,6 +272,11 @@ NS_ASSUME_NONNULL_BEGIN
break;
}
[self finishProcessingEnvelope:envelope transaction:transaction];
}
- (void)finishProcessingEnvelope:(SSKProtoEnvelope *)envelope transaction:(SDSAnyWriteTransaction *)transaction
{
// If we reach here, we were able to successfully handle the message.
// We need to check to make sure that we clear any placeholders that may have been
// inserted for this message. This would happen if:
@ -271,7 +288,9 @@ NS_ASSUME_NONNULL_BEGIN
}
}
- (void)handleDeliveryReceipt:(SSKProtoEnvelope *)envelope transaction:(SDSAnyWriteTransaction *)transaction
- (void)handleDeliveryReceipt:(SSKProtoEnvelope *)envelope
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction
{
if (!envelope) {
OWSFailDebug(@"Missing envelope.");
@ -293,6 +312,7 @@ NS_ASSUME_NONNULL_BEGIN
@(envelope.timestamp),
]
deliveryTimestamp:nil
context:context
transaction:transaction];
}
@ -304,6 +324,7 @@ NS_ASSUME_NONNULL_BEGIN
recipientDeviceId:(uint32_t)deviceId
sentTimestamps:(NSArray<NSNumber *> *)sentTimestamps
deliveryTimestamp:(NSNumber *_Nullable)deliveryTimestamp
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction
{
if (!address.isValid) {
@ -332,17 +353,7 @@ NS_ASSUME_NONNULL_BEGIN
continue;
}
NSError *error;
NSArray<TSOutgoingMessage *> *messages = (NSArray<TSOutgoingMessage *> *)[InteractionFinder
interactionsWithTimestamp:timestamp
filter:^(TSInteraction *interaction) {
return [interaction isKindOfClass:[TSOutgoingMessage class]];
}
transaction:transaction
error:&error];
if (error != nil) {
OWSFailDebug(@"Error loading interactions: %@", error);
}
NSArray<TSOutgoingMessage *> *messages = [context messagesWithTimestamp:timestamp transaction:transaction];
if (messages.count < 1) {
OWSLogInfo(@"Missing message for delivery receipt: %llu", timestamp);
@ -357,6 +368,7 @@ NS_ASSUME_NONNULL_BEGIN
[outgoingMessage updateWithDeliveredRecipient:address
recipientDeviceId:deviceId
deliveryTimestamp:deliveryTimestamp
context:context
transaction:transaction];
}
}
@ -365,6 +377,31 @@ NS_ASSUME_NONNULL_BEGIN
return earlyTimestamps;
}
- (MessageManagerRequest *_Nullable)requestForEnvelope:(SSKProtoEnvelope *)envelope
plaintextData:(NSData *)plaintextData
wasReceivedByUD:(BOOL)wasReceivedByUD
serverDeliveryTimestamp:(uint64_t)serverDeliveryTimestamp
shouldDiscardVisibleMessages:(BOOL)shouldDiscardVisibleMessages
transaction:(SDSAnyWriteTransaction *)transaction
{
if (transaction == nil) {
OWSFail(@"Missing transaction.");
return nil;
}
MessageManagerRequest *request = [[MessageManagerRequest alloc] initWithEnvelope:envelope
plaintextData:plaintextData
wasReceivedByUD:wasReceivedByUD
serverDeliveryTimestamp:serverDeliveryTimestamp
shouldDiscardVisibleMessages:shouldDiscardVisibleMessages
transaction:transaction];
return request;
}
- (void)logUnactionablePayload:(SSKProtoEnvelope *)envelope
{
OWSProdInfoWEnvelope([OWSAnalyticsEvents messageManagerErrorEnvelopeNoActionablePayload], envelope);
}
- (void)throws_handleEnvelope:(SSKProtoEnvelope *)envelope
plaintextData:(NSData *)plaintextData
wasReceivedByUD:(BOOL)wasReceivedByUD
@ -372,134 +409,129 @@ NS_ASSUME_NONNULL_BEGIN
shouldDiscardVisibleMessages:(BOOL)shouldDiscardVisibleMessages
transaction:(SDSAnyWriteTransaction *)transaction
{
if (![self isValidEnvelope:envelope]) {
OWSFailDebug(@"Missing envelope.");
return;
}
if (!plaintextData) {
OWSFailDebug(@"Missing plaintextData.");
return;
}
if (!transaction) {
OWSFail(@"Missing transaction.");
OWSAssert(plaintextData != nil);
MessageManagerRequest *request = [self requestForEnvelope:envelope
plaintextData:plaintextData
wasReceivedByUD:wasReceivedByUD
serverDeliveryTimestamp:serverDeliveryTimestamp
shouldDiscardVisibleMessages:shouldDiscardVisibleMessages
transaction:transaction];
if (request == nil) {
return;
}
BOOL duplicateEnvelope = [InteractionFinder existsIncomingMessageWithTimestamp:envelope.timestamp
address:envelope.sourceAddress
sourceDeviceId:envelope.sourceDevice
transaction:transaction];
[self throws_handleRequest:request
context:[[PassthroughDeliveryReceiptContext alloc] init]
transaction:transaction];
}
if (duplicateEnvelope) {
OWSLogInfo(@"Ignoring previously received envelope from %@ with timestamp: %llu",
envelopeAddress(envelope),
envelope.timestamp);
- (BOOL)handleRequest:(MessageManagerRequest *)request
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction
{
@try {
[self throws_handleRequest:request context:context transaction:transaction];
return YES;
} @catch (NSException *exception) {
OWSFailDebug(@"Received an invalid envelope: %@", exception.debugDescription);
return NO;
}
}
- (void)throws_handleRequest:(MessageManagerRequest *)request
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction
{
SSKProtoContent *contentProto = request.protoContent;
if (contentProto == nil) {
return;
}
OWSLogInfo(@"handling content: <Content: %@>", [self descriptionForContent:contentProto]);
if (envelope.content != nil) {
NSError *error;
SSKProtoContent *_Nullable contentProto = [[SSKProtoContent alloc] initWithSerializedData:plaintextData
error:&error];
if (error || !contentProto) {
OWSFailDebug(@"could not parse proto: %@", error);
return;
}
OWSLogInfo(@"handling content: <Content: %@>", [self descriptionForContent:contentProto]);
if (contentProto.syncMessage) {
[self throws_handleIncomingEnvelope:envelope
switch (request.messageType) {
case OWSMessageManagerMessageTypeSyncMessage:
[self throws_handleIncomingEnvelope:request.envelope
withSyncMessage:contentProto.syncMessage
plaintextData:plaintextData
wasReceivedByUD:wasReceivedByUD
serverDeliveryTimestamp:serverDeliveryTimestamp
plaintextData:request.plaintextData
wasReceivedByUD:request.wasReceivedByUD
serverDeliveryTimestamp:request.serverDeliveryTimestamp
transaction:transaction];
[[OWSDeviceManager shared] setHasReceivedSyncMessage];
} else if (contentProto.dataMessage) {
[self handleIncomingEnvelope:envelope
break;
case OWSMessageManagerMessageTypeDataMessage:
[self handleIncomingEnvelope:request.envelope
withDataMessage:contentProto.dataMessage
plaintextData:plaintextData
wasReceivedByUD:wasReceivedByUD
serverDeliveryTimestamp:serverDeliveryTimestamp
shouldDiscardVisibleMessages:shouldDiscardVisibleMessages
plaintextData:request.plaintextData
wasReceivedByUD:request.wasReceivedByUD
serverDeliveryTimestamp:request.serverDeliveryTimestamp
shouldDiscardVisibleMessages:request.shouldDiscardVisibleMessages
transaction:transaction];
} else if (contentProto.callMessage) {
if (shouldDiscardVisibleMessages) {
OWSLogInfo(@"Discarding message with timestamp: %llu", envelope.timestamp);
return;
}
OWSCallMessageAction action = [self.callMessageHandler actionForEnvelope:envelope
break;
case OWSMessageManagerMessageTypeCallMessage:
OWSAssertDebug(!request.shouldDiscardVisibleMessages);
OWSCallMessageAction action = [self.callMessageHandler actionForEnvelope:request.envelope
callMessage:contentProto.callMessage
serverDeliveryTimestamp:serverDeliveryTimestamp];
serverDeliveryTimestamp:request.serverDeliveryTimestamp];
switch (action) {
case OWSCallMessageActionIgnore:
OWSLogInfo(@"Ignoring call message with timestamp: %llu", envelope.timestamp);
OWSLogInfo(@"Ignoring call message with timestamp: %llu", request.envelope.timestamp);
break;
case OWSCallMessageActionHandoff:
[self.callMessageHandler externallyHandleCallMessageWithEnvelope:envelope
plaintextData:plaintextData
wasReceivedByUD:wasReceivedByUD
serverDeliveryTimestamp:serverDeliveryTimestamp
[self.callMessageHandler externallyHandleCallMessageWithEnvelope:request.envelope
plaintextData:request.plaintextData
wasReceivedByUD:request.wasReceivedByUD
serverDeliveryTimestamp:request.serverDeliveryTimestamp
transaction:transaction];
break;
case OWSCallMessageActionProcess:
[self handleIncomingEnvelope:envelope
[self handleIncomingEnvelope:request.envelope
withCallMessage:contentProto.callMessage
serverDeliveryTimestamp:serverDeliveryTimestamp
serverDeliveryTimestamp:request.serverDeliveryTimestamp
transaction:transaction];
break;
}
} else if (contentProto.typingMessage) {
[self handleIncomingEnvelope:envelope
break;
case OWSMessageManagerMessageTypeTypingMessage:
[self handleIncomingEnvelope:request.envelope
withTypingMessage:contentProto.typingMessage
serverDeliveryTimestamp:serverDeliveryTimestamp
serverDeliveryTimestamp:request.serverDeliveryTimestamp
transaction:transaction];
} else if (contentProto.nullMessage) {
break;
case OWSMessageManagerMessageTypeNullMessage:
OWSLogInfo(@"Received null message.");
} else if (contentProto.receiptMessage) {
[self handleIncomingEnvelope:envelope
break;
case OWSMessageManagerMessageTypeReceiptMessage:
[self handleIncomingEnvelope:request.envelope
withReceiptMessage:contentProto.receiptMessage
context:context
transaction:transaction];
} else if (contentProto.decryptionErrorMessage) {
[self handleIncomingEnvelope:envelope
break;
case OWSMessageManagerMessageTypeDecryptionErrorMessage:
[self handleIncomingEnvelope:request.envelope
withDecryptionErrorMessage:contentProto.decryptionErrorMessage
transaction:transaction];
} else if (contentProto.storyMessage) {
[self handleIncomingEnvelope:envelope withStoryMessage:contentProto.storyMessage transaction:transaction];
} else if (contentProto.hasSenderKeyDistributionMessage) {
break;
case OWSMessageManagerMessageTypeStoryMessage:
[self handleIncomingEnvelope:request.envelope
withStoryMessage:contentProto.storyMessage
transaction:transaction];
break;
case OWSMessageManagerMessageTypeHasSenderKeyDistributionMessage:
// Sender key distribution messages are not mutually exclusive. They can be
// included with any message type. However, they're not processed here. They're
// processed in the -preprocess phase that occurs post-decryption.
//
// See: OWSMessageManager.preprocessEnvelope(envelope:plaintext:transaction:)
} else {
break;
case OWSMessageManagerMessageTypeUnknown:
OWSLogWarn(@"Ignoring envelope. Content with no known payload");
}
} else if (envelope.legacyMessage != nil) { // DEPRECATED - Remove after all clients have been upgraded.
NSError *error;
SSKProtoDataMessage *_Nullable dataMessageProto =
[[SSKProtoDataMessage alloc] initWithSerializedData:plaintextData error:&error];
if (error || !dataMessageProto) {
OWSFailDebug(@"could not parse proto: %@", error);
return;
}
OWSLogInfo(@"handling message: <DataMessage: %@ />", [self descriptionForDataMessage:dataMessageProto]);
[self handleIncomingEnvelope:envelope
withDataMessage:dataMessageProto
plaintextData:plaintextData
wasReceivedByUD:wasReceivedByUD
serverDeliveryTimestamp:serverDeliveryTimestamp
shouldDiscardVisibleMessages:shouldDiscardVisibleMessages
transaction:transaction];
} else {
OWSProdInfoWEnvelope([OWSAnalyticsEvents messageManagerErrorEnvelopeNoActionablePayload], envelope);
break;
}
if (SSKDebugFlags.internalLogging || CurrentAppContext().isNSE) {
OWSLogInfo(@"Done timestamp: %llu, serviceTimestamp: %llu, ", envelope.timestamp, envelope.serverTimestamp);
OWSLogInfo(@"Done timestamp: %llu, serviceTimestamp: %llu, ",
request.envelope.timestamp,
request.envelope.serverTimestamp);
}
}
@ -863,6 +895,7 @@ NS_ASSUME_NONNULL_BEGIN
- (void)handleIncomingEnvelope:(SSKProtoEnvelope *)envelope
withReceiptMessage:(SSKProtoReceiptMessage *)receiptMessage
context:(id<DeliveryReceiptContext>)context
transaction:(SDSAnyWriteTransaction *)transaction
{
if (!envelope) {
@ -903,6 +936,7 @@ NS_ASSUME_NONNULL_BEGIN
recipientDeviceId:envelope.sourceDevice
sentTimestamps:sentTimestamps
deliveryTimestamp:@(envelope.timestamp)
context:context
transaction:transaction];
break;
case SSKProtoReceiptMessageTypeRead:

352
SignalServiceKit/src/Messages/OWSMessageManager.swift
View File

@ -94,23 +94,7 @@ extension OWSMessageManager {
@objc
func isValidEnvelope(_ envelope: SSKProtoEnvelope) -> Bool {
guard envelope.timestamp >= 1 else {
owsFailDebug("Invalid timestamp")
return false
}
guard SDS.fitsInInt64(envelope.timestamp) else {
owsFailDebug("Invalid timestamp")
return false
}
guard envelope.hasValidSource else {
owsFailDebug("Invalid source")
return false
}
guard envelope.sourceDevice >= 1 else {
owsFailDebug("Invalid source device")
return false
}
return true
return envelope.isValid
}
/// Performs a limited amount of time sensitive processing before scheduling the remainder of message processing
@ -126,6 +110,7 @@ extension OWSMessageManager {
/// to decrypt.
/// - This *needs* to happen in the very same write transaction where the message was decrypted. It's important to keep in mind
/// that the NSE could race with the main app when processing messages. The write transaction is used to protect us from any races.
@objc
func preprocessEnvelope(envelope: SSKProtoEnvelope, plaintext: Data?, transaction: SDSAnyWriteTransaction) {
guard CurrentAppContext().shouldProcessIncomingMessages else {
owsFail("Should not process messages")
@ -317,3 +302,336 @@ extension OWSMessageManager {
return "[" + splits.joined(separator: ", ") + "]"
}
}
extension SSKProtoEnvelope {
var isValid: Bool {
guard timestamp >= 1 else {
owsFailDebug("Invalid timestamp")
return false
}
guard SDS.fitsInInt64(timestamp) else {
owsFailDebug("Invalid timestamp")
return false
}
guard hasValidSource else {
owsFailDebug("Invalid source")
return false
}
guard sourceDevice >= 1 else {
owsFailDebug("Invalid source device")
return false
}
return true
}
@objc
var formattedAddress: String {
return "\(String(describing: sourceAddress)).\(sourceDevice)"
}
}
extension SSKProtoContent {
@objc
var contentDescription: String {
var message = String()
if let syncMessage = self.syncMessage {
message.append("<SyncMessage: \(syncMessage.contentDescription) />")
} else if let dataMessage = self.dataMessage {
message.append("<DataMessage: \(dataMessage.contentDescription) />")
} else if let callMessage = self.callMessage {
message.append("<CallMessage \(callMessage.contentDescription) />")
} else if let nullMessage = self.nullMessage {
message.append("<NullMessage: \(nullMessage) />")
} else if let receiptMessage = self.receiptMessage {
message.append("<ReceiptMessage: \(receiptMessage) />")
} else if let typingMessage = self.typingMessage {
message.append("<TypingMessage: \(typingMessage) />")
} else if let decryptionErrorMessage = self.decryptionErrorMessage {
message.append("<DecryptionErrorMessage: \(decryptionErrorMessage) />")
} else if let storyMessage = self.storyMessage {
message.append("<StoryMessage: \(storyMessage) />")
}
// SKDM's are not mutually exclusive with other content types
if hasSenderKeyDistributionMessage {
if !message.isEmpty {
message.append(" + ")
}
message.append("SenderKeyDistributionMessage")
}
if message.isEmpty {
// Don't fire an analytics event; if we ever add a new content type, we'd generate a ton of
// analytics traffic.
owsFailDebug("Unknown content type.")
return "UnknownContent"
}
return message
}
}
extension SSKProtoCallMessage {
@objc
var contentDescription: String {
let messageType: String
let callId: UInt64
if let offer = self.offer {
messageType = "Offer"
callId = offer.id
} else if let busy = self.busy {
messageType = "Busy"
callId = busy.id
} else if let answer = self.answer {
messageType = "Answer"
callId = answer.id
} else if let legacyHangup = self.legacyHangup {
messageType = "legacyHangup"
callId = legacyHangup.id
} else if let hangup = self.hangup {
messageType = "Hangup"
callId = hangup.id
} else if let firstICEUpdate = iceUpdate.first {
messageType = "Ice Updates \(iceUpdate.count)"
callId = firstICEUpdate.id
} else if let opaque = self.opaque {
if opaque.hasUrgency {
messageType = "Opaque (\(opaque.unwrappedUrgency.contentDescription))"
} else {
messageType = "Opaque"
}
callId = 0
} else {
owsFailDebug("failure: unexpected call message type: \(self)")
messageType = "Unknown"
callId = 0
}
return "type: \(messageType), id: \(callId)"
}
}
extension SSKProtoCallMessageOpaqueUrgency {
var contentDescription: String {
switch self {
case .droppable:
return "Droppable"
case .handleImmediately:
return "HandleImmediately"
default:
return "Unknown"
}
}
}
extension SSKProtoDataMessage {
@objc
var contentDescription: String {
var parts = [String]()
if group != nil {
parts.append("(Group:YES) )")
}
if Int64(flags) & Int64(SSKProtoDataMessageFlags.endSession.rawValue) != 0 {
parts.append("EndSession")
} else if Int64(flags) & Int64(SSKProtoDataMessageFlags.expirationTimerUpdate.rawValue) != 0 {
parts.append("ExpirationTimerUpdate")
} else if Int64(flags) & Int64(SSKProtoDataMessageFlags.profileKeyUpdate.rawValue) != 0 {
parts.append("ProfileKey")
} else if attachments.count > 0 {
parts.append("MessageWithAttachment")
} else {
parts.append("Plain")
}
return "<\(parts.joined(separator: " ")) />"
}
}
extension SSKProtoSyncMessage {
@objc
var contentDescription: String {
if sent != nil {
return "SentTranscript"
}
if let request = self.request {
if !request.hasType {
return "Unknown sync request."
}
switch request.unwrappedType {
case .contacts:
return "ContactRequest"
case .groups:
return "GroupRequest"
case .blocked:
return "BlockedRequest"
case .configuration:
return "ConfigurationRequest"
case .keys:
return "KeysRequest"
case .pniIdentity:
return "PniIdentityRequest"
default:
owsFailDebug("Unknown sync message request type")
return "UnknownRequest"
}
}
if blocked != nil {
return "Blocked"
}
if pniIdentity != nil {
return "PniIdentity"
}
if read.count > 0 {
return "ReadReceipt"
}
if let verified = verified {
return "Verification for: \(String(describing: verified.destinationAddress))"
}
if !stickerPackOperation.isEmpty {
var operationTypes = [String]()
for packOperationProto in stickerPackOperation {
if !packOperationProto.hasType {
operationTypes.append("unknown")
continue
}
switch packOperationProto.unwrappedType {
case .install:
operationTypes.append("install")
case .remove:
operationTypes.append("remove")
default:
operationTypes.append("unknown")
}
}
return "StickerPackOperation: \(operationTypes.joined(separator: ", "))"
}
if let fetchLatest = fetchLatest {
switch fetchLatest.unwrappedType {
case .unknown:
return "FetchLatest_Unknown"
case .localProfile:
return "FetchLatest_LocalProfile"
case .storageManifest:
return "FetchLatest_StorageManifest"
case .subscriptionStatus:
return "FetchLatest_SubscriptionStatus"
}
} else {
owsFailDebug("Unknown sync message type")
return "Unknown"
}
}
}
extension SSKProtoCallMessageOpaque {
@objc
static func urgencyDescription(_ value: SSKProtoCallMessageOpaqueUrgency) -> String {
return value.contentDescription
}
}
@objc
class MessageManagerRequest: NSObject {
@objc let envelope: SSKProtoEnvelope
@objc let plaintextData: Data
@objc let wasReceivedByUD: Bool
@objc let serverDeliveryTimestamp: UInt64
@objc let shouldDiscardVisibleMessages: Bool
enum Kind {
case modern(SSKProtoContent)
case unactionable
}
private let kind: Kind
@objc
var protoContent: SSKProtoContent? {
switch kind {
case .modern(let content):
return content
case .unactionable:
return nil
}
}
@objc
var messageType: OWSMessageManagerMessageType {
if protoContent?.syncMessage != nil {
return .syncMessage
}
if protoContent?.dataMessage != nil {
return .dataMessage
}
if protoContent?.callMessage != nil {
return .callMessage
}
if protoContent?.typingMessage != nil {
return .typingMessage
}
if protoContent?.nullMessage != nil {
return .nullMessage
}
if protoContent?.receiptMessage != nil {
return .receiptMessage
}
if protoContent?.decryptionErrorMessage != nil {
return .decryptionErrorMessage
}
if protoContent?.storyMessage != nil {
return .storyMessage
}
if protoContent?.hasSenderKeyDistributionMessage ?? false {
return .hasSenderKeyDistributionMessage
}
return .unknown
}
@objc
init?(envelope: SSKProtoEnvelope,
plaintextData: Data,
wasReceivedByUD: Bool,
serverDeliveryTimestamp: UInt64,
shouldDiscardVisibleMessages: Bool,
transaction: SDSAnyWriteTransaction) {
guard envelope.isValid, let sourceAddress = envelope.sourceAddress else {
owsFailDebug("Missing envelope")
return nil
}
self.envelope = envelope
self.plaintextData = plaintextData
self.wasReceivedByUD = wasReceivedByUD
self.serverDeliveryTimestamp = serverDeliveryTimestamp
self.shouldDiscardVisibleMessages = shouldDiscardVisibleMessages
if Self.isDuplicate(envelope, address: sourceAddress, transaction: transaction) {
Logger.info("Ignoring previously received envelope from \(envelope.formattedAddress) with timestamp: \(envelope.timestamp)")
return nil
}
if envelope.content != nil {
do {
let contentProto = try SSKProtoContent(serializedData: self.plaintextData)
Logger.info("handling content: <Content: \(contentProto.contentDescription)>")
if contentProto.callMessage != nil && shouldDiscardVisibleMessages {
Logger.info("Discarding message with timestamp \(envelope.timestamp)")
return nil
}
kind = .modern(contentProto)
} catch {
owsFailDebug("could not parse proto: \(error)")
return nil
}
} else {
kind = .unactionable
}
}
private static func isDuplicate(_ envelope: SSKProtoEnvelope,
address: SignalServiceAddress,
transaction: SDSAnyReadTransaction) -> Bool {
return InteractionFinder.existsIncomingMessage(timestamp: envelope.timestamp,
address: address,
sourceDeviceId: envelope.sourceDevice,
transaction: transaction)
}
}

95
SignalServiceKit/src/TestUtils/TestProtocolRunner.swift
View File

@ -270,6 +270,53 @@ public struct FakeService: Dependencies {
return builder
}
public func envelopeBuilder(fromSenderClient senderClient: TestSignalClient, groupV2Context: SSKProtoGroupContextV2) throws -> SSKProtoEnvelopeBuilder {
envelopeId += 1
let builder = SSKProtoEnvelope.builder(timestamp: envelopeId)
builder.setType(.ciphertext)
builder.setSourceDevice(senderClient.deviceId)
let content = try buildEncryptedContentData(fromSenderClient: senderClient, groupV2Context: groupV2Context)
builder.setContent(content)
// builder.setServerTimestamp(serverTimestamp)
// builder.setServerGuid(serverGuid)
return builder
}
public func envelopeBuilderForServerGeneratedDeliveryReceipt(fromSenderClient senderClient: TestSignalClient) -> SSKProtoEnvelopeBuilder {
envelopeId += 1
let builder = SSKProtoEnvelope.builder(timestamp: envelopeId)
builder.setType(.receipt)
builder.setSourceDevice(senderClient.deviceId)
return builder
}
public func envelopeBuilderForInvalidEnvelope(fromSenderClient senderClient: TestSignalClient) -> SSKProtoEnvelopeBuilder {
envelopeId += 1
let builder = SSKProtoEnvelope.builder(timestamp: envelopeId)
builder.setType(.unknown)
builder.setSourceDevice(senderClient.deviceId)
builder.setContent("Hello world".data(using: .utf8)!)
return builder
}
public func envelopeBuilderForUDDeliveryReceipt(fromSenderClient senderClient: TestSignalClient,
timestamp: UInt64) -> SSKProtoEnvelopeBuilder {
envelopeId += 1
let builder = SSKProtoEnvelope.builder(timestamp: envelopeId)
builder.setType(.ciphertext)
builder.setSourceDevice(senderClient.deviceId)
let content = try! buildEncryptedContentData(fromSenderClient: senderClient, deliveryReceiptForMessage: timestamp)
builder.setContent(content)
return builder
}
public func buildEncryptedContentData(fromSenderClient senderClient: TestSignalClient, bodyText: String?) throws -> Data {
let plaintext = try buildContentData(bodyText: bodyText)
let cipherMessage: CiphertextMessage = databaseStorage.write { transaction in
@ -283,6 +330,33 @@ public struct FakeService: Dependencies {
return Data(cipherMessage.serialize())
}
public func buildEncryptedContentData(fromSenderClient senderClient: TestSignalClient, groupV2Context: SSKProtoGroupContextV2) throws -> Data {
let plaintext = try buildContentData(groupV2Context: groupV2Context)
let cipherMessage: CiphertextMessage = databaseStorage.write { transaction in
return try! self.runner.encrypt(plaintext,
senderClient: senderClient,
recipient: self.localClient.protocolAddress,
context: transaction)
}
assert(cipherMessage.messageType == .whisper)
return Data(cipherMessage.serialize())
}
public func buildEncryptedContentData(fromSenderClient senderClient: TestSignalClient,
deliveryReceiptForMessage timestamp: UInt64) throws -> Data {
let plaintext = try buildContentData(deliveryReceiptForMessage: timestamp)
let cipherMessage: CiphertextMessage = databaseStorage.write { transaction in
return try! self.runner.encrypt(plaintext,
senderClient: senderClient,
recipient: self.localClient.protocolAddress,
context: transaction)
}
assert(cipherMessage.messageType == .whisper)
return Data(cipherMessage.serialize())
}
public func buildContentData(bodyText: String?) throws -> Data {
let dataMessageBuilder = SSKProtoDataMessage.builder()
if let bodyText = bodyText {
@ -296,6 +370,27 @@ public struct FakeService: Dependencies {
return try contentBuilder.buildSerializedData()
}
public func buildContentData(groupV2Context: SSKProtoGroupContextV2) throws -> Data {
let dataMessageBuilder = SSKProtoDataMessage.builder()
dataMessageBuilder.setGroupV2(groupV2Context)
let contentBuilder = SSKProtoContent.builder()
contentBuilder.setDataMessage(try dataMessageBuilder.build())
return try contentBuilder.buildSerializedData()
}
public func buildContentData(deliveryReceiptForMessage timestamp: UInt64) throws -> Data {
let receiptMessageBuilder = SSKProtoReceiptMessage.builder()
receiptMessageBuilder.setType(.delivery)
receiptMessageBuilder.setTimestamp([timestamp])
let contentBuilder = SSKProtoContent.builder()
contentBuilder.setReceiptMessage(try receiptMessageBuilder.build())
return try contentBuilder.buildSerializedData()
}
}
#endif

6
SignalServiceKit/tests/Messages/Interactions/TSOutgoingMessageTest.swift
View File

@ -123,6 +123,7 @@ class TSOutgoingMessageTest: SSKBaseTestSwift {
message.update(withDeliveredRecipient: otherAddress,
recipientDeviceId: 1,
deliveryTimestamp: nil,
context: PassthroughDeliveryReceiptContext(),
transaction: transaction)
XCTAssertFalse(identityManager.shouldSharePhoneNumber(with: otherAddress, transaction: transaction))
@ -160,6 +161,7 @@ class TSOutgoingMessageTest: SSKBaseTestSwift {
message.update(withDeliveredRecipient: otherAddress,
recipientDeviceId: 1,
deliveryTimestamp: nil,
context: PassthroughDeliveryReceiptContext(),
transaction: transaction)
// Still waiting on device #2!
@ -168,6 +170,7 @@ class TSOutgoingMessageTest: SSKBaseTestSwift {
message.update(withDeliveredRecipient: otherAddress,
recipientDeviceId: 2,
deliveryTimestamp: nil,
context: PassthroughDeliveryReceiptContext(),
transaction: transaction)
// There we go.
@ -201,6 +204,7 @@ class TSOutgoingMessageTest: SSKBaseTestSwift {
message.update(withDeliveredRecipient: otherAddress,
recipientDeviceId: 1,
deliveryTimestamp: nil,
context: PassthroughDeliveryReceiptContext(),
transaction: transaction)
// Still not changed!
@ -234,6 +238,7 @@ class TSOutgoingMessageTest: SSKBaseTestSwift {
message.update(withDeliveredRecipient: otherAddress,
recipientDeviceId: 1,
deliveryTimestamp: nil,
context: PassthroughDeliveryReceiptContext(),
transaction: transaction)
// ...it should stay active.
@ -278,6 +283,7 @@ class TSOutgoingMessageTest: SSKBaseTestSwift {
message.update(withDeliveredRecipient: otherAddress,
recipientDeviceId: 1,
deliveryTimestamp: nil,
context: PassthroughDeliveryReceiptContext(),
transaction: transaction)
// Still on, because our PNI changed!

84
SignalServiceKit/tests/Messages/MessageProcessingIntegrationTest.swift
View File

@ -5,6 +5,7 @@
import XCTest
@testable import SignalServiceKit
import GRDB
import LibSignalClient
class MessageProcessingIntegrationTest: SSKBaseTestSwift {
@ -231,6 +232,89 @@ class MessageProcessingIntegrationTest: SSKBaseTestSwift {
waitForExpectations(timeout: 1.0)
}
private func deliveryReceiptProtoEnvelope(timestamp: UInt64) throws -> SSKProtoEnvelope {
let envelopeBuilder = fakeService.envelopeBuilderForServerGeneratedDeliveryReceipt(fromSenderClient: bobClient)
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
envelopeBuilder.setServerGuid(UUID().uuidString)
envelopeBuilder.setDestinationUuid(localClient.uuidIdentifier)
envelopeBuilder.setTimestamp(timestamp)
return try envelopeBuilder.build()
}
private func invalidProtoEnvelope() throws -> SSKProtoEnvelope {
let envelopeBuilder = fakeService.envelopeBuilderForInvalidEnvelope(fromSenderClient: bobClient)
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
envelopeBuilder.setServerGuid(UUID().uuidString)
envelopeBuilder.setDestinationUuid(localClient.uuidIdentifier)
return try envelopeBuilder.build()
}
private func udDeliveryReceiptProtoEnvelope(timestamp: UInt64) throws -> SSKProtoEnvelope {
let envelopeBuilder = fakeService.envelopeBuilderForUDDeliveryReceipt(fromSenderClient: bobClient,
timestamp: timestamp)
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
envelopeBuilder.setServerGuid(UUID().uuidString)
envelopeBuilder.setDestinationUuid(localClient.uuidIdentifier)
return try envelopeBuilder.build()
}
private func normalMessageProtoEnvelope() throws -> SSKProtoEnvelope {
let envelopeBuilder = try! fakeService.envelopeBuilder(fromSenderClient: bobClient, bodyText: "Those who stands for nothing will fall for anything")
envelopeBuilder.setSourceE164(bobClient.e164Identifier!)
envelopeBuilder.setSourceUuid(bobClient.uuidIdentifier)
envelopeBuilder.setServerTimestamp(NSDate.ows_millisecondTimeStamp())
envelopeBuilder.setServerGuid(UUID().uuidString)
return try envelopeBuilder.build()
}
private func decrytpedEnvelopeForDeliveryReceipt(timestamp: UInt64) throws -> DecryptedEnvelope {
let envelope = try deliveryReceiptProtoEnvelope(timestamp: timestamp)
let data = try envelope.serializedData()
return DecryptedEnvelope(envelope: envelope,
envelopeData: data,
plaintextData: nil,
serverDeliveryTimestamp: 1234,
wasReceivedByUD: false,
identity: .pni,
completion: { _ in })
}
private func decryptedEnvelopeForNormalMessage() throws -> DecryptedEnvelope {
let envelope = try normalMessageProtoEnvelope()
let data = try envelope.serializedData()
return DecryptedEnvelope(envelope: envelope,
envelopeData: data,
plaintextData: nil,
serverDeliveryTimestamp: 1234,
wasReceivedByUD: false,
identity: .pni,
completion: { _ in })
}
private func deliveryReceiptProcessingRequest(timestamp: UInt64) throws -> ProcessingRequest {
let decryptedEnvelope = try decrytpedEnvelopeForDeliveryReceipt(timestamp: 1)
return ProcessingRequest(decryptedEnvelope,
state: .plaintextReceipt(decryptedEnvelope.envelope))
}
private func normalMessageProcessingRequest() throws -> ProcessingRequest {
let envelope = try! normalMessageProtoEnvelope()
let mmr = write { transaction in
MessageManagerRequest(envelope: envelope,
plaintextData: Data(),
wasReceivedByUD: false,
serverDeliveryTimestamp: 1,
shouldDiscardVisibleMessages: false,
transaction: transaction)!
}
let decryptedEnvelope = try decrytpedEnvelopeForDeliveryReceipt(timestamp: 1)
return ProcessingRequest(decryptedEnvelope,
state: .messageManagerRequest(mmr))
}
func testPniMessage() {
let localPniClient = LocalSignalClient(identity: .pni)
write { transaction in

3
SignalServiceKit/tests/Messages/OWSDisappearingMessageFinderTest.m
View File

@ -1,5 +1,5 @@
//
// Copyright (c) 2021 Open Whisper Systems. All rights reserved.
// Copyright (c) 2022 Open Whisper Systems. All rights reserved.
//
#import "MockSSKEnvironment.h"
@ -216,6 +216,7 @@ NS_ASSUME_NONNULL_BEGIN
[expiringDeliveredOutgoingMessage updateWithDeliveredRecipient:self.otherAddress
recipientDeviceId:0
deliveryTimestamp:nil
context:[[PassthroughDeliveryReceiptContext alloc] init]
transaction:transaction];
uint64_t nowMs = [NSDate ows_millisecondTimeStamp];
[expiringDeliveredAndReadOutgoingMessage updateWithReadRecipient:self.otherAddress