Accept source ACI at /v1/messages/report

2026-04-19 21:38:05 +01:00 · 2022-03-08 17:34:20 -08:00
parent fa3a9570d6
commit 77fd01bd9f
14 changed files with 477 additions and 112 deletions
--- a/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
@@ -637,8 +637,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
        new DirectoryV2Controller(directoryV2CredentialsGenerator),
        new DonationController(clock, zkReceiptOperations, redeemedReceiptsManager, accountsManager, config.getBadges(),
            ReceiptCredentialPresentation::new, stripeExecutor, config.getDonationConfiguration(), config.getStripe()),
-        new MessageController(rateLimiters, messageSender, receiptSender, accountsManager, messagesManager, apnFallbackManager,
-            reportMessageManager, multiRecipientMessageExecutor),
+        new MessageController(rateLimiters, messageSender, receiptSender, accountsManager, deletedAccountsManager,
+            messagesManager, apnFallbackManager, reportMessageManager, multiRecipientMessageExecutor),
        new PaymentsController(currencyManager, paymentsCredentialsGenerator),
        new ProfileController(clock, rateLimiters, accountsManager, profilesManager, dynamicConfigurationManager, profileBadgeConverter, config.getBadges(), cdnS3Client, profileCdnPolicyGenerator, profileCdnPolicySigner, config.getCdnConfiguration().getBucket(), zkProfileOperations),
        new ProvisioningController(rateLimiters, provisioningManager),
--- a/service/src/main/java/org/whispersystems/textsecuregcm/controllers/MessageController.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/controllers/MessageController.java
@@ -88,6 +88,7 @@ import org.whispersystems.textsecuregcm.push.ReceiptSender;
 import org.whispersystems.textsecuregcm.redis.RedisOperation;
 import org.whispersystems.textsecuregcm.storage.Account;
 import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.DeletedAccountsManager;
 import org.whispersystems.textsecuregcm.storage.Device;
 import org.whispersystems.textsecuregcm.storage.MessagesManager;
 import org.whispersystems.textsecuregcm.storage.ReportMessageManager;
