Add `/v1/verification`

service/config/sample.yml

@@ -72,6 +72,9 @@ dynamoDbTables:
   redeemedReceipts:
     tableName: Example_RedeemedReceipts
     expiration: P30D # Duration of time until rows expire
+  registrationRecovery:
+    tableName: Example_RegistrationRecovery
+    expiration: P300D # Duration of time until rows expire
   remoteConfig:
     tableName: Example_RemoteConfig
   reportMessage:
@@ -80,9 +83,8 @@ dynamoDbTables:
     tableName: Example_ReservedUsernames
   subscriptions:
     tableName: Example_Subscriptions
-  registrationRecovery:
+  verificationSessions:
-    tableName: Example_RegistrationRecovery
+    tableName: Example_VerificationSessions
-    expiration: P300D # Duration of time until rows expire
 
 cacheCluster: # Redis server configuration for cache cluster
   configurationUri: redis://redis.example.com:6379/

service/pom.xml

@@ -567,6 +567,16 @@
         </executions>
       </plugin>
 
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <version>3.0.0-M7</version>
+        <configuration>
+          <!-- work around PATCH not being a supported method on HttpUrlConnection -->
+          <argLine>--add-opens=java.base/java.net=ALL-UNNAMED</argLine>
+        </configuration>
+      </plugin>
+
       <plugin>
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-jar-plugin</artifactId>

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

@@ -85,6 +85,7 @@ import org.whispersystems.textsecuregcm.badges.ResourceBundleLevelTranslator;
 import org.whispersystems.textsecuregcm.captcha.CaptchaChecker;
 import org.whispersystems.textsecuregcm.captcha.HCaptchaClient;
 import org.whispersystems.textsecuregcm.captcha.RecaptchaClient;
+import org.whispersystems.textsecuregcm.captcha.RegistrationCaptchaManager;
 import org.whispersystems.textsecuregcm.configuration.DirectoryServerConfiguration;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.controllers.AccountController;
@@ -111,6 +112,7 @@ import org.whispersystems.textsecuregcm.controllers.SecureStorageController;
 import org.whispersystems.textsecuregcm.controllers.SecureValueRecovery2Controller;
 import org.whispersystems.textsecuregcm.controllers.StickerController;
 import org.whispersystems.textsecuregcm.controllers.SubscriptionController;
+import org.whispersystems.textsecuregcm.controllers.VerificationController;
 import org.whispersystems.textsecuregcm.currency.CoinMarketCapClient;
 import org.whispersystems.textsecuregcm.currency.CurrencyConversionManager;
 import org.whispersystems.textsecuregcm.currency.FixerClient;
@@ -130,6 +132,7 @@ import org.whispersystems.textsecuregcm.mappers.InvalidWebsocketAddressException
 import org.whispersystems.textsecuregcm.mappers.JsonMappingExceptionMapper;
 import org.whispersystems.textsecuregcm.mappers.NonNormalizedPhoneNumberExceptionMapper;
 import org.whispersystems.textsecuregcm.mappers.RateLimitExceededExceptionMapper;
+import org.whispersystems.textsecuregcm.mappers.RegistrationServiceSenderExceptionMapper;
 import org.whispersystems.textsecuregcm.mappers.ServerRejectedExceptionMapper;
 import org.whispersystems.textsecuregcm.metrics.ApplicationShutdownMonitor;
 import org.whispersystems.textsecuregcm.metrics.BufferPoolGauges;
@@ -210,6 +213,8 @@ import org.whispersystems.textsecuregcm.storage.ReportMessageManager;
 import org.whispersystems.textsecuregcm.storage.StoredVerificationCodeManager;
 import org.whispersystems.textsecuregcm.storage.SubscriptionManager;
 import org.whispersystems.textsecuregcm.storage.VerificationCodeStore;
+import org.whispersystems.textsecuregcm.storage.VerificationSessionManager;
+import org.whispersystems.textsecuregcm.storage.VerificationSessions;
 import org.whispersystems.textsecuregcm.subscriptions.BraintreeManager;
 import org.whispersystems.textsecuregcm.subscriptions.StripeManager;
 import org.whispersystems.textsecuregcm.util.Constants;
@@ -382,6 +387,8 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         dynamoDbAsyncClient
     );
 
+    final VerificationSessions verificationSessions = new VerificationSessions(dynamoDbAsyncClient,
+        config.getDynamoDbTables().getVerificationSessions().getTableName(), clock);
     reactor.util.Metrics.MicrometerConfiguration.useRegistry(Metrics.globalRegistry);
     Schedulers.enableMetrics();
 
@@ -632,6 +639,9 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     environment.lifecycle().manage(directoryQueue);
     environment.lifecycle().manage(registrationServiceClient);
 
+    final RegistrationCaptchaManager registrationCaptchaManager = new RegistrationCaptchaManager(captchaChecker,
+        rateLimiters, config.getTestDevices(), dynamicConfigurationManager);
+
     StaticCredentialsProvider cdnCredentialsProvider = StaticCredentialsProvider
         .create(AwsBasicCredentials.create(
             config.getCdnConfiguration().getAccessKey(),
@@ -687,9 +697,9 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     // these should be common, but use @Auth DisabledPermittedAccount, which isn’t supported yet on websocket
     environment.jersey().register(
         new AccountController(pendingAccountsManager, accountsManager, rateLimiters,
-            registrationServiceClient, dynamicConfigurationManager, turnTokenGenerator, config.getTestDevices(),
+            registrationServiceClient, dynamicConfigurationManager, turnTokenGenerator,
-            captchaChecker, pushNotificationManager, changeNumberManager, registrationLockVerificationManager,
+            registrationCaptchaManager, pushNotificationManager, changeNumberManager,
-            registrationRecoveryPasswordsManager, usernameHashZkProofVerifier, clock));
+            registrationLockVerificationManager, registrationRecoveryPasswordsManager, usernameHashZkProofVerifier, clock));
 
     environment.jersey().register(new KeysController(rateLimiters, keys, accountsManager));
 
