Key transparency search and monitor endpoints

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

@@ -37,6 +37,7 @@ import org.whispersystems.textsecuregcm.configuration.FcmConfiguration;
 import org.whispersystems.textsecuregcm.configuration.GcpAttachmentsConfiguration;
 import org.whispersystems.textsecuregcm.configuration.GenericZkConfig;
 import org.whispersystems.textsecuregcm.configuration.HCaptchaClientFactory;
+import org.whispersystems.textsecuregcm.configuration.KeyTransparencyServiceConfiguration;
 import org.whispersystems.textsecuregcm.configuration.LinkDeviceSecretConfiguration;
 import org.whispersystems.textsecuregcm.configuration.MaxDeviceConfiguration;
 import org.whispersystems.textsecuregcm.configuration.MessageByteLimitCardinalityEstimatorConfiguration;
@@ -341,6 +342,11 @@ public class WhisperServerConfiguration extends Configuration {
   @JsonProperty
   private ExternalRequestFilterConfiguration externalRequestFilter;
 
+  @Valid
+  @NotNull
+  @JsonProperty
+  private KeyTransparencyServiceConfiguration keyTransparencyService;
+
   public TlsKeyStoreConfiguration getTlsKeyStoreConfiguration() {
     return tlsKeyStore;
   }
@@ -567,4 +573,8 @@ public class WhisperServerConfiguration extends Configuration {
   public ExternalRequestFilterConfiguration getExternalRequestFilterConfiguration() {
     return externalRequestFilter;
   }
+
+  public KeyTransparencyServiceConfiguration getKeyTransparencyServiceConfiguration() {
+    return keyTransparencyService;
+  }
 }

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

@@ -119,6 +119,7 @@ import org.whispersystems.textsecuregcm.controllers.DeviceController;
 import org.whispersystems.textsecuregcm.controllers.DirectoryV2Controller;
 import org.whispersystems.textsecuregcm.controllers.DonationController;
 import org.whispersystems.textsecuregcm.controllers.KeepAliveController;
+import org.whispersystems.textsecuregcm.controllers.KeyTransparencyController;
 import org.whispersystems.textsecuregcm.controllers.KeysController;
 import org.whispersystems.textsecuregcm.controllers.MessageController;
 import org.whispersystems.textsecuregcm.controllers.PaymentsController;
@@ -159,6 +160,7 @@ import org.whispersystems.textsecuregcm.grpc.net.ManagedLocalGrpcServer;
 import org.whispersystems.textsecuregcm.grpc.net.ManagedNioEventLoopGroup;
 import org.whispersystems.textsecuregcm.grpc.net.NoiseWebSocketTunnelServer;
 import org.whispersystems.textsecuregcm.jetty.JettyHttpConfigurationCustomizer;
+import org.whispersystems.textsecuregcm.keytransparency.KeyTransparencyServiceClient;
 import org.whispersystems.textsecuregcm.limits.CardinalityEstimator;
 import org.whispersystems.textsecuregcm.limits.PushChallengeManager;
 import org.whispersystems.textsecuregcm.limits.RateLimitChallengeManager;
@@ -572,6 +574,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         .maxThreads(2)
         .minThreads(2)
         .build();
+    ExecutorService keyTransparencyCallbackExecutor = new VirtualExecutorServiceProvider(name(getClass(), "keyTransparency-%d"))
+        .getExecutorService();
 
     ScheduledExecutorService subscriptionProcessorRetryExecutor = environment.lifecycle()
         .scheduledExecutorService(name(getClass(), "subscriptionProcessorRetry-%d")).threads(1).build();
@@ -607,6 +611,11 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
 
     RegistrationServiceClient registrationServiceClient = config.getRegistrationServiceConfiguration()
         .build(environment, registrationCallbackExecutor, registrationIdentityTokenRefreshExecutor);
+    KeyTransparencyServiceClient keyTransparencyServiceClient = new KeyTransparencyServiceClient(
+        config.getKeyTransparencyServiceConfiguration().host(),
+        config.getKeyTransparencyServiceConfiguration().port(),
+        config.getKeyTransparencyServiceConfiguration().tlsCertificate(),
+        keyTransparencyCallbackExecutor);
     SecureValueRecovery2Client secureValueRecovery2Client = new SecureValueRecovery2Client(svr2CredentialsGenerator,
         secureValueRecoveryServiceExecutor, secureValueRecoveryServiceRetryExecutor, config.getSvr2Configuration());
     SecureStorageClient secureStorageClient = new SecureStorageClient(storageCredentialsGenerator,
@@ -737,6 +746,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.lifecycle().manage(clientPresenceManager);
     environment.lifecycle().manage(currencyManager);
     environment.lifecycle().manage(registrationServiceClient);
+    environment.lifecycle().manage(keyTransparencyServiceClient);
     environment.lifecycle().manage(clientReleaseManager);
     environment.lifecycle().manage(virtualThreadPinEventMonitor);
 
@@ -978,6 +988,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     final MessageMetrics messageMetrics = new MessageMetrics();
 
     environment.jersey().register(new BufferingInterceptor());
+    environment.jersey().register(keyTransparencyCallbackExecutor);
     environment.jersey().register(new VirtualExecutorServiceProvider("managed-async-virtual-thread-"));
     environment.jersey().register(new RequestStatisticsFilter(TrafficSource.HTTP));
     environment.jersey().register(MultiRecipientMessageProvider.class);
@@ -1084,6 +1095,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new DonationController(clock, zkReceiptOperations, redeemedReceiptsManager, accountsManager, config.getBadges(),
             ReceiptCredentialPresentation::new),
         new KeysController(rateLimiters, keysManager, accountsManager, zkSecretParams, Clock.systemUTC()),
+        new KeyTransparencyController(keyTransparencyServiceClient),
         new MessageController(rateLimiters, messageByteLimitCardinalityEstimator, messageSender, receiptSender,
             accountsManager, messagesManager, pushNotificationManager, reportMessageManager,
             multiRecipientMessageExecutor, messageDeliveryScheduler, reportSpamTokenProvider, clientReleaseManager,

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

@@ -0,0 +1,13 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.configuration;
+
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.Positive;
+
+public record KeyTransparencyServiceConfiguration(@NotBlank String host,
+                                                  @Positive int port,
+                                                  @NotBlank String tlsCertificate) {}

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

@@ -0,0 +1,244 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.controllers;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.ByteString;
+import io.dropwizard.auth.Auth;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import katie.MonitorKey;
+import katie.MonitorProof;
+import katie.MonitorResponse;
+import katie.SearchResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
+import org.whispersystems.textsecuregcm.entities.KeyTransparencyMonitorRequest;
+import org.whispersystems.textsecuregcm.entities.KeyTransparencyMonitorResponse;
+import org.whispersystems.textsecuregcm.entities.KeyTransparencySearchRequest;
+import org.whispersystems.textsecuregcm.entities.KeyTransparencySearchResponse;
+import org.whispersystems.textsecuregcm.keytransparency.KeyTransparencyServiceClient;
+import org.whispersystems.textsecuregcm.limits.RateLimitedByIp;
+import org.whispersystems.textsecuregcm.limits.RateLimiters;
+import org.whispersystems.textsecuregcm.util.ExceptionUtils;
+import org.whispersystems.websocket.auth.ReadOnly;
+
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+import javax.ws.rs.BadRequestException;
+import javax.ws.rs.ForbiddenException;
+import javax.ws.rs.NotFoundException;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.ServerErrorException;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+
+@Path("/v1/key-transparency")
+@Tag(name = "KeyTransparency")
+public class KeyTransparencyController {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(KeyTransparencyController.class);
+  private static final Duration KEY_TRANSPARENCY_RPC_TIMEOUT = Duration.ofSeconds(15);
+  private static final byte USERNAME_PREFIX = (byte) 'u';
+  private static final byte E164_PREFIX = (byte) 'n';
+  @VisibleForTesting
+  static final byte ACI_PREFIX = (byte) 'a';
+  private final KeyTransparencyServiceClient keyTransparencyServiceClient;
+
+  public KeyTransparencyController(
+      final KeyTransparencyServiceClient keyTransparencyServiceClient) {
+    this.keyTransparencyServiceClient = keyTransparencyServiceClient;
+  }
+
+  @Operation(
+      summary = "Search for the given search keys in the key transparency log",
+      description = """
+          Enforced unauthenticated endpoint. Returns a response if all search keys exist in the key transparency log.
+          """
+  )
+  @ApiResponse(responseCode = "200", description = "All search key lookups were successful", useReturnTypeSchema = true)
+  @ApiResponse(responseCode = "403", description = "At least one search key lookup to value mapping was invalid")
+  @ApiResponse(responseCode = "404", description = "At least one search key lookup did not find the key")
+  @ApiResponse(responseCode = "413", description = "Ratelimited")
+  @ApiResponse(responseCode = "422", description = "Invalid request format")
+  @POST
+  @Path("/search")
+  @RateLimitedByIp(RateLimiters.For.KEY_TRANSPARENCY_SEARCH_PER_IP)
+  @Produces(MediaType.APPLICATION_JSON)
+  public KeyTransparencySearchResponse search(
+      @ReadOnly @Auth final Optional<AuthenticatedAccount> authenticatedAccount,
+      @NotNull @Valid final KeyTransparencySearchRequest request) {
+
+    // Disallow clients from making authenticated requests to this endpoint
+    requireNotAuthenticated(authenticatedAccount);
+
+    try {
+      final CompletableFuture<SearchResponse> aciSearchKeyResponseFuture = keyTransparencyServiceClient.search(
+          getFullSearchKeyByteString(ACI_PREFIX, request.aci().toCompactByteArray()),
+          request.lastTreeHeadSize(),
+          KEY_TRANSPARENCY_RPC_TIMEOUT);
+
+      final CompletableFuture<SearchResponse> e164SearchKeyResponseFuture = request.e164()
+          .map(e164 -> keyTransparencyServiceClient.search(
+              getFullSearchKeyByteString(E164_PREFIX, e164.getBytes(StandardCharsets.UTF_8)),
+              request.lastTreeHeadSize(),
+              KEY_TRANSPARENCY_RPC_TIMEOUT))
+          .orElse(CompletableFuture.completedFuture(null));
+
+      final CompletableFuture<SearchResponse> usernameHashSearchKeyResponseFuture = request.usernameHash()
+          .map(usernameHash -> keyTransparencyServiceClient.search(
+              getFullSearchKeyByteString(USERNAME_PREFIX, request.usernameHash().get()),
+              request.lastTreeHeadSize(),
+              KEY_TRANSPARENCY_RPC_TIMEOUT))
+          .orElse(CompletableFuture.completedFuture(null));
+
+      return CompletableFuture.allOf(aciSearchKeyResponseFuture, e164SearchKeyResponseFuture,
+              usernameHashSearchKeyResponseFuture)
+          .thenApply(ignored ->
+              new KeyTransparencySearchResponse(aciSearchKeyResponseFuture.join(),
+                  Optional.ofNullable(e164SearchKeyResponseFuture.join()),
+                  Optional.ofNullable(usernameHashSearchKeyResponseFuture.join())))
+          .join();
+    } catch (final CancellationException exception) {
+      LOGGER.error("Unexpected cancellation from key transparency service", exception);
+      throw new ServerErrorException(Response.Status.SERVICE_UNAVAILABLE, exception);
+    } catch (final CompletionException exception) {
+      handleKeyTransparencyServiceError(exception);
+    }
+    // This is unreachable
+    return null;
+  }
+
+  @Operation(
+      summary = "Monitor the given search keys in the key transparency log",
+      description = """
+          Enforced unauthenticated endpoint. Return proofs proving that the log tree
+          has been constructed correctly in later entries for each of the given search keys .
+          """
+  )
+  @ApiResponse(responseCode = "200", description = "All search keys exist in the log", useReturnTypeSchema = true)
+  @ApiResponse(responseCode = "404", description = "At least one search key lookup did not find the key")
+  @ApiResponse(responseCode = "413", description = "Ratelimited")
+  @ApiResponse(responseCode = "422", description = "Invalid request format")
+  @POST
+  @Path("/monitor")
+  @RateLimitedByIp(RateLimiters.For.KEY_TRANSPARENCY_MONITOR_PER_IP)
+  @Produces(MediaType.APPLICATION_JSON)
+  public KeyTransparencyMonitorResponse monitor(
+      @ReadOnly @Auth final Optional<AuthenticatedAccount> authenticatedAccount,
+      @NotNull @Valid final KeyTransparencyMonitorRequest request) {
+
+    // Disallow clients from making authenticated requests to this endpoint
+    requireNotAuthenticated(authenticatedAccount);
+
+    try {
+      final List<MonitorKey> monitorKeys = new ArrayList<>(List.of(
+          createMonitorKey(getFullSearchKeyByteString(ACI_PREFIX, request.aci().toCompactByteArray()),
+              request.aciPositions())
+      ));
+
+      request.usernameHash().ifPresent(usernameHash ->
+          monitorKeys.add(createMonitorKey(getFullSearchKeyByteString(USERNAME_PREFIX, usernameHash),
+              request.usernameHashPositions().get()))
+      );
+
+      request.e164().ifPresent(e164 ->
+          monitorKeys.add(
+              createMonitorKey(getFullSearchKeyByteString(E164_PREFIX, e164.getBytes(StandardCharsets.UTF_8)),
+                  request.e164Positions().get()))
+      );
+
+      final MonitorResponse monitorResponse = keyTransparencyServiceClient.monitor(
+          monitorKeys,
+          request.lastTreeHeadSize(),
+          KEY_TRANSPARENCY_RPC_TIMEOUT).join();
+
+      MonitorProof usernameHashMonitorProof = null;
+      MonitorProof e164MonitorProof = null;
+
+      // In the future we'll update KT's monitor response structure to enumerate each monitor key proof
+      // rather than returning everything in a list
+      if (monitorResponse.getContactProofsCount() == 3) {
+        e164MonitorProof = monitorResponse.getContactProofs(1);
+        usernameHashMonitorProof = monitorResponse.getContactProofs(2);
+      } else if (monitorResponse.getContactProofsCount() == 2) {
+        if (request.usernameHash().isPresent()) {
+          usernameHashMonitorProof = monitorResponse.getContactProofs(1);
+        } else if (request.e164().isPresent()) {
+          e164MonitorProof = monitorResponse.getContactProofs(1);
+        }
+      }
+      return new KeyTransparencyMonitorResponse(monitorResponse.getTreeHead(),
+          monitorResponse.getContactProofs(0),
+          Optional.ofNullable(e164MonitorProof),
+          Optional.ofNullable(usernameHashMonitorProof),
+          monitorResponse.getInclusionList().stream().map(ByteString::toByteArray).toList());
+    } catch (final CancellationException exception) {
+      LOGGER.error("Unexpected cancellation from key transparency service", exception);
+      throw new ServerErrorException(Response.Status.SERVICE_UNAVAILABLE, exception);
+    } catch (final CompletionException exception) {
+      handleKeyTransparencyServiceError(exception);
+    }
+    // This is unreachable
+    return null;
+  }
+
+  private void handleKeyTransparencyServiceError(final CompletionException exception) {
+    final Throwable unwrapped = ExceptionUtils.unwrap(exception);
+
+    if (unwrapped instanceof StatusRuntimeException e) {
+      final Status.Code code = e.getStatus().getCode();
+      final String description = e.getStatus().getDescription();
+      switch (code) {
+        case NOT_FOUND -> throw new NotFoundException(description);
+        case PERMISSION_DENIED -> throw new ForbiddenException(description);
+        case INVALID_ARGUMENT -> throw new WebApplicationException(description, 422);
+        default -> throw new ServerErrorException(Response.Status.INTERNAL_SERVER_ERROR, unwrapped);
+      }
+    }
+    LOGGER.error("Unexpected key transparency service failure", unwrapped);
+    throw new ServerErrorException(Response.Status.INTERNAL_SERVER_ERROR, unwrapped);
+  }
+
+  private static MonitorKey createMonitorKey(final ByteString fullSearchKey, final List<Long> positions) {
+    return MonitorKey.newBuilder()
+        .setSearchKey(fullSearchKey)
+        .addAllEntries(positions)
+        .build();
+  }
+
+  private void requireNotAuthenticated(final Optional<AuthenticatedAccount> authenticatedAccount) {
+    if (authenticatedAccount.isPresent()) {
+      throw new BadRequestException("Endpoint requires unauthenticated access");
+    }
+  }
+
+  @VisibleForTesting
+  static ByteString getFullSearchKeyByteString(final byte prefix, final byte[] searchKeyBytes) {
+    final ByteBuffer fullSearchKeyBuffer = ByteBuffer.allocate(searchKeyBytes.length + 1);
+    fullSearchKeyBuffer.put(prefix);
+    fullSearchKeyBuffer.put(searchKeyBytes);
+    fullSearchKeyBuffer.flip();
+
+    return ByteString.copyFrom(fullSearchKeyBuffer.array());
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/entities/KeyTransparencyMonitorRequest.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.swagger.v3.oas.annotations.media.Schema;
+import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier;
+import org.whispersystems.textsecuregcm.util.ByteArrayBase64UrlAdapter;
+import org.whispersystems.textsecuregcm.util.ServiceIdentifierAdapter;
+
+import javax.validation.constraints.AssertTrue;
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+import javax.validation.constraints.Positive;
+import java.util.List;
+import java.util.Optional;
+
+public record KeyTransparencyMonitorRequest(
+    @NotNull
+    @JsonSerialize(using = ServiceIdentifierAdapter.ServiceIdentifierSerializer.class)
+    @JsonDeserialize(using = ServiceIdentifierAdapter.AciServiceIdentifierDeserializer.class)
+    @Schema(description = "The aci identifier to monitor")
+    AciServiceIdentifier aci,
+
+    @NotEmpty
+    @Schema(description = "A list of log tree positions maintained by the client for the aci search key.")
+    List<@Positive Long> aciPositions,
+
+    @Schema(description = "The e164-formatted phone number to monitor")
+    Optional<String> e164,
+
+    @Schema(description = "A list of log tree positions maintained by the client for the e164 search key.")
+    Optional<List<@Positive Long>> e164Positions,
+
+    @JsonSerialize(contentUsing = ByteArrayBase64UrlAdapter.Serializing.class)
+    @JsonDeserialize(contentUsing = ByteArrayBase64UrlAdapter.Deserializing.class)
+    @Schema(description = "The username hash to monitor, encoded in url-safe unpadded base64.")
+    Optional<byte[]> usernameHash,
+
+    @Schema(description = "A list of log tree positions maintained by the client for the username hash search key.")
+    Optional<List<@Positive Long>> usernameHashPositions,
+
+    @Schema(description = "The tree head size to prove consistency against.")
+    Optional<@Positive Long> lastTreeHeadSize
+) {
+
+  @AssertTrue
+  public boolean isUsernameHashFieldsValid() {
+    return (usernameHash.isEmpty() && usernameHashPositions.isEmpty()) ||
+        (usernameHash.isPresent() && usernameHashPositions.isPresent() && !usernameHashPositions.get().isEmpty());
+  }
+
+  @AssertTrue
+  public boolean isE164VFieldsValid() {
+    return (e164.isEmpty() && e164Positions.isEmpty()) ||
+        (e164.isPresent() && e164Positions.isPresent() && !e164Positions.get().isEmpty());
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/entities/KeyTransparencyMonitorResponse.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.swagger.v3.oas.annotations.media.Schema;
+import katie.FullTreeHead;
+import katie.MonitorProof;
+import org.whispersystems.textsecuregcm.util.ByteArrayAdapter;
+import org.whispersystems.textsecuregcm.util.FullTreeHeadProtobufAdapter;
+import org.whispersystems.textsecuregcm.util.MonitorProofProtobufAdapter;
+
+import javax.validation.constraints.NotNull;
+import java.util.List;
+import java.util.Optional;
+
+public record KeyTransparencyMonitorResponse(
+    @NotNull
+    @JsonSerialize(using = FullTreeHeadProtobufAdapter.Serializer.class)
+    @JsonDeserialize(using = FullTreeHeadProtobufAdapter.Deserializer.class)
+    @Schema(description = """
+        The key transparency log's tree head along with a consistency proof and possibly an auditor-signed tree head
+        """)
+    FullTreeHead fullTreeHead,
+
+    @NotNull
+    @JsonSerialize(using = MonitorProofProtobufAdapter.Serializer.class)
+    @JsonDeserialize(using = MonitorProofProtobufAdapter.Deserializer.class)
+    @Schema(description = "The monitor proof for the aci search key")
+    MonitorProof aciMonitorProof,
+
+    @JsonSerialize(contentUsing = MonitorProofProtobufAdapter.Serializer.class)
+    @JsonDeserialize(contentUsing = MonitorProofProtobufAdapter.Deserializer.class)
+    @Schema(description = "The monitor proof for the e164 search key")
+    Optional<MonitorProof> e164MonitorProof,
+
+    @JsonSerialize(contentUsing = MonitorProofProtobufAdapter.Serializer.class)
+    @JsonDeserialize(contentUsing = MonitorProofProtobufAdapter.Deserializer.class)
+    @Schema(description = "The monitor proof for the username hash search key")
+    Optional<MonitorProof> usernameHashMonitorProof,
+
+    @NotNull
+    @JsonSerialize(contentUsing = ByteArrayAdapter.Serializing.class)
+    @JsonDeserialize(contentUsing = ByteArrayAdapter.Deserializing.class)
+    @Schema(description = "A list of hashes encoded in standard, unpadded base64 that prove inclusion across all monitor proofs ")
+    List<byte[]> inclusionProof
+) {}

service/src/main/java/org/whispersystems/textsecuregcm/entities/KeyTransparencySearchRequest.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.swagger.v3.oas.annotations.media.Schema;
+import org.whispersystems.textsecuregcm.identity.AciServiceIdentifier;
+import org.whispersystems.textsecuregcm.util.ByteArrayBase64UrlAdapter;
+import org.whispersystems.textsecuregcm.util.E164;
+import org.whispersystems.textsecuregcm.util.ServiceIdentifierAdapter;
+
+import javax.validation.constraints.NotNull;
+import javax.validation.constraints.Positive;
+import java.util.Optional;
+
+public record KeyTransparencySearchRequest(
+    @NotNull
+    @JsonSerialize(using = ServiceIdentifierAdapter.ServiceIdentifierSerializer.class)
+    @JsonDeserialize(using = ServiceIdentifierAdapter.AciServiceIdentifierDeserializer.class)
+    @Schema(description = "The aci identifier to look up")
+    AciServiceIdentifier aci,
+
+    @E164
+    @Schema(description = "The e164-formatted phone number to look up")
+    Optional<String> e164,
+
+    @JsonSerialize(contentUsing = ByteArrayBase64UrlAdapter.Serializing.class)
+    @JsonDeserialize(contentUsing = ByteArrayBase64UrlAdapter.Deserializing.class)
+    @Schema(description = "The username hash to look up, encoded in web-safe unpadded base64.")
+    Optional<byte[]> usernameHash,
+
+    @Schema(description = "The tree head size to prove consistency against.")
+    Optional<@Positive Long> lastTreeHeadSize
+) {}

service/src/main/java/org/whispersystems/textsecuregcm/entities/KeyTransparencySearchResponse.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.swagger.v3.oas.annotations.media.Schema;
+import katie.SearchResponse;
+import org.whispersystems.textsecuregcm.util.SearchResponseProtobufAdapter;
+
+import javax.validation.constraints.NotNull;
+import java.util.Optional;
+
+public record KeyTransparencySearchResponse(
+    @NotNull
+    @JsonSerialize(using = SearchResponseProtobufAdapter.Serializer.class)
+    @JsonDeserialize(using = SearchResponseProtobufAdapter.Deserializer.class)
+    @Schema(description = "The search response for the aci search key")
+    SearchResponse aciSearchResponse,
+
+    @JsonSerialize(contentUsing = SearchResponseProtobufAdapter.Serializer.class)
+    @JsonDeserialize(contentUsing = SearchResponseProtobufAdapter.Deserializer.class)
+    @Schema(description = "The search response for the e164 search key")
+    Optional<SearchResponse> e164SearchResponse,
+
+    @JsonSerialize(contentUsing = SearchResponseProtobufAdapter.Serializer.class)
+    @JsonDeserialize(contentUsing = SearchResponseProtobufAdapter.Deserializer.class)
+    @Schema(description = "The search response for the username hash search key")
+    Optional<SearchResponse> usernameHashSearchResponse
+) {}

service/src/main/java/org/whispersystems/textsecuregcm/keytransparency/KeyTransparencyServiceClient.java

@@ -0,0 +1,98 @@
+package org.whispersystems.textsecuregcm.keytransparency;
+
+import com.google.protobuf.ByteString;
+import io.dropwizard.lifecycle.Managed;
+import io.grpc.ChannelCredentials;
+import io.grpc.Deadline;
+import io.grpc.Grpc;
+import io.grpc.ManagedChannel;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.TlsChannelCredentials;
+import katie.KatieGrpc;
+import katie.MonitorKey;
+import katie.MonitorRequest;
+import katie.MonitorResponse;
+import katie.SearchRequest;
+import katie.SearchResponse;
+import org.whispersystems.textsecuregcm.util.CompletableFutureUtil;
+
+public class KeyTransparencyServiceClient implements Managed {
+
+  private final Executor callbackExecutor;
+  private final String host;
+  private final int port;
+  private final ChannelCredentials tlsChannelCredentials;
+  private ManagedChannel channel;
+  private KatieGrpc.KatieFutureStub stub;
+
+  public KeyTransparencyServiceClient(
+      final String host,
+      final int port,
+      final String tlsCertificate,
+      final Executor callbackExecutor
+  ) throws IOException {
+    this.host = host;
+    this.port = port;
+    try (final ByteArrayInputStream certificateInputStream = new ByteArrayInputStream(
+        tlsCertificate.getBytes(StandardCharsets.UTF_8))) {
+      tlsChannelCredentials = TlsChannelCredentials.newBuilder()
+          .trustManager(certificateInputStream)
+          .build();
+    }
+    this.callbackExecutor = callbackExecutor;
+  }
+
+  public CompletableFuture<SearchResponse> search(
+      final ByteString searchKey,
+      final Optional<Long> lastTreeHeadSize,
+      final Duration timeout) {
+    final SearchRequest.Builder searchRequestBuilder = SearchRequest.newBuilder()
+        .setSearchKey(searchKey);
+
+    lastTreeHeadSize.ifPresent(searchRequestBuilder::setLast);
+
+    return CompletableFutureUtil.toCompletableFuture(stub.withDeadline(toDeadline(timeout))
+        .search(searchRequestBuilder.build()), callbackExecutor);
+  }
+
+  public CompletableFuture<MonitorResponse> monitor(final List<MonitorKey> monitorKeys,
+      final Optional<Long> lastTreeHeadSize,
+      final Duration timeout) {
+    final MonitorRequest.Builder monitorRequestBuilder = MonitorRequest.newBuilder()
+        .addAllContactKeys(monitorKeys);
+
+    lastTreeHeadSize.ifPresent(monitorRequestBuilder::setLast);
+
+    return CompletableFutureUtil.toCompletableFuture(stub.withDeadline(toDeadline(timeout))
+        .monitor(monitorRequestBuilder.build()), callbackExecutor);
+  }
+
+  private static Deadline toDeadline(final Duration timeout) {
+    return Deadline.after(timeout.toMillis(), TimeUnit.MILLISECONDS);
+  }
+
+  @Override
+  public void start() throws Exception {
+    channel = Grpc.newChannelBuilderForAddress(host, port, tlsChannelCredentials)
+        .idleTimeout(1, TimeUnit.MINUTES)
+        .build();
+    stub = KatieGrpc.newFutureStub(channel);
+  }
+
+  @Override
+  public void stop() throws Exception {
+    if (channel != null) {
+      channel.shutdown();
+    }
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/limits/RateLimiters.java

@@ -48,6 +48,8 @@ public class RateLimiters extends BaseRateLimiters<RateLimiters.For> {
     CREATE_CALL_LINK("createCallLink", false, new RateLimiterConfig(100, Duration.ofMinutes(15))),
     INBOUND_MESSAGE_BYTES("inboundMessageBytes", true, new RateLimiterConfig(128 * 1024 * 1024, Duration.ofNanos(500_000))),
     EXTERNAL_SERVICE_CREDENTIALS("externalServiceCredentials", true, new RateLimiterConfig(100, Duration.ofMinutes(15))),
+    KEY_TRANSPARENCY_SEARCH_PER_IP("keyTransparencySearch", true, new RateLimiterConfig(100, Duration.ofSeconds(15))),
+    KEY_TRANSPARENCY_MONITOR_PER_IP("keyTransparencyMonitor", true, new RateLimiterConfig(100, Duration.ofSeconds(15))),
     ;
 
     private final String id;

service/src/main/java/org/whispersystems/textsecuregcm/registration/RegistrationServiceClient.java

@@ -1,8 +1,5 @@
 package org.whispersystems.textsecuregcm.registration;
 
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 import com.google.i18n.phonenumbers.NumberParseException;
 import com.google.i18n.phonenumbers.PhoneNumberUtil;
 import com.google.i18n.phonenumbers.Phonenumber;
@@ -34,6 +31,7 @@ import org.signal.registration.rpc.SendVerificationCodeRequest;
 import org.whispersystems.textsecuregcm.controllers.RateLimitExceededException;
 import org.whispersystems.textsecuregcm.controllers.VerificationSessionRateLimitExceededException;
 import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
+import org.whispersystems.textsecuregcm.util.CompletableFutureUtil;
 
 public class RegistrationServiceClient implements Managed {
 
@@ -84,11 +82,11 @@ public class RegistrationServiceClient implements Managed {
     final long e164 = Long.parseLong(
         PhoneNumberUtil.getInstance().format(phoneNumber, PhoneNumberUtil.PhoneNumberFormat.E164).substring(1));
 
-    return toCompletableFuture(stub.withDeadline(toDeadline(timeout))
+    return CompletableFutureUtil.toCompletableFuture(stub.withDeadline(toDeadline(timeout))
         .createSession(CreateRegistrationSessionRequest.newBuilder()
             .setE164(e164)
             .setAccountExistsWithE164(accountExistsWithPhoneNumber)
-            .build()))
+            .build()), callbackExecutor)
         .thenApply(response -> switch (response.getResponseCase()) {
           case SESSION_METADATA -> buildSessionResponseFromMetadata(response.getSessionMetadata());
 
@@ -129,8 +127,8 @@ public class RegistrationServiceClient implements Managed {
       requestBuilder.setSenderName(senderOverride);
     }
 
-    return toCompletableFuture(stub.withDeadline(toDeadline(timeout))
-        .sendVerificationCode(requestBuilder.build()))
+    return CompletableFutureUtil.toCompletableFuture(stub.withDeadline(toDeadline(timeout))
+        .sendVerificationCode(requestBuilder.build()), callbackExecutor)
         .thenApply(response -> {
           if (response.hasError()) {
             switch (response.getError().getErrorType()) {
@@ -172,11 +170,11 @@ public class RegistrationServiceClient implements Managed {
   public CompletableFuture<RegistrationServiceSession> checkVerificationCode(final byte[] sessionId,
       final String verificationCode,
       final Duration timeout) {
-    return toCompletableFuture(stub.withDeadline(toDeadline(timeout))
+    return CompletableFutureUtil.toCompletableFuture(stub.withDeadline(toDeadline(timeout))
         .checkVerificationCode(CheckVerificationCodeRequest.newBuilder()
             .setSessionId(ByteString.copyFrom(sessionId))
             .setVerificationCode(verificationCode)
-            .build()))
+            .build()), callbackExecutor)
         .thenApply(response -> {
           if (response.hasError()) {
             switch (response.getError().getErrorType()) {
@@ -208,9 +206,9 @@ public class RegistrationServiceClient implements Managed {
 
   public CompletableFuture<Optional<RegistrationServiceSession>> getSession(final byte[] sessionId,
       final Duration timeout) {
-    return toCompletableFuture(stub.withDeadline(toDeadline(timeout)).getSessionMetadata(
+    return CompletableFutureUtil.toCompletableFuture(stub.withDeadline(toDeadline(timeout)).getSessionMetadata(
         GetRegistrationSessionMetadataRequest.newBuilder()
-            .setSessionId(ByteString.copyFrom(sessionId)).build()))
+            .setSessionId(ByteString.copyFrom(sessionId)).build()), callbackExecutor)
         .thenApply(response -> {
           if (response.hasError()) {
             switch (response.getError().getErrorType()) {
@@ -251,24 +249,6 @@ public class RegistrationServiceClient implements Managed {
     };
   }
 
-  private <T> CompletableFuture<T> toCompletableFuture(final ListenableFuture<T> listenableFuture) {
-    final CompletableFuture<T> completableFuture = new CompletableFuture<>();
-
-    Futures.addCallback(listenableFuture, new FutureCallback<T>() {
-      @Override
-      public void onSuccess(@Nullable final T result) {
-        completableFuture.complete(result);
-      }
-
-      @Override
-      public void onFailure(final Throwable throwable) {
-        completableFuture.completeExceptionally(throwable);
-      }
-    }, callbackExecutor);
-
-    return completableFuture;
-  }
-
   @Override
   public void start() throws Exception {
   }

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+
+import javax.annotation.Nullable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+public class CompletableFutureUtil {
+
+  public static <T> CompletableFuture<T> toCompletableFuture(final ListenableFuture<T> listenableFuture,
+      final Executor callbackExecutor) {
+    final CompletableFuture<T> completableFuture = new CompletableFuture<>();
+
+    Futures.addCallback(listenableFuture, new FutureCallback<T>() {
+      @Override
+      public void onSuccess(@Nullable final T result) {
+        completableFuture.complete(result);
+      }
+
+      @Override
+      public void onFailure(final Throwable throwable) {
+        completableFuture.completeExceptionally(throwable);
+      }
+    }, callbackExecutor);
+
+    return completableFuture;
+  }
+}

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

@@ -14,6 +14,7 @@ import java.lang.annotation.Documented;
 import java.lang.annotation.Retention;
 import java.lang.annotation.Target;
 import java.util.Objects;
+import java.util.Optional;
 import javax.validation.Constraint;
 import javax.validation.ConstraintValidator;
 import javax.validation.ConstraintValidatorContext;
@@ -25,7 +26,10 @@ import javax.validation.Payload;
  */
 @Target({ FIELD, PARAMETER, METHOD })
 @Retention(RUNTIME)
-@Constraint(validatedBy = E164.Validator.class)
+@Constraint(validatedBy = {
+    E164.Validator.class,
+    E164.OptionalValidator.class
+})
 @Documented
 public @interface E164 {
 
@@ -53,4 +57,12 @@ public @interface E164 {
       return true;
     }
   }
+
+  class OptionalValidator implements ConstraintValidator<E164, Optional<String>> {
+
+    @Override
+    public boolean isValid(final Optional<String> value, final ConstraintValidatorContext context) {
+        return value.map(s -> new Validator().isValid(s, context)).orElse(true);
+    }
+  }
 }

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

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+import katie.FullTreeHead;
+
+public class FullTreeHeadProtobufAdapter {
+
+  public static class Serializer extends ProtobufAdapter.Serializer<FullTreeHead> {}
+
+  public static class Deserializer extends ProtobufAdapter.Deserializer<FullTreeHead> {
+
+    public Deserializer() {
+      super(FullTreeHead::newBuilder);
+    }
+  }
+}

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

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+import katie.MonitorProof;
+
+public class MonitorProofProtobufAdapter {
+
+  public static class Serializer extends ProtobufAdapter.Serializer<MonitorProof> {}
+
+  public static class Deserializer extends ProtobufAdapter.Deserializer<MonitorProof> {
+
+    public Deserializer() {
+      super(MonitorProof::newBuilder);
+    }
+  }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.google.protobuf.Message;
+import com.google.protobuf.util.JsonFormat;
+
+import java.io.IOException;
+import java.util.function.Supplier;
+
+public class ProtobufAdapter {
+
+  public static class Serializer<T extends Message> extends JsonSerializer<T> {
+
+    @Override
+    public void serialize(T message, JsonGenerator jsonGenerator, SerializerProvider serializerProvider)
+        throws IOException {
+      jsonGenerator.writeString(JsonFormat.printer().print(message));
+    }
+  }
+
+  public static class Deserializer<T extends Message> extends JsonDeserializer<T> {
+
+    private final Supplier<Message.Builder> builderSupplier;
+
+    public Deserializer(Supplier<Message.Builder> builderSupplier) {
+      this.builderSupplier = builderSupplier;
+    }
+
+    @Override
+    public T deserialize(JsonParser jsonParser, DeserializationContext deserializationContext) throws IOException {
+      Message.Builder builder = builderSupplier.get();
+      JsonFormat.parser().ignoringUnknownFields().merge(jsonParser.getValueAsString(), builder);
+      return (T) builder.build();
+    }
+  }
+}

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

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2024 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+import katie.SearchResponse;
+
+public class SearchResponseProtobufAdapter {
+
+  public static class Serializer extends ProtobufAdapter.Serializer<SearchResponse> {}
+
+  public static class Deserializer extends ProtobufAdapter.Deserializer<SearchResponse> {
+
+    public Deserializer() {
+      super(SearchResponse::newBuilder);
+    }
+  }
+}