Retrieve Cloudflare Turn Credentials from Cloudflare

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

@@ -28,6 +28,11 @@ import io.micrometer.core.instrument.Metrics;
 import io.micrometer.core.instrument.binder.grpc.MetricCollectingServerInterceptor;
 import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
 import io.netty.channel.local.LocalAddress;
+import io.netty.channel.socket.nio.NioDatagramChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.resolver.ResolvedAddressTypes;
+import io.netty.resolver.dns.DnsNameResolver;
+import io.netty.resolver.dns.DnsNameResolverBuilder;
 import java.io.FileInputStream;
 import java.net.http.HttpClient;
 import java.security.KeyStore;
@@ -73,6 +78,7 @@ import org.whispersystems.textsecuregcm.attachments.TusAttachmentGenerator;
 import org.whispersystems.textsecuregcm.auth.AccountAuthenticator;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
 import org.whispersystems.textsecuregcm.auth.CertificateGenerator;
+import org.whispersystems.textsecuregcm.auth.CloudflareTurnCredentialsManager;
 import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
 import org.whispersystems.textsecuregcm.auth.PhoneVerificationTokenManager;
 import org.whispersystems.textsecuregcm.auth.RegistrationLockVerificationManager;
@@ -539,9 +545,24 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         .workQueue(new SynchronousQueue<>())
         .keepAliveTime(io.dropwizard.util.Duration.seconds(60L))
         .build();
+    ExecutorService cloudflareTurnHttpExecutor = environment.lifecycle()
+        .executorService(name(getClass(), "cloudflareTurn-%d"))
+        .maxThreads(2)
+        .minThreads(2)
+        .build();
 
     ScheduledExecutorService subscriptionProcessorRetryExecutor = environment.lifecycle()
         .scheduledExecutorService(name(getClass(), "subscriptionProcessorRetry-%d")).threads(1).build();
+    ScheduledExecutorService cloudflareTurnRetryExecutor = environment.lifecycle()
+        .scheduledExecutorService(name(getClass(), "cloudflareTurnRetry-%d")).threads(1).build();
+
+    final ManagedNioEventLoopGroup dnsResolutionEventLoopGroup = new ManagedNioEventLoopGroup();
+    final DnsNameResolver cloudflareDnsResolver = new DnsNameResolverBuilder(dnsResolutionEventLoopGroup.next())
+            .resolvedAddressTypes(ResolvedAddressTypes.IPV6_PREFERRED)
+            .completeOncePreferredResolved(false)
+            .channelType(NioDatagramChannel.class)
+            .socketChannelType(NioSocketChannel.class)
+            .build();
 
     ExternalServiceCredentialsGenerator directoryV2CredentialsGenerator = DirectoryV2Controller.credentialsGenerator(
         config.getDirectoryV2Configuration().getDirectoryV2ClientConfiguration());
@@ -635,7 +656,20 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         pushLatencyManager);
     final ReceiptSender receiptSender = new ReceiptSender(accountsManager, messageSender, receiptSenderExecutor);
     final TurnTokenGenerator turnTokenGenerator = new TurnTokenGenerator(dynamicConfigurationManager,
-        config.getTurnConfiguration().secret().value(), config.getTurnConfiguration().cloudflare());
+        config.getTurnConfiguration().secret().value());
+    final CloudflareTurnCredentialsManager cloudflareTurnCredentialsManager = new CloudflareTurnCredentialsManager(
+        config.getTurnConfiguration().cloudflare().apiToken().value(),
+        config.getTurnConfiguration().cloudflare().endpoint(),
+        config.getTurnConfiguration().cloudflare().ttl(),
+        config.getTurnConfiguration().cloudflare().urls(),
+        config.getTurnConfiguration().cloudflare().urlsWithIps(),
+        config.getTurnConfiguration().cloudflare().hostname(),
+        config.getTurnConfiguration().cloudflare().circuitBreaker(),
+        cloudflareTurnHttpExecutor,
+        config.getTurnConfiguration().cloudflare().retry(),
+        cloudflareTurnRetryExecutor,
+        cloudflareDnsResolver
+        );
 
     final CardinalityEstimator messageByteLimitCardinalityEstimator = new CardinalityEstimator(
         rateLimitersCluster,
@@ -887,6 +921,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         config.getNoiseWebSocketTunnelConfiguration().recognizedProxySecret().value());
 
     environment.lifecycle().manage(localEventLoopGroup);