@@ -769,7 +779,10 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new SecureValueRecovery2Controller(svr2CredentialsGenerator),
         new StickerController(rateLimiters, config.getCdnConfiguration().getAccessKey(),
             config.getCdnConfiguration().getAccessSecret(), config.getCdnConfiguration().getRegion(),
-            config.getCdnConfiguration().getBucket())
+            config.getCdnConfiguration().getBucket()),
+        new VerificationController(registrationServiceClient, new VerificationSessionManager(verificationSessions),
+            pushNotificationManager, registrationCaptchaManager, registrationRecoveryPasswordsManager, rateLimiters,
+            clock)
     );
     if (config.getSubscription() != null && config.getOneTimeDonations() != null) {
       commonControllers.add(new SubscriptionController(clock, config.getSubscription(), config.getOneTimeDonations(),
@@ -846,6 +859,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new ServerRejectedExceptionMapper(),
         new ImpossiblePhoneNumberExceptionMapper(),
         new NonNormalizedPhoneNumberExceptionMapper(),
+        new RegistrationServiceSenderExceptionMapper(),
         new JsonMappingExceptionMapper()
     ).forEach(exceptionMapper -> {
       environment.jersey().register(exceptionMapper);

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

@@ -5,6 +5,8 @@
 
 package org.whispersystems.textsecuregcm.auth;
 
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
 import java.security.MessageDigest;
 import java.time.Duration;
 import java.util.concurrent.CancellationException;
@@ -19,7 +21,7 @@ import javax.ws.rs.core.Response;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.whispersystems.textsecuregcm.entities.PhoneVerificationRequest;
-import org.whispersystems.textsecuregcm.entities.RegistrationSession;
+import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
 import org.whispersystems.textsecuregcm.registration.RegistrationServiceClient;
 import org.whispersystems.textsecuregcm.storage.RegistrationRecoveryPasswordsManager;
 
@@ -46,7 +48,8 @@ public class PhoneVerificationTokenManager {
    * @param request the request with exactly one verification token (RegistrationService sessionId or registration
    *                recovery password)
    * @return if verification was successful, returns the verification type
-   * @throws BadRequestException    if the number does not match the sessionId’s number
+   * @throws BadRequestException    if the number does not match the sessionId’s number, or the remote service rejects
+   *                                the session ID as invalid
    * @throws NotAuthorizedException if the session is not verified
    * @throws ForbiddenException     if the recovery password is not valid
    * @throws InterruptedException   if verification did not complete before a timeout
@@ -65,7 +68,7 @@ public class PhoneVerificationTokenManager {
 
   private void verifyBySessionId(final String number, final byte[] sessionId) throws InterruptedException {
     try {
-      final RegistrationSession session = registrationServiceClient
+      final RegistrationServiceSession session = registrationServiceClient
           .getSession(sessionId, REGISTRATION_RPC_TIMEOUT)
           .get(VERIFICATION_TIMEOUT_SECONDS, TimeUnit.SECONDS)
           .orElseThrow(() -> new NotAuthorizedException("session not verified"));
@@ -76,7 +79,19 @@ public class PhoneVerificationTokenManager {
       if (!session.verified()) {
         throw new NotAuthorizedException("session not verified");
       }
-    } catch (final CancellationException | ExecutionException | TimeoutException e) {
+    } catch (final ExecutionException e) {
+
+      if (e.getCause() instanceof StatusRuntimeException grpcRuntimeException) {
+        if (grpcRuntimeException.getStatus().getCode() == Status.Code.INVALID_ARGUMENT) {
+          throw new BadRequestException();
+        }
+      }
+
+      logger.error("Registration service failure", e);
+      throw new ServerErrorException(Response.Status.SERVICE_UNAVAILABLE);
+
+    } catch (final CancellationException | TimeoutException e) {
+
       logger.error("Registration service failure", e);
       throw new ServerErrorException(Response.Status.SERVICE_UNAVAILABLE);
     }

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

@@ -10,9 +10,9 @@ import java.time.Duration;
 import javax.annotation.Nullable;
 import org.whispersystems.textsecuregcm.util.Util;
 
-public record StoredVerificationCode(String code,
+public record StoredVerificationCode(@Nullable String code,
                                      long timestamp,
-                                     String pushCode,
+                                     @Nullable String pushCode,
                                      @Nullable byte[] sessionId) {
 
   public static final Duration EXPIRATION = Duration.ofMinutes(10);

service/src/main/java/org/whispersystems/textsecuregcm/captcha/RegistrationCaptchaManager.java

@@ -0,0 +1,106 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.captcha;
+
+import static com.codahale.metrics.MetricRegistry.name;
+
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.SharedMetricRegistries;
+import java.io.IOException;
+import java.util.Map;
+import java.util.Optional;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicCaptchaConfiguration;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.controllers.AccountController;
+import org.whispersystems.textsecuregcm.controllers.RateLimitExceededException;
+import org.whispersystems.textsecuregcm.limits.RateLimiters;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
+import org.whispersystems.textsecuregcm.util.Constants;
+import org.whispersystems.textsecuregcm.util.Util;
+
+public class RegistrationCaptchaManager {
+
+  private static final Logger logger = LoggerFactory.getLogger(RegistrationCaptchaManager.class);
+
+  private final MetricRegistry metricRegistry = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
+  private final Meter countryFilteredHostMeter = metricRegistry.meter(
+      name(AccountController.class, "country_limited_host"));
+  private final Meter rateLimitedHostMeter = metricRegistry.meter(name(AccountController.class, "rate_limited_host"));
+  private final Meter rateLimitedPrefixMeter = metricRegistry.meter(
+      name(AccountController.class, "rate_limited_prefix"));
+
+  private final CaptchaChecker captchaChecker;
+  private final RateLimiters rateLimiters;
+  private final Map<String, Integer> testDevices;
+  private final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
+
+
+  public RegistrationCaptchaManager(final CaptchaChecker captchaChecker, final RateLimiters rateLimiters,
+      final Map<String, Integer> testDevices,
+      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager) {
+    this.captchaChecker = captchaChecker;
+    this.rateLimiters = rateLimiters;
+    this.testDevices = testDevices;
+    this.dynamicConfigurationManager = dynamicConfigurationManager;
+  }
+
+  @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+  public Optional<AssessmentResult> assessCaptcha(final Optional<String> captcha, final String sourceHost)
+      throws IOException {
+    return captcha.isPresent()
+        ? Optional.of(captchaChecker.verify(captcha.get(), sourceHost))
+        : Optional.empty();
+  }
+
+  public boolean requiresCaptcha(final String number, final String forwardedFor, String sourceHost,
+      final boolean pushChallengeMatch) {
+    if (testDevices.containsKey(number)) {
+      return false;
+    }
+
+    if (!pushChallengeMatch) {
+      return true;
+    }
+
+    final String countryCode = Util.getCountryCode(number);
+    final String region = Util.getRegion(number);
+
+    DynamicCaptchaConfiguration captchaConfig = dynamicConfigurationManager.getConfiguration()
+        .getCaptchaConfiguration();
+
+    boolean countryFiltered = captchaConfig.getSignupCountryCodes().contains(countryCode) ||
+        captchaConfig.getSignupRegions().contains(region);
+
+    try {
+      rateLimiters.getSmsVoiceIpLimiter().validate(sourceHost);
+    } catch (RateLimitExceededException e) {
+      logger.info("Rate limit exceeded: {}, {} ({})", number, sourceHost, forwardedFor);
+      rateLimitedHostMeter.mark();
+
+      return true;
+    }
+
+    try {
+      rateLimiters.getSmsVoicePrefixLimiter().validate(Util.getNumberPrefix(number));
+    } catch (RateLimitExceededException e) {
+      logger.info("Prefix rate limit exceeded: {}, {} ({})", number, sourceHost, forwardedFor);
+      rateLimitedPrefixMeter.mark();
+
+      return true;
+    }
+
+    if (countryFiltered) {
+      countryFilteredHostMeter.mark();
+      return true;
+    }
+
+    return false;
+  }
+
+}

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

@@ -58,11 +58,12 @@ public class DynamoDbTables {
   private final Table profiles;
   private final Table pushChallenge;
   private final TableWithExpiration redeemedReceipts;
+  private final TableWithExpiration registrationRecovery;
   private final Table remoteConfig;
   private final Table reportMessage;
   private final Table reservedUsernames;
   private final Table subscriptions;
-  private final TableWithExpiration registrationRecovery;
+  private final Table verificationSessions;
 
   public DynamoDbTables(
       @JsonProperty("accounts") final AccountsTableConfiguration accounts,
@@ -77,11 +78,12 @@ public class DynamoDbTables {
       @JsonProperty("profiles") final Table profiles,
       @JsonProperty("pushChallenge") final Table pushChallenge,
       @JsonProperty("redeemedReceipts") final TableWithExpiration redeemedReceipts,
+      @JsonProperty("registrationRecovery") final TableWithExpiration registrationRecovery,
       @JsonProperty("remoteConfig") final Table remoteConfig,
       @JsonProperty("reportMessage") final Table reportMessage,
       @JsonProperty("reservedUsernames") final Table reservedUsernames,
       @JsonProperty("subscriptions") final Table subscriptions,
-      @JsonProperty("registrationRecovery") final TableWithExpiration registrationRecovery) {
+      @JsonProperty("verificationSessions") final Table verificationSessions) {
 
     this.accounts = accounts;
     this.deletedAccounts = deletedAccounts;
@@ -95,11 +97,12 @@ public class DynamoDbTables {
     this.profiles = profiles;
     this.pushChallenge = pushChallenge;
     this.redeemedReceipts = redeemedReceipts;
+    this.registrationRecovery = registrationRecovery;
     this.remoteConfig = remoteConfig;
     this.reportMessage = reportMessage;
     this.reservedUsernames = reservedUsernames;
     this.subscriptions = subscriptions;
-    this.registrationRecovery = registrationRecovery;
+    this.verificationSessions = verificationSessions;
   }
 
   @NotNull
@@ -174,6 +177,12 @@ public class DynamoDbTables {
     return redeemedReceipts;
   }
 
+  @NotNull
+  @Valid
+  public TableWithExpiration getRegistrationRecovery() {
+    return registrationRecovery;
+  }
+
   @NotNull
   @Valid
   public Table getRemoteConfig() {
@@ -200,7 +209,7 @@ public class DynamoDbTables {
 
   @NotNull
   @Valid
-  public TableWithExpiration getRegistrationRecovery() {
+  public Table getVerificationSessions() {
-    return registrationRecovery;
+    return verificationSessions;
   }
 }

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

@@ -29,6 +29,12 @@ public class RateLimitsConfiguration {
   @JsonProperty
   private RateLimitConfiguration verifyPin = new RateLimitConfiguration(10, 1 / (24.0 * 60.0));
 
+  @JsonProperty
+  private RateLimitConfiguration verificationCaptcha = new RateLimitConfiguration(10, 2);
+
+  @JsonProperty
+  private RateLimitConfiguration verificationPushChallenge = new RateLimitConfiguration(5, 2);
+
   @JsonProperty
   private RateLimitConfiguration registration = new RateLimitConfiguration(2, 2);
 
@@ -122,6 +128,14 @@ public class RateLimitsConfiguration {
     return verifyPin;
   }
 
+  public RateLimitConfiguration getVerificationCaptcha() {
+    return verificationCaptcha;
+  }
+
+  public RateLimitConfiguration getVerificationPushChallenge() {
+    return verificationPushChallenge;
+  }
+
   public RateLimitConfiguration getRegistration() {
     return registration;
   }

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

@@ -28,7 +28,6 @@ import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.HexFormat;
-import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletionException;
@@ -67,8 +66,7 @@ import org.whispersystems.textsecuregcm.auth.StoredVerificationCode;
 import org.whispersystems.textsecuregcm.auth.TurnToken;
 import org.whispersystems.textsecuregcm.auth.TurnTokenGenerator;
 import org.whispersystems.textsecuregcm.captcha.AssessmentResult;
-import org.whispersystems.textsecuregcm.captcha.CaptchaChecker;
+import org.whispersystems.textsecuregcm.captcha.RegistrationCaptchaManager;
-import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicCaptchaConfiguration;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
 import org.whispersystems.textsecuregcm.entities.AccountIdentifierResponse;
@@ -118,9 +116,6 @@ public class AccountController {
   public static final int USERNAME_HASH_LENGTH = 32;
   private final Logger         logger                   = LoggerFactory.getLogger(AccountController.class);
   private final MetricRegistry metricRegistry           = SharedMetricRegistries.getOrCreate(Constants.METRICS_NAME);
-  private final Meter          countryFilteredHostMeter = metricRegistry.meter(name(AccountController.class, "country_limited_host"     ));
-  private final Meter          rateLimitedHostMeter     = metricRegistry.meter(name(AccountController.class, "rate_limited_host"        ));
-  private final Meter          rateLimitedPrefixMeter   = metricRegistry.meter(name(AccountController.class, "rate_limited_prefix"      ));
   private final Meter          captchaRequiredMeter     = metricRegistry.meter(name(AccountController.class, "captcha_required"         ));
 
   private static final String PUSH_CHALLENGE_COUNTER_NAME = name(AccountController.class, "pushChallenge");
@@ -155,8 +150,7 @@ public class AccountController {
   private final RegistrationServiceClient registrationServiceClient;
   private final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
   private final TurnTokenGenerator turnTokenGenerator;
-  private final Map<String, Integer> testDevices;
+  private final RegistrationCaptchaManager registrationCaptchaManager;
-  private final CaptchaChecker captchaChecker;
   private final PushNotificationManager pushNotificationManager;
   private final RegistrationLockVerificationManager registrationLockVerificationManager;
   private final RegistrationRecoveryPasswordsManager registrationRecoveryPasswordsManager;
@@ -175,8 +169,7 @@ public class AccountController {
       RegistrationServiceClient registrationServiceClient,
       DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager,
       TurnTokenGenerator turnTokenGenerator,
-      Map<String, Integer> testDevices,
+      RegistrationCaptchaManager registrationCaptchaManager,
-      CaptchaChecker captchaChecker,
       PushNotificationManager pushNotificationManager,
       ChangeNumberManager changeNumberManager,
       RegistrationLockVerificationManager registrationLockVerificationManager,
@@ -189,9 +182,8 @@ public class AccountController {
     this.rateLimiters = rateLimiters;
     this.registrationServiceClient = registrationServiceClient;
     this.dynamicConfigurationManager = dynamicConfigurationManager;
-    this.testDevices = testDevices;
     this.turnTokenGenerator = turnTokenGenerator;
-    this.captchaChecker = captchaChecker;
+    this.registrationCaptchaManager = registrationCaptchaManager;
     this.pushNotificationManager = pushNotificationManager;
     this.registrationLockVerificationManager = registrationLockVerificationManager;
     this.changeNumberManager = changeNumberManager;
@@ -245,6 +237,7 @@ public class AccountController {
       } else {
         final byte[] sessionId = createRegistrationSession(phoneNumber);
         storedVerificationCode = new StoredVerificationCode(null, clock.millis(), generatePushChallenge(), sessionId);
+        new StoredVerificationCode(null, clock.millis(), generatePushChallenge(), sessionId);
       }
     }
 
@@ -278,9 +271,7 @@ public class AccountController {
     final String region = Util.getRegion(number);
 
     // if there's a captcha, assess it, otherwise check if we need a captcha
-    final Optional<AssessmentResult> assessmentResult = captcha.isPresent()
+    final Optional<AssessmentResult> assessmentResult = registrationCaptchaManager.assessCaptcha(captcha, sourceHost);
-        ? Optional.of(captchaChecker.verify(captcha.get(), sourceHost))
-        : Optional.empty();
 
     assessmentResult.ifPresent(result ->
         Metrics.counter(CAPTCHA_ATTEMPT_COUNTER_NAME, Tags.of(
@@ -300,7 +291,8 @@ public class AccountController {
 
     final boolean requiresCaptcha = assessmentResult
         .map(result -> !result.valid())
-        .orElseGet(() -> requiresCaptcha(number, transport, forwardedFor, sourceHost, pushChallengeMatch));
+        .orElseGet(
+            () -> registrationCaptchaManager.requiresCaptcha(number, forwardedFor, sourceHost, pushChallengeMatch));
 
     if (requiresCaptcha) {
       captchaRequiredMeter.mark();
@@ -357,8 +349,7 @@ public class AccountController {
 
     final StoredVerificationCode storedVerificationCode = new StoredVerificationCode(null,
         clock.millis(),
-        maybeStoredVerificationCode.map(StoredVerificationCode::pushCode).orElse(null),
+        maybeStoredVerificationCode.map(StoredVerificationCode::pushCode).orElse(null), sessionId);
-        sessionId);
 
     pendingAccounts.store(number, storedVerificationCode);
 
@@ -844,50 +835,6 @@ public class AccountController {
     return match;
   }
 
-  private boolean requiresCaptcha(String number, String transport, String forwardedFor, String sourceHost, boolean pushChallengeMatch) {
-    if (testDevices.containsKey(number)) {
-      return false;
-    }
-
-    if (!pushChallengeMatch) {
-      return true;
-    }
-
-    final String countryCode = Util.getCountryCode(number);
-    final String region = Util.getRegion(number);
-
-    DynamicCaptchaConfiguration captchaConfig = dynamicConfigurationManager.getConfiguration()
-        .getCaptchaConfiguration();
-
-    boolean countryFiltered = captchaConfig.getSignupCountryCodes().contains(countryCode) ||
-        captchaConfig.getSignupRegions().contains(region);
-
-    try {
-      rateLimiters.getSmsVoiceIpLimiter().validate(sourceHost);
-    } catch (RateLimitExceededException e) {
-      logger.info("Rate limit exceeded: {}, {}, {} ({})", transport, number, sourceHost, forwardedFor);
-      rateLimitedHostMeter.mark();
-
-      return true;
-    }
-
-    try {
-      rateLimiters.getSmsVoicePrefixLimiter().validate(Util.getNumberPrefix(number));
-    } catch (RateLimitExceededException e) {
-      logger.info("Prefix rate limit exceeded: {}, {}, {} ({})", transport, number, sourceHost, forwardedFor);
-      rateLimitedPrefixMeter.mark();
-
-      return true;
-    }
-
-    if (countryFiltered) {
-      countryFilteredHostMeter.mark();
-      return true;
-    }
-
-    return false;
-  }
-
   @Timed
   @DELETE
   @Path("/me")

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

@@ -0,0 +1,676 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.controllers;
+
+import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
+
+import com.codahale.metrics.annotation.Timed;
+import com.google.i18n.phonenumbers.NumberParseException;
+import com.google.i18n.phonenumbers.PhoneNumberUtil;
+import com.google.i18n.phonenumbers.Phonenumber;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.SecureRandom;
+import java.time.Clock;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.TimeUnit;
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+import javax.ws.rs.BadRequestException;
+import javax.ws.rs.ClientErrorException;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.ForbiddenException;
+import javax.ws.rs.GET;
+import javax.ws.rs.HeaderParam;
+import javax.ws.rs.NotFoundException;
+import javax.ws.rs.PATCH;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.ServerErrorException;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.HttpHeaders;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.http.HttpStatus;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.captcha.AssessmentResult;
+import org.whispersystems.textsecuregcm.captcha.RegistrationCaptchaManager;
+import org.whispersystems.textsecuregcm.entities.CreateVerificationSessionRequest;
+import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
+import org.whispersystems.textsecuregcm.entities.SubmitVerificationCodeRequest;
+import org.whispersystems.textsecuregcm.entities.UpdateVerificationSessionRequest;
+import org.whispersystems.textsecuregcm.entities.VerificationCodeRequest;
+import org.whispersystems.textsecuregcm.entities.VerificationSessionResponse;
+import org.whispersystems.textsecuregcm.limits.RateLimiter;
+import org.whispersystems.textsecuregcm.limits.RateLimiters;
+import org.whispersystems.textsecuregcm.metrics.UserAgentTagUtil;
+import org.whispersystems.textsecuregcm.push.PushNotification;
+import org.whispersystems.textsecuregcm.push.PushNotificationManager;
+import org.whispersystems.textsecuregcm.registration.ClientType;
+import org.whispersystems.textsecuregcm.registration.MessageTransport;
+import org.whispersystems.textsecuregcm.registration.RegistrationServiceClient;
+import org.whispersystems.textsecuregcm.registration.RegistrationServiceException;
+import org.whispersystems.textsecuregcm.registration.RegistrationServiceSenderException;
+import org.whispersystems.textsecuregcm.registration.VerificationSession;
+import org.whispersystems.textsecuregcm.spam.FilterSpam;
+import org.whispersystems.textsecuregcm.storage.RegistrationRecoveryPasswordsManager;
+import org.whispersystems.textsecuregcm.storage.VerificationSessionManager;
+import org.whispersystems.textsecuregcm.util.ExceptionUtils;
+import org.whispersystems.textsecuregcm.util.HeaderUtils;
+import org.whispersystems.textsecuregcm.util.Pair;
+import org.whispersystems.textsecuregcm.util.Util;
+
+@Path("/v1/verification")
+public class VerificationController {
+
+  private static final Logger logger = LoggerFactory.getLogger(VerificationController.class);
+
+  private static final Duration REGISTRATION_RPC_TIMEOUT = Duration.ofSeconds(15);
+  private static final Duration DYNAMODB_TIMEOUT = Duration.ofSeconds(5);
+
+  private static final SecureRandom RANDOM = new SecureRandom();
+
+  private static final String PUSH_CHALLENGE_COUNTER_NAME = name(VerificationController.class, "pushChallenge");
+  private static final String CHALLENGE_PRESENT_TAG_NAME = "present";
+  private static final String CHALLENGE_MATCH_TAG_NAME = "matches";
+  private static final String CAPTCHA_ATTEMPT_COUNTER_NAME = name(VerificationController.class, "captcha");
+  private static final String COUNTRY_CODE_TAG_NAME = "countryCode";
+  private static final String REGION_CODE_TAG_NAME = "regionCode";
+  private static final String SCORE_TAG_NAME = "score";
+  private static final String CODE_REQUESTED_COUNTER_NAME = name(VerificationController.class, "codeRequested");
+  private static final String VERIFICATION_TRANSPORT_TAG_NAME = "transport";
+  private static final String VERIFIED_COUNTER_NAME = name(VerificationController.class, "verified");
+  private static final String SUCCESS_TAG_NAME = "success";
+
+  private final RegistrationServiceClient registrationServiceClient;
+  private final VerificationSessionManager verificationSessionManager;
+  private final PushNotificationManager pushNotificationManager;
+  private final RegistrationCaptchaManager registrationCaptchaManager;
+  private final RegistrationRecoveryPasswordsManager registrationRecoveryPasswordsManager;
+  private final RateLimiters rateLimiters;
+
+  private final Clock clock;
+
+  public VerificationController(final RegistrationServiceClient registrationServiceClient,
+      final VerificationSessionManager verificationSessionManager,
+      final PushNotificationManager pushNotificationManager,
+      final RegistrationCaptchaManager registrationCaptchaManager,
+      final RegistrationRecoveryPasswordsManager registrationRecoveryPasswordsManager, final RateLimiters rateLimiters,
+      final Clock clock) {
+    this.registrationServiceClient = registrationServiceClient;
+    this.verificationSessionManager = verificationSessionManager;
+    this.pushNotificationManager = pushNotificationManager;
+    this.registrationCaptchaManager = registrationCaptchaManager;
+    this.registrationRecoveryPasswordsManager = registrationRecoveryPasswordsManager;
+    this.rateLimiters = rateLimiters;
+    this.clock = clock;
+  }
+
+  @Timed
+  @POST
+  @Path("/session")
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Produces(MediaType.APPLICATION_JSON)
+  public VerificationSessionResponse createSession(@NotNull @Valid CreateVerificationSessionRequest request)
+      throws RateLimitExceededException {
+
+    final Pair<String, PushNotification.TokenType> pushTokenAndType = validateAndExtractPushToken(
+        request.getUpdateVerificationSessionRequest());
+
+    final Phonenumber.PhoneNumber phoneNumber;
+    try {
+      phoneNumber = PhoneNumberUtil.getInstance().parse(request.getNumber(), null);
+    } catch (final NumberParseException e) {
+      throw new ServerErrorException("could not parse already validated number", Response.Status.INTERNAL_SERVER_ERROR);
+    }
+
+    final RegistrationServiceSession registrationServiceSession;
+    try {
+      registrationServiceSession = registrationServiceClient.createRegistrationSessionSession(phoneNumber,
+          REGISTRATION_RPC_TIMEOUT).join();
+    } catch (final CancellationException e) {
+
+      throw new ServerErrorException("registration service unavailable", Response.Status.SERVICE_UNAVAILABLE);
+    } catch (final CompletionException e) {
+
+      if (ExceptionUtils.unwrap(e) instanceof RateLimitExceededException re) {
+        RateLimiter.adaptLegacyException(() -> {
+          throw re;
+        });
+      }
+
+      throw new ServerErrorException(Response.Status.INTERNAL_SERVER_ERROR, e);
+    }
+
+    VerificationSession verificationSession = new VerificationSession(null, new ArrayList<>(),
+        Collections.emptyList(), false,
+        clock.millis(), clock.millis(), registrationServiceSession.expiration());
+
+    verificationSession = handlePushToken(pushTokenAndType, verificationSession);
+    // unconditionally request a captcha -- it will either be the only requested information, or a fallback
+    // if a push challenge sent in `handlePushToken` doesn't arrive in time
+    verificationSession.requestedInformation().add(VerificationSession.Information.CAPTCHA);
+
+    storeVerificationSession(registrationServiceSession, verificationSession);
+
+    return buildResponse(registrationServiceSession, verificationSession);
+  }
+
+  @Timed
+  @FilterSpam
+  @PATCH
+  @Path("/session/{sessionId}")
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Produces(MediaType.APPLICATION_JSON)
+  public VerificationSessionResponse updateSession(@PathParam("sessionId") final String encodedSessionId,
+      @HeaderParam(com.google.common.net.HttpHeaders.X_FORWARDED_FOR) String forwardedFor,
+      @HeaderParam(HttpHeaders.USER_AGENT) final String userAgent,
+      @NotNull @Valid final UpdateVerificationSessionRequest updateVerificationSessionRequest) {
+
+    final String sourceHost = HeaderUtils.getMostRecentProxy(forwardedFor).orElseThrow();
+
+    final Pair<String, PushNotification.TokenType> pushTokenAndType = validateAndExtractPushToken(
+        updateVerificationSessionRequest);
+
+    final RegistrationServiceSession registrationServiceSession = retrieveRegistrationServiceSession(encodedSessionId);
+    VerificationSession verificationSession = retrieveVerificationSession(registrationServiceSession);
+
+    try {
+      // these handle* methods ordered from least likely to fail to most, so take care when considering a change
+      verificationSession = handlePushToken(pushTokenAndType, verificationSession);
+
+      verificationSession = handlePushChallenge(updateVerificationSessionRequest, registrationServiceSession,
+          verificationSession);
+
+      verificationSession = handleCaptcha(sourceHost, updateVerificationSessionRequest, registrationServiceSession,
+          verificationSession, userAgent);
+    } catch (final RateLimitExceededException e) {
+
+      final Response response = buildResponseForRateLimitExceeded(verificationSession, registrationServiceSession,
+          e.getRetryDuration());
+      throw new ClientErrorException(response);
+
+    } catch (final ForbiddenException e) {
+
+      throw new ClientErrorException(Response.status(Response.Status.FORBIDDEN)
+          .entity(buildResponse(registrationServiceSession, verificationSession))
+          .build());
+
+    } finally {
+      // Each of the handle* methods may update requestedInformation, submittedInformation, and allowedToRequestCode,
+      // and we want to be sure to store a changes, even if a later method throws
+      updateStoredVerificationSession(registrationServiceSession, verificationSession);
+    }
+
+    return buildResponse(registrationServiceSession, verificationSession);
+  }
+
+  private void storeVerificationSession(final RegistrationServiceSession registrationServiceSession,
+      final VerificationSession verificationSession) {
+    verificationSessionManager.insert(registrationServiceSession.encodedSessionId(), verificationSession)
+        .orTimeout(DYNAMODB_TIMEOUT.toSeconds(), TimeUnit.SECONDS)
+        .join();
+  }
+
+  private void updateStoredVerificationSession(final RegistrationServiceSession registrationServiceSession,
+      final VerificationSession verificationSession) {
+    verificationSessionManager.update(registrationServiceSession.encodedSessionId(), verificationSession)
+        .orTimeout(DYNAMODB_TIMEOUT.toSeconds(), TimeUnit.SECONDS)
+        .join();
+  }
+
+  /**
+   * If {@code pushTokenAndType} values are not {@code null}, sends a push challenge. If there is no existing push
+   * challenge in the session, one will be created, set on the returned session record, and
+   * {@link VerificationSession#requestedInformation()} will be updated.
+   */
+  private VerificationSession handlePushToken(
+      final Pair<String, PushNotification.TokenType> pushTokenAndType, VerificationSession verificationSession) {
+
+    if (pushTokenAndType.first() != null) {
+
+      if (verificationSession.pushChallenge() == null) {
+
+        final List<VerificationSession.Information> requestedInformation = new ArrayList<>();
+        requestedInformation.add(VerificationSession.Information.PUSH_CHALLENGE);
+        requestedInformation.addAll(verificationSession.requestedInformation());
+
+        verificationSession = new VerificationSession(generatePushChallenge(), requestedInformation,
+            verificationSession.submittedInformation(), verificationSession.allowedToRequestCode(),
+            verificationSession.createdTimestamp(), clock.millis(), verificationSession.remoteExpirationSeconds()
+        );
+      }
+
+      pushNotificationManager.sendRegistrationChallengeNotification(pushTokenAndType.first(), pushTokenAndType.second(),
+          verificationSession.pushChallenge());
+    }
+
+    return verificationSession;
+  }
+
+  /**
+   * If a push challenge value is present, compares against the stored value. If they match, then
+   * {@link VerificationSession.Information#PUSH_CHALLENGE} is removed from requested information, added to submitted
+   * information, and {@link VerificationSession#allowedToRequestCode()} is re-evaluated.
+   *
+   * @throws ForbiddenException         if values to not match.
+   * @throws RateLimitExceededException if too many push challenges have been submitted
+   */
+  private VerificationSession handlePushChallenge(
+      final UpdateVerificationSessionRequest updateVerificationSessionRequest,
+      final RegistrationServiceSession registrationServiceSession,
+      VerificationSession verificationSession) throws RateLimitExceededException {
+
+    if (verificationSession.submittedInformation()
+        .contains(VerificationSession.Information.PUSH_CHALLENGE)) {
+      // skip if a challenge has already been submitted
+      return verificationSession;
+    }
+
+    final boolean pushChallengePresent = updateVerificationSessionRequest.pushChallenge() != null;
+    if (pushChallengePresent) {
+      RateLimiter.adaptLegacyException(
+          () -> rateLimiters.getVerificationPushChallengeLimiter()
+              .validate(registrationServiceSession.encodedSessionId()));
+    }
+
+    final boolean pushChallengeMatches;
+    if (pushChallengePresent && verificationSession.pushChallenge() != null) {
+      pushChallengeMatches = MessageDigest.isEqual(
+          updateVerificationSessionRequest.pushChallenge().getBytes(StandardCharsets.UTF_8),
+          verificationSession.pushChallenge().getBytes(StandardCharsets.UTF_8));
+    } else {
+      pushChallengeMatches = false;
+    }
+
+    Metrics.counter(PUSH_CHALLENGE_COUNTER_NAME,
+            COUNTRY_CODE_TAG_NAME, Util.getCountryCode(registrationServiceSession.number()),
+            REGION_CODE_TAG_NAME, Util.getRegion(registrationServiceSession.number()),
+            CHALLENGE_PRESENT_TAG_NAME, Boolean.toString(pushChallengePresent),
+            CHALLENGE_MATCH_TAG_NAME, Boolean.toString(pushChallengeMatches))
+        .increment();
+
+    if (pushChallengeMatches) {
+      final List<VerificationSession.Information> submittedInformation = new ArrayList<>(
+          verificationSession.submittedInformation());
+      submittedInformation.add(VerificationSession.Information.PUSH_CHALLENGE);
+
+      final List<VerificationSession.Information> requestedInformation = new ArrayList<>(
+          verificationSession.requestedInformation());
+      // a push challenge satisfies a requested captcha
+      requestedInformation.remove(VerificationSession.Information.CAPTCHA);
+      final boolean allowedToRequestCode = (verificationSession.allowedToRequestCode()
+          || requestedInformation.remove(VerificationSession.Information.PUSH_CHALLENGE))
+          && requestedInformation.isEmpty();
+
+      verificationSession = new VerificationSession(verificationSession.pushChallenge(), requestedInformation,
+          submittedInformation, allowedToRequestCode, verificationSession.createdTimestamp(), clock.millis(),
+          verificationSession.remoteExpirationSeconds());
+
+    } else if (pushChallengePresent) {
+      throw new ForbiddenException();
+    }
+    return verificationSession;
+  }
+
+  /**
+   * If a captcha value is present, it is assessed. If it is valid, then {@link VerificationSession.Information#CAPTCHA}
+   * is removed from requested information, added to submitted information, and
+   * {@link VerificationSession#allowedToRequestCode()} is re-evaluated.
+   *
+   * @throws ForbiddenException         if assessment is not valid.
+   * @throws RateLimitExceededException if too many captchas have been submitted
+   */
+  private VerificationSession handleCaptcha(final String sourceHost,
+      final UpdateVerificationSessionRequest updateVerificationSessionRequest,
+      final RegistrationServiceSession registrationServiceSession,
+      VerificationSession verificationSession,
+      final String userAgent) throws RateLimitExceededException {
+
+    if (updateVerificationSessionRequest.captcha() == null) {
+      return verificationSession;
+    }
+
+    RateLimiter.adaptLegacyException(
+        () -> rateLimiters.getVerificationCaptchaLimiter().validate(registrationServiceSession.encodedSessionId()));
+
+    final AssessmentResult assessmentResult;
+    try {
+      assessmentResult = registrationCaptchaManager.assessCaptcha(
+              Optional.of(updateVerificationSessionRequest.captcha()), sourceHost)
+          .orElseThrow(() -> new ServerErrorException(Response.Status.INTERNAL_SERVER_ERROR));
+
+      Metrics.counter(CAPTCHA_ATTEMPT_COUNTER_NAME, Tags.of(
+              Tag.of(SUCCESS_TAG_NAME, String.valueOf(assessmentResult.valid())),
+              UserAgentTagUtil.getPlatformTag(userAgent),
+              Tag.of(COUNTRY_CODE_TAG_NAME, Util.getCountryCode(registrationServiceSession.number())),
+              Tag.of(REGION_CODE_TAG_NAME, Util.getRegion(registrationServiceSession.number())),
+              Tag.of(SCORE_TAG_NAME, assessmentResult.score())))
+          .increment();
+
+    } catch (IOException e) {
+      throw new ServerErrorException(Response.Status.SERVICE_UNAVAILABLE);
+    }
+
+    if (assessmentResult.valid()) {
+      final List<VerificationSession.Information> submittedInformation = new ArrayList<>(
+          verificationSession.submittedInformation());
+      submittedInformation.add(VerificationSession.Information.CAPTCHA);
+
+      final List<VerificationSession.Information> requestedInformation = new ArrayList<>(
+          verificationSession.requestedInformation());
+      // a captcha satisfies a push challenge, in case of push deliverability issues
+      requestedInformation.remove(VerificationSession.Information.PUSH_CHALLENGE);
+      final boolean allowedToRequestCode = (verificationSession.allowedToRequestCode()
+          || requestedInformation.remove(VerificationSession.Information.CAPTCHA))
+          && requestedInformation.isEmpty();
+
+      verificationSession = new VerificationSession(verificationSession.pushChallenge(), requestedInformation,
+          submittedInformation, allowedToRequestCode, verificationSession.createdTimestamp(), clock.millis(),
+          verificationSession.remoteExpirationSeconds());
+    } else {
+      throw new ForbiddenException();
+    }
+
+    return verificationSession;
+  }
+
+  @Timed
+  @GET
+  @Path("/session/{sessionId}")
+  @Produces(MediaType.APPLICATION_JSON)
+  public VerificationSessionResponse getSession(@PathParam("sessionId") final String encodedSessionId) {
+
+    final RegistrationServiceSession registrationServiceSession = retrieveRegistrationServiceSession(encodedSessionId);
+    final VerificationSession verificationSession = retrieveVerificationSession(registrationServiceSession);
+
+    return buildResponse(registrationServiceSession, verificationSession);
+  }
+
+  @Timed
+  @POST
+  @Path("/session/{sessionId}/code")
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Produces(MediaType.APPLICATION_JSON)
+  public VerificationSessionResponse requestVerificationCode(@PathParam("sessionId") final String encodedSessionId,
+      @HeaderParam(HttpHeaders.USER_AGENT) final String userAgent,
+      @HeaderParam(HttpHeaders.ACCEPT_LANGUAGE) Optional<String> acceptLanguage,
+      @NotNull @Valid VerificationCodeRequest verificationCodeRequest) throws Throwable {
+
+    final RegistrationServiceSession registrationServiceSession = retrieveRegistrationServiceSession(encodedSessionId);
+    final VerificationSession verificationSession = retrieveVerificationSession(registrationServiceSession);
+
+    if (registrationServiceSession.verified()) {
+      throw new ClientErrorException(
+          Response.status(Response.Status.CONFLICT)
+              .entity(buildResponse(registrationServiceSession, verificationSession))
+              .build());
+    }
+
+    if (!verificationSession.allowedToRequestCode()) {
+      final Response.Status status = verificationSession.requestedInformation().isEmpty()
+          ? Response.Status.TOO_MANY_REQUESTS
+          : Response.Status.CONFLICT;
+
+      throw new ClientErrorException(
+          Response.status(status)
+              .entity(buildResponse(registrationServiceSession, verificationSession))
+              .build());
+    }
+
+    final MessageTransport messageTransport = verificationCodeRequest.transport().toMessageTransport();
+
+    final ClientType clientType = switch (verificationCodeRequest.client()) {
+      case "ios" -> ClientType.IOS;
+      case "android-2021-03" -> ClientType.ANDROID_WITH_FCM;
+      default -> {
+        if (StringUtils.startsWithIgnoreCase(verificationCodeRequest.client(), "android")) {
+          yield ClientType.ANDROID_WITHOUT_FCM;
+        }
+        yield ClientType.UNKNOWN;
+      }
+    };
+
+    final RegistrationServiceSession resultSession;
+    try {
+      resultSession = registrationServiceClient.sendVerificationCode(registrationServiceSession.id(),
+          messageTransport,
+          clientType,
+          acceptLanguage.orElse(null), REGISTRATION_RPC_TIMEOUT).join();
+    } catch (final CancellationException e) {
+      throw new ServerErrorException("registration service unavailable", Response.Status.SERVICE_UNAVAILABLE);
+    } catch (final CompletionException e) {
+      final Throwable unwrappedException = ExceptionUtils.unwrap(e);
+      if (unwrappedException instanceof RateLimitExceededException rateLimitExceededException) {
+        if (rateLimitExceededException instanceof VerificationSessionRateLimitExceededException ve) {
+          final Response response = buildResponseForRateLimitExceeded(verificationSession, ve.getRegistrationSession(),
+              ve.getRetryDuration());
+          throw new ClientErrorException(response);
+        }
+
+        throw new RateLimitExceededException(rateLimitExceededException.getRetryDuration().orElse(null), false);
+      } else if (unwrappedException instanceof RegistrationServiceException registrationServiceException) {
+
+        throw registrationServiceException.getRegistrationSession()
+            .map(s -> buildResponse(s, verificationSession))
+            .map(verificationSessionResponse -> new ClientErrorException(
+                Response.status(Response.Status.CONFLICT).entity(verificationSessionResponse).build()))
+            .orElseGet(NotFoundException::new);
+
+      } else if (unwrappedException instanceof RegistrationServiceSenderException) {
+
+        throw unwrappedException;
+
+      } else {
+        logger.error("Registration service failure", unwrappedException);
+        throw new ServerErrorException(Response.Status.INTERNAL_SERVER_ERROR);
+      }
+    }
+
+    Metrics.counter(CODE_REQUESTED_COUNTER_NAME, Tags.of(
+            UserAgentTagUtil.getPlatformTag(userAgent),
+            Tag.of(COUNTRY_CODE_TAG_NAME, Util.getCountryCode(registrationServiceSession.number())),
+            Tag.of(REGION_CODE_TAG_NAME, Util.getRegion(registrationServiceSession.number())),
+            Tag.of(VERIFICATION_TRANSPORT_TAG_NAME, verificationCodeRequest.transport().toString())))
+        .increment();
+
+    return buildResponse(resultSession, verificationSession);
+  }
+
+  @Timed
+  @PUT
+  @Path("/session/{sessionId}/code")
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Produces(MediaType.APPLICATION_JSON)
+  public VerificationSessionResponse verifyCode(@PathParam("sessionId") final String encodedSessionId,
+      @HeaderParam(HttpHeaders.USER_AGENT) String userAgent,
+      @NotNull @Valid final SubmitVerificationCodeRequest submitVerificationCodeRequest)
+      throws RateLimitExceededException {
+
+    final RegistrationServiceSession registrationServiceSession = retrieveRegistrationServiceSession(encodedSessionId);
+    final VerificationSession verificationSession = retrieveVerificationSession(registrationServiceSession);
+
+    if (registrationServiceSession.verified()) {
+      final VerificationSessionResponse verificationSessionResponse = buildResponse(registrationServiceSession,
+          verificationSession);
+
+      throw new ClientErrorException(
+          Response.status(Response.Status.CONFLICT).entity(verificationSessionResponse).build());
+    }
+
+    final RegistrationServiceSession resultSession;
+    try {
+      resultSession = registrationServiceClient.checkVerificationCodeSession(registrationServiceSession.id(),
+              submitVerificationCodeRequest.code(),
+              REGISTRATION_RPC_TIMEOUT)
+          .join();
+    } catch (final CancellationException e) {
+      logger.warn("Unexpected cancellation from registration service", e);
+      throw new ServerErrorException(Response.Status.SERVICE_UNAVAILABLE);
+    } catch (final CompletionException e) {
+      final Throwable unwrappedException = ExceptionUtils.unwrap(e);
+      if (unwrappedException instanceof RateLimitExceededException rateLimitExceededException) {
+
+        if (rateLimitExceededException instanceof VerificationSessionRateLimitExceededException ve) {
+          final Response response = buildResponseForRateLimitExceeded(verificationSession, ve.getRegistrationSession(),
+              ve.getRetryDuration());
+          throw new ClientErrorException(response);
+        }
+
+        throw new RateLimitExceededException(rateLimitExceededException.getRetryDuration().orElse(null), false);
+
+      } else if (unwrappedException instanceof RegistrationServiceException registrationServiceException) {
+
+        throw registrationServiceException.getRegistrationSession()
+            .map(s -> buildResponse(s, verificationSession))
+            .map(verificationSessionResponse -> new ClientErrorException(
+                Response.status(Response.Status.CONFLICT).entity(verificationSessionResponse).build()))
+            .orElseGet(NotFoundException::new);
+
+      } else {
+        logger.error("Registration service failure", unwrappedException);
+        throw new ServerErrorException(Response.Status.INTERNAL_SERVER_ERROR);
+      }
+    }
+
+    if (resultSession.verified()) {
+      registrationRecoveryPasswordsManager.removeForNumber(registrationServiceSession.number());
+    }
+
+    Metrics.counter(VERIFIED_COUNTER_NAME, Tags.of(
+            UserAgentTagUtil.getPlatformTag(userAgent),
+            Tag.of(COUNTRY_CODE_TAG_NAME, Util.getCountryCode(registrationServiceSession.number())),
+            Tag.of(REGION_CODE_TAG_NAME, Util.getRegion(registrationServiceSession.number())),
+            Tag.of(SUCCESS_TAG_NAME, Boolean.toString(resultSession.verified()))))
+        .increment();
+
+    return buildResponse(resultSession, verificationSession);
+  }
+
+  private Response buildResponseForRateLimitExceeded(final VerificationSession verificationSession,
+      final RegistrationServiceSession registrationServiceSession,
+      final Optional<Duration> retryDuration) {
+
+    final Response.ResponseBuilder responseBuilder = Response.status(Response.Status.TOO_MANY_REQUESTS)
+        .entity(buildResponse(registrationServiceSession, verificationSession));
+
+    retryDuration
+        .filter(d -> !d.isNegative())
+        .ifPresent(d -> responseBuilder.header(HttpHeaders.RETRY_AFTER, d.toSeconds()));
+
+    return responseBuilder.build();
+  }
+
+  /**
+   * @throws ClientErrorException          with {@code 422} status if the ID cannot be decoded
+   * @throws javax.ws.rs.NotFoundException if the ID cannot be found
+   */
+  private RegistrationServiceSession retrieveRegistrationServiceSession(final String encodedSessionId) {
+    final byte[] sessionId;
+
+    try {
+      sessionId = decodeSessionId(encodedSessionId);
+    } catch (final IllegalArgumentException e) {
+      throw new ClientErrorException("Malformed session ID", HttpStatus.SC_UNPROCESSABLE_ENTITY);
+    }
+
+    try {
+      final RegistrationServiceSession registrationServiceSession = registrationServiceClient.getSession(sessionId,
+              REGISTRATION_RPC_TIMEOUT).join()
+          .orElseThrow(NotFoundException::new);
+
+      if (registrationServiceSession.verified()) {
+        registrationRecoveryPasswordsManager.removeForNumber(registrationServiceSession.number());
+      }
+
+      return registrationServiceSession;
+
+    } catch (final CompletionException | CancellationException e) {
+      final Throwable unwrapped = ExceptionUtils.unwrap(e);
+
+      if (unwrapped.getCause() instanceof StatusRuntimeException grpcRuntimeException) {
+        if (grpcRuntimeException.getStatus().getCode() == Status.Code.INVALID_ARGUMENT) {
+          throw new BadRequestException();
+        }
+      }
+      logger.error("Registration service failure", e);
+      throw new ServerErrorException(Response.Status.SERVICE_UNAVAILABLE, e);
+    }
+  }
+
+  /**
+   * @throws NotFoundException if the session is has no record
+   */
+  private VerificationSession retrieveVerificationSession(final RegistrationServiceSession registrationServiceSession) {
+
+    return verificationSessionManager.findForId(registrationServiceSession.encodedSessionId())
+        .orTimeout(5, TimeUnit.SECONDS)
+        .join().orElseThrow(NotFoundException::new);
+  }
+
+  /**
+   * @throws ClientErrorException with {@code 422} status if the only one of token and type are present
+   */
+  private Pair<String, PushNotification.TokenType> validateAndExtractPushToken(
+      final UpdateVerificationSessionRequest request) {
+
+    final String pushToken;
+    final PushNotification.TokenType pushTokenType;
+    if (Objects.isNull(request.pushToken())
+        != Objects.isNull(request.pushTokenType())) {
+      throw new WebApplicationException("must specify both pushToken and pushTokenType or neither",
+          HttpStatus.SC_UNPROCESSABLE_ENTITY);
+    } else {
+      pushToken = request.pushToken();
+      pushTokenType = pushToken == null
+          ? null
+          : request.pushTokenType().toTokenType();
+    }
+
+    return new Pair<>(pushToken, pushTokenType);
+  }
+
+  private VerificationSessionResponse buildResponse(final RegistrationServiceSession registrationServiceSession,
+      final VerificationSession verificationSession) {
+    return new VerificationSessionResponse(registrationServiceSession.encodedSessionId(),
+        registrationServiceSession.nextSms(),
+        registrationServiceSession.nextVoiceCall(), registrationServiceSession.nextVerificationAttempt(),
+        verificationSession.allowedToRequestCode(), verificationSession.requestedInformation(),
+        registrationServiceSession.verified());
+  }
+
+  public static byte[] decodeSessionId(final String sessionId) {
+    return Base64.getUrlDecoder().decode(sessionId);
+  }
+
+  private static String generatePushChallenge() {
+    final byte[] challenge = new byte[16];
+    RANDOM.nextBytes(challenge);
+
+    return HexFormat.of().formatHex(challenge);
+  }
+
+}

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.controllers;
+
+import org.jetbrains.annotations.Nullable;
+import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
+import java.time.Duration;
+
+public class VerificationSessionRateLimitExceededException extends RateLimitExceededException {
+
+  private final RegistrationServiceSession registrationServiceSession;
+
+  /**
+   * Constructs a new exception indicating when it may become safe to retry
+   *
+   * @param registrationServiceSession the associated registration session
+   * @param retryDuration              A duration to wait before retrying, null if no duration can be indicated
+   * @param legacy                     whether to use a legacy status code when mapping the exception to an HTTP
+   *                                   response
+   */
+  public VerificationSessionRateLimitExceededException(
+      final RegistrationServiceSession registrationServiceSession, @Nullable final Duration retryDuration,
+      final boolean legacy) {
+    super(retryDuration, legacy);
+    this.registrationServiceSession = registrationServiceSession;
+  }
+
+  public RegistrationServiceSession getRegistrationSession() {
+    return registrationServiceSession;
+  }
+}

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

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonUnwrapped;
+import javax.validation.Valid;
+import javax.validation.constraints.NotBlank;
+import org.whispersystems.textsecuregcm.util.E164;
+
+// Not a record, because Jackson does not support @JsonUnwrapped with records
+// https://github.com/FasterXML/jackson-databind/issues/1497
+public final class CreateVerificationSessionRequest {
+
+  @E164
+  @NotBlank
+  @JsonProperty
+  private String number;
+
+  @Valid
+  @JsonUnwrapped
+  private UpdateVerificationSessionRequest updateVerificationSessionRequest;
+
+  public String getNumber() {
+    return number;
+  }
+
+  public UpdateVerificationSessionRequest getUpdateVerificationSessionRequest() {
+    return updateVerificationSessionRequest;
+  }
+
+}

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import java.util.Base64;
+import javax.annotation.Nullable;
+import org.signal.registration.rpc.RegistrationSessionMetadata;
+
+public record RegistrationServiceSession(byte[] id, String number, boolean verified,
+                                         @Nullable Long nextSms, @Nullable Long nextVoiceCall,
+                                         @Nullable Long nextVerificationAttempt,
+                                         long expiration) {
+
+
+  public String encodedSessionId() {
+    return encodeSessionId(id);
+  }
+
+  public static String encodeSessionId(final byte[] sessionId) {
+    return Base64.getUrlEncoder().encodeToString(sessionId);
+  }
+
+  public RegistrationServiceSession(byte[] id, String number, RegistrationSessionMetadata remoteSession) {
+    this(id, number, remoteSession.getVerified(),
+        remoteSession.getMayRequestSms() ? remoteSession.getNextSmsSeconds() : null,
+        remoteSession.getMayRequestVoiceCall() ? remoteSession.getNextVoiceCallSeconds() : null,
+        remoteSession.getMayCheckCode() ? remoteSession.getNextCodeCheckSeconds() : null,
+        remoteSession.getExpirationSeconds());
+  }
+
+}

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

@@ -5,6 +5,8 @@
 
 package org.whispersystems.textsecuregcm.entities;
 
-public record RegistrationSession(String number, boolean verified) {
+import javax.validation.constraints.NotBlank;
+
+public record SubmitVerificationCodeRequest(@NotBlank String code) {
 
 }

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import javax.annotation.Nullable;
+import org.whispersystems.textsecuregcm.push.PushNotification;
+
+public record UpdateVerificationSessionRequest(@Nullable String pushToken,
+                                               @Nullable PushTokenType pushTokenType,
+                                               @Nullable String pushChallenge,
+                                               @Nullable String captcha,
+                                               @Nullable String mcc,
+                                               @Nullable String mnc) {
+
+  public enum PushTokenType {
+    @JsonProperty("apn")
+    APN,
+    @JsonProperty("fcm")
+    FCM;
+
+    public PushNotification.TokenType toTokenType() {
+      return switch (this) {
+
+        case APN -> PushNotification.TokenType.APN;
+        case FCM -> PushNotification.TokenType.FCM;
+      };
+    }
+  }
+
+}

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import javax.validation.constraints.NotNull;
+import org.whispersystems.textsecuregcm.registration.MessageTransport;
+
+public record VerificationCodeRequest(@NotNull Transport transport, @NotNull String client) {
+
+  public enum Transport {
+    @JsonProperty("sms")
+    SMS,
+    @JsonProperty("voice")
+    VOICE;
+
+    public MessageTransport toMessageTransport() {
+      return switch (this) {
+        case SMS -> MessageTransport.SMS;
+        case VOICE -> MessageTransport.VOICE;
+      };
+    }
+  }
+
+}

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

@@ -0,0 +1,17 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.entities;
+
+import java.util.List;
+import javax.annotation.Nullable;
+import org.whispersystems.textsecuregcm.registration.VerificationSession;
+
+public record VerificationSessionResponse(String id, @Nullable Long nextSms, @Nullable Long nextCall,
+                                          @Nullable Long nextVerificationAttempt, boolean allowedToRequestCode,
+                                          List<VerificationSession.Information> requestedInformation,
+                                          boolean verified) {
+
+}

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

@@ -150,4 +150,5 @@ public class RateLimiter {
 
     void validate() throws RateLimitExceededException;
   }
+
 }

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

@@ -42,6 +42,8 @@ public class RateLimiters {
   private final RateLimiter smsVoiceIpLimiter;
   private final RateLimiter smsVoicePrefixLimiter;
   private final RateLimiter verifyLimiter;
+  private final RateLimiter verificationCaptchaLimiter;
+  private final RateLimiter verificationPushChallengeLimiter;
   private final RateLimiter pinLimiter;
   private final RateLimiter registrationLimiter;
   private final RateLimiter attachmentLimiter;
@@ -61,10 +63,14 @@ public class RateLimiters {
   public RateLimiters(final RateLimitsConfiguration config, final FaultTolerantRedisCluster cacheCluster) {
     this.smsDestinationLimiter = fromConfig("smsDestination", config.getSmsDestination(), cacheCluster);
     this.voiceDestinationLimiter = fromConfig("voxDestination", config.getVoiceDestination(), cacheCluster);
-    this.voiceDestinationDailyLimiter = fromConfig("voxDestinationDaily", config.getVoiceDestinationDaily(), cacheCluster);
+    this.voiceDestinationDailyLimiter = fromConfig("voxDestinationDaily", config.getVoiceDestinationDaily(),
+        cacheCluster);
     this.smsVoiceIpLimiter = fromConfig("smsVoiceIp", config.getSmsVoiceIp(), cacheCluster);
     this.smsVoicePrefixLimiter = fromConfig("smsVoicePrefix", config.getSmsVoicePrefix(), cacheCluster);
     this.verifyLimiter = fromConfig("verify", config.getVerifyNumber(), cacheCluster);
+    this.verificationCaptchaLimiter = fromConfig("verificationCaptcha", config.getVerificationCaptcha(), cacheCluster);
+    this.verificationPushChallengeLimiter = fromConfig("verificationPushChallenge",
+        config.getVerificationPushChallenge(), cacheCluster);
     this.pinLimiter = fromConfig("pin", config.getVerifyPin(), cacheCluster);
     this.registrationLimiter = fromConfig("registration", config.getRegistration(), cacheCluster);
     this.attachmentLimiter = fromConfig("attachmentCreate", config.getAttachments(), cacheCluster);
@@ -134,6 +140,14 @@ public class RateLimiters {
     return verifyLimiter;
   }
 
+  public RateLimiter getVerificationCaptchaLimiter() {
+    return verificationCaptchaLimiter;
+  }
+
+  public RateLimiter getVerificationPushChallengeLimiter() {
+    return verificationPushChallengeLimiter;
+  }
+
   public RateLimiter getPinLimiter() {
     return pinLimiter;
   }

service/src/main/java/org/whispersystems/textsecuregcm/mappers/RegistrationServiceSenderExceptionMapper.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.mappers;
+
+import com.google.common.annotations.VisibleForTesting;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.ext.ExceptionMapper;
+import org.whispersystems.textsecuregcm.registration.RegistrationServiceSenderException;
+
+public class RegistrationServiceSenderExceptionMapper implements ExceptionMapper<RegistrationServiceSenderException> {
+
+  @Override
+  public Response toResponse(final RegistrationServiceSenderException exception) {
+    return Response.status(Response.Status.BAD_GATEWAY)
+        .entity(new SendVerificationCodeFailureResponse(exception.getReason(), exception.isPermanent()))
+        .build();
+  }
+
+  @VisibleForTesting
+  public record SendVerificationCodeFailureResponse(RegistrationServiceSenderException.Reason reason,
+                                                    boolean permanentFailure) {
+
+  }
+}

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

@@ -28,9 +28,11 @@ import org.signal.registration.rpc.CheckVerificationCodeRequest;
 import org.signal.registration.rpc.CreateRegistrationSessionRequest;
 import org.signal.registration.rpc.GetRegistrationSessionMetadataRequest;
 import org.signal.registration.rpc.RegistrationServiceGrpc;
+import org.signal.registration.rpc.RegistrationSessionMetadata;
 import org.signal.registration.rpc.SendVerificationCodeRequest;
 import org.whispersystems.textsecuregcm.controllers.RateLimitExceededException;
-import org.whispersystems.textsecuregcm.entities.RegistrationSession;
+import org.whispersystems.textsecuregcm.controllers.VerificationSessionRateLimitExceededException;
+import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
 
 public class RegistrationServiceClient implements Managed {
 
@@ -76,7 +78,10 @@ public class RegistrationServiceClient implements Managed {
     this.callbackExecutor = callbackExecutor;
   }
 
-  public CompletableFuture<byte[]> createRegistrationSession(final Phonenumber.PhoneNumber phoneNumber, final Duration timeout) {
+  // The …Session suffix methods distinguish the new methods, which return Sessions, from the old.
+  // Once the deprecated methods are removed, the names can be streamlined.
+  public CompletableFuture<RegistrationServiceSession> createRegistrationSessionSession(
+      final Phonenumber.PhoneNumber phoneNumber, final Duration timeout) {
     final long e164 = Long.parseLong(
         PhoneNumberUtil.getInstance().format(phoneNumber, PhoneNumberUtil.PhoneNumberFormat.E164).substring(1));
 
@@ -85,12 +90,14 @@ public class RegistrationServiceClient implements Managed {
             .setE164(e164)
             .build()))
         .thenApply(response -> switch (response.getResponseCase()) {
-          case SESSION_METADATA -> response.getSessionMetadata().getSessionId().toByteArray();
+          case SESSION_METADATA -> buildSessionResponseFromMetadata(response.getSessionMetadata());
 
           case ERROR -> {
             switch (response.getError().getErrorType()) {
               case CREATE_REGISTRATION_SESSION_ERROR_TYPE_RATE_LIMITED -> throw new CompletionException(
-                  new RateLimitExceededException(Duration.ofSeconds(response.getError().getRetryAfterSeconds()),
+                  new RateLimitExceededException(response.getError().getMayRetry()
+                      ? Duration.ofSeconds(response.getError().getRetryAfterSeconds())
+                      : null,
                       true));
               case CREATE_REGISTRATION_SESSION_ERROR_TYPE_ILLEGAL_PHONE_NUMBER -> throw new IllegalArgumentException();
               default -> throw new RuntimeException(
@@ -102,7 +109,14 @@ public class RegistrationServiceClient implements Managed {
         });
   }
 
-  public CompletableFuture<byte[]> sendRegistrationCode(final byte[] sessionId,
+  @Deprecated
+  public CompletableFuture<byte[]> createRegistrationSession(final Phonenumber.PhoneNumber phoneNumber,
+      final Duration timeout) {
+    return createRegistrationSessionSession(phoneNumber, timeout)
+        .thenApply(RegistrationServiceSession::id);
+  }
+
+  public CompletableFuture<RegistrationServiceSession> sendVerificationCode(final byte[] sessionId,
       final MessageTransport messageTransport,
       final ClientType clientType,
       @Nullable final String acceptLanguage,
@@ -123,21 +137,57 @@ public class RegistrationServiceClient implements Managed {
           if (response.hasError()) {
             switch (response.getError().getErrorType()) {
               case SEND_VERIFICATION_CODE_ERROR_TYPE_RATE_LIMITED -> throw new CompletionException(
-                  new RateLimitExceededException(Duration.ofSeconds(response.getError().getRetryAfterSeconds()),
+                  new VerificationSessionRateLimitExceededException(
+                      buildSessionResponseFromMetadata(response.getSessionMetadata()),
+                      response.getError().getMayRetry()
+                          ? Duration.ofSeconds(response.getError().getRetryAfterSeconds())
+                          : null,
                       true));
 
-              default -> throw new CompletionException(new RuntimeException("Failed to send verification code: " + response.getError().getErrorType()));
+              case SEND_VERIFICATION_CODE_ERROR_TYPE_SESSION_NOT_FOUND -> throw new CompletionException(
+                  new RegistrationServiceException(null));
+
+              case SEND_VERIFICATION_CODE_ERROR_TYPE_SESSION_ALREADY_VERIFIED -> throw new CompletionException(
+                  new RegistrationServiceException(buildSessionResponseFromMetadata(response.getSessionMetadata())));
+
+              case SEND_VERIFICATION_CODE_ERROR_TYPE_SENDER_REJECTED -> throw new CompletionException(
+                  RegistrationServiceSenderException.rejected(response.getError().getMayRetry()));
+              case SEND_VERIFICATION_CODE_ERROR_TYPE_SENDER_ILLEGAL_ARGUMENT -> throw new CompletionException(
+                  RegistrationServiceSenderException.illegalArgument(response.getError().getMayRetry()));
+              case SEND_VERIFICATION_CODE_ERROR_TYPE_UNSPECIFIED -> throw new CompletionException(
+                  RegistrationServiceSenderException.unknown(response.getError().getMayRetry()));
+
+              default -> throw new CompletionException(
+                  new RuntimeException("Failed to send verification code: " + response.getError().getErrorType()));
             }
           } else {
-            return response.getSessionId().toByteArray();
+            return buildSessionResponseFromMetadata(response.getSessionMetadata());
           }
         });
   }
 
+  @Deprecated
+  public CompletableFuture<byte[]> sendRegistrationCode(final byte[] sessionId,
+      final MessageTransport messageTransport,
+      final ClientType clientType,
+      @Nullable final String acceptLanguage,
+      final Duration timeout) {
+    return sendVerificationCode(sessionId, messageTransport, clientType, acceptLanguage, timeout)
+        .thenApply(RegistrationServiceSession::id);
+  }
+
+  @Deprecated
   public CompletableFuture<Boolean> checkVerificationCode(final byte[] sessionId,
       final String verificationCode,
       final Duration timeout) {
 
+    return checkVerificationCodeSession(sessionId, verificationCode, timeout)
+        .thenApply(RegistrationServiceSession::verified);
+  }
+
+  public CompletableFuture<RegistrationServiceSession> checkVerificationCodeSession(final byte[] sessionId,
+      final String verificationCode,
+      final Duration timeout) {
     return toCompletableFuture(stub.withDeadline(toDeadline(timeout))
         .checkVerificationCode(CheckVerificationCodeRequest.newBuilder()
             .setSessionId(ByteString.copyFrom(sessionId))
@@ -147,18 +197,32 @@ public class RegistrationServiceClient implements Managed {
           if (response.hasError()) {
             switch (response.getError().getErrorType()) {
               case CHECK_VERIFICATION_CODE_ERROR_TYPE_RATE_LIMITED -> throw new CompletionException(
-                  new RateLimitExceededException(Duration.ofSeconds(response.getError().getRetryAfterSeconds()),
+                  new VerificationSessionRateLimitExceededException(
+                      buildSessionResponseFromMetadata(response.getSessionMetadata()),
+                      response.getError().getMayRetry()
+                          ? Duration.ofSeconds(response.getError().getRetryAfterSeconds())
+                          : null,
                       true));
 
-              default -> throw new CompletionException(new RuntimeException("Failed to check verification code: " + response.getError().getErrorType()));
+              case CHECK_VERIFICATION_CODE_ERROR_TYPE_NO_CODE_SENT, CHECK_VERIFICATION_CODE_ERROR_TYPE_ATTEMPT_EXPIRED ->
+                  throw new CompletionException(
+                      new RegistrationServiceException(buildSessionResponseFromMetadata(response.getSessionMetadata()))
+                  );
+
+              case CHECK_VERIFICATION_CODE_ERROR_TYPE_SESSION_NOT_FOUND -> throw new CompletionException(
+                  new RegistrationServiceException(null)
+              );
+
+              default -> throw new CompletionException(
+                  new RuntimeException("Failed to check verification code: " + response.getError().getErrorType()));
             }
           } else {
-            return response.getVerified() || response.getSessionMetadata().getVerified();
+            return buildSessionResponseFromMetadata(response.getSessionMetadata());
           }
         });
   }
 
-  public CompletableFuture<Optional<RegistrationSession>> getSession(final byte[] sessionId,
+  public CompletableFuture<Optional<RegistrationServiceSession>> getSession(final byte[] sessionId,
       final Duration timeout) {
     return toCompletableFuture(stub.withDeadline(toDeadline(timeout)).getSessionMetadata(
         GetRegistrationSessionMetadataRequest.newBuilder()
@@ -173,11 +237,16 @@ public class RegistrationServiceClient implements Managed {
             }
           }
 
-          final String number = convertNumeralE164ToString(response.getSessionMetadata().getE164());
+          return Optional.of(buildSessionResponseFromMetadata(response.getSessionMetadata()));
-          return Optional.of(new RegistrationSession(number, response.getSessionMetadata().getVerified()));
         });
   }
 
+  private static RegistrationServiceSession buildSessionResponseFromMetadata(
+      final RegistrationSessionMetadata sessionMetadata) {
+    return new RegistrationServiceSession(sessionMetadata.getSessionId().toByteArray(),
+        convertNumeralE164ToString(sessionMetadata.getE164()), sessionMetadata);
+  }
+
   private static Deadline toDeadline(final Duration timeout) {
     return Deadline.after(timeout.toMillis(), TimeUnit.MILLISECONDS);
   }

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.registration;
+
+import java.util.Optional;
+import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
+
+/**
+ * When the Registration Service returns an error, it will also return the latest {@link RegistrationServiceSession}
+ * data, so that clients may have the latest details on requesting and submitting codes.
+ */
+public class RegistrationServiceException extends Exception {
+
+  private final RegistrationServiceSession registrationServiceSession;
+
+  public RegistrationServiceException(final RegistrationServiceSession registrationServiceSession) {
+    super(null, null, true, false);
+    this.registrationServiceSession = registrationServiceSession;
+  }
+
+  /**
+   * @return if empty, the session that encountered should be considered non-existent and may be discarded
+   */
+  public Optional<RegistrationServiceSession> getRegistrationSession() {
+    return Optional.ofNullable(registrationServiceSession);
+  }
+}

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

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.registration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+/**
+ * An error from an SMS/voice provider (“sender”) downstream of Registration Service is mapped to a {@link Reason}, and
+ * may be permanent.
+ */
+public class RegistrationServiceSenderException extends Exception {
+
+  private final Reason reason;
+  private final boolean permanent;
+
+  public static RegistrationServiceSenderException illegalArgument(final boolean permanent) {
+    return new RegistrationServiceSenderException(Reason.ILLEGAL_ARGUMENT, permanent);
+  }
+
+  public static RegistrationServiceSenderException rejected(final boolean permanent) {
+    return new RegistrationServiceSenderException(Reason.PROVIDER_REJECTED, permanent);
+  }
+
+  public static RegistrationServiceSenderException unknown(final boolean permanent) {
+    return new RegistrationServiceSenderException(Reason.PROVIDER_UNAVAILABLE, permanent);
+  }
+
+  private RegistrationServiceSenderException(final Reason reason, final boolean permanent) {
+    super(null, null, true, false);
+    this.reason = reason;
+    this.permanent = permanent;
+  }
+
+  public Reason getReason() {
+    return reason;
+  }
+
+  public boolean isPermanent() {
+    return permanent;
+  }
+
+  public enum Reason {
+
+    @JsonProperty("providerUnavailable")
+    PROVIDER_UNAVAILABLE,
+    @JsonProperty("providerRejected")
+    PROVIDER_REJECTED,
+    @JsonProperty("illegalArgument")
+    ILLEGAL_ARGUMENT
+  }
+}

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

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.registration;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import java.time.Instant;
+import java.util.List;
+import javax.annotation.Nullable;
+import org.whispersystems.textsecuregcm.storage.SerializedExpireableJsonDynamoStore;
+
+/**
+ * Server-internal stored session object. Primarily used by
+ * {@link org.whispersystems.textsecuregcm.controllers.VerificationController} to manage the steps required to begin
+ * requesting codes from Registration Service, in order to get a verified session to be provided to
+ * {@link org.whispersystems.textsecuregcm.controllers.RegistrationController}.
+ *
+ * @param pushChallenge           the value of a push challenge sent to a client, after it submitted a push token
+ * @param requestedInformation    information requested that a client send to the server
+ * @param submittedInformation    information that a client has submitted and that the server has verified
+ * @param allowedToRequestCode    whether the client is allowed to request a code. This request will be forwarded to
+ *                                Registration Service
+ * @param createdTimestamp        when this session was created
+ * @param updatedTimestamp        when this session was updated
+ * @param remoteExpirationSeconds when the remote
+ *                                {@link org.whispersystems.textsecuregcm.entities.RegistrationServiceSession} expires
+ * @see org.whispersystems.textsecuregcm.entities.RegistrationServiceSession
+ * @see org.whispersystems.textsecuregcm.entities.VerificationSessionResponse
+ */
+public record VerificationSession(@Nullable String pushChallenge,
+                                  List<Information> requestedInformation, List<Information> submittedInformation,
+                                  boolean allowedToRequestCode, long createdTimestamp, long updatedTimestamp,
+                                  long remoteExpirationSeconds) implements
+    SerializedExpireableJsonDynamoStore.Expireable {
+
+  @Override
+  public long getExpirationEpochSeconds() {
+    return Instant.ofEpochMilli(updatedTimestamp).plusSeconds(remoteExpirationSeconds).getEpochSecond();
+  }
+
+  public enum Information {
+    @JsonProperty("pushChallenge")
+    PUSH_CHALLENGE,
+    @JsonProperty("captcha")
+    CAPTCHA
+  }
+}

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

@@ -0,0 +1,151 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.storage;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.annotations.VisibleForTesting;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.time.Clock;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.util.AttributeValues;
+import org.whispersystems.textsecuregcm.util.SystemMapper;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import software.amazon.awssdk.services.dynamodb.model.DeleteItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
+
+public abstract class SerializedExpireableJsonDynamoStore<T> {
+
+  public interface Expireable {
+
+    @JsonIgnore
+    long getExpirationEpochSeconds();
+  }
+
+  private final DynamoDbAsyncClient dynamoDbClient;
+  private final String tableName;
+  private final Clock clock;
+  private final Class<T> deserializationTargetClass;
+
+  @VisibleForTesting
+  static final String KEY_KEY = "K";
+
+  private static final String ATTR_SERIALIZED_VALUE = "V";
+  private static final String ATTR_TTL = "E";
+
+  private static final Logger log = LoggerFactory.getLogger(VerificationCodeStore.class);
+
+  public SerializedExpireableJsonDynamoStore(final DynamoDbAsyncClient dynamoDbClient, final String tableName,
+      final Clock clock) {
+    this.dynamoDbClient = dynamoDbClient;
+    this.tableName = tableName;
+    this.clock = clock;
+
+    if (getClass().getGenericSuperclass() instanceof ParameterizedType pt) {
+      // Extract the parameterized class declared by concrete implementations, so that it can
+      // be passed to future deserialization calls
+      final Type[] actualTypeArguments = pt.getActualTypeArguments();
+      if (actualTypeArguments.length != 1) {
+        throw new RuntimeException("Unexpected number of type arguments: " + actualTypeArguments.length);
+      }
+      deserializationTargetClass = (Class<T>) actualTypeArguments[0];
+    } else {
+      throw new RuntimeException(
+          "Unable to determine target class for deserialization - generic superclass is not a ParameterizedType");
+    }
+  }
+
+  public CompletableFuture<Void> insert(final String key, final T v) {
+    return put(key, v, builder -> builder.expressionAttributeNames(Map.of(
+        "#key", KEY_KEY
+    )).conditionExpression("attribute_not_exists(#key)"));
+  }
+
+  public CompletableFuture<Void> update(final String key, final T v) {
+    return put(key, v, ignored -> {
+    });
+  }
+
+  private CompletableFuture<Void> put(final String key, final T v,
+      final Consumer<PutItemRequest.Builder> putRequestCustomizer) {
+    try {
+      final Map<String, AttributeValue> attributeValueMap = new HashMap<>(Map.of(
+          KEY_KEY, AttributeValues.fromString(key),
+          ATTR_SERIALIZED_VALUE,
+          AttributeValues.fromString(SystemMapper.getMapper().writeValueAsString(v))));
+      if (v instanceof Expireable ev) {
+        attributeValueMap.put(ATTR_TTL, AttributeValues.fromLong(getExpirationTimestamp(ev)));
+      }
+      final PutItemRequest.Builder builder = PutItemRequest.builder()
+          .tableName(tableName)
+          .item(attributeValueMap);
+      putRequestCustomizer.accept(builder);
+
+      return dynamoDbClient.putItem(builder.build())
+          .thenRun(() -> {
+          });
+    } catch (final JsonProcessingException e) {
+      // This should never happen when writing directly to a string except in cases of serious misconfiguration, which
+      // would be caught by tests.
+      throw new AssertionError(e);
+    }
+  }
+
+  private long getExpirationTimestamp(final Expireable v) {
+    return v.getExpirationEpochSeconds();
+  }
+
+  public CompletableFuture<Optional<T>> findForKey(final String key) {
+    return dynamoDbClient.getItem(GetItemRequest.builder()
+            .tableName(tableName)
+            .consistentRead(true)
+            .key(Map.of(KEY_KEY, AttributeValues.fromString(key)))
+            .build())
+        .thenApply(response -> {
+          try {
+            return response.hasItem()
+                ? filterMaybeExpiredValue(
+                SystemMapper.getMapper()
+                    .readValue(response.item().get(ATTR_SERIALIZED_VALUE).s(), deserializationTargetClass))
+                : Optional.empty();
+          } catch (final JsonProcessingException e) {
+            log.error("Failed to parse stored value", e);
+            return Optional.empty();
+          }
+        });
+  }
+
+  private Optional<T> filterMaybeExpiredValue(T v) {
+    // It's possible for DynamoDB to return items after their expiration time (although it is very unlikely for small
+    // tables)
+    if (v instanceof Expireable ev) {
+      if (getExpirationTimestamp(ev) < clock.instant().getEpochSecond()) {
+        return Optional.empty();
+      }
+    }
+
+    return Optional.of(v);
+  }
+
+  public CompletableFuture<Void> remove(final String key) {
+    return dynamoDbClient.deleteItem(DeleteItemRequest.builder()
+            .tableName(tableName)
+            .key(Map.of(KEY_KEY, AttributeValues.fromString(key)))
+            .build())
+        .thenRun(() -> {
+        });
+  }
+
+}

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.storage;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import org.whispersystems.textsecuregcm.registration.VerificationSession;
+
+public class VerificationSessionManager {
+
+  private final VerificationSessions verificationSessions;
+
+  public VerificationSessionManager(final VerificationSessions verificationSessions) {
+    this.verificationSessions = verificationSessions;
+  }
+
+  public CompletableFuture<Void> insert(final String encodedSessionId, final VerificationSession verificationSession) {
+    return verificationSessions.insert(encodedSessionId, verificationSession);
+  }
+
+  public CompletableFuture<Void> update(final String encodedSessionId, final VerificationSession verificationSession) {
+    return verificationSessions.update(encodedSessionId, verificationSession);
+  }
+
+  public CompletableFuture<Optional<VerificationSession>> findForId(final String encodedSessionId) {
+    return verificationSessions.findForKey(encodedSessionId);
+  }
+
+}

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

@@ -0,0 +1,17 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.storage;
+
+import java.time.Clock;
+import org.whispersystems.textsecuregcm.registration.VerificationSession;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+
+public class VerificationSessions extends SerializedExpireableJsonDynamoStore<VerificationSession> {
+
+  public VerificationSessions(final DynamoDbAsyncClient dynamoDbClient, final String tableName, final Clock clock) {
+    super(dynamoDbClient, tableName, clock);
+  }
+}

service/src/main/proto/RegistrationService.proto

@@ -66,6 +66,66 @@ message RegistrationSessionMetadata {
    * The phone number associated with this registration session.
    */
   uint64 e164 = 3;
+
+  /**
+   * Indicates whether the caller may request delivery of a verification code
+   * via SMS now or at some time in the future. If true, the time a caller must
+   * wait before requesting a verification code via SMS is given in the
+   * `next_sms_seconds` field.
+   */
+  bool may_request_sms = 4;
+
+  /**
+   * The duration, in seconds, after which a caller will next be allowed to
+   * request delivery of a verification code via SMS if `may_request_sms` is
+   * true. If zero, a caller may request a verification code via SMS
+   * immediately. If `may_request_sms` is false, this field has no meaning.
+   */
+  uint64 next_sms_seconds = 5;
+
+  /**
+   * Indicates whether the caller may request delivery of a verification code
+   * via a phone call now or at some time in the future. If true, the time a
+   * caller must wait before requesting a verification code via SMS is given in
+   * the `next_voice_call_seconds` field. If false, simply waiting will not
+   * allow the caller to request a phone call and the caller may need to
+   * perform some other action (like attempting verification code delivery via
+   * SMS) before requesting a voice call.
+   */
+  bool may_request_voice_call = 6;
+
+  /**
+   * The duration, in seconds, after which a caller will next be allowed to
+   * request delivery of a verification code via a phone call if
+   * `may_request_voice_call` is true. If zero, a caller may request a
+   * verification code via a phone call immediately. If `may_request_voice_call`
+   * is false, this field has no meaning.
+   */
+  uint64 next_voice_call_seconds = 7;
+
+  /**
+   * Indicates whether the caller may submit new verification codes now or at
+   * some time in the future. If true, the time a caller must wait before
+   * submitting a verification code is given in the `next_code_check_seconds`
+   * field. If false, simply waiting will not allow the caller to submit a
+   * verification code and the caller may need to perform some other action
+   * (like requesting delivery of a verification code) before checking a
+   * verification code.
+   */
+  bool may_check_code = 8;
+
+  /**
+   * The duration, in seconds, after which a caller will next be allowed to
+   * submit a verification code if `may_check_code` is true. If zero, a caller
+   * may submit a verification code immediately. If `may_check_code` is false,
+   * this field has no meaning.
+   */
+  uint64 next_code_check_seconds = 9;
+
+  /**
+   * The duration, in seconds, after which this session will expire.
+   */
+  uint64 expiration_seconds = 10;
 }
 
 message CreateRegistrationSessionError {
@@ -315,29 +375,30 @@ message CheckVerificationCodeError {
 enum CheckVerificationCodeErrorType {
   CHECK_VERIFICATION_CODE_ERROR_TYPE_UNSPECIFIED = 0;
 
-  /**
-   * The caller has made too many incorrect guesses within the scope of this
-   * session and may not make any further guesses.
-   */
-  CHECK_VERIFICATION_CODE_ERROR_TYPE_ATTEMPTS_EXHAUSTED = 1;
-
   /**
    * The caller has attempted to submit a verification code even though no
    * verification codes have been sent within the scope of this session. The
    * caller must issue a "send code" request before trying again.
    */
-  CHECK_VERIFICATION_CODE_ERROR_TYPE_NO_CODE_SENT = 2;
+  CHECK_VERIFICATION_CODE_ERROR_TYPE_NO_CODE_SENT = 1;
 
   /**
    * The caller has made too many guesses within some period of time. Callers
    * should wait for the duration prescribed in the session metadata object
    * elsewhere in the response before trying again.
    */
-  CHECK_VERIFICATION_CODE_ERROR_TYPE_RATE_LIMITED = 3;
+  CHECK_VERIFICATION_CODE_ERROR_TYPE_RATE_LIMITED = 2;
 
   /**
    * The session identified in this request could not be found (possibly due to
    * session expiration).
    */
-  CHECK_VERIFICATION_CODE_ERROR_TYPE_SESSION_NOT_FOUND = 4;
+  CHECK_VERIFICATION_CODE_ERROR_TYPE_SESSION_NOT_FOUND = 3;
+
+  /**
+   * The session identified in this request is still active, but the most
+   * recently-sent code has expired. Callers should request a new code, then
+   * try again.
+   */
+  CHECK_VERIFICATION_CODE_ERROR_TYPE_ATTEMPT_EXPIRED = 4;
 }

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

@@ -22,8 +22,7 @@ class StoredVerificationCodeTest {
 
   private static Stream<Arguments> isValid() {
     return Stream.of(
-        Arguments.of(
+        Arguments.of(new StoredVerificationCode("code", System.currentTimeMillis(), null, null), "code", true),
-            new StoredVerificationCode("code", System.currentTimeMillis(), null, null), "code", true),
         Arguments.of(new StoredVerificationCode("code", System.currentTimeMillis(), null, null), "incorrect", false),
         Arguments.of(new StoredVerificationCode("", System.currentTimeMillis(), null, null), "", false)
     );

service/src/test/java/org/whispersystems/textsecuregcm/controllers/AccountControllerTest.java

@@ -75,6 +75,7 @@ import org.whispersystems.textsecuregcm.auth.StoredVerificationCode;
 import org.whispersystems.textsecuregcm.auth.TurnTokenGenerator;
 import org.whispersystems.textsecuregcm.captcha.AssessmentResult;
 import org.whispersystems.textsecuregcm.captcha.CaptchaChecker;
+import org.whispersystems.textsecuregcm.captcha.RegistrationCaptchaManager;
 import org.whispersystems.textsecuregcm.configuration.SecureBackupServiceConfiguration;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicCaptchaConfiguration;
 import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
@@ -197,6 +198,8 @@ class AccountControllerTest {
 
   private static final RegistrationLockVerificationManager registrationLockVerificationManager = new RegistrationLockVerificationManager(
       accountsManager, clientPresenceManager, backupCredentialsGenerator, rateLimiters);
+  private static final RegistrationCaptchaManager registrationCaptchaManager = new RegistrationCaptchaManager(
+      captchaChecker, rateLimiters, Map.of(TEST_NUMBER, 123456), dynamicConfigurationManager);
 
   private static final ResourceExtension resources = ResourceExtension.builder()
       .addProvider(AuthHelper.getAuthFilter())
@@ -217,8 +220,7 @@ class AccountControllerTest {
           registrationServiceClient,
           dynamicConfigurationManager,
           turnTokenGenerator,
-          Map.of(TEST_NUMBER, 123456),
+          registrationCaptchaManager,
-          captchaChecker,
           pushNotificationManager,
           changeNumberManager,
           registrationLockVerificationManager,
@@ -250,30 +252,43 @@ class AccountControllerTest {
     when(rateLimiters.getUsernameLookupLimiter()).thenReturn(usernameLookupLimiter);
 
     when(senderPinAccount.getLastSeen()).thenReturn(System.currentTimeMillis());
-    when(senderPinAccount.getRegistrationLock()).thenReturn(new StoredRegistrationLock(Optional.empty(), Optional.empty(), System.currentTimeMillis()));
+    when(senderPinAccount.getRegistrationLock()).thenReturn(
+        new StoredRegistrationLock(Optional.empty(), Optional.empty(), System.currentTimeMillis()));
 
     when(senderHasStorage.getUuid()).thenReturn(UUID.randomUUID());
     when(senderHasStorage.isStorageSupported()).thenReturn(true);
-    when(senderHasStorage.getRegistrationLock()).thenReturn(new StoredRegistrationLock(Optional.empty(), Optional.empty(), System.currentTimeMillis()));
+    when(senderHasStorage.getRegistrationLock()).thenReturn(
+        new StoredRegistrationLock(Optional.empty(), Optional.empty(), System.currentTimeMillis()));
 
-    when(senderRegLockAccount.getRegistrationLock()).thenReturn(new StoredRegistrationLock(Optional.of(registrationLockCredentials.hash()), Optional.of(registrationLockCredentials.salt()), System.currentTimeMillis()));
+    when(senderRegLockAccount.getRegistrationLock()).thenReturn(
+        new StoredRegistrationLock(Optional.of(registrationLockCredentials.hash()),
+            Optional.of(registrationLockCredentials.salt()), System.currentTimeMillis()));
     when(senderRegLockAccount.getLastSeen()).thenReturn(System.currentTimeMillis());
     when(senderRegLockAccount.getUuid()).thenReturn(SENDER_REG_LOCK_UUID);
     when(senderRegLockAccount.getNumber()).thenReturn(SENDER_REG_LOCK);
 
-    when(senderTransfer.getRegistrationLock()).thenReturn(new StoredRegistrationLock(Optional.empty(), Optional.empty(), System.currentTimeMillis()));
+    when(senderTransfer.getRegistrationLock()).thenReturn(
+        new StoredRegistrationLock(Optional.empty(), Optional.empty(), System.currentTimeMillis()));
     when(senderTransfer.getUuid()).thenReturn(SENDER_TRANSFER_UUID);
     when(senderTransfer.getNumber()).thenReturn(SENDER_TRANSFER);
 
-    when(pendingAccountsManager.getCodeForNumber(SENDER)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER)).thenReturn(
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", null)));
     when(pendingAccountsManager.getCodeForNumber(SENDER_OLD)).thenReturn(Optional.empty());
-    when(pendingAccountsManager.getCodeForNumber(SENDER_PIN)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER_PIN)).thenReturn(
-    when(pendingAccountsManager.getCodeForNumber(SENDER_REG_LOCK)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
-    when(pendingAccountsManager.getCodeForNumber(SENDER_OVER_PIN)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER_REG_LOCK)).thenReturn(
-    when(pendingAccountsManager.getCodeForNumber(SENDER_OVER_PREFIX)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", null)));
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
-    when(pendingAccountsManager.getCodeForNumber(SENDER_PREAUTH)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "validchallenge", null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER_OVER_PIN)).thenReturn(
-    when(pendingAccountsManager.getCodeForNumber(SENDER_HAS_STORAGE)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
-    when(pendingAccountsManager.getCodeForNumber(SENDER_TRANSFER)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER_OVER_PREFIX)).thenReturn(
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER_PREAUTH)).thenReturn(
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "validchallenge", null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER_HAS_STORAGE)).thenReturn(
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER_TRANSFER)).thenReturn(
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), null, null)));
 
     when(accountsManager.getByE164(eq(SENDER_PIN))).thenReturn(Optional.of(senderPinAccount));
     when(accountsManager.getByE164(eq(SENDER_REG_LOCK))).thenReturn(Optional.of(senderRegLockAccount));
@@ -953,7 +968,8 @@ class AccountControllerTest {
     final String challenge = "challenge";
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
-    when(pendingAccountsManager.getCodeForNumber(SENDER)).thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), challenge, null)));
+    when(pendingAccountsManager.getCodeForNumber(SENDER)).thenReturn(Optional.of(
+        new StoredVerificationCode(null, System.currentTimeMillis(), challenge, null)));
     when(registrationServiceClient.sendRegistrationCode(any(), any(), any(), any(), any()))
         .thenReturn(CompletableFuture.completedFuture(sessionId));
 
@@ -1103,8 +1119,8 @@ class AccountControllerTest {
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
     when(pendingAccountsManager.getCodeForNumber(SENDER_REG_LOCK))
-        .thenReturn(Optional.of(
+        .thenReturn(
-            new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
+            Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(sessionId, "666666", AccountController.REGISTRATION_RPC_TIMEOUT))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1137,8 +1153,8 @@ class AccountControllerTest {
       final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
       when(pendingAccountsManager.getCodeForNumber(SENDER_REG_LOCK))
-          .thenReturn(Optional.of(
+          .thenReturn(
-              new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
+              Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
 
       when(registrationServiceClient.checkVerificationCode(sessionId, "666666", AccountController.REGISTRATION_RPC_TIMEOUT))
           .thenReturn(CompletableFuture.completedFuture(true));
@@ -1164,8 +1180,8 @@ class AccountControllerTest {
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
     when(pendingAccountsManager.getCodeForNumber(SENDER_REG_LOCK))
-        .thenReturn(Optional.of(
+        .thenReturn(
-            new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
+            Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(sessionId, "666666", AccountController.REGISTRATION_RPC_TIMEOUT))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1191,8 +1207,8 @@ class AccountControllerTest {
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
     when(pendingAccountsManager.getCodeForNumber(SENDER_REG_LOCK))
-        .thenReturn(Optional.of(
+        .thenReturn(
-            new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
+            Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "666666-push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(sessionId, "666666", AccountController.REGISTRATION_RPC_TIMEOUT))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1224,8 +1240,7 @@ class AccountControllerTest {
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
     when(pendingAccountsManager.getCodeForNumber(SENDER_TRANSFER))
-        .thenReturn(Optional.of(
+        .thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", sessionId)));
-            new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(sessionId, "1234", AccountController.REGISTRATION_RPC_TIMEOUT))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1249,8 +1264,7 @@ class AccountControllerTest {
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
     when(pendingAccountsManager.getCodeForNumber(SENDER_TRANSFER))
-        .thenReturn(Optional.of(
+        .thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", sessionId)));
-            new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(sessionId, "1234", AccountController.REGISTRATION_RPC_TIMEOUT))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1274,8 +1288,7 @@ class AccountControllerTest {
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
     when(pendingAccountsManager.getCodeForNumber(SENDER_TRANSFER))
-        .thenReturn(Optional.of(
+        .thenReturn(Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", sessionId)));
-            new StoredVerificationCode(null, System.currentTimeMillis(), "1234-push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(sessionId, "1234", AccountController.REGISTRATION_RPC_TIMEOUT))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1299,8 +1312,8 @@ class AccountControllerTest {
     final String code = "987654";
     final byte[] sessionId = "session".getBytes(StandardCharsets.UTF_8);
 
-    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(Optional.of(
+    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(
-        new StoredVerificationCode(null, System.currentTimeMillis(), "push", sessionId)));
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(any(), any(), any()))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1400,8 +1413,8 @@ class AccountControllerTest {
     final String code = "987654";
     final byte[] sessionId = "session-id".getBytes(StandardCharsets.UTF_8);
 
-    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(Optional.of(
+    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(
-        new StoredVerificationCode(code, System.currentTimeMillis(), "push", sessionId)));
+        Optional.of(new StoredVerificationCode(code, System.currentTimeMillis(), "push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(any(), any(), any()))
         .thenReturn(CompletableFuture.completedFuture(false));
@@ -1426,8 +1439,8 @@ class AccountControllerTest {
     final String code = "987654";
     final byte[] sessionId = "session-id".getBytes(StandardCharsets.UTF_8);
 
-    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(Optional.of(
+    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(
-        new StoredVerificationCode(code, System.currentTimeMillis(), "push", sessionId)));
+        Optional.of(new StoredVerificationCode(code, System.currentTimeMillis(), "push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(any(), any(), any()))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1460,8 +1473,8 @@ class AccountControllerTest {
     final String code = "987654";
     final byte[] sessionId = "session-id".getBytes(StandardCharsets.UTF_8);
 
-    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(Optional.of(
+    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(
-        new StoredVerificationCode(code, System.currentTimeMillis(), "push", sessionId)));
+        Optional.of(new StoredVerificationCode(code, System.currentTimeMillis(), "push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(any(), any(), any()))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1539,8 +1552,8 @@ class AccountControllerTest {
     final String reglock = "setec-astronomy";
     final byte[] sessionId = "session-id".getBytes(StandardCharsets.UTF_8);
 
-    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(Optional.of(
+    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(
-        new StoredVerificationCode(null, System.currentTimeMillis(), "push", sessionId)));
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(any(), any(), any()))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -1591,8 +1604,8 @@ class AccountControllerTest {
     when(AuthHelper.VALID_ACCOUNT.getDevice(2L)).thenReturn(Optional.of(device2));
     when(AuthHelper.VALID_ACCOUNT.getDevice(3L)).thenReturn(Optional.of(device3));
 
-    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(Optional.of(
+    when(pendingAccountsManager.getCodeForNumber(number)).thenReturn(
-        new StoredVerificationCode(null, System.currentTimeMillis(), "push", sessionId)));
+        Optional.of(new StoredVerificationCode(null, System.currentTimeMillis(), "push", sessionId)));
 
     when(registrationServiceClient.checkVerificationCode(any(), any(), any()))
         .thenReturn(CompletableFuture.completedFuture(true));
@@ -2231,8 +2244,10 @@ class AccountControllerTest {
         Arguments.of("123456", null, false),
         Arguments.of(null, new StoredVerificationCode(null, 0, null, null), false),
         Arguments.of(null, new StoredVerificationCode(null, 0, "123456", null), false),
-        Arguments.of("654321", new StoredVerificationCode(null, 0, "123456", null), false),
+        Arguments.of("654321", new StoredVerificationCode(null, 0, "123456", null),
-        Arguments.of("123456", new StoredVerificationCode(null, 0, "123456", null), true)
+            false),
+        Arguments.of("123456", new StoredVerificationCode(null, 0, "123456", null),
+            true)
     );
   }
 }

service/src/test/java/org/whispersystems/textsecuregcm/controllers/AccountControllerV2Test.java

@@ -24,6 +24,7 @@ import io.dropwizard.auth.PolymorphicAuthValueFactoryProvider;
 import io.dropwizard.testing.junit5.DropwizardExtensionsSupport;
 import io.dropwizard.testing.junit5.ResourceExtension;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.List;
@@ -59,7 +60,7 @@ import org.whispersystems.textsecuregcm.auth.RegistrationLockVerificationManager
 import org.whispersystems.textsecuregcm.entities.AccountIdentityResponse;
 import org.whispersystems.textsecuregcm.entities.ChangeNumberRequest;
 import org.whispersystems.textsecuregcm.entities.PhoneNumberDiscoverabilityRequest;
-import org.whispersystems.textsecuregcm.entities.RegistrationSession;
+import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
 import org.whispersystems.textsecuregcm.limits.RateLimiter;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.mappers.ImpossiblePhoneNumberExceptionMapper;
@@ -78,7 +79,9 @@ import org.whispersystems.textsecuregcm.util.SystemMapper;
 @ExtendWith(DropwizardExtensionsSupport.class)
 class AccountControllerV2Test {
 
-  public static final String NEW_NUMBER = PhoneNumberUtil.getInstance().format(
+  private static final long SESSION_EXPIRATION_SECONDS = Duration.ofMinutes(10).toSeconds();
+
+  private static final String NEW_NUMBER = PhoneNumberUtil.getInstance().format(
       PhoneNumberUtil.getInstance().getExampleNumber("US"),
       PhoneNumberUtil.PhoneNumberFormat.E164);
 
@@ -146,7 +149,9 @@ class AccountControllerV2Test {
     void changeNumberSuccess() throws Exception {
 
       when(registrationServiceClient.getSession(any(), any()))
-          .thenReturn(CompletableFuture.completedFuture(Optional.of(new RegistrationSession(NEW_NUMBER, true))));
+          .thenReturn(CompletableFuture.completedFuture(
+              Optional.of(new RegistrationServiceSession(new byte[16], NEW_NUMBER, true, null, null, null,
+                  SESSION_EXPIRATION_SECONDS))));
 
       final AccountIdentityResponse accountIdentityResponse =
           resources.getJerseyTest()
@@ -245,7 +250,7 @@ class AccountControllerV2Test {
 
     @ParameterizedTest
     @MethodSource
-    void registrationServiceSessionCheck(@Nullable final RegistrationSession session, final int expectedStatus,
+    void registrationServiceSessionCheck(@Nullable final RegistrationServiceSession session, final int expectedStatus,
         final String message) {
       when(registrationServiceClient.getSession(any(), any()))
           .thenReturn(CompletableFuture.completedFuture(Optional.ofNullable(session)));
@@ -263,8 +268,14 @@ class AccountControllerV2Test {
     static Stream<Arguments> registrationServiceSessionCheck() {
       return Stream.of(
           Arguments.of(null, 401, "session not found"),
-          Arguments.of(new RegistrationSession("+18005551234", false), 400, "session number mismatch"),
+          Arguments.of(new RegistrationServiceSession(new byte[16], "+18005551234", false, null, null, null,
-          Arguments.of(new RegistrationSession(NEW_NUMBER, false), 401, "session not verified")
+                  SESSION_EXPIRATION_SECONDS), 400,
+              "session number mismatch"),
+          Arguments.of(
+              new RegistrationServiceSession(new byte[16], NEW_NUMBER, false, null, null, null,
+                  SESSION_EXPIRATION_SECONDS),
+              401,
+              "session not verified")
       );
     }
 
@@ -273,7 +284,9 @@ class AccountControllerV2Test {
     void registrationLock(final RegistrationLockError error) throws Exception {
       when(registrationServiceClient.getSession(any(), any()))
           .thenReturn(
-              CompletableFuture.completedFuture(Optional.of(new RegistrationSession(NEW_NUMBER, true))));
+              CompletableFuture.completedFuture(
+                  Optional.of(new RegistrationServiceSession(new byte[16], NEW_NUMBER, true, null, null, null,
+                      SESSION_EXPIRATION_SECONDS))));
 
       when(accountsManager.getByE164(any())).thenReturn(Optional.of(mock(Account.class)));
 

service/src/test/java/org/whispersystems/textsecuregcm/controllers/RegistrationControllerTest.java

@@ -17,6 +17,7 @@ import com.google.i18n.phonenumbers.PhoneNumberUtil;
 import io.dropwizard.testing.junit5.DropwizardExtensionsSupport;
 import io.dropwizard.testing.junit5.ResourceExtension;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
 import java.util.Base64;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -43,7 +44,7 @@ import org.whispersystems.textsecuregcm.auth.RegistrationLockError;
 import org.whispersystems.textsecuregcm.auth.RegistrationLockVerificationManager;
 import org.whispersystems.textsecuregcm.entities.AccountAttributes;
 import org.whispersystems.textsecuregcm.entities.RegistrationRequest;
-import org.whispersystems.textsecuregcm.entities.RegistrationSession;
+import org.whispersystems.textsecuregcm.entities.RegistrationServiceSession;
 import org.whispersystems.textsecuregcm.limits.RateLimiter;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.mappers.ImpossiblePhoneNumberExceptionMapper;
@@ -59,11 +60,12 @@ import org.whispersystems.textsecuregcm.util.SystemMapper;
 @ExtendWith(DropwizardExtensionsSupport.class)
 class RegistrationControllerTest {
 
+  private static final long SESSION_EXPIRATION_SECONDS = Duration.ofMinutes(10).toSeconds();
+
   private static final String NUMBER = PhoneNumberUtil.getInstance().format(
       PhoneNumberUtil.getInstance().getExampleNumber("US"),
       PhoneNumberUtil.PhoneNumberFormat.E164);
-
+  private static final String PASSWORD = "password";
-  public static final String PASSWORD = "password";
 
   private final AccountsManager accountsManager = mock(AccountsManager.class);
   private final RegistrationServiceClient registrationServiceClient = mock(RegistrationServiceClient.class);
@@ -187,7 +189,7 @@ class RegistrationControllerTest {
 
   @ParameterizedTest
   @MethodSource
-  void registrationServiceSessionCheck(@Nullable final RegistrationSession session, final int expectedStatus,
+  void registrationServiceSessionCheck(@Nullable final RegistrationServiceSession session, final int expectedStatus,
       final String message) {
     when(registrationServiceClient.getSession(any(), any()))
         .thenReturn(CompletableFuture.completedFuture(Optional.ofNullable(session)));
@@ -204,8 +206,15 @@ class RegistrationControllerTest {
   static Stream<Arguments> registrationServiceSessionCheck() {
     return Stream.of(
         Arguments.of(null, 401, "session not found"),
-        Arguments.of(new RegistrationSession("+18005551234", false), 400, "session number mismatch"),
+        Arguments.of(
-        Arguments.of(new RegistrationSession(NUMBER, false), 401, "session not verified")
+            new RegistrationServiceSession(new byte[16], "+18005551234", false, null, null, null,
+                SESSION_EXPIRATION_SECONDS),
+            400,
+            "session number mismatch"),
+        Arguments.of(
+            new RegistrationServiceSession(new byte[16], NUMBER, false, null, null, null, SESSION_EXPIRATION_SECONDS),
+            401,
+            "session not verified")
     );
   }
 
@@ -244,7 +253,10 @@ class RegistrationControllerTest {
   @EnumSource(RegistrationLockError.class)
   void registrationLock(final RegistrationLockError error) throws Exception {
     when(registrationServiceClient.getSession(any(), any()))
-        .thenReturn(CompletableFuture.completedFuture(Optional.of(new RegistrationSession(NUMBER, true))));
+        .thenReturn(
+            CompletableFuture.completedFuture(
+                Optional.of(new RegistrationServiceSession(new byte[16], NUMBER, true, null, null, null,
+                    SESSION_EXPIRATION_SECONDS))));
 
     when(accountsManager.getByE164(any())).thenReturn(Optional.of(mock(Account.class)));
 
@@ -275,7 +287,10 @@ class RegistrationControllerTest {
   void deviceTransferAvailable(final boolean existingAccount, final boolean transferSupported,
       final boolean skipDeviceTransfer, final int expectedStatus) throws Exception {
     when(registrationServiceClient.getSession(any(), any()))
-        .thenReturn(CompletableFuture.completedFuture(Optional.of(new RegistrationSession(NUMBER, true))));
+        .thenReturn(
+            CompletableFuture.completedFuture(
+                Optional.of(new RegistrationServiceSession(new byte[16], NUMBER, true, null, null, null,
+                    SESSION_EXPIRATION_SECONDS))));
 
     final Optional<Account> maybeAccount;
     if (existingAccount) {
@@ -301,7 +316,10 @@ class RegistrationControllerTest {
   @Test
   void registrationSuccess() throws Exception {
     when(registrationServiceClient.getSession(any(), any()))
-        .thenReturn(CompletableFuture.completedFuture(Optional.of(new RegistrationSession(NUMBER, true))));
+        .thenReturn(
+            CompletableFuture.completedFuture(
+                Optional.of(new RegistrationServiceSession(new byte[16], NUMBER, true, null, null, null,
+                    SESSION_EXPIRATION_SECONDS))));
     when(accountsManager.create(any(), any(), any(), any(), any()))
         .thenReturn(mock(Account.class));
 

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

@@ -0,0 +1,183 @@
+/*
+ * Copyright 2013-2021 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.storage;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition;
+import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
+
+class SerializedExpireableJsonDynamoStoreTest {
+
+  static abstract class Tests<T> {
+
+    private static final String TABLE_NAME = "test";
+    private static final String KEY = "foo";
+
+    static final Clock clock = Clock.systemUTC();
+
+    interface Value {
+
+      String v();
+    }
+
+    @RegisterExtension
+    static final DynamoDbExtension DYNAMO_DB_EXTENSION = DynamoDbExtension.builder()
+        .tableName(TABLE_NAME)
+        .hashKey(SerializedExpireableJsonDynamoStore.KEY_KEY)
+        .attributeDefinition(AttributeDefinition.builder()
+            .attributeName(SerializedExpireableJsonDynamoStore.KEY_KEY)
+            .attributeType(ScalarAttributeType.S)
+            .build())
+        .build();
+
+    private SerializedExpireableJsonDynamoStore<T> store;
+
+    abstract SerializedExpireableJsonDynamoStore<T> getStore(final DynamoDbAsyncClient dynamoDbClient,
+        final String tableName);
+
+    abstract T testValue(final String v);
+
+    abstract T maybeExpiredTestValue(final String v);
+
+    @BeforeEach
+    void setUp() {
+      store = getStore(DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(), TABLE_NAME);
+    }
+
+    @Test
+    void testStoreAndFind() throws Exception {
+      assertEquals(Optional.empty(), store.findForKey(KEY).get(1, TimeUnit.SECONDS));
+
+      final T original = testValue("1234");
+      final T second = testValue("5678");
+
+      store.insert(KEY, original).get(1, TimeUnit.SECONDS);
+      {
+        final Optional<T> maybeValue = store.findForKey(KEY).get(1, TimeUnit.SECONDS);
+
+        assertTrue(maybeValue.isPresent());
+        assertEquals(original, maybeValue.get());
+      }
+
+      assertThrows(Exception.class, () -> store.insert(KEY, second).get(1, TimeUnit.SECONDS));
+
+      assertDoesNotThrow(() -> store.update(KEY, second).get(1, TimeUnit.SECONDS));
+      {
+        final Optional<T> maybeValue = store.findForKey(KEY).get(1, TimeUnit.SECONDS);
+
+        assertTrue(maybeValue.isPresent());
+        assertEquals(second, maybeValue.get());
+      }
+    }
+
+    @Test
+    void testRemove() throws Exception {
+      assertEquals(Optional.empty(), store.findForKey(KEY).get(1, TimeUnit.SECONDS));
+
+      store.insert(KEY, testValue("1234")).get(1, TimeUnit.SECONDS);
+      assertTrue(store.findForKey(KEY).get(1, TimeUnit.SECONDS).isPresent());
+
+      store.remove(KEY).get(1, TimeUnit.SECONDS);
+      assertFalse(store.findForKey(KEY).get(1, TimeUnit.SECONDS).isPresent());
+
+      final T v = maybeExpiredTestValue("1234");
+      store.insert(KEY, v).get(1, TimeUnit.SECONDS);
+
+      assertEquals(v instanceof SerializedExpireableJsonDynamoStore.Expireable,
+          store.findForKey(KEY).get(1, TimeUnit.SECONDS).isEmpty());
+    }
+
+  }
+
+  record Expires(String v, long timestamp) implements SerializedExpireableJsonDynamoStore.Expireable, Tests.Value {
+
+    static final Duration EXPIRATION = Duration.ofSeconds(30);
+
+    @Override
+    public long getExpirationEpochSeconds() {
+      return Instant.ofEpochMilli(timestamp()).plus(EXPIRATION).getEpochSecond();
+    }
+  }
+
+  @Nested
+  class Expireable extends Tests<Expires> {
+
+    class ExpiresStore extends SerializedExpireableJsonDynamoStore<Expires> {
+
+      public ExpiresStore(final DynamoDbAsyncClient dynamoDbClient, final String tableName) {
+        super(dynamoDbClient, tableName, clock);
+      }
+    }
+
+    private static final long VALID_TIMESTAMP = Instant.now().toEpochMilli();
+    private static final long EXPIRED_TIMESTAMP = Instant.now().minus(Expires.EXPIRATION).minus(
+        Duration.ofHours(1)).toEpochMilli();
+
+    @Override
+    SerializedExpireableJsonDynamoStore<Expires> getStore(final DynamoDbAsyncClient dynamoDbClient,
+        final String tableName) {
+      return new ExpiresStore(dynamoDbClient, tableName);
+    }
+
+    @Override
+    Expires testValue(final String v) {
+      return new Expires(v, VALID_TIMESTAMP);
+    }
+
+    @Override
+    Expires maybeExpiredTestValue(final String v) {
+      return new Expires(v, EXPIRED_TIMESTAMP);
+    }
+  }
+
+  record DoesNotExpire(String v) implements Tests.Value {
+
+  }
+
+
+  @Nested
+  class NotExpireable extends Tests<DoesNotExpire> {
+
+    class DoesNotExpireStore extends SerializedExpireableJsonDynamoStore<DoesNotExpire> {
+
+      public DoesNotExpireStore(final DynamoDbAsyncClient dynamoDbClient, final String tableName) {
+        super(dynamoDbClient, tableName, clock);
+      }
+    }
+
+    @Override
+    SerializedExpireableJsonDynamoStore<DoesNotExpire> getStore(final DynamoDbAsyncClient dynamoDbClient,
+        final String tableName) {
+      return new DoesNotExpireStore(dynamoDbClient, tableName);
+    }
+
+    @Override
+    DoesNotExpire testValue(final String v) {
+      return new DoesNotExpire(v);
+    }
+
+    @Override
+    DoesNotExpire maybeExpiredTestValue(final String v) {
+      return new DoesNotExpire(v);
+    }
+  }
+
+}

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

@@ -53,8 +53,10 @@ class VerificationCodeStoreTest {
   void testStoreAndFind() {
     assertEquals(Optional.empty(), verificationCodeStore.findForNumber(PHONE_NUMBER));
 
-    final StoredVerificationCode originalCode = new StoredVerificationCode("1234", VALID_TIMESTAMP, "abcd", "session".getBytes(StandardCharsets.UTF_8));
+    final StoredVerificationCode originalCode = new StoredVerificationCode("1234", VALID_TIMESTAMP, "abcd",
-    final StoredVerificationCode secondCode = new StoredVerificationCode("5678", VALID_TIMESTAMP, "efgh", "changed-session".getBytes(StandardCharsets.UTF_8));
+        "session".getBytes(StandardCharsets.UTF_8));
+    final StoredVerificationCode secondCode = new StoredVerificationCode("5678", VALID_TIMESTAMP, "efgh",
+        "changed-session".getBytes(StandardCharsets.UTF_8));
 
     verificationCodeStore.insert(PHONE_NUMBER, originalCode);
     {
@@ -77,13 +79,15 @@ class VerificationCodeStoreTest {
   void testRemove() {
     assertEquals(Optional.empty(), verificationCodeStore.findForNumber(PHONE_NUMBER));
 
-    verificationCodeStore.insert(PHONE_NUMBER, new StoredVerificationCode("1234", VALID_TIMESTAMP, "abcd", "session".getBytes(StandardCharsets.UTF_8)));
+    verificationCodeStore.insert(PHONE_NUMBER,
+        new StoredVerificationCode("1234", VALID_TIMESTAMP, "abcd", "session".getBytes(StandardCharsets.UTF_8)));
     assertTrue(verificationCodeStore.findForNumber(PHONE_NUMBER).isPresent());
 
     verificationCodeStore.remove(PHONE_NUMBER);
     assertFalse(verificationCodeStore.findForNumber(PHONE_NUMBER).isPresent());
 
-    verificationCodeStore.insert(PHONE_NUMBER, new StoredVerificationCode("1234", EXPIRED_TIMESTAMP, "abcd", "session".getBytes(StandardCharsets.UTF_8)));
+    verificationCodeStore.insert(PHONE_NUMBER,
+        new StoredVerificationCode("1234", EXPIRED_TIMESTAMP, "abcd", "session".getBytes(StandardCharsets.UTF_8)));
     assertFalse(verificationCodeStore.findForNumber(PHONE_NUMBER).isPresent());
   }
 

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

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.storage;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletionException;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.whispersystems.textsecuregcm.registration.VerificationSession;
+import org.whispersystems.textsecuregcm.util.ExceptionUtils;
+import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition;
+import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException;
+import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
+
+class VerificationSessionsTest {
+
+  private static final String TABLE_NAME = "verification_sessions_test";
+
+  private static final Clock clock = Clock.systemUTC();
+
+  @RegisterExtension
+  static final DynamoDbExtension DYNAMO_DB_EXTENSION = DynamoDbExtension.builder()
+      .tableName(TABLE_NAME)
+      .hashKey(VerificationSessions.KEY_KEY)
+      .attributeDefinition(AttributeDefinition.builder()
+          .attributeName(VerificationSessions.KEY_KEY)
+          .attributeType(ScalarAttributeType.S)
+          .build())
+      .build();
+
+  private VerificationSessions verificationSessions;
+
+  @BeforeEach
+  void setUp() {
+    verificationSessions = new VerificationSessions(DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(), TABLE_NAME, clock);
+  }
+
+  @Test
+  void testExpiration() {
+    final Instant created = Instant.now().minusSeconds(60);
+    final Instant updates = Instant.now();
+    final Duration remoteExpiration = Duration.ofMinutes(2);
+
+    final VerificationSession verificationSession = new VerificationSession(null,
+        List.of(VerificationSession.Information.PUSH_CHALLENGE), Collections.emptyList(), true,
+        created.toEpochMilli(), updates.toEpochMilli(), remoteExpiration.toSeconds());
+
+    assertEquals(updates.plus(remoteExpiration).getEpochSecond(), verificationSession.getExpirationEpochSeconds());
+  }
+
+  @Test
+  void testStore() {
+
+    assertTimeoutPreemptively(Duration.ofSeconds(5), () -> {
+
+      final String sessionId = "sessionId";
+
+      final Optional<VerificationSession> absentSession = verificationSessions.findForKey(sessionId).join();
+      assertTrue(absentSession.isEmpty());
+
+      final VerificationSession session = new VerificationSession(null,
+          List.of(VerificationSession.Information.PUSH_CHALLENGE), Collections.emptyList(), true,
+          clock.millis(), clock.millis(), Duration.ofMinutes(1).toSeconds());
+
+      verificationSessions.insert(sessionId, session).join();
+
+      assertEquals(session, verificationSessions.findForKey(sessionId).join().orElseThrow());
+
+      final CompletionException ce = assertThrows(CompletionException.class,
+          () -> verificationSessions.insert(sessionId, session).join());
+
+      final Throwable t = ExceptionUtils.unwrap(ce);
+      assertTrue(t instanceof ConditionalCheckFailedException,
+          "inserting with the same key should fail conditional checks");
+
+      final VerificationSession updatedSession = new VerificationSession(null, Collections.emptyList(),
+          List.of(VerificationSession.Information.PUSH_CHALLENGE), true, clock.millis(), clock.millis(),
+          Duration.ofMinutes(2).toSeconds());
+      verificationSessions.update(sessionId, updatedSession).join();
+
+      assertEquals(updatedSession, verificationSessions.findForKey(sessionId).join().orElseThrow());
+    });
+  }
+
+}

service/src/test/java/org/whispersystems/textsecuregcm/tests/controllers/DeviceControllerTest.java

@@ -25,7 +25,6 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.stream.Stream;
 import javax.ws.rs.Path;
 import javax.ws.rs.client.Entity;
 import javax.ws.rs.core.MediaType;
@@ -36,8 +35,6 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
 import org.whispersystems.textsecuregcm.auth.DisabledPermittedAuthenticatedAccount;