Signal/Android
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Android.git synced 2026-08-06 13:31:18 +01:00
Code Issues Packages Projects Releases Wiki Activity

Allow partial group sends after SKDM send failures.

Browse Source
This commit is contained in:
Cody Henthorne
2026-07-27 16:12:16 -04:00
committed by Alex Hart
parent 472cd107d4
commit b614f69c64
5 changed files with 54 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/NetworkResultUtil.kt
+1 -1
View File
@@ -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
}
lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java
+45 -43
View File
@@ -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<SignalServiceAddress> workingRecipients = new ArrayList<>(recipients);
List<SendMessageResult> deferredResults = new LinkedList<>();
Set<ServiceId> quarantined = new HashSet<>();
for (int i = 0; i < RETRY_COUNT; i++) {
GroupTargetInfo targetInfo = buildGroupTargetInfo(recipients);
GroupTargetInfo targetInfo = buildGroupTargetInfo(workingRecipients);
final GroupTargetInfo targetInfoSnapshot = targetInfo;
Set<SignalProtocolAddress> sharedWith = aciStore.getSenderKeySharedWith(distributionId);
@@ -2516,7 +2520,7 @@ public class SignalServiceMessageSender {
.collect(Collectors.toList());
List<GroupSendFullToken> needsSenderKeyGroupSendTokens = groupSendEndorsements != null ? groupSendEndorsements.forIndividuals(needsSenderKeyTargets) : null;
List<SealedSenderAccess> needsSenderKeySealedSenderAccesses = SealedSenderAccess.forFanOutGroupSend(needsSenderKeyGroupSendTokens, sealedSenderAccess.getSenderCertificate(), needsSenderKeyAccesses);
List<SealedSenderAccess> needsSenderKeySealedSenderAccesses = SealedSenderAccess.forFanOutGroupSend(needsSenderKeyGroupSendTokens, sealedSenderCertificate, needsSenderKeyAccesses);
List<SendMessageResult> 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<SendMessageResult> 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<ServiceId> failedAddresses = trueFailures.stream()
.map(result -> result.getAddress().getServiceId())
.collect(Collectors.toSet());
workingRecipients = workingRecipients.stream()
.filter(r -> !quarantined.contains(r.getServiceId()))
.collect(Collectors.toList());
List<SendMessageResult> fakeNetworkFailures = recipients.stream()
.filter(r -> !failedAddresses.contains(r.getServiceId()))
.map(SendMessageResult::networkFailure)
.collect(Collectors.toList());
List<SendMessageResult> 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<MultiRecipientMessageResponse, MultiRecipientSendFailure> result = messageApi.sendGroupMessage(ciphertext, multiRecipientAuth, timestamp, online, urgent);
if (result instanceof RequestResult.Success) {
MultiRecipientMessageResponse response = ((RequestResult.Success<MultiRecipientMessageResponse>) result).getResult();
return transformGroupResponseToMessageResults(targetInfo.devices, MessageApiKt.unsentTargets(response), content);
MultiRecipientMessageResponse response = ((RequestResult.Success<MultiRecipientMessageResponse>) result).getResult();
List<SendMessageResult> 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<MultiRecipientSendFailure>) result).getError();
if (error instanceof MismatchedDeviceException) {
MismatchedDeviceException mismatchedDeviceException = (MismatchedDeviceException) error;
Log.w(TAG, "[sendGroupMessage][" + timestamp + "] Handling mismatched devices. (" + mismatchedDeviceException.getMessage() + ")");
List<SendMessageResult> 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<ServiceId> 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<SendMessageResult> networkFailures = recipients.stream()
.filter(r -> !failedAddresses.contains(r.getServiceId()))
.map(SendMessageResult::networkFailure)
.collect(Collectors.toList());
List<SendMessageResult> 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.");
lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/crypto/SealedSenderAccess.kt
-39
View File
@@ -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) {
lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/groupsv2/GroupSendEndorsements.kt
+7 -7
View File
@@ -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<SignalServiceAddress>): 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<SignalServiceAddress>): List<GroupSendFullToken?> {
lib/libsignal-service/src/main/java/org/whispersystems/signalservice/api/push/exceptions/RateLimitException.java
+1 -2
View File
@@ -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<Long> retryAfterMilliseconds) {
super(status, message);
super(status, retryAfterMilliseconds.map(ms -> message + " retry-after: " + ms).orElse(message));
this.retryAfterMilliseconds = retryAfterMilliseconds;
}