Add ArchiveController

Adds endpoints for creating and managing backup objects with ZK
anonymous credentials.

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

@@ -224,7 +224,12 @@ public class WhisperServerConfiguration extends Configuration {
   @Valid
   @NotNull
   @JsonProperty
-  private GenericZkConfig genericZkConfig;
+  private GenericZkConfig callingZkConfig;
+
+  @Valid
+  @NotNull
+  @JsonProperty
+  private GenericZkConfig backupsZkConfig;
 
   @Valid
   @NotNull
@@ -435,8 +440,12 @@ public class WhisperServerConfiguration extends Configuration {
     return zkConfig;
   }
 
-  public GenericZkConfig getGenericZkConfig() {
-    return genericZkConfig;
+  public GenericZkConfig getCallingZkConfig() {
+    return callingZkConfig;
+  }
+
+  public GenericZkConfig getBackupsZkConfig() {
+    return backupsZkConfig;
   }
 
   public RemoteConfigConfiguration getRemoteConfigConfiguration() {

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

@@ -76,6 +76,9 @@ import org.whispersystems.textsecuregcm.auth.RegistrationLockVerificationManager
 import org.whispersystems.textsecuregcm.auth.TurnTokenGenerator;
 import org.whispersystems.textsecuregcm.auth.WebsocketRefreshApplicationEventListener;
 import org.whispersystems.textsecuregcm.auth.grpc.BasicCredentialAuthenticationInterceptor;
+import org.whispersystems.textsecuregcm.backup.BackupAuthManager;
+import org.whispersystems.textsecuregcm.backup.BackupManager;
+import org.whispersystems.textsecuregcm.backup.TusBackupCredentialGenerator;
 import org.whispersystems.textsecuregcm.badges.ConfiguredProfileBadgeConverter;
 import org.whispersystems.textsecuregcm.badges.ResourceBundleLevelTranslator;
 import org.whispersystems.textsecuregcm.captcha.CaptchaChecker;
@@ -92,6 +95,7 @@ import org.whispersystems.textsecuregcm.controllers.ArtController;
 import org.whispersystems.textsecuregcm.controllers.AttachmentControllerV2;
 import org.whispersystems.textsecuregcm.controllers.AttachmentControllerV3;
 import org.whispersystems.textsecuregcm.controllers.AttachmentControllerV4;
+import org.whispersystems.textsecuregcm.controllers.ArchiveController;
 import org.whispersystems.textsecuregcm.controllers.CallLinkController;
 import org.whispersystems.textsecuregcm.controllers.CertificateController;
 import org.whispersystems.textsecuregcm.controllers.ChallengeController;
@@ -137,6 +141,7 @@ import org.whispersystems.textsecuregcm.limits.RateLimitChallengeManager;
 import org.whispersystems.textsecuregcm.limits.RateLimiters;
 import org.whispersystems.textsecuregcm.mappers.CompletionExceptionMapper;
 import org.whispersystems.textsecuregcm.mappers.DeviceLimitExceededExceptionMapper;
+import org.whispersystems.textsecuregcm.mappers.GrpcStatusRuntimeExceptionMapper;
 import org.whispersystems.textsecuregcm.mappers.IOExceptionMapper;
 import org.whispersystems.textsecuregcm.mappers.ImpossiblePhoneNumberExceptionMapper;
 import org.whispersystems.textsecuregcm.mappers.InvalidWebsocketAddressExceptionMapper;
@@ -638,11 +643,18 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         config.getCdnConfiguration().region());
 
     ServerSecretParams zkSecretParams = new ServerSecretParams(config.getZkConfig().serverSecret().value());
-    GenericServerSecretParams genericZkSecretParams = new GenericServerSecretParams(config.getGenericZkConfig().serverSecret().value());
+    GenericServerSecretParams callingGenericZkSecretParams = new GenericServerSecretParams(config.getCallingZkConfig().serverSecret().value());
+    GenericServerSecretParams backupsGenericZkSecretParams = new GenericServerSecretParams(config.getBackupsZkConfig().serverSecret().value());
     ServerZkProfileOperations zkProfileOperations = new ServerZkProfileOperations(zkSecretParams);
     ServerZkAuthOperations zkAuthOperations = new ServerZkAuthOperations(zkSecretParams);
     ServerZkReceiptOperations zkReceiptOperations = new ServerZkReceiptOperations(zkSecretParams);
 
+    TusBackupCredentialGenerator tusBackupCredentialGenerator = new TusBackupCredentialGenerator(config.getTus());
+    BackupAuthManager backupAuthManager = new BackupAuthManager(dynamicConfigurationManager, rateLimiters, accountsManager, backupsGenericZkSecretParams, clock);
+    BackupManager backupManager = new BackupManager(backupsGenericZkSecretParams, tusBackupCredentialGenerator, dynamoDbAsyncClient,
+        config.getDynamoDbTables().getBackups().getTableName(),
+        clock);
+
     AuthFilter<BasicCredentials, AuthenticatedAccount> accountAuthFilter = new BasicCredentialAuthFilter.Builder<AuthenticatedAccount>().setAuthenticator(
         accountAuthenticator).buildAuthFilter();
     AuthFilter<BasicCredentials, DisabledPermittedAuthenticatedAccount> disabledPermittedAccountAuthFilter = new BasicCredentialAuthFilter.Builder<DisabledPermittedAuthenticatedAccount>().setAuthenticator(
@@ -767,8 +779,9 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
         new AttachmentControllerV2(rateLimiters, config.getAwsAttachmentsConfiguration().accessKey().value(), config.getAwsAttachmentsConfiguration().accessSecret().value(), config.getAwsAttachmentsConfiguration().region(), config.getAwsAttachmentsConfiguration().bucket()),
         new AttachmentControllerV3(rateLimiters, gcsAttachmentGenerator),
         new AttachmentControllerV4(rateLimiters, gcsAttachmentGenerator, new TusAttachmentGenerator(config.getTus()), experimentEnrollmentManager),
-        new CallLinkController(rateLimiters, genericZkSecretParams),
-        new CertificateController(new CertificateGenerator(config.getDeliveryCertificate().certificate().value(), config.getDeliveryCertificate().ecPrivateKey(), config.getDeliveryCertificate().expiresDays()), zkAuthOperations, genericZkSecretParams, clock),
+        new ArchiveController(backupAuthManager, backupManager),
+        new CallLinkController(rateLimiters, callingGenericZkSecretParams),
+        new CertificateController(new CertificateGenerator(config.getDeliveryCertificate().certificate().value(), config.getDeliveryCertificate().ecPrivateKey(), config.getDeliveryCertificate().expiresDays()), zkAuthOperations, callingGenericZkSecretParams, clock),
         new ChallengeController(rateLimitChallengeManager),
         new DeviceController(config.getLinkDeviceSecretConfiguration().secret().value(), accountsManager, messagesManager, keys, rateLimiters,
             rateLimitersCluster, config.getMaxDevices(), clock),
@@ -869,6 +882,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     List.of(
         new LoggingUnhandledExceptionMapper(),
         new CompletionExceptionMapper(),
+        new GrpcStatusRuntimeExceptionMapper(),
         new IOExceptionMapper(),
         new RateLimitExceededExceptionMapper(),
         new InvalidWebsocketAddressExceptionMapper(),

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

@@ -0,0 +1,10 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.auth;
+
+import org.whispersystems.textsecuregcm.backup.BackupTier;
+
+public record AuthenticatedBackupUser(byte[] backupId, BackupTier backupTier) {}

service/src/main/java/org/whispersystems/textsecuregcm/backup/BackupAuthManager.java

@@ -0,0 +1,165 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.backup;
+
+import io.grpc.Status;
+import java.security.MessageDigest;
+import java.time.Clock;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
+import org.signal.libsignal.zkgroup.GenericServerSecretParams;
+import org.signal.libsignal.zkgroup.InvalidInputException;
+import org.signal.libsignal.zkgroup.backups.BackupAuthCredentialRequest;
+import org.signal.libsignal.zkgroup.backups.BackupAuthCredentialResponse;
+import org.whispersystems.textsecuregcm.configuration.dynamic.DynamicConfiguration;
+import org.whispersystems.textsecuregcm.controllers.RateLimitExceededException;
+import org.whispersystems.textsecuregcm.limits.RateLimiters;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.AccountsManager;
+import org.whispersystems.textsecuregcm.storage.DynamicConfigurationManager;
+import org.whispersystems.textsecuregcm.util.Util;
+
+/**
+ * Issues ZK backup auth credentials for authenticated accounts
+ * <p>
+ * Authenticated callers can create ZK credentials that contain a blinded backup-id, so that they can later use that
+ * backup id without the verifier learning that the id is associated with this account.
+ * <p>
+ * First use {@link #commitBackupId} to provide a blinded backup-id. This is stored in durable storage. Then the caller
+ * can use {@link #getBackupAuthCredentials} to retrieve credentials that can subsequently be used to make anonymously
+ * authenticated requests against their backup-id.
+ */
+public class BackupAuthManager {
+
+  private static final Duration MAX_REDEMPTION_DURATION = Duration.ofDays(7);
+  final static String BACKUP_EXPERIMENT_NAME = "backup";
+  final static String BACKUP_MEDIA_EXPERIMENT_NAME = "backupMedia";
+
+  private final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager;
+  private final GenericServerSecretParams serverSecretParams;
+  private final Clock clock;
+  private final RateLimiters rateLimiters;
+  private final AccountsManager accountsManager;
+
+  public BackupAuthManager(
+      final DynamicConfigurationManager<DynamicConfiguration> dynamicConfigurationManager,
+      final RateLimiters rateLimiters,
+      final AccountsManager accountsManager,
+      final GenericServerSecretParams serverSecretParams,
+      final Clock clock) {
+    this.dynamicConfigurationManager = dynamicConfigurationManager;
+    this.rateLimiters = rateLimiters;
+    this.accountsManager = accountsManager;
+    this.serverSecretParams = serverSecretParams;
+    this.clock = clock;
+  }
+
+  /**
+   * Store a credential request containing a blinded backup-id for future use.
+   *
+   * @param account                  The account using the backup-id
+   * @param backupAuthCredentialRequest A request containing the blinded backup-id
+   * @return A future that completes when the credentialRequest has been stored
+   * @throws RateLimitExceededException If too many backup-ids have been committed
+   */
+  public CompletableFuture<Void> commitBackupId(final Account account,
+      final BackupAuthCredentialRequest backupAuthCredentialRequest) throws RateLimitExceededException {
+    if (receiptLevel(account).isEmpty()) {
+      throw Status.PERMISSION_DENIED.withDescription("Backups not allowed on account").asRuntimeException();
+    }
+
+    byte[] serializedRequest = backupAuthCredentialRequest.serialize();
+    byte[] existingRequest = account.getBackupCredentialRequest();
+    if (existingRequest != null && MessageDigest.isEqual(serializedRequest, existingRequest)) {
+      // No need to update or enforce rate limits, this is the credential that the user has already
+      // committed to.
+      return CompletableFuture.completedFuture(null);
+    }
+
+    rateLimiters.forDescriptor(RateLimiters.For.SET_BACKUP_ID).validate(account.getUuid());
+
+    return this.accountsManager
+        .updateAsync(account, acc -> acc.setBackupCredentialRequest(serializedRequest))
+        .thenRun(Util.NOOP);
+  }
+
+  public record Credential(BackupAuthCredentialResponse credential, Instant redemptionTime) {}
+
+  /**
+   * Create a credential for every day between redemptionStart and redemptionEnd
+   * <p>
+   * This uses a {@link BackupAuthCredentialRequest} previous stored via {@link this#commitBackupId} to generate the
+   * credentials.
+   *
+   * @param account         The account to create the credentials for
+   * @param redemptionStart The day (must be truncated to a day boundary) the first credential should be valid
+   * @param redemptionEnd   The day (must be truncated to a day boundary) the last credential should be valid
+   * @return Credentials and the day on which they may be redeemed
+   */
+  public CompletableFuture<List<Credential>> getBackupAuthCredentials(
+      final Account account,
+      final Instant redemptionStart,
+      final Instant redemptionEnd) {
+
+    final long receiptLevel = receiptLevel(account).orElseThrow(
+        () -> Status.PERMISSION_DENIED.withDescription("Backups not allowed on account").asRuntimeException());
+
+    final Instant startOfDay = clock.instant().truncatedTo(ChronoUnit.DAYS);
+    if (redemptionStart.isAfter(redemptionEnd) ||
+        redemptionStart.isBefore(startOfDay) ||
+        redemptionEnd.isAfter(startOfDay.plus(MAX_REDEMPTION_DURATION)) ||
+        !redemptionStart.equals(redemptionStart.truncatedTo(ChronoUnit.DAYS)) ||
+        !redemptionEnd.equals(redemptionEnd.truncatedTo(ChronoUnit.DAYS))) {
+
+      throw Status.INVALID_ARGUMENT.withDescription("invalid redemption window").asRuntimeException();
+    }
+
+    // fetch the blinded backup-id the account should have previously committed to
+    final byte[] committedBytes = account.getBackupCredentialRequest();
+    if (committedBytes == null) {
+      throw Status.NOT_FOUND.withDescription("No blinded backup-id has been added to the account").asRuntimeException();
+    }
+
+    try {
+      // create a credential for every day in the requested period
+      final BackupAuthCredentialRequest credentialReq = new BackupAuthCredentialRequest(committedBytes);
+      return CompletableFuture.completedFuture(Stream
+          .iterate(redemptionStart, curr -> curr.plus(Duration.ofDays(1)))
+          .takeWhile(redemptionTime -> !redemptionTime.isAfter(redemptionEnd))
+          .map(redemption -> new Credential(
+              credentialReq.issueCredential(redemption, receiptLevel, serverSecretParams),
+              redemption))
+          .toList());
+    } catch (InvalidInputException e) {
+      throw Status.INTERNAL
+          .withDescription("Could not deserialize stored request credential")
+          .withCause(e)
+          .asRuntimeException();
+    }
+  }
+
+  private Optional<Long> receiptLevel(final Account account) {
+    if (inExperiment(BACKUP_MEDIA_EXPERIMENT_NAME, account)) {
+      return Optional.of(BackupTier.MEDIA.getReceiptLevel());
+    }
+    if (inExperiment(BACKUP_EXPERIMENT_NAME, account)) {
+      return Optional.of(BackupTier.MESSAGES.getReceiptLevel());
+    }
+    return Optional.empty();
+  }
+
+  private boolean inExperiment(final String experimentName, final Account account) {
+    return dynamicConfigurationManager.getConfiguration()
+        .getExperimentEnrollmentConfiguration(experimentName)
+        .map(config -> config.getEnrolledUuids().contains(account.getUuid()))
+        .orElse(false);
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/backup/BackupManager.java

@@ -0,0 +1,391 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.backup;
+
+import io.grpc.Status;
+import io.micrometer.core.instrument.Metrics;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.Clock;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import org.signal.libsignal.protocol.InvalidKeyException;
+import org.signal.libsignal.protocol.ecc.ECPublicKey;
+import org.signal.libsignal.zkgroup.GenericServerSecretParams;
+import org.signal.libsignal.zkgroup.VerificationFailedException;
+import org.signal.libsignal.zkgroup.backups.BackupAuthCredentialPresentation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.auth.AuthenticatedBackupUser;
+import org.whispersystems.textsecuregcm.metrics.MetricsUtil;
+import org.whispersystems.textsecuregcm.util.AttributeValues;
+import org.whispersystems.textsecuregcm.util.ExceptionUtils;
+import org.whispersystems.textsecuregcm.util.Util;
+import software.amazon.awssdk.core.SdkBytes;
+import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
+import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
+import software.amazon.awssdk.services.dynamodb.model.ConditionalCheckFailedException;
+import software.amazon.awssdk.services.dynamodb.model.GetItemRequest;
+import software.amazon.awssdk.services.dynamodb.model.UpdateItemRequest;
+
+public class BackupManager {
+
+  private static final Logger logger = LoggerFactory.getLogger(BackupManager.class);
+
+  static final String MESSAGE_BACKUP_NAME = "messageBackup";
+  private static final int BACKUP_CDN = 3;
+  private static final String ZK_AUTHN_COUNTER_NAME = MetricsUtil.name(BackupManager.class, "authentication");
+  private static final String ZK_AUTHZ_FAILURE_COUNTER_NAME = MetricsUtil.name(BackupManager.class, "authorizationFailure");
+  private static final String SUCCESS_TAG_NAME = "success";
+  private static final String FAILURE_REASON_TAG_NAME = "reason";
+
+  private final GenericServerSecretParams serverSecretParams;
+  private final TusBackupCredentialGenerator tusBackupCredentialGenerator;
+  private final DynamoDbAsyncClient dynamoClient;
+  private final String backupTableName;
+  private final Clock clock;
+
+  // The backups table
+
+  // B: 16 bytes that identifies the backup
+  public static final String KEY_BACKUP_ID_HASH = "U";
+  // N: Time in seconds since epoch of the last backup refresh. This timestamp must be periodically updated to avoid
+  // garbage collection of archive objects.
+  public static final String ATTR_LAST_REFRESH = "R";
+  // N: Time in seconds since epoch of the last backup media refresh. This timestamp can only be updated if the client
+  // has BackupTier.MEDIA, and must be periodically updated to avoid garbage collection of media objects.
+  public static final String ATTR_LAST_MEDIA_REFRESH = "MR";
+  // B: A 32 byte public key that should be used to sign the presentation used to authenticate requests against the
+  // backup-id
+  public static final String ATTR_PUBLIC_KEY = "P";
+  // N: Bytes consumed by this backup
+  public static final String ATTR_MEDIA_BYTES_USED = "MB";
+  // N: Number of media objects in the backup
+  public static final String ATTR_MEDIA_COUNT = "MC";
+  // N: The cdn number where the message backup is stored
+  public static final String ATTR_CDN = "CDN";
+
+  public BackupManager(
+      final GenericServerSecretParams serverSecretParams,
+      final TusBackupCredentialGenerator tusBackupCredentialGenerator,
+      final DynamoDbAsyncClient dynamoClient,
+      final String backupTableName,
+      final Clock clock) {
+    this.serverSecretParams = serverSecretParams;
+    this.dynamoClient = dynamoClient;
+    this.tusBackupCredentialGenerator = tusBackupCredentialGenerator;
+    this.backupTableName = backupTableName;
+    this.clock = clock;
+  }
+
+  /**
+   * Set the public key for the backup-id.
+   * <p>
+   * Once set, calls {@link BackupManager#authenticateBackupUser} can succeed if the presentation is signed with the
+   * private key corresponding to this public key.
+   *
+   * @param presentation a ZK credential presentation that encodes the backupId
+   * @param signature    the signature of the presentation
+   * @param publicKey    the public key of a key-pair that the presentation must be signed with
+   */
+  public CompletableFuture<Void> setPublicKey(
+      final BackupAuthCredentialPresentation presentation,
+      final byte[] signature,
+      final ECPublicKey publicKey) {
+
+    // Note: this is a special case where we can't validate the presentation signature against the stored public key
+    // because we are currently setting it. We check against the provided public key, but we must also verify that
+    // there isn't an existing, different stored public key for the backup-id (verified with a condition expression)
+    final BackupTier backupTier = verifySignatureAndCheckPresentation(presentation, signature, publicKey);
+    if (backupTier.compareTo(BackupTier.MESSAGES) < 0) {
+      Metrics.counter(ZK_AUTHZ_FAILURE_COUNTER_NAME).increment();
+      throw Status.PERMISSION_DENIED
+          .withDescription("credential does not support setting public key")
+          .asRuntimeException();
+    }
+
+    final byte[] hashedBackupId = hashedBackupId(presentation.getBackupId());
+    return dynamoClient.updateItem(UpdateItemRequest.builder()
+            .tableName(backupTableName)
+            .key(Map.of(KEY_BACKUP_ID_HASH, AttributeValues.b(hashedBackupId)))
+            .updateExpression("SET #publicKey = :publicKey")
+            .expressionAttributeNames(Map.of("#publicKey", ATTR_PUBLIC_KEY))
+            .expressionAttributeValues(Map.of(":publicKey", AttributeValues.b(publicKey.serialize())))
+            .conditionExpression("attribute_not_exists(#publicKey) OR #publicKey = :publicKey")
+            .build())
+        .exceptionally(throwable -> {
+          // There was already a row for this backup-id and it contained a different publicKey
+          if (ExceptionUtils.unwrap(throwable) instanceof ConditionalCheckFailedException) {
+            Metrics.counter(ZK_AUTHN_COUNTER_NAME,
+                    SUCCESS_TAG_NAME, String.valueOf(false),
+                    FAILURE_REASON_TAG_NAME, "public_key_conflict")
+                .increment();
+            throw Status.UNAUTHENTICATED
+                .withDescription("public key does not match existing public key for the backup-id")
+                .asRuntimeException();
+          }
+          throw ExceptionUtils.wrap(throwable);
+        })
+        .thenRun(Util.NOOP);
+  }
+
+
+  /**
+   * Create a form that may be used to upload a backup file for the backupId encoded in the presentation.
+   * <p>
+   * If successful, this also updates the TTL of the backup.
+   *
+   * @param backupUser an already ZK authenticated backup user
+   * @return the upload form
+   */
+  public CompletableFuture<MessageBackupUploadDescriptor> createMessageBackupUploadDescriptor(
+      final AuthenticatedBackupUser backupUser) {
+    final byte[] hashedBackupId = hashedBackupId(backupUser);
+    final String encodedBackupId = encodeForCdn(hashedBackupId);
+
+    final long refreshTimeSecs = clock.instant().getEpochSecond();
+
+    final List<String> updates = new ArrayList<>(List.of("#cdn = :cdn", "#lastRefresh = :expiration"));
+    final Map<String, String> expressionAttributeNames = new HashMap<>(Map.of(
+        "#cdn", ATTR_CDN,
+        "#lastRefresh", ATTR_LAST_REFRESH));
+    if (backupUser.backupTier().compareTo(BackupTier.MEDIA) >= 0) {
+      updates.add("#lastMediaRefresh = :expiration");
+      expressionAttributeNames.put("#lastMediaRefresh", ATTR_LAST_MEDIA_REFRESH);
+    }
+
+    // this could race with concurrent updates, but the only effect would be last-writer-wins on the timestamp
+    return dynamoClient.updateItem(UpdateItemRequest.builder()
+            .tableName(backupTableName)
+            .key(Map.of(KEY_BACKUP_ID_HASH, AttributeValues.b(hashedBackupId)))
+            .updateExpression("SET %s".formatted(String.join(",", updates)))
+            .expressionAttributeNames(expressionAttributeNames)
+            .expressionAttributeValues(Map.of(
+                ":cdn", AttributeValues.n(BACKUP_CDN),
+                ":expiration", AttributeValues.n(refreshTimeSecs)))
+            .build())
+        .thenApply(result -> tusBackupCredentialGenerator.generateUpload(encodedBackupId, MESSAGE_BACKUP_NAME));
+  }
+
+  /**
+   * Update the last update timestamps for the backupId in the presentation
+   *
+   * @param backupUser an already ZK authenticated backup user
+   */
+  public CompletableFuture<Void> ttlRefresh(final AuthenticatedBackupUser backupUser) {
+    if (backupUser.backupTier().compareTo(BackupTier.MESSAGES) < 0) {
+      Metrics.counter(ZK_AUTHZ_FAILURE_COUNTER_NAME).increment();
+      throw Status.PERMISSION_DENIED
+          .withDescription("credential does not support ttl operation")
+          .asRuntimeException();
+    }
+    final long refreshTimeSecs = clock.instant().getEpochSecond();
+    // update message backup TTL
+    final List<String> updates = new ArrayList<>(Collections.singletonList("#lastRefresh = :expiration"));
+    final Map<String, String> expressionAttributeNames = new HashMap<>(Map.of("#lastRefresh", ATTR_LAST_REFRESH));
+    if (backupUser.backupTier().compareTo(BackupTier.MEDIA) >= 0) {
+      // update media TTL
+      expressionAttributeNames.put("#lastMediaRefresh", ATTR_LAST_MEDIA_REFRESH);
+      updates.add("#lastMediaRefresh = :expiration");
+    }
+    return dynamoClient.updateItem(UpdateItemRequest.builder()
+            .tableName(backupTableName)
+            .key(Map.of(KEY_BACKUP_ID_HASH, AttributeValues.b(hashedBackupId(backupUser))))
+            .updateExpression("SET %s".formatted(String.join(",", updates)))
+            .expressionAttributeNames(expressionAttributeNames)
+            .expressionAttributeValues(Map.of(":expiration", AttributeValues.n(refreshTimeSecs)))
+            .build())
+        .thenRun(Util.NOOP);
+  }
+
+  public record BackupInfo(int cdn, String backupSubdir, String messageBackupKey, Optional<Long> mediaUsedSpace) {}
+
+  /**
+   * Retrieve information about the existing backup
+   *
+   * @param backupUser an already ZK authenticated backup user
+   * @return Information about the existing backup
+   */
+  public CompletableFuture<BackupInfo> backupInfo(final AuthenticatedBackupUser backupUser) {
+    if (backupUser.backupTier().compareTo(BackupTier.MESSAGES) < 0) {
+      Metrics.counter(ZK_AUTHZ_FAILURE_COUNTER_NAME).increment();
+      throw Status.PERMISSION_DENIED.withDescription("credential does not support info operation")
+          .asRuntimeException();
+    }
+    return backupInfoHelper(backupUser);
+  }
+
+  private CompletableFuture<BackupInfo> backupInfoHelper(final AuthenticatedBackupUser backupUser) {
+    return dynamoClient.getItem(GetItemRequest.builder()
+            .tableName(backupTableName)
+            .key(Map.of(KEY_BACKUP_ID_HASH, AttributeValues.b(hashedBackupId(backupUser))))
+            .projectionExpression("#cdn,#bytesUsed")
+            .expressionAttributeNames(Map.of("#cdn", ATTR_CDN, "#bytesUsed", ATTR_MEDIA_BYTES_USED))
+            .build())
+        .thenApply(response -> {
+          if (!response.hasItem()) {
+            throw Status.NOT_FOUND.withDescription("Backup not found").asRuntimeException();
+          }
+          final int cdn = AttributeValues.get(response.item(), ATTR_CDN)
+              .map(AttributeValue::n)
+              .map(Integer::parseInt)
+              .orElseThrow(() -> Status.NOT_FOUND.withDescription("Stored backup not found").asRuntimeException());
+
+          final Optional<Long> mediaUsed = AttributeValues.get(response.item(), ATTR_MEDIA_BYTES_USED)
+              .map(AttributeValue::n)
+              .map(Long::parseLong);
+
+          return new BackupInfo(cdn, encodeForCdn(hashedBackupId(backupUser)), MESSAGE_BACKUP_NAME, mediaUsed);
+        });
+  }
+
+  /**
+   * Generate credentials that can be used to read from the backup CDN
+   *
+   * @param backupUser an already ZK authenticated backup user
+   * @return A map of headers to include with CDN requests
+   */
+  public Map<String, String> generateReadAuth(final AuthenticatedBackupUser backupUser) {
+    if (backupUser.backupTier().compareTo(BackupTier.MESSAGES) < 0) {
+      Metrics.counter(ZK_AUTHZ_FAILURE_COUNTER_NAME).increment();
+      throw Status.PERMISSION_DENIED
+          .withDescription("credential does not support read auth operation")
+          .asRuntimeException();
+
+    }
+    final String encodedBackupId = encodeForCdn(hashedBackupId(backupUser));
+    return tusBackupCredentialGenerator.readHeaders(encodedBackupId);
+  }
+
+  /**
+   * Authenticate the ZK anonymous backup credential's presentation
+   * <p>
+   * This validates:
+   * <li> The presentation was for a credential issued by the server </li>
+   * <li> The credential is in its redemption window </li>
+   * <li> The backup-id matches a previously committed blinded backup-id and server issued receipt level </li>
+   * <li> The signature of the credential matches an existing publicKey associated with this backup-id </li>
+   *
+   * @param presentation A {@link BackupAuthCredentialPresentation}
+   * @param signature    An XEd25519 signature of the presentation bytes
+   * @return On authentication success, the authenticated backup-id and backup-tier encoded in the presentation
+   */
+  public CompletableFuture<AuthenticatedBackupUser> authenticateBackupUser(
+      final BackupAuthCredentialPresentation presentation,
+      final byte[] signature) {
+    final byte[] hashedBackupId = hashedBackupId(presentation.getBackupId());
+    return dynamoClient.getItem(GetItemRequest.builder()
+            .tableName(backupTableName)
+            .key(Map.of(KEY_BACKUP_ID_HASH, AttributeValues.b(hashedBackupId)))
+            .projectionExpression("#publicKey")
+            .expressionAttributeNames(Map.of("#publicKey", ATTR_PUBLIC_KEY))
+            .build())
+        .thenApply(response -> {
+          if (!response.hasItem()) {
+            Metrics.counter(ZK_AUTHN_COUNTER_NAME,
+                    SUCCESS_TAG_NAME, String.valueOf(false),
+                    FAILURE_REASON_TAG_NAME, "missing_public_key")
+                .increment();
+            throw Status.NOT_FOUND.withDescription("Backup not found").asRuntimeException();
+          }
+          final byte[] publicKeyBytes = AttributeValues.get(response.item(), ATTR_PUBLIC_KEY)
+              .map(AttributeValue::b)
+              .map(SdkBytes::asByteArray)
+              .orElseThrow(() -> Status.INTERNAL
+                  .withDescription("Stored backup missing public key")
+                  .asRuntimeException());
+          try {
+            final ECPublicKey publicKey = new ECPublicKey(publicKeyBytes);
+            return new AuthenticatedBackupUser(
+                presentation.getBackupId(),
+                verifySignatureAndCheckPresentation(presentation, signature, publicKey));
+          } catch (InvalidKeyException e) {
+            Metrics.counter(ZK_AUTHN_COUNTER_NAME,
+                    SUCCESS_TAG_NAME, String.valueOf(false),
+                    FAILURE_REASON_TAG_NAME, "invalid_public_key")
+                .increment();
+            logger.error("Invalid publicKey for backupId hash {}",
+                HexFormat.of().formatHex(hashedBackupId), e);
+            throw Status.INTERNAL
+                .withCause(e)
+                .withDescription("Could not deserialize stored public key")
+                .asRuntimeException();
+          }
+        })
+        .thenApply(result -> {
+          Metrics.counter(ZK_AUTHN_COUNTER_NAME, SUCCESS_TAG_NAME, String.valueOf(true)).increment();
+          return result;
+        });
+  }
+
+
+  /**
+   * Verify the presentation and return the extracted backup tier
+   *
+   * @param presentation A ZK credential presentation that encodes the backupId and the receipt level of the requester
+   * @return The backup tier this presentation supports
+   */
+  private BackupTier verifySignatureAndCheckPresentation(
+      final BackupAuthCredentialPresentation presentation,
+      final byte[] signature,
+      final ECPublicKey publicKey) {
+    if (!publicKey.verifySignature(presentation.serialize(), signature)) {
+      Metrics.counter(ZK_AUTHN_COUNTER_NAME,
+              SUCCESS_TAG_NAME, String.valueOf(false),
+              FAILURE_REASON_TAG_NAME, "signature_validation")
+          .increment();
+      throw Status.UNAUTHENTICATED
+          .withDescription("backup auth credential presentation signature verification failed")
+          .asRuntimeException();
+    }
+    try {
+      presentation.verify(clock.instant(), serverSecretParams);
+    } catch (VerificationFailedException e) {
+      Metrics.counter(ZK_AUTHN_COUNTER_NAME,
+              SUCCESS_TAG_NAME, String.valueOf(false),
+              FAILURE_REASON_TAG_NAME, "presentation_verification")
+          .increment();
+      throw Status.UNAUTHENTICATED
+          .withDescription("backup auth credential presentation verification failed")
+          .withCause(e)
+          .asRuntimeException();
+    }
+
+    return BackupTier
+        .fromReceiptLevel(presentation.getReceiptLevel())
+        .orElseThrow(() -> {
+          Metrics.counter(ZK_AUTHN_COUNTER_NAME,
+                  SUCCESS_TAG_NAME, String.valueOf(false),
+                  FAILURE_REASON_TAG_NAME, "invalid_receipt_level")
+              .increment();
+          return Status.PERMISSION_DENIED.withDescription("invalid receipt level").asRuntimeException();
+        });
+  }
+
+  private static byte[] hashedBackupId(final AuthenticatedBackupUser backupId) {
+    return hashedBackupId(backupId.backupId());
+  }
+
+  private static byte[] hashedBackupId(final byte[] backupId) {
+    try {
+      return Arrays.copyOf(MessageDigest.getInstance("SHA-256").digest(backupId), 16);
+    } catch (NoSuchAlgorithmException e) {
+      throw new AssertionError(e);
+    }
+  }
+
+  private static String encodeForCdn(final byte[] bytes) {
+    return Base64.getUrlEncoder().encodeToString(bytes);
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/backup/BackupTier.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.backup;
+
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public enum BackupTier {
+  NONE(0),
+  MESSAGES(10),
+  MEDIA(20);
+
+  private static Map<Long, BackupTier> LOOKUP = Arrays.stream(BackupTier.values())
+      .collect(Collectors.toMap(BackupTier::getReceiptLevel, Function.identity()));
+  private long receiptLevel;
+
+  private BackupTier(long receiptLevel) {
+    this.receiptLevel = receiptLevel;
+  }
+
+  long getReceiptLevel() {
+    return receiptLevel;
+  }
+
+  static Optional<BackupTier> fromReceiptLevel(long receiptLevel) {
+    return Optional.ofNullable(LOOKUP.get(receiptLevel));
+  }
+}

service/src/main/java/org/whispersystems/textsecuregcm/backup/MessageBackupUploadDescriptor.java

@@ -0,0 +1,14 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.backup;
+
+import java.util.Map;
+
+public record MessageBackupUploadDescriptor(
+    int cdn,
+    String key,
+    Map<String, String> headers,
+    String signedUploadLocation) {}

service/src/main/java/org/whispersystems/textsecuregcm/backup/TusBackupCredentialGenerator.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.backup;
+
+import org.apache.http.HttpHeaders;
+import org.whispersystems.textsecuregcm.attachments.TusConfiguration;
+import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentials;
+import org.whispersystems.textsecuregcm.auth.ExternalServiceCredentialsGenerator;
+import org.whispersystems.textsecuregcm.util.HeaderUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Clock;
+import java.util.Base64;
+import java.util.Map;
+
+public class TusBackupCredentialGenerator {
+
+  private static final int BACKUP_CDN = 3;
+
+  private static String READ_PERMISSION = "read";
+  private static String WRITE_PERMISSION = "write";
+  private static String CDN_PATH = "backups";
+  private static String PERMISSION_SEPARATOR = "$";
+
+  // Write entities will be of the form 'write$backups/<string>
+  private static final String WRITE_ENTITY_PREFIX = String.format("%s%s%s/", WRITE_PERMISSION, PERMISSION_SEPARATOR,
+      CDN_PATH);
+  // Read entities will be of the form 'read$backups/<string>
+  private static final String READ_ENTITY_PREFIX = String.format("%s%s%s/", READ_PERMISSION, PERMISSION_SEPARATOR,
+      CDN_PATH);
+
+  private final ExternalServiceCredentialsGenerator credentialsGenerator;
+  private final String tusUri;
+
+  public TusBackupCredentialGenerator(final TusConfiguration cfg) {
+    this.tusUri = cfg.uploadUri();
+    this.credentialsGenerator = credentialsGenerator(Clock.systemUTC(), cfg);
+  }
+
+  private static ExternalServiceCredentialsGenerator credentialsGenerator(final Clock clock,
+      final TusConfiguration cfg) {
+    return ExternalServiceCredentialsGenerator
+        .builder(cfg.userAuthenticationTokenSharedSecret())
+        .prependUsername(false)
+        .withClock(clock)
+        .build();
+  }
+
+  public MessageBackupUploadDescriptor generateUpload(final String hashedBackupId, final String objectName) {
+    if (hashedBackupId.isBlank() || objectName.isBlank()) {
+      throw new IllegalArgumentException("Upload descriptors must have non-empty keys");
+    }
+    final String key = "%s/%s".formatted(hashedBackupId, objectName);
+    final String entity = WRITE_ENTITY_PREFIX + key;
+    final ExternalServiceCredentials credentials = credentialsGenerator.generateFor(entity);
+    final String b64Key = Base64.getEncoder().encodeToString(key.getBytes(StandardCharsets.UTF_8));
+    final Map<String, String> headers = Map.of(
+        HttpHeaders.AUTHORIZATION, HeaderUtils.basicAuthHeader(credentials),
+        "Upload-Metadata", String.format("filename %s", b64Key));
+
+    return new MessageBackupUploadDescriptor(
+        BACKUP_CDN,
+        key,
+        headers,
+        tusUri + "/" + CDN_PATH);
+  }
+
+  public Map<String, String> readHeaders(final String hashedBackupId) {
+    if (hashedBackupId.isBlank()) {
+      throw new IllegalArgumentException("Backup subdir name must be non-empty");
+    }
+    final ExternalServiceCredentials credentials = credentialsGenerator.generateFor(
+        READ_ENTITY_PREFIX + hashedBackupId);
+    return Map.of(HttpHeaders.AUTHORIZATION, HeaderUtils.basicAuthHeader(credentials));
+  }
+}

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

@@ -47,6 +47,8 @@ public class DynamoDbTables {
   }
 
   private final AccountsTableConfiguration accounts;
+
+  private final Table backups;
   private final Table clientReleases;
   private final Table deletedAccounts;
   private final Table deletedAccountsLock;
@@ -68,6 +70,7 @@ public class DynamoDbTables {
 
   public DynamoDbTables(
       @JsonProperty("accounts") final AccountsTableConfiguration accounts,
+      @JsonProperty("backups") final Table backups,
       @JsonProperty("clientReleases") final Table clientReleases,
       @JsonProperty("deletedAccounts") final Table deletedAccounts,
       @JsonProperty("deletedAccountsLock") final Table deletedAccountsLock,
@@ -88,6 +91,7 @@ public class DynamoDbTables {
       @JsonProperty("verificationSessions") final Table verificationSessions) {
 
     this.accounts = accounts;
+    this.backups = backups;
     this.clientReleases = clientReleases;
     this.deletedAccounts = deletedAccounts;
     this.deletedAccountsLock = deletedAccountsLock;
@@ -114,6 +118,12 @@ public class DynamoDbTables {
     return accounts;
   }
 
+  @NotNull
+  @Valid
+  public Table getBackups() {
+    return backups;
+  }
+
   @NotNull
   @Valid
   public Table getClientReleases() {

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

@@ -0,0 +1,365 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.controllers;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.dropwizard.auth.Auth;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
+import io.swagger.v3.oas.annotations.media.Content;
+import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.responses.ApiResponse;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+import javax.annotation.Nullable;
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+import javax.ws.rs.BadRequestException;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.HeaderParam;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.core.MediaType;
+import org.signal.libsignal.protocol.ecc.ECPublicKey;
+import org.signal.libsignal.zkgroup.InvalidInputException;
+import org.signal.libsignal.zkgroup.backups.BackupAuthCredentialPresentation;
+import org.signal.libsignal.zkgroup.backups.BackupAuthCredentialRequest;
+import org.whispersystems.textsecuregcm.auth.AuthenticatedAccount;
+import org.whispersystems.textsecuregcm.backup.BackupAuthManager;
+import org.whispersystems.textsecuregcm.backup.BackupManager;
+import org.whispersystems.textsecuregcm.util.BackupAuthCredentialAdapter;
+import org.whispersystems.textsecuregcm.util.ECPublicKeyAdapter;
+
+@Path("/v1/archives")
+@Tag(name = "Archive")
+public class ArchiveController {
+
+  public final static String X_SIGNAL_ZK_AUTH = "X-Signal-ZK-Auth";
+  public final static String X_SIGNAL_ZK_AUTH_SIGNATURE = "X-Signal-ZK-Auth-Signature";
+
+  private final BackupAuthManager backupAuthManager;
+  private final BackupManager backupManager;
+
+  public ArchiveController(
+      final BackupAuthManager backupAuthManager,
+      final BackupManager backupManager) {
+    this.backupAuthManager = backupAuthManager;
+    this.backupManager = backupManager;
+  }
+
+  public record SetBackupIdRequest(
+      @Schema(description = """
+          A BackupAuthCredentialRequest containing a blinded encrypted backup-id, encoded as a base64 string
+          """, implementation = String.class)
+      @JsonDeserialize(using = BackupAuthCredentialAdapter.CredentialRequestDeserializer.class)
+      @JsonSerialize(using = BackupAuthCredentialAdapter.CredentialRequestSerializer.class)
+      @NotNull BackupAuthCredentialRequest backupAuthCredentialRequest) {}
+
+  @PUT
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Produces(MediaType.APPLICATION_JSON)
+  @Path("/backupid")
+  @Operation(
+      summary = "Set backup id",
+      description = """
+          Set a (blinded) backup-id for the account. Each account may have a single active backup-id that can be used
+          to store and retrieve backups. Once the backup-id is set, BackupAuthCredentials can be generated
+          using /v1/archives/auth.
+
+          The blinded backup-id and the key-pair used to blind it should be derived from a recoverable secret.
+          """)
+  @ApiResponse(responseCode = "204", description = "The backup-id was set")
+  @ApiResponse(responseCode = "400", description = "The provided backup auth credential request was invalid")
+  @ApiResponse(responseCode = "429", description = "Rate limited. Too many attempts to change the backup-id have been made")
+  public CompletionStage<Void> setBackupId(
+      @Auth final AuthenticatedAccount account,
+      @Valid @NotNull final SetBackupIdRequest setBackupIdRequest) throws RateLimitExceededException {
+    return this.backupAuthManager.commitBackupId(account.getAccount(), setBackupIdRequest.backupAuthCredentialRequest);
+  }
+
+  public record BackupAuthCredentialsResponse(
+      @Schema(description = "A list of BackupAuthCredentials and their validity periods")
+      List<BackupAuthCredential> credentials) {
+
+    public record BackupAuthCredential(
+        @Schema(description = "A base64 encoded BackupAuthCredential")
+        byte[] credential,
+        @Schema(description = "The day on which this credential is valid. Seconds since epoch truncated to day boundary")
+        long redemptionTime) {}
+  }
+
+  @GET
+  @Produces(MediaType.APPLICATION_JSON)
+  @Path("/auth")
+  @Operation(
+      summary = "Fetch ZK credentials ",
+      description = """
+          After setting a blinded backup-id with PUT /v1/archives/, this fetches credentials that can be used to perform
+          operations against that backup-id. Clients may (and should) request up to 7 days of credentials at a time.
+                    
+          The redemptionStart and redemptionEnd seconds must be UTC day aligned, and must not span more than 7 days.
+          """)
+  @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = BackupAuthCredentialsResponse.class)))
+  @ApiResponse(responseCode = "400", description = "The start/end did not meet alignment/duration requirements")
+  @ApiResponse(responseCode = "404", description = "Could not find an existing blinded backup id")
+  @ApiResponse(responseCode = "429", description = "Rate limited.")
+  public CompletionStage<BackupAuthCredentialsResponse> getBackupZKCredentials(
+      @Auth AuthenticatedAccount auth,
+      @NotNull @QueryParam("redemptionStartSeconds") Integer startSeconds,
+      @NotNull @QueryParam("redemptionEndSeconds") Integer endSeconds) {
+
+    return this.backupAuthManager.getBackupAuthCredentials(
+            auth.getAccount(),
+            Instant.ofEpochSecond(startSeconds), Instant.ofEpochSecond(endSeconds))
+        .thenApply(creds -> new BackupAuthCredentialsResponse(creds.stream()
+            .map(cred -> new BackupAuthCredentialsResponse.BackupAuthCredential(
+                cred.credential().serialize(),
+                cred.redemptionTime().getEpochSecond()))
+            .toList()));
+  }
+
+
+  /**
+   * API annotation for endpoints that take anonymous auth. All anonymous endpoints
+   * <li> 400 if regular auth is used by accident </li>
+   * <li> 401 if the anonymous auth invalid </li>
+   * <li> 403 if the anonymous credential does not have sufficient permissions </li>
+   */
+  @Target(ElementType.METHOD)
+  @Retention(RetentionPolicy.RUNTIME)
+  @ApiResponse(
+      responseCode = "403",
+      description = "Forbidden. The request had insufficient permissions to perform the requested action")
+  @ApiResponse(responseCode = "401", description = "The provided backup auth credential presentation could not be verified")
+  @ApiResponse(responseCode = "400", description = "Bad arguments. The request may have been made on an authenticated channel")
+  @interface ApiResponseZkAuth {}
+
+  public record BackupAuthCredentialPresentationHeader(BackupAuthCredentialPresentation presentation) {
+
+    private static final String DESCRIPTION = "Presentation of a ZK backup auth credential acquired from /v1/archives/auth as a base64 encoded string";
+
+    public BackupAuthCredentialPresentationHeader(final String header) {
+      this(deserialize(header));
+    }
+
+    private static BackupAuthCredentialPresentation deserialize(final String base64Presentation) {
+      byte[] bytes = Base64.getDecoder().decode(base64Presentation);
+      try {
+        return new BackupAuthCredentialPresentation(bytes);
+      } catch (InvalidInputException e) {
+        throw new IllegalArgumentException(e);
+      }
+    }
+  }
+
+  public record BackupAuthCredentialPresentationSignature(byte[] signature) {
+
+    private static final String DESCRIPTION = "Signature of the ZK auth credential's presentation as a base64 encoded string";
+
+    public BackupAuthCredentialPresentationSignature(final String header) {
+      this(Base64.getDecoder().decode(header));
+    }
+  }
+
+  public record ReadAuthResponse(
+      @Schema(description = "Auth headers to include with cdn read requests") Map<String, String> headers) {}
+
+  @GET
+  @Path("/auth/read")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Operation(
+      summary = "Get CDN read credentials",
+      description = "Retrieve credentials used to read objects stored on the backup cdn")
+  @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = ReadAuthResponse.class)))
+  @ApiResponse(responseCode = "429", description = "Rate limited.")
+  @ApiResponseZkAuth
+  public CompletionStage<ReadAuthResponse> readAuth(
+      @Auth final Optional<AuthenticatedAccount> account,
+
+      @Parameter(description = BackupAuthCredentialPresentationHeader.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH) final ArchiveController.BackupAuthCredentialPresentationHeader presentation,
+
+      @Parameter(description = BackupAuthCredentialPresentationSignature.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH_SIGNATURE) final BackupAuthCredentialPresentationSignature signature) {
+    if (account.isPresent()) {
+      throw new BadRequestException("must not use authenticated connection for anonymous operations");
+    }
+    return backupManager.authenticateBackupUser(presentation.presentation, signature.signature)
+        .thenApply(backupManager::generateReadAuth)
+        .thenApply(ReadAuthResponse::new);
+  }
+
+  public record BackupInfoResponse(
+      @Schema(description = "If present, the CDN type where the message backup is stored")
+      int cdn,
+
+      @Schema(description = "If present, the directory of your backup data on the cdn.")
+      String backupDir,
+
+      @Schema(description = "If present, the name of the most recent message backup on the cdn. The backup is at /backupDir/backupName")
+      String backupName,
+
+      @Nullable
+      @Schema(description = "The amount of space used to store media")
+      Long usedSpace) {}
+
+  @GET
+  @Produces(MediaType.APPLICATION_JSON)
+  @Operation(
+      summary = "Fetch backup info",
+      description = "Retrieve information about the currently stored backup")
+  @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = BackupInfoResponse.class)))
+  @ApiResponse(responseCode = "404", description = "No existing backups found")
+  @ApiResponse(responseCode = "429", description = "Rate limited.")
+  @ApiResponseZkAuth
+  public CompletionStage<BackupInfoResponse> backupInfo(
+      @Auth final Optional<AuthenticatedAccount> account,
+
+      @Parameter(description = BackupAuthCredentialPresentationHeader.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH) final BackupAuthCredentialPresentationHeader presentation,
+
+      @Parameter(description = BackupAuthCredentialPresentationSignature.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH_SIGNATURE) final BackupAuthCredentialPresentationSignature signature) {
+    if (account.isPresent()) {
+      throw new BadRequestException("must not use authenticated connection for anonymous operations");
+    }
+
+    return backupManager.authenticateBackupUser(presentation.presentation, signature.signature)
+        .thenCompose(backupManager::backupInfo)
+        .thenApply(backupInfo -> new BackupInfoResponse(
+            backupInfo.cdn(),
+            backupInfo.backupSubdir(),
+            backupInfo.messageBackupKey(),
+            backupInfo.mediaUsedSpace().orElse(null)));
+  }
+
+  public record SetPublicKeyRequest(
+      @JsonSerialize(using = ECPublicKeyAdapter.Serializer.class)
+      @JsonDeserialize(using = ECPublicKeyAdapter.Deserializer.class)
+      @Schema(type = "string", description = "The public key, serialized in libsignal's elliptic-curve public key format and then base64-encoded.")
+      ECPublicKey backupIdPublicKey) {}
+
+  @PUT
+  @Path("/keys")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Consumes(MediaType.APPLICATION_JSON)
+  @Operation(
+      summary = "Set public key",
+      description = """
+          Permanently set the public key of an ED25519 key-pair for the backup-id. All requests that provide a anonymous
+          BackupAuthCredentialPresentation (including this one!) must also sign the presentation with the private key 
+          corresponding to the provided public key.
+          """)
+  @ApiResponseZkAuth
+  @ApiResponse(responseCode = "204", description = "The public key was set")
+  @ApiResponse(responseCode = "429", description = "Rate limited.")
+  public CompletionStage<Void> setPublicKey(
+      @Auth final Optional<AuthenticatedAccount> account,
+
+      @Parameter(description = BackupAuthCredentialPresentationHeader.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH) final ArchiveController.BackupAuthCredentialPresentationHeader presentation,
+
+      @Parameter(description = BackupAuthCredentialPresentationSignature.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH_SIGNATURE) final BackupAuthCredentialPresentationSignature signature,
+
+      @NotNull SetPublicKeyRequest setPublicKeyRequest) {
+    return backupManager.setPublicKey(
+        presentation.presentation, signature.signature,
+        setPublicKeyRequest.backupIdPublicKey);
+  }
+
+
+  public record MessageBackupResponse(
+      @Schema(description = "Indicates the CDN type. 3 indicates resumable uploads using TUS")
+      int cdn,
+      @Schema(description = "The location within the specified cdn where the finished upload can be found.")
+      String key,
+      @Schema(description = "A map of headers to include with all upload requests. Potentially contains time-limited upload credentials")
+      Map<String, String> headers,
+      @Schema(description = "The URL to upload to with the appropriate protocol")
+      String signedUploadLocation) {}
+
+  @GET
+  @Path("/upload/form")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Operation(
+      summary = "Fetch message backup upload form",
+      description = "Retrieve an upload form that can be used to perform a resumable upload of a message backup.")
+  @ApiResponse(responseCode = "200", content = @Content(schema = @Schema(implementation = MessageBackupResponse.class)))
+  @ApiResponse(responseCode = "429", description = "Rate limited.")
+  @ApiResponseZkAuth
+  public CompletionStage<MessageBackupResponse> backup(
+      @Auth final Optional<AuthenticatedAccount> account,
+
+      @Parameter(description = BackupAuthCredentialPresentationHeader.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH) final ArchiveController.BackupAuthCredentialPresentationHeader presentation,
+
+      @Parameter(description = BackupAuthCredentialPresentationSignature.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH_SIGNATURE) final BackupAuthCredentialPresentationSignature signature) {
+    if (account.isPresent()) {
+      throw new BadRequestException("must not use authenticated connection for anonymous operations");
+    }
+    return backupManager.authenticateBackupUser(presentation.presentation, signature.signature)
+        .thenCompose(backupManager::createMessageBackupUploadDescriptor)
+        .thenApply(result -> new MessageBackupResponse(
+            result.cdn(),
+            result.key(),
+            result.headers(),
+            result.signedUploadLocation()));
+  }
+
+
+  @POST
+  @Produces(MediaType.APPLICATION_JSON)
+  @Operation(
+      summary = "Refresh backup",
+      description = """
+          Indicate that this backup is still active. Clients must periodically upload new backups or perform a refresh
+          via a POST request. If a backup is not refreshed, after 30 days it may be deleted.
+          """)
+  @ApiResponse(responseCode = "204", description = "The backup was successfully refreshed")
+  @ApiResponse(responseCode = "429", description = "Rate limited.")
+  @ApiResponseZkAuth
+  public CompletionStage<Void> refresh(
+      @Auth final Optional<AuthenticatedAccount> account,
+
+      @Parameter(description = BackupAuthCredentialPresentationHeader.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH) final BackupAuthCredentialPresentationHeader presentation,
+
+      @Parameter(description = BackupAuthCredentialPresentationSignature.DESCRIPTION, schema = @Schema(implementation = String.class))
+      @NotNull
+      @HeaderParam(X_SIGNAL_ZK_AUTH_SIGNATURE) final BackupAuthCredentialPresentationSignature signature) {
+    if (account.isPresent()) {
+      throw new BadRequestException("must not use authenticated connection for anonymous operations");
+    }
+    return backupManager
+        .authenticateBackupUser(presentation.presentation, signature.signature)
+        .thenCompose(backupManager::ttlRefresh);
+  }
+}

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

@@ -46,6 +46,7 @@ public class RateLimiters extends BaseRateLimiters<RateLimiters.For> {
     RATE_LIMIT_RESET("rateLimitReset", true, new RateLimiterConfig(2, Duration.ofHours(12))),
     RECAPTCHA_CHALLENGE_ATTEMPT("recaptchaChallengeAttempt", true, new RateLimiterConfig(10, Duration.ofMinutes(144))),
     RECAPTCHA_CHALLENGE_SUCCESS("recaptchaChallengeSuccess", true, new RateLimiterConfig(2, Duration.ofHours(12))),
+    SET_BACKUP_ID("setBackupId", true, new RateLimiterConfig(2, Duration.ofDays(7))),
     PUSH_CHALLENGE_ATTEMPT("pushChallengeAttempt", true, new RateLimiterConfig(10, Duration.ofMinutes(144))),
     PUSH_CHALLENGE_SUCCESS("pushChallengeSuccess", true, new RateLimiterConfig(2, Duration.ofHours(12))),
     CREATE_CALL_LINK("createCallLink", false, new RateLimiterConfig(100, Duration.ofMinutes(15))),

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.mappers;
+
+import io.dropwizard.jersey.errors.ErrorMessage;
+import io.grpc.StatusRuntimeException;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.ext.ExceptionMapper;
+import javax.ws.rs.ext.Provider;
+
+@Provider
+public class GrpcStatusRuntimeExceptionMapper implements ExceptionMapper<StatusRuntimeException> {
+
+  @Override
+  public Response toResponse(final StatusRuntimeException exception) {
+    int httpCode = switch (exception.getStatus().getCode()) {
+      case OK -> 200;
+      case INVALID_ARGUMENT, FAILED_PRECONDITION, OUT_OF_RANGE -> 400;
+      case UNAUTHENTICATED -> 401;
+      case PERMISSION_DENIED -> 403;
+      case NOT_FOUND -> 404;
+      case ALREADY_EXISTS, ABORTED -> 409;
+      case CANCELLED -> 499;
+      case UNKNOWN, UNIMPLEMENTED, DEADLINE_EXCEEDED, RESOURCE_EXHAUSTED, INTERNAL, UNAVAILABLE, DATA_LOSS -> 500;
+    };
+
+    return Response.status(httpCode)
+        .entity(new ErrorMessage(httpCode, exception.getMessage()))
+        .type(MediaType.APPLICATION_JSON_TYPE)
+        .build();
+  }
+}

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

@@ -100,6 +100,9 @@ public class Account {
   @JsonProperty("inCds")
   private boolean discoverableByPhoneNumber = true;
 
+  @JsonProperty("bcr")
+  private byte[] backupCredentialRequest;
+
   @JsonProperty
   private int version;
 
@@ -486,6 +489,14 @@ public class Account {
     this.version = version;
   }
 
+  public byte[] getBackupCredentialRequest() {
+    return backupCredentialRequest;
+  }
+
+  public void setBackupCredentialRequest(final byte[] backupCredentialRequest) {
+    this.backupCredentialRequest = backupCredentialRequest;
+  }
+
 
   /**
    * Have all this account's devices been manually locked?

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

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.util;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Metrics;
+import java.io.IOException;
+import java.util.Base64;
+import org.signal.libsignal.zkgroup.InvalidInputException;
+import org.signal.libsignal.zkgroup.backups.BackupAuthCredentialRequest;
+import org.signal.libsignal.zkgroup.internal.ByteArray;
+import org.whispersystems.textsecuregcm.metrics.MetricsUtil;
+
+public class BackupAuthCredentialAdapter {
+
+  private static final Counter INVALID_BASE64_COUNTER =
+      Metrics.counter(MetricsUtil.name(BackupAuthCredentialAdapter.class, "invalidBase64"));
+
+  private static final Counter INVALID_BYTES_COUNTER =
+      Metrics.counter(MetricsUtil.name(BackupAuthCredentialAdapter.class, "invalidBackupAuthObject"));
+
+  abstract static class GenericDeserializer<T> extends JsonDeserializer<T> {
+
+    abstract T deserialize(final byte[] bytes) throws InvalidInputException;
+
+    @Override
+    public T deserialize(final JsonParser parser, final DeserializationContext deserializationContext)
+        throws IOException {
+      final byte[] bytes;
+      try {
+        bytes = Base64.getDecoder().decode(parser.getValueAsString());
+      } catch (final IllegalArgumentException e) {
+        INVALID_BASE64_COUNTER.increment();
+        throw new JsonParseException(parser, "Could not parse string as a base64-encoded value", e);
+      }
+      try {
+        return deserialize(bytes);
+      } catch (InvalidInputException e) {
+        INVALID_BYTES_COUNTER.increment();
+        throw new JsonParseException(parser, "Could not interpret bytes as a BackupAuth object");
+      }
+    }
+  }
+
+  static class GenericSerializer<T extends ByteArray> extends JsonSerializer<T> {
+
+    @Override
+    public void serialize(final T t, final JsonGenerator jsonGenerator, final SerializerProvider serializerProvider)
+        throws IOException {
+      jsonGenerator.writeString(Base64.getEncoder().encodeToString(t.serialize()));
+    }
+  }
+
+  public static class CredentialRequestSerializer extends GenericSerializer<BackupAuthCredentialRequest> {}
+  public static class CredentialRequestDeserializer extends GenericDeserializer<BackupAuthCredentialRequest> {
+    @Override
+    BackupAuthCredentialRequest deserialize(final byte[] bytes) throws InvalidInputException {
+      return new BackupAuthCredentialRequest(bytes);
+    }
+  }
+
+
+}