Count reported messages per sender

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerConfiguration.java

@@ -43,6 +43,7 @@ import org.whispersystems.textsecuregcm.configuration.RecaptchaV2Configuration;
 import org.whispersystems.textsecuregcm.configuration.RedisClusterConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RedisConfiguration;
 import org.whispersystems.textsecuregcm.configuration.RemoteConfigConfiguration;
+import org.whispersystems.textsecuregcm.configuration.ReportMessageConfiguration;
 import org.whispersystems.textsecuregcm.configuration.SecureBackupServiceConfiguration;
 import org.whispersystems.textsecuregcm.configuration.SecureStorageServiceConfiguration;
 import org.whispersystems.textsecuregcm.configuration.StripeConfiguration;
@@ -318,6 +319,11 @@ public class WhisperServerConfiguration extends Configuration {
   // TODO: Mark as @NotNull when enabled for production.
   private SubscriptionConfiguration subscription;
 
+  @Valid
+  @NotNull
+  @JsonProperty
+  private ReportMessageConfiguration reportMessage;
+
   private Map<String, String> transparentDataIndex = new HashMap<>();
 
   public StripeConfiguration getStripe() {
@@ -545,4 +551,8 @@ public class WhisperServerConfiguration extends Configuration {
   public SubscriptionConfiguration getSubscription() {
     return subscription;
   }
+
+  public ReportMessageConfiguration getReportMessageConfiguration() {
+    return reportMessage;
+  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -364,7 +364,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     AbusiveHostRules abusiveHostRules = new AbusiveHostRules(abuseDatabase);
     RemoteConfigs remoteConfigs = new RemoteConfigs(accountDatabase);
     PushChallengeDynamoDb pushChallengeDynamoDb = new PushChallengeDynamoDb(pushChallengeDynamoDbClient, config.getPushChallengeDynamoDbConfiguration().getTableName());
-    ReportMessageDynamoDb reportMessageDynamoDb = new ReportMessageDynamoDb(reportMessageDynamoDbClient, config.getReportMessageDynamoDbConfiguration().getTableName());
+    ReportMessageDynamoDb reportMessageDynamoDb = new ReportMessageDynamoDb(reportMessageDynamoDbClient, config.getReportMessageDynamoDbConfiguration().getTableName(), config.getReportMessageConfiguration().getReportTtl());
     VerificationCodeStore pendingAccounts = new VerificationCodeStore(pendingAccountsDynamoDbClient, config.getPendingAccountsDynamoDbConfiguration().getTableName());
     VerificationCodeStore pendingDevices = new VerificationCodeStore(pendingDevicesDynamoDbClient, config.getPendingDevicesDynamoDbConfiguration().getTableName());
 
@@ -438,7 +438,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     ProfilesManager            profilesManager            = new ProfilesManager(profiles, cacheCluster);
     MessagesCache              messagesCache              = new MessagesCache(messagesCluster, messagesCluster, keyspaceNotificationDispatchExecutor);
     PushLatencyManager         pushLatencyManager         = new PushLatencyManager(metricsCluster);
-    ReportMessageManager       reportMessageManager       = new ReportMessageManager(reportMessageDynamoDb, Metrics.globalRegistry);
+    ReportMessageManager       reportMessageManager       = new ReportMessageManager(reportMessageDynamoDb, rateLimitersCluster, Metrics.globalRegistry, config.getReportMessageConfiguration().getCounterTtl());
     MessagesManager            messagesManager            = new MessagesManager(messagesDynamoDb, messagesCache, pushLatencyManager, reportMessageManager);
     DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,
         deletedAccountsLockDynamoDbClient, config.getDeletedAccountsLockDynamoDbConfiguration().getTableName());

service/src/main/java/org/whispersystems/textsecuregcm/configuration/ReportMessageConfiguration.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2013-2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import javax.validation.constraints.NotNull;
+import java.time.Duration;
+
+public class ReportMessageConfiguration {
+
+  @JsonProperty
+  @NotNull
+  private final Duration reportTtl = Duration.ofDays(7);
+
+  @JsonProperty
+  @NotNull
+  private final Duration counterTtl = Duration.ofDays(1);
+
+  public Duration getReportTtl() {
+    return reportTtl;
+  }
+
+  public Duration getCounterTtl() {
+    return counterTtl;
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/storage/ReportMessageDynamoDb.java

@@ -15,14 +15,14 @@ public class ReportMessageDynamoDb {
   static final String KEY_HASH = "H";
   static final String ATTR_TTL = "E";
 
-  static final Duration TIME_TO_LIVE = Duration.ofDays(7);
-
   private final DynamoDbClient db;
   private final String tableName;
+  private final Duration ttl;
 
-  public ReportMessageDynamoDb(final DynamoDbClient dynamoDB, final String tableName) {
+  public ReportMessageDynamoDb(final DynamoDbClient dynamoDB, final String tableName, final Duration ttl) {
     this.db = dynamoDB;
     this.tableName = tableName;
+    this.ttl = ttl;
   }
 
   public void store(byte[] hash) {
@@ -30,7 +30,7 @@ public class ReportMessageDynamoDb {
         .tableName(tableName)
         .item(Map.of(
             KEY_HASH, AttributeValues.fromByteArray(hash),
-            ATTR_TTL, AttributeValues.fromLong(Instant.now().plus(TIME_TO_LIVE).getEpochSecond())
+            ATTR_TTL, AttributeValues.fromLong(Instant.now().plus(ttl).getEpochSecond())
         ))
         .build());
   }

service/src/main/java/org/whispersystems/textsecuregcm/storage/ReportMessageManager.java

@@ -3,15 +3,18 @@ package org.whispersystems.textsecuregcm.storage;
 import static com.codahale.metrics.MetricRegistry.name;
 
 import com.google.common.annotations.VisibleForTesting;
+import io.lettuce.core.RedisException;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
 import java.util.Objects;
 import java.util.UUID;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 import org.whispersystems.textsecuregcm.util.UUIDUtil;
 import org.whispersystems.textsecuregcm.util.Util;
 
@@ -21,14 +24,23 @@ public class ReportMessageManager {
   static final String REPORT_COUNTER_NAME = name(ReportMessageManager.class, "reported");
 
   private final ReportMessageDynamoDb reportMessageDynamoDb;
+  private final FaultTolerantRedisCluster rateLimitCluster;
   private final MeterRegistry meterRegistry;
 
+  private final Duration counterTtl;
+
   private static final Logger logger = LoggerFactory.getLogger(ReportMessageManager.class);
 
-  public ReportMessageManager(ReportMessageDynamoDb reportMessageDynamoDb, final MeterRegistry meterRegistry) {
+  public ReportMessageManager(final ReportMessageDynamoDb reportMessageDynamoDb,
+      final FaultTolerantRedisCluster rateLimitCluster,
+      final MeterRegistry meterRegistry,
+      final Duration counterTtl) {
 
     this.reportMessageDynamoDb = reportMessageDynamoDb;
+    this.rateLimitCluster = rateLimitCluster;
     this.meterRegistry = meterRegistry;
+
+    this.counterTtl = counterTtl;
   }
 
   public void store(String sourceNumber, UUID messageGuid) {
@@ -47,6 +59,13 @@ public class ReportMessageManager {
     final boolean found = reportMessageDynamoDb.remove(hash(messageGuid, sourceNumber));
 
     if (found) {
+      rateLimitCluster.useCluster(connection -> {
+        final String reportedSenderKey = getReportedSenderKey(sourceNumber);
+
+        connection.sync().pfadd(reportedSenderKey, sourceNumber);
+        connection.sync().expire(reportedSenderKey, counterTtl.toSeconds());
+      });
+
       Counter.builder(REPORT_COUNTER_NAME)
           .tag("countryCode", Util.getCountryCode(sourceNumber))
           .register(meterRegistry)
@@ -54,6 +73,23 @@ public class ReportMessageManager {
     }
   }
 
+  /**
+   * Returns the number of times messages from the given number 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
+   *
+   * @return the number of times the given number has been reported recently
+   */
+  public int getRecentReportCount(final String number) {
+    try {
+      return rateLimitCluster.withCluster(connection -> connection.sync().pfcount(getReportedSenderKey(number)).intValue());
+    } catch (final RedisException e) {
+      return 0;
+    }
+  }
+
   private byte[] hash(UUID messageGuid, String otherId) {
     final MessageDigest sha256;
     try {
@@ -67,4 +103,8 @@ public class ReportMessageManager {
 
     return sha256.digest();
   }
+
+  private static String getReportedSenderKey(final String senderNumber) {
+    return "reported_number::" + senderNumber;
+  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/workers/DeleteUserCommand.java

@@ -172,6 +172,8 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
           configuration.getMetricsClusterConfiguration(), redisClusterClientResources);
       FaultTolerantRedisCluster clientPresenceCluster    = new FaultTolerantRedisCluster("client_presence_cluster",
           configuration.getClientPresenceClusterConfiguration(), redisClusterClientResources);
+      FaultTolerantRedisCluster rateLimitersCluster = new FaultTolerantRedisCluster("rate_limiters",
+          configuration.getRateLimitersCluster(), redisClusterClientResources);
       SecureBackupClient secureBackupClient = new SecureBackupClient(backupCredentialsGenerator, backupServiceExecutor,
           configuration.getSecureBackupServiceConfiguration());
       SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
@@ -186,9 +188,10 @@ public class DeleteUserCommand extends EnvironmentCommand<WhisperServerConfigura
       UsernamesManager usernamesManager = new UsernamesManager(usernames, reservedUsernames, cacheCluster);
       ProfilesManager profilesManager = new ProfilesManager(profiles, cacheCluster);
       ReportMessageDynamoDb reportMessageDynamoDb = new ReportMessageDynamoDb(reportMessagesDynamoDb,
-          configuration.getReportMessageDynamoDbConfiguration().getTableName());
+          configuration.getReportMessageDynamoDbConfiguration().getTableName(),
-      ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb,
+          configuration.getReportMessageConfiguration().getReportTtl());
-          Metrics.globalRegistry);
+      ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb, rateLimitersCluster,
+          Metrics.globalRegistry, configuration.getReportMessageConfiguration().getCounterTtl());
       MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, pushLatencyManager,
           reportMessageManager);
       DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,

service/src/main/java/org/whispersystems/textsecuregcm/workers/SetUserDiscoverabilityCommand.java

@@ -114,6 +114,8 @@ public class SetUserDiscoverabilityCommand extends EnvironmentCommand<WhisperSer
 
       FaultTolerantRedisCluster cacheCluster = new FaultTolerantRedisCluster("main_cache_cluster",
           configuration.getCacheClusterConfiguration(), redisClusterClientResources);
+      FaultTolerantRedisCluster rateLimitersCluster = new FaultTolerantRedisCluster("rate_limiters",
+          configuration.getRateLimitersCluster(), redisClusterClientResources);
 
       ExecutorService keyspaceNotificationDispatchExecutor = environment.lifecycle()
           .executorService(name(getClass(), "keyspaceNotification-%d")).maxThreads(4).build();
@@ -189,9 +191,10 @@ public class SetUserDiscoverabilityCommand extends EnvironmentCommand<WhisperSer
       UsernamesManager usernamesManager = new UsernamesManager(usernames, reservedUsernames, cacheCluster);
       ProfilesManager profilesManager = new ProfilesManager(profiles, cacheCluster);
       ReportMessageDynamoDb reportMessageDynamoDb = new ReportMessageDynamoDb(reportMessagesDynamoDb,
-          configuration.getReportMessageDynamoDbConfiguration().getTableName());
+          configuration.getReportMessageDynamoDbConfiguration().getTableName(),
-      ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb,
+          configuration.getReportMessageConfiguration().getReportTtl());
-          Metrics.globalRegistry);
+      ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb, rateLimitersCluster,
+          Metrics.globalRegistry, configuration.getReportMessageConfiguration().getCounterTtl());
       MessagesManager messagesManager = new MessagesManager(messagesDynamoDb, messagesCache, pushLatencyManager,
           reportMessageManager);
       DeletedAccountsManager deletedAccountsManager = new DeletedAccountsManager(deletedAccounts,

service/src/test/java/org/whispersystems/textsecuregcm/storage/ReportMessageDynamoDbTest.java

@@ -4,6 +4,7 @@ import static org.junit.jupiter.api.Assertions.assertAll;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.time.Duration;
 import java.util.UUID;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -31,7 +32,7 @@ class ReportMessageDynamoDbTest {
 
   @BeforeEach
   void setUp() {
-    this.reportMessageDynamoDb = new ReportMessageDynamoDb(dynamoDbExtension.getDynamoDbClient(), TABLE_NAME);
+    this.reportMessageDynamoDb = new ReportMessageDynamoDb(dynamoDbExtension.getDynamoDbClient(), TABLE_NAME, Duration.ofDays(1));
   }
 
   @Test

service/src/test/java/org/whispersystems/textsecuregcm/storage/ReportMessageManagerTest.java

@@ -12,15 +12,18 @@ import static org.mockito.Mockito.when;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import java.time.Duration;
 import java.util.UUID;
 import org.junit.jupiter.api.Test;
+import org.whispersystems.textsecuregcm.redis.FaultTolerantRedisCluster;
 
 class ReportMessageManagerTest {
 
   private final ReportMessageDynamoDb reportMessageDynamoDb = mock(ReportMessageDynamoDb.class);
   private final MeterRegistry meterRegistry = new SimpleMeterRegistry();
 
-  private final ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb, meterRegistry);
+  private final ReportMessageManager reportMessageManager = new ReportMessageManager(reportMessageDynamoDb,
+      mock(FaultTolerantRedisCluster.class), meterRegistry, Duration.ofDays(1));
 
   @Test
   void testStore() {
@@ -28,7 +31,7 @@ class ReportMessageManagerTest {
     final UUID messageGuid = UUID.randomUUID();
     final String number = "+15105551111";
 
-    assertDoesNotThrow(() ->  reportMessageManager.store(null, messageGuid));
+    assertDoesNotThrow(() -> reportMessageManager.store(null, messageGuid));
 
     verifyZeroInteractions(reportMessageDynamoDb);
 
@@ -37,7 +40,7 @@ class ReportMessageManagerTest {
     verify(reportMessageDynamoDb).store(any());
 
     doThrow(RuntimeException.class)
-      .when(reportMessageDynamoDb).store(any());
+        .when(reportMessageDynamoDb).store(any());
 
     assertDoesNotThrow(() -> reportMessageManager.store(number, messageGuid));
   }