feat(pm_auth): pm_auth service migration (#3047)

Co-authored-by: Sarthak Soni <76486416+Sarthak1799@users.noreply.github.com>
Co-authored-by: hyperswitch-bot[bot] <148525504+hyperswitch-bot[bot]@users.noreply.github.com>
Co-authored-by: Sarthak Soni <sarthak.soni@juspay.in>

.github/workflows/CI-pr.yml

@@ -203,6 +203,11 @@ jobs:
           else
             echo "test_utils_changes_exist=true" >> $GITHUB_ENV
           fi
+          if git diff --submodule=diff --exit-code --quiet origin/$GITHUB_BASE_REF -- crates/pm_auth/; then
+            echo "pm_auth_changes_exist=false" >> $GITHUB_ENV
+          else
+            echo "pm_auth_changes_exist=true" >> $GITHUB_ENV
+          fi
 
       - name: Cargo hack api_models
         if: env.api_models_changes_exist == 'true'
@@ -249,6 +254,11 @@ jobs:
         shell: bash
         run: cargo hack check --each-feature --no-dev-deps -p redis_interface
 
+      - name: Cargo hack pm_auth
+        if: env.pm_auth_changes_exist == 'true'
+        shell: bash
+        run: cargo hack check --each-feature --no-dev-deps -p pm_auth
+
       - name: Cargo hack router
         if: env.router_changes_exist == 'true'
         shell: bash
@@ -456,6 +466,11 @@ jobs:
           else
             echo "test_utils_changes_exist=true" >> $GITHUB_ENV
           fi
+          if git diff --submodule=diff --exit-code --quiet origin/$GITHUB_BASE_REF -- crates/pm_auth/; then
+            echo "pm_auth_changes_exist=false" >> $GITHUB_ENV
+          else
+            echo "pm_auth_changes_exist=true" >> $GITHUB_ENV
+          fi
 
       - name: Cargo hack api_models
         if: env.api_models_changes_exist == 'true'
@@ -502,6 +517,11 @@ jobs:
         shell: bash
         run: cargo hack check --each-feature --no-dev-deps -p redis_interface
 
+      - name: Cargo hack pm_auth
+        if: env.pm_auth_changes_exist == 'true'
+        shell: bash
+        run: cargo hack check --each-feature --no-dev-deps -p pm_auth
+
       - name: Cargo hack router
         if: env.router_changes_exist == 'true'
         shell: bash

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",

config/config.example.toml

@@ -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

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

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

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" }

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()
+}

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;

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
+);

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"

crates/pm_auth/README.md

@@ -0,0 +1,3 @@
+# Payment Method Auth Services
+
+An open banking services for payment method auth validation

crates/pm_auth/src/connector.rs

@@ -0,0 +1,3 @@
+pub mod plaid;
+
+pub use self::plaid::Plaid;

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)
+    }
+}

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>,
+}

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";

crates/pm_auth/src/core.rs

@@ -0,0 +1 @@
+pub mod errors;

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,
+}

crates/pm_auth/src/lib.rs

@@ -0,0 +1,4 @@
+pub mod connector;
+pub mod consts;
+pub mod core;
+pub mod types;

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>,
+}

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,
+        })
+    }
+}

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,
+>
+{
+}

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" }

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,

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;

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)
+        api_enums::Connector::Plaid => {
-            .attach_printable(format!("invalid connector name: {connector_name}"))),
+            PlaidAuthType::foreign_try_from(val)?;
+            Ok(())
+        }
     }
 }
 

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(_) => {
+            storage::PaymentTokenData::AuthBankDebit(auth_token) => {
-                Err(errors::ApiErrorResponse::NotImplemented {
+                core_pm_auth::retrieve_payment_method_from_auth_service(
-                    message: errors::NotImplementedMessage::Default,
+                    state,
-                })
+                    merchant_key_store,
-                .into_report()
+                    auth_token,
+                    payment_intent,
+                )
+                .await
             }
         }
     }

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 {
+    for mca in &filtered_mcas {
-        let payment_methods = match mca.payment_methods_enabled {
+        let payment_methods = match &mca.payment_methods_enabled {
-            Some(pm) => pm,
+            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(),
             }
         })
     }

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)))
+}

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")
+}

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),
+        }
+    }
+}

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"))]

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)))
     }
 }
 

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

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
+}

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;

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<(

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),
+    }
+}

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;
 

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)),
+        }
+    }
+}

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