+    environment.lifecycle().manage(dnsResolutionEventLoopGroup);
     environment.lifecycle().manage(anonymousGrpcServer);
     environment.lifecycle().manage(authenticatedGrpcServer);
     environment.lifecycle().manage(noiseWebSocketEventLoopGroup);
@@ -1018,7 +1053,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new AttachmentControllerV4(rateLimiters, gcsAttachmentGenerator, tusAttachmentGenerator,
             experimentEnrollmentManager),
         new ArchiveController(backupAuthManager, backupManager),
-        new CallRoutingController(rateLimiters, callRouter, turnTokenGenerator, experimentEnrollmentManager),
+        new CallRoutingController(rateLimiters, callRouter, turnTokenGenerator, experimentEnrollmentManager, cloudflareTurnCredentialsManager),
         new CallLinkController(rateLimiters, callingGenericZkSecretParams),
         new CertificateController(new CertificateGenerator(config.getDeliveryCertificate().certificate().value(),
             config.getDeliveryCertificate().ecPrivateKey(), config.getDeliveryCertificate().expiresDays()),

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

@@ -0,0 +1,116 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.auth;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import io.netty.resolver.dns.DnsNameResolver;
+import java.io.IOException;
+import java.net.Inet6Address;
+import java.net.URI;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.util.List;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import javax.ws.rs.core.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.configuration.CircuitBreakerConfiguration;
+import org.whispersystems.textsecuregcm.configuration.RetryConfiguration;
+import org.whispersystems.textsecuregcm.http.FaultTolerantHttpClient;
+import org.whispersystems.textsecuregcm.util.ExceptionUtils;
+import org.whispersystems.textsecuregcm.util.SystemMapper;
+
+public class CloudflareTurnCredentialsManager {
+
+  private static final Logger logger = LoggerFactory.getLogger(CloudflareTurnCredentialsManager.class);
+
+  private final List<String> cloudflareTurnUrls;
+  private final List<String> cloudflareTurnUrlsWithIps;
+  private final String cloudflareTurnHostname;
+  private final HttpRequest request;
+
+  private final FaultTolerantHttpClient cloudflareTurnClient;
+  private final DnsNameResolver dnsNameResolver;
+
+  record CredentialRequest(long ttl) {}
+
+  record CloudflareTurnResponse(IceServer iceServers) {
+
+    record IceServer(
+        String username,
+        String credential,
+        List<String> urls) {
+    }
+  }
+
+  public CloudflareTurnCredentialsManager(final String cloudflareTurnApiToken,
+      final String cloudflareTurnEndpoint, final long cloudflareTurnTtl, final List<String> cloudflareTurnUrls,
+      final List<String> cloudflareTurnUrlsWithIps, final String cloudflareTurnHostname,
+      final CircuitBreakerConfiguration circuitBreaker, final ExecutorService executor, final RetryConfiguration retry,
+      final ScheduledExecutorService retryExecutor, final DnsNameResolver dnsNameResolver) {
+
+    this.cloudflareTurnClient = FaultTolerantHttpClient.newBuilder()
+        .withName("cloudflare-turn")
+        .withCircuitBreaker(circuitBreaker)
+        .withExecutor(executor)
+        .withRetry(retry)
+        .withRetryExecutor(retryExecutor)
+        .build();
+    this.cloudflareTurnUrls = cloudflareTurnUrls;
+    this.cloudflareTurnUrlsWithIps = cloudflareTurnUrlsWithIps;
+    this.cloudflareTurnHostname = cloudflareTurnHostname;
+    this.dnsNameResolver = dnsNameResolver;
+
+    try {
+      final String body = SystemMapper.jsonMapper().writeValueAsString(new CredentialRequest(cloudflareTurnTtl));
+      this.request = HttpRequest.newBuilder()
+          .uri(URI.create(cloudflareTurnEndpoint))
+          .header("Content-Type", "application/json")
+          .header("Authorization", String.format("Bearer %s", cloudflareTurnApiToken))
+          .POST(HttpRequest.BodyPublishers.ofString(body))
+          .build();
+    } catch (JsonProcessingException e) {
+      throw new IllegalArgumentException(e);
+    }
+  }
+
+  public TurnToken retrieveFromCloudflare() throws IOException {
+    final List<String> cloudflareTurnComposedUrls;
+    try {
+      cloudflareTurnComposedUrls = dnsNameResolver.resolveAll(cloudflareTurnHostname).get().stream()
+          .map(i -> switch (i) {
+            case Inet6Address i6 -> "[" + i6.getHostAddress() + "]";
+            default -> i.getHostAddress();
+          })
+          .flatMap(i -> cloudflareTurnUrlsWithIps.stream().map(u -> u.formatted(i)))
+          .toList();
+    } catch (Exception e) {
+      throw new IOException(e);
+    }
+
+    final HttpResponse<String> response;
+    try {
+      response = cloudflareTurnClient.sendAsync(request, HttpResponse.BodyHandlers.ofString()).join();
+    } catch (CompletionException e) {
+      logger.warn("failed to make http request to Cloudflare Turn: {}", e.getMessage());
+      throw new IOException(ExceptionUtils.unwrap(e));
+    }
+
+    if (response.statusCode() != Response.Status.CREATED.getStatusCode()) {
+      logger.warn("failure request credentials from Cloudflare Turn (code={}): {}", response.statusCode(), response);
+      throw new IOException("Cloudflare Turn http failure : " + response.statusCode());
+    }
+
+    final CloudflareTurnResponse cloudflareTurnResponse = SystemMapper.jsonMapper()
+        .readValue(response.body(), CloudflareTurnResponse.class);
+
+    return new TurnToken(cloudflareTurnResponse.iceServers().username(),
+        cloudflareTurnResponse.iceServers().credential(),
+        cloudflareTurnUrls, cloudflareTurnComposedUrls, cloudflareTurnHostname);
+  }
+}

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

@@ -17,7 +17,6 @@ import java.util.UUID;
 import javax.crypto.Mac;
 import javax.crypto.spec.SecretKeySpec;
 import org.whispersystems.textsecuregcm.calls.routing.TurnServerOptions;
-import org.whispersystems.textsecuregcm.configuration.CloudflareTurnConfiguration;
 import org.whispersystems.textsecuregcm.configuration.TurnUriConfiguration;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicTurnConfiguration;
@@ -38,21 +37,10 @@ public class TurnTokenGenerator {
 
   private static final String WithIpsProtocol = "01";
 
-  private final String cloudflareTurnUsername;
-  private final String cloudflareTurnPassword;
-  private final List<String> cloudflareTurnUrls;
-  private final String cloudflareTurnHostname;
-
   public TurnTokenGenerator(final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager,
-      final byte[] turnSecret, final CloudflareTurnConfiguration cloudflareTurnConfiguration) {
-
+      final byte[] turnSecret) {
     this.dynamicConfigurationManager = dynamicConfigurationManager;
     this.turnSecret = turnSecret;
-
-    this.cloudflareTurnUsername = cloudflareTurnConfiguration.username().value();
-    this.cloudflareTurnPassword = cloudflareTurnConfiguration.password().value();
-    this.cloudflareTurnUrls = cloudflareTurnConfiguration.urls();
-    this.cloudflareTurnHostname = cloudflareTurnConfiguration.hostname();
   }
 
   @Deprecated
@@ -64,10 +52,6 @@ public class TurnTokenGenerator {
     return generateToken(options.hostname(), options.urlsWithIps(), options.urlsWithHostname());
   }
 
-  public TurnToken generateForCloudflareBeta() {
-    return new TurnToken(cloudflareTurnUsername, cloudflareTurnPassword, cloudflareTurnUrls, null, cloudflareTurnHostname);
-  }
-
   private TurnToken generateToken(String hostname, List<String> urlsWithIps, List<String> urlsWithHostname) {
     try {
       final Mac mac = Mac.getInstance(ALGORITHM);

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

@@ -11,7 +11,25 @@ import javax.validation.Valid;
 import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
 
-public record CloudflareTurnConfiguration(@NotNull SecretString username, @NotNull SecretString password,
-                                          @Valid @NotNull List<@NotBlank String> urls, @NotBlank String hostname) {
+public record CloudflareTurnConfiguration(@NotNull SecretString apiToken,
+                                          @NotBlank String endpoint,
+                                          @NotBlank long ttl,
+                                          @NotBlank List<String> urls,
+                                          @NotBlank List<String> urlsWithIps,
+                                          @NotNull @Valid CircuitBreakerConfiguration circuitBreaker,
+                                          @NotNull @Valid RetryConfiguration retry,
+                                          @NotBlank String hostname) {
 
+  public CloudflareTurnConfiguration {
+    if (circuitBreaker == null) {
+      // It’s a little counter-intuitive, but this compact constructor allows a default value
+      // to be used when one isn’t specified (e.g. in YAML), allowing the field to still be
+      // validated as @NotNull
+      circuitBreaker = new CircuitBreakerConfiguration();
+    }
+
+    if (retry == null) {
+      retry = new RetryConfiguration();
+    }
+  }
 }

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

@@ -7,6 +7,7 @@ import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Metrics;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import java.io.IOException;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.Optional;
@@ -20,6 +21,7 @@ import javax.ws.rs.core.MediaType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
+import org.whispersystems.textsecuregcm.auth.CloudflareTurnCredentialsManager;
 import org.whispersystems.textsecuregcm.auth.TurnToken;
 import org.whispersystems.textsecuregcm.auth.TurnTokenGenerator;
 import org.whispersystems.textsecuregcm.calls.routing.TurnCallRouter;
@@ -40,17 +42,20 @@ public class CallRoutingController {
   private final TurnCallRouter turnCallRouter;
   private final TurnTokenGenerator tokenGenerator;
   private final ExperimentEnrollmentManager experimentEnrollmentManager;
+  private final CloudflareTurnCredentialsManager cloudflareTurnCredentialsManager;
 
   public CallRoutingController(
       final RateLimiters rateLimiters,
       final TurnCallRouter turnCallRouter,
       final TurnTokenGenerator tokenGenerator,
-      final ExperimentEnrollmentManager experimentEnrollmentManager
+      final ExperimentEnrollmentManager experimentEnrollmentManager,
+      final CloudflareTurnCredentialsManager cloudflareTurnCredentialsManager
   ) {
     this.rateLimiters = rateLimiters;
     this.turnCallRouter = turnCallRouter;
     this.tokenGenerator = tokenGenerator;
     this.experimentEnrollmentManager = experimentEnrollmentManager;
+    this.cloudflareTurnCredentialsManager = cloudflareTurnCredentialsManager;
   }
 
   @GET
@@ -70,12 +75,12 @@ public class CallRoutingController {
   public TurnToken getCallingRelays(
       final @ReadOnly @Auth AuthenticatedAccount auth,
       @Context ContainerRequestContext requestContext
-  ) throws RateLimitExceededException {
+  ) throws RateLimitExceededException, IOException {
     UUID aci = auth.getAccount().getUuid();
     rateLimiters.getCallEndpointLimiter().validate(aci);
 
     if (experimentEnrollmentManager.isEnrolled(aci, "cloudflareTurn")) {
-      return tokenGenerator.generateForCloudflareBeta();
+      return cloudflareTurnCredentialsManager.retrieveFromCloudflare();
     }
 
     Optional<InetAddress> address = Optional.empty();