Make TURN configuration dynamic

Also enables conditionally including more TURN servers for gradual
rollouts

service/config/sample.yml

@@ -78,14 +78,6 @@ twilio: # Twilio gateway configuration
   androidAppHash: example # Hash appended to Android
   verifyServiceFriendlyName: example # Service name used in template. Requires Twilio account rep to enable
 
-turn: # TURN server configuration
-  secret: example # TURN server secret
-  uris:
-    - stun:example.com:80
-    - stun:another.example.com:443
-    - turn:example.com:443?transport=udp
-    - turn:ya.example.com:80?transport=udp
-
 cacheCluster: # Redis server configuration for cache cluster
   configurationUri: redis://redis.example.com:6379/
 

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

@@ -43,7 +43,6 @@ import org.whispersystems.textsecuregcm.configuration.SecureStorageServiceConfig
 import org.whispersystems.textsecuregcm.configuration.StripeConfiguration;
 import org.whispersystems.textsecuregcm.configuration.SubscriptionConfiguration;
 import org.whispersystems.textsecuregcm.configuration.TestDeviceConfiguration;
-import org.whispersystems.textsecuregcm.configuration.TurnConfiguration;
 import org.whispersystems.textsecuregcm.configuration.TwilioConfiguration;
 import org.whispersystems.textsecuregcm.configuration.UnidentifiedDeliveryConfiguration;
 import org.whispersystems.textsecuregcm.configuration.VoiceVerificationConfiguration;
