Bug: [FEATURE] Migrate PaymentMethodAuth to OSS



### Feature Description

Migrating PM auth services to OSS

### Possible Implementation

Migration of PM auth crate and dependent changes

### Have you spent some time checking if this feature request has been raised before?

- [X] I checked and didn't find a similar issue

### Have you read the Contributing Guidelines?

- [X] I have read the [Contributing Guidelines](https://github.com/juspay/hyperswitch/blob/main/docs/CONTRIBUTING.md)

### Are you willing to submit a PR?

Yes, I am willing to submit a PR!
diff --git a/Cargo.lock b/Cargo.lock
index d2e8d9dd5df..307a5ca2398 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -405,6 +405,8 @@ dependencies = [
  "common_utils",
  "error-stack",
  "euclid",
+ "frunk",
+ "frunk_core",
  "masking",
  "mime",
  "reqwest",
@@ -4436,6 +4438,27 @@ dependencies = [
  "plotters-backend",
 ]
 
+[[package]]
+name = "pm_auth"
+version = "0.1.0"
+dependencies = [
+ "api_models",
+ "async-trait",
+ "bytes 1.5.0",
+ "common_enums",
+ "common_utils",
+ "error-stack",
+ "http",
+ "masking",
+ "mime",
+ "router_derive",
+ "router_env",
+ "serde",
+ "serde_json",
+ "strum 0.24.1",
+ "thiserror",
+]
+
 [[package]]
 name = "png"
 version = "0.16.8"
@@ -5110,6 +5133,7 @@ dependencies = [
  "num_cpus",
  "once_cell",
  "openssl",
+ "pm_auth",
  "qrcode",
  "rand 0.8.5",
  "rand_chacha 0.3.1",
diff --git a/config/config.example.toml b/config/config.example.toml
index 1897c935581..7a50c23f484 100644
--- a/config/config.example.toml
+++ b/config/config.example.toml
@@ -122,7 +122,7 @@ kms_encrypted_recon_admin_api_key = ""   # Base64-encoded (KMS encrypted) cipher
 # like card details
 [locker]
 host = ""                      # Locker host
-host_rs = ""                   # Rust Locker host 
+host_rs = ""                   # Rust Locker host
 mock_locker = true             # Emulate a locker locally using Postgres
 basilisk_host = ""             # Basilisk host
 locker_signing_key_id = "1"    # Key_id to sign basilisk hs locker
@@ -461,6 +461,10 @@ apple_pay_merchant_cert_key = "APPLE_PAY_MERCHNAT_CERTIFICATE_KEY" #Private key
 [payment_link]
 sdk_url = "http://localhost:9090/dist/HyperLoader.js"
 
+[payment_method_auth]
+redis_expiry = 900
+pm_auth_key = "Some_pm_auth_key"
+
 # Analytics configuration.
 [analytics]
 source = "sqlx" # The Analytics source/strategy to be used
diff --git a/config/development.toml b/config/development.toml
index 4ee33795676..15acfdee9b7 100644
--- a/config/development.toml
+++ b/config/development.toml
@@ -470,6 +470,10 @@ apple_pay_merchant_cert_key = "APPLE_PAY_MERCHNAT_CERTIFICATE_KEY"
 [payment_link]
 sdk_url = "http://localhost:9090/dist/HyperLoader.js"
 
+[payment_method_auth]
+redis_expiry = 900
+pm_auth_key = "Some_pm_auth_key"
+
 [lock_settings]
 redis_lock_expiry_seconds = 180 # 3 * 60 seconds
 delay_between_retries_in_milliseconds = 500
diff --git a/config/docker_compose.toml b/config/docker_compose.toml
index 55fc62329d4..5eec8d733d6 100644
--- a/config/docker_compose.toml
+++ b/config/docker_compose.toml
@@ -330,6 +330,10 @@ payout_connector_list = "wise"
 [multiple_api_version_supported_connectors]
 supported_connectors = "braintree"
 
+[payment_method_auth]
+redis_expiry = 900
+pm_auth_key = "Some_pm_auth_key"
+
 [lock_settings]
 redis_lock_expiry_seconds = 180 # 3 * 60 seconds
 delay_between_retries_in_milliseconds = 500
diff --git a/crates/api_models/Cargo.toml b/crates/api_models/Cargo.toml
index 116aad25d5c..afba129b601 100644
--- a/crates/api_models/Cargo.toml
+++ b/crates/api_models/Cargo.toml
@@ -30,6 +30,8 @@ strum = { version = "0.25", features = ["derive"] }
 time = { version = "0.3.21", features = ["serde", "serde-well-known", "std"] }
 url = { version = "2.4.0", features = ["serde"] }
 utoipa = { version = "3.3.0", features = ["preserve_order"] }
+frunk = "0.4.1"
+frunk_core = "0.4.1"
 
 # First party crates
 cards = { version = "0.1.0", path = "../cards" }
diff --git a/crates/api_models/src/enums.rs b/crates/api_models/src/enums.rs
index 17787929a46..21586054055 100644
--- a/crates/api_models/src/enums.rs
+++ b/crates/api_models/src/enums.rs
@@ -1,3 +1,5 @@
+use std::str::FromStr;
+
 pub use common_enums::*;
 use utoipa::ToSchema;
 
@@ -500,3 +502,26 @@ pub enum LockerChoice {
     Basilisk,
     Tartarus,
 }
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Eq,
+    PartialEq,
+    serde::Serialize,
+    serde::Deserialize,
+    strum::Display,
+    strum::EnumString,
+    frunk::LabelledGeneric,
+    ToSchema,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum PmAuthConnectors {
+    Plaid,
+}
+
+pub fn convert_pm_auth_connector(connector_name: &str) -> Option<PmAuthConnectors> {
+    PmAuthConnectors::from_str(connector_name).ok()
+}
diff --git a/crates/api_models/src/lib.rs b/crates/api_models/src/lib.rs
index ce3c11d9c2f..935944cf74c 100644
--- a/crates/api_models/src/lib.rs
+++ b/crates/api_models/src/lib.rs
@@ -23,6 +23,7 @@ pub mod payment_methods;
 pub mod payments;
 #[cfg(feature = "payouts")]
 pub mod payouts;
+pub mod pm_auth;
 pub mod refunds;
 pub mod routing;
 pub mod surcharge_decision_configs;
diff --git a/crates/api_models/src/pm_auth.rs b/crates/api_models/src/pm_auth.rs
new file mode 100644
index 00000000000..7044bd8d335
--- /dev/null
+++ b/crates/api_models/src/pm_auth.rs
@@ -0,0 +1,57 @@
+use common_enums::{PaymentMethod, PaymentMethodType};
+use common_utils::{
+    events::{ApiEventMetric, ApiEventsType},
+    impl_misc_api_event_type,
+};
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+pub struct LinkTokenCreateRequest {
+    pub language: Option<String>, // optional language field to be passed
+    pub client_secret: Option<String>, // client secret to be passed in req body
+    pub payment_id: String, // payment_id to be passed in req body for redis pm_auth connector name fetch
+    pub payment_method: PaymentMethod, // payment_method to be used for filtering pm_auth connector
+    pub payment_method_type: PaymentMethodType, // payment_method_type to be used for filtering pm_auth connector
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct LinkTokenCreateResponse {
+    pub link_token: String, // link_token received in response
+    pub connector: String,  // pm_auth connector name in response
+}
+
+#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "snake_case")]
+
+pub struct ExchangeTokenCreateRequest {
+    pub public_token: String,
+    pub client_secret: Option<String>,
+    pub payment_id: String,
+    pub payment_method: PaymentMethod,
+    pub payment_method_type: PaymentMethodType,
+}
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct ExchangeTokenCreateResponse {
+    pub access_token: String,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct PaymentMethodAuthConfig {
+    pub enabled_payment_methods: Vec<PaymentMethodAuthConnectorChoice>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct PaymentMethodAuthConnectorChoice {
+    pub payment_method: PaymentMethod,
+    pub payment_method_type: PaymentMethodType,
+    pub connector_name: String,
+    pub mca_id: String,
+}
+
+impl_misc_api_event_type!(
+    LinkTokenCreateRequest,
+    LinkTokenCreateResponse,
+    ExchangeTokenCreateRequest,
+    ExchangeTokenCreateResponse
+);
diff --git a/crates/pm_auth/Cargo.toml b/crates/pm_auth/Cargo.toml
new file mode 100644
index 00000000000..a9aebc5b540
--- /dev/null
+++ b/crates/pm_auth/Cargo.toml
@@ -0,0 +1,27 @@
+[package]
+name = "pm_auth"
+description = "Open banking services"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+readme = "README.md"
+
+[dependencies]
+# First party crates
+api_models = { version = "0.1.0", path = "../api_models" }
+common_enums = { version = "0.1.0", path = "../common_enums" }
+common_utils = { version = "0.1.0", path = "../common_utils" }
+masking = { version = "0.1.0", path = "../masking" }
+router_derive = { version = "0.1.0", path = "../router_derive" }
+router_env = { version = "0.1.0", path = "../router_env", features = ["log_extra_implicit_fields", "log_custom_entries_to_extra"] }
+
+# Third party crates
+async-trait = "0.1.66"
+bytes = "1.4.0"
+error-stack = "0.3.1"
+http = "0.2.9"
+mime = "0.3.17"
+serde = "1.0.159"
+serde_json = "1.0.91"
+strum = { version = "0.24.1", features = ["derive"] }
+thiserror = "1.0.43"
diff --git a/crates/pm_auth/README.md b/crates/pm_auth/README.md
new file mode 100644
index 00000000000..c630a7fe676
--- /dev/null
+++ b/crates/pm_auth/README.md
@@ -0,0 +1,3 @@
+# Payment Method Auth Services
+
+An open banking services for payment method auth validation
diff --git a/crates/pm_auth/src/connector.rs b/crates/pm_auth/src/connector.rs
new file mode 100644
index 00000000000..56aad846e24
--- /dev/null
+++ b/crates/pm_auth/src/connector.rs
@@ -0,0 +1,3 @@
+pub mod plaid;
+
+pub use self::plaid::Plaid;
diff --git a/crates/pm_auth/src/connector/plaid.rs b/crates/pm_auth/src/connector/plaid.rs
new file mode 100644
index 00000000000..d25aba881d2
--- /dev/null
+++ b/crates/pm_auth/src/connector/plaid.rs
@@ -0,0 +1,353 @@
+pub mod transformers;
+
+use std::fmt::Debug;
+
+use common_utils::{
+    ext_traits::{BytesExt, Encode},
+    request::{Method, Request, RequestBody, RequestBuilder},
+};
+use error_stack::ResultExt;
+use masking::{Mask, Maskable};
+use transformers as plaid;
+
+use crate::{
+    core::errors,
+    types::{
+        self as auth_types,
+        api::{
+            auth_service::{self, BankAccountCredentials, ExchangeToken, LinkToken},
+            ConnectorCommon, ConnectorCommonExt, ConnectorIntegration,
+        },
+    },
+};
+
+#[derive(Debug, Clone)]
+pub struct Plaid;
+
+impl<Flow, Request, Response> ConnectorCommonExt<Flow, Request, Response> for Plaid
+where
+    Self: ConnectorIntegration<Flow, Request, Response>,
+{
+    fn build_headers(
+        &self,
+        req: &auth_types::PaymentAuthRouterData<Flow, Request, Response>,
+        _connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
+        let mut header = vec![(
+            "Content-Type".to_string(),
+            self.get_content_type().to_string().into(),
+        )];
+
+        let mut auth = self.get_auth_header(&req.connector_auth_type)?;
+        header.append(&mut auth);
+        Ok(header)
+    }
+}
+
+impl ConnectorCommon for Plaid {
+    fn id(&self) -> &'static str {
+        "plaid"
+    }
+
+    fn common_get_content_type(&self) -> &'static str {
+        "application/json"
+    }
+    fn base_url<'a>(&self, _connectors: &'a auth_types::PaymentMethodAuthConnectors) -> &'a str {
+        "https://sandbox.plaid.com"
+    }
+
+    fn get_auth_header(
+        &self,
+        auth_type: &auth_types::ConnectorAuthType,
+    ) -> errors::CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
+        let auth = plaid::PlaidAuthType::try_from(auth_type)
+            .change_context(errors::ConnectorError::FailedToObtainAuthType)?;
+        let client_id = auth.client_id.into_masked();
+        let secret = auth.secret.into_masked();
+
+        Ok(vec![
+            ("PLAID-CLIENT-ID".to_string(), client_id),
+            ("PLAID-SECRET".to_string(), secret),
+        ])
+    }
+
+    fn build_error_response(
+        &self,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::ErrorResponse, errors::ConnectorError> {
+        let response: plaid::PlaidErrorResponse =
+            res.response
+                .parse_struct("PlaidErrorResponse")
+                .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        Ok(auth_types::ErrorResponse {
+            status_code: res.status_code,
+            code: crate::consts::NO_ERROR_CODE.to_string(),
+            message: response.error_message,
+            reason: response.display_message,
+        })
+    }
+}
+
+impl auth_service::AuthService for Plaid {}
+impl auth_service::AuthServiceLinkToken for Plaid {}
+
+impl ConnectorIntegration<LinkToken, auth_types::LinkTokenRequest, auth_types::LinkTokenResponse>
+    for Plaid
+{
+    fn get_headers(
+        &self,
+        req: &auth_types::LinkTokenRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        _req: &auth_types::LinkTokenRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<String, errors::ConnectorError> {
+        Ok(format!(
+            "{}{}",
+            self.base_url(connectors),
+            "/link/token/create"
+        ))
+    }
+
+    fn get_request_body(
+        &self,
+        req: &auth_types::LinkTokenRouterData,
+    ) -> errors::CustomResult<Option<RequestBody>, errors::ConnectorError> {
+        let req_obj = plaid::PlaidLinkTokenRequest::try_from(req)?;
+        let plaid_req = RequestBody::log_and_get_request_body(
+            &req_obj,
+            Encode::<plaid::PlaidLinkTokenRequest>::encode_to_string_of_json,
+        )
+        .change_context(errors::ConnectorError::RequestEncodingFailed)?;
+        Ok(Some(plaid_req))
+    }
+
+    fn build_request(
+        &self,
+        req: &auth_types::LinkTokenRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Post)
+                .url(&auth_types::PaymentAuthLinkTokenType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(auth_types::PaymentAuthLinkTokenType::get_headers(
+                    self, req, connectors,
+                )?)
+                .body(auth_types::PaymentAuthLinkTokenType::get_request_body(
+                    self, req,
+                )?)
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &auth_types::LinkTokenRouterData,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::LinkTokenRouterData, errors::ConnectorError> {
+        let response: plaid::PlaidLinkTokenResponse = res
+            .response
+            .parse_struct("PlaidLinkTokenResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        <auth_types::LinkTokenRouterData>::try_from(auth_types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+    fn get_error_response(
+        &self,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res)
+    }
+}
+
+impl auth_service::AuthServiceExchangeToken for Plaid {}
+
+impl
+    ConnectorIntegration<
+        ExchangeToken,
+        auth_types::ExchangeTokenRequest,
+        auth_types::ExchangeTokenResponse,
+    > for Plaid
+{
+    fn get_headers(
+        &self,
+        req: &auth_types::ExchangeTokenRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        _req: &auth_types::ExchangeTokenRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<String, errors::ConnectorError> {
+        Ok(format!(
+            "{}{}",
+            self.base_url(connectors),
+            "/item/public_token/exchange"
+        ))
+    }
+
+    fn get_request_body(
+        &self,
+        req: &auth_types::ExchangeTokenRouterData,
+    ) -> errors::CustomResult<Option<RequestBody>, errors::ConnectorError> {
+        let req_obj = plaid::PlaidExchangeTokenRequest::try_from(req)?;
+        let plaid_req = RequestBody::log_and_get_request_body(
+            &req_obj,
+            Encode::<plaid::PlaidExchangeTokenRequest>::encode_to_string_of_json,
+        )
+        .change_context(errors::ConnectorError::RequestEncodingFailed)?;
+        Ok(Some(plaid_req))
+    }
+
+    fn build_request(
+        &self,
+        req: &auth_types::ExchangeTokenRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Post)
+                .url(&auth_types::PaymentAuthExchangeTokenType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(auth_types::PaymentAuthExchangeTokenType::get_headers(
+                    self, req, connectors,
+                )?)
+                .body(auth_types::PaymentAuthExchangeTokenType::get_request_body(
+                    self, req,
+                )?)
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &auth_types::ExchangeTokenRouterData,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::ExchangeTokenRouterData, errors::ConnectorError> {
+        let response: plaid::PlaidExchangeTokenResponse = res
+            .response
+            .parse_struct("PlaidExchangeTokenResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        <auth_types::ExchangeTokenRouterData>::try_from(auth_types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+    fn get_error_response(
+        &self,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res)
+    }
+}
+
+impl auth_service::AuthServiceBankAccountCredentials for Plaid {}
+
+impl
+    ConnectorIntegration<
+        BankAccountCredentials,
+        auth_types::BankAccountCredentialsRequest,
+        auth_types::BankAccountCredentialsResponse,
+    > for Plaid
+{
+    fn get_headers(
+        &self,
+        req: &auth_types::BankDetailsRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Vec<(String, Maskable<String>)>, errors::ConnectorError> {
+        self.build_headers(req, connectors)
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        self.common_get_content_type()
+    }
+
+    fn get_url(
+        &self,
+        _req: &auth_types::BankDetailsRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<String, errors::ConnectorError> {
+        Ok(format!("{}{}", self.base_url(connectors), "/auth/get"))
+    }
+
+    fn get_request_body(
+        &self,
+        req: &auth_types::BankDetailsRouterData,
+    ) -> errors::CustomResult<Option<RequestBody>, errors::ConnectorError> {
+        let req_obj = plaid::PlaidBankAccountCredentialsRequest::try_from(req)?;
+        let plaid_req = RequestBody::log_and_get_request_body(
+            &req_obj,
+            Encode::<plaid::PlaidBankAccountCredentialsRequest>::encode_to_string_of_json,
+        )
+        .change_context(errors::ConnectorError::RequestEncodingFailed)?;
+        Ok(Some(plaid_req))
+    }
+
+    fn build_request(
+        &self,
+        req: &auth_types::BankDetailsRouterData,
+        connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<Option<Request>, errors::ConnectorError> {
+        Ok(Some(
+            RequestBuilder::new()
+                .method(Method::Post)
+                .url(&auth_types::PaymentAuthBankAccountDetailsType::get_url(
+                    self, req, connectors,
+                )?)
+                .attach_default_headers()
+                .headers(auth_types::PaymentAuthBankAccountDetailsType::get_headers(
+                    self, req, connectors,
+                )?)
+                .body(auth_types::PaymentAuthBankAccountDetailsType::get_request_body(self, req)?)
+                .build(),
+        ))
+    }
+
+    fn handle_response(
+        &self,
+        data: &auth_types::BankDetailsRouterData,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::BankDetailsRouterData, errors::ConnectorError> {
+        let response: plaid::PlaidBankAccountCredentialsResponse = res
+            .response
+            .parse_struct("PlaidBankAccountCredentialsResponse")
+            .change_context(errors::ConnectorError::ResponseDeserializationFailed)?;
+        <auth_types::BankDetailsRouterData>::try_from(auth_types::ResponseRouterData {
+            response,
+            data: data.clone(),
+            http_code: res.status_code,
+        })
+    }
+    fn get_error_response(
+        &self,
+        res: auth_types::Response,
+    ) -> errors::CustomResult<auth_types::ErrorResponse, errors::ConnectorError> {
+        self.build_error_response(res)
+    }
+}
diff --git a/crates/pm_auth/src/connector/plaid/transformers.rs b/crates/pm_auth/src/connector/plaid/transformers.rs
new file mode 100644
index 00000000000..5e1ad67aead
--- /dev/null
+++ b/crates/pm_auth/src/connector/plaid/transformers.rs
@@ -0,0 +1,294 @@
+use std::collections::HashMap;
+
+use common_enums::PaymentMethodType;
+use masking::Secret;
+use serde::{Deserialize, Serialize};
+
+use crate::{core::errors, types};
+
+#[derive(Debug, Serialize, Eq, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub struct PlaidLinkTokenRequest {
+    client_name: String,
+    country_codes: Vec<String>,
+    language: String,
+    products: Vec<String>,
+    user: User,
+}
+
+#[derive(Debug, Serialize, Eq, PartialEq)]
+
+pub struct User {
+    pub client_user_id: String,
+}
+
+impl TryFrom<&types::LinkTokenRouterData> for PlaidLinkTokenRequest {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(item: &types::LinkTokenRouterData) -> Result<Self, Self::Error> {
+        Ok(Self {
+            client_name: item.request.client_name.clone(),
+            country_codes: item.request.country_codes.clone().ok_or(
+                errors::ConnectorError::MissingRequiredField {
+                    field_name: "country_codes",
+                },
+            )?,
+            language: item.request.language.clone().unwrap_or("en".to_string()),
+            products: vec!["auth".to_string()],
+            user: User {
+                client_user_id: item.request.user_info.clone().ok_or(
+                    errors::ConnectorError::MissingRequiredField {
+                        field_name: "country_codes",
+                    },
+                )?,
+            },
+        })
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub struct PlaidLinkTokenResponse {
+    link_token: String,
+}
+
+impl<F, T>
+    TryFrom<types::ResponseRouterData<F, PlaidLinkTokenResponse, T, types::LinkTokenResponse>>
+    for types::PaymentAuthRouterData<F, T, types::LinkTokenResponse>
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: types::ResponseRouterData<F, PlaidLinkTokenResponse, T, types::LinkTokenResponse>,
+    ) -> Result<Self, Self::Error> {
+        Ok(Self {
+            response: Ok(types::LinkTokenResponse {
+                link_token: item.response.link_token,
+            }),
+            ..item.data
+        })
+    }
+}
+
+#[derive(Debug, Serialize, Eq, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub struct PlaidExchangeTokenRequest {
+    public_token: String,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+
+pub struct PlaidExchangeTokenResponse {
+    pub access_token: String,
+}
+
+impl<F, T>
+    TryFrom<
+        types::ResponseRouterData<F, PlaidExchangeTokenResponse, T, types::ExchangeTokenResponse>,
+    > for types::PaymentAuthRouterData<F, T, types::ExchangeTokenResponse>
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: types::ResponseRouterData<
+            F,
+            PlaidExchangeTokenResponse,
+            T,
+            types::ExchangeTokenResponse,
+        >,
+    ) -> Result<Self, Self::Error> {
+        Ok(Self {
+            response: Ok(types::ExchangeTokenResponse {
+                access_token: item.response.access_token,
+            }),
+            ..item.data
+        })
+    }
+}
+
+impl TryFrom<&types::ExchangeTokenRouterData> for PlaidExchangeTokenRequest {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(item: &types::ExchangeTokenRouterData) -> Result<Self, Self::Error> {
+        Ok(Self {
+            public_token: item.request.public_token.clone(),
+        })
+    }
+}
+
+#[derive(Debug, Serialize, Eq, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub struct PlaidBankAccountCredentialsRequest {
+    access_token: String,
+    options: Option<BankAccountCredentialsOptions>,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+
+pub struct PlaidBankAccountCredentialsResponse {
+    pub accounts: Vec<PlaidBankAccountCredentialsAccounts>,
+    pub numbers: PlaidBankAccountCredentialsNumbers,
+    // pub item: PlaidBankAccountCredentialsItem,
+    pub request_id: String,
+}
+
+#[derive(Debug, Serialize, Eq, PartialEq)]
+#[serde(rename_all = "snake_case")]
+pub struct BankAccountCredentialsOptions {
+    account_ids: Vec<String>,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+
+pub struct PlaidBankAccountCredentialsAccounts {
+    pub account_id: String,
+    pub name: String,
+    pub subtype: Option<String>,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidBankAccountCredentialsBalances {
+    pub available: Option<i32>,
+    pub current: Option<i32>,
+    pub limit: Option<i32>,
+    pub iso_currency_code: Option<String>,
+    pub unofficial_currency_code: Option<String>,
+    pub last_updated_datetime: Option<String>,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidBankAccountCredentialsNumbers {
+    pub ach: Vec<PlaidBankAccountCredentialsACH>,
+    pub eft: Vec<PlaidBankAccountCredentialsEFT>,
+    pub international: Vec<PlaidBankAccountCredentialsInternational>,
+    pub bacs: Vec<PlaidBankAccountCredentialsBacs>,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidBankAccountCredentialsItem {
+    pub item_id: String,
+    pub institution_id: Option<String>,
+    pub webhook: Option<String>,
+    pub error: Option<PlaidErrorResponse>,
+}
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidBankAccountCredentialsACH {
+    pub account_id: String,
+    pub account: String,
+    pub routing: String,
+    pub wire_routing: Option<String>,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidBankAccountCredentialsEFT {
+    pub account_id: String,
+    pub account: String,
+    pub institution: String,
+    pub branch: String,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidBankAccountCredentialsInternational {
+    pub account_id: String,
+    pub iban: String,
+    pub bic: String,
+}
+
+#[derive(Debug, Deserialize, Eq, PartialEq)]
+pub struct PlaidBankAccountCredentialsBacs {
+    pub account_id: String,
+    pub account: String,
+    pub sort_code: String,
+}
+
+impl TryFrom<&types::BankDetailsRouterData> for PlaidBankAccountCredentialsRequest {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(item: &types::BankDetailsRouterData) -> Result<Self, Self::Error> {
+        Ok(Self {
+            access_token: item.request.access_token.clone(),
+            options: item.request.optional_ids.as_ref().map(|bank_account_ids| {
+                BankAccountCredentialsOptions {
+                    account_ids: bank_account_ids.ids.clone(),
+                }
+            }),
+        })
+    }
+}
+
+impl<F, T>
+    TryFrom<
+        types::ResponseRouterData<
+            F,
+            PlaidBankAccountCredentialsResponse,
+            T,
+            types::BankAccountCredentialsResponse,
+        >,
+    > for types::PaymentAuthRouterData<F, T, types::BankAccountCredentialsResponse>
+{
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(
+        item: types::ResponseRouterData<
+            F,
+            PlaidBankAccountCredentialsResponse,
+            T,
+            types::BankAccountCredentialsResponse,
+        >,
+    ) -> Result<Self, Self::Error> {
+        let (account_numbers, accounts_info) = (item.response.numbers, item.response.accounts);
+        let mut bank_account_vec = Vec::new();
+        let mut id_to_suptype = HashMap::new();
+
+        accounts_info.into_iter().for_each(|acc| {
+            id_to_suptype.insert(acc.account_id, (acc.subtype, acc.name));
+        });
+
+        account_numbers.ach.into_iter().for_each(|ach| {
+            let (acc_type, acc_name) =
+                if let Some((_type, name)) = id_to_suptype.get(&ach.account_id) {
+                    (_type.to_owned(), Some(name.clone()))
+                } else {
+                    (None, None)
+                };
+
+            let bank_details_new = types::BankAccountDetails {
+                account_name: acc_name,
+                account_number: ach.account,
+                routing_number: ach.routing,
+                payment_method_type: PaymentMethodType::Ach,
+                account_id: ach.account_id,
+                account_type: acc_type,
+            };
+
+            bank_account_vec.push(bank_details_new);
+        });
+
+        Ok(Self {
+            response: Ok(types::BankAccountCredentialsResponse {
+                credentials: bank_account_vec,
+            }),
+            ..item.data
+        })
+    }
+}
+pub struct PlaidAuthType {
+    pub client_id: Secret<String>,
+    pub secret: Secret<String>,
+}
+
+impl TryFrom<&types::ConnectorAuthType> for PlaidAuthType {
+    type Error = error_stack::Report<errors::ConnectorError>;
+    fn try_from(auth_type: &types::ConnectorAuthType) -> Result<Self, Self::Error> {
+        match auth_type {
+            types::ConnectorAuthType::BodyKey { client_id, secret } => Ok(Self {
+                client_id: client_id.to_owned(),
+                secret: secret.to_owned(),
+            }),
+            _ => Err(errors::ConnectorError::FailedToObtainAuthType.into()),
+        }
+    }
+}
+
+#[derive(Debug, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub struct PlaidErrorResponse {
+    pub display_message: Option<String>,
+    pub error_code: Option<String>,
+    pub error_message: String,
+    pub error_type: Option<String>,
+}
diff --git a/crates/pm_auth/src/consts.rs b/crates/pm_auth/src/consts.rs
new file mode 100644
index 00000000000..dac3485ec8f
--- /dev/null
+++ b/crates/pm_auth/src/consts.rs
@@ -0,0 +1,5 @@
+pub const REQUEST_TIME_OUT: u64 = 30; // will timeout after the mentioned limit
+pub const REQUEST_TIMEOUT_ERROR_CODE: &str = "TIMEOUT"; // timeout error code
+pub const REQUEST_TIMEOUT_ERROR_MESSAGE: &str = "Connector did not respond in specified time"; // error message for timed out request
+pub const NO_ERROR_CODE: &str = "No error code";
+pub const NO_ERROR_MESSAGE: &str = "No error message";
diff --git a/crates/pm_auth/src/core.rs b/crates/pm_auth/src/core.rs
new file mode 100644
index 00000000000..629e98fbf87
--- /dev/null
+++ b/crates/pm_auth/src/core.rs
@@ -0,0 +1 @@
+pub mod errors;
diff --git a/crates/pm_auth/src/core/errors.rs b/crates/pm_auth/src/core/errors.rs
new file mode 100644
index 00000000000..31b178a6276
--- /dev/null
+++ b/crates/pm_auth/src/core/errors.rs
@@ -0,0 +1,27 @@
+#[derive(Debug, thiserror::Error, PartialEq)]
+pub enum ConnectorError {
+    #[error("Failed to obtain authentication type")]
+    FailedToObtainAuthType,
+    #[error("Missing required field: {field_name}")]
+    MissingRequiredField { field_name: &'static str },
+    #[error("Failed to execute a processing step: {0:?}")]
+    ProcessingStepFailed(Option<bytes::Bytes>),
+    #[error("Failed to deserialize connector response")]
+    ResponseDeserializationFailed,
+    #[error("Failed to encode connector request")]
+    RequestEncodingFailed,
+}
+
+pub type CustomResult<T, E> = error_stack::Result<T, E>;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ParsingError {
+    #[error("Failed to parse enum: {0}")]
+    EnumParseFailure(&'static str),
+    #[error("Failed to parse struct: {0}")]
+    StructParseFailure(&'static str),
+    #[error("Failed to serialize to {0} format")]
+    EncodeError(&'static str),
+    #[error("Unknown error while parsing")]
+    UnknownError,
+}
diff --git a/crates/pm_auth/src/lib.rs b/crates/pm_auth/src/lib.rs
new file mode 100644
index 00000000000..60d0e06a1e0
--- /dev/null
+++ b/crates/pm_auth/src/lib.rs
@@ -0,0 +1,4 @@
+pub mod connector;
+pub mod consts;
+pub mod core;
+pub mod types;
diff --git a/crates/pm_auth/src/types.rs b/crates/pm_auth/src/types.rs
new file mode 100644
index 00000000000..6f5875247f1
--- /dev/null
+++ b/crates/pm_auth/src/types.rs
@@ -0,0 +1,152 @@
+pub mod api;
+
+use std::marker::PhantomData;
+
+use api::auth_service::{BankAccountCredentials, ExchangeToken, LinkToken};
+use common_enums::PaymentMethodType;
+use masking::Secret;
+#[derive(Debug, Clone)]
+pub struct PaymentAuthRouterData<F, Request, Response> {
+    pub flow: PhantomData<F>,
+    pub merchant_id: Option<String>,
+    pub connector: Option<String>,
+    pub request: Request,
+    pub response: Result<Response, ErrorResponse>,
+    pub connector_auth_type: ConnectorAuthType,
+    pub connector_http_status_code: Option<u16>,
+}
+
+#[derive(Debug, Clone)]
+pub struct LinkTokenRequest {
+    pub client_name: String,
+    pub country_codes: Option<Vec<String>>,
+    pub language: Option<String>,
+    pub user_info: Option<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct LinkTokenResponse {
+    pub link_token: String,
+}
+
+pub type LinkTokenRouterData =
+    PaymentAuthRouterData<LinkToken, LinkTokenRequest, LinkTokenResponse>;
+
+#[derive(Debug, Clone)]
+pub struct ExchangeTokenRequest {
+    pub public_token: String,
+}
+
+#[derive(Debug, Clone)]
+pub struct ExchangeTokenResponse {
+    pub access_token: String,
+}
+
+impl From<ExchangeTokenResponse> for api_models::pm_auth::ExchangeTokenCreateResponse {
+    fn from(value: ExchangeTokenResponse) -> Self {
+        Self {
+            access_token: value.access_token,
+        }
+    }
+}
+
+pub type ExchangeTokenRouterData =
+    PaymentAuthRouterData<ExchangeToken, ExchangeTokenRequest, ExchangeTokenResponse>;
+
+#[derive(Debug, Clone)]
+pub struct BankAccountCredentialsRequest {
+    pub access_token: String,
+    pub optional_ids: Option<BankAccountOptionalIDs>,
+}
+
+#[derive(Debug, Clone)]
+pub struct BankAccountOptionalIDs {
+    pub ids: Vec<String>,
+}
+
+#[derive(Debug, Clone)]
+pub struct BankAccountCredentialsResponse {
+    pub credentials: Vec<BankAccountDetails>,
+}
+
+#[derive(Debug, Clone)]
+pub struct BankAccountDetails {
+    pub account_name: Option<String>,
+    pub account_number: String,
+    pub routing_number: String,
+    pub payment_method_type: PaymentMethodType,
+    pub account_id: String,
+    pub account_type: Option<String>,
+}
+
+pub type BankDetailsRouterData = PaymentAuthRouterData<
+    BankAccountCredentials,
+    BankAccountCredentialsRequest,
+    BankAccountCredentialsResponse,
+>;
+
+pub type PaymentAuthLinkTokenType =
+    dyn self::api::ConnectorIntegration<LinkToken, LinkTokenRequest, LinkTokenResponse>;
+
+pub type PaymentAuthExchangeTokenType =
+    dyn self::api::ConnectorIntegration<ExchangeToken, ExchangeTokenRequest, ExchangeTokenResponse>;
+
+pub type PaymentAuthBankAccountDetailsType = dyn self::api::ConnectorIntegration<
+    BankAccountCredentials,
+    BankAccountCredentialsRequest,
+    BankAccountCredentialsResponse,
+>;
+
+#[derive(Clone, Debug, strum::EnumString, strum::Display)]
+#[strum(serialize_all = "snake_case")]
+pub enum PaymentMethodAuthConnectors {
+    Plaid,
+}
+
+#[derive(Debug, Clone)]
+pub struct ResponseRouterData<Flow, R, Request, Response> {
+    pub response: R,
+    pub data: PaymentAuthRouterData<Flow, Request, Response>,
+    pub http_code: u16,
+}
+
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct ErrorResponse {
+    pub code: String,
+    pub message: String,
+    pub reason: Option<String>,
+    pub status_code: u16,
+}
+
+impl ErrorResponse {
+    fn get_not_implemented() -> Self {
+        Self {
+            code: "IR_00".to_string(),
+            message: "This API is under development and will be made available soon.".to_string(),
+            reason: None,
+            status_code: http::StatusCode::INTERNAL_SERVER_ERROR.as_u16(),
+        }
+    }
+}
+
+#[derive(Default, Debug, Clone, serde::Deserialize)]
+pub enum ConnectorAuthType {
+    BodyKey {
+        client_id: Secret<String>,
+        secret: Secret<String>,
+    },
+    #[default]
+    NoKey,
+}
+
+#[derive(Clone, Debug)]
+pub struct Response {
+    pub headers: Option<http::HeaderMap>,
+    pub response: bytes::Bytes,
+    pub status_code: u16,
+}
+
+#[derive(serde::Deserialize, Clone)]
+pub struct AuthServiceQueryParam {
+    pub client_secret: Option<String>,
+}
diff --git a/crates/pm_auth/src/types/api.rs b/crates/pm_auth/src/types/api.rs
new file mode 100644
index 00000000000..2416d0fee1d
--- /dev/null
+++ b/crates/pm_auth/src/types/api.rs
@@ -0,0 +1,167 @@
+pub mod auth_service;
+
+use std::fmt::Debug;
+
+use common_utils::{
+    errors::CustomResult,
+    request::{Request, RequestBody},
+};
+use masking::Maskable;
+
+use crate::{
+    core::errors::ConnectorError,
+    types::{self as auth_types, api::auth_service::AuthService},
+};
+
+#[async_trait::async_trait]
+pub trait ConnectorIntegration<T, Req, Resp>: ConnectorIntegrationAny<T, Req, Resp> + Sync {
+    fn get_headers(
+        &self,
+        _req: &super::PaymentAuthRouterData<T, Req, Resp>,
+        _connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> CustomResult<Vec<(String, Maskable<String>)>, ConnectorError> {
+        Ok(vec![])
+    }
+
+    fn get_content_type(&self) -> &'static str {
+        mime::APPLICATION_JSON.essence_str()
+    }
+
+    fn get_url(
+        &self,
+        _req: &super::PaymentAuthRouterData<T, Req, Resp>,
+        _connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> CustomResult<String, ConnectorError> {
+        Ok(String::new())
+    }
+
+    fn get_request_body(
+        &self,
+        _req: &super::PaymentAuthRouterData<T, Req, Resp>,
+    ) -> CustomResult<Option<RequestBody>, ConnectorError> {
+        Ok(None)
+    }
+
+    fn build_request(
+        &self,
+        _req: &super::PaymentAuthRouterData<T, Req, Resp>,
+        _connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> CustomResult<Option<Request>, ConnectorError> {
+        Ok(None)
+    }
+
+    fn handle_response(
+        &self,
+        data: &super::PaymentAuthRouterData<T, Req, Resp>,
+        _res: auth_types::Response,
+    ) -> CustomResult<super::PaymentAuthRouterData<T, Req, Resp>, ConnectorError>
+    where
+        T: Clone,
+        Req: Clone,
+        Resp: Clone,
+    {
+        Ok(data.clone())
+    }
+
+    fn get_error_response(
+        &self,
+        _res: auth_types::Response,
+    ) -> CustomResult<auth_types::ErrorResponse, ConnectorError> {
+        Ok(auth_types::ErrorResponse::get_not_implemented())
+    }
+
+    fn get_5xx_error_response(
+        &self,
+        res: auth_types::Response,
+    ) -> CustomResult<auth_types::ErrorResponse, ConnectorError> {
+        let error_message = match res.status_code {
+            500 => "internal_server_error",
+            501 => "not_implemented",
+            502 => "bad_gateway",
+            503 => "service_unavailable",
+            504 => "gateway_timeout",
+            505 => "http_version_not_supported",
+            506 => "variant_also_negotiates",
+            507 => "insufficient_storage",
+            508 => "loop_detected",
+            510 => "not_extended",
+            511 => "network_authentication_required",
+            _ => "unknown_error",
+        };
+        Ok(auth_types::ErrorResponse {
+            code: res.status_code.to_string(),
+            message: error_message.to_string(),
+            reason: String::from_utf8(res.response.to_vec()).ok(),
+            status_code: res.status_code,
+        })
+    }
+}
+
+pub trait ConnectorCommonExt<Flow, Req, Resp>:
+    ConnectorCommon + ConnectorIntegration<Flow, Req, Resp>
+{
+    fn build_headers(
+        &self,
+        _req: &auth_types::PaymentAuthRouterData<Flow, Req, Resp>,
+        _connectors: &auth_types::PaymentMethodAuthConnectors,
+    ) -> CustomResult<Vec<(String, Maskable<String>)>, ConnectorError> {
+        Ok(Vec::new())
+    }
+}
+
+pub type BoxedConnectorIntegration<'a, T, Req, Resp> =
+    Box<&'a (dyn ConnectorIntegration<T, Req, Resp> + Send + Sync)>;
+
+pub trait ConnectorIntegrationAny<T, Req, Resp>: Send + Sync + 'static {
+    fn get_connector_integration(&self) -> BoxedConnectorIntegration<'_, T, Req, Resp>;
+}
+
+impl<S, T, Req, Resp> ConnectorIntegrationAny<T, Req, Resp> for S
+where
+    S: ConnectorIntegration<T, Req, Resp>,
+{
+    fn get_connector_integration(&self) -> BoxedConnectorIntegration<'_, T, Req, Resp> {
+        Box::new(self)
+    }
+}
+
+pub trait AuthServiceConnector: AuthService + Send + Debug {}
+
+impl<T: Send + Debug + AuthService> AuthServiceConnector for T {}
+
+pub type BoxedPaymentAuthConnector = Box<&'static (dyn AuthServiceConnector + Sync)>;
+
+#[derive(Clone, Debug)]
+pub struct PaymentAuthConnectorData {
+    pub connector: BoxedPaymentAuthConnector,
+    pub connector_name: super::PaymentMethodAuthConnectors,
+}
+
+pub trait ConnectorCommon {
+    fn id(&self) -> &'static str;
+
+    fn get_auth_header(
+        &self,
+        _auth_type: &auth_types::ConnectorAuthType,
+    ) -> CustomResult<Vec<(String, Maskable<String>)>, ConnectorError> {
+        Ok(Vec::new())
+    }
+
+    fn common_get_content_type(&self) -> &'static str {
+        "application/json"
+    }
+
+    fn base_url<'a>(&self, connectors: &'a auth_types::PaymentMethodAuthConnectors) -> &'a str;
+
+    fn build_error_response(
+        &self,
+        res: auth_types::Response,
+    ) -> CustomResult<auth_types::ErrorResponse, ConnectorError> {
+        Ok(auth_types::ErrorResponse {
+            status_code: res.status_code,
+            code: crate::consts::NO_ERROR_CODE.to_string(),
+            message: crate::consts::NO_ERROR_MESSAGE.to_string(),
+            reason: None,
+        })
+    }
+}
diff --git a/crates/pm_auth/src/types/api/auth_service.rs b/crates/pm_auth/src/types/api/auth_service.rs
new file mode 100644
index 00000000000..35d44970d51
--- /dev/null
+++ b/crates/pm_auth/src/types/api/auth_service.rs
@@ -0,0 +1,40 @@
+use crate::types::{
+    BankAccountCredentialsRequest, BankAccountCredentialsResponse, ExchangeTokenRequest,
+    ExchangeTokenResponse, LinkTokenRequest, LinkTokenResponse,
+};
+
+pub trait AuthService:
+    super::ConnectorCommon
+    + AuthServiceLinkToken
+    + AuthServiceExchangeToken
+    + AuthServiceBankAccountCredentials
+{
+}
+
+#[derive(Debug, Clone)]
+pub struct LinkToken;
+
+pub trait AuthServiceLinkToken:
+    super::ConnectorIntegration<LinkToken, LinkTokenRequest, LinkTokenResponse>
+{
+}
+
+#[derive(Debug, Clone)]
+pub struct ExchangeToken;
+
+pub trait AuthServiceExchangeToken:
+    super::ConnectorIntegration<ExchangeToken, ExchangeTokenRequest, ExchangeTokenResponse>
+{
+}
+
+#[derive(Debug, Clone)]
+pub struct BankAccountCredentials;
+
+pub trait AuthServiceBankAccountCredentials:
+    super::ConnectorIntegration<
+    BankAccountCredentials,
+    BankAccountCredentialsRequest,
+    BankAccountCredentialsResponse,
+>
+{
+}
diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml
index 791f617b30d..e498658e457 100644
--- a/crates/router/Cargo.toml
+++ b/crates/router/Cargo.toml
@@ -111,6 +111,7 @@ currency_conversion = { version = "0.1.0", path = "../currency_conversion" }
 data_models = { version = "0.1.0", path = "../data_models", default-features = false }
 diesel_models = { version = "0.1.0", path = "../diesel_models", features = ["kv_store"] }
 euclid = { version = "0.1.0", path = "../euclid", features = ["valued_jit"] }
+pm_auth = { version = "0.1.0", path = "../pm_auth", package = "pm_auth" }
 external_services = { version = "0.1.0", path = "../external_services" }
 kgraph_utils = { version = "0.1.0", path = "../kgraph_utils" }
 masking = { version = "0.1.0", path = "../masking" }
diff --git a/crates/router/src/configs/settings.rs b/crates/router/src/configs/settings.rs
index 6cbffc186d2..0d098b9acab 100644
--- a/crates/router/src/configs/settings.rs
+++ b/crates/router/src/configs/settings.rs
@@ -100,6 +100,7 @@ pub struct Settings {
     pub required_fields: RequiredFields,
     pub delayed_session_response: DelayedSessionConfig,
     pub webhook_source_verification_call: WebhookSourceVerificationCall,
+    pub payment_method_auth: PaymentMethodAuth,
     pub connector_request_reference_id_config: ConnectorRequestReferenceIdConfig,
     #[cfg(feature = "payouts")]
     pub payouts: Payouts,
@@ -154,6 +155,12 @@ pub struct ForexApi {
     pub redis_lock_timeout: u64,
 }
 
+#[derive(Debug, Deserialize, Clone, Default)]
+pub struct PaymentMethodAuth {
+    pub redis_expiry: i64,
+    pub pm_auth_key: String,
+}
+
 #[derive(Debug, Deserialize, Clone, Default)]
 pub struct DefaultExchangeRates {
     pub base_currency: String,
diff --git a/crates/router/src/core.rs b/crates/router/src/core.rs
index be83de84916..0bd197ee22e 100644
--- a/crates/router/src/core.rs
+++ b/crates/router/src/core.rs
@@ -24,6 +24,7 @@ pub mod payment_methods;
 pub mod payments;
 #[cfg(feature = "payouts")]
 pub mod payouts;
+pub mod pm_auth;
 pub mod refunds;
 pub mod routing;
 pub mod surcharge_decision_config;
diff --git a/crates/router/src/core/admin.rs b/crates/router/src/core/admin.rs
index 5ab543d382f..113bc7d677d 100644
--- a/crates/router/src/core/admin.rs
+++ b/crates/router/src/core/admin.rs
@@ -10,9 +10,10 @@ use common_utils::{
     ext_traits::{AsyncExt, ConfigExt, Encode, ValueExt},
     pii,
 };
-use error_stack::{report, FutureExt, ResultExt};
+use error_stack::{report, FutureExt, IntoReport, ResultExt};
 use futures::future::try_join_all;
 use masking::{PeekInterface, Secret};
+use pm_auth::connector::plaid::transformers::PlaidAuthType;
 use uuid::Uuid;
 
 use crate::{
@@ -762,7 +763,7 @@ pub async fn create_payment_connector(
     )
     .await?;
 
-    let routable_connector =
+    let mut routable_connector =
         api_enums::RoutableConnectors::from_str(&req.connector_name.to_string()).ok();
 
     let business_profile = state
@@ -773,6 +774,30 @@ pub async fn create_payment_connector(
             id: profile_id.to_owned(),
         })?;
 
+    let pm_auth_connector =
+        api_enums::convert_pm_auth_connector(req.connector_name.to_string().as_str());
+
+    let is_unroutable_connector = if pm_auth_connector.is_some() {
+        if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth {
+            return Err(errors::ApiErrorResponse::InvalidRequestData {
+                message: "Invalid connector type given".to_string(),
+            })
+            .into_report();
+        }
+        true
+    } else {
+        let routable_connector_option = req
+            .connector_name
+            .to_string()
+            .parse()
+            .into_report()
+            .change_context(errors::ApiErrorResponse::InvalidRequestData {
+                message: "Invalid connector name given".to_string(),
+            })?;
+        routable_connector = Some(routable_connector_option);
+        false
+    };
+
     // If connector label is not passed in the request, generate one
     let connector_label = req
         .connector_label
@@ -877,6 +902,20 @@ pub async fn create_payment_connector(
         api_enums::ConnectorStatus::Active,
     )?;
 
+    if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth {
+        if let Some(val) = req.pm_auth_config.clone() {
+            validate_pm_auth(
+                val,
+                &*state.clone().store,
+                merchant_id.clone().as_str(),
+                &key_store,
+                merchant_account,
+                &Some(profile_id.clone()),
+            )
+            .await?;
+        }
+    }
+
     let merchant_connector_account = domain::MerchantConnectorAccount {
         merchant_id: merchant_id.to_string(),
         connector_type: req.connector_type,
@@ -948,7 +987,7 @@ pub async fn create_payment_connector(
             #[cfg(feature = "connector_choice_mca_id")]
             merchant_connector_id: Some(mca.merchant_connector_id.clone()),
             #[cfg(not(feature = "connector_choice_mca_id"))]
-            sub_label: req.business_sub_label,
+            sub_label: req.business_sub_label.clone(),
         };
 
         if !default_routing_config.contains(&choice) {
@@ -956,7 +995,7 @@ pub async fn create_payment_connector(
             routing_helpers::update_merchant_default_config(
                 &*state.store,
                 merchant_id,
-                default_routing_config,
+                default_routing_config.clone(),
             )
             .await?;
         }
@@ -965,7 +1004,7 @@ pub async fn create_payment_connector(
             routing_helpers::update_merchant_default_config(
                 &*state.store,
                 &profile_id.clone(),
-                default_routing_config_for_profile,
+                default_routing_config_for_profile.clone(),
             )
             .await?;
         }
@@ -980,10 +1019,92 @@ pub async fn create_payment_connector(
         ],
     );
 
+    if !is_unroutable_connector {
+        if let Some(routable_connector_val) = routable_connector {
+            let choice = routing_types::RoutableConnectorChoice {
+                #[cfg(feature = "backwards_compatibility")]
+                choice_kind: routing_types::RoutableChoiceKind::FullStruct,
+                connector: routable_connector_val,
+                #[cfg(feature = "connector_choice_mca_id")]
+                merchant_connector_id: Some(mca.merchant_connector_id.clone()),
+                #[cfg(not(feature = "connector_choice_mca_id"))]
+                sub_label: req.business_sub_label.clone(),
+            };
+
+            if !default_routing_config.contains(&choice) {
+                default_routing_config.push(choice.clone());
+                routing_helpers::update_merchant_default_config(
+                    &*state.clone().store,
+                    merchant_id,
+                    default_routing_config,
+                )
+                .await?;
+            }
+
+            if !default_routing_config_for_profile.contains(&choice) {
+                default_routing_config_for_profile.push(choice);
+                routing_helpers::update_merchant_default_config(
+                    &*state.store,
+                    &profile_id,
+                    default_routing_config_for_profile,
+                )
+                .await?;
+            }
+        }
+    };
+
     let mca_response = mca.try_into()?;
     Ok(service_api::ApplicationResponse::Json(mca_response))
 }
 
+async fn validate_pm_auth(
+    val: serde_json::Value,
+    db: &dyn StorageInterface,
+    merchant_id: &str,
+    key_store: &domain::MerchantKeyStore,
+    merchant_account: domain::MerchantAccount,
+    profile_id: &Option<String>,
+) -> RouterResponse<()> {
+    let config = serde_json::from_value::<api_models::pm_auth::PaymentMethodAuthConfig>(val)
+        .into_report()
+        .change_context(errors::ApiErrorResponse::InvalidRequestData {
+            message: "invalid data received for payment method auth config".to_string(),
+        })
+        .attach_printable("Failed to deserialize Payment Method Auth config")?;
+
+    let all_mcas = db
+        .find_merchant_connector_account_by_merchant_id_and_disabled_list(
+            merchant_id,
+            true,
+            key_store,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: merchant_account.merchant_id.clone(),
+        })?;
+
+    for conn_choice in config.enabled_payment_methods {
+        let pm_auth_mca = all_mcas
+            .clone()
+            .into_iter()
+            .find(|mca| mca.merchant_connector_id == conn_choice.mca_id)
+            .ok_or(errors::ApiErrorResponse::GenericNotFoundError {
+                message: "payment method auth connector account not found".to_string(),
+            })
+            .into_report()?;
+
+        if &pm_auth_mca.profile_id != profile_id {
+            return Err(errors::ApiErrorResponse::GenericNotFoundError {
+                message: "payment method auth profile_id differs from connector profile_id"
+                    .to_string(),
+            })
+            .into_report();
+        }
+    }
+
+    Ok(services::ApplicationResponse::StatusOk)
+}
+
 pub async fn retrieve_payment_connector(
     state: AppState,
     merchant_id: String,
@@ -1066,7 +1187,7 @@ pub async fn update_payment_connector(
         .await
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
 
-    let _merchant_account = db
+    let merchant_account = db
         .find_merchant_account_by_merchant_id(merchant_id, &key_store)
         .await
         .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
@@ -1106,6 +1227,20 @@ pub async fn update_payment_connector(
     let (connector_status, disabled) =
         validate_status_and_disabled(req.status, req.disabled, auth, mca.status)?;
 
+    if req.connector_type != api_enums::ConnectorType::PaymentMethodAuth {
+        if let Some(val) = req.pm_auth_config.clone() {
+            validate_pm_auth(
+                val,
+                db,
+                merchant_id,
+                &key_store,
+                merchant_account,
+                &mca.profile_id,
+            )
+            .await?;
+        }
+    }
+
     let payment_connector = storage::MerchantConnectorAccountUpdate::Update {
         merchant_id: None,
         connector_type: Some(req.connector_type),
@@ -1720,8 +1855,10 @@ pub(crate) fn validate_auth_and_metadata_type(
             signifyd::transformers::SignifydAuthType::try_from(val)?;
             Ok(())
         }
-        api_enums::Connector::Plaid => Err(report!(errors::ConnectorError::InvalidConnectorName)
-            .attach_printable(format!("invalid connector name: {connector_name}"))),
+        api_enums::Connector::Plaid => {
+            PlaidAuthType::foreign_try_from(val)?;
+            Ok(())
+        }
     }
 }
 
diff --git a/crates/router/src/core/payment_methods.rs b/crates/router/src/core/payment_methods.rs
index a2dbfb1480c..14a39f1d955 100644
--- a/crates/router/src/core/payment_methods.rs
+++ b/crates/router/src/core/payment_methods.rs
@@ -11,12 +11,12 @@ pub use api_models::{
 pub use common_utils::request::RequestBody;
 use data_models::payments::{payment_attempt::PaymentAttempt, PaymentIntent};
 use diesel_models::enums;
-use error_stack::IntoReport;
 
 use crate::{
     core::{
-        errors::{self, RouterResult},
+        errors::RouterResult,
         payments::helpers,
+        pm_auth::{self as core_pm_auth},
     },
     routes::AppState,
     types::{
@@ -172,11 +172,14 @@ impl PaymentMethodRetrieve for Oss {
                 .map(|card| Some((card, enums::PaymentMethod::Card)))
             }
 
-            storage::PaymentTokenData::AuthBankDebit(_) => {
-                Err(errors::ApiErrorResponse::NotImplemented {
-                    message: errors::NotImplementedMessage::Default,
-                })
-                .into_report()
+            storage::PaymentTokenData::AuthBankDebit(auth_token) => {
+                core_pm_auth::retrieve_payment_method_from_auth_service(
+                    state,
+                    merchant_key_store,
+                    auth_token,
+                    payment_intent,
+                )
+                .await
             }
         }
     }
diff --git a/crates/router/src/core/payment_methods/cards.rs b/crates/router/src/core/payment_methods/cards.rs
index bbcfe45a1d0..84aef952a53 100644
--- a/crates/router/src/core/payment_methods/cards.rs
+++ b/crates/router/src/core/payment_methods/cards.rs
@@ -13,6 +13,7 @@ use api_models::{
         ResponsePaymentMethodsEnabled,
     },
     payments::BankCodeResponse,
+    pm_auth::PaymentMethodAuthConfig,
     surcharge_decision_configs as api_surcharge_decision_configs,
 };
 use common_utils::{
@@ -29,6 +30,8 @@ use super::surcharge_decision_configs::{
     perform_surcharge_decision_management_for_payment_method_list,
     perform_surcharge_decision_management_for_saved_cards,
 };
+#[cfg(not(feature = "connector_choice_mca_id"))]
+use crate::core::utils::get_connector_label;
 use crate::{
     configs::settings,
     core::{
@@ -1081,9 +1084,9 @@ pub async fn list_payment_methods(
     logger::debug!(mca_before_filtering=?filtered_mcas);
 
     let mut response: Vec<ResponsePaymentMethodIntermediate> = vec![];
-    for mca in filtered_mcas {
-        let payment_methods = match mca.payment_methods_enabled {
-            Some(pm) => pm,
+    for mca in &filtered_mcas {
+        let payment_methods = match &mca.payment_methods_enabled {
+            Some(pm) => pm.clone(),
             None => continue,
         };
 
@@ -1094,13 +1097,15 @@ pub async fn list_payment_methods(
             payment_intent.as_ref(),
             payment_attempt.as_ref(),
             billing_address.as_ref(),
-            mca.connector_name,
+            mca.connector_name.clone(),
             pm_config_mapping,
             &state.conf.mandates.supported_payment_methods,
         )
         .await?;
     }
 
+    let mut pmt_to_auth_connector = HashMap::new();
+
     if let Some((payment_attempt, payment_intent)) =
         payment_attempt.as_ref().zip(payment_intent.as_ref())
     {
@@ -1204,6 +1209,84 @@ pub async fn list_payment_methods(
             pre_routing_results.insert(pm_type, routable_choice);
         }
 
+        let redis_conn = db
+            .get_redis_conn()
+            .map_err(|redis_error| logger::error!(?redis_error))
+            .ok();
+
+        let mut val = Vec::new();
+
+        for (payment_method_type, routable_connector_choice) in &pre_routing_results {
+            #[cfg(not(feature = "connector_choice_mca_id"))]
+            let connector_label = get_connector_label(
+                payment_intent.business_country,
+                payment_intent.business_label.as_ref(),
+                #[cfg(not(feature = "connector_choice_mca_id"))]
+                routable_connector_choice.sub_label.as_ref(),
+                #[cfg(feature = "connector_choice_mca_id")]
+                None,
+                routable_connector_choice.connector.to_string().as_str(),
+            );
+            #[cfg(not(feature = "connector_choice_mca_id"))]
+            let matched_mca = filtered_mcas
+                .iter()
+                .find(|m| connector_label == m.connector_label);
+
+            #[cfg(feature = "connector_choice_mca_id")]
+            let matched_mca = filtered_mcas.iter().find(|m| {
+                routable_connector_choice.merchant_connector_id.as_ref()
+                    == Some(&m.merchant_connector_id)
+            });
+
+            if let Some(m) = matched_mca {
+                let pm_auth_config = m
+                    .pm_auth_config
+                    .as_ref()
+                    .map(|config| {
+                        serde_json::from_value::<PaymentMethodAuthConfig>(config.clone())
+                            .into_report()
+                            .change_context(errors::StorageError::DeserializationFailed)
+                            .attach_printable("Failed to deserialize Payment Method Auth config")
+                    })
+                    .transpose()
+                    .unwrap_or_else(|err| {
+                        logger::error!(error=?err);
+                        None
+                    });
+
+                let matched_config = match pm_auth_config {
+                    Some(config) => {
+                        let internal_config = config
+                            .enabled_payment_methods
+                            .iter()
+                            .find(|config| config.payment_method_type == *payment_method_type)
+                            .cloned();
+
+                        internal_config
+                    }
+                    None => None,
+                };
+
+                if let Some(config) = matched_config {
+                    pmt_to_auth_connector
+                        .insert(*payment_method_type, config.connector_name.clone());
+                    val.push(config);
+                }
+            }
+        }
+
+        let pm_auth_key = format!("pm_auth_{}", payment_intent.payment_id);
+        let redis_expiry = state.conf.payment_method_auth.redis_expiry;
+
+        if let Some(rc) = redis_conn {
+            rc.serialize_and_set_key_with_expiry(pm_auth_key.as_str(), val, redis_expiry)
+                .await
+                .attach_printable("Failed to store pm auth data in redis")
+                .unwrap_or_else(|err| {
+                    logger::error!(error=?err);
+                })
+        };
+
         routing_info.pre_routing_results = Some(pre_routing_results);
 
         let encoded = utils::Encode::<storage::PaymentRoutingInfo>::encode_to_value(&routing_info)
@@ -1461,7 +1544,9 @@ pub async fn list_payment_methods(
                     .and_then(|inner_hm| inner_hm.get(payment_method_types_hm.0))
                     .cloned(),
                 surcharge_details: None,
-                pm_auth_connector: None,
+                pm_auth_connector: pmt_to_auth_connector
+                    .get(payment_method_types_hm.0)
+                    .cloned(),
             })
         }
 
@@ -1496,7 +1581,9 @@ pub async fn list_payment_methods(
                     .and_then(|inner_hm| inner_hm.get(payment_method_types_hm.0))
                     .cloned(),
                 surcharge_details: None,
-                pm_auth_connector: None,
+                pm_auth_connector: pmt_to_auth_connector
+                    .get(payment_method_types_hm.0)
+                    .cloned(),
             })
         }
 
@@ -1526,7 +1613,7 @@ pub async fn list_payment_methods(
                     .and_then(|inner_hm| inner_hm.get(key.0))
                     .cloned(),
                 surcharge_details: None,
-                pm_auth_connector: None,
+                pm_auth_connector: pmt_to_auth_connector.get(&payment_method_type).cloned(),
             }
         })
     }
@@ -1559,7 +1646,7 @@ pub async fn list_payment_methods(
                     .and_then(|inner_hm| inner_hm.get(key.0))
                     .cloned(),
                 surcharge_details: None,
-                pm_auth_connector: None,
+                pm_auth_connector: pmt_to_auth_connector.get(&payment_method_type).cloned(),
             }
         })
     }
@@ -1592,7 +1679,7 @@ pub async fn list_payment_methods(
                     .and_then(|inner_hm| inner_hm.get(key.0))
                     .cloned(),
                 surcharge_details: None,
-                pm_auth_connector: None,
+                pm_auth_connector: pmt_to_auth_connector.get(&payment_method_type).cloned(),
             }
         })
     }
diff --git a/crates/router/src/core/pm_auth.rs b/crates/router/src/core/pm_auth.rs
new file mode 100644
index 00000000000..821f049d8cf
--- /dev/null
+++ b/crates/router/src/core/pm_auth.rs
@@ -0,0 +1,729 @@
+use std::{collections::HashMap, str::FromStr};
+
+use api_models::{
+    enums,
+    payment_methods::{self, BankAccountAccessCreds},
+    payments::{AddressDetails, BankDebitBilling, BankDebitData, PaymentMethodData},
+};
+use hex;
+pub mod helpers;
+pub mod transformers;
+
+use common_utils::{
+    consts,
+    crypto::{HmacSha256, SignMessage},
+    ext_traits::AsyncExt,
+    generate_id,
+};
+use data_models::payments::PaymentIntent;
+use error_stack::{IntoReport, ResultExt};
+#[cfg(feature = "kms")]
+pub use external_services::kms;
+use helpers::PaymentAuthConnectorDataExt;
+use masking::{ExposeInterface, PeekInterface};
+use pm_auth::{
+    connector::plaid::transformers::PlaidAuthType,
+    types::{
+        self as pm_auth_types,
+        api::{
+            auth_service::{BankAccountCredentials, ExchangeToken, LinkToken},
+            BoxedConnectorIntegration, PaymentAuthConnectorData,
+        },
+    },
+};
+
+use crate::{
+    core::{
+        errors::{self, ApiErrorResponse, RouterResponse, RouterResult, StorageErrorExt},
+        payment_methods::cards,
+        payments::helpers as oss_helpers,
+        pm_auth::helpers::{self as pm_auth_helpers},
+    },
+    db::StorageInterface,
+    logger,
+    routes::AppState,
+    services::{
+        pm_auth::{self as pm_auth_services},
+        ApplicationResponse,
+    },
+    types::{
+        self,
+        domain::{self, types::decrypt},
+        storage,
+        transformers::ForeignTryFrom,
+    },
+};
+
+pub async fn create_link_token(
+    state: AppState,
+    merchant_account: domain::MerchantAccount,
+    key_store: domain::MerchantKeyStore,
+    payload: api_models::pm_auth::LinkTokenCreateRequest,
+) -> RouterResponse<api_models::pm_auth::LinkTokenCreateResponse> {
+    let db = &*state.store;
+
+    let redis_conn = db
+        .get_redis_conn()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get redis connection")?;
+
+    let pm_auth_key = format!("pm_auth_{}", payload.payment_id);
+
+    let pm_auth_configs = redis_conn
+        .get_and_deserialize_key::<Vec<api_models::pm_auth::PaymentMethodAuthConnectorChoice>>(
+            pm_auth_key.as_str(),
+            "Vec<PaymentMethodAuthConnectorChoice>",
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get payment method auth choices from redis")?;
+
+    let selected_config = pm_auth_configs
+        .into_iter()
+        .find(|config| {
+            config.payment_method == payload.payment_method
+                && config.payment_method_type == payload.payment_method_type
+        })
+        .ok_or(ApiErrorResponse::GenericNotFoundError {
+            message: "payment method auth connector name not found".to_string(),
+        })
+        .into_report()?;
+
+    let connector_name = selected_config.connector_name.as_str();
+
+    let connector = PaymentAuthConnectorData::get_connector_by_name(connector_name)?;
+    let connector_integration: BoxedConnectorIntegration<
+        '_,
+        LinkToken,
+        pm_auth_types::LinkTokenRequest,
+        pm_auth_types::LinkTokenResponse,
+    > = connector.connector.get_connector_integration();
+
+    let payment_intent = oss_helpers::verify_payment_intent_time_and_client_secret(
+        &*state.store,
+        &merchant_account,
+        payload.client_secret,
+    )
+    .await?;
+
+    let billing_country = payment_intent
+        .as_ref()
+        .async_map(|pi| async {
+            oss_helpers::get_address_by_id(
+                &*state.store,
+                pi.billing_address_id.clone(),
+                &key_store,
+                pi.payment_id.clone(),
+                merchant_account.merchant_id.clone(),
+                merchant_account.storage_scheme,
+            )
+            .await
+        })
+        .await
+        .transpose()?
+        .flatten()
+        .and_then(|address| address.country)
+        .map(|country| country.to_string());
+
+    let merchant_connector_account = state
+        .store
+        .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+            merchant_account.merchant_id.as_str(),
+            &selected_config.mca_id,
+            &key_store,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: merchant_account.merchant_id.clone(),
+        })?;
+
+    let auth_type = helpers::get_connector_auth_type(merchant_connector_account)?;
+
+    let router_data = pm_auth_types::LinkTokenRouterData {
+        flow: std::marker::PhantomData,
+        merchant_id: Some(merchant_account.merchant_id),
+        connector: Some(connector_name.to_string()),
+        request: pm_auth_types::LinkTokenRequest {
+            client_name: "HyperSwitch".to_string(),
+            country_codes: Some(vec![billing_country.ok_or(
+                errors::ApiErrorResponse::MissingRequiredField {
+                    field_name: "billing_country",
+                },
+            )?]),
+            language: payload.language,
+            user_info: payment_intent.and_then(|pi| pi.customer_id),
+        },
+        response: Ok(pm_auth_types::LinkTokenResponse {
+            link_token: "".to_string(),
+        }),
+        connector_http_status_code: None,
+        connector_auth_type: auth_type,
+    };
+
+    let connector_resp = pm_auth_services::execute_connector_processing_step(
+        state.as_ref(),
+        connector_integration,
+        &router_data,
+        &connector.connector_name,
+    )
+    .await
+    .change_context(ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed while calling link token creation connector api")?;
+
+    let link_token_resp =
+        connector_resp
+            .response
+            .map_err(|err| ApiErrorResponse::ExternalConnectorError {
+                code: err.code,
+                message: err.message,
+                connector: connector.connector_name.to_string(),
+                status_code: err.status_code,
+                reason: err.reason,
+            })?;
+
+    let response = api_models::pm_auth::LinkTokenCreateResponse {
+        link_token: link_token_resp.link_token,
+        connector: connector.connector_name.to_string(),
+    };
+
+    Ok(ApplicationResponse::Json(response))
+}
+
+impl ForeignTryFrom<&types::ConnectorAuthType> for PlaidAuthType {
+    type Error = errors::ConnectorError;
+
+    fn foreign_try_from(auth_type: &types::ConnectorAuthType) -> Result<Self, Self::Error> {
+        match auth_type {
+            types::ConnectorAuthType::BodyKey { api_key, key1 } => {
+                Ok::<Self, errors::ConnectorError>(Self {
+                    client_id: api_key.to_owned(),
+                    secret: key1.to_owned(),
+                })
+            }
+            _ => Err(errors::ConnectorError::FailedToObtainAuthType),
+        }
+    }
+}
+
+pub async fn exchange_token_core(
+    state: AppState,
+    merchant_account: domain::MerchantAccount,
+    key_store: domain::MerchantKeyStore,
+    payload: api_models::pm_auth::ExchangeTokenCreateRequest,
+) -> RouterResponse<()> {
+    let db = &*state.store;
+
+    let config = get_selected_config_from_redis(db, &payload).await?;
+
+    let connector_name = config.connector_name.as_str();
+
+    let connector =
+        pm_auth_types::api::PaymentAuthConnectorData::get_connector_by_name(connector_name)?;
+
+    let merchant_connector_account = state
+        .store
+        .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+            merchant_account.merchant_id.as_str(),
+            &config.mca_id,
+            &key_store,
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: merchant_account.merchant_id.clone(),
+        })?;
+
+    let auth_type = helpers::get_connector_auth_type(merchant_connector_account.clone())?;
+
+    let access_token = get_access_token_from_exchange_api(
+        &connector,
+        connector_name,
+        &payload,
+        &auth_type,
+        &state,
+    )
+    .await?;
+
+    let bank_account_details_resp = get_bank_account_creds(
+        connector,
+        &merchant_account,
+        connector_name,
+        &access_token,
+        auth_type,
+        &state,
+        None,
+    )
+    .await?;
+
+    Box::pin(store_bank_details_in_payment_methods(
+        key_store,
+        payload,
+        merchant_account,
+        state,
+        bank_account_details_resp,
+        (connector_name, access_token),
+        merchant_connector_account.merchant_connector_id,
+    ))
+    .await?;
+
+    Ok(ApplicationResponse::StatusOk)
+}
+
+async fn store_bank_details_in_payment_methods(
+    key_store: domain::MerchantKeyStore,
+    payload: api_models::pm_auth::ExchangeTokenCreateRequest,
+    merchant_account: domain::MerchantAccount,
+    state: AppState,
+    bank_account_details_resp: pm_auth_types::BankAccountCredentialsResponse,
+    connector_details: (&str, String),
+    mca_id: String,
+) -> RouterResult<()> {
+    let key = key_store.key.get_inner().peek();
+    let db = &*state.clone().store;
+    let (connector_name, access_token) = connector_details;
+
+    let payment_intent = db
+        .find_payment_intent_by_payment_id_merchant_id(
+            &payload.payment_id,
+            &merchant_account.merchant_id,
+            merchant_account.storage_scheme,
+        )
+        .await
+        .to_not_found_response(ApiErrorResponse::PaymentNotFound)?;
+
+    let customer_id = payment_intent
+        .customer_id
+        .ok_or(ApiErrorResponse::CustomerNotFound)?;
+
+    let payment_methods = db
+        .find_payment_method_by_customer_id_merchant_id_list(
+            &customer_id,
+            &merchant_account.merchant_id,
+        )
+        .await
+        .change_context(ApiErrorResponse::InternalServerError)?;
+
+    let mut hash_to_payment_method: HashMap<
+        String,
+        (
+            storage::PaymentMethod,
+            payment_methods::PaymentMethodDataBankCreds,
+        ),
+    > = HashMap::new();
+
+    for pm in payment_methods {
+        if pm.payment_method == enums::PaymentMethod::BankDebit {
+            let bank_details_pm_data = decrypt::<serde_json::Value, masking::WithType>(
+                pm.payment_method_data.clone(),
+                key,
+            )
+            .await
+            .change_context(ApiErrorResponse::InternalServerError)
+            .attach_printable("unable to decrypt bank account details")?
+            .map(|x| x.into_inner().expose())
+            .map(|v| {
+                serde_json::from_value::<payment_methods::PaymentMethodsData>(v)
+                    .into_report()
+                    .change_context(errors::StorageError::DeserializationFailed)
+                    .attach_printable("Failed to deserialize Payment Method Auth config")
+            })
+            .transpose()
+            .unwrap_or_else(|err| {
+                logger::error!(error=?err);
+                None
+            })
+            .and_then(|pmd| match pmd {
+                payment_methods::PaymentMethodsData::BankDetails(bank_creds) => Some(bank_creds),
+                _ => None,
+            })
+            .ok_or(ApiErrorResponse::InternalServerError)?;
+
+            hash_to_payment_method.insert(
+                bank_details_pm_data.hash.clone(),
+                (pm, bank_details_pm_data),
+            );
+        }
+    }
+
+    #[cfg(feature = "kms")]
+    let pm_auth_key = kms::get_kms_client(&state.conf.kms)
+        .await
+        .decrypt(state.conf.payment_method_auth.pm_auth_key.clone())
+        .await
+        .change_context(ApiErrorResponse::InternalServerError)?;
+
+    #[cfg(not(feature = "kms"))]
+    let pm_auth_key = state.conf.payment_method_auth.pm_auth_key.clone();
+
+    let mut update_entries: Vec<(storage::PaymentMethod, storage::PaymentMethodUpdate)> =
+        Vec::new();
+    let mut new_entries: Vec<storage::PaymentMethodNew> = Vec::new();
+
+    for creds in bank_account_details_resp.credentials {
+        let hash_string = format!("{}-{}", creds.account_number, creds.routing_number);
+        let generated_hash = hex::encode(
+            HmacSha256::sign_message(&HmacSha256, pm_auth_key.as_bytes(), hash_string.as_bytes())
+                .change_context(ApiErrorResponse::InternalServerError)
+                .attach_printable("Failed to sign the message")?,
+        );
+
+        let contains_account = hash_to_payment_method.get(&generated_hash);
+        let mut pmd = payment_methods::PaymentMethodDataBankCreds {
+            mask: creds
+                .account_number
+                .chars()
+                .rev()
+                .take(4)
+                .collect::<String>()
+                .chars()
+                .rev()
+                .collect::<String>(),
+            hash: generated_hash,
+            account_type: creds.account_type,
+            account_name: creds.account_name,
+            payment_method_type: creds.payment_method_type,
+            connector_details: vec![payment_methods::BankAccountConnectorDetails {
+                connector: connector_name.to_string(),
+                mca_id: mca_id.clone(),
+                access_token: payment_methods::BankAccountAccessCreds::AccessToken(
+                    access_token.clone(),
+                ),
+                account_id: creds.account_id,
+            }],
+        };
+
+        if let Some((pm, details)) = contains_account {
+            pmd.connector_details.extend(
+                details
+                    .connector_details
+                    .clone()
+                    .into_iter()
+                    .filter(|conn| conn.mca_id != mca_id),
+            );
+
+            let payment_method_data = payment_methods::PaymentMethodsData::BankDetails(pmd);
+            let encrypted_data =
+                cards::create_encrypted_payment_method_data(&key_store, Some(payment_method_data))
+                    .await
+                    .ok_or(ApiErrorResponse::InternalServerError)?;
+            let pm_update = storage::PaymentMethodUpdate::PaymentMethodDataUpdate {
+                payment_method_data: Some(encrypted_data),
+            };
+
+            update_entries.push((pm.clone(), pm_update));
+        } else {
+            let payment_method_data = payment_methods::PaymentMethodsData::BankDetails(pmd);
+            let encrypted_data =
+                cards::create_encrypted_payment_method_data(&key_store, Some(payment_method_data))
+                    .await
+                    .ok_or(ApiErrorResponse::InternalServerError)?;
+            let pm_id = generate_id(consts::ID_LENGTH, "pm");
+            let pm_new = storage::PaymentMethodNew {
+                customer_id: customer_id.clone(),
+                merchant_id: merchant_account.merchant_id.clone(),
+                payment_method_id: pm_id,
+                payment_method: enums::PaymentMethod::BankDebit,
+                payment_method_type: Some(creds.payment_method_type),
+                payment_method_issuer: None,
+                scheme: None,
+                metadata: None,
+                payment_method_data: Some(encrypted_data),
+                ..storage::PaymentMethodNew::default()
+            };
+
+            new_entries.push(pm_new);
+        };
+    }
+
+    store_in_db(update_entries, new_entries, db).await?;
+
+    Ok(())
+}
+
+async fn store_in_db(
+    update_entries: Vec<(storage::PaymentMethod, storage::PaymentMethodUpdate)>,
+    new_entries: Vec<storage::PaymentMethodNew>,
+    db: &dyn StorageInterface,
+) -> RouterResult<()> {
+    let update_entries_futures = update_entries
+        .into_iter()
+        .map(|(pm, pm_update)| db.update_payment_method(pm, pm_update))
+        .collect::<Vec<_>>();
+
+    let new_entries_futures = new_entries
+        .into_iter()
+        .map(|pm_new| db.insert_payment_method(pm_new))
+        .collect::<Vec<_>>();
+
+    let update_futures = futures::future::join_all(update_entries_futures);
+    let new_futures = futures::future::join_all(new_entries_futures);
+
+    let (update, new) = tokio::join!(update_futures, new_futures);
+
+    let _ = update
+        .into_iter()
+        .map(|res| res.map_err(|err| logger::error!("Payment method storage failed {err:?}")));
+
+    let _ = new
+        .into_iter()
+        .map(|res| res.map_err(|err| logger::error!("Payment method storage failed {err:?}")));
+
+    Ok(())
+}
+
+pub async fn get_bank_account_creds(
+    connector: PaymentAuthConnectorData,
+    merchant_account: &domain::MerchantAccount,
+    connector_name: &str,
+    access_token: &str,
+    auth_type: pm_auth_types::ConnectorAuthType,
+    state: &AppState,
+    bank_account_id: Option<String>,
+) -> RouterResult<pm_auth_types::BankAccountCredentialsResponse> {
+    let connector_integration_bank_details: BoxedConnectorIntegration<
+        '_,
+        BankAccountCredentials,
+        pm_auth_types::BankAccountCredentialsRequest,
+        pm_auth_types::BankAccountCredentialsResponse,
+    > = connector.connector.get_connector_integration();
+
+    let router_data_bank_details = pm_auth_types::BankDetailsRouterData {
+        flow: std::marker::PhantomData,
+        merchant_id: Some(merchant_account.merchant_id.clone()),
+        connector: Some(connector_name.to_string()),
+        request: pm_auth_types::BankAccountCredentialsRequest {
+            access_token: access_token.to_string(),
+            optional_ids: bank_account_id
+                .map(|id| pm_auth_types::BankAccountOptionalIDs { ids: vec![id] }),
+        },
+        response: Ok(pm_auth_types::BankAccountCredentialsResponse {
+            credentials: Vec::new(),
+        }),
+        connector_http_status_code: None,
+        connector_auth_type: auth_type,
+    };
+
+    let bank_details_resp = pm_auth_services::execute_connector_processing_step(
+        state,
+        connector_integration_bank_details,
+        &router_data_bank_details,
+        &connector.connector_name,
+    )
+    .await
+    .change_context(ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed while calling bank account details connector api")?;
+
+    let bank_account_details_resp =
+        bank_details_resp
+            .response
+            .map_err(|err| ApiErrorResponse::ExternalConnectorError {
+                code: err.code,
+                message: err.message,
+                connector: connector.connector_name.to_string(),
+                status_code: err.status_code,
+                reason: err.reason,
+            })?;
+
+    Ok(bank_account_details_resp)
+}
+
+async fn get_access_token_from_exchange_api(
+    connector: &PaymentAuthConnectorData,
+    connector_name: &str,
+    payload: &api_models::pm_auth::ExchangeTokenCreateRequest,
+    auth_type: &pm_auth_types::ConnectorAuthType,
+    state: &AppState,
+) -> RouterResult<String> {
+    let connector_integration: BoxedConnectorIntegration<
+        '_,
+        ExchangeToken,
+        pm_auth_types::ExchangeTokenRequest,
+        pm_auth_types::ExchangeTokenResponse,
+    > = connector.connector.get_connector_integration();
+
+    let router_data = pm_auth_types::ExchangeTokenRouterData {
+        flow: std::marker::PhantomData,
+        merchant_id: None,
+        connector: Some(connector_name.to_string()),
+        request: pm_auth_types::ExchangeTokenRequest {
+            public_token: payload.public_token.clone(),
+        },
+        response: Ok(pm_auth_types::ExchangeTokenResponse {
+            access_token: "".to_string(),
+        }),
+        connector_http_status_code: None,
+        connector_auth_type: auth_type.clone(),
+    };
+
+    let resp = pm_auth_services::execute_connector_processing_step(
+        state,
+        connector_integration,
+        &router_data,
+        &connector.connector_name,
+    )
+    .await
+    .change_context(ApiErrorResponse::InternalServerError)
+    .attach_printable("Failed while calling exchange token connector api")?;
+
+    let exchange_token_resp =
+        resp.response
+            .map_err(|err| ApiErrorResponse::ExternalConnectorError {
+                code: err.code,
+                message: err.message,
+                connector: connector.connector_name.to_string(),
+                status_code: err.status_code,
+                reason: err.reason,
+            })?;
+
+    let access_token = exchange_token_resp.access_token;
+    Ok(access_token)
+}
+
+async fn get_selected_config_from_redis(
+    db: &dyn StorageInterface,
+    payload: &api_models::pm_auth::ExchangeTokenCreateRequest,
+) -> RouterResult<api_models::pm_auth::PaymentMethodAuthConnectorChoice> {
+    let redis_conn = db
+        .get_redis_conn()
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get redis connection")?;
+
+    let pm_auth_key = format!("pm_auth_{}", payload.payment_id);
+
+    let pm_auth_configs = redis_conn
+        .get_and_deserialize_key::<Vec<api_models::pm_auth::PaymentMethodAuthConnectorChoice>>(
+            pm_auth_key.as_str(),
+            "Vec<PaymentMethodAuthConnectorChoice>",
+        )
+        .await
+        .change_context(errors::ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed to get payment method auth choices from redis")?;
+
+    let selected_config = pm_auth_configs
+        .iter()
+        .find(|conf| {
+            conf.payment_method == payload.payment_method
+                && conf.payment_method_type == payload.payment_method_type
+        })
+        .ok_or(ApiErrorResponse::GenericNotFoundError {
+            message: "connector name not found".to_string(),
+        })
+        .into_report()?
+        .clone();
+
+    Ok(selected_config)
+}
+
+pub async fn retrieve_payment_method_from_auth_service(
+    state: &AppState,
+    key_store: &domain::MerchantKeyStore,
+    auth_token: &payment_methods::BankAccountConnectorDetails,
+    payment_intent: &PaymentIntent,
+) -> RouterResult<Option<(PaymentMethodData, enums::PaymentMethod)>> {
+    let db = state.store.as_ref();
+
+    let connector = pm_auth_types::api::PaymentAuthConnectorData::get_connector_by_name(
+        auth_token.connector.as_str(),
+    )?;
+
+    let merchant_account = db
+        .find_merchant_account_by_merchant_id(&payment_intent.merchant_id, key_store)
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::MerchantAccountNotFound)?;
+
+    let mca = db
+        .find_by_merchant_connector_account_merchant_id_merchant_connector_id(
+            &payment_intent.merchant_id,
+            &auth_token.mca_id,
+            key_store,
+        )
+        .await
+        .to_not_found_response(errors::ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: auth_token.mca_id.clone(),
+        })
+        .attach_printable(
+            "error while fetching merchant_connector_account from merchant_id and connector name",
+        )?;
+
+    let auth_type = pm_auth_helpers::get_connector_auth_type(mca)?;
+
+    let BankAccountAccessCreds::AccessToken(access_token) = &auth_token.access_token;
+
+    let bank_account_creds = get_bank_account_creds(
+        connector,
+        &merchant_account,
+        &auth_token.connector,
+        access_token,
+        auth_type,
+        state,
+        Some(auth_token.account_id.clone()),
+    )
+    .await?;
+
+    logger::debug!("bank_creds: {:?}", bank_account_creds);
+
+    let bank_account = bank_account_creds
+        .credentials
+        .first()
+        .ok_or(errors::ApiErrorResponse::InternalServerError)
+        .into_report()
+        .attach_printable("Bank account details not found")?;
+
+    let mut bank_type = None;
+    if let Some(account_type) = bank_account.account_type.clone() {
+        bank_type = api_models::enums::BankType::from_str(account_type.as_str())
+            .map_err(|error| logger::error!(%error,"unable to parse account_type {account_type:?}"))
+            .ok();
+    }
+
+    let address = oss_helpers::get_address_by_id(
+        &*state.store,
+        payment_intent.billing_address_id.clone(),
+        key_store,
+        payment_intent.payment_id.clone(),
+        merchant_account.merchant_id.clone(),
+        merchant_account.storage_scheme,
+    )
+    .await?;
+
+    let name = address
+        .as_ref()
+        .and_then(|addr| addr.first_name.clone().map(|name| name.into_inner()));
+
+    let address_details = address.clone().map(|addr| {
+        let line1 = addr.line1.map(|line1| line1.into_inner());
+        let line2 = addr.line2.map(|line2| line2.into_inner());
+        let line3 = addr.line3.map(|line3| line3.into_inner());
+        let zip = addr.zip.map(|zip| zip.into_inner());
+        let state = addr.state.map(|state| state.into_inner());
+        let first_name = addr.first_name.map(|first_name| first_name.into_inner());
+        let last_name = addr.last_name.map(|last_name| last_name.into_inner());
+
+        AddressDetails {
+            city: addr.city,
+            country: addr.country,
+            line1,
+            line2,
+            line3,
+            zip,
+            state,
+            first_name,
+            last_name,
+        }
+    });
+    let payment_method_data = PaymentMethodData::BankDebit(BankDebitData::AchBankDebit {
+        billing_details: BankDebitBilling {
+            name: name.unwrap_or_default(),
+            email: common_utils::pii::Email::from(masking::Secret::new("".to_string())),
+            address: address_details,
+        },
+        account_number: masking::Secret::new(bank_account.account_number.clone()),
+        routing_number: masking::Secret::new(bank_account.routing_number.clone()),
+        card_holder_name: None,
+        bank_account_holder_name: None,
+        bank_name: None,
+        bank_type,
+        bank_holder_type: None,
+    });
+
+    Ok(Some((payment_method_data, enums::PaymentMethod::BankDebit)))
+}
diff --git a/crates/router/src/core/pm_auth/helpers.rs b/crates/router/src/core/pm_auth/helpers.rs
new file mode 100644
index 00000000000..43d30705a80
--- /dev/null
+++ b/crates/router/src/core/pm_auth/helpers.rs
@@ -0,0 +1,33 @@
+use common_utils::ext_traits::ValueExt;
+use error_stack::{IntoReport, ResultExt};
+use pm_auth::types::{self as pm_auth_types, api::BoxedPaymentAuthConnector};
+
+use crate::{
+    core::errors::{self, ApiErrorResponse},
+    types::{self, domain, transformers::ForeignTryFrom},
+};
+
+pub trait PaymentAuthConnectorDataExt {
+    fn get_connector_by_name(name: &str) -> errors::CustomResult<Self, ApiErrorResponse>
+    where
+        Self: Sized;
+    fn convert_connector(
+        connector_name: pm_auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<BoxedPaymentAuthConnector, ApiErrorResponse>;
+}
+
+pub fn get_connector_auth_type(
+    merchant_connector_account: domain::MerchantConnectorAccount,
+) -> errors::CustomResult<pm_auth_types::ConnectorAuthType, ApiErrorResponse> {
+    let auth_type: types::ConnectorAuthType = merchant_connector_account
+        .connector_account_details
+        .parse_value("ConnectorAuthType")
+        .change_context(ApiErrorResponse::MerchantConnectorAccountNotFound {
+            id: "ConnectorAuthType".to_string(),
+        })?;
+
+    pm_auth_types::ConnectorAuthType::foreign_try_from(auth_type)
+        .into_report()
+        .change_context(ApiErrorResponse::InternalServerError)
+        .attach_printable("Failed while converting ConnectorAuthType")
+}
diff --git a/crates/router/src/core/pm_auth/transformers.rs b/crates/router/src/core/pm_auth/transformers.rs
new file mode 100644
index 00000000000..8a1369c2e02
--- /dev/null
+++ b/crates/router/src/core/pm_auth/transformers.rs
@@ -0,0 +1,18 @@
+use pm_auth::types::{self as pm_auth_types};
+
+use crate::{core::errors, types, types::transformers::ForeignTryFrom};
+
+impl ForeignTryFrom<types::ConnectorAuthType> for pm_auth_types::ConnectorAuthType {
+    type Error = errors::ConnectorError;
+    fn foreign_try_from(auth_type: types::ConnectorAuthType) -> Result<Self, Self::Error> {
+        match auth_type {
+            types::ConnectorAuthType::BodyKey { api_key, key1 } => {
+                Ok::<Self, errors::ConnectorError>(Self::BodyKey {
+                    client_id: api_key.to_owned(),
+                    secret: key1.to_owned(),
+                })
+            }
+            _ => Err(errors::ConnectorError::FailedToObtainAuthType),
+        }
+    }
+}
diff --git a/crates/router/src/routes.rs b/crates/router/src/routes.rs
index ce1717c9e93..ec718b2dde9 100644
--- a/crates/router/src/routes.rs
+++ b/crates/router/src/routes.rs
@@ -40,6 +40,8 @@ pub mod verify_connector;
 pub mod webhooks;
 
 pub mod locker_migration;
+#[cfg(any(feature = "olap", feature = "oltp"))]
+pub mod pm_auth;
 #[cfg(feature = "dummy_connector")]
 pub use self::app::DummyConnector;
 #[cfg(any(feature = "olap", feature = "oltp"))]
diff --git a/crates/router/src/routes/app.rs b/crates/router/src/routes/app.rs
index 6b72e69b9f4..a7c394b7b6c 100644
--- a/crates/router/src/routes/app.rs
+++ b/crates/router/src/routes/app.rs
@@ -20,6 +20,8 @@ use super::currency;
 use super::dummy_connector::*;
 #[cfg(feature = "payouts")]
 use super::payouts::*;
+#[cfg(feature = "oltp")]
+use super::pm_auth;
 #[cfg(feature = "olap")]
 use super::routing as cloud_routing;
 #[cfg(all(feature = "olap", feature = "kms"))]
@@ -555,6 +557,8 @@ impl PaymentMethods {
                     .route(web::post().to(payment_method_update_api))
                     .route(web::delete().to(payment_method_delete_api)),
             )
+            .service(web::resource("/auth/link").route(web::post().to(pm_auth::link_token_create)))
+            .service(web::resource("/auth/exchange").route(web::post().to(pm_auth::exchange_token)))
     }
 }
 
diff --git a/crates/router/src/routes/lock_utils.rs b/crates/router/src/routes/lock_utils.rs
index 88c35bb0a13..533d1d3a629 100644
--- a/crates/router/src/routes/lock_utils.rs
+++ b/crates/router/src/routes/lock_utils.rs
@@ -13,6 +13,7 @@ pub enum ApiIdentifier {
     Ephemeral,
     Mandates,
     PaymentMethods,
+    PaymentMethodAuth,
     Payouts,
     Disputes,
     CardsInfo,
@@ -86,6 +87,8 @@ impl From<Flow> for ApiIdentifier {
             | Flow::PaymentMethodsDelete
             | Flow::ValidatePaymentMethod => Self::PaymentMethods,
 
+            Flow::PmAuthLinkTokenCreate | Flow::PmAuthExchangeToken => Self::PaymentMethodAuth,
+
             Flow::PaymentsCreate
             | Flow::PaymentsRetrieve
             | Flow::PaymentsUpdate
diff --git a/crates/router/src/routes/pm_auth.rs b/crates/router/src/routes/pm_auth.rs
new file mode 100644
index 00000000000..cfadd787c31
--- /dev/null
+++ b/crates/router/src/routes/pm_auth.rs
@@ -0,0 +1,73 @@
+use actix_web::{web, HttpRequest, Responder};
+use api_models as api_types;
+use router_env::{instrument, tracing, types::Flow};
+
+use crate::{core::api_locking, routes::AppState, services::api as oss_api};
+
+#[instrument(skip_all, fields(flow = ?Flow::PmAuthLinkTokenCreate))]
+pub async fn link_token_create(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    json_payload: web::Json<api_types::pm_auth::LinkTokenCreateRequest>,
+) -> impl Responder {
+    let payload = json_payload.into_inner();
+    let flow = Flow::PmAuthLinkTokenCreate;
+    let (auth, _) = match crate::services::authentication::check_client_secret_and_get_auth(
+        req.headers(),
+        &payload,
+    ) {
+        Ok((auth, _auth_flow)) => (auth, _auth_flow),
+        Err(e) => return oss_api::log_and_return_error_response(e),
+    };
+    Box::pin(oss_api::server_wrap(
+        flow,
+        state,
+        &req,
+        payload,
+        |state, auth, payload| {
+            crate::core::pm_auth::create_link_token(
+                state,
+                auth.merchant_account,
+                auth.key_store,
+                payload,
+            )
+        },
+        &*auth,
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
+
+#[instrument(skip_all, fields(flow = ?Flow::PmAuthExchangeToken))]
+pub async fn exchange_token(
+    state: web::Data<AppState>,
+    req: HttpRequest,
+    json_payload: web::Json<api_types::pm_auth::ExchangeTokenCreateRequest>,
+) -> impl Responder {
+    let payload = json_payload.into_inner();
+    let flow = Flow::PmAuthExchangeToken;
+    let (auth, _) = match crate::services::authentication::check_client_secret_and_get_auth(
+        req.headers(),
+        &payload,
+    ) {
+        Ok((auth, _auth_flow)) => (auth, _auth_flow),
+        Err(e) => return oss_api::log_and_return_error_response(e),
+    };
+    Box::pin(oss_api::server_wrap(
+        flow,
+        state,
+        &req,
+        payload,
+        |state, auth, payload| {
+            crate::core::pm_auth::exchange_token_core(
+                state,
+                auth.merchant_account,
+                auth.key_store,
+                payload,
+            )
+        },
+        &*auth,
+        api_locking::LockAction::NotApplicable,
+    ))
+    .await
+}
diff --git a/crates/router/src/services.rs b/crates/router/src/services.rs
index e46612b95df..57f3b802bd5 100644
--- a/crates/router/src/services.rs
+++ b/crates/router/src/services.rs
@@ -6,6 +6,7 @@ pub mod encryption;
 pub mod jwt;
 pub mod kafka;
 pub mod logger;
+pub mod pm_auth;
 
 #[cfg(feature = "email")]
 pub mod email;
diff --git a/crates/router/src/services/authentication.rs b/crates/router/src/services/authentication.rs
index 8a0cd7c729e..b48465ebd17 100644
--- a/crates/router/src/services/authentication.rs
+++ b/crates/router/src/services/authentication.rs
@@ -641,6 +641,18 @@ impl ClientSecretFetch for api_models::payments::RetrievePaymentLinkRequest {
     }
 }
 
+impl ClientSecretFetch for api_models::pm_auth::LinkTokenCreateRequest {
+    fn get_client_secret(&self) -> Option<&String> {
+        self.client_secret.as_ref()
+    }
+}
+
+impl ClientSecretFetch for api_models::pm_auth::ExchangeTokenCreateRequest {
+    fn get_client_secret(&self) -> Option<&String> {
+        self.client_secret.as_ref()
+    }
+}
+
 pub fn get_auth_type_and_flow<A: AppStateInfo + Sync>(
     headers: &HeaderMap,
 ) -> RouterResult<(
diff --git a/crates/router/src/services/pm_auth.rs b/crates/router/src/services/pm_auth.rs
new file mode 100644
index 00000000000..7487b12663b
--- /dev/null
+++ b/crates/router/src/services/pm_auth.rs
@@ -0,0 +1,95 @@
+use pm_auth::{
+    consts,
+    core::errors::ConnectorError,
+    types::{self as pm_auth_types, api::BoxedConnectorIntegration, PaymentAuthRouterData},
+};
+
+use crate::{
+    core::errors::{self},
+    logger,
+    routes::AppState,
+    services::{self},
+};
+
+pub async fn execute_connector_processing_step<
+    'b,
+    'a,
+    T: 'static,
+    Req: Clone + 'static,
+    Resp: Clone + 'static,
+>(
+    state: &'b AppState,
+    connector_integration: BoxedConnectorIntegration<'a, T, Req, Resp>,
+    req: &'b PaymentAuthRouterData<T, Req, Resp>,
+    connector: &pm_auth_types::PaymentMethodAuthConnectors,
+) -> errors::CustomResult<PaymentAuthRouterData<T, Req, Resp>, ConnectorError>
+where
+    T: Clone,
+    Req: Clone,
+    Resp: Clone,
+{
+    let mut router_data = req.clone();
+
+    let connector_request = connector_integration.build_request(req, connector)?;
+
+    match connector_request {
+        Some(request) => {
+            logger::debug!(connector_request=?request);
+            let response = services::api::call_connector_api(state, request).await;
+            logger::debug!(connector_response=?response);
+            match response {
+                Ok(body) => {
+                    let response = match body {
+                        Ok(body) => {
+                            let body = pm_auth_types::Response {
+                                headers: body.headers,
+                                response: body.response,
+                                status_code: body.status_code,
+                            };
+                            let connector_http_status_code = Some(body.status_code);
+                            let mut data =
+                                connector_integration.handle_response(&router_data, body)?;
+                            data.connector_http_status_code = connector_http_status_code;
+
+                            data
+                        }
+                        Err(body) => {
+                            let body = pm_auth_types::Response {
+                                headers: body.headers,
+                                response: body.response,
+                                status_code: body.status_code,
+                            };
+                            router_data.connector_http_status_code = Some(body.status_code);
+
+                            let error = match body.status_code {
+                                500..=511 => connector_integration.get_5xx_error_response(body)?,
+                                _ => connector_integration.get_error_response(body)?,
+                            };
+
+                            router_data.response = Err(error);
+
+                            router_data
+                        }
+                    };
+                    Ok(response)
+                }
+                Err(error) => {
+                    if error.current_context().is_upstream_timeout() {
+                        let error_response = pm_auth_types::ErrorResponse {
+                            code: consts::REQUEST_TIMEOUT_ERROR_CODE.to_string(),
+                            message: consts::REQUEST_TIMEOUT_ERROR_MESSAGE.to_string(),
+                            reason: Some(consts::REQUEST_TIMEOUT_ERROR_MESSAGE.to_string()),
+                            status_code: 504,
+                        };
+                        router_data.response = Err(error_response);
+                        router_data.connector_http_status_code = Some(504);
+                        Ok(router_data)
+                    } else {
+                        Err(error.change_context(ConnectorError::ProcessingStepFailed(None)))
+                    }
+                }
+            }
+        }
+        None => Ok(router_data),
+    }
+}
diff --git a/crates/router/src/types.rs b/crates/router/src/types.rs
index de28c1a3188..aa563c647ea 100644
--- a/crates/router/src/types.rs
+++ b/crates/router/src/types.rs
@@ -10,6 +10,8 @@ pub mod api;
 pub mod domain;
 #[cfg(feature = "frm")]
 pub mod fraud_check;
+pub mod pm_auth;
+
 pub mod storage;
 pub mod transformers;
 
diff --git a/crates/router/src/types/pm_auth.rs b/crates/router/src/types/pm_auth.rs
new file mode 100644
index 00000000000..e2d08c6afea
--- /dev/null
+++ b/crates/router/src/types/pm_auth.rs
@@ -0,0 +1,38 @@
+use std::str::FromStr;
+
+use error_stack::{IntoReport, ResultExt};
+use pm_auth::{
+    connector::plaid,
+    types::{
+        self as pm_auth_types,
+        api::{BoxedPaymentAuthConnector, PaymentAuthConnectorData},
+    },
+};
+
+use crate::core::{
+    errors::{self, ApiErrorResponse},
+    pm_auth::helpers::PaymentAuthConnectorDataExt,
+};
+
+impl PaymentAuthConnectorDataExt for PaymentAuthConnectorData {
+    fn get_connector_by_name(name: &str) -> errors::CustomResult<Self, ApiErrorResponse> {
+        let connector_name = pm_auth_types::PaymentMethodAuthConnectors::from_str(name)
+            .into_report()
+            .change_context(ApiErrorResponse::IncorrectConnectorNameGiven)
+            .attach_printable_lazy(|| {
+                format!("unable to parse connector: {:?}", name.to_string())
+            })?;
+        let connector = Self::convert_connector(connector_name.clone())?;
+        Ok(Self {
+            connector,
+            connector_name,
+        })
+    }
+    fn convert_connector(
+        connector_name: pm_auth_types::PaymentMethodAuthConnectors,
+    ) -> errors::CustomResult<BoxedPaymentAuthConnector, ApiErrorResponse> {
+        match connector_name {
+            pm_auth_types::PaymentMethodAuthConnectors::Plaid => Ok(Box::new(&plaid::Plaid)),
+        }
+    }
+}
diff --git a/crates/router_env/src/logger/types.rs b/crates/router_env/src/logger/types.rs
index 13ca344e9c5..b682bcb12e6 100644
--- a/crates/router_env/src/logger/types.rs
+++ b/crates/router_env/src/logger/types.rs
@@ -295,6 +295,10 @@ pub enum Flow {
     UserMerchantAccountList,
     /// Get users for merchant account
     GetUserDetails,
+    /// PaymentMethodAuth Link token create
+    PmAuthLinkTokenCreate,
+    /// PaymentMethodAuth Exchange token create
+    PmAuthExchangeToken,
     /// Get reset password link
     ForgotPassword,
     /// Reset password using link

## Type of Change
<!-- Put an `x` in the boxes that apply -->

- [ ] Bugfix
- [x] New feature
- [ ] Enhancement
- [ ] Refactoring
- [ ] Dependency updates
- [ ] Documentation
- [ ] CI/CD

## Description
<!-- Describe your changes in detail -->
PM Auth service migration

### Additional Changes

- [x] This PR modifies the API contract
- [ ] This PR modifies the database schema
- [x] This PR modifies application configuration/environment variables

<!--
Provide links to the files with corresponding changes.

Following are the paths where you can find config files:
1. `config`
2. `crates/router/src/configs`
3. `loadtest/config`
-->


## Motivation and Context
<!--
Why is this change required? What problem does it solve?
If it fixes an open issue, please link to the issue here.

If you don't have an issue, we'd recommend starting with one first so the PR
can focus on the implementation (unless it is an obvious bug or documentation fix
that will have little conversation).
-->


## How did you test it?
<!--
Did you write an integration/unit/API test to verify the code changes?
Or did you test this change manually (provide relevant screenshots)?
-->
1. Create merchant account - 
```
curl --location --request POST 'http://localhost:8080/accounts' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data-raw '{
  "merchant_id": "merchant_1701764398",
  "locker_id": "m0010",
  "merchant_name": "Sarthak",
  "merchant_details": {
    "primary_contact_person": "John Test",
    "primary_email": "JohnTest@test.com",
    "primary_phone": "sunt laborum",
    "secondary_contact_person": "John Test2",
    "secondary_email": "JohnTest2@test.com",
    "secondary_phone": "cillum do dolor id",
    "website": "www.example.com",
    "about_business": "Online Retail with a wide selection of organic products for North America",
    "address": {
      "line1": "1467",
      "line2": "Harrison Street",
      "line3": "Harrison Street",
      "city": "San Fransico",
      "state": "California",
      "zip": "94122",
      "country": "US"
    }
  },
  "return_url": "https://google.com/success",
  "webhook_details": {
     "webhook_username": "ekart_retail",
    "webhook_password": "password_ekart@123",
    "payment_created_enabled": true,
    "payment_succeeded_enabled": true,
    "payment_failed_enabled": true
  },
  "sub_merchants_enabled": false,
  "metadata": {
    "city": "NY",
    "unit": "245"
  },
  "primary_business_details": [
    {
      "country": "US",
      "business": "default"
    }
  ]
}'
```
2. Create API key
3. Create Plaid Connector account
```
curl --location --request POST 'http://localhost:8080/account/merchant_1701763754/connectors' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data-raw '{
  "connector_type": "payment_method_auth",
  "connector_name": "plaid",
  "connector_account_details": {
    "auth_type": "BodyKey",
    "api_key": "some key",
    "key1": "some key"
  },
  "test_mode": false,
  "disabled": false,
  "business_country": "US",
  "business_label":"default",
  "status": "active",
  "metadata": {
    "city": "NY",
    "unit": "245"
  }
}'
```
4. Create payment connector account - 
```
curl --location --request POST 'http://localhost:8080/account/merchant_1701763754/connectors' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: test_admin' \
--data-raw '{
    "connector_type": "fiz_operations",
    "connector_name": "stripe",
    "connector_account_details": {
        "auth_type": "HeaderKey",
        "api_key": "some key"
    },
    "test_mode": false,
    "disabled": false,
    "payment_methods_enabled": [
        {
            "payment_method": "card",
            "payment_method_types": [
                {
                    "payment_method_type": "credit",
                    "card_networks": [
                        "Visa",
                        "Mastercard"
                    ],
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "debit",
                    "card_networks": [
                        "Visa",
                        "Mastercard"
                    ],
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        },
        {
            "payment_method": "bank_debit",
            "payment_method_types": [
                {
                    "payment_method_type": "ach",
                    "recurring_enabled": true,
                    "installment_payment_enabled": true,
                    "minimum_amount": 0,
                    "maximum_amount": 10000
                }
            ]
        },
        {
            "payment_method": "pay_later",
            "payment_method_types": [
                {
                    "payment_method_type": "klarna",
                    "payment_experience": "redirect_to_url",
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "affirm",
                    "payment_experience": "redirect_to_url",
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                },
                {
                    "payment_method_type": "afterpay_clearpay",
                    "payment_experience": "redirect_to_url",
                    "minimum_amount": 1,
                    "maximum_amount": 68607706,
                    "recurring_enabled": true,
                    "installment_payment_enabled": true
                }
            ]
        }
    ],
    "metadata": {
        "city": "NY",
        "unit": "245"
    },
    "connector_webhook_details": {
        "merchant_secret": "MyWebhookSecret"
    },
    "business_country": "US",
    "business_label": "default",
    "pm_auth_config": {
        "enabled_payment_methods": [
           {
                "payment_method": "bank_debit",
                "payment_method_type": "ach",
                "connector_name": "plaid",
                "mca_id": "mca_F6MPzhuQvzhHhOsVdgxi"
            }
        ]
    }
}'
```
5. Payments create - 
```
curl --location --request POST 'http://localhost:8080/payments' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_O08Uv8O5PjAfrxsSLuduC1bn3FTWEpk0N2vvAKnvhAfc7jISAnrNgkTtcbibjO00' \
--data-raw '{
    "amount": 6540,
    "currency": "USD",
    "confirm": false,
    "capture_method": "automatic",
    "capture_on": "2022-09-10T10:11:12Z",
    "amount_to_capture": 6540,
    "customer_id": "StripeCustomer",
    "business_country": "US",
    "business_label": "default",
    "email": "guest@example.com",
    "name": "John Doe",
    "phone": "999999999",
    "phone_country_code": "+1",
    "description": "Its my first payment request",
    "authentication_type": "no_three_ds",
    "return_url": "https://google.com",
    "shipping": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "8056594430",
            "country_code": "+91"
        }
    },
    "billing": {
        "address": {
            "line1": "1467",
            "line2": "Harrison Street",
            "line3": "Harrison Street",
            "city": "San Fransico",
            "state": "California",
            "zip": "94122",
            "country": "US",
            "first_name": "joseph",
            "last_name": "Doe"
        },
        "phone": {
            "number": "8056594427",
            "country_code": "+91"
        }
    },
    "statement_descriptor_name": "joseph",
    "statement_descriptor_suffix": "JS",
    "metadata": {
        "udf1": "value1",
        "new_customer": "true",
        "login_date": "2019-09-10T10:11:12Z"
    }
}'
```
6. List PM for merchant - 
```
curl --location --request GET 'http://localhost:8080/account/payment_methods?client_secret=pay_57JSBXgwKCHdgIyvLXDi_secret_1arVvWIGuP7XlWPMrt8W' \
--header 'Accept: application/json' \
--header 'api-key: pk_dev_588a676d99f8470ca0eb3a6d5b955f2a' \
--data-raw ''
```
7. Link token API - 
```
curl --location --request POST 'http://localhost:8080/payment_methods/auth/link' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: pk_dev_588a676d99f8470ca0eb3a6d5b955f2a' \
--data-raw '{
    "client_secret": "pay_57JSBXgwKCHdgIyvLXDi_secret_1arVvWIGuP7XlWPMrt8W",
    "payment_id": "pay_57JSBXgwKCHdgIyvLXDi",
    "payment_method": "bank_debit",
    "payment_method_type": "ach"
}'
```
8. Get public token from plaid -
```
curl --location --request POST 'https://sandbox.plaid.com/sandbox/public_token/create' \
--header 'Content-Type: application/json' \
--header 'api-key: dev_O08Uv8O5PjAfrxsSLuduC1bn3FTWEpk0N2vvAKnvhAfc7jISAnrNgkTtcbibjO00' \
--data-raw '{
    "client_id": "some key",
    "secret": "some key",
    "institution_id": "ins_20",
    "initial_products": [
        "auth"
    ],
    "options": {
        "webhook": "https://www.genericwebhookurl.com/webhook"
    }
}'
``` 
9. Exchange token API - 
```
curl --location --request POST 'http://localhost:8080/payment_methods/auth/exchange' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: pk_dev_588a676d99f8470ca0eb3a6d5b955f2a' \
--data-raw '{
    "client_secret": "pay_57JSBXgwKCHdgIyvLXDi_secret_1arVvWIGuP7XlWPMrt8W",
    "public_token": "public-sandbox-19e07c9f-8b96-4e96-81a1-e3e3d5c70dc6",
    "payment_id": "pay_57JSBXgwKCHdgIyvLXDi",
    "payment_method": "bank_debit",
    "payment_method_type": "ach"
}'
```
10. Pm list for customer (take any payment token) - 
```
curl --location --request GET 'http://localhost:8080/customers/StripeCustomer/payment_methods' \
--header 'Accept: application/json' \
--header 'api-key: dev_O08Uv8O5PjAfrxsSLuduC1bn3FTWEpk0N2vvAKnvhAfc7jISAnrNgkTtcbibjO00'
```
11. Payments confirm with payment_token - 
```
curl --location --request POST 'http://localhost:8080/payments/pay_57JSBXgwKCHdgIyvLXDi/confirm' \
--header 'Content-Type: application/json' \
--header 'Accept: application/json' \
--header 'api-key: dev_O08Uv8O5PjAfrxsSLuduC1bn3FTWEpk0N2vvAKnvhAfc7jISAnrNgkTtcbibjO00' \
--data-raw '{
    "payment_method": "bank_debit",
    "payment_method_type": "ach",
    "payment_token": "token_2OqGjTK08Gen3iTEJq6u"
}'
```

## Checklist
<!-- Put an `x` in the boxes that apply -->

- [x] I formatted the code `cargo +nightly fmt --all`
- [x] I addressed lints thrown by `cargo clippy`
- [x] I reviewed the submitted code
- [ ] I added unit tests for my changes where possible
- [ ] I added a [CHANGELOG](/CHANGELOG.md) entry if applicable