@@ -105,14 +106,15 @@ public class MessageController {

  private static final Logger logger = LoggerFactory.getLogger(MessageController.class);

-  private final RateLimiters                rateLimiters;
-  private final MessageSender               messageSender;
-  private final ReceiptSender               receiptSender;
-  private final AccountsManager             accountsManager;
-  private final MessagesManager             messagesManager;
-  private final ApnFallbackManager          apnFallbackManager;
-  private final ReportMessageManager        reportMessageManager;
-  private final ExecutorService             multiRecipientMessageExecutor;
+  private final RateLimiters rateLimiters;
+  private final MessageSender messageSender;
+  private final ReceiptSender receiptSender;
+  private final AccountsManager accountsManager;
+  private final DeletedAccountsManager deletedAccountsManager;
+  private final MessagesManager messagesManager;
+  private final ApnFallbackManager apnFallbackManager;
+  private final ReportMessageManager reportMessageManager;
+  private final ExecutorService multiRecipientMessageExecutor;

  @VisibleForTesting
  static final Semver FIRST_IOS_VERSION_WITH_INCORRECT_ENVELOPE_TYPE = new Semver("5.22.0");
@@ -146,6 +148,7 @@ public class MessageController {
      MessageSender messageSender,
      ReceiptSender receiptSender,
      AccountsManager accountsManager,
+      DeletedAccountsManager deletedAccountsManager,
      MessagesManager messagesManager,
      ApnFallbackManager apnFallbackManager,
      ReportMessageManager reportMessageManager,
@@ -154,6 +157,7 @@ public class MessageController {
    this.messageSender = messageSender;
    this.receiptSender = receiptSender;
    this.accountsManager = accountsManager;
+    this.deletedAccountsManager = deletedAccountsManager;
    this.messagesManager = messagesManager;
    this.apnFallbackManager = apnFallbackManager;
    this.reportMessageManager = reportMessageManager;
@@ -556,11 +560,39 @@ public class MessageController {

  @Timed
  @POST
-  @Path("/report/{sourceNumber}/{messageGuid}")
-  public Response reportMessage(@Auth AuthenticatedAccount auth, @PathParam("sourceNumber") String sourceNumber,
+  @Path("/report/{source}/{messageGuid}")
+  public Response reportMessage(@Auth AuthenticatedAccount auth, @PathParam("source") String source,
      @PathParam("messageGuid") UUID messageGuid) {

-    reportMessageManager.report(sourceNumber, messageGuid, auth.getAccount().getUuid());
+    final Optional<String> sourceNumber;
+    final Optional<UUID> sourceAci;
+    final Optional<UUID> sourcePni;
+    if (source.startsWith("+")) {
+      sourceNumber = Optional.of(source);
+      final Optional<Account> maybeAccount = accountsManager.getByE164(source);
+      if (maybeAccount.isPresent()) {
+        sourceAci = maybeAccount.map(Account::getUuid);
+        sourcePni = maybeAccount.map(Account::getPhoneNumberIdentifier);
+      } else {
+        sourceAci = deletedAccountsManager.findDeletedAccountAci(source);
+        sourcePni = Optional.ofNullable(accountsManager.getPhoneNumberIdentifier(source));
+      }
+    } else {
+      sourceAci = Optional.of(UUID.fromString(source));
+
+      final Optional<Account> sourceAccount = accountsManager.getByAccountIdentifier(sourceAci.get());
+
+      if (sourceAccount.isEmpty()) {
+        logger.warn("Could not find source: {}", sourceAci.get());
+        sourceNumber = deletedAccountsManager.findDeletedAccountE164(sourceAci.get());
+        sourcePni = sourceNumber.map(accountsManager::getPhoneNumberIdentifier);
+      } else {
+        sourceNumber = sourceAccount.map(Account::getNumber);
+        sourcePni = sourceAccount.map(Account::getPhoneNumberIdentifier);
+      }
+    }
+
+    reportMessageManager.report(sourceNumber, sourceAci, sourcePni, messageGuid, auth.getAccount().getUuid());

    return Response.status(Status.ACCEPTED)
        .build();
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
@@ -513,6 +513,14 @@ public class AccountsManager {
    }
  }

+  public Optional<String> getNumberForPhoneNumberIdentifier(UUID pni) {
+    return phoneNumberIdentifiers.getPhoneNumber(pni);
+  }
+
+  public UUID getPhoneNumberIdentifier(String e164) {
+    return phoneNumberIdentifiers.getPhoneNumberIdentifier(e164);
+  }
+
  public AccountCrawlChunk getAllFromDynamo(int length) {
    return accounts.getAllFromStart(length);
  }
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccounts.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccounts.java
@@ -21,14 +21,16 @@ import java.util.stream.Collectors;
 import org.whispersystems.textsecuregcm.util.AttributeValues;
 import org.whispersystems.textsecuregcm.util.Pair;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
-import software.amazon.awssdk.services.dynamodb.model.DeleteItemRequest;
-import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
-import software.amazon.awssdk.services.dynamodb.model.GetItemResponse;
 import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
 import software.amazon.awssdk.services.dynamodb.model.BatchGetItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.BatchGetItemResponse;
+import software.amazon.awssdk.services.dynamodb.model.DeleteItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.GetItemResponse;
 import software.amazon.awssdk.services.dynamodb.model.KeysAndAttributes;
 import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.QueryRequest;
+import software.amazon.awssdk.services.dynamodb.model.QueryResponse;
 import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
 import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest;

@@ -40,6 +42,8 @@ public class DeletedAccounts extends AbstractDynamoDbStore {
  static final String ATTR_EXPIRES = "E";
  static final String ATTR_NEEDS_CDS_RECONCILIATION = "R";

+  static final String UUID_TO_E164_INDEX_NAME = "u_to_p";
+
  static final Duration TIME_TO_LIVE = Duration.ofDays(30);

  // Note that this limit is imposed by DynamoDB itself; going above 100 will result in errors
@@ -76,6 +80,28 @@ public class DeletedAccounts extends AbstractDynamoDbStore {
    return Optional.ofNullable(AttributeValues.getUUID(response.item(), ATTR_ACCOUNT_UUID, null));
  }

+  Optional<String> findE164(final UUID uuid) {
+    final QueryResponse response = db().query(QueryRequest.builder()
+        .tableName(tableName)
+        .indexName(UUID_TO_E164_INDEX_NAME)
+        .keyConditionExpression("#uuid = :uuid")
+        .projectionExpression("#e164")
+        .expressionAttributeNames(Map.of("#uuid", ATTR_ACCOUNT_UUID,
+            "#e164", KEY_ACCOUNT_E164))
+        .expressionAttributeValues(Map.of(":uuid", AttributeValues.fromUUID(uuid))).build());
+
+    if (response.count() == 0) {
+      return Optional.empty();
+    }
+
+    if (response.count() > 1) {
+      throw new RuntimeException(
+          "Impossible result: more than one phone number returned for UUID: " + uuid);
+    }
+
+    return Optional.ofNullable(response.items().get(0).get(KEY_ACCOUNT_E164).s());
+  }
+
  void remove(final String e164) {
    db().deleteItem(DeleteItemRequest.builder()
        .tableName(tableName)
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/DeletedAccountsManager.java
@@ -173,8 +173,7 @@ public class DeletedAccountsManager {
          }

          return lockAcquired;
-        })
-        .collect(Collectors.toList());
+        }).toList();

    assert lockItems.size() == reconciliationCandidates.size();

@@ -192,7 +191,16 @@ public class DeletedAccountsManager {
    try {
      deletedAccounts.markReconciled(consumer.reconcile(accountsToReconcile));
    } finally {
-      lockItems.forEach(lockItem -> lockClient.releaseLock(ReleaseLockOptions.builder(lockItem).withBestEffort(true).build()));
+      lockItems.forEach(
+          lockItem -> lockClient.releaseLock(ReleaseLockOptions.builder(lockItem).withBestEffort(true).build()));
    }
  }
+
+  public Optional<UUID> findDeletedAccountAci(final String e164) {
+    return deletedAccounts.findUuid(e164);
+  }
+
+  public Optional<String> findDeletedAccountE164(final UUID uuid) {
+    return deletedAccounts.findE164(uuid);
+  }
 }
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java
@@ -14,6 +14,8 @@ import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
 import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntity;
 import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntityList;
@@ -23,6 +25,8 @@ import org.whispersystems.textsecuregcm.util.Constants;

 public class MessagesManager {

+  private static final Logger logger = LoggerFactory.getLogger(MessagesManager.class);
+
  private static final int RESULT_SET_CHUNK_SIZE = 100;

  private static final MetricRegistry metricRegistry = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
@@ -53,7 +57,11 @@ public class MessagesManager {
    messagesCache.insert(messageGuid, destinationUuid, destinationDevice, message);

    if (message.hasSource() && !destinationUuid.toString().equals(message.getSourceUuid())) {
-      reportMessageManager.store(message.getSource(), messageGuid);
+      if (message.hasSourceUuid()) {
+        reportMessageManager.store(message.getSource(), message.getSourceUuid(), messageGuid);
+      } else {
+        logger.warn("Message missing source UUID");
+      }
    }
  }

--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/PhoneNumberIdentifiers.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/PhoneNumberIdentifiers.java
@@ -5,20 +5,23 @@

 package org.whispersystems.textsecuregcm.storage;

+import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
+
 import com.google.common.annotations.VisibleForTesting;
 import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.Timer;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
 import org.whispersystems.textsecuregcm.util.AttributeValues;
 import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
 import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.GetItemResponse;
+import software.amazon.awssdk.services.dynamodb.model.QueryRequest;
+import software.amazon.awssdk.services.dynamodb.model.QueryResponse;
 import software.amazon.awssdk.services.dynamodb.model.ReturnValue;
 import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.UpdateItemResponse;
-import java.util.Map;
-import java.util.UUID;
-
-import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;

 /**
 * Manages a global, persistent mapping of phone numbers to phone number identifiers regardless of whether those
@@ -31,7 +34,10 @@ public class PhoneNumberIdentifiers {

  @VisibleForTesting
  static final String KEY_E164 = "P";
-  private static final String ATTR_PHONE_NUMBER_IDENTIFIER = "PNI";
+  @VisibleForTesting
+  static final String INDEX_NAME = "pni_to_p";
+  @VisibleForTesting
+  static final String ATTR_PHONE_NUMBER_IDENTIFIER = "PNI";

  private static final Timer GET_PNI_TIMER = Metrics.timer(name(PhoneNumberIdentifiers.class, "get"));
  private static final Timer SET_PNI_TIMER = Metrics.timer(name(PhoneNumberIdentifiers.class, "set"));
@@ -69,6 +75,34 @@ public class PhoneNumberIdentifiers {
    return phoneNumberIdentifier;
  }

+  public Optional<String> getPhoneNumber(final UUID phoneNumberIdentifier) {
+    final QueryResponse response = dynamoDbClient.query(QueryRequest.builder()
+        .tableName(tableName)
+        .indexName(INDEX_NAME)
+        .keyConditionExpression("#pni = :pni")
+        .projectionExpression("#phone_number")
+        .expressionAttributeNames(Map.of(
+            "#phone_number", KEY_E164,
+            "#pni", ATTR_PHONE_NUMBER_IDENTIFIER
+        ))
+        .expressionAttributeValues(Map.of(
+            ":pni", AttributeValues.fromUUID(phoneNumberIdentifier)
+        ))
+        .build());
+
+    if (response.count() == 0) {
+      return Optional.empty();
+    }
+
+    if (response.count() > 1) {
+      throw new RuntimeException(
+          "Impossible result: more than one phone number returned for PNI: " + phoneNumberIdentifier);
+    }
+
+    return Optional.ofNullable(response.items().get(0).get(KEY_E164).s());
+  }
+
+
  @VisibleForTesting
  UUID generatePhoneNumberIdentifierIfNotExists(final String phoneNumber) {
    final UpdateItemResponse response = SET_PNI_TIMER.record(() -> dynamoDbClient.updateItem(UpdateItemRequest.builder()
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/ReportMessageManager.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/ReportMessageManager.java
@@ -1,6 +1,9 @@
 package org.whispersystems.textsecuregcm.storage;

+import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
+
 import io.lettuce.core.RedisException;
+import io.micrometer.core.instrument.Metrics;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
@@ -8,6 +11,7 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.UUID;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -16,6 +20,8 @@ import org.whispersystems.textsecuregcm.util.UUIDUtil;

 public class ReportMessageManager {

+  private static final String MIGRATION_COUNTER_NAME = name(ReportMessageManager.class, "migration");
+
  private final ReportMessageDynamoDb reportMessageDynamoDb;
  private final FaultTolerantRedisCluster rateLimitCluster;

@@ -39,51 +45,91 @@ public class ReportMessageManager {
    this.reportedMessageListeners.add(listener);
  }

-  public void store(String sourceNumber, UUID messageGuid) {
+  // TODO sourceNumber can be removed after 2022-04-01
+  public void store(String sourceNumber, String sourceAci, UUID messageGuid) {

    try {
      Objects.requireNonNull(sourceNumber);
+      Objects.requireNonNull(sourceAci);

      reportMessageDynamoDb.store(hash(messageGuid, sourceNumber));
+      reportMessageDynamoDb.store(hash(messageGuid, sourceAci));
    } catch (final Exception e) {
      logger.warn("Failed to store hash", e);
    }
  }

-  public void report(String sourceNumber, UUID messageGuid, UUID reporterUuid) {
+  public void report(Optional<String> sourceNumber, Optional<UUID> sourceAci, Optional<UUID> sourcePni,
+      UUID messageGuid, UUID reporterUuid) {

-    final boolean found = reportMessageDynamoDb.remove(hash(messageGuid, sourceNumber));
+    // TODO sourceNumber can be removed after 2022-04-15
+    final boolean foundByNumber = sourceNumber.map(number -> reportMessageDynamoDb.remove(hash(messageGuid, number)))
+        .orElse(false);

-    if (found) {
+    final boolean foundByAci = sourceAci.map(uuid -> reportMessageDynamoDb.remove(hash(messageGuid, uuid.toString()))).
+        orElse(false);
+
+    if (foundByNumber || foundByAci) {
      rateLimitCluster.useCluster(connection -> {
-        final String reportedSenderKey = getReportedSenderKey(sourceNumber);
+        sourceNumber.ifPresent(number -> {
+          final String reportedSenderKey = getReportedSenderKey(number);
+          connection.sync().pfadd(reportedSenderKey, reporterUuid.toString());
+          connection.sync().expire(reportedSenderKey, counterTtl.toSeconds());
+        });

-        connection.sync().pfadd(reportedSenderKey, reporterUuid.toString());
-        connection.sync().expire(reportedSenderKey, counterTtl.toSeconds());
+        sourcePni.ifPresent(pni -> {
+          final String reportedSenderKey = getReportedSenderPniKey(pni);
+          connection.sync().pfadd(reportedSenderKey, reporterUuid.toString());
+          connection.sync().expire(reportedSenderKey, counterTtl.toSeconds());
+        });
+
+        sourceAci.ifPresent(aci -> {
+          final String reportedSenderKey = getReportedSenderAciKey(aci);
+          connection.sync().pfadd(reportedSenderKey, reporterUuid.toString());
+          connection.sync().expire(reportedSenderKey, counterTtl.toSeconds());
+        });
      });

-      reportedMessageListeners.forEach(listener -> {
-        try {
-          listener.handleMessageReported(sourceNumber, messageGuid, reporterUuid);
-        } catch (final Exception e) {
-          logger.error("Failed to notify listener of reported message", e);
-        }
-      });
+      sourceNumber.ifPresent(number ->
+          reportedMessageListeners.forEach(listener -> {
+            try {
+              // TODO should listener take the source Aci?
+              listener.handleMessageReported(number, messageGuid, reporterUuid);
+            } catch (final Exception e) {
+              logger.error("Failed to notify listener of reported message", e);
+            }
+          }));
    }
+
+    Metrics.counter(
+        MIGRATION_COUNTER_NAME,
+        "foundByNumber", String.valueOf(foundByNumber),
+        "foundByAci", String.valueOf(foundByAci),
+        "sourceAciPresent", String.valueOf(sourceAci.isPresent()),
+        "sourcePniPresent", String.valueOf(sourcePni.isPresent()),
+        "sourceNumberPresent", String.valueOf(sourceNumber.isPresent())
+    ).increment();
  }

  /**
-   * Returns the number of times messages from the given number have been reported by recipients as abusive. Note that
+   * Returns the number of times messages from the given account have been reported by recipients as abusive. Note that
   * this method makes a call to an external service, and callers should take care to memoize calls where possible and
   * avoid unnecessary calls.
   *
-   * @param number the number to check for recent reports
-   *
+   * @param account the account to check for recent reports
   * @return the number of times the given number has been reported recently
   */
-  public int getRecentReportCount(final String number) {
+  public int getRecentReportCount(final Account account) {
    try {
-      return rateLimitCluster.withCluster(connection -> connection.sync().pfcount(getReportedSenderKey(number)).intValue());
+      return rateLimitCluster.withCluster(
+          connection ->
+              Math.max(
+                  Math.max(
+                      // TODO number can be removed after 2022-04-15
+                      connection.sync().pfcount(getReportedSenderKey(account.getNumber())).intValue(),
+                      connection.sync().pfcount(getReportedSenderPniKey(account.getPhoneNumberIdentifier()))
+                          .intValue()),
+                  connection.sync().pfcount(getReportedSenderAciKey(account.getUuid())).intValue()));
    } catch (final RedisException e) {
      return 0;
    }
@@ -106,4 +152,12 @@ public class ReportMessageManager {
  private static String getReportedSenderKey(final String senderNumber) {
    return "reported_number::" + senderNumber;
  }
+
+  private static String getReportedSenderAciKey(final UUID aci) {
+    return "reported_account::" + aci.toString();
+  }
+
+  private static String getReportedSenderPniKey(final UUID pni) {
+    return "reported_pni::" + pni.toString();
+  }
 }