@@ -168,11 +167,6 @@ public class WhisperServerConfiguration extends Configuration {
   @JsonProperty
   private WebSocketConfiguration webSocket = new WebSocketConfiguration();
 
-  @Valid
-  @NotNull
-  @JsonProperty
-  private TurnConfiguration turn;
-
   @Valid
   @NotNull
   @JsonProperty
@@ -345,10 +339,6 @@ public class WhisperServerConfiguration extends Configuration {
     return limits;
   }
 
-  public TurnConfiguration getTurnConfiguration() {
-    return turn;
-  }
-
   public GcmConfiguration getGcmConfiguration() {
     return gcm;
   }

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

@@ -484,7 +484,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     SmsSender                smsSender          = new SmsSender(twilioSmsSender);
     MessageSender            messageSender      = new MessageSender(apnFallbackManager, clientPresenceManager, messagesManager, gcmSender, apnSender, pushLatencyManager);
     ReceiptSender            receiptSender      = new ReceiptSender(accountsManager, messageSender, receiptSenderExecutor);
-    TurnTokenGenerator       turnTokenGenerator = new TurnTokenGenerator(config.getTurnConfiguration());
+    TurnTokenGenerator       turnTokenGenerator = new TurnTokenGenerator(dynamicConfigurationManager);
     RecaptchaClient recaptchaClient = new RecaptchaClient(
         config.getRecaptchaConfiguration().getProjectPath(),
         config.getRecaptchaConfiguration().getCredentialConfigurationJson(),

service/src/main/java/org/whispersystems/textsecuregcm/auth/TurnToken.java

@@ -6,6 +6,7 @@
 package org.whispersystems.textsecuregcm.auth;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.annotations.VisibleForTesting;
 
 import java.util.List;
 
@@ -25,4 +26,9 @@ public class TurnToken {
     this.password = password;
     this.urls     = urls;
   }
+
+  @VisibleForTesting
+  List<String> getUrls() {
+    return urls;
+  }
 }

service/src/main/java/org/whispersystems/textsecuregcm/auth/TurnTokenGenerator.java

@@ -5,7 +5,12 @@
 
 package org.whispersystems.textsecuregcm.auth;
 
-import org.whispersystems.textsecuregcm.configuration.TurnConfiguration;
+import org.whispersystems.textsecuregcm.configuration.TurnUriConfiguration;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicTurnConfiguration;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
+import org.whispersystems.textsecuregcm.util.Pair;
+import org.whispersystems.textsecuregcm.util.WeightedRandomSelect;
 
 import javax.crypto.Mac;
 import javax.crypto.spec.SecretKeySpec;
@@ -14,20 +19,21 @@ import java.security.NoSuchAlgorithmException;
 import java.security.SecureRandom;
 import java.util.Base64;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 
 public class TurnTokenGenerator {
 
-  private final byte[]       key;
+  private final DynamicConfigurationManager<DynamicConfiguration> dynamicConfiguration;
-  private final List<String> urls;
 
-  public TurnTokenGenerator(TurnConfiguration configuration) {
+  public TurnTokenGenerator(final DynamicConfigurationManager<DynamicConfiguration> config) {
-    this.key  = configuration.getSecret().getBytes();
+    this.dynamicConfiguration = config;
-    this.urls = configuration.getUris();
   }
 
-  public TurnToken generate() {
+  public TurnToken generate(final String e164) {
     try {
+      byte[] key                = dynamicConfiguration.getConfiguration().getTurnConfiguration().getSecret().getBytes();
+      List<String> urls         = urls(e164);
       Mac    mac                = Mac.getInstance("HmacSHA1");
       long   validUntilSeconds  = (System.currentTimeMillis() + TimeUnit.DAYS.toMillis(1)) / 1000;
       long   user               = Math.abs(new SecureRandom().nextInt());
@@ -41,4 +47,22 @@ public class TurnTokenGenerator {
       throw new AssertionError(e);
     }
   }
+
+  private List<String> urls(final String e164) {
+    final DynamicTurnConfiguration turnConfig = dynamicConfiguration.getConfiguration().getTurnConfiguration();
+
+    // Check if number is enrolled to test out specific turn servers
+    final Optional<TurnUriConfiguration> enrolled = turnConfig.getUriConfigs().stream()
+        .filter(config -> config.getEnrolledNumbers().contains(e164))
+        .findFirst();
+    if (enrolled.isPresent()) {
+      return enrolled.get().getUris();
+    }
+
+    // Otherwise, select from turn server sets by weighted choice
+    return WeightedRandomSelect.select(turnConfig
+        .getUriConfigs()
+        .stream()
+        .map(c -> new Pair<List<String>, Long>(c.getUris(), c.getWeight())).toList());
+  }
 }

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

@@ -1,30 +0,0 @@
-/*
- * Copyright 2013-2020 Signal Messenger, LLC
- * SPDX-License-Identifier: AGPL-3.0-only
- */
-
-package org.whispersystems.textsecuregcm.configuration;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-import java.util.List;
-import javax.validation.constraints.NotEmpty;
-import javax.validation.constraints.NotNull;
-
-public class TurnConfiguration {
-
-  @JsonProperty
-  @NotEmpty
-  private String secret;
-
-  @JsonProperty
-  @NotNull
-  private List<String> uris;
-
-  public List<String> getUris() {
-    return uris;
-  }
-
-  public String getSecret() {
-    return secret;
-  }
-}

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

@@ -0,0 +1,38 @@
+package org.whispersystems.textsecuregcm.configuration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+public class TurnUriConfiguration {
+  @JsonProperty
+  @NotNull
+  private List<String> uris;
+
+  /**
+   * The weight of this entry for weighted random selection
+   */
+  @JsonProperty
+  @Min(0)
+  private long weight = 1;
+
+  /**
+   * Enrolled numbers will always get this uri list
+   */
+  private Set<String> enrolledNumbers = Collections.emptySet();
+
+  public List<String> getUris() {
+    return uris;
+  }
+
+  public long getWeight() {
+    return weight;
+  }
+
+  public Set<String> getEnrolledNumbers() {
+    return Collections.unmodifiableSet(enrolledNumbers);
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicConfiguration.java

@@ -56,6 +56,10 @@ public class DynamicConfiguration {
   @Valid
   private DynamicUakMigrationConfiguration uakMigrationConfiguration = new DynamicUakMigrationConfiguration();
 
+  @JsonProperty
+  @Valid
+  private DynamicTurnConfiguration turn = new DynamicTurnConfiguration();
+
   public Optional<DynamicExperimentEnrollmentConfiguration> getExperimentEnrollmentConfiguration(
       final String experimentName) {
     return Optional.ofNullable(experiments.get(experimentName));
@@ -109,4 +113,8 @@ public class DynamicConfiguration {
 
   public DynamicUakMigrationConfiguration getUakMigrationConfiguration() { return uakMigrationConfiguration; }
 
+  public DynamicTurnConfiguration getTurnConfiguration() {
+    return turn;
+  }
+
 }

service/src/main/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicTurnConfiguration.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2013-2020 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.configuration.dynamic;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.whispersystems.textsecuregcm.configuration.TurnUriConfiguration;
+import java.util.Collections;
+import java.util.List;
+import javax.validation.Valid;
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+
+public class DynamicTurnConfiguration {
+
+  @JsonProperty
+  private String secret;
+
+  @JsonProperty
+  private List<@Valid TurnUriConfiguration> uriConfigs = Collections.emptyList();
+
+  public List<TurnUriConfiguration> getUriConfigs() {
+    return uriConfigs;
+  }
+
+  public String getSecret() {
+    return secret;
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/controllers/AccountController.java

@@ -449,7 +449,7 @@ public class AccountController {
   @Produces(MediaType.APPLICATION_JSON)
   public TurnToken getTurnToken(@Auth AuthenticatedAccount auth) throws RateLimitExceededException {
     rateLimiters.getTurnLimiter().validate(auth.getAccount().getUuid());
-    return turnTokenGenerator.generate();
+    return turnTokenGenerator.generate(auth.getAccount().getNumber());
   }
 
   @Timed

service/src/main/java/org/whispersystems/textsecuregcm/util/WeightedRandomSelect.java

@@ -0,0 +1,54 @@
+package org.whispersystems.textsecuregcm.util;
+
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Select a random item according to its weight
+ *
+ * @param <T> the type of the objects to select from
+ */
+public class WeightedRandomSelect<T> {
+
+  List<Pair<T, Long>> weightedItems;
+  long totalWeight;
+
+  public WeightedRandomSelect(List<Pair<T, Long>> weightedItems) throws IllegalArgumentException {
+    this.weightedItems = weightedItems;
+    this.totalWeight = weightedItems.stream().mapToLong(Pair::second).sum();
+
+    weightedItems.stream().map(Pair::second).filter(w -> w < 0).findFirst().ifPresent(invalid -> {
+      throw new IllegalArgumentException("Illegal selection weight " + invalid);
+    });
+
+    if (weightedItems.isEmpty() || totalWeight == 0) {
+      throw new IllegalArgumentException("Cannot create an empty weighted random selector");
+    }
+  }
+
+  public T select() {
+    if (weightedItems.size() == 1) {
+      return weightedItems.get(0).first();
+    }
+    long select = ThreadLocalRandom.current().nextLong(0, totalWeight);
+    long current = 0;
+    for (Pair<T, Long> item : weightedItems) {
+      /*
+        Accumulate weights for each item and select the first item whose
+        cumulative weight exceeds the selected value. nextLong() is exclusive,
+        so by the last item we're guaranteed to find a value as the
+        last item's weight is one more than the maximum value of select.
+      */
+      current += item.second();
+      if (current > select) {
+        return item.first();
+      }
+    }
+    throw new IllegalStateException("totalWeight " + totalWeight + " exceeds item weights");
+  }
+
+  public static <T> T select(List<Pair<T, Long>> weightedItems) {
+    return new WeightedRandomSelect<T>(weightedItems).select();
+  }
+
+}

service/src/test/java/org/whispersystems/textsecuregcm/auth/TurnTokenGeneratorTest.java

@@ -0,0 +1,133 @@
+package org.whispersystems.textsecuregcm.auth;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import org.junit.jupiter.api.Test;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TurnTokenGeneratorTest {
+
+  @Test
+  public void testAlwaysSelectFirst() throws JsonProcessingException {
+    final String configString = """
+        captcha:
+          scoreFloor: 1.0
+        turn:
+          secret: bloop
+          uriConfigs:
+            - uris:
+                - always1.org
+                - always2.org
+            - uris:
+                - never.org
+              weight: 0
+         """;
+    DynamicConfiguration config = DynamicConfigurationManager
+        .parseConfiguration(configString, DynamicConfiguration.class)
+        .orElseThrow();
+
+    @SuppressWarnings("unchecked")
+    DynamicConfigurationManager<DynamicConfiguration> mockDynamicConfigManager = mock(
+        DynamicConfigurationManager.class);
+
+    when(mockDynamicConfigManager.getConfiguration()).thenReturn(config);
+    final TurnTokenGenerator turnTokenGenerator = new TurnTokenGenerator(mockDynamicConfigManager);
+
+    final long COUNT = 1000;
+
+    final Map<String, Long> urlCounts = Stream
+        .generate(() -> turnTokenGenerator.generate(""))
+        .limit(COUNT)
+        .flatMap(token -> token.getUrls().stream())
+        .collect(Collectors.groupingBy(i -> i, Collectors.counting()));
+
+    assertThat(urlCounts.get("always1.org")).isEqualTo(COUNT);
+    assertThat(urlCounts.get("always2.org")).isEqualTo(COUNT);
+    assertThat(urlCounts).doesNotContainKey("never.org");
+  }
+
+  @Test
+  public void testProbabilisticUrls() throws JsonProcessingException {
+    final String configString = """
+        captcha:
+          scoreFloor: 1.0
+        turn:
+          secret: bloop
+          uriConfigs:
+            - uris:
+                - always.org
+                - sometimes1.org
+              weight: 5
+            - uris:
+                - always.org
+                - sometimes2.org
+              weight: 5
+         """;
+    DynamicConfiguration config = DynamicConfigurationManager
+        .parseConfiguration(configString, DynamicConfiguration.class)
+        .orElseThrow();
+
+    @SuppressWarnings("unchecked")
+    DynamicConfigurationManager<DynamicConfiguration> mockDynamicConfigManager = mock(
+        DynamicConfigurationManager.class);
+
+    when(mockDynamicConfigManager.getConfiguration()).thenReturn(config);
+    final TurnTokenGenerator turnTokenGenerator = new TurnTokenGenerator(mockDynamicConfigManager);
+
+    final long COUNT = 1000;
+
+    final Map<String, Long> urlCounts = Stream
+        .generate(() -> turnTokenGenerator.generate(""))
+        .limit(COUNT)
+        .flatMap(token -> token.getUrls().stream())
+        .collect(Collectors.groupingBy(i -> i, Collectors.counting()));
+
+    assertThat(urlCounts.get("always.org")).isEqualTo(COUNT);
+    assertThat(urlCounts.get("sometimes1.org")).isGreaterThan(0);
+    assertThat(urlCounts.get("sometimes2.org")).isGreaterThan(0);
+  }
+
+  @Test
+  public void testExplicitEnrollment() throws JsonProcessingException {
+    final String configString = """
+        captcha:
+          scoreFloor: 1.0
+        turn:
+          secret: bloop
+          uriConfigs:
+            - uris:
+                - enrolled.org
+              weight: 0
+              enrolledNumbers:
+                - +15555555555
+            - uris:
+                - unenrolled.org
+              weight: 1
+         """;
+    DynamicConfiguration config = DynamicConfigurationManager
+        .parseConfiguration(configString, DynamicConfiguration.class)
+        .orElseThrow();
+
+    @SuppressWarnings("unchecked")
+    DynamicConfigurationManager<DynamicConfiguration> mockDynamicConfigManager = mock(
+        DynamicConfigurationManager.class);
+
+    when(mockDynamicConfigManager.getConfiguration()).thenReturn(config);
+
+    final TurnTokenGenerator turnTokenGenerator = new TurnTokenGenerator(mockDynamicConfigManager);
+    TurnToken token = turnTokenGenerator.generate("+15555555555");
+    assertThat(token.getUrls().get(0)).isEqualTo("enrolled.org");
+    token = turnTokenGenerator.generate("+15555555556");
+    assertThat(token.getUrls().get(0)).isEqualTo("unenrolled.org");
+
+  }
+
+}

service/src/test/java/org/whispersystems/textsecuregcm/configuration/dynamic/DynamicConfigurationTest.java

@@ -395,4 +395,46 @@ class DynamicConfigurationTest {
       assertThat(directoryReconcilerConfiguration.isEnabled()).isFalse();
     }
   }
+
+  @Test
+  void testParseTurnConfig() throws JsonProcessingException {
+    {
+      final String config = REQUIRED_CONFIG.concat("""
+          turn:
+            secret: bloop
+            uriConfigs:
+                - uris:
+                    - turn:test.org
+                  weight: -1
+           """);
+      assertThat(DynamicConfigurationManager.parseConfiguration(config, DynamicConfiguration.class)).isEmpty();
+    }
+    {
+      final String config = REQUIRED_CONFIG.concat("""
+          turn:
+            secret: bloop
+            uriConfigs:
+                - uris:
+                    - turn:test0.org
+                    - turn:test1.org
+                - uris:
+                    - turn:test2.org
+                  weight: 2
+                  enrolledNumbers:
+                    - +15555555555
+           """);
+      DynamicTurnConfiguration turnConfiguration = DynamicConfigurationManager
+          .parseConfiguration(config, DynamicConfiguration.class)
+          .orElseThrow()
+          .getTurnConfiguration();
+      assertThat(turnConfiguration.getSecret()).isEqualTo("bloop");
+      assertThat(turnConfiguration.getUriConfigs().get(0).getUris()).hasSize(2);
+      assertThat(turnConfiguration.getUriConfigs().get(1).getUris()).hasSize(1);
+      assertThat(turnConfiguration.getUriConfigs().get(0).getWeight()).isEqualTo(1);
+      assertThat(turnConfiguration.getUriConfigs().get(1).getWeight()).isEqualTo(2);
+      assertThat(turnConfiguration.getUriConfigs().get(1).getEnrolledNumbers()).containsExactly("+15555555555");
+
+    }
+
+  }
 }

service/src/test/java/org/whispersystems/textsecuregcm/util/WeightedRandomSelectTest.java

@@ -0,0 +1,46 @@
+package org.whispersystems.textsecuregcm.util;
+
+import org.junit.jupiter.api.Test;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class WeightedRandomSelectTest {
+
+  @Test
+  public void test5050() {
+    final WeightedRandomSelect<String> selector = new WeightedRandomSelect<>(
+        List.of(new Pair<>("a", 1L), new Pair<>("b", 1L)));
+    final Map<String, Long> counts = Stream.generate(selector::select)
+        .limit(1000)
+        .collect(Collectors.groupingBy(s -> s, Collectors.counting()));
+    assertThat(counts.get("a")).isGreaterThan(1);
+    assertThat(counts.get("b")).isGreaterThan(1);
+  }
+
+  @Test
+  public void testAlways() {
+    final WeightedRandomSelect<String> selector = new WeightedRandomSelect<>(
+        List.of(new Pair<>("a", 1L), new Pair<>("b", 0L)));
+    final Map<String, Long> counts = Stream.generate(selector::select)
+        .limit(1000)
+        .collect(Collectors.groupingBy(s -> s, Collectors.counting()));
+    assertThat(counts.get("a")).isEqualTo(1000);
+    assertThat(counts).doesNotContainKey("b");
+  }
+
+  @Test
+  public void testThree() {
+    final WeightedRandomSelect<String> selector = new WeightedRandomSelect<>(
+        List.of(new Pair<>("a", 33L), new Pair<>("b", 33L), new Pair<>("c", 33L)));
+    final Map<String, Long> counts = Stream.generate(selector::select)
+        .limit(1000)
+        .collect(Collectors.groupingBy(s -> s, Collectors.counting()));
+    assertThat(counts.get("a")).isGreaterThan(1);
+    assertThat(counts.get("b")).isGreaterThan(1);
+    assertThat(counts.get("c")).isGreaterThan(1);
+  }
+}