diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/NetworkResultUtil.kt b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/NetworkResultUtil.kt index db09899aa7..a79759c5cb 100644 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/NetworkResultUtil.kt +++ b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/NetworkResultUtil.kt @@ -123,7 +123,7 @@ object NetworkResultUtil { throw when (result.code) { 400, 401 -> AuthorizationFailedException(result.code, "Authorization failed!") 404 -> NotFoundException("Not found") - 429 -> RateLimitException(result.code, "Rate limit exceeded: ${result.code}", Optional.empty()) + 429 -> RateLimitException(result.code, "Rate limit exceeded: ${result.code}", Optional.ofNullable(result.retryAfter()?.inWholeMilliseconds)) 508 -> ServerRejectedException() else -> result.exception } diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java index e4b9a17882..43e74f9a70 100644 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java +++ b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java @@ -2494,11 +2494,15 @@ public class SignalServiceMessageSender { accessBySid.put(addressIterator.next().getServiceId(), accessIterator.next()); } - SenderCertificate senderCertificate = unidentifiedAccess.stream().filter(Objects::nonNull).findFirst().map(UnidentifiedAccess::getUnidentifiedCertificate).orElse(null); - SealedSenderAccess sealedSenderAccess = SealedSenderAccess.forGroupSend(senderCertificate, groupSendEndorsements, story); + SenderCertificate senderCertificate = unidentifiedAccess.stream().filter(Objects::nonNull).findFirst().map(UnidentifiedAccess::getUnidentifiedCertificate).orElse(null); + SenderCertificate sealedSenderCertificate = story ? senderCertificate : groupSendEndorsements.getSealedSenderCertificate(); + + List workingRecipients = new ArrayList<>(recipients); + List deferredResults = new LinkedList<>(); + Set quarantined = new HashSet<>(); for (int i = 0; i < RETRY_COUNT; i++) { - GroupTargetInfo targetInfo = buildGroupTargetInfo(recipients); + GroupTargetInfo targetInfo = buildGroupTargetInfo(workingRecipients); final GroupTargetInfo targetInfoSnapshot = targetInfo; Set sharedWith = aciStore.getSenderKeySharedWith(distributionId); @@ -2516,7 +2520,7 @@ public class SignalServiceMessageSender { .collect(Collectors.toList()); List needsSenderKeyGroupSendTokens = groupSendEndorsements != null ? groupSendEndorsements.forIndividuals(needsSenderKeyTargets) : null; - List needsSenderKeySealedSenderAccesses = SealedSenderAccess.forFanOutGroupSend(needsSenderKeyGroupSendTokens, sealedSenderAccess.getSenderCertificate(), needsSenderKeyAccesses); + List needsSenderKeySealedSenderAccesses = SealedSenderAccess.forFanOutGroupSend(needsSenderKeyGroupSendTokens, sealedSenderCertificate, needsSenderKeyAccesses); List results = sendSenderKeyDistributionMessage(distributionId, needsSenderKeyTargets, @@ -2540,29 +2544,24 @@ public class SignalServiceMessageSender { int failureCount = results.size() - successes.size(); if (failureCount > 0) { - Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Failed to send sender keys to " + failureCount + " recipients. Sending back failed results now."); + Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Failed to send sender keys to " + failureCount + " recipient(s). Quarantining and continuing send to the rest."); - List trueFailures = results.stream() - .filter(r -> !r.isSuccess()) - .collect(Collectors.toList()); + for (SendMessageResult failure : results) { + if (!failure.isSuccess() && quarantined.add(failure.getAddress().getServiceId())) { + deferredResults.add(failure); + } + } - Set failedAddresses = trueFailures.stream() - .map(result -> result.getAddress().getServiceId()) - .collect(Collectors.toSet()); + workingRecipients = workingRecipients.stream() + .filter(r -> !quarantined.contains(r.getServiceId())) + .collect(Collectors.toList()); - List fakeNetworkFailures = recipients.stream() - .filter(r -> !failedAddresses.contains(r.getServiceId())) - .map(SendMessageResult::networkFailure) - .collect(Collectors.toList()); - - List modifiedResults = new LinkedList<>(); - modifiedResults.addAll(trueFailures); - modifiedResults.addAll(fakeNetworkFailures); - - return modifiedResults; - } else { - targetInfo = buildGroupTargetInfo(recipients); + if (workingRecipients.isEmpty()) { + return deferredResults; + } } + + targetInfo = buildGroupTargetInfo(workingRecipients); } sendEvents.onSenderKeyShared(); @@ -2571,7 +2570,7 @@ public class SignalServiceMessageSender { byte[] ciphertext; try { - ciphertext = cipher.encryptForGroup(distributionId, targetInfo.destinations, targetInfo.sessions, sealedSenderAccess.getSenderCertificate(), content.encode(), contentHint, groupId); + ciphertext = cipher.encryptForGroup(distributionId, targetInfo.destinations, targetInfo.sessions, sealedSenderCertificate, content.encode(), contentHint, groupId); } catch (org.signal.libsignal.protocol.UntrustedIdentityException e) { throw new UntrustedIdentityException("Untrusted during group encrypt", e.getName(), e.getUntrustedIdentity()); } @@ -2579,19 +2578,20 @@ public class SignalServiceMessageSender { sendEvents.onMessageEncrypted(); MultiRecipientSendAuthorization multiRecipientAuth = story ? MultiRecipientSendAuthorization.Story.INSTANCE - : new MultiRecipientSendAuthorization.GroupSend(groupSendEndorsements.toFullToken()); + : new MultiRecipientSendAuthorization.GroupSend(groupSendEndorsements.toFullToken(workingRecipients)); RequestResult result = messageApi.sendGroupMessage(ciphertext, multiRecipientAuth, timestamp, online, urgent); if (result instanceof RequestResult.Success) { - MultiRecipientMessageResponse response = ((RequestResult.Success) result).getResult(); - return transformGroupResponseToMessageResults(targetInfo.devices, MessageApiKt.unsentTargets(response), content); + MultiRecipientMessageResponse response = ((RequestResult.Success) result).getResult(); + List sendResults = new LinkedList<>(transformGroupResponseToMessageResults(targetInfo.devices, MessageApiKt.unsentTargets(response), content)); + sendResults.addAll(deferredResults); + return sendResults; } else if (result instanceof RequestResult.NonSuccess) { MultiRecipientSendFailure error = ((RequestResult.NonSuccess) result).getError(); if (error instanceof MismatchedDeviceException) { MismatchedDeviceException mismatchedDeviceException = (MismatchedDeviceException) error; Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Handling mismatched devices. (" + mismatchedDeviceException.getMessage() + ")"); - List invalidPreKeyResults = new LinkedList<>(); for (MismatchedDeviceException.Entry entry : mismatchedDeviceException.getEntries()) { SignalServiceAddress address = new SignalServiceAddress(ServiceId.fromLibSignal(entry.getAccount())); @@ -2599,8 +2599,16 @@ public class SignalServiceMessageSender { try { handleMismatchedDevices(address, devices); } catch (InvalidPreKeyException e) { - Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Invalid prekey for " + address.getIdentifier() + " during mismatch handling."); - invalidPreKeyResults.add(SendMessageResult.invalidPreKeyFailure(address)); + Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Invalid prekey for " + address.getIdentifier() + " during mismatch handling. Quarantining."); + if (quarantined.add(address.getServiceId())) { + deferredResults.add(SendMessageResult.invalidPreKeyFailure(address)); + } + continue; + } catch (RateLimitException e) { + Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Rate limited fetching prekeys for " + address.getIdentifier() + " during mismatch handling. Quarantining."); + if (quarantined.add(address.getServiceId())) { + deferredResults.add(SendMessageResult.rateLimitFailure(address, e)); + } continue; } if (entry.getStaleDevices().length > 0) { @@ -2609,20 +2617,14 @@ public class SignalServiceMessageSender { } } - if (!invalidPreKeyResults.isEmpty()) { - Set failedAddresses = invalidPreKeyResults.stream() - .map(r -> r.getAddress().getServiceId()) - .collect(Collectors.toSet()); + if (!quarantined.isEmpty()) { + workingRecipients = workingRecipients.stream() + .filter(r -> !quarantined.contains(r.getServiceId())) + .collect(Collectors.toList()); - List networkFailures = recipients.stream() - .filter(r -> !failedAddresses.contains(r.getServiceId())) - .map(SendMessageResult::networkFailure) - .collect(Collectors.toList()); - - List combinedResults = new LinkedList<>(); - combinedResults.addAll(invalidPreKeyResults); - combinedResults.addAll(networkFailures); - return combinedResults; + if (workingRecipients.isEmpty()) { + return deferredResults; + } } } else if (error instanceof RequestUnauthorizedException) { Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Invalid access header."); diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/crypto/SealedSenderAccess.kt b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/crypto/SealedSenderAccess.kt index 564a592a47..60fde0f516 100644 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/crypto/SealedSenderAccess.kt +++ b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/crypto/SealedSenderAccess.kt @@ -8,7 +8,6 @@ package org.whispersystems.signalservice.api.crypto import org.signal.core.util.Base64 import org.signal.libsignal.metadata.certificate.SenderCertificate import org.signal.libsignal.zkgroup.groupsend.GroupSendFullToken -import org.whispersystems.signalservice.api.groupsv2.GroupSendEndorsements /** * Provides single interface for the various ways to send via sealed sender. @@ -73,35 +72,6 @@ sealed class SealedSenderAccess { } } - /** - * For sending to a "group" of recipients using group send endorsements/tokens. - */ - class GroupGroupSendToken( - private val groupSendEndorsements: GroupSendEndorsements - ) : SealedSenderAccess() { - - override val headerName: String = "Group-Send-Token" - override val headerValue: String by lazy { Base64.encodeWithPadding(groupSendEndorsements.serialize()) } - - override val senderCertificate: SenderCertificate - get() = groupSendEndorsements.sealedSenderCertificate - - override fun switchToFallback(): SealedSenderAccess? { - return null - } - } - - class StorySendNoop(override val senderCertificate: SenderCertificate) : SealedSenderAccess() { - override val headerName: String = "" - override val headerValue: String = "" - - override fun switchToFallback(): SealedSenderAccess? { - return null - } - - override fun applyHeader(): Boolean = false - } - /** * Provide a lazy way to create a group send token. */ @@ -157,15 +127,6 @@ sealed class SealedSenderAccess { } } - @JvmStatic - fun forGroupSend(senderCertificate: SenderCertificate?, groupSendEndorsements: GroupSendEndorsements?, forStory: Boolean): SealedSenderAccess { - if (forStory) { - return StorySendNoop(senderCertificate!!) - } - - return GroupGroupSendToken(groupSendEndorsements!!) - } - @JvmStatic fun isUnrestrictedForStory(sealedSenderAccess: SealedSenderAccess?): Boolean { return when (sealedSenderAccess) { diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/groupsv2/GroupSendEndorsements.kt b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/groupsv2/GroupSendEndorsements.kt index 64a5e27c9e..64855a1191 100644 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/groupsv2/GroupSendEndorsements.kt +++ b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/groupsv2/GroupSendEndorsements.kt @@ -11,6 +11,7 @@ import org.signal.libsignal.zkgroup.groups.GroupSecretParams import org.signal.libsignal.zkgroup.groupsend.GroupSendEndorsement import org.signal.libsignal.zkgroup.groupsend.GroupSendFullToken import org.whispersystems.signalservice.api.push.SignalServiceAddress +import java.io.IOException import java.time.Instant /** @@ -24,14 +25,13 @@ data class GroupSendEndorsements( ) { private val expiration: Instant by lazy { Instant.ofEpochMilli(expirationMs) } - private val combinedEndorsement: GroupSendEndorsement by lazy { GroupSendEndorsement.combine(endorsements.values) } - fun toFullToken(): GroupSendFullToken { - return combinedEndorsement.toFullToken(groupSecretParams, expiration) - } - - fun serialize(): ByteArray { - return toFullToken().serialize() + @Throws(IOException::class) + fun toFullToken(addresses: List): GroupSendFullToken { + val combined = GroupSendEndorsement.combine( + addresses.map { endorsements[it.serviceId] ?: throw IOException("Missing group send endorsement for a group-send recipient") } + ) + return combined.toFullToken(groupSecretParams, expiration) } fun forIndividuals(addresses: List): List { diff --git a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/push/exceptions/RateLimitException.java b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/push/exceptions/RateLimitException.java index 86f45fc314..c159e803a5 100644 --- a/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/push/exceptions/RateLimitException.java +++ b/lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/push/exceptions/RateLimitException.java @@ -8,7 +8,6 @@ package org.whispersystems.signalservice.api.push.exceptions; import org.signal.network.exceptions.NonSuccessfulResponseCodeException; - import java.util.Optional; public class RateLimitException extends NonSuccessfulResponseCodeException { @@ -19,7 +18,7 @@ public class RateLimitException extends NonSuccessfulResponseCodeException { } public RateLimitException(int status, String message, Optional retryAfterMilliseconds) { - super(status, message); + super(status, retryAfterMilliseconds.map(ms -> message + " retry-after: " + ms).orElse(message)); this.retryAfterMilliseconds = retryAfterMilliseconds; }