feat: create key in encryption service for merchant and user (#4910)

Co-authored-by: Arjun Karthik <m.arjunkarthik@gmail.com>
Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>

config/config.example.toml

@@ -28,6 +28,10 @@ certificate = "/path/to/certificate.pem"
 idle_pool_connection_timeout = 90 # Timeout for idle pool connections (defaults to 90s)
 
 
+# Configuration for the Key Manager Service
+[key_manager]
+url = "http://localhost:5000" # URL of the encryption service
+
 # Main SQL data store credentials
 [master_database]
 username = "db_user"      # DB Username

config/deployments/env_specific.toml

@@ -246,6 +246,10 @@ payment_intents = "hyperswitch-payment-intent-events"
 refunds = "hyperswitch-refund-events"
 disputes = "hyperswitch-dispute-events"
 
+# Configuration for the Key Manager Service
+[key_manager]
+url = "http://localhost:5000" # URL of the encryption service
+
 # This section provides some secret values.
 [secrets]
 master_enc_key = "sample_key"            # Master Encryption key used to encrypt merchant wise encryption key. Should be 32-byte long.

config/development.toml

@@ -12,6 +12,9 @@ metrics_enabled = false
 use_xray_generator = false
 bg_metrics_collection_interval_in_secs = 15
 
+[key_manager]
+url = "http://localhost:5000"
+
 # TODO: Update database credentials before running application
 [master_database]
 username = "db_user"

config/docker_compose.toml

@@ -90,6 +90,9 @@ default_command_timeout = 30
 unresponsive_timeout = 10
 max_feed_count = 200
 
+[key_manager]
+url = "http://localhost:5000"
+
 [cors]
 max_age = 30
 # origins = "http://localhost:8080,http://localhost:9000"

crates/api_models/src/admin.rs

@@ -994,6 +994,12 @@ pub struct ToggleKVResponse {
     pub kv_enabled: bool,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct TransferKeyResponse {
+    /// The identifier for the Merchant Account
+    #[schema(example = 32)]
+    pub total_transferred: usize,
+}
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 pub struct ToggleKVRequest {
     #[serde(skip_deserializing)]

crates/api_models/src/events.rs

@@ -80,6 +80,7 @@ impl_misc_api_event_type!(
     ToggleKVRequest,
     ToggleAllKVRequest,
     ToggleAllKVResponse,
+    TransferKeyResponse,
     MerchantAccountDeleteResponse,
     MerchantAccountUpdate,
     CardInfoResponse,

crates/api_models/src/events/user.rs

@@ -20,7 +20,7 @@ use crate::user::{
     SsoSignInRequest, SwitchMerchantIdRequest, TokenOrPayloadResponse, TokenResponse,
     TwoFactorAuthStatusResponse, UpdateUserAccountDetailsRequest,
     UpdateUserAuthenticationMethodRequest, UserFromEmailRequest, UserMerchantCreate,
-    VerifyEmailRequest, VerifyRecoveryCodeRequest, VerifyTotpRequest,
+    UserTransferKeyResponse, VerifyEmailRequest, VerifyRecoveryCodeRequest, VerifyTotpRequest,
 };
 
 impl ApiEventMetric for DashboardEntryResponse {
@@ -85,6 +85,7 @@ common_utils::impl_misc_api_event_type!(
     UpdateUserAuthenticationMethodRequest,
     GetSsoAuthUrlRequest,
     SsoSignInRequest,
+    UserTransferKeyResponse,
     AuthSelectRequest
 );
 

crates/api_models/src/user.rs

@@ -379,3 +379,8 @@ pub struct AuthIdQueryParam {
 pub struct AuthSelectRequest {
     pub id: Option<String>,
 }
+
+#[derive(Debug, serde::Deserialize, serde::Serialize)]
+pub struct UserTransferKeyResponse {
+    pub total_transferred: usize,
+}

crates/common_utils/Cargo.toml

@@ -8,6 +8,8 @@ readme = "README.md"
 license.workspace = true
 
 [features]
+keymanager = ["dep:router_env"]
+keymanager_mtls = ["reqwest/rustls-tls"]
 signals = ["dep:signal-hook-tokio", "dep:signal-hook", "dep:tokio", "dep:router_env", "dep:futures"]
 async_ext = ["dep:async-trait", "dep:futures"]
 logs = ["dep:router_env"]

crates/common_utils/src/errors.rs

@@ -145,6 +145,42 @@ where
     }
 }
 
+#[allow(missing_docs)]
+#[derive(Debug, thiserror::Error)]
+pub enum KeyManagerClientError {
+    #[error("Failed to construct header from the given value")]
+    FailedtoConstructHeader,
+    #[error("Failed to send request to Keymanager")]
+    RequestNotSent(String),
+    #[error("URL encoding of request failed")]
+    UrlEncodingFailed,
+    #[error("Failed to build the reqwest client ")]
+    ClientConstructionFailed,
+    #[error("Failed to send the request to Keymanager")]
+    RequestSendFailed,
+    #[error("Internal Server Error Received {0:?}")]
+    InternalServerError(bytes::Bytes),
+    #[error("Bad request received {0:?}")]
+    BadRequest(bytes::Bytes),
+    #[error("Unexpected Error occurred while calling the KeyManager")]
+    Unexpected(bytes::Bytes),
+    #[error("Response Decoding failed")]
+    ResponseDecodingFailed,
+}
+
+#[allow(missing_docs)]
+#[derive(Debug, thiserror::Error)]
+pub enum KeyManagerError {
+    #[error("Failed to add key to the KeyManager")]
+    KeyAddFailed,
+    #[error("Failed to transfer the key to the KeyManager")]
+    KeyTransferFailed,
+    #[error("Failed to Encrypt the data in the KeyManager")]
+    EncryptionFailed,
+    #[error("Failed to Decrypt the data in the KeyManager")]
+    DecryptionFailed,
+}
+
 /// Allow [error_stack::Report] to convert between error types
 /// This auto-implements [ReportSwitchExt] for the corresponding errors
 pub trait ErrorSwitch<T> {

crates/common_utils/src/keymanager.rs

@@ -0,0 +1,175 @@
+//! Consists of all the common functions to use the Keymanager.
+
+use core::fmt::Debug;
+use std::str::FromStr;
+
+use error_stack::ResultExt;
+use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
+#[cfg(feature = "keymanager_mtls")]
+use masking::PeekInterface;
+use once_cell::sync::OnceCell;
+use router_env::{instrument, logger, tracing};
+
+use crate::{
+    errors,
+    types::keymanager::{
+        DataKeyCreateResponse, EncryptionCreateRequest, EncryptionTransferRequest, KeyManagerState,
+    },
+};
+
+const CONTENT_TYPE: &str = "Content-Type";
+static ENCRYPTION_API_CLIENT: OnceCell<reqwest::Client> = OnceCell::new();
+
+/// Get keymanager client constructed from the url and state
+#[instrument(skip_all)]
+#[allow(unused_mut)]
+fn get_api_encryption_client(
+    state: &KeyManagerState,
+) -> errors::CustomResult<reqwest::Client, errors::KeyManagerClientError> {
+    let get_client = || {
+        let mut client = reqwest::Client::builder()
+            .redirect(reqwest::redirect::Policy::none())
+            .pool_idle_timeout(std::time::Duration::from_secs(
+                state.client_idle_timeout.unwrap_or_default(),
+            ));
+
+        #[cfg(feature = "keymanager_mtls")]
+        {
+            let cert = state.cert.clone();
+            let ca = state.ca.clone();
+
+            let identity = reqwest::Identity::from_pem(cert.peek().as_ref())
+                .change_context(errors::KeyManagerClientError::ClientConstructionFailed)?;
+            let ca_cert = reqwest::Certificate::from_pem(ca.peek().as_ref())
+                .change_context(errors::KeyManagerClientError::ClientConstructionFailed)?;
+
+            client = client
+                .use_rustls_tls()
+                .identity(identity)
+                .add_root_certificate(ca_cert)
+                .https_only(true);
+        }
+
+        client
+            .build()
+            .change_context(errors::KeyManagerClientError::ClientConstructionFailed)
+    };
+
+    Ok(ENCRYPTION_API_CLIENT.get_or_try_init(get_client)?.clone())
+}
+
+/// Generic function to send the request to keymanager
+#[instrument(skip_all)]
+pub async fn send_encryption_request<T>(
+    state: &KeyManagerState,
+    headers: HeaderMap,
+    url: String,
+    method: Method,
+    request_body: T,
+) -> errors::CustomResult<reqwest::Response, errors::KeyManagerClientError>
+where
+    T: serde::Serialize,
+{
+    let client = get_api_encryption_client(state)?;
+    let url = reqwest::Url::parse(&url)
+        .change_context(errors::KeyManagerClientError::UrlEncodingFailed)?;
+
+    client
+        .request(method, url)
+        .json(&request_body)
+        .headers(headers)
+        .send()
+        .await
+        .change_context(errors::KeyManagerClientError::RequestNotSent(
+            "Unable to send request to encryption service".to_string(),
+        ))
+}
+
+/// Generic function to call the Keymanager and parse the response back
+#[instrument(skip_all)]
+pub async fn call_encryption_service<T, R>(
+    state: &KeyManagerState,
+    method: Method,
+    endpoint: &str,
+    request_body: T,
+) -> errors::CustomResult<R, errors::KeyManagerClientError>
+where
+    T: serde::Serialize + Send + Sync + 'static + Debug,
+    R: serde::de::DeserializeOwned,
+{
+    let url = format!("{}/{endpoint}", &state.url);
+
+    logger::info!(key_manager_request=?request_body);
+
+    let response = send_encryption_request(
+        state,
+        HeaderMap::from_iter(
+            vec![(
+                HeaderName::from_str(CONTENT_TYPE)
+                    .change_context(errors::KeyManagerClientError::FailedtoConstructHeader)?,
+                HeaderValue::from_str("application/json")
+                    .change_context(errors::KeyManagerClientError::FailedtoConstructHeader)?,
+            )]
+            .into_iter(),
+        ),
+        url,
+        method,
+        request_body,
+    )
+    .await
+    .map_err(|err| err.change_context(errors::KeyManagerClientError::RequestSendFailed))?;
+
+    logger::info!(key_manager_response=?response);
+
+    match response.status() {
+        StatusCode::OK => response
+            .json::<R>()
+            .await
+            .change_context(errors::KeyManagerClientError::ResponseDecodingFailed),
+        StatusCode::INTERNAL_SERVER_ERROR => {
+            Err(errors::KeyManagerClientError::InternalServerError(
+                response
+                    .bytes()
+                    .await
+                    .change_context(errors::KeyManagerClientError::ResponseDecodingFailed)?,
+            )
+            .into())
+        }
+        StatusCode::BAD_REQUEST => Err(errors::KeyManagerClientError::BadRequest(
+            response
+                .bytes()
+                .await
+                .change_context(errors::KeyManagerClientError::ResponseDecodingFailed)?,
+        )
+        .into()),
+        _ => Err(errors::KeyManagerClientError::Unexpected(
+            response
+                .bytes()
+                .await
+                .change_context(errors::KeyManagerClientError::ResponseDecodingFailed)?,
+        )
+        .into()),
+    }
+}
+
+/// A function to create the key in keymanager
+#[instrument(skip_all)]
+pub async fn create_key_in_key_manager(
+    state: &KeyManagerState,
+    request_body: EncryptionCreateRequest,
+) -> errors::CustomResult<DataKeyCreateResponse, errors::KeyManagerError> {
+    call_encryption_service(state, Method::POST, "key/create", request_body)
+        .await
+        .change_context(errors::KeyManagerError::KeyAddFailed)
+}
+
+/// A function to transfer the key in keymanager
+#[instrument(skip_all)]
+pub async fn transfer_key_to_key_manager(
+    state: &KeyManagerState,
+    request_body: EncryptionTransferRequest,
+) -> errors::CustomResult<DataKeyCreateResponse, errors::KeyManagerError> {
+    call_encryption_service(state, Method::POST, "key/transfer", request_body)
+        .await
+        .change_context(errors::KeyManagerError::KeyTransferFailed)
+}

crates/common_utils/src/lib.rs

@@ -19,6 +19,8 @@ pub mod events;
 pub mod ext_traits;
 pub mod fp_utils;
 pub mod id_type;
+#[cfg(feature = "keymanager")]
+pub mod keymanager;
 pub mod link_utils;
 pub mod macros;
 pub mod new_type;

crates/common_utils/src/types.rs

@@ -1,4 +1,6 @@
 //! Types that can be used in other crates
+pub mod keymanager;
+
 use std::{
     fmt::Display,
     ops::{Add, Sub},

crates/common_utils/src/types/keymanager.rs

@@ -0,0 +1,41 @@
+#![allow(missing_docs)]
+
+#[cfg(feature = "keymanager_mtls")]
+use masking::Secret;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug)]
+pub struct KeyManagerState {
+    pub url: String,
+    pub client_idle_timeout: Option<u64>,
+    #[cfg(feature = "keymanager_mtls")]
+    pub ca: Secret<String>,
+    #[cfg(feature = "keymanager_mtls")]
+    pub cert: Secret<String>,
+}
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
+#[serde(tag = "data_identifier", content = "key_identifier")]
+pub enum Identifier {
+    User(String),
+    Merchant(String),
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
+pub struct EncryptionCreateRequest {
+    #[serde(flatten)]
+    pub identifier: Identifier,
+}
+
+#[derive(Serialize, Deserialize, Debug, Eq, PartialEq, Clone)]
+pub struct EncryptionTransferRequest {
+    #[serde(flatten)]
+    pub identifier: Identifier,
+    pub key: String,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub struct DataKeyCreateResponse {
+    #[serde(flatten)]
+    pub identifier: Identifier,
+    pub key_version: String,
+}

crates/diesel_models/src/query/merchant_key_store.rs

@@ -54,4 +54,20 @@ impl MerchantKeyStore {
         )
         .await
     }
+
+    pub async fn list_all_key_stores(conn: &PgPooledConn) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<
+            <Self as HasTable>::Table,
+            _,
+            <<Self as HasTable>::Table as diesel::Table>::PrimaryKey,
+            _,
+        >(
+            conn,
+            dsl::merchant_id.ne_all(vec!["".to_string()]),
+            None,
+            None,
+            None,
+        )
+        .await
+    }
 }

crates/diesel_models/src/query/user_key_store.rs

@@ -14,6 +14,22 @@ impl UserKeyStoreNew {
 }
 
 impl UserKeyStore {
+    pub async fn get_all_user_key_stores(conn: &PgPooledConn) -> StorageResult<Vec<Self>> {
+        generics::generic_filter::<
+            <Self as HasTable>::Table,
+            _,
+            <<Self as HasTable>::Table as diesel::Table>::PrimaryKey,
+            _,
+        >(
+            conn,
+            dsl::user_id.ne_all(vec!["".to_string()]),
+            None,
+            None,
+            None,
+        )
+        .await
+    }
+
     pub async fn find_by_user_id(conn: &PgPooledConn, user_id: &str) -> StorageResult<Self> {
         generics::generic_find_one::<<Self as HasTable>::Table, _, _>(
             conn,

crates/router/Cargo.toml

@@ -11,10 +11,12 @@ license.workspace = true
 [features]
 default = ["kv_store", "stripe", "oltp", "olap", "backwards_compatibility", "accounts_cache", "dummy_connector", "payouts", "payout_retry", "business_profile_routing", "connector_choice_mca_id", "profile_specific_fallback_routing", "retry", "frm", "tls"]
 tls = ["actix-web/rustls-0_22"]
+keymanager_mtls = ["reqwest/rustls-tls","common_utils/keymanager_mtls"]
 email = ["external_services/email", "scheduler/email", "olap"]
+keymanager_create = []
 frm = ["api_models/frm", "hyperswitch_domain_models/frm"]
 stripe = ["dep:serde_qs"]
-release = ["stripe", "email", "backwards_compatibility", "business_profile_routing", "accounts_cache", "kv_store", "connector_choice_mca_id", "profile_specific_fallback_routing", "vergen", "recon", "external_services/aws_kms", "external_services/aws_s3"]
+release = ["stripe", "email", "backwards_compatibility", "business_profile_routing", "accounts_cache", "kv_store", "connector_choice_mca_id", "profile_specific_fallback_routing", "vergen", "recon", "external_services/aws_kms", "external_services/aws_s3","keymanager_mtls","keymanager_create"]
 olap = ["hyperswitch_domain_models/olap", "storage_impl/olap", "scheduler/olap", "api_models/olap", "dep:analytics"]
 oltp = ["storage_impl/oltp"]
 kv_store = ["scheduler/kv_store"]
@@ -107,7 +109,7 @@ api_models = { version = "0.1.0", path = "../api_models", features = ["errors"]
 analytics = { version = "0.1.0", path = "../analytics", optional = true }
 cards = { version = "0.1.0", path = "../cards" }
 common_enums = { version = "0.1.0", path = "../common_enums" }
-common_utils = { version = "0.1.0", path = "../common_utils", features = ["signals", "async_ext", "logs", "metrics"] }
+common_utils = { version = "0.1.0", path = "../common_utils", features = ["signals", "async_ext", "logs", "metrics","keymanager"] }
 hyperswitch_constraint_graph = { version = "0.1.0", path = "../hyperswitch_constraint_graph" }
 currency_conversion = { version = "0.1.0", path = "../currency_conversion" }
 hyperswitch_domain_models = { version = "0.1.0", path = "../hyperswitch_domain_models", default-features = false }

crates/router/src/configs/defaults.rs

@@ -8810,3 +8810,15 @@ impl Default for super::settings::ApiKeys {
         }
     }
 }
+
+impl Default for super::settings::KeyManagerConfig {
+    fn default() -> Self {
+        Self {
+            url: String::from("localhost:5000"),
+            #[cfg(feature = "keymanager_mtls")]
+            ca: String::default().into(),
+            #[cfg(feature = "keymanager_mtls")]
+            cert: String::default().into(),
+        }
+    }
+}

crates/router/src/configs/secrets_transformers.rs

@@ -197,6 +197,35 @@ impl SecretsHandler for settings::PaymentMethodAuth {
     }
 }
 
+#[async_trait::async_trait]
+impl SecretsHandler for settings::KeyManagerConfig {
+    async fn convert_to_raw_secret(
+        value: SecretStateContainer<Self, SecuredSecret>,
+        _secret_management_client: &dyn SecretManagementInterface,
+    ) -> CustomResult<SecretStateContainer<Self, RawSecret>, SecretsManagementError> {
+        #[cfg(feature = "keymanager_mtls")]
+        let keyconfig = value.get_inner();
+
+        #[cfg(feature = "keymanager_mtls")]
+        let ca = _secret_management_client
+            .get_secret(keyconfig.ca.clone())
+            .await?;
+
+        #[cfg(feature = "keymanager_mtls")]
+        let cert = _secret_management_client
+            .get_secret(keyconfig.cert.clone())
+            .await?;
+
+        Ok(value.transition_state(|keyconfig| Self {
+            #[cfg(feature = "keymanager_mtls")]
+            ca,
+            #[cfg(feature = "keymanager_mtls")]
+            cert,
+            ..keyconfig
+        }))
+    }
+}
+
 #[async_trait::async_trait]
 impl SecretsHandler for settings::Secrets {
     async fn convert_to_raw_secret(
@@ -318,6 +347,14 @@ pub(crate) async fn fetch_raw_secrets(
     .await
     .expect("Failed to decrypt payment method auth configs");
 
+    #[allow(clippy::expect_used)]
+    let key_manager = settings::KeyManagerConfig::convert_to_raw_secret(
+        conf.key_manager,
+        secret_management_client,
+    )
+    .await
+    .expect("Failed to decrypt keymanager configs");
+
     #[allow(clippy::expect_used)]
     let user_auth_methods = settings::UserAuthMethodSettings::convert_to_raw_secret(
         conf.user_auth_methods,
@@ -337,6 +374,7 @@ pub(crate) async fn fetch_raw_secrets(
         secrets_management: conf.secrets_management,
         proxy: conf.proxy,
         env: conf.env,
+        key_manager,
         #[cfg(feature = "olap")]
         replica_database,
         secrets,

crates/router/src/configs/settings.rs

@@ -69,6 +69,7 @@ pub struct Settings<S: SecretState> {
     pub log: Log,
     pub secrets: SecretStateContainer<Secrets, S>,
     pub locker: Locker,
+    pub key_manager: SecretStateContainer<KeyManagerConfig, S>,
     pub connectors: Connectors,
     pub forex_api: SecretStateContainer<ForexApi, S>,
     pub refund: Refund,
@@ -214,6 +215,15 @@ pub struct KvConfig {
     pub soft_kill: Option<bool>,
 }
 
+#[derive(Debug, Deserialize, Clone)]
+pub struct KeyManagerConfig {
+    pub url: String,
+    #[cfg(feature = "keymanager_mtls")]
+    pub cert: Secret<String>,
+    #[cfg(feature = "keymanager_mtls")]
+    pub ca: Secret<String>,
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 pub struct GenericLink {
     pub payment_method_collect: GenericLinkEnvConfig,

crates/router/src/core.rs

@@ -14,6 +14,7 @@ pub mod connector_onboarding;
 pub mod currency;
 pub mod customers;
 pub mod disputes;
+pub mod encryption;
 pub mod errors;
 pub mod files;
 #[cfg(feature = "frm")]

crates/router/src/core/admin.rs

@@ -9,6 +9,8 @@ use common_utils::{
     ext_traits::{AsyncExt, ConfigExt, Encode, ValueExt},
     pii,
 };
+#[cfg(all(feature = "keymanager_create", feature = "olap"))]
+use common_utils::{keymanager, types::keymanager as km_types};
 use diesel_models::configs;
 use error_stack::{report, FutureExt, ResultExt};
 use futures::future::try_join_all;
@@ -22,6 +24,7 @@ use crate::types::transformers::ForeignFrom;
 use crate::{
     consts,
     core::{
+        encryption::transfer_encryption_key,
         errors::{self, RouterResponse, RouterResult, StorageErrorExt},
         payments::helpers,
         routing::helpers as routing_helpers,
@@ -125,6 +128,19 @@ pub async fn create_merchant_account(
         .create_domain_model_from_request(db, key_store.clone())
         .await?;
 
+    #[cfg(feature = "keymanager_create")]
+    {
+        keymanager::create_key_in_key_manager(
+            &(&state).into(),
+            km_types::EncryptionCreateRequest {
+                identifier: km_types::Identifier::Merchant(merchant_id.clone()),
+            },
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::DuplicateMerchantAccount)
+        .attach_printable("Failed to insert key to KeyManager")?;
+    }
+
     db.insert_merchant_key_store(key_store.clone(), &master_key.to_vec().into())
         .await
         .to_duplicate_response(errors::ApiErrorResponse::DuplicateMerchantAccount)?;
@@ -2491,6 +2507,18 @@ pub(crate) fn validate_connector_auth_type(
     }
 }
 
+pub async fn transfer_key_store_to_key_manager(
+    state: SessionState,
+) -> RouterResponse<admin_types::TransferKeyResponse> {
+    let resp = transfer_encryption_key(&state).await?;
+
+    Ok(service_api::ApplicationResponse::Json(
+        admin_types::TransferKeyResponse {
+            total_transferred: resp,
+        },
+    ))
+}
+
 #[cfg(feature = "dummy_connector")]
 pub async fn validate_dummy_connector_enabled(
     state: &SessionState,

crates/router/src/core/encryption.rs

@@ -0,0 +1,55 @@
+use base64::Engine;
+use common_utils::{
+    keymanager::transfer_key_to_key_manager,
+    types::keymanager::{EncryptionTransferRequest, Identifier},
+};
+use error_stack::ResultExt;
+use hyperswitch_domain_models::merchant_key_store::MerchantKeyStore;
+use masking::ExposeInterface;
+
+use crate::{consts::BASE64_ENGINE, errors, types::domain::UserKeyStore, SessionState};
+
+pub async fn transfer_encryption_key(
+    state: &SessionState,
+) -> errors::CustomResult<usize, errors::ApiErrorResponse> {
+    let db = &*state.store;
+    let key_stores = db
+        .get_all_key_stores(&db.get_master_key().to_vec().into())
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)?;
+    send_request_to_key_service_for_merchant(state, key_stores).await
+}
+
+pub async fn send_request_to_key_service_for_merchant(
+    state: &SessionState,
+    keys: Vec<MerchantKeyStore>,
+) -> errors::CustomResult<usize, errors::ApiErrorResponse> {
+    futures::future::try_join_all(keys.into_iter().map(|key| async move {
+        let key_encoded = BASE64_ENGINE.encode(key.key.clone().into_inner().expose());
+        let req = EncryptionTransferRequest {
+            identifier: Identifier::Merchant(key.merchant_id.clone()),
+            key: key_encoded,
+        };
+        transfer_key_to_key_manager(&state.into(), req).await
+    }))
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .map(|v| v.len())
+}
+
+pub async fn send_request_to_key_service_for_user(
+    state: &SessionState,
+    keys: Vec<UserKeyStore>,
+) -> errors::CustomResult<usize, errors::ApiErrorResponse> {
+    futures::future::try_join_all(keys.into_iter().map(|key| async move {
+        let key_encoded = BASE64_ENGINE.encode(key.key.clone().into_inner().expose());
+        let req = EncryptionTransferRequest {
+            identifier: Identifier::User(key.user_id.clone()),
+            key: key_encoded,
+        };
+        transfer_key_to_key_manager(&state.into(), req).await
+    }))
+    .await
+    .change_context(errors::ApiErrorResponse::InternalServerError)
+    .map(|v| v.len())
+}

crates/router/src/core/user.rs

@@ -27,6 +27,7 @@ use super::errors::{StorageErrorExt, UserErrors, UserResponse, UserResult};
 use crate::services::email::types as email_types;
 use crate::{
     consts,
+    core::encryption::send_request_to_key_service_for_user,
     db::domain::user_authentication_method::DEFAULT_USER_AUTH_METHOD,
     routes::{app::ReqState, SessionState},
     services::{authentication as auth, authorization::roles, openidconnect, ApplicationResponse},
@@ -1911,6 +1912,25 @@ pub async fn generate_recovery_codes(
     }))
 }
 
+pub async fn transfer_user_key_store_keymanager(
+    state: SessionState,
+) -> UserResponse<user_api::UserTransferKeyResponse> {
+    let db = &state.global_store;
+
+    let key_stores = db
+        .get_all_user_key_store(&state.store.get_master_key().to_vec().into())
+        .await
+        .change_context(UserErrors::InternalServerError)?;
+
+    Ok(ApplicationResponse::Json(
+        user_api::UserTransferKeyResponse {
+            total_transferred: send_request_to_key_service_for_user(&state, key_stores)
+                .await
+                .change_context(UserErrors::InternalServerError)?,
+        },
+    ))
+}
+
 pub async fn verify_recovery_code(
     state: SessionState,
     user_token: auth::UserIdFromAuth,

crates/router/src/db/kafka_store.rs

@@ -2096,6 +2096,12 @@ impl MerchantKeyStoreInterface for KafkaStore {
             .list_multiple_key_stores(merchant_ids, key)
             .await
     }
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
+        self.diesel_store.get_all_key_stores(key).await
+    }
 }
 
 #[async_trait::async_trait]
@@ -2958,6 +2964,13 @@ impl UserKeyStoreInterface for KafkaStore {
             .get_user_key_store_by_user_id(user_id, key)
             .await
     }
+
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
+        self.diesel_store.get_all_user_key_store(key).await
+    }
 }
 
 #[async_trait::async_trait]

crates/router/src/db/merchant_key_store.rs

@@ -40,6 +40,11 @@ pub trait MerchantKeyStoreInterface {
         merchant_ids: Vec<String>,
         key: &Secret<Vec<u8>>,
     ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError>;
+
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -163,6 +168,26 @@ impl MerchantKeyStoreInterface for Store {
         }))
         .await
     }
+
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+        let fetch_func = || async {
+            diesel_models::merchant_key_store::MerchantKeyStore::list_all_key_stores(&conn)
+                .await
+                .map_err(|err| report!(errors::StorageError::from(err)))
+        };
+
+        futures::future::try_join_all(fetch_func().await?.into_iter().map(|key_store| async {
+            key_store
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -251,6 +276,21 @@ impl MerchantKeyStoreInterface for MockDb {
         )
         .await
     }
+    async fn get_all_key_stores(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::MerchantKeyStore>, errors::StorageError> {
+        let merchant_key_stores = self.merchant_key_store.lock().await;
+
+        futures::future::try_join_all(merchant_key_stores.iter().map(|merchant_key| async {
+            merchant_key
+                .to_owned()
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
 }
 
 #[cfg(test)]

crates/router/src/db/user_key_store.rs

@@ -27,6 +27,11 @@ pub trait UserKeyStoreInterface {
         user_id: &str,
         key: &Secret<Vec<u8>>,
     ) -> CustomResult<domain::UserKeyStore, errors::StorageError>;
+
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError>;
 }
 
 #[async_trait::async_trait]
@@ -65,6 +70,27 @@ impl UserKeyStoreInterface for Store {
             .await
             .change_context(errors::StorageError::DecryptionError)
     }
+
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
+        let conn = connection::pg_connection_read(self).await?;
+
+        let fetch_func = || async {
+            diesel_models::user_key_store::UserKeyStore::get_all_user_key_stores(&conn)
+                .await
+                .map_err(|err| report!(errors::StorageError::from(err)))
+        };
+
+        futures::future::try_join_all(fetch_func().await?.into_iter().map(|key_store| async {
+            key_store
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
 }
 
 #[async_trait::async_trait]
@@ -98,6 +124,22 @@ impl UserKeyStoreInterface for MockDb {
             .change_context(errors::StorageError::DecryptionError)
     }
 
+    async fn get_all_user_key_store(
+        &self,
+        key: &Secret<Vec<u8>>,
+    ) -> CustomResult<Vec<domain::UserKeyStore>, errors::StorageError> {
+        let user_key_store = self.user_key_store.lock().await;
+
+        futures::future::try_join_all(user_key_store.iter().map(|user_key| async {
+            user_key
+                .to_owned()
+                .convert(key)
+                .await
+                .change_context(errors::StorageError::DecryptionError)
+        }))
+        .await
+    }
+
     #[instrument(skip_all)]
     async fn get_user_key_store_by_user_id(
         &self,

crates/router/src/routes/admin.rs

@@ -446,16 +446,16 @@ pub async fn merchant_account_toggle_kv(
     .await
 }
 
-/// Merchant Account - Toggle KV
+/// Merchant Account - Transfer Keys
 ///
-/// Toggle KV mode for all Merchant Accounts
+/// Transfer Merchant Encryption key to keymanager
 #[instrument(skip_all)]
 pub async fn merchant_account_toggle_all_kv(
     state: web::Data<AppState>,
     req: HttpRequest,
     json_payload: web::Json<admin::ToggleAllKVRequest>,
 ) -> HttpResponse {
-    let flow = Flow::ConfigKeyUpdate;
+    let flow = Flow::MerchantTransferKey;
     let payload = json_payload.into_inner();
 
     api::server_wrap(
@@ -651,6 +651,27 @@ pub async fn merchant_account_kv_status(
     .await
 }
 
+/// Merchant Account - KV Status
+///
+/// Toggle KV mode for the Merchant Account
+#[instrument(skip_all)]
+pub async fn merchant_account_transfer_keys(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+) -> HttpResponse {
+    let flow = Flow::ConfigKeyFetch;
+    api::server_wrap(
+        flow,
+        state,
+        &req,
+        (),
+        |state, _, _, _| transfer_key_store_to_key_manager(state),
+        &auth::AdminApiAuth,
+        api_locking::LockAction::NotApplicable,
+    )
+    .await
+}
+
 #[instrument(skip_all, fields(flow = ?Flow::ToggleExtendedCardInfo))]
 pub async fn toggle_extended_card_info(
     state: web::Data<AppState>,

crates/router/src/routes/app.rs

@@ -1028,6 +1028,9 @@ impl MerchantAccount {
                     .route(web::post().to(merchant_account_toggle_kv))
                     .route(web::get().to(merchant_account_kv_status)),
             )
+            .service(
+                web::resource("/transfer").route(web::post().to(merchant_account_transfer_keys)),
+            )
             .service(web::resource("/kv").route(web::post().to(merchant_account_toggle_all_kv)))
             .service(
                 web::resource("/{id}")
@@ -1398,6 +1401,11 @@ impl User {
                     .route(web::post().to(set_dashboard_metadata)),
             );
 
+        route = route.service(
+            web::scope("/key")
+                .service(web::resource("/transfer").route(web::post().to(transfer_user_key))),
+        );
+
         // Two factor auth routes
         route = route.service(
             web::scope("/2fa")

crates/router/src/routes/lock_utils.rs

@@ -45,6 +45,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::MerchantsAccountRetrieve
             | Flow::MerchantsAccountUpdate
             | Flow::MerchantsAccountDelete
+            | Flow::MerchantTransferKey
             | Flow::MerchantAccountList => Self::MerchantAccount,
 
             Flow::RoutingCreateConfig
@@ -233,6 +234,7 @@ impl From<Flow> for ApiIdentifier {
             | Flow::CreateUserAuthenticationMethod
             | Flow::UpdateUserAuthenticationMethod
             | Flow::ListUserAuthenticationMethods
+            | Flow::UserTransferKey
             | Flow::GetSsoAuthUrl
             | Flow::SignInWithSso
             | Flow::AuthSelect => Self::User,

crates/router/src/routes/user.rs

@@ -895,3 +895,18 @@ pub async fn terminate_auth_select(
     ))
     .await
 }
+
+pub async fn transfer_user_key(state: web::Data<AppState>, req: HttpRequest) -> HttpResponse {
+    let flow = Flow::UserTransferKey;
+
+    Box::pin(api::server_wrap(
+        flow,
+        state.clone(),
+        &req,
+        (),
+        |state, _, _, _| user_core::transfer_user_key_store_keymanager(state),
+        &auth::AdminApiAuth,
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}

crates/router/src/types/domain/types.rs

@@ -1,3 +1,18 @@
+use common_utils::types::keymanager::KeyManagerState;
 pub use hyperswitch_domain_models::type_encryption::{
     decrypt, encrypt, encrypt_optional, AsyncLift, Lift, TypeEncryption,
 };
+
+impl From<&crate::SessionState> for KeyManagerState {
+    fn from(state: &crate::SessionState) -> Self {
+        let conf = state.conf.key_manager.get_inner();
+        Self {
+            url: conf.url.clone(),
+            client_idle_timeout: state.conf.proxy.idle_pool_connection_timeout,
+            #[cfg(feature = "keymanager_mtls")]
+            cert: conf.cert.clone(),
+            #[cfg(feature = "keymanager_mtls")]
+            ca: conf.ca.clone(),
+        }
+    }
+}

crates/router/src/types/domain/user.rs

@@ -6,6 +6,8 @@ use api_models::{
 use common_enums::TokenPurpose;
 #[cfg(not(feature = "v2"))]
 use common_utils::id_type;
+#[cfg(feature = "keymanager_create")]
+use common_utils::types::keymanager::{EncryptionCreateRequest, Identifier};
 use common_utils::{crypto::Encryptable, errors::CustomResult, new_type::MerchantName, pii};
 use diesel_models::{
     enums::{TotpStatus, UserStatus},
@@ -971,6 +973,19 @@ impl UserFromStorage {
                     .change_context(UserErrors::InternalServerError)?,
                 created_at: common_utils::date_time::now(),
             };
+
+            #[cfg(feature = "keymanager_create")]
+            {
+                common_utils::keymanager::create_key_in_key_manager(
+                    &state.into(),
+                    EncryptionCreateRequest {
+                        identifier: Identifier::User(key_store.user_id.clone()),
+                    },
+                )
+                .await
+                .change_context(UserErrors::InternalServerError)?;
+            }
+
             state
                 .global_store
                 .insert_user_key_store(key_store, &master_key.to_vec().into())

crates/router_env/src/logger/types.rs

@@ -81,6 +81,8 @@ pub enum Flow {
     MerchantConnectorsDelete,
     /// Merchant Connectors list flow.
     MerchantConnectorsList,
+    /// Merchant Transfer Keys
+    MerchantTransferKey,
     /// ConfigKey create flow.
     ConfigKeyCreate,
     /// ConfigKey fetch flow.
@@ -316,6 +318,8 @@ pub enum Flow {
     UserSignUpWithMerchantId,
     /// User Sign In
     UserSignIn,
+    /// User transfer key
+    UserTransferKey,
     /// User connect account
     UserConnectAccount,
     /// Upsert Decision Manager Config

loadtest/config/development.toml

@@ -9,6 +9,9 @@ traces_enabled = true
 metrics_enabled = true
 ignore_errors = false
 
+[key_manager]
+url = "http://localhost:5000"
+
 [master_database]
 username = "postgres"
 password = "postgres"