diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java index 6ec136df5..44c8ce21c 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java @@ -509,7 +509,7 @@ public class WhisperServerService extends Application LookupUsernameHashResponse.newBuilder() - .setServiceIdentifier(ServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(account.getUuid()))) + .setServiceIdentifier(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(account.getUuid()))) .build()) .orElseGet(() -> LookupUsernameHashResponse.newBuilder().setNotFound(NotFound.getDefaultInstance()).build()); } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcService.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcService.java index cb4bd789f..760782551 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcService.java @@ -13,7 +13,6 @@ import java.util.HexFormat; import java.util.List; import java.util.Objects; import java.util.UUID; -import java.util.concurrent.ThreadLocalRandom; import org.signal.chat.account.ClearRegistrationLockRequest; import org.signal.chat.account.ClearRegistrationLockResponse; import org.signal.chat.account.ConfigureUnidentifiedAccessRequest; @@ -88,8 +87,8 @@ public class AccountsGrpcService extends SimpleAccountsGrpc.AccountsImplBase { final Account account = getAuthenticatedAccount(); final AccountIdentifiers.Builder accountIdentifiersBuilder = AccountIdentifiers.newBuilder() - .addServiceIdentifiers(ServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(account.getUuid()))) - .addServiceIdentifiers(ServiceIdentifierUtil.toGrpcServiceIdentifier(new PniServiceIdentifier(account.getPhoneNumberIdentifier()))) + .addServiceIdentifiers(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(account.getUuid()))) + .addServiceIdentifiers(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(new PniServiceIdentifier(account.getPhoneNumberIdentifier()))) .setE164(account.getNumber()); account.getUsernameHash().ifPresent(usernameHash -> diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ServiceIdentifierUtil.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/GrpcServiceIdentifierUtil.java similarity index 76% rename from service/src/main/java/org/whispersystems/textsecuregcm/grpc/ServiceIdentifierUtil.java rename to service/src/main/java/org/whispersystems/textsecuregcm/grpc/GrpcServiceIdentifierUtil.java index dc8af3c81..86d828a4c 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ServiceIdentifierUtil.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/GrpcServiceIdentifierUtil.java @@ -5,17 +5,15 @@ package org.whispersystems.textsecuregcm.grpc; -import com.google.protobuf.ByteString; -import io.grpc.Status; import java.util.UUID; import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; import org.whispersystems.textsecuregcm.identity.PniServiceIdentifier; import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; import org.whispersystems.textsecuregcm.util.UUIDUtil; -public class ServiceIdentifierUtil { +public class GrpcServiceIdentifierUtil { - private ServiceIdentifierUtil() { + private GrpcServiceIdentifierUtil() { } public static ServiceIdentifier fromGrpcServiceIdentifier(final org.signal.chat.common.ServiceIdentifier serviceIdentifier) { @@ -42,12 +40,4 @@ public class ServiceIdentifierUtil { .setUuid(UUIDUtil.toByteString(serviceIdentifier.uuid())) .build(); } - - public static ByteString toCompactByteString(final ServiceIdentifier serviceIdentifier) { - return ByteString.copyFrom(serviceIdentifier.toCompactByteArray()); - } - - public static ServiceIdentifier fromByteString(final ByteString byteString) { - return ServiceIdentifier.fromBytes(byteString.toByteArray()); - } } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysAnonymousGrpcService.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysAnonymousGrpcService.java index 93dac9430..784bd5eff 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysAnonymousGrpcService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysAnonymousGrpcService.java @@ -13,7 +13,6 @@ import java.util.Arrays; import java.util.concurrent.Flow; import org.signal.chat.errors.FailedUnidentifiedAuthorization; import org.signal.chat.errors.NotFound; -import org.signal.chat.keys.AccountPreKeyBundles; import org.signal.chat.keys.CheckIdentityKeyRequest; import org.signal.chat.keys.CheckIdentityKeyResponse; import org.signal.chat.keys.GetPreKeysAnonymousRequest; @@ -46,7 +45,7 @@ public class KeysAnonymousGrpcService extends SimpleKeysAnonymousGrpc.KeysAnonym @Override public GetPreKeysAnonymousResponse getPreKeys(final GetPreKeysAnonymousRequest request) { final ServiceIdentifier serviceIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getTargetIdentifier()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getTargetIdentifier()); final byte deviceId = request.getRequest().hasDeviceId() ? DeviceIdUtil.validate(request.getRequest().getDeviceId()) @@ -91,7 +90,7 @@ public class KeysAnonymousGrpcService extends SimpleKeysAnonymousGrpc.KeysAnonym @Override public Flow.Publisher checkIdentityKeys(final Flow.Publisher requests) { return JdkFlowAdapter.publisherToFlowPublisher(JdkFlowAdapter.flowPublisherToFlux(requests) - .map(request -> Tuples.of(ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getTargetIdentifier()), + .map(request -> Tuples.of(GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getTargetIdentifier()), request.getFingerprint().toByteArray())) .flatMap(serviceIdentifierAndFingerprint -> Mono.fromFuture( () -> accountsManager.getByServiceIdentifierAsync(serviceIdentifierAndFingerprint.getT1())) @@ -100,7 +99,7 @@ public class KeysAnonymousGrpcService extends SimpleKeysAnonymousGrpc.KeysAnonym .identityType()), serviceIdentifierAndFingerprint.getT2())) .map(account -> CheckIdentityKeyResponse.newBuilder() .setTargetIdentifier( - ServiceIdentifierUtil.toGrpcServiceIdentifier(serviceIdentifierAndFingerprint.getT1())) + GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(serviceIdentifierAndFingerprint.getT1())) .setIdentityKey(ByteString.copyFrom(account.getIdentityKey(serviceIdentifierAndFingerprint.getT1() .identityType()).serialize())) .build()))); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysGrpcService.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysGrpcService.java index 6f5ad167a..59b5094cb 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysGrpcService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/KeysGrpcService.java @@ -96,7 +96,7 @@ public class KeysGrpcService extends SimpleKeysGrpc.KeysImplBase { final AuthenticatedDevice authenticatedDevice = AuthenticationUtil.requireAuthenticatedDevice(); final ServiceIdentifier targetIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getTargetIdentifier()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getTargetIdentifier()); final Optional maybeTargetAccount = accountsManager.getByServiceIdentifier(targetIdentifier); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesAnonymousGrpcService.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesAnonymousGrpcService.java index 457be793e..669498a9c 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesAnonymousGrpcService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesAnonymousGrpcService.java @@ -6,19 +6,18 @@ package org.whispersystems.textsecuregcm.grpc; import com.google.protobuf.ByteString; +import com.google.protobuf.Empty; import java.time.Clock; import java.util.Map; import java.util.Optional; import java.util.stream.Collectors; -import com.google.protobuf.Empty; -import io.grpc.Status; import org.signal.chat.errors.FailedUnidentifiedAuthorization; import org.signal.chat.errors.NotFound; import org.signal.chat.messages.IndividualRecipientMessageBundle; -import org.signal.chat.messages.SendMessageType; import org.signal.chat.messages.MultiRecipientMismatchedDevices; import org.signal.chat.messages.MultiRecipientSuccess; import org.signal.chat.messages.SendMessageResponse; +import org.signal.chat.messages.SendMessageType; import org.signal.chat.messages.SendMultiRecipientMessageRequest; import org.signal.chat.messages.SendMultiRecipientMessageResponse; import org.signal.chat.messages.SendMultiRecipientStoryRequest; @@ -85,7 +84,7 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me throws RateLimitExceededException { final ServiceIdentifier destinationServiceIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getDestination()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getDestination()); final Optional maybeDestination = accountsManager.getByServiceIdentifier(destinationServiceIdentifier); @@ -133,7 +132,7 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me throws RateLimitExceededException { final ServiceIdentifier destinationServiceIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getDestination()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getDestination()); final Optional maybeDestination = accountsManager.getByServiceIdentifier(destinationServiceIdentifier); @@ -194,7 +193,7 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me .setType(MessageProtos.Envelope.Type.UNIDENTIFIED_SENDER) .setClientTimestamp(messages.getTimestamp()) .setServerTimestamp(clock.millis()) - .setDestinationServiceId(destinationServiceIdentifier.toServiceIdentifierString()) + .setDestinationServiceId(destinationServiceIdentifier.toCompactByteString()) .setEphemeral(ephemeral) .setUrgent(urgent) .setContent(entry.getValue().getPayload()); @@ -313,7 +312,7 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me final MultiRecipientSuccess.Builder responseBuilder = MultiRecipientSuccess.newBuilder(); MessageUtil.getUnresolvedRecipients(multiRecipientMessage, resolvedRecipients).stream() - .map(ServiceIdentifierUtil::toGrpcServiceIdentifier) + .map(GrpcServiceIdentifierUtil::toGrpcServiceIdentifier) .forEach(responseBuilder::addUnresolvedRecipients); return SendMultiRecipientMessageResponse.newBuilder().setSuccess(responseBuilder).build(); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcHelper.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcHelper.java index 424469213..ca9c217b6 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcHelper.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcHelper.java @@ -23,7 +23,7 @@ public class MessagesGrpcHelper { final org.whispersystems.textsecuregcm.controllers.MismatchedDevices mismatchedDevices) { final MismatchedDevices.Builder mismatchedDevicesBuilder = MismatchedDevices.newBuilder() - .setServiceIdentifier(ServiceIdentifierUtil.toGrpcServiceIdentifier(serviceIdentifier)); + .setServiceIdentifier(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(serviceIdentifier)); mismatchedDevices.missingDeviceIds().forEach(mismatchedDevicesBuilder::addMissingDevices); mismatchedDevices.extraDeviceIds().forEach(mismatchedDevicesBuilder::addExtraDevices); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcService.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcService.java index 124302def..935e29733 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesGrpcService.java @@ -76,7 +76,7 @@ public class MessagesGrpcService extends SimpleMessagesGrpc.MessagesImplBase { .orElseThrow(() -> GrpcExceptions.invalidCredentials("invalid credentials")); final ServiceIdentifier destinationServiceIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getDestination()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getDestination()); if (sender.isIdentifiedBy(destinationServiceIdentifier)) { throw GrpcExceptions.invalidArguments("use `sendSyncMessage` to send messages to own account"); @@ -159,8 +159,8 @@ public class MessagesGrpcService extends SimpleMessagesGrpc.MessagesImplBase { .setType(getEnvelopeType(entry.getValue().getType())) .setClientTimestamp(messages.getTimestamp()) .setServerTimestamp(clock.millis()) - .setDestinationServiceId(destinationServiceIdentifier.toServiceIdentifierString()) - .setSourceServiceId(new AciServiceIdentifier(sender.accountIdentifier()).toServiceIdentifierString()) + .setDestinationServiceId(destinationServiceIdentifier.toCompactByteString()) + .setSourceServiceId(new AciServiceIdentifier(sender.accountIdentifier()).toCompactByteString()) .setSourceDevice(sender.deviceId()) .setEphemeral(ephemeral) .setUrgent(urgent) diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileAnonymousGrpcService.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileAnonymousGrpcService.java index 9f8cf8c72..d94623c86 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileAnonymousGrpcService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileAnonymousGrpcService.java @@ -63,7 +63,7 @@ public class ProfileAnonymousGrpcService extends SimpleProfileAnonymousGrpc.Prof @Override public GetUnversionedProfileAnonymousResponse getUnversionedProfile(final GetUnversionedProfileAnonymousRequest request) { final ServiceIdentifier targetIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getServiceIdentifier()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getServiceIdentifier()); // Callers must be authenticated to request unversioned profiles by PNI if (targetIdentifier.identityType() == IdentityType.PNI) { @@ -100,7 +100,7 @@ public class ProfileAnonymousGrpcService extends SimpleProfileAnonymousGrpc.Prof @Override public GetVersionedProfileAnonymousResponse getVersionedProfile(final GetVersionedProfileAnonymousRequest request) { - final ServiceIdentifier targetIdentifier = ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getAccountIdentifier()); + final ServiceIdentifier targetIdentifier = GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getAccountIdentifier()); final Optional targetAccount = getTargetAccountAndValidateUnidentifiedAccess(targetIdentifier, request.getUnidentifiedAccessKey().toByteArray()); @@ -123,7 +123,7 @@ public class ProfileAnonymousGrpcService extends SimpleProfileAnonymousGrpc.Prof @Override public GetExpiringProfileKeyCredentialAnonymousResponse getExpiringProfileKeyCredential( final GetExpiringProfileKeyCredentialAnonymousRequest request) { - final ServiceIdentifier targetIdentifier = ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getAccountIdentifier()); + final ServiceIdentifier targetIdentifier = GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getRequest().getAccountIdentifier()); if (request.getRequest().getCredentialType() != CredentialType.CREDENTIAL_TYPE_EXPIRING_PROFILE_KEY) { throw GrpcExceptions.constraintViolation("invalid credential type"); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileGrpcService.java b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileGrpcService.java index 6549eb236..4b3326269 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileGrpcService.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/grpc/ProfileGrpcService.java @@ -211,7 +211,7 @@ public class ProfileGrpcService extends SimpleProfileGrpc.ProfileImplBase { public GetUnversionedProfileResponse getUnversionedProfile(final GetUnversionedProfileRequest request) throws RateLimitExceededException { final AuthenticatedDevice authenticatedDevice = AuthenticationUtil.requireAuthenticatedDevice(); final ServiceIdentifier targetIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getServiceIdentifier()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getServiceIdentifier()); final Optional maybeAccount = validateRateLimitAndGetAccount(authenticatedDevice.accountIdentifier(), targetIdentifier); return maybeAccount.map(account -> GetUnversionedProfileResponse.newBuilder() @@ -228,7 +228,7 @@ public class ProfileGrpcService extends SimpleProfileGrpc.ProfileImplBase { public GetVersionedProfileResponse getVersionedProfile(final GetVersionedProfileRequest request) throws RateLimitExceededException { final AuthenticatedDevice authenticatedDevice = AuthenticationUtil.requireAuthenticatedDevice(); final ServiceIdentifier targetIdentifier = - ServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getAccountIdentifier()); + GrpcServiceIdentifierUtil.fromGrpcServiceIdentifier(request.getAccountIdentifier()); final Optional maybeAccount = validateRateLimitAndGetAccount(authenticatedDevice.accountIdentifier(), targetIdentifier); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/identity/AciServiceIdentifier.java b/service/src/main/java/org/whispersystems/textsecuregcm/identity/AciServiceIdentifier.java index 0cf04fee0..11bc8c218 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/identity/AciServiceIdentifier.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/identity/AciServiceIdentifier.java @@ -5,6 +5,7 @@ package org.whispersystems.textsecuregcm.identity; +import com.google.protobuf.ByteString; import io.micrometer.core.instrument.Metrics; import io.swagger.v3.oas.annotations.media.Schema; import java.nio.ByteBuffer; @@ -62,6 +63,10 @@ public record AciServiceIdentifier(UUID uuid) implements ServiceIdentifier { return new AciServiceIdentifier(UUID.fromString(string)); } + public static AciServiceIdentifier fromByteString(final ByteString byteString) { + return fromBytes(byteString.toByteArray()); + } + public static AciServiceIdentifier fromBytes(final byte[] bytes) { final UUID uuid; diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/identity/ServiceIdentifier.java b/service/src/main/java/org/whispersystems/textsecuregcm/identity/ServiceIdentifier.java index 341a7f0eb..4b84516de 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/identity/ServiceIdentifier.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/identity/ServiceIdentifier.java @@ -5,6 +5,7 @@ package org.whispersystems.textsecuregcm.identity; +import com.google.protobuf.ByteString; import io.swagger.v3.oas.annotations.media.Schema; import java.util.UUID; import org.signal.libsignal.protocol.ServiceId; @@ -49,6 +50,15 @@ public sealed interface ServiceIdentifier permits AciServiceIdentifier, PniServi */ byte[] toCompactByteArray(); + /** + * Returns a compact binary representation of this account identifier as a {@link ByteString}. + * + * @return a binary representation of this account identifier + */ + default ByteString toCompactByteString() { + return ByteString.copyFrom(toCompactByteArray()); + } + /** * Returns a fixed-width binary representation of this account identifier. * @@ -56,11 +66,20 @@ public sealed interface ServiceIdentifier permits AciServiceIdentifier, PniServi */ byte[] toFixedWidthByteArray(); + /** + * Returns a fixed-width binary representation of this account identifier as a {@link ByteString}. + * + * @return a binary representation of this account identifier + */ + default ByteString toFixedWidthByteString() { + return ByteString.copyFrom(toFixedWidthByteArray()); + } + /** * Parse a service identifier string, which should be a plain UUID string for ACIs and a prefixed UUID string for PNIs * - * @param string A service identifier string - * @return The parsed {@link ServiceIdentifier} + * @param string a service identifier string + * @return the parsed service identifier */ static ServiceIdentifier valueOf(final String string) { try { @@ -70,6 +89,12 @@ public sealed interface ServiceIdentifier permits AciServiceIdentifier, PniServi } } + /** + * Parse a service identifier from a byte array. + * + * @param bytes the byte array from which to parse a service identifier + * @return the parsed service identifier + */ static ServiceIdentifier fromBytes(final byte[] bytes) { try { return AciServiceIdentifier.fromBytes(bytes); @@ -78,6 +103,16 @@ public sealed interface ServiceIdentifier permits AciServiceIdentifier, PniServi } } + /** + * Parse a service identifier from a byte string. + * + * @param byteString the byte string from which to parse a service identifier + * @return the parsed service identifier + */ + static ServiceIdentifier fromByteString(final ByteString byteString) { + return fromBytes(byteString.toByteArray()); + } + static ServiceIdentifier fromLibsignal(final ServiceId libsignalServiceId) { if (libsignalServiceId instanceof ServiceId.Aci) { return new AciServiceIdentifier(libsignalServiceId.getRawUUID()); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/metrics/MessageMetrics.java b/service/src/main/java/org/whispersystems/textsecuregcm/metrics/MessageMetrics.java index 58f737697..05813eaf8 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/metrics/MessageMetrics.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/metrics/MessageMetrics.java @@ -15,6 +15,7 @@ import io.micrometer.core.instrument.Timer; import java.time.Duration; import java.time.Instant; import java.util.ArrayList; +import java.util.HexFormat; import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,9 +47,10 @@ public final class MessageMetrics { final MessageProtos.Envelope envelope) { if (envelope.hasDestinationServiceId()) { try { - measureAccountDestinationUuidMismatches(account, ServiceIdentifier.valueOf(envelope.getDestinationServiceId())); + measureAccountDestinationUuidMismatches(account, ServiceIdentifier.fromByteString(envelope.getDestinationServiceId())); } catch (final IllegalArgumentException ignored) { - logger.warn("Envelope had invalid destination UUID: {}", envelope.getDestinationServiceId()); + logger.warn("Envelope had invalid destination service ID: {}", + HexFormat.of().formatHex(envelope.getDestinationServiceId().toByteArray())); } } } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java b/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java index ca8625a68..9a2ce91d6 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/push/MessageSender.java @@ -4,8 +4,8 @@ */ package org.whispersystems.textsecuregcm.push; -import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name; import static org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope; +import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name; import com.google.common.annotations.VisibleForTesting; import io.dropwizard.util.DataSize; @@ -24,7 +24,6 @@ import java.util.concurrent.CompletableFuture; import java.util.stream.Collectors; import javax.annotation.Nullable; import kotlin.Pair; -import org.apache.commons.lang3.StringUtils; import org.signal.libsignal.protocol.SealedSenderMultiRecipientMessage; import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration; import org.whispersystems.textsecuregcm.controllers.MessageDeliveryNotAllowedException; @@ -327,15 +326,15 @@ public class MessageSender { final byte excludedDeviceId; if (syncMessageSenderDeviceId.isPresent()) { - if (messagesByDeviceId.values().stream().anyMatch(message -> StringUtils.isBlank(message.getSourceServiceId()) || - !destination.isIdentifiedBy(ServiceIdentifier.valueOf(message.getSourceServiceId())))) { + if (messagesByDeviceId.values().stream().anyMatch(message -> !message.hasSourceServiceId() || + !destination.isIdentifiedBy(ServiceIdentifier.fromByteString(message.getSourceServiceId())))) { throw new IllegalArgumentException("Sync message sender device ID specified, but one or more messages are not addressed to sender"); } excludedDeviceId = syncMessageSenderDeviceId.get(); } else { - if (messagesByDeviceId.values().stream().anyMatch(message -> StringUtils.isNotBlank(message.getSourceServiceId()) && - destination.isIdentifiedBy(ServiceIdentifier.valueOf(message.getSourceServiceId())))) { + if (messagesByDeviceId.values().stream().anyMatch(message -> message.hasSourceServiceId() && + destination.isIdentifiedBy(ServiceIdentifier.fromByteString(message.getSourceServiceId())))) { throw new IllegalArgumentException("Sync message sender device ID not specified, but one or more messages are addressed to sender"); } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/push/ReceiptSender.java b/service/src/main/java/org/whispersystems/textsecuregcm/push/ReceiptSender.java index 2f34a1ba4..9fd934e07 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/push/ReceiptSender.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/push/ReceiptSender.java @@ -44,9 +44,9 @@ public class ReceiptSender { destinationAccount -> { final Envelope message = Envelope.newBuilder() .setServerTimestamp(System.currentTimeMillis()) - .setSourceServiceId(sourceIdentifier.toServiceIdentifierString()) + .setSourceServiceId(sourceIdentifier.toCompactByteString()) .setSourceDevice(sourceDeviceId) - .setDestinationServiceId(destinationIdentifier.toServiceIdentifierString()) + .setDestinationServiceId(destinationIdentifier.toCompactByteString()) .setClientTimestamp(messageId) .setType(Envelope.Type.SERVER_DELIVERY_RECEIPT) .setUrgent(false) diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManager.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManager.java index 935501165..6b1177b02 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManager.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManager.java @@ -40,6 +40,7 @@ import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil; import org.whispersystems.textsecuregcm.push.MessageSender; import org.whispersystems.textsecuregcm.push.MessageTooLargeException; import org.whispersystems.textsecuregcm.util.Pair; +import org.whispersystems.textsecuregcm.util.UUIDUtil; public class ChangeNumberManager { @@ -159,10 +160,9 @@ public class ChangeNumberManager { try { // Now that we've actually updated the account, populate the "updated PNI" field on all envelopes - final String updatedPniString = updatedAccount.getIdentifier(IdentityType.PNI).toString(); - messagesByDeviceId.replaceAll((_, envelope) -> - envelope.toBuilder().setUpdatedPni(updatedPniString).build()); + envelope.toBuilder().setUpdatedPni(UUIDUtil.toByteString(updatedAccount.getIdentifier(IdentityType.PNI))) + .build()); messageSender.sendMessages(updatedAccount, serviceIdentifier, diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/EnvelopeUtil.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/EnvelopeUtil.java deleted file mode 100644 index a22bb1a11..000000000 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/EnvelopeUtil.java +++ /dev/null @@ -1,135 +0,0 @@ -/* - * Copyright 2025 Signal Messenger, LLC - * SPDX-License-Identifier: AGPL-3.0-only - */ - -package org.whispersystems.textsecuregcm.storage; - -import java.util.UUID; -import com.google.common.annotations.VisibleForTesting; -import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; -import org.whispersystems.textsecuregcm.grpc.ServiceIdentifierUtil; -import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; -import org.whispersystems.textsecuregcm.util.UUIDUtil; - -/** - * Provides utility methods for "compressing" and "expanding" envelopes. Historically UUID-like fields in envelopes have - * been represented as strings (e.g. "c15f1dfb-ae2c-43a8-9bb9-baba97ac416c"), but could be represented as more - * compact byte arrays instead. Existing clients generally expect string representations (though that should change in - * the near future), but we can use the more compressed forms at rest for more efficient storage and transfer. - */ -public class EnvelopeUtil { - - @VisibleForTesting - static final String INCLUDE_BINARY_SERVICE_ID_EXPERIMENT_NAME = "envelopeIncludeBinaryServiceIdentifier"; - - /** - * Converts all "compressible" UUID-like fields in the given envelope to more compact binary representations. - * - * @param envelope the envelope to compress - * - * @return an envelope with string-based UUID-like fields compressed to binary representations - */ - public static MessageProtos.Envelope compress(final MessageProtos.Envelope envelope) { - final MessageProtos.Envelope.Builder builder = envelope.toBuilder(); - - if (builder.hasSourceServiceId()) { - final ServiceIdentifier sourceServiceId = ServiceIdentifier.valueOf(builder.getSourceServiceId()); - - builder.setSourceServiceIdBinary(ServiceIdentifierUtil.toCompactByteString(sourceServiceId)); - builder.clearSourceServiceId(); - } - - if (builder.hasDestinationServiceId()) { - final ServiceIdentifier destinationServiceId = ServiceIdentifier.valueOf(builder.getDestinationServiceId()); - - builder.setDestinationServiceIdBinary(ServiceIdentifierUtil.toCompactByteString(destinationServiceId)); - builder.clearDestinationServiceId(); - } - - if (builder.hasServerGuid()) { - final UUID serverGuid = UUID.fromString(builder.getServerGuid()); - - builder.setServerGuidBinary(UUIDUtil.toByteString(serverGuid)); - builder.clearServerGuid(); - } - - if (builder.hasUpdatedPni()) { - final UUID updatedPni = UUID.fromString(builder.getUpdatedPni()); - - builder.setUpdatedPniBinary(UUIDUtil.toByteString(updatedPni)); - builder.clearUpdatedPni(); - } - - return builder.build(); - } - - /** - * "Expands" all binary representations of UUID-like fields to string representations to meet current client - * expectations. - * - * @param envelope the envelope to expand - * - * @return an envelope with binary representations of UUID-like fields expanded to string representations - */ - public static MessageProtos.Envelope expand(final MessageProtos.Envelope envelope, - final ExperimentEnrollmentManager experimentEnrollmentManager) { - - final boolean includeBinaryServiceIdentifiers; - - if (envelope.hasDestinationServiceIdBinary() || envelope.hasDestinationServiceId()) { - final ServiceIdentifier destinationIdentifier = envelope.hasDestinationServiceIdBinary() - ? ServiceIdentifier.fromBytes(envelope.getDestinationServiceIdBinary().toByteArray()) - : ServiceIdentifier.valueOf(envelope.getDestinationServiceId()); - - includeBinaryServiceIdentifiers = - experimentEnrollmentManager.isEnrolled(destinationIdentifier.uuid(), INCLUDE_BINARY_SERVICE_ID_EXPERIMENT_NAME); - } else { - includeBinaryServiceIdentifiers = false; - } - - final MessageProtos.Envelope.Builder builder = envelope.toBuilder(); - - if (builder.hasSourceServiceIdBinary()) { - final ServiceIdentifier sourceServiceId = - ServiceIdentifierUtil.fromByteString(builder.getSourceServiceIdBinary()); - - builder.setSourceServiceId(sourceServiceId.toServiceIdentifierString()); - - if (!includeBinaryServiceIdentifiers) { - builder.clearSourceServiceIdBinary(); - } - } - - if (builder.hasDestinationServiceIdBinary()) { - final ServiceIdentifier destinationServiceId = - ServiceIdentifierUtil.fromByteString(builder.getDestinationServiceIdBinary()); - - builder.setDestinationServiceId(destinationServiceId.toServiceIdentifierString()); - - if (!includeBinaryServiceIdentifiers) { - builder.clearDestinationServiceIdBinary(); - } - } - - if (builder.hasServerGuidBinary()) { - final UUID serverGuid = UUIDUtil.fromByteString(builder.getServerGuidBinary()); - - builder.setServerGuid(serverGuid.toString()); - - if (!includeBinaryServiceIdentifiers) { - builder.clearServerGuidBinary(); - } - } - - if (builder.hasUpdatedPniBinary()) { - final UUID updatedPni = UUIDUtil.fromByteString(builder.getUpdatedPniBinary()); - - // Note that expanded envelopes include BOTH forms of the `updatedPni` field - builder.setUpdatedPni(updatedPni.toString()); - } - - return builder.build(); - } -} diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagePersister.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagePersister.java index 3eb38a615..fa941a713 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagePersister.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagePersister.java @@ -32,6 +32,7 @@ import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfigurati import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.identity.IdentityType; import org.whispersystems.textsecuregcm.metrics.DevicePlatformUtil; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import org.whispersystems.textsecuregcm.util.Util; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -366,7 +367,7 @@ public class MessagePersister implements Managed { TRIMMED_MESSAGE_BYTES_COUNTER.increment(envelope.getSerializedSize()); return Mono .fromCompletionStage(() -> messagesManager - .delete(aci, device, UUID.fromString(envelope.getServerGuid()), envelope.getServerTimestamp())) + .delete(aci, device, UUIDUtil.fromByteString(envelope.getServerGuid()), envelope.getServerTimestamp())) .retryWhen(retryBackoffSpec) .map(Optional::isPresent); }) diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java index 40216bac0..a306538bc 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java @@ -48,12 +48,12 @@ import org.signal.libsignal.protocol.ServiceId; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; import org.whispersystems.textsecuregcm.metrics.MetricsUtil; import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClusterClient; import org.whispersystems.textsecuregcm.util.Pair; import org.whispersystems.textsecuregcm.util.ResilienceUtil; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.core.observability.micrometer.Micrometer; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -121,7 +121,6 @@ public class MessagesCache { private final ExecutorService messageDeletionExecutorService; // messageDeletionExecutorService wrapped into a reactor Scheduler private final Scheduler messageDeletionScheduler; - private final ExperimentEnrollmentManager experimentEnrollmentManager; private final MessagesCacheInsertScript insertScript; private final MessagesCacheInsertSharedMultiRecipientPayloadAndViewsScript insertMrmScript; @@ -175,16 +174,13 @@ public class MessagesCache { final Scheduler messageDeliveryScheduler, final ExecutorService messageDeletionExecutorService, final ScheduledExecutorService retryExecutor, - final Clock clock, - final ExperimentEnrollmentManager experimentEnrollmentManager) - throws IOException { + final Clock clock) throws IOException { this( redisCluster, messageDeliveryScheduler, messageDeletionExecutorService, clock, - experimentEnrollmentManager, new MessagesCacheInsertScript(redisCluster, retryExecutor), new MessagesCacheInsertSharedMultiRecipientPayloadAndViewsScript(redisCluster, retryExecutor), new MessagesCacheGetItemsScript(redisCluster), @@ -200,7 +196,6 @@ public class MessagesCache { MessagesCache(final FaultTolerantRedisClusterClient redisCluster, final Scheduler messageDeliveryScheduler, final ExecutorService messageDeletionExecutorService, final Clock clock, - final ExperimentEnrollmentManager experimentEnrollmentManager, final MessagesCacheInsertScript insertScript, final MessagesCacheInsertSharedMultiRecipientPayloadAndViewsScript insertMrmScript, final MessagesCacheGetItemsScript getItemsScript, @@ -216,7 +211,6 @@ public class MessagesCache { this.messageDeliveryScheduler = messageDeliveryScheduler; this.messageDeletionExecutorService = messageDeletionExecutorService; this.messageDeletionScheduler = Schedulers.fromExecutorService(messageDeletionExecutorService, "messageDeletion"); - this.experimentEnrollmentManager = experimentEnrollmentManager; this.insertScript = insertScript; this.insertMrmScript = insertMrmScript; @@ -233,7 +227,7 @@ public class MessagesCache { final byte destinationDeviceId, final MessageProtos.Envelope message) { - final MessageProtos.Envelope messageWithGuid = message.toBuilder().setServerGuid(messageGuid.toString()).build(); + final MessageProtos.Envelope messageWithGuid = message.toBuilder().setServerGuid(UUIDUtil.toByteString(messageGuid)).build(); final Timer.Sample sample = Timer.start(); return insertScript.executeAsync(destinationAccountIdentifier, destinationDeviceId, messageWithGuid) @@ -278,7 +272,7 @@ public class MessagesCache { removedMessages.add(RemovedMessage.fromEnvelope(envelope)); if (envelope.hasSharedMrmKey()) { serviceIdentifierToMrmKeys.computeIfAbsent( - ServiceIdentifier.valueOf(envelope.getDestinationServiceId()), _ -> new ArrayList<>()) + ServiceIdentifier.fromByteString(envelope.getDestinationServiceId()), _ -> new ArrayList<>()) .add(envelope.getSharedMrmKey().toByteArray()); } } catch (final InvalidProtocolBufferException e) { @@ -398,7 +392,7 @@ public class MessagesCache { private void discardStaleMessages(final UUID destinationUuid, final byte destinationDevice, Flux staleMessages, final Counter counter, final String context) { staleMessages - .map(e -> UUID.fromString(e.getServerGuid())) + .map(e -> UUIDUtil.fromByteString(e.getServerGuid())) .buffer(PAGE_SIZE) .subscribeOn(messageDeletionScheduler) .subscribe(messageGuids -> @@ -479,7 +473,7 @@ public class MessagesCache { final byte[] key = mrmMessage.getSharedMrmKey().toByteArray(); final byte[] sharedMrmViewKey = MessagesCache.getSharedMrmViewKey( // the message might be addressed to the account's PNI, so use the service ID from the envelope - ServiceIdentifier.valueOf(mrmMessage.getDestinationServiceId()), destinationDevice); + ServiceIdentifier.fromByteString(mrmMessage.getDestinationServiceId()), destinationDevice); return Mono.from(redisCluster.withBinaryCluster( conn -> conn.reactive().hmget(key, "data".getBytes(StandardCharsets.UTF_8), sharedMrmViewKey) @@ -670,10 +664,10 @@ public class MessagesCache { try { final MessageProtos.Envelope message = parseEnvelope(serialized); - processedMessages.add(message.getServerGuid()); + processedMessages.add(UUIDUtil.fromByteString(message.getServerGuid()).toString()); if (message.hasSharedMrmKey()) { - serviceIdentifierToMrmKeys.computeIfAbsent(ServiceIdentifier.valueOf(message.getDestinationServiceId()), + serviceIdentifierToMrmKeys.computeIfAbsent(ServiceIdentifier.fromByteString(message.getDestinationServiceId()), _ -> new ArrayList<>()) .add(message.getSharedMrmKey().toByteArray()); } @@ -782,6 +776,6 @@ public class MessagesCache { private MessageProtos.Envelope parseEnvelope(final byte[] envelopeBytes) throws InvalidProtocolBufferException { - return EnvelopeUtil.expand(MessageProtos.Envelope.parseFrom(envelopeBytes), experimentEnrollmentManager); + return MessageProtos.Envelope.parseFrom(envelopeBytes); } } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScript.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScript.java index d832d348c..3496597d3 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScript.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScript.java @@ -21,6 +21,7 @@ import org.whispersystems.textsecuregcm.push.RedisMessageAvailabilityManager; import org.whispersystems.textsecuregcm.redis.ClusterLuaScript; import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClusterClient; import org.whispersystems.textsecuregcm.util.ResilienceUtil; +import org.whispersystems.textsecuregcm.util.UUIDUtil; /** * Inserts an envelope into the message queue for a destination device and publishes a "new message available" event. @@ -62,8 +63,8 @@ class MessagesCacheInsertScript { ); final List args = new ArrayList<>(Arrays.asList( - EnvelopeUtil.compress(envelope).toByteArray(), // message - envelope.getServerGuid().getBytes(StandardCharsets.UTF_8), // guid + envelope.toByteArray(), // message + UUIDUtil.fromByteString(envelope.getServerGuid()).toString().getBytes(StandardCharsets.UTF_8), // guid NEW_MESSAGE_EVENT_BYTES // eventPayload )); diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDb.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDb.java index 0e14961b5..14314d37c 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDb.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDb.java @@ -27,8 +27,8 @@ import org.reactivestreams.Publisher; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; import org.whispersystems.textsecuregcm.util.AttributeValues; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.core.publisher.Flux; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient; @@ -63,13 +63,11 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore { private final String tableName; private final Duration timeToLive; private final ExecutorService messageDeletionExecutor; - private final ExperimentEnrollmentManager experimentEnrollmentManager; private static final Logger logger = LoggerFactory.getLogger(MessagesDynamoDb.class); public MessagesDynamoDb(DynamoDbClient dynamoDb, DynamoDbAsyncClient dynamoDbAsyncClient, String tableName, - Duration timeToLive, ExecutorService messageDeletionExecutor, - final ExperimentEnrollmentManager experimentEnrollmentManager) { + Duration timeToLive, ExecutorService messageDeletionExecutor) { super(dynamoDb); this.dbAsyncClient = dynamoDbAsyncClient; @@ -77,7 +75,6 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore { this.timeToLive = timeToLive; this.messageDeletionExecutor = messageDeletionExecutor; - this.experimentEnrollmentManager = experimentEnrollmentManager; } public void store(final List messages, final UUID destinationAccountUuid, @@ -94,14 +91,14 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore { final AttributeValue partitionKey = convertPartitionKey(destinationAccountUuid, destinationDevice); List writeItems = new ArrayList<>(); for (MessageProtos.Envelope message : messages) { - final UUID messageUuid = UUID.fromString(message.getServerGuid()); + final UUID messageUuid = UUIDUtil.fromByteString(message.getServerGuid()); final ImmutableMap.Builder item = ImmutableMap.builder() .put(KEY_PARTITION, partitionKey) .put(KEY_SORT, convertSortKey(message.getServerTimestamp(), messageUuid)) .put(LOCAL_INDEX_MESSAGE_UUID_KEY_SORT, convertLocalIndexMessageUuidSortKey(messageUuid)) .put(KEY_TTL, AttributeValues.fromLong(getTtlForMessage(message))) - .put(KEY_ENVELOPE_BYTES, AttributeValue.builder().b(SdkBytes.fromByteArray(EnvelopeUtil.compress(message).toByteArray())).build()); + .put(KEY_ENVELOPE_BYTES, AttributeValue.builder().b(SdkBytes.fromByteArray(message.toByteArray())).build()); writeItems.add(WriteRequest.builder().putRequest(PutRequest.builder() .item(item.build()) @@ -147,7 +144,7 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore { return dbAsyncClient.queryPaginator(queryRequest).items() .map(message -> { try { - return convertItemToEnvelope(message, experimentEnrollmentManager); + return convertItemToEnvelope(message); } catch (final InvalidProtocolBufferException e) { logger.error("Failed to parse envelope", e); return null; @@ -167,7 +164,7 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore { .thenApplyAsync(deleteItemResponse -> { if (deleteItemResponse.attributes() != null && deleteItemResponse.attributes().containsKey(KEY_PARTITION)) { try { - return Optional.of(convertItemToEnvelope(deleteItemResponse.attributes(), experimentEnrollmentManager)); + return Optional.of(convertItemToEnvelope(deleteItemResponse.attributes())); } catch (final InvalidProtocolBufferException e) { logger.error("Failed to parse envelope", e); } @@ -178,11 +175,10 @@ public class MessagesDynamoDb extends AbstractDynamoDbStore { } @VisibleForTesting - static MessageProtos.Envelope convertItemToEnvelope(final Map item, - final ExperimentEnrollmentManager experimentEnrollmentManager) throws InvalidProtocolBufferException { + static MessageProtos.Envelope convertItemToEnvelope(final Map item) + throws InvalidProtocolBufferException { - return EnvelopeUtil.expand(MessageProtos.Envelope.parseFrom(item.get(KEY_ENVELOPE_BYTES).b().asByteArray()), - experimentEnrollmentManager); + return MessageProtos.Envelope.parseFrom(item.get(KEY_ENVELOPE_BYTES).b().asByteArray()); } private long getTtlForMessage(MessageProtos.Envelope message) { diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java index 7abf50104..b58c82729 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java @@ -34,6 +34,7 @@ import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; import org.whispersystems.textsecuregcm.metrics.MetricsUtil; import org.whispersystems.textsecuregcm.push.RedisMessageAvailabilityManager; import org.whispersystems.textsecuregcm.util.Pair; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.core.observability.micrometer.Micrometer; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -102,10 +103,14 @@ public class MessagesManager { return messagesCache.insert(messageGuid, accountIdentifier, deviceId, message) .thenAccept(present -> { - if (message.hasSourceServiceId() && !accountIdentifier.toString() - .equals(message.getSourceServiceId())) { - // Note that this is an asynchronous, best-effort, fire-and-forget operation - reportMessageManager.store(message.getSourceServiceId(), messageGuid); + if (message.hasSourceServiceId()) { + final ServiceIdentifier sourceServiceIdentifier = + ServiceIdentifier.fromByteString(message.getSourceServiceId()); + + if (!accountIdentifier.equals(sourceServiceIdentifier.uuid())) { + // Note that this is an asynchronous, best-effort, fire-and-forget operation + reportMessageManager.store(sourceServiceIdentifier.toServiceIdentifierString(), messageGuid); + } } devicePresenceById.put(deviceId, present); @@ -170,8 +175,8 @@ public class MessagesManager { return insertAsync(resolvedRecipients.get(recipient).getIdentifier(IdentityType.ACI), IntStream.range(0, devices.length).mapToObj(i -> devices[i]) - .collect(Collectors.toMap(deviceId -> deviceId, deviceId -> prototypeMessage.toBuilder() - .setDestinationServiceId(serviceIdentifier.toServiceIdentifierString()) + .collect(Collectors.toMap(deviceId -> deviceId, _ -> prototypeMessage.toBuilder() + .setDestinationServiceId(serviceIdentifier.toCompactByteString()) .build()))) .thenAccept(clientPresenceByDeviceId -> clientPresenceByAccountAndDevice.put(resolvedRecipients.get(recipient), @@ -275,7 +280,7 @@ public class MessagesManager { messagesDynamoDb.store(messages, destinationUuid, destinationDevice); - final List messageGuids = messages.stream().map(message -> UUID.fromString(message.getServerGuid())) + final List messageGuids = messages.stream().map(message -> UUIDUtil.fromByteString(message.getServerGuid())) .collect(Collectors.toList()); int messagesRemovedFromCache = 0; try { diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStream.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStream.java index 125182aa9..f567b4141 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStream.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStream.java @@ -11,6 +11,7 @@ import java.util.concurrent.Flow; import com.google.common.annotations.VisibleForTesting; import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.push.RedisMessageAvailabilityManager; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import org.whispersystems.textsecuregcm.util.Util; /// A [MessageStream] implementation that produces message from a joint DynamoDB/Redis message store. @@ -58,7 +59,7 @@ public class RedisDynamoDbMessageStream implements MessageStream { @Override public CompletableFuture acknowledgeMessage(final MessageProtos.Envelope message) { - final UUID guid = UUID.fromString(message.getServerGuid()); + final UUID guid = UUIDUtil.fromByteString(message.getServerGuid()); return messagesCache.remove(accountIdentifier, device.getId(), guid) .thenCompose(removed -> removed.map(_ -> CompletableFuture.completedFuture(null)) diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/RemovedMessage.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/RemovedMessage.java index b37dc8248..0e5da9db1 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/RemovedMessage.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/RemovedMessage.java @@ -10,6 +10,7 @@ import java.util.Optional; import java.util.UUID; import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; +import org.whispersystems.textsecuregcm.util.UUIDUtil; public record RemovedMessage(Optional sourceServiceId, ServiceIdentifier destinationServiceId, @VisibleForTesting UUID serverGuid, long serverTimestamp, long clientTimestamp, @@ -18,10 +19,10 @@ public record RemovedMessage(Optional sourceServiceId, Servic public static RemovedMessage fromEnvelope(MessageProtos.Envelope envelope) { return new RemovedMessage( envelope.hasSourceServiceId() - ? Optional.of(ServiceIdentifier.valueOf(envelope.getSourceServiceId())) + ? Optional.of(ServiceIdentifier.fromByteString(envelope.getSourceServiceId())) : Optional.empty(), - ServiceIdentifier.valueOf(envelope.getDestinationServiceId()), - UUID.fromString(envelope.getServerGuid()), + ServiceIdentifier.fromByteString(envelope.getDestinationServiceId()), + UUIDUtil.fromByteString(envelope.getServerGuid()), envelope.getServerTimestamp(), envelope.getClientTimestamp(), envelope.getType() diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStream.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStream.java index df8d13d99..f97958585 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStream.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStream.java @@ -160,7 +160,7 @@ public class FoundationDbMessageStream implements MessageStream { @Override public CompletableFuture acknowledgeMessage(final MessageProtos.Envelope message) { acknowledgedMessageBuffer.acknowledgeMessage( - messageGuidCodec.decodeMessageGuid(UUIDUtil.fromByteString(message.getServerGuidBinary()))); + messageGuidCodec.decodeMessageGuid(UUIDUtil.fromByteString(message.getServerGuid()))); return CompletableFuture.completedFuture(null); } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStreamEntry.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStreamEntry.java index badec3ca5..ac64909ae 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStreamEntry.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStreamEntry.java @@ -16,7 +16,7 @@ sealed interface FoundationDbMessageStreamEntry permits FoundationDbMessageStrea return switch (this) { case Message(final Versionstamp versionstamp, final MessageProtos.Envelope partialEnvelope): { yield new MessageStreamEntry.Envelope(partialEnvelope.toBuilder() - .setServerGuidBinary(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp))) + .setServerGuid(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp))) .build()); } case QueueEmpty(): diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java b/service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java index 6c6d72576..83c21667a 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java @@ -15,9 +15,9 @@ import io.micrometer.core.instrument.Tags; import io.micrometer.core.instrument.Timer; import java.time.Duration; import java.util.Collections; +import java.util.HexFormat; import java.util.List; import java.util.Optional; -import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @@ -44,11 +44,11 @@ import org.whispersystems.textsecuregcm.storage.Account; import org.whispersystems.textsecuregcm.storage.ClientReleaseManager; import org.whispersystems.textsecuregcm.storage.ConflictingMessageConsumerException; import org.whispersystems.textsecuregcm.storage.Device; -import org.whispersystems.textsecuregcm.storage.EnvelopeUtil; import org.whispersystems.textsecuregcm.storage.MessageStream; import org.whispersystems.textsecuregcm.storage.MessageStreamEntry; import org.whispersystems.textsecuregcm.storage.MessagesManager; import org.whispersystems.textsecuregcm.util.HeaderUtils; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import org.whispersystems.websocket.WebSocketClient; import org.whispersystems.websocket.WebSocketResourceProvider; import org.whispersystems.websocket.messages.WebSocketResponseMessage; @@ -93,8 +93,6 @@ public class WebSocketConnection implements DisconnectionRequestListener { private static final Duration CLOSE_WITH_PENDING_MESSAGES_NOTIFICATION_DELAY = Duration.ofMinutes(1); - private static final String RECOMPRESS_ENVELOPE_EXPERIMENT_NAME = "recompressEnvelopes"; - private static final Logger logger = LoggerFactory.getLogger(WebSocketConnection.class); private final ReceiptSender receiptSender; @@ -173,7 +171,7 @@ public class WebSocketConnection implements DisconnectionRequestListener { if (hasSentFirstMessage.compareAndSet(false, true)) { messageDeliveryLoopMonitor.recordDeliveryAttempt(authenticatedAccount.getIdentifier(IdentityType.ACI), authenticatedDevice.getId(), - UUID.fromString(message.getServerGuid()), + UUIDUtil.fromByteString(message.getServerGuid()), client.getUserAgent(), "websocket"); } @@ -241,11 +239,7 @@ public class WebSocketConnection implements DisconnectionRequestListener { return messageStream.acknowledgeMessage(message); } - final boolean recompressEnvelope = - experimentEnrollmentManager.isEnrolled(authenticatedAccount.getIdentifier(IdentityType.ACI), RECOMPRESS_ENVELOPE_EXPERIMENT_NAME); - - final Optional body = - Optional.of(serializeMessage(recompressEnvelope ? EnvelopeUtil.compress(message) : message)); + final Optional body = Optional.of(serializeMessage(message)); sendMessageCounter.increment(); sentMessageCounter.increment(); @@ -308,11 +302,11 @@ public class WebSocketConnection implements DisconnectionRequestListener { } try { - receiptSender.sendReceipt(ServiceIdentifier.valueOf(message.getDestinationServiceId()), - authenticatedDevice.getId(), AciServiceIdentifier.valueOf(message.getSourceServiceId()), + receiptSender.sendReceipt(ServiceIdentifier.fromByteString(message.getDestinationServiceId()), + authenticatedDevice.getId(), AciServiceIdentifier.fromByteString(message.getSourceServiceId()), message.getClientTimestamp()); } catch (final IllegalArgumentException e) { - logger.error("Could not parse UUID: {}", message.getSourceServiceId()); + logger.error("Could not parse UUID: {}", HexFormat.of().formatHex(message.getSourceServiceId().toByteArray())); } catch (final Exception e) { logger.warn("Failed to send receipt", e); } diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java b/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java index 2a3610433..605024598 100644 --- a/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java +++ b/service/src/main/java/org/whispersystems/textsecuregcm/workers/CommandDependencies.java @@ -245,7 +245,7 @@ public record CommandDependencies( MessagesDynamoDb messagesDynamoDb = new MessagesDynamoDb(dynamoDbClient, dynamoDbAsyncClient, configuration.getDynamoDbTables().getMessages().getTableName(), configuration.getDynamoDbTables().getMessages().getExpiration(), - messageDeletionExecutor, experimentEnrollmentManager); + messageDeletionExecutor); FaultTolerantRedisClusterClient messagesCluster = configuration.getMessageCacheConfiguration() .getRedisClusterConfiguration().build("messages", redisClientResourcesBuilder); FaultTolerantRedisClusterClient rateLimitersCluster = configuration.getRateLimitersCluster().build("rate_limiters", @@ -267,7 +267,7 @@ public record CommandDependencies( DisconnectionRequestManager disconnectionRequestManager = new DisconnectionRequestManager(pubsubClient, disconnectionRequestListenerExecutor, retryExecutor); MessagesCache messagesCache = new MessagesCache(messagesCluster, - messageDeliveryScheduler, messageDeletionExecutor, retryExecutor, Clock.systemUTC(), experimentEnrollmentManager); + messageDeliveryScheduler, messageDeletionExecutor, retryExecutor, Clock.systemUTC()); ProfilesManager profilesManager = new ProfilesManager(profilesV1, profiles, cacheCluster, retryExecutor, asyncCdnS3Client, configuration.getCdnConfiguration().bucket()); ReportMessageDynamoDb reportMessageDynamoDb = new ReportMessageDynamoDb(dynamoDbClient, dynamoDbAsyncClient, diff --git a/service/src/main/proto/TextSecure.proto b/service/src/main/proto/TextSecure.proto index 7f578dc8b..4a8845f4b 100644 --- a/service/src/main/proto/TextSecure.proto +++ b/service/src/main/proto/TextSecure.proto @@ -21,25 +21,26 @@ message Envelope { PLAINTEXT_CONTENT = 8; // for decryption error receipts } + reserved 9; // Formerly string server_guid; + reserved 11; // Formerly string source_service_id + reserved 13; // Formerly string destination_service_id + reserved 15; // Formerly string updated_pni; + optional Type type = 1; - optional string source_service_id = 11; optional uint32 source_device = 7; optional uint64 client_timestamp = 5; optional bytes content = 8; // Contains an encrypted Content - optional string server_guid = 9; optional uint64 server_timestamp = 10; optional bool ephemeral = 12; // indicates that the message should not be persisted if the recipient is offline - optional string destination_service_id = 13; optional bool urgent = 14 [default=true]; - optional string updated_pni = 15; optional bool story = 16; // indicates that the content is a story. optional bytes report_spam_token = 17; // token sent when reporting spam optional bytes shared_mrm_key = 18; // indicates content should be fetched from multi-recipient message datastore - optional bytes source_service_id_binary = 19; // service ID binary (i.e. 16 byte UUID for ACI, 1 byte prefix + 16 byte UUID for PNI) - optional bytes destination_service_id_binary = 20; // service ID binary (i.e. 16 byte UUID for ACI, 1 byte prefix + 16 byte UUID for PNI) - optional bytes server_guid_binary = 21; // 16-byte UUID - optional bytes updated_pni_binary = 22; // 16-byte UUID - // next: 22 + optional bytes source_service_id = 19; // service ID binary (i.e. 16 byte UUID for ACI, 1 byte prefix + 16 byte UUID for PNI) + optional bytes destination_service_id = 20; // service ID binary (i.e. 16 byte UUID for ACI, 1 byte prefix + 16 byte UUID for PNI) + optional bytes server_guid = 21; // 16-byte UUID + optional bytes updated_pni = 22; // 16-byte UUID + // next: 23 } message ProvisioningAddress { diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsAnonymousGrpcServiceTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsAnonymousGrpcServiceTest.java index 7c4848cc8..3c69ee015 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsAnonymousGrpcServiceTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsAnonymousGrpcServiceTest.java @@ -90,11 +90,11 @@ class AccountsAnonymousGrpcServiceTest extends .thenReturn(Optional.of(mock(Account.class))); assertTrue(unauthenticatedServiceStub().checkAccountExistence(CheckAccountExistenceRequest.newBuilder() - .setServiceIdentifier(ServiceIdentifierUtil.toGrpcServiceIdentifier(serviceIdentifier)) + .setServiceIdentifier(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(serviceIdentifier)) .build()).getAccountExists()); assertFalse(unauthenticatedServiceStub().checkAccountExistence(CheckAccountExistenceRequest.newBuilder() - .setServiceIdentifier(ServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(UUID.randomUUID()))) + .setServiceIdentifier(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(UUID.randomUUID()))) .build()).getAccountExists()); } @@ -131,7 +131,7 @@ class AccountsAnonymousGrpcServiceTest extends //noinspection ResultOfMethodCallIgnored GrpcTestUtils.assertRateLimitExceeded(retryAfter, () -> unauthenticatedServiceStub().checkAccountExistence(CheckAccountExistenceRequest.newBuilder() - .setServiceIdentifier(ServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(UUID.randomUUID()))) + .setServiceIdentifier(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(UUID.randomUUID()))) .build()), accountsManager); } @@ -148,7 +148,7 @@ class AccountsAnonymousGrpcServiceTest extends when(accountsManager.getByUsernameHash(usernameHash)) .thenReturn(CompletableFuture.completedFuture(Optional.of(account))); - assertEquals(ServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(accountIdentifier)), + assertEquals(GrpcServiceIdentifierUtil.toGrpcServiceIdentifier(new AciServiceIdentifier(accountIdentifier)), unauthenticatedServiceStub().lookupUsernameHash(LookupUsernameHashRequest.newBuilder() .setUsernameHash(ByteString.copyFrom(usernameHash)) .build()) diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcServiceTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcServiceTest.java index 8120c1f45..a008eab2a 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcServiceTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/grpc/AccountsGrpcServiceTest.java @@ -132,8 +132,8 @@ class AccountsGrpcServiceTest extends SimpleBaseGrpcTest expectedEnvelopes = new HashMap<>(Map.of( LINKED_DEVICE_ID, MessageProtos.Envelope.newBuilder() .setType(expectedEnvelopeType) - .setSourceServiceId(serviceIdentifier.toServiceIdentifierString()) + .setSourceServiceId(serviceIdentifier.toCompactByteString()) .setSourceDevice(AUTHENTICATED_DEVICE_ID) - .setDestinationServiceId(serviceIdentifier.toServiceIdentifierString()) + .setDestinationServiceId(serviceIdentifier.toCompactByteString()) .setClientTimestamp(CLOCK.millis()) .setServerTimestamp(CLOCK.millis()) .setEphemeral(false) @@ -572,9 +572,9 @@ class MessagesGrpcServiceTest extends SimpleBaseGrpcTest unauthenticatedServiceStub().getUnversionedProfile(request)); @@ -303,7 +303,7 @@ public class ProfileAnonymousGrpcServiceTest extends SimpleBaseGrpcTest messageSender.sendMessages(account, serviceIdentifier, Map.of(deviceId, MessageProtos.Envelope.newBuilder() - .setSourceServiceId(serviceIdentifier.toServiceIdentifierString()) + .setSourceServiceId(serviceIdentifier.toCompactByteString()) .setSourceDevice(deviceId) .build()), Map.of(deviceId, 17), diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManagerTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManagerTest.java index ce169d2a2..a644096af 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManagerTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/ChangeNumberManagerTest.java @@ -60,6 +60,7 @@ import org.whispersystems.textsecuregcm.push.MessageTooLargeException; import org.whispersystems.textsecuregcm.tests.util.KeysHelper; import org.whispersystems.textsecuregcm.util.Pair; import org.whispersystems.textsecuregcm.util.TestClock; +import org.whispersystems.textsecuregcm.util.UUIDUtil; public class ChangeNumberManagerTest { private MessageSender messageSender; @@ -245,11 +246,11 @@ public class ChangeNumberManagerTest { .setType(MessageProtos.Envelope.Type.forNumber(incomingMessage.type())) .setClientTimestamp(CLOCK.millis()) .setServerTimestamp(CLOCK.millis()) - .setDestinationServiceId(new AciServiceIdentifier(aci).toServiceIdentifierString()) + .setDestinationServiceId(new AciServiceIdentifier(aci).toCompactByteString()) .setContent(ByteString.copyFrom(incomingMessage.content())) - .setSourceServiceId(new AciServiceIdentifier(aci).toServiceIdentifierString()) + .setSourceServiceId(new AciServiceIdentifier(aci).toCompactByteString()) .setSourceDevice(primaryDeviceId) - .setUpdatedPni(updatedPhoneNumberIdentifiersByAccount.get(account).toString()) + .setUpdatedPni(UUIDUtil.toByteString(updatedPhoneNumberIdentifiersByAccount.get(account))) .setUrgent(true) .setEphemeral(false) .build(); diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/EnvelopeUtilTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/EnvelopeUtilTest.java deleted file mode 100644 index bcde6067b..000000000 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/EnvelopeUtilTest.java +++ /dev/null @@ -1,135 +0,0 @@ -/* - * Copyright 2025 Signal Messenger, LLC - * SPDX-License-Identifier: AGPL-3.0-only - */ - -package org.whispersystems.textsecuregcm.storage; - -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -import com.google.protobuf.ByteString; -import java.util.UUID; -import java.util.concurrent.ThreadLocalRandom; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; -import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; -import org.whispersystems.textsecuregcm.grpc.ServiceIdentifierUtil; -import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; -import org.whispersystems.textsecuregcm.identity.IdentityType; -import org.whispersystems.textsecuregcm.identity.PniServiceIdentifier; -import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; -import org.whispersystems.textsecuregcm.util.TestRandomUtil; -import org.whispersystems.textsecuregcm.util.UUIDUtil; - -class EnvelopeUtilTest { - - @ParameterizedTest - @ValueSource(booleans = {true, false}) - void compressExpand(final boolean includeBinaryServiceIdentifiers) { - final ExperimentEnrollmentManager experimentEnrollmentManager = mock(ExperimentEnrollmentManager.class); - when(experimentEnrollmentManager.isEnrolled(any(UUID.class), eq(EnvelopeUtil.INCLUDE_BINARY_SERVICE_ID_EXPERIMENT_NAME))) - .thenReturn(includeBinaryServiceIdentifiers); - - { - final MessageProtos.Envelope compressibleFieldsNullMessage = generateRandomMessageBuilder().build(); - final MessageProtos.Envelope compressed = EnvelopeUtil.compress(compressibleFieldsNullMessage); - - assertFalse(compressed.hasSourceServiceId()); - assertFalse(compressed.hasSourceServiceIdBinary()); - assertFalse(compressed.hasDestinationServiceId()); - assertFalse(compressed.hasDestinationServiceIdBinary()); - assertFalse(compressed.hasServerGuid()); - assertFalse(compressed.hasServerGuidBinary()); - assertFalse(compressed.hasUpdatedPni()); - assertFalse(compressed.hasUpdatedPniBinary()); - - final MessageProtos.Envelope expanded = EnvelopeUtil.expand(compressed, experimentEnrollmentManager); - - assertFalse(expanded.hasSourceServiceId()); - assertFalse(expanded.hasSourceServiceIdBinary()); - assertFalse(expanded.hasDestinationServiceId()); - assertFalse(expanded.hasDestinationServiceIdBinary()); - assertFalse(expanded.hasServerGuid()); - assertFalse(expanded.hasServerGuidBinary()); - assertFalse(compressed.hasUpdatedPni()); - assertFalse(compressed.hasUpdatedPniBinary()); - } - - { - final ServiceIdentifier sourceServiceId = generateRandomServiceIdentifier(); - final ServiceIdentifier destinationServiceId = generateRandomServiceIdentifier(); - final UUID serverGuid = UUID.randomUUID(); - final UUID updatedPni = UUID.randomUUID(); - - final MessageProtos.Envelope compressibleFieldsExpandedMessage = generateRandomMessageBuilder() - .setSourceServiceId(sourceServiceId.toServiceIdentifierString()) - .setDestinationServiceId(destinationServiceId.toServiceIdentifierString()) - .setServerGuid(serverGuid.toString()) - .setUpdatedPni(updatedPni.toString()) - .build(); - - final MessageProtos.Envelope compressed = EnvelopeUtil.compress(compressibleFieldsExpandedMessage); - - assertFalse(compressed.hasSourceServiceId()); - assertEquals(ServiceIdentifierUtil.toCompactByteString(sourceServiceId), compressed.getSourceServiceIdBinary()); - assertFalse(compressed.hasDestinationServiceId()); - assertEquals(ServiceIdentifierUtil.toCompactByteString(destinationServiceId), compressed.getDestinationServiceIdBinary()); - assertFalse(compressed.hasServerGuid()); - assertEquals(UUIDUtil.toByteString(serverGuid), compressed.getServerGuidBinary()); - assertFalse(compressed.hasUpdatedPni()); - assertEquals(UUIDUtil.toByteString(updatedPni), compressed.getUpdatedPniBinary()); - - assertEquals(compressed, EnvelopeUtil.compress(compressed), "Double compression should make no changes"); - - final MessageProtos.Envelope expanded = EnvelopeUtil.expand(compressed, experimentEnrollmentManager); - - assertEquals(sourceServiceId.toServiceIdentifierString(), expanded.getSourceServiceId()); - assertEquals(destinationServiceId.toServiceIdentifierString(), expanded.getDestinationServiceId()); - assertEquals(serverGuid.toString(), expanded.getServerGuid()); - assertEquals(updatedPni.toString(), expanded.getUpdatedPni()); - assertEquals(UUIDUtil.toByteString(updatedPni), expanded.getUpdatedPniBinary()); - - if (includeBinaryServiceIdentifiers) { - assertEquals(ServiceIdentifierUtil.toCompactByteString(sourceServiceId), expanded.getSourceServiceIdBinary()); - assertEquals(ServiceIdentifierUtil.toCompactByteString(destinationServiceId), expanded.getDestinationServiceIdBinary()); - assertEquals(UUIDUtil.toByteString(serverGuid), expanded.getServerGuidBinary()); - } else { - assertFalse(expanded.hasSourceServiceIdBinary()); - assertFalse(expanded.hasDestinationServiceIdBinary()); - assertFalse(expanded.hasServerGuidBinary()); - } - - assertEquals(expanded, EnvelopeUtil.expand(expanded, experimentEnrollmentManager), - "Double expansion should make no changes"); - - // Expanded envelopes include both representations of the `updatedPni` field - assertTrue(expanded.hasUpdatedPni()); - assertTrue(expanded.hasUpdatedPniBinary()); - assertEquals(UUID.fromString(expanded.getUpdatedPni()), UUIDUtil.fromByteString(expanded.getUpdatedPniBinary())); - } - } - - private static ServiceIdentifier generateRandomServiceIdentifier() { - final IdentityType identityType = ThreadLocalRandom.current().nextBoolean() ? IdentityType.ACI : IdentityType.PNI; - - return switch (identityType) { - case ACI -> new AciServiceIdentifier(UUID.randomUUID()); - case PNI -> new PniServiceIdentifier(UUID.randomUUID()); - }; - } - - private MessageProtos.Envelope.Builder generateRandomMessageBuilder() { - return MessageProtos.Envelope.newBuilder() - .setClientTimestamp(ThreadLocalRandom.current().nextLong()) - .setServerTimestamp(ThreadLocalRandom.current().nextLong()) - .setContent(ByteString.copyFrom(TestRandomUtil.nextBytes(256))) - .setType(MessageProtos.Envelope.Type.CIPHERTEXT); - } -} diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterIntegrationTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterIntegrationTest.java index 769deaa95..833675da5 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterIntegrationTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterIntegrationTest.java @@ -36,12 +36,13 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration; import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicMessagePersisterConfiguration; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; +import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; import org.whispersystems.textsecuregcm.push.MessageAvailabilityListener; import org.whispersystems.textsecuregcm.push.RedisMessageAvailabilityManager; import org.whispersystems.textsecuregcm.redis.RedisClusterExtension; import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables; import org.whispersystems.textsecuregcm.tests.util.DevicesHelper; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; import software.amazon.awssdk.services.dynamodb.DynamoDbClient; @@ -64,7 +65,6 @@ class MessagePersisterIntegrationTest { private RedisMessageAvailabilityManager redisMessageAvailabilityManager; private MessagePersister messagePersister; private Account account; - private ExperimentEnrollmentManager experimentEnrollmentManager; private static final Duration PERSIST_DELAY = Duration.ofMinutes(10); @@ -85,14 +85,13 @@ class MessagePersisterIntegrationTest { messageDeliveryScheduler = Schedulers.newBoundedElastic(10, 10_000, "messageDelivery"); persistQueueScheduler = Schedulers.newBoundedElastic(10, 10_000, "persistQueue"); messageDeletionExecutorService = Executors.newSingleThreadExecutor(); - experimentEnrollmentManager = mock(ExperimentEnrollmentManager.class); final MessagesDynamoDb messagesDynamoDb = new MessagesDynamoDb(DYNAMO_DB_EXTENSION.getDynamoDbClient(), DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(), Tables.MESSAGES.tableName(), Duration.ofDays(14), - messageDeletionExecutorService, experimentEnrollmentManager); + messageDeletionExecutorService); final AccountsManager accountsManager = mock(AccountsManager.class); messagesCache = new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(), - messageDeliveryScheduler, messageDeletionExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC(), experimentEnrollmentManager); + messageDeliveryScheduler, messageDeletionExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC()); final MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, @@ -199,7 +198,7 @@ class MessagePersisterIntegrationTest { dynamoDB.scan(ScanRequest.builder().tableName(Tables.MESSAGES.tableName()).build()).items().stream() .map(item -> { try { - return MessagesDynamoDb.convertItemToEnvelope(item, experimentEnrollmentManager); + return MessagesDynamoDb.convertItemToEnvelope(item); } catch (InvalidProtocolBufferException e) { fail("Could not parse stored message", e); return null; @@ -246,7 +245,7 @@ class MessagePersisterIntegrationTest { dynamoDB.scan(ScanRequest.builder().tableName(Tables.MESSAGES.tableName()).build()).items().stream() .map(item -> { try { - return MessagesDynamoDb.convertItemToEnvelope(item, experimentEnrollmentManager); + return MessagesDynamoDb.convertItemToEnvelope(item); } catch (InvalidProtocolBufferException e) { fail("Could not parse stored message", e); return null; @@ -264,8 +263,8 @@ class MessagePersisterIntegrationTest { .setServerTimestamp(serverTimestamp) .setContent(ByteString.copyFromUtf8(RandomStringUtils.secure().nextAlphanumeric(256))) .setType(MessageProtos.Envelope.Type.CIPHERTEXT) - .setServerGuid(messageGuid.toString()) - .setDestinationServiceId(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(messageGuid)) + .setDestinationServiceId(new AciServiceIdentifier(UUID.randomUUID()).toCompactByteString()) .setEphemeral(ephemeral) .build(); } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterTest.java index d1d2fbf49..4729de06e 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterTest.java @@ -55,11 +55,12 @@ import org.mockito.stubbing.Answer; import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration; import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicMessagePersisterConfiguration; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; +import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; import org.whispersystems.textsecuregcm.identity.IdentityType; import org.whispersystems.textsecuregcm.redis.RedisClusterExtension; import org.whispersystems.textsecuregcm.tests.util.DevicesHelper; import org.whispersystems.textsecuregcm.util.TestClock; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.scheduler.Scheduler; @@ -132,7 +133,7 @@ class MessagePersisterTest { messageDeliveryScheduler = Schedulers.newBoundedElastic(10, 10_000, "messageDelivery"); persistQueueScheduler = Schedulers.newBoundedElastic(10, 10_000, "persistQueue"); messagesCache = spy(new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(), - messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), CLOCK, mock(ExperimentEnrollmentManager.class))); + messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), CLOCK)); messagePersister = new MessagePersister(messagesCache, messagesManager, accountsManager, @@ -154,7 +155,7 @@ class MessagePersisterTest { messagesDynamoDb.store(messages, destinationUuid, destinationDevice); for (final MessageProtos.Envelope message : messages) { - messagesCache.remove(destinationUuid, destinationDevice.getId(), UUID.fromString(message.getServerGuid())).get(); + messagesCache.remove(destinationUuid, destinationDevice.getId(), UUIDUtil.fromByteString(message.getServerGuid())).get(); } return messages.size(); @@ -476,7 +477,7 @@ class MessagePersisterTest { .toList(); final long cacheSize = cachedMessages.stream().mapToLong(MessageProtos.Envelope::getSerializedSize).sum(); for (final MessageProtos.Envelope envelope : cachedMessages) { - messagesCache.insert(UUID.fromString(envelope.getServerGuid()), DESTINATION_ACCOUNT_UUID, Device.PRIMARY_ID, envelope).join(); + messagesCache.insert(UUIDUtil.fromByteString(envelope.getServerGuid()), DESTINATION_ACCOUNT_UUID, Device.PRIMARY_ID, envelope).join(); } final long expectedClearedBytes = (long) (cacheSize * EXTRA_ROOM_RATIO); @@ -575,12 +576,12 @@ class MessagePersisterTest { private MessageProtos.Envelope generateMessage(UUID accountUuid, UUID messageGuid, long messageTimestamp, int contentSize) { return MessageProtos.Envelope.newBuilder() - .setDestinationServiceId(accountUuid.toString()) + .setDestinationServiceId(new AciServiceIdentifier(accountUuid).toCompactByteString()) .setClientTimestamp(messageTimestamp) .setServerTimestamp(messageTimestamp) .setContent(ByteString.copyFromUtf8(RandomStringUtils.secure().nextAlphanumeric(contentSize))) .setType(MessageProtos.Envelope.Type.CIPHERTEXT) - .setServerGuid(messageGuid.toString()) + .setServerGuid(UUIDUtil.toByteString(messageGuid)) .build(); } } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheGetItemsScriptTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheGetItemsScriptTest.java index 01bc2a3c4..0133ddd6f 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheGetItemsScriptTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheGetItemsScriptTest.java @@ -21,9 +21,9 @@ import java.util.concurrent.ScheduledExecutorService; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; import org.whispersystems.textsecuregcm.redis.ClusterLuaScript; import org.whispersystems.textsecuregcm.redis.RedisClusterExtension; +import org.whispersystems.textsecuregcm.util.UUIDUtil; class MessagesCacheGetItemsScriptTest { @@ -38,10 +38,10 @@ class MessagesCacheGetItemsScriptTest { final UUID destinationUuid = UUID.randomUUID(); final byte deviceId = 1; - final String serverGuid = UUID.randomUUID().toString(); + final UUID serverGuid = UUID.randomUUID(); final MessageProtos.Envelope envelope1 = MessageProtos.Envelope.newBuilder() .setServerTimestamp(Instant.now().getEpochSecond()) - .setServerGuid(serverGuid) + .setServerGuid(UUIDUtil.toByteString(serverGuid)) .build(); insertScript.executeAsync(destinationUuid, deviceId, envelope1).toCompletableFuture().join(); @@ -54,11 +54,9 @@ class MessagesCacheGetItemsScriptTest { assertNotNull(messageAndScores); assertEquals(2, messageAndScores.size()); - final MessageProtos.Envelope resultEnvelope = - EnvelopeUtil.expand(MessageProtos.Envelope.parseFrom(messageAndScores.getFirst()), - mock(ExperimentEnrollmentManager.class)); + final MessageProtos.Envelope resultEnvelope = MessageProtos.Envelope.parseFrom(messageAndScores.getFirst()); - assertEquals(serverGuid, resultEnvelope.getServerGuid()); + assertEquals(serverGuid, UUIDUtil.fromByteString(resultEnvelope.getServerGuid())); } @Test diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScriptTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScriptTest.java index 73b0b732d..0c4d3dd8a 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScriptTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheInsertScriptTest.java @@ -25,6 +25,7 @@ import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.push.RedisMessageAvailabilityManager; import org.whispersystems.textsecuregcm.redis.FaultTolerantPubSubClusterConnection; import org.whispersystems.textsecuregcm.redis.RedisClusterExtension; +import org.whispersystems.textsecuregcm.util.UUIDUtil; class MessagesCacheInsertScriptTest { @@ -40,26 +41,25 @@ class MessagesCacheInsertScriptTest { final byte deviceId = 1; final MessageProtos.Envelope envelope1 = MessageProtos.Envelope.newBuilder() .setServerTimestamp(Instant.now().getEpochSecond()) - .setServerGuid(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) .build(); insertScript.executeAsync(destinationUuid, deviceId, envelope1).toCompletableFuture().join(); - assertEquals(List.of(EnvelopeUtil.compress(envelope1)), getStoredMessages(destinationUuid, deviceId)); + assertEquals(List.of(envelope1), getStoredMessages(destinationUuid, deviceId)); final MessageProtos.Envelope envelope2 = MessageProtos.Envelope.newBuilder() .setServerTimestamp(Instant.now().getEpochSecond()) - .setServerGuid(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) .build(); insertScript.executeAsync(destinationUuid, deviceId, envelope2).toCompletableFuture().join(); - assertEquals(List.of(EnvelopeUtil.compress(envelope1), EnvelopeUtil.compress(envelope2)), - getStoredMessages(destinationUuid, deviceId)); + assertEquals(List.of(envelope1, envelope2), getStoredMessages(destinationUuid, deviceId)); insertScript.executeAsync(destinationUuid, deviceId, envelope1).toCompletableFuture().join(); - assertEquals(List.of(EnvelopeUtil.compress(envelope1), EnvelopeUtil.compress(envelope2)), + assertEquals(List.of(envelope1, envelope2), getStoredMessages(destinationUuid, deviceId), "Messages with same GUID should be deduplicated"); } @@ -95,7 +95,7 @@ class MessagesCacheInsertScriptTest { assertFalse(insertScript.executeAsync(destinationUuid, deviceId, MessageProtos.Envelope.newBuilder() .setServerTimestamp(Instant.now().getEpochSecond()) - .setServerGuid(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) .build()) .toCompletableFuture() .join()); @@ -108,7 +108,7 @@ class MessagesCacheInsertScriptTest { assertTrue(insertScript.executeAsync(destinationUuid, deviceId, MessageProtos.Envelope.newBuilder() .setServerTimestamp(Instant.now().getEpochSecond()) - .setServerGuid(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) .build()) .toCompletableFuture() .join()); diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveByGuidScriptTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveByGuidScriptTest.java index aca714e2f..0b48777f1 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveByGuidScriptTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveByGuidScriptTest.java @@ -35,7 +35,7 @@ class MessagesCacheRemoveByGuidScriptTest { final UUID serverGuid = UUID.randomUUID(); final MessageProtos.Envelope envelope1 = MessageProtos.Envelope.newBuilder() .setServerTimestamp(Instant.now().getEpochSecond()) - .setServerGuid(serverGuid.toString()) + .setServerGuid(UUIDUtil.toByteString(serverGuid)) .build(); insertScript.executeAsync(destinationUuid, deviceId, envelope1).toCompletableFuture().join(); @@ -52,6 +52,6 @@ class MessagesCacheRemoveByGuidScriptTest { final MessageProtos.Envelope resultMessage = MessageProtos.Envelope.parseFrom(removedMessages.getFirst()); - assertEquals(serverGuid, UUIDUtil.fromByteString(resultMessage.getServerGuidBinary())); + assertEquals(serverGuid, UUIDUtil.fromByteString(resultMessage.getServerGuid())); } } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveQueueScriptTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveQueueScriptTest.java index 1757904b9..73329f0a6 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveQueueScriptTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheRemoveQueueScriptTest.java @@ -18,6 +18,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.redis.RedisClusterExtension; +import org.whispersystems.textsecuregcm.util.UUIDUtil; class MessagesCacheRemoveQueueScriptTest { @@ -35,7 +36,7 @@ class MessagesCacheRemoveQueueScriptTest { final byte deviceId = 1; final MessageProtos.Envelope envelope1 = MessageProtos.Envelope.newBuilder() .setServerTimestamp(Instant.now().getEpochSecond()) - .setServerGuid(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) .build(); insertScript.executeAsync(destinationUuid, deviceId, envelope1).toCompletableFuture().join(); diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheTest.java index 0df85ee19..0800c6299 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesCacheTest.java @@ -71,12 +71,12 @@ import org.junit.jupiter.params.provider.ValueSource; import org.reactivestreams.Publisher; import org.signal.libsignal.protocol.SealedSenderMultiRecipientMessage; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisClusterClient; import org.whispersystems.textsecuregcm.redis.RedisClusterExtension; import org.whispersystems.textsecuregcm.tests.util.RedisClusterHelper; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.core.publisher.Flux; import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; @@ -110,7 +110,7 @@ class MessagesCacheTest { resubscribeRetryExecutorService = Executors.newSingleThreadScheduledExecutor(); messageDeliveryScheduler = Schedulers.newBoundedElastic(10, 10_000, "messageDelivery"); messagesCache = new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(), - messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC(), mock(ExperimentEnrollmentManager.class)); + messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC()); } @AfterEach @@ -176,21 +176,21 @@ class MessagesCacheTest { } assertEquals(Collections.emptyList(), messagesCache.remove(DESTINATION_UUID, DESTINATION_DEVICE_ID, - messagesToRemove.stream().map(message -> UUID.fromString(message.getServerGuid())) + messagesToRemove.stream().map(message -> UUIDUtil.fromByteString(message.getServerGuid())) .collect(Collectors.toList())).get(5, TimeUnit.SECONDS)); for (final MessageProtos.Envelope message : messagesToRemove) { - messagesCache.insert(UUID.fromString(message.getServerGuid()), DESTINATION_UUID, DESTINATION_DEVICE_ID, + messagesCache.insert(UUIDUtil.fromByteString(message.getServerGuid()), DESTINATION_UUID, DESTINATION_DEVICE_ID, message).join(); } for (final MessageProtos.Envelope message : messagesToPreserve) { - messagesCache.insert(UUID.fromString(message.getServerGuid()), DESTINATION_UUID, DESTINATION_DEVICE_ID, + messagesCache.insert(UUIDUtil.fromByteString(message.getServerGuid()), DESTINATION_UUID, DESTINATION_DEVICE_ID, message).join(); } final List removedMessages = messagesCache.remove(DESTINATION_UUID, DESTINATION_DEVICE_ID, - messagesToRemove.stream().map(message -> UUID.fromString(message.getServerGuid())) + messagesToRemove.stream().map(message -> UUIDUtil.fromByteString(message.getServerGuid())) .collect(Collectors.toList())).get(5, TimeUnit.SECONDS); assertEquals(messagesToRemove.stream().map(RemovedMessage::fromEnvelope).toList(), removedMessages); @@ -227,7 +227,7 @@ class MessagesCacheTest { for (final MessageProtos.Envelope message : expectedMessages) { assertEquals(expectedOldestTimestamp, messagesCache.getEarliestUndeliveredTimestamp(DESTINATION_UUID, DESTINATION_DEVICE_ID).block()); - messagesCache.remove(DESTINATION_UUID, DESTINATION_DEVICE_ID, UUID.fromString(message.getServerGuid())).join(); + messagesCache.remove(DESTINATION_UUID, DESTINATION_DEVICE_ID, UUIDUtil.fromByteString(message.getServerGuid())).join(); expectedOldestTimestamp += 1; } assertNull(messagesCache.getEarliestUndeliveredTimestamp(DESTINATION_UUID, DESTINATION_DEVICE_ID).block()); @@ -252,7 +252,7 @@ class MessagesCacheTest { messagesCache.remove(DESTINATION_UUID, DESTINATION_DEVICE_ID, expectedMessages.stream() .map(MessageProtos.Envelope::getServerGuid) - .map(UUID::fromString) + .map(UUIDUtil::fromByteString) .collect(Collectors.toList())); final UUID message1Guid = UUID.randomUUID(); @@ -385,7 +385,7 @@ class MessagesCacheTest { } final MessagesCache messagesCache = new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(), - messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), cacheClock, mock(ExperimentEnrollmentManager.class)); + messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), cacheClock); final List actualMessages = Flux.from( messagesCache.get(DESTINATION_UUID, DESTINATION_DEVICE_ID)) @@ -406,7 +406,7 @@ class MessagesCacheTest { // delete all of these messages and call `getAll()`, to confirm that ephemeral messages have been discarded CompletableFuture.allOf(actualMessages.stream() .map(message -> messagesCache.remove(DESTINATION_UUID, DESTINATION_DEVICE_ID, - UUID.fromString(message.getServerGuid()))) + UUIDUtil.fromByteString(message.getServerGuid()))) .toArray(CompletableFuture[]::new)) .get(5, TimeUnit.SECONDS); @@ -592,7 +592,7 @@ class MessagesCacheTest { } else { assertEquals(1, messages.size()); - assertEquals(guid, UUID.fromString(messages.getFirst().getServerGuid())); + assertEquals(guid, UUIDUtil.fromByteString(messages.getFirst().getServerGuid())); assertFalse(messages.getFirst().hasSharedMrmKey()); final SealedSenderMultiRecipientMessage.Recipient recipient = mrm.getRecipients() .get(destinationServiceId.toLibsignal()); @@ -672,7 +672,7 @@ class MessagesCacheTest { default -> throw new IllegalStateException(); }; - messagesCache.insert(UUID.fromString(messageToInsert.getServerGuid()), destinationUuid, deviceId, messageToInsert).join(); + messagesCache.insert(UUIDUtil.fromByteString(messageToInsert.getServerGuid()), destinationUuid, deviceId, messageToInsert).join(); } long actualQueueSize = messagesCache.estimatePersistedQueueSizeBytes(destinationUuid, deviceId).join(); assertEquals(expectedQueueSize, actualQueueSize); @@ -730,7 +730,7 @@ class MessagesCacheTest { } assertEquals(message.toBuilder() - .setServerGuid(messageGuid.toString()) + .setServerGuid(UUIDUtil.toByteString(messageGuid)) .build(), messages.getFirst()); } @@ -765,7 +765,7 @@ class MessagesCacheTest { messageDeliveryScheduler = Schedulers.newBoundedElastic(10, 10_000, "messageDelivery"); messagesCache = new MessagesCache(mockCluster, messageDeliveryScheduler, - Executors.newSingleThreadExecutor(), mock(ScheduledExecutorService.class), Clock.systemUTC(), mock(ExperimentEnrollmentManager.class)); + Executors.newSingleThreadExecutor(), mock(ScheduledExecutorService.class), Clock.systemUTC()); } @AfterEach @@ -1003,13 +1003,13 @@ class MessagesCacheTest { .setServerTimestamp(timestamp) .setContent(ByteString.copyFromUtf8(RandomStringUtils.secure().nextAlphanumeric(256))) .setType(MessageProtos.Envelope.Type.CIPHERTEXT) - .setServerGuid(messageGuid.toString()) - .setDestinationServiceId(destinationServiceId.toServiceIdentifierString()) + .setServerGuid(UUIDUtil.toByteString(messageGuid)) + .setDestinationServiceId(destinationServiceId.toCompactByteString()) .setEphemeral(ephemeral); if (!sealedSender) { envelopeBuilder.setSourceDevice(random.nextInt(Device.MAXIMUM_DEVICE_ID) + 1) - .setSourceServiceId(UUID.randomUUID().toString()); + .setSourceServiceId(new AciServiceIdentifier(UUID.randomUUID()).toCompactByteString()); } return envelopeBuilder.build(); diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDbTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDbTest.java index c862824a3..bdafd85dc 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDbTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesDynamoDbTest.java @@ -6,9 +6,9 @@ package org.whispersystems.textsecuregcm.storage; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; import com.google.protobuf.ByteString; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.ArrayList; import java.util.List; @@ -26,10 +26,10 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import org.reactivestreams.Publisher; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; +import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; import org.whispersystems.textsecuregcm.storage.DynamoDbExtensionSchema.Tables; import org.whispersystems.textsecuregcm.tests.util.DevicesHelper; -import org.whispersystems.textsecuregcm.tests.util.MessageHelper; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.core.publisher.Flux; import reactor.test.StepVerifier; @@ -47,19 +47,19 @@ class MessagesDynamoDbTest { builder.setType(MessageProtos.Envelope.Type.UNIDENTIFIED_SENDER); builder.setClientTimestamp(123456789L); builder.setContent(ByteString.copyFrom(new byte[]{(byte) 0xDE, (byte) 0xAD, (byte) 0xBE, (byte) 0xEF})); - builder.setServerGuid(UUID.randomUUID().toString()); + builder.setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())); builder.setServerTimestamp(serverTimestamp); - builder.setDestinationServiceId(UUID.randomUUID().toString()); + builder.setDestinationServiceId(UUIDUtil.toByteString(UUID.randomUUID())); MESSAGE1 = builder.build(); builder.setType(MessageProtos.Envelope.Type.CIPHERTEXT); - builder.setSourceServiceId(UUID.randomUUID().toString()); + builder.setSourceServiceId(UUIDUtil.toByteString(UUID.randomUUID())); builder.setSourceDevice(1); builder.setContent(ByteString.copyFromUtf8("MOO")); - builder.setServerGuid(UUID.randomUUID().toString()); + builder.setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())); builder.setServerTimestamp(serverTimestamp + 1); - builder.setDestinationServiceId(UUID.randomUUID().toString()); + builder.setDestinationServiceId(UUIDUtil.toByteString(UUID.randomUUID())); MESSAGE2 = builder.build(); @@ -67,9 +67,9 @@ class MessagesDynamoDbTest { builder.clearSourceDevice(); builder.clearSourceDevice(); builder.setContent(ByteString.copyFromUtf8("COW")); - builder.setServerGuid(UUID.randomUUID().toString()); + builder.setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())); builder.setServerTimestamp(serverTimestamp); // Test same millisecond arrival for two different messages - builder.setDestinationServiceId(UUID.randomUUID().toString()); + builder.setDestinationServiceId(UUIDUtil.toByteString(UUID.randomUUID())); MESSAGE3 = builder.build(); } @@ -85,7 +85,7 @@ class MessagesDynamoDbTest { messageDeletionExecutorService = Executors.newSingleThreadExecutor(); messagesDynamoDb = new MessagesDynamoDb(DYNAMO_DB_EXTENSION.getDynamoDbClient(), DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(), Tables.MESSAGES.tableName(), Duration.ofDays(14), - messageDeletionExecutorService, mock(ExperimentEnrollmentManager.class)); + messageDeletionExecutorService); } @AfterEach @@ -107,8 +107,14 @@ class MessagesDynamoDbTest { final List messagesStored = load(destinationUuid, destinationDevice, MessagesDynamoDb.RESULT_SET_CHUNK_SIZE); assertThat(messagesStored).isNotNull().hasSize(3); + + // Danger: `UUID#compareTo` does NOT do what you might expect because it uses + // `Long.compare(a.mostSignificantBits(), b.mostSignificantBits())`, which means that a UUID whose binary + // representation is lexicographically "greater than" another UUID might wind up "less than" the other UUID if its + // most significant bit is `1`, which would cause `UUID#compareTo` to treat it as a negative number. To avoid that + // surprise (and match DynamoDB's lexicographical sorting), we just do string ordering. final MessageProtos.Envelope firstMessage = - MESSAGE1.getServerGuid().compareTo(MESSAGE3.getServerGuid()) < 0 ? MESSAGE1 : MESSAGE3; + UUIDUtil.fromByteString(MESSAGE1.getServerGuid()).toString().compareTo(UUIDUtil.fromByteString(MESSAGE3.getServerGuid()).toString()) < 0 ? MESSAGE1 : MESSAGE3; final MessageProtos.Envelope secondMessage = firstMessage == MESSAGE1 ? MESSAGE3 : MESSAGE1; assertThat(messagesStored).element(0).isEqualTo(firstMessage); assertThat(messagesStored).element(1).isEqualTo(secondMessage); @@ -124,8 +130,7 @@ class MessagesDynamoDbTest { final List messages = new ArrayList<>(messageCount); for (int i = 0; i < messageCount; i++) { - messages.add(MessageHelper.createMessage(UUID.randomUUID(), Device.PRIMARY_ID, destinationUuid, (i + 1L) * 1000, - "message " + i)); + messages.add(createMessage(UUID.randomUUID(), Device.PRIMARY_ID, destinationUuid, (i + 1L) * 1000, "message " + i)); } messagesDynamoDb.store(messages, destinationUuid, destinationDevice); @@ -158,8 +163,7 @@ class MessagesDynamoDbTest { final List messages = new ArrayList<>(messageCount); for (int i = 0; i < messageCount; i++) { - messages.add(MessageHelper.createMessage(UUID.randomUUID(), Device.PRIMARY_ID, destinationUuid, (i + 1L) * 1000, - "message " + i)); + messages.add(createMessage(UUID.randomUUID(), Device.PRIMARY_ID, destinationUuid, (i + 1L) * 1000, "message " + i)); } messagesDynamoDb.store(messages, destinationUuid, destinationDevice); @@ -207,7 +211,7 @@ class MessagesDynamoDbTest { .hasSize(1).element(0).isEqualTo(MESSAGE2); messagesDynamoDb.deleteMessage(secondDestinationUuid, primary, - UUID.fromString(MESSAGE2.getServerGuid()), MESSAGE2.getServerTimestamp()).get(1, TimeUnit.SECONDS); + UUIDUtil.fromByteString(MESSAGE2.getServerGuid()), MESSAGE2.getServerTimestamp()).get(1, TimeUnit.SECONDS); assertThat(load(destinationUuid, primary, MessagesDynamoDb.RESULT_SET_CHUNK_SIZE)).isNotNull().hasSize(1) .element(0).isEqualTo(MESSAGE1); @@ -236,7 +240,7 @@ class MessagesDynamoDbTest { assertThat(load(destinationUuid, primary, MessagesDynamoDb.RESULT_SET_CHUNK_SIZE)) .as("load should return all messages stored").containsOnly(MESSAGE1, MESSAGE2); - messagesDynamoDb.deleteMessage(destinationUuid, primary, UUID.fromString(MESSAGE1.getServerGuid()), MESSAGE1.getServerTimestamp()) + messagesDynamoDb.deleteMessage(destinationUuid, primary, UUIDUtil.fromByteString(MESSAGE1.getServerGuid()), MESSAGE1.getServerTimestamp()) .get(1, TimeUnit.SECONDS); assertThat(load(destinationUuid, primary, MessagesDynamoDb.RESULT_SET_CHUNK_SIZE)) .as("deleting message by guid and timestamp should work").containsExactly(MESSAGE2); @@ -273,8 +277,8 @@ class MessagesDynamoDbTest { { final MessageProtos.Envelope nonUrgentMessage = MessageProtos.Envelope.newBuilder() .setUrgent(false) - .setServerGuid(UUID.randomUUID().toString()) - .setDestinationServiceId(destinationUuid.toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) + .setDestinationServiceId(new AciServiceIdentifier(destinationUuid).toCompactByteString()) .setServerTimestamp(serverTimestamp++) .build(); @@ -289,17 +293,17 @@ class MessagesDynamoDbTest { for (int i = 0; i < MessagesDynamoDb.MAY_HAVE_URGENT_MESSAGES_QUERY_LIMIT * 5; i++) { messages.add(MessageProtos.Envelope.newBuilder() .setUrgent(false) - .setServerGuid(UUID.randomUUID().toString()) - .setDestinationServiceId(destinationUuid.toString()) - .setServerTimestamp(serverTimestamp++) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) + .setDestinationServiceId(new AciServiceIdentifier(destinationUuid).toCompactByteString()) + .setServerTimestamp(serverTimestamp++) .build()); } // and one urgent message messages.add(MessageProtos.Envelope.newBuilder() .setUrgent(true) - .setServerGuid(UUID.randomUUID().toString()) - .setDestinationServiceId(destinationUuid.toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) + .setDestinationServiceId(new AciServiceIdentifier(destinationUuid).toCompactByteString()) .setServerTimestamp(serverTimestamp++) .build()); @@ -308,4 +312,22 @@ class MessagesDynamoDbTest { assertThat(messagesDynamoDb.mayHaveUrgentMessages(destinationUuid, destinationDevice).join()).isTrue(); } + + private static MessageProtos.Envelope createMessage(final UUID senderUuid, + final byte senderDeviceId, + final UUID destinationUuid, + long timestamp, + final String content) { + + return MessageProtos.Envelope.newBuilder() + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) + .setType(MessageProtos.Envelope.Type.CIPHERTEXT) + .setClientTimestamp(timestamp) + .setServerTimestamp(0) + .setSourceServiceId(new AciServiceIdentifier(senderUuid).toCompactByteString()) + .setSourceDevice(senderDeviceId) + .setDestinationServiceId(new AciServiceIdentifier(destinationUuid).toCompactByteString()) + .setContent(ByteString.copyFrom(content.getBytes(StandardCharsets.UTF_8))) + .build(); + } } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesManagerTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesManagerTest.java index 2dd366264..4e82e4160 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesManagerTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagesManagerTest.java @@ -66,7 +66,7 @@ class MessagesManagerTest { void insert() { final UUID sourceAci = UUID.randomUUID(); final Envelope message = Envelope.newBuilder() - .setSourceServiceId(sourceAci.toString()) + .setSourceServiceId(new AciServiceIdentifier(sourceAci).toCompactByteString()) .build(); final UUID destinationUuid = UUID.randomUUID(); @@ -76,7 +76,7 @@ class MessagesManagerTest { verify(reportMessageManager).store(eq(sourceAci.toString()), any(UUID.class)); final Envelope syncMessage = Envelope.newBuilder(message) - .setSourceServiceId(destinationUuid.toString()) + .setSourceServiceId(new AciServiceIdentifier(destinationUuid).toCompactByteString()) .build(); messagesManager.insert(destinationUuid, Map.of(Device.PRIMARY_ID, syncMessage)); @@ -170,17 +170,17 @@ class MessagesManagerTest { verify(messagesCache).insert(any(), eq(singleDeviceAccountAciServiceIdentifier.uuid()), eq(Device.PRIMARY_ID), - eq(prototypeExpectedMessage.toBuilder().setDestinationServiceId(singleDeviceAccountAciServiceIdentifier.toServiceIdentifierString()).build())); + eq(prototypeExpectedMessage.toBuilder().setDestinationServiceId(singleDeviceAccountAciServiceIdentifier.toCompactByteString()).build())); verify(messagesCache).insert(any(), eq(singleDeviceAccountAciServiceIdentifier.uuid()), eq(Device.PRIMARY_ID), - eq(prototypeExpectedMessage.toBuilder().setDestinationServiceId(singleDeviceAccountPniServiceIdentifier.toServiceIdentifierString()).build())); + eq(prototypeExpectedMessage.toBuilder().setDestinationServiceId(singleDeviceAccountPniServiceIdentifier.toCompactByteString()).build())); verify(messagesCache).insert(any(), eq(multiDeviceAccountAciServiceIdentifier.uuid()), eq((byte) (Device.PRIMARY_ID + 1)), - eq(prototypeExpectedMessage.toBuilder().setDestinationServiceId(multiDeviceAccountAciServiceIdentifier.toServiceIdentifierString()).build())); + eq(prototypeExpectedMessage.toBuilder().setDestinationServiceId(multiDeviceAccountAciServiceIdentifier.toCompactByteString()).build())); verify(messagesCache, never()).insert(any(), eq(unresolvedAccountAciServiceIdentifier.uuid()), diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessagePublisherTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessagePublisherTest.java index b3a7627da..9d4894fd6 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessagePublisherTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessagePublisherTest.java @@ -37,11 +37,11 @@ import org.junit.jupiter.api.extension.RegisterExtension; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.whispersystems.textsecuregcm.entities.MessageProtos; -import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; import org.whispersystems.textsecuregcm.identity.ServiceIdentifier; import org.whispersystems.textsecuregcm.push.RedisMessageAvailabilityManager; import org.whispersystems.textsecuregcm.redis.RedisClusterExtension; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import reactor.adapter.JdkFlowAdapter; import reactor.core.Disposable; import reactor.core.scheduler.Scheduler; @@ -82,11 +82,10 @@ class RedisDynamoDbMessagePublisherTest { DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(), DynamoDbExtensionSchema.Tables.MESSAGES.tableName(), Duration.ofDays(14), - sharedExecutorService, - mock(ExperimentEnrollmentManager.class)); + sharedExecutorService); messagesCache = new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(), - messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC(), mock(ExperimentEnrollmentManager.class)); + messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC()); redisMessageAvailabilityManager = mock(RedisMessageAvailabilityManager.class); @@ -398,7 +397,7 @@ class RedisDynamoDbMessagePublisherTest { } private MessageProtos.Envelope insertRedisMessage(final MessageProtos.Envelope message) { - messagesCache.insert(UUID.fromString(message.getServerGuid()), + messagesCache.insert(UUIDUtil.fromByteString(message.getServerGuid()), DESTINATION_SERVICE_IDENTIFIER.uuid(), destinationDevice.getId(), message) @@ -408,7 +407,7 @@ class RedisDynamoDbMessagePublisherTest { } private void deleteRedisMessage(final MessageProtos.Envelope message) { - messagesCache.remove(DESTINATION_SERVICE_IDENTIFIER.uuid(), destinationDevice.getId(), UUID.fromString(message.getServerGuid())).join(); + messagesCache.remove(DESTINATION_SERVICE_IDENTIFIER.uuid(), destinationDevice.getId(), UUIDUtil.fromByteString(message.getServerGuid())).join(); } private MessageProtos.Envelope insertDynamoDbMessage(final MessageProtos.Envelope message) { @@ -420,7 +419,7 @@ class RedisDynamoDbMessagePublisherTest { private void deleteDynamoDbMessage(final MessageProtos.Envelope message) { messagesDynamoDb.deleteMessage(DESTINATION_SERVICE_IDENTIFIER.uuid(), destinationDevice, - UUID.fromString(message.getServerGuid()), + UUIDUtil.fromByteString(message.getServerGuid()), message.getServerTimestamp()) .join(); } @@ -434,8 +433,8 @@ class RedisDynamoDbMessagePublisherTest { .setServerTimestamp(timestamp) .setContent(ByteString.copyFromUtf8(RandomStringUtils.secure().nextAlphanumeric(256))) .setType(MessageProtos.Envelope.Type.CIPHERTEXT) - .setServerGuid(UUID.randomUUID().toString()) - .setDestinationServiceId(DESTINATION_SERVICE_IDENTIFIER.toServiceIdentifierString()); + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) + .setDestinationServiceId(DESTINATION_SERVICE_IDENTIFIER.toCompactByteString()); return envelopeBuilder.build(); } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStreamTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStreamTest.java index d19625a02..59dd10412 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStreamTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisDynamoDbMessageStreamTest.java @@ -20,6 +20,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; +import org.whispersystems.textsecuregcm.util.UUIDUtil; class RedisDynamoDbMessageStreamTest { @@ -57,7 +58,7 @@ class RedisDynamoDbMessageStreamTest { @Test void acknowledgeMessageDynamoDb() { final MessageProtos.Envelope message = generateMessage(); - final UUID messageGuid = UUID.fromString(message.getServerGuid()); + final UUID messageGuid = UUIDUtil.fromByteString(message.getServerGuid()); final long serverTimestamp = message.getServerTimestamp(); when(messagesDynamoDb.deleteMessage(ACCOUNT_IDENTIFIER, device, messageGuid, serverTimestamp)) @@ -72,7 +73,7 @@ class RedisDynamoDbMessageStreamTest { @Test void acknowledgeMessageRedis() { final MessageProtos.Envelope message = generateMessage(); - final UUID messageGuid = UUID.fromString(message.getServerGuid()); + final UUID messageGuid = UUIDUtil.fromByteString(message.getServerGuid()); when(messagesCache.remove(ACCOUNT_IDENTIFIER, DEVICE_ID, messageGuid)) .thenReturn(CompletableFuture.completedFuture(Optional.of(RemovedMessage.fromEnvelope(message)))); @@ -85,8 +86,8 @@ class RedisDynamoDbMessageStreamTest { private static MessageProtos.Envelope generateMessage() { return MessageProtos.Envelope.newBuilder() - .setServerGuid(UUID.randomUUID().toString()) - .setDestinationServiceId(new AciServiceIdentifier(ACCOUNT_IDENTIFIER).toServiceIdentifierString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) + .setDestinationServiceId(new AciServiceIdentifier(ACCOUNT_IDENTIFIER).toCompactByteString()) .setServerTimestamp(System.currentTimeMillis()) .setClientTimestamp(System.currentTimeMillis()) .setType(MessageProtos.Envelope.Type.CIPHERTEXT) diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java index 041481bb6..5a4965a2c 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/storage/foundationdb/FoundationDbMessageStoreTest.java @@ -423,7 +423,7 @@ class FoundationDbMessageStoreTest { assertInstanceOf(MessageStreamEntry.Envelope.class, retrievedEntries.get(i)); assertEquals(expectedVersionstamps.get(i), - messageGuidCodec.decodeMessageGuid(UUIDUtil.fromByteString(envelopeEntry.message().getServerGuidBinary()))); + messageGuidCodec.decodeMessageGuid(UUIDUtil.fromByteString(envelopeEntry.message().getServerGuid()))); } } @@ -483,18 +483,18 @@ class FoundationDbMessageStoreTest { StepVerifier.create(JdkFlowAdapter.flowPublisherToFlux(messageStream.getMessages())) .expectNext(new MessageStreamEntry.Envelope(message1 .toBuilder() - .setServerGuidBinary(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp1))) + .setServerGuid(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp1))) .build())) .expectNext(new MessageStreamEntry.Envelope(message2 .toBuilder() - .setServerGuidBinary(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp2))) + .setServerGuid(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp2))) .build())) .expectNext(new MessageStreamEntry.QueueEmpty()) // Trigger insertion of another message .then(latch::countDown) .expectNextMatches(entry -> entry.equals(new MessageStreamEntry.Envelope(message3 .toBuilder() - .setServerGuidBinary(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp3.join()))) + .setServerGuid(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(versionstamp3.join()))) .build()))) .verifyTimeout(Duration.ofSeconds(3)); } @@ -602,7 +602,7 @@ class FoundationDbMessageStoreTest { assertEquals(expectedRedeliveredVersionstamps, retrievedEntries.stream() .filter(e -> e instanceof MessageStreamEntry.Envelope) - .map(e -> messageGuidCodec.decodeMessageGuid(UUIDUtil.fromByteString(((MessageStreamEntry.Envelope) e).message().getServerGuidBinary()))) + .map(e -> messageGuidCodec.decodeMessageGuid(UUIDUtil.fromByteString(((MessageStreamEntry.Envelope) e).message().getServerGuid()))) .toList()); } @@ -692,7 +692,7 @@ class FoundationDbMessageStoreTest { .then(queueEmptyLatch::countDown) .expectNextMatches(entry -> entry.equals(new MessageStreamEntry.Envelope(freshEphemeralMessage .toBuilder() - .setServerGuidBinary(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(deliveredUnacknowledgedVersionstamp.join()))) + .setServerGuid(UUIDUtil.toByteString(messageGuidCodec.encodeMessageGuid(deliveredUnacknowledgedVersionstamp.join()))) .build()))) .verifyTimeout(Duration.ofSeconds(3)); diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/tests/util/MessageHelper.java b/service/src/test/java/org/whispersystems/textsecuregcm/tests/util/MessageHelper.java deleted file mode 100644 index e7cdf6fd4..000000000 --- a/service/src/test/java/org/whispersystems/textsecuregcm/tests/util/MessageHelper.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Copyright 2022 Signal Messenger, LLC - * SPDX-License-Identifier: AGPL-3.0-only - */ - -package org.whispersystems.textsecuregcm.tests.util; - -import com.google.protobuf.ByteString; -import java.nio.charset.StandardCharsets; -import java.util.UUID; -import org.whispersystems.textsecuregcm.entities.MessageProtos; - -public class MessageHelper { - - public static MessageProtos.Envelope createMessage(UUID senderUuid, final byte senderDeviceId, UUID destinationUuid, - long timestamp, String content) { - return MessageProtos.Envelope.newBuilder() - .setServerGuid(UUID.randomUUID().toString()) - .setType(MessageProtos.Envelope.Type.CIPHERTEXT) - .setClientTimestamp(timestamp) - .setServerTimestamp(0) - .setSourceServiceId(senderUuid.toString()) - .setSourceDevice(senderDeviceId) - .setDestinationServiceId(destinationUuid.toString()) - .setContent(ByteString.copyFrom(content.getBytes(StandardCharsets.UTF_8))) - .build(); - } -} diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionIntegrationTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionIntegrationTest.java index 718cf2bfb..1d2fb6d94 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionIntegrationTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionIntegrationTest.java @@ -49,6 +49,7 @@ import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfigurati import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope; import org.whispersystems.textsecuregcm.experiment.ExperimentEnrollmentManager; +import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier; import org.whispersystems.textsecuregcm.identity.IdentityType; import org.whispersystems.textsecuregcm.limits.MessageDeliveryLoopMonitor; import org.whispersystems.textsecuregcm.metrics.MessageMetrics; @@ -67,6 +68,7 @@ import org.whispersystems.textsecuregcm.storage.MessagesCache; import org.whispersystems.textsecuregcm.storage.MessagesDynamoDb; import org.whispersystems.textsecuregcm.storage.MessagesManager; import org.whispersystems.textsecuregcm.storage.ReportMessageManager; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import org.whispersystems.websocket.WebSocketClient; import org.whispersystems.websocket.messages.WebSocketResponseMessage; import reactor.core.publisher.Mono; @@ -106,10 +108,10 @@ class WebSocketConnectionIntegrationTest { when(dynamicConfigurationManager.getConfiguration()).thenReturn(new DynamicConfiguration()); messagesCache = new MessagesCache(REDIS_CLUSTER_EXTENSION.getRedisCluster(), - messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC(), mock(ExperimentEnrollmentManager.class)); + messageDeliveryScheduler, sharedExecutorService, mock(ScheduledExecutorService.class), Clock.systemUTC()); messagesDynamoDb = new MessagesDynamoDb(DYNAMO_DB_EXTENSION.getDynamoDbClient(), DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(), Tables.MESSAGES.tableName(), Duration.ofDays(7), - sharedExecutorService, mock(ExperimentEnrollmentManager.class)); + sharedExecutorService); redisMessageAvailabilityManager = new RedisMessageAvailabilityManager(REDIS_CLUSTER_EXTENSION.getRedisCluster(), sharedExecutorService, sharedExecutorService); reportMessageManager = mock(ReportMessageManager.class); account = mock(Account.class); @@ -392,8 +394,8 @@ class WebSocketConnectionIntegrationTest { .setServerTimestamp(timestamp) .setContent(ByteString.copyFromUtf8(RandomStringUtils.secure().nextAlphanumeric(256))) .setType(MessageProtos.Envelope.Type.CIPHERTEXT) - .setServerGuid(messageGuid.toString()) - .setDestinationServiceId(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(messageGuid)) + .setDestinationServiceId(new AciServiceIdentifier(UUID.randomUUID()).toCompactByteString()) .build(); } } diff --git a/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java b/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java index d2f962f26..2c921a68a 100644 --- a/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java +++ b/service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionTest.java @@ -56,6 +56,7 @@ import org.whispersystems.textsecuregcm.storage.Device; import org.whispersystems.textsecuregcm.storage.MessageStream; import org.whispersystems.textsecuregcm.storage.MessageStreamEntry; import org.whispersystems.textsecuregcm.storage.MessagesManager; +import org.whispersystems.textsecuregcm.util.UUIDUtil; import org.whispersystems.websocket.WebSocketClient; import org.whispersystems.websocket.messages.WebSocketResponseMessage; import reactor.adapter.JdkFlowAdapter; @@ -177,13 +178,13 @@ class WebSocketConnectionTest { verify(receiptSender) .sendReceipt(new AciServiceIdentifier(destinationAccountIdentifier), deviceId, - AciServiceIdentifier.valueOf(successfulMessage.getSourceServiceId()), + AciServiceIdentifier.fromByteString(successfulMessage.getSourceServiceId()), successfulMessage.getClientTimestamp()); verify(receiptSender) .sendReceipt(new AciServiceIdentifier(destinationAccountIdentifier), deviceId, - AciServiceIdentifier.valueOf(secondSuccessfulMessage.getSourceServiceId()), + AciServiceIdentifier.fromByteString(secondSuccessfulMessage.getSourceServiceId()), secondSuccessfulMessage.getClientTimestamp()); webSocketConnection.stop(); @@ -253,19 +254,19 @@ class WebSocketConnectionTest { verify(receiptSender) .sendReceipt(new AciServiceIdentifier(destinationAccountIdentifier), deviceId, - AciServiceIdentifier.valueOf(successfulMessage.getSourceServiceId()), + AciServiceIdentifier.fromByteString(successfulMessage.getSourceServiceId()), successfulMessage.getClientTimestamp()); verify(receiptSender, never()) .sendReceipt(new AciServiceIdentifier(destinationAccountIdentifier), deviceId, - AciServiceIdentifier.valueOf(failedMessage.getSourceServiceId()), + AciServiceIdentifier.fromByteString(failedMessage.getSourceServiceId()), failedMessage.getClientTimestamp()); verify(receiptSender, never()) .sendReceipt(new AciServiceIdentifier(destinationAccountIdentifier), deviceId, - AciServiceIdentifier.valueOf(secondSuccessfulMessage.getSourceServiceId()), + AciServiceIdentifier.fromByteString(secondSuccessfulMessage.getSourceServiceId()), secondSuccessfulMessage.getClientTimestamp()); verify(client, timeout(500)).close(eq(1011), anyString()); @@ -572,13 +573,13 @@ class WebSocketConnectionTest { final String content) { return Envelope.newBuilder() - .setServerGuid(UUID.randomUUID().toString()) + .setServerGuid(UUIDUtil.toByteString(UUID.randomUUID())) .setType(Envelope.Type.CIPHERTEXT) .setClientTimestamp(timestamp) .setServerTimestamp(0) - .setSourceServiceId(senderUuid.toString()) + .setSourceServiceId(new AciServiceIdentifier(senderUuid).toCompactByteString()) .setSourceDevice(SOURCE_DEVICE_ID) - .setDestinationServiceId(destinationUuid.toString()) + .setDestinationServiceId(new AciServiceIdentifier(destinationUuid).toCompactByteString()) .setContent(ByteString.copyFrom(content.getBytes(StandardCharsets.UTF_8))) .build(); }