fully migrate AS storage provider to backend

backend/migrations/20231124125801_create_as_batched_keys_table.sql

@@ -0,0 +1,10 @@
+-- SPDX-FileCopyrightText: 2023 Phoenix R&D GmbH <[email protected]>
+--
+-- SPDX-License-Identifier: AGPL-3.0-or-later
+
+-- migrations/{timestamp}_create_as_batched_keys_table.sql
+CREATE TABLE as_batched_keys(
+token_key_id smallint NOT NULL,
+PRIMARY KEY (token_key_id),
+voprf_server BYTEA NOT NULL
+);

backend/migrations/20231129125623_create_opaque_setup_table.sql

@@ -0,0 +1,12 @@
+-- SPDX-FileCopyrightText: 2023 Phoenix R&D GmbH <[email protected]>
+--
+-- SPDX-License-Identifier: AGPL-3.0-or-later
+
+-- migrations/{timestamp}_create_opaque_setup_table.sql
+-- Create opaque setup Table
+CREATE TABLE opaque_setup(
+id uuid NOT NULL,
+PRIMARY KEY (id),
+opaque_setup BYTEA NOT NULL
+);
+

backend/src/auth_service/client_api/anonymous.rs

@@ -14,42 +14,53 @@ use phnxtypes::{
 
 use crate::auth_service::{
     client_record::ClientRecord,
+    connection_package::StorableConnectionPackage,
     credentials::{intermediate_signing_key::IntermediateCredential, signing_key::Credential},
-    storage_provider_trait::AsStorageProvider,
+    queue::Queue,
+    user_record::UserRecord,
     AuthService,
 };
 
 impl AuthService {
-    pub(crate) async fn as_user_clients<S: AsStorageProvider>(
-        storage_provider: &S,
+    pub(crate) async fn as_user_clients(
+        &self,
         params: UserClientsParams,
     ) -> Result<UserClientsResponse, UserClientsError> {
         let UserClientsParams { user_name } = params;
 
         // Look up the user entry in the DB
-        let client_credentials = storage_provider.client_credentials(&user_name).await;
+        let client_credentials = UserRecord::client_credentials(&self.db_pool, &user_name)
+            .await
+            .map_err(|e| {
+                tracing::warn!("Failed to load client credentials: {:?}", e);
+                UserClientsError::StorageError
+            })?;
 
         let response = UserClientsResponse { client_credentials };
 
         Ok(response)
     }
 
-    pub async fn as_user_connection_packages<S: AsStorageProvider>(
-        storage_provider: &S,
+    pub async fn as_user_connection_packages(
+        &self,
         params: UserConnectionPackagesParams,
     ) -> Result<UserConnectionPackagesResponse, UserConnectionPackagesError> {
         let UserConnectionPackagesParams { user_name } = params;
 
-        let connection_packages = storage_provider
-            .load_user_connection_packages(&user_name)
-            .await
-            .map_err(|e| {
-                tracing::warn!(
-                    "Failed to load connection packages due to storage error: {:?}",
-                    e
-                );
-                UserConnectionPackagesError::StorageError
-            })?;
+        let mut connection = self.db_pool.acquire().await.map_err(|e| {
+            tracing::warn!("Failed to acquire connection from pool: {:?}", e);
+            UserConnectionPackagesError::StorageError
+        })?;
+        let connection_packages =
+            StorableConnectionPackage::user_connection_packages(&mut connection, &user_name)
+                .await
+                .map_err(|e| {
+                    tracing::warn!(
+                        "Failed to load connection packages due to storage error: {:?}",
+                        e
+                    );
+                    UserConnectionPackagesError::StorageError
+                })?;
 
         // If there are no connection packages, we have to conclude that there
         // is no user.
@@ -63,9 +74,8 @@ impl AuthService {
         Ok(response)
     }
 
-    pub(crate) async fn as_enqueue_message<S: AsStorageProvider>(
+    pub(crate) async fn as_enqueue_message(
         &self,
-        storage_provider: &S,
         params: EnqueueMessageParams,
     ) -> Result<(), EnqueueMessageError> {
         let EnqueueMessageParams {
@@ -94,8 +104,11 @@ impl AuthService {
         // TODO: Future work: PCS
 
         tracing::trace!("Enqueueing message in storage provider");
-        storage_provider
-            .enqueue(&client_id, queue_message)
+        let mut connection = self.db_pool.acquire().await.map_err(|e| {
+            tracing::warn!("Failed to acquire connection from pool: {:?}", e);
+            EnqueueMessageError::StorageError
+        })?;
+        Queue::enqueue(&mut connection, &client_id, queue_message)
             .await
             .map_err(|e| {
                 tracing::warn!("Failed to enqueue message: {:?}", e);

backend/src/auth_service/client_api/client.rs

@@ -25,15 +25,15 @@ use crate::auth_service::{
     client_record::ClientRecord,
     connection_package::StorableConnectionPackage,
     credentials::intermediate_signing_key::{IntermediateCredential, IntermediateSigningKey},
-    storage_provider_trait::AsStorageProvider,
+    opaque::OpaqueSetup,
+    queue::Queue,
     user_record::UserRecord,
     AuthService,
 };
 
 impl AuthService {
-    pub(crate) async fn as_init_client_addition<S: AsStorageProvider>(
+    pub(crate) async fn as_init_client_addition(
         &self,
-        storage_provider: &S,
         params: InitiateClientAdditionParams,
     ) -> Result<InitClientAdditionResponse, InitClientAdditionError> {
         let InitiateClientAdditionParams {
@@ -42,7 +42,7 @@ impl AuthService {
         } = params;
 
         // Load the server setup from storage
-        let server_setup = storage_provider.load_opaque_setup().await.map_err(|e| {
+        let server_setup = OpaqueSetup::load(&self.db_pool).await.map_err(|e| {
             tracing::error!("Storage provider error: {:?}", e);
             InitClientAdditionError::StorageError
         })?;
@@ -187,8 +187,8 @@ impl AuthService {
             .map_err(|_| FinishClientAdditionError::InvalidConnectionPackage)?;
 
         StorableConnectionPackage::store_multiple(
-            &mut connection,
-            vec![verified_connection_package].into_iter(),
+            &self.db_pool,
+            vec![verified_connection_package],
             &client_id,
         )
         .await
@@ -221,8 +221,8 @@ impl AuthService {
         Ok(())
     }
 
-    pub(crate) async fn as_dequeue_messages<S: AsStorageProvider>(
-        storage_provider: &S,
+    pub(crate) async fn as_dequeue_messages(
+        &self,
         params: DequeueMessagesParamsTbs,
     ) -> Result<DequeueMessagesResponse, AsDequeueError> {
         let DequeueMessagesParamsTbs {
@@ -234,13 +234,21 @@ impl AuthService {
         // TODO: The backend should have its own value for max_messages and use
         // that one if the client-given one exceeds it.
         tracing::trace!("Reading and deleting messages from storage provider");
-        let (messages, remaining_messages_number) = storage_provider
-            .read_and_delete(&sender, sequence_number_start, max_message_number)
-            .await
-            .map_err(|e| {
-                tracing::error!("Storage provider error: {:?}", e);
-                AsDequeueError::StorageError
-            })?;
+        let mut connection = self.db_pool.acquire().await.map_err(|e| {
+            tracing::error!("Error acquiring connection: {:?}", e);
+            AsDequeueError::StorageError
+        })?;
+        let (messages, remaining_messages_number) = Queue::read_and_delete(
+            &mut connection,
+            &sender,
+            sequence_number_start,
+            max_message_number,
+        )
+        .await
+        .map_err(|e| {
+            tracing::error!("Storage provider error: {:?}", e);
+            AsDequeueError::StorageError
+        })?;
 
         let response = DequeueMessagesResponse {
             messages,

backend/src/auth_service/client_api/key_packages.rs

@@ -11,14 +11,13 @@ use phnxtypes::{
 };
 
 use crate::auth_service::{
-    credentials::intermediate_signing_key::IntermediateCredential,
-    storage_provider_trait::AsStorageProvider, AuthService,
+    connection_package::StorableConnectionPackage,
+    credentials::intermediate_signing_key::IntermediateCredential, AuthService,
 };
 
 impl AuthService {
-    pub(crate) async fn as_publish_connection_packages<S: AsStorageProvider>(
+    pub(crate) async fn as_publish_connection_packages(
         &self,
-        storage_provider: &S,
         params: AsPublishConnectionPackagesParamsTbs,
     ) -> Result<(), PublishConnectionPackageError> {
         let AsPublishConnectionPackagesParamsTbs {
@@ -33,7 +32,7 @@ impl AuthService {
                 PublishConnectionPackageError::StorageError
             })?;
 
-        // TODO: Last resort key package
+        // TODO: Last resort connection package
         let connection_packages = connection_packages
             .into_iter()
             .map(|cp| {
@@ -46,26 +45,33 @@ impl AuthService {
             })
             .collect::<Result<Vec<ConnectionPackage>, PublishConnectionPackageError>>()?;
 
-        storage_provider
-            .store_connection_packages(&client_id, connection_packages)
-            .await
-            .map_err(|_| PublishConnectionPackageError::StorageError)?;
+        StorableConnectionPackage::store_multiple(
+            &self.db_pool,
+            connection_packages.into_iter(),
+            &client_id,
+        )
+        .await
+        .map_err(|_| PublishConnectionPackageError::StorageError)?;
         Ok(())
     }
 
-    pub(crate) async fn as_client_key_package<S: AsStorageProvider>(
-        storage_provider: &S,
+    pub(crate) async fn as_client_key_package(
+        &self,
         params: ClientConnectionPackageParamsTbs,
     ) -> Result<AsClientConnectionPackageResponse, ClientKeyPackageError> {
         let client_id = params.0;
 
-        let connection_package = storage_provider
-            .client_connection_package(&client_id)
-            .await
-            .map_err(|e| {
-                tracing::error!("Storage provider error: {:?}", e);
-                ClientKeyPackageError::StorageError
-            })?;
+        let mut connection = self.db_pool.acquire().await.map_err(|e| {
+            tracing::error!("Can't acquire a connection: {:?}", e);
+            ClientKeyPackageError::StorageError
+        })?;
+        let connection_package =
+            StorableConnectionPackage::client_connection_package(&mut connection, &client_id)
+                .await
+                .map_err(|e| {
+                    tracing::error!("Storage provider error: {:?}", e);
+                    ClientKeyPackageError::StorageError
+                })?;
 
         let response = AsClientConnectionPackageResponse {
             connection_package: Some(connection_package),

backend/src/auth_service/client_api/mod.rs

@@ -9,7 +9,7 @@ use phnxtypes::{
     messages::client_as::{Init2FactorAuthParamsTbs, Init2FactorAuthResponse},
 };
 
-use super::{storage_provider_trait::AsStorageProvider, user_record::UserRecord, AuthService};
+use super::{opaque::OpaqueSetup, user_record::UserRecord, AuthService};
 
 use tls_codec::Serialize;
 
@@ -20,9 +20,8 @@ pub mod privacypass;
 pub mod user;
 
 impl AuthService {
-    pub(crate) async fn as_init_two_factor_auth<S: AsStorageProvider>(
+    pub(crate) async fn as_init_two_factor_auth(
         &self,
-        storage_provider: &S,
         params: Init2FactorAuthParamsTbs,
     ) -> Result<Init2FactorAuthResponse, Init2FactorAuthError> {
         let Init2FactorAuthParamsTbs {
@@ -31,7 +30,7 @@ impl AuthService {
         } = params;
 
         // Load the server setup from storage
-        let server_setup = storage_provider.load_opaque_setup().await.map_err(|e| {
+        let server_setup = OpaqueSetup::load(&self.db_pool).await.map_err(|e| {
             tracing::error!("Storage provider error: {:?}", e);
             Init2FactorAuthError::StorageError
         })?;

backend/src/auth_service/client_api/privacypass.rs

@@ -8,13 +8,15 @@ use phnxtypes::{
 };
 use privacypass::batched_tokens_ristretto255::server::Server;
 
-use crate::auth_service::{storage_provider_trait::AsStorageProvider, AuthService};
+use crate::auth_service::{
+    client_record::ClientRecord, privacy_pass::AuthServiceBatchedKeyStoreProvider, AuthService,
+};
 
-const MAX_TOKENS_PER_REQUEST: usize = 100;
+const MAX_TOKENS_PER_REQUEST: i32 = 100;
 
 impl AuthService {
-    pub(crate) async fn as_issue_tokens<S: AsStorageProvider>(
-        storage_provider: &S,
+    pub(crate) async fn as_issue_tokens(
+        &self,
         params: IssueTokensParamsTbs,
     ) -> Result<IssueTokensResponse, IssueTokensError> {
         let IssueTokensParamsTbs {
@@ -24,22 +26,33 @@ impl AuthService {
             token_type: _,
             token_request,
         } = params;
+        let tokens_requested = token_request.nr() as i32;
 
-        // Load current token allowance from storage provider
-        let token_allowance = storage_provider
-            .load_client_token_allowance(&client_id)
+        // Start a transaction
+        let mut transaction = self
+            .db_pool
+            .begin()
             .await
             .map_err(|_| IssueTokensError::StorageError)?;
 
-        let tokens_requested = token_request.nr();
+        // Load current token allowance from storage provider
+        let mut client_record = ClientRecord::load(&mut *transaction, &client_id)
+            .await
+            .map_err(|e| {
+                tracing::error!("Error loading client record: {:?}", e);
+                IssueTokensError::StorageError
+            })?
+            .ok_or(IssueTokensError::UnknownClient)?;
+
+        let token_allowance = client_record.token_allowance;
         if tokens_requested > token_allowance || tokens_requested > MAX_TOKENS_PER_REQUEST {
             return Err(IssueTokensError::TooManyTokens);
         }
 
-        let pp_key_store = storage_provider.privacy_pass_key_store().await;
         let pp_server = Server::new();
+        let key_store = AuthServiceBatchedKeyStoreProvider::new(&mut transaction);
         let token_response = pp_server
-            .issue_token_response(pp_key_store, token_request)
+            .issue_token_response(&key_store, token_request)
             .await
             .map_err(|_| IssueTokensError::PrivacyPassError)?;
 
@@ -48,8 +61,14 @@ impl AuthService {
         };
 
         // Reduce the token allowance by the number of tokens issued.
-        storage_provider
-            .set_client_token_allowance(&client_id, token_allowance - tokens_requested)
+        client_record.token_allowance -= tokens_requested;
+        client_record.update(&mut *transaction).await.map_err(|e| {
+            tracing::error!("Error updating client record: {:?}", e);
+            IssueTokensError::StorageError
+        })?;
+
+        transaction
+            .commit()
             .await
             .map_err(|_| IssueTokensError::StorageError)?;