From 8aa3a02bcf4ae97ad34b1cca781f4325659ebc6b Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Thu, 14 Nov 2024 10:47:01 -0800
Subject: [PATCH 01/13] Add authentication methods and unit tests.

---
 dbt/adapters/redshift/connections.py | 270 +++++++++++++++++----------
 tests/unit/test_auth_method.py       | 162 +++++++++++++++-
 2 files changed, 329 insertions(+), 103 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 521bcff86..ac12adf2a 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -37,10 +37,17 @@ def get_message(self) -> str:
 logger = AdapterLogger("Redshift")
 
 
+class IdentityCenterTokenType(StrEnum):
+    ACCESS_TOKEN = "ACCESS_TOKEN"
+    EXT_JWT = "EXT_JWT"
+
+
 class RedshiftConnectionMethod(StrEnum):
     DATABASE = "database"
     IAM = "iam"
     IAM_ROLE = "iam_role"
+    IAM_IDENTITY_CENTER_BROWSER = "iam_idc_browser"
+    IAM_IDENTITY_CENTER_TOKEN = "iam_idc_token"
 
 
 class UserSSLMode(StrEnum):
@@ -128,6 +135,22 @@ class RedshiftCredentials(Credentials):
     access_key_id: Optional[str] = None
     secret_access_key: Optional[str] = None
 
+    #
+    # IAM identity center methods
+    #
+
+    # browser
+    credentials_provider: Optional[str] = None
+    idc_region: Optional[str] = None
+    issuer_url: Optional[str] = None
+    listen_port: int = 7890
+    idc_client_display_name: Optional[str] = "Amazon Redshift driver"
+    idp_response_timeout: int = 60
+
+    # token
+    token: Optional[str] = None
+    token_type: Optional[str] = None
+
     _ALIASES = {"dbname": "database", "pass": "password"}
 
     @property
@@ -163,131 +186,181 @@ def unique_field(self) -> str:
         return self.host
 
 
-class RedshiftConnectMethodFactory:
-    credentials: RedshiftCredentials
+def get_connection_method(
+    credentials: RedshiftCredentials,
+) -> Callable[[], redshift_connector.Connection]:
+    #
+    # Helper Methods
+    #
+    def __assert_required_fields(credentials, required_fields, method_name):
+        missing_fields = [
+            field for field in required_fields if getattr(credentials, field, None) is None
+        ]
+        if missing_fields:
+            fields_str = "', '".join(missing_fields)
+            raise FailedToConnectError(
+                f"'{fields_str}' field(s) are required for '{method_name}' credentials method"
+            )
 
-    def __init__(self, credentials) -> None:
-        self.credentials = credentials
+    def __base_kwargs(credentials) -> Dict[str, Any]:
+        redshift_ssl_config = RedshiftSSLConfig.parse(credentials.sslmode).to_dict()
+        return {
+            "host": credentials.host,
+            "port": int(credentials.port) if credentials.port else 5439,
+            "database": credentials.database,
+            "region": credentials.region,
+            "auto_create": credentials.autocreate,
+            "db_groups": credentials.db_groups,
+            "timeout": credentials.connect_timeout,
+            **redshift_ssl_config,
+        }
 
-    def get_connect_method(self) -> Callable[[], redshift_connector.Connection]:
+    def __iam_kwargs(credentials) -> Dict[str, Any]:
 
-        # Support missing 'method' for backwards compatibility
-        method = self.credentials.method or RedshiftConnectionMethod.DATABASE
-        if method == RedshiftConnectionMethod.DATABASE:
-            kwargs = self._database_kwargs
-        elif method == RedshiftConnectionMethod.IAM:
-            kwargs = self._iam_user_kwargs
-        elif method == RedshiftConnectionMethod.IAM_ROLE:
-            kwargs = self._iam_role_kwargs
+        if "serverless" in credentials.host:
+            cluster_identifier = None
+        elif credentials.cluster_id:
+            cluster_identifier = credentials.cluster_id
         else:
-            raise FailedToConnectError(f"Invalid 'method' in profile: '{method}'")
+            raise FailedToConnectError(
+                "Failed to use IAM method:"
+                " 'cluster_id' must be provided for provisioned cluster"
+                " 'host' must be provided for serverless endpoint"
+            )
+
+        iam_specific_kwargs = {
+            "iam": True,
+            "user": "",
+            "password": "",
+            "cluster_identifier": cluster_identifier,
+        }
 
-        def connect() -> redshift_connector.Connection:
-            c = redshift_connector.connect(**kwargs)
-            if self.credentials.autocommit:
-                c.autocommit = True
-            if self.credentials.role:
-                c.cursor().execute(f"set role {self.credentials.role}")
-            return c
+        return __base_kwargs(credentials) | iam_specific_kwargs
 
-        return connect
+    def __database_kwargs(credentials) -> Dict[str, Any]:
+        logger.debug("Connecting to Redshift with 'database' credentials method")
 
-    @property
-    def _database_kwargs(self) -> Dict[str, Any]:
-        logger.debug("Connecting to redshift with 'database' credentials method")
-        kwargs = self._base_kwargs
-
-        if self.credentials.user and self.credentials.password:
-            kwargs.update(
-                user=self.credentials.user,
-                password=self.credentials.password,
-            )
-        else:
-            raise FailedToConnectError(
-                "'user' and 'password' fields are required for 'database' credentials method"
-            )
+        __assert_required_fields(credentials, ["user", "password"], "database")
 
-        return kwargs
+        db_credentials = {
+            "user": credentials.user,
+            "password": credentials.password,
+        }
 
-    @property
-    def _iam_user_kwargs(self) -> Dict[str, Any]:
-        logger.debug("Connecting to redshift with 'iam' credentials method")
-        kwargs = self._iam_kwargs
-
-        if self.credentials.access_key_id and self.credentials.secret_access_key:
-            kwargs.update(
-                access_key_id=self.credentials.access_key_id,
-                secret_access_key=self.credentials.secret_access_key,
-            )
-        elif self.credentials.access_key_id or self.credentials.secret_access_key:
+        return __base_kwargs(credentials) | db_credentials
+
+    def __iam_user_kwargs(credentials) -> Dict[str, Any]:
+        logger.debug("Connecting to Redshift with 'iam' credentials method")
+
+        if credentials.access_key_id and credentials.secret_access_key:
+            iam_credentials = {
+                "access_key_id": credentials.access_key_id,
+                "secret_access_key": credentials.secret_access_key,
+            }
+        elif credentials.access_key_id or credentials.secret_access_key:
             raise FailedToConnectError(
                 "'access_key_id' and 'secret_access_key' are both needed if providing explicit credentials"
             )
         else:
-            kwargs.update(profile=self.credentials.iam_profile)
+            iam_credentials = {"profile": credentials.iam_profile}
 
-        if user := self.credentials.user:
-            kwargs.update(db_user=user)
-        else:
-            raise FailedToConnectError("'user' field is required for 'iam' credentials method")
+        __assert_required_fields(credentials, ["user"], "iam")
+        iam_credentials["db_user"] = credentials.user
 
-        return kwargs
+        return __iam_kwargs(credentials) | iam_credentials
 
-    @property
-    def _iam_role_kwargs(self) -> Dict[str, Optional[Any]]:
-        logger.debug("Connecting to redshift with 'iam_role' credentials method")
-        kwargs = self._iam_kwargs
+    def __iam_role_kwargs(credentials) -> Dict[str, Any]:
+        logger.debug("Connecting to Redshift with 'iam_role' credentials method")
+        role_kwargs = {
+            "db_user": None,
+            "group_federation": "serverless" not in credentials.host,
+        }
 
-        # It's a role, we're ignoring the user
-        kwargs.update(db_user=None)
+        if credentials.iam_profile:
+            role_kwargs["profile"] = credentials.iam_profile
 
-        # Serverless shouldn't get group_federation, Provisoned clusters should
-        if "serverless" in self.credentials.host:
-            kwargs.update(group_federation=False)
-        else:
-            kwargs.update(group_federation=True)
+        return __iam_kwargs(credentials) | role_kwargs
 
-        if iam_profile := self.credentials.iam_profile:
-            kwargs.update(profile=iam_profile)
+    def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
+        logger.debug("Connecting to Redshift with 'iam_idc_browser' credentials method")
+        identity_center_method_name = "BrowserIdcAuthPlugin"
 
-        return kwargs
+        if credentials.credentials_provider != identity_center_method_name:
+            raise FailedToConnectError(
+                f"'credentials_provider' must be set to '{identity_center_method_name}'"
+            )
 
-    @property
-    def _iam_kwargs(self) -> Dict[str, Any]:
-        kwargs = self._base_kwargs
-        kwargs.update(
-            iam=True,
-            user="",
-            password="",
+        __assert_required_fields(
+            credentials, ["credentials_provider", "idc_region", "issuer_url"], "iam_idc_browser"
         )
 
-        if "serverless" in self.credentials.host:
-            kwargs.update(cluster_identifier=None)
-        elif cluster_id := self.credentials.cluster_id:
-            kwargs.update(cluster_identifier=cluster_id)
-        else:
+        idc_kwargs = {
+            "credentials_provider": identity_center_method_name,
+            "idc_region": credentials.idc_region,
+            "issuer_url": credentials.issuer_url,
+            "idc_client_display_name": credentials.idc_client_display_name,
+            "idp_response_timeout": credentials.idp_response_timeout,
+        }
+
+        return __iam_kwargs(credentials) | idc_kwargs
+
+    def __iam_idc_token_kwargs(credentials) -> Dict[str, Any]:
+        logger.debug("Connecting to Redshift with 'iam_idc_token' credentials method")
+        identity_center_method_name = "IdpTokenAuthPlugin"
+
+        if credentials.credentials_provider != identity_center_method_name:
             raise FailedToConnectError(
-                "Failed to use IAM method:"
-                "    'cluster_id' must be provided for provisioned cluster"
-                "    'host' must be provided for serverless endpoint"
+                f"'credentials_provider' must be set to '{identity_center_method_name}'"
             )
 
-        return kwargs
+        __assert_required_fields(
+            credentials, ["credentials_provider", "token", "token_type"], "iam_idc_token"
+        )
 
-    @property
-    def _base_kwargs(self) -> Dict[str, Any]:
-        kwargs = {
-            "host": self.credentials.host,
-            "port": int(self.credentials.port) if self.credentials.port else int(5439),
-            "database": self.credentials.database,
-            "region": self.credentials.region,
-            "auto_create": self.credentials.autocreate,
-            "db_groups": self.credentials.db_groups,
-            "timeout": self.credentials.connect_timeout,
+        try:
+            _ = IdentityCenterTokenType(credentials.token_type)
+        except ValueError:
+            raise FailedToConnectError(
+                f"'token_type' must be set to one of {[token.value for token in iter(IdentityCenterTokenType)]}"
+            )
+
+        idc_token_kwargs = {
+            "credentials_provider": identity_center_method_name,
+            "token": credentials.token,
+            "token_type": credentials.token_type,
         }
-        redshift_ssl_config = RedshiftSSLConfig.parse(self.credentials.sslmode)
-        kwargs.update(redshift_ssl_config.to_dict())
-        return kwargs
+
+        return __iam_kwargs(credentials) | idc_token_kwargs
+
+    #
+    # Head of function execution
+    #
+
+    method_to_kwargs_function = {
+        None: __database_kwargs,
+        RedshiftConnectionMethod.DATABASE: __database_kwargs,
+        RedshiftConnectionMethod.IAM: __iam_user_kwargs,
+        RedshiftConnectionMethod.IAM_ROLE: __iam_role_kwargs,
+        RedshiftConnectionMethod.IAM_IDENTITY_CENTER_BROWSER: __iam_idc_browser_kwargs,
+        RedshiftConnectionMethod.IAM_IDENTITY_CENTER_TOKEN: __iam_idc_token_kwargs,
+    }
+
+    try:
+        kwargs_function = method_to_kwargs_function[credentials.method]
+    except KeyError:
+        raise FailedToConnectError(f"Invalid 'method' in profile: '{credentials.method}'")
+
+    kwargs = kwargs_function(credentials)
+
+    def connect() -> redshift_connector.Connection:
+        c = redshift_connector.connect(**kwargs)
+        if credentials.autocommit:
+            c.autocommit = True
+        if credentials.role:
+            c.cursor().execute(f"set role {credentials.role}")
+        return c
+
+    return connect
 
 
 class RedshiftConnectionManager(SQLConnectionManager):
@@ -373,7 +446,6 @@ def open(cls, connection):
             return connection
 
         credentials = connection.credentials
-        connect_method_factory = RedshiftConnectMethodFactory(credentials)
 
         def exponential_backoff(attempt: int):
             return attempt * attempt
@@ -387,7 +459,7 @@ def exponential_backoff(attempt: int):
 
         open_connection = cls.retry_connection(
             connection,
-            connect=connect_method_factory.get_connect_method(),
+            connect=get_connection_method(credentials),
             logger=logger,
             retry_limit=credentials.retries,
             retry_timeout=exponential_backoff,
diff --git a/tests/unit/test_auth_method.py b/tests/unit/test_auth_method.py
index 55b1aad74..bc2876672 100644
--- a/tests/unit/test_auth_method.py
+++ b/tests/unit/test_auth_method.py
@@ -9,7 +9,7 @@
     Plugin as RedshiftPlugin,
     RedshiftAdapter,
 )
-from dbt.adapters.redshift.connections import RedshiftConnectMethodFactory, RedshiftSSLConfig
+from dbt.adapters.redshift.connections import get_connection_method, RedshiftSSLConfig
 from tests.unit.utils import config_from_parts_or_dicts, inject_adapter
 
 
@@ -61,7 +61,7 @@ def test_invalid_auth_method(self):
         # we have to set method this way, otherwise it won't validate
         self.config.credentials.method = "badmethod"
         with self.assertRaises(FailedToConnectError) as context:
-            connect_method_factory = RedshiftConnectMethodFactory(self.config.credentials)
+            connect_method_factory = get_connection_method(self.config.credentials)
             connect_method_factory.get_connect_method()
         self.assertTrue("badmethod" in context.exception.msg)
 
@@ -221,7 +221,7 @@ def test_iam_optionals(self):
     def test_no_cluster_id(self):
         self.config.credentials = self.config.credentials.replace(method="iam")
         with self.assertRaises(FailedToConnectError) as context:
-            connect_method_factory = RedshiftConnectMethodFactory(self.config.credentials)
+            connect_method_factory = get_connection_method(self.config.credentials)
             connect_method_factory.get_connect_method()
 
         self.assertTrue("'cluster_id' must be provided" in context.exception.msg)
@@ -400,7 +400,7 @@ class TestIAMRoleMethod(AuthMethod):
     def test_no_cluster_id(self):
         self.config.credentials = self.config.credentials.replace(method="iam_role")
         with self.assertRaises(FailedToConnectError) as context:
-            connect_method_factory = RedshiftConnectMethodFactory(self.config.credentials)
+            connect_method_factory = get_connection_method(self.config.credentials)
             connect_method_factory.get_connect_method()
 
         self.assertTrue("'cluster_id' must be provided" in context.exception.msg)
@@ -573,3 +573,157 @@ def test_profile_invalid_serverless(self):
                 **DEFAULT_SSL_CONFIG,
             )
         self.assertTrue("'host' must be provided" in context.exception.msg)
+
+
+class TestIAMIdcBrowser(AuthMethod):
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_profile_idc_browser_all_fields(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_idc_browser",
+            credentials_provider="BrowserIdcAuthPlugin",
+            idc_region="us-east-1",
+            issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
+            idc_client_display_name="display name",
+            idp_response_timeout=0,
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+        )
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            iam=True,
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+            database="redshift",
+            cluster_identifier=None,
+            region=None,
+            auto_create=False,
+            db_groups=[],
+            password="",
+            user="",
+            timeout=None,
+            port=5439,
+            **DEFAULT_SSL_CONFIG,
+            idp_response_timeout=0,
+            idc_client_display_name="display name",
+            credentials_provider="BrowserIdcAuthPlugin",
+            idc_region="us-east-1",
+            issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
+        )
+
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_profile_idc_browser_required_fields_only(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_idc_browser",
+            credentials_provider="BrowserIdcAuthPlugin",
+            idc_region="us-east-1",
+            issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+        )
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            iam=True,
+            host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+            database="redshift",
+            cluster_identifier=None,
+            region=None,
+            auto_create=False,
+            db_groups=[],
+            password="",
+            user="",
+            timeout=None,
+            port=5439,
+            **DEFAULT_SSL_CONFIG,
+            idp_response_timeout=60,
+            idc_client_display_name="Amazon Redshift driver",
+            credentials_provider="BrowserIdcAuthPlugin",
+            idc_region="us-east-1",
+            issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
+        )
+
+    def test_invalid_plugin_for_idc_browser_auth_method(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_idc_browser",
+            credentials_provider="IdpTokenAuthPlugin",
+        )
+        with self.assertRaises(FailedToConnectError) as context:
+            connection = self.adapter.acquire_connection("dummy")
+            connection.handle
+
+        assert "BrowserIdcAuthPlugin" in context.exception.msg
+
+    def test_invalid_adapter_missing_fields(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_idc_browser",
+            credentials_provider="BrowserIdcAuthPlugin",
+            idc_client_display_name="my display",
+        )
+        with self.assertRaises(FailedToConnectError) as context:
+            connection = self.adapter.acquire_connection("dummy")
+            connection.handle
+
+        assert (
+            "'idc_region', 'issuer_url' field(s) are required for 'iam_idc_browser' credentials method"
+            in context.exception.msg
+        )
+
+
+class TestIAMIdcToken(AuthMethod):
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_profile_idc_token_all_required_fields(self):
+        """Same as all possible fields"""
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_idc_token",
+            credentials_provider="IdpTokenAuthPlugin",
+            token="token",
+            token_type="ACCESS_TOKEN",
+            host="doesnotexist.1235.us-east-2.redshift-serverless.amazonaws.com",
+        )
+        connection = self.adapter.acquire_connection("dummy")
+        connection.handle
+        redshift_connector.connect.assert_called_once_with(
+            iam=True,
+            host="doesnotexist.1235.us-east-2.redshift-serverless.amazonaws.com",
+            database="redshift",
+            cluster_identifier=None,
+            region=None,
+            auto_create=False,
+            db_groups=[],
+            password="",
+            user="",
+            timeout=None,
+            port=5439,
+            **DEFAULT_SSL_CONFIG,
+            credentials_provider="IdpTokenAuthPlugin",
+            token="token",
+            token_type="ACCESS_TOKEN",
+        )
+
+    def test_invalid_plugin_for_idc_token_auth_method(self):
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_idc_token",
+            token="token",
+            token_type="ACCESS_TOKEN",
+            credentials_provider="BrowserIdcAuthPlugin",
+        )
+        with self.assertRaises(FailedToConnectError) as context:
+            connection = self.adapter.acquire_connection("dummy")
+            connection.handle
+
+        assert "IdpTokenAuthPlugin" in context.exception.msg
+
+    @mock.patch("redshift_connector.connect", MagicMock())
+    def test_invalid_idc_token_missing_field(self):
+        # Successful test
+        self.config.credentials = self.config.credentials.replace(
+            method="iam_idc_token",
+            credentials_provider="IdpTokenAuthPlugin",
+            token_type="ACCESS_TOKEN",
+            host="doesnotexist.1235.us-east-2.redshift-serverless.amazonaws.com",
+        )
+        with self.assertRaises(FailedToConnectError) as context:
+            connection = self.adapter.acquire_connection("dummy")
+            connection.handle
+        assert (
+            "'token' field(s) are required for 'iam_idc_token' credentials method"
+            in context.exception.msg
+        )

From 5fe5af1d3e9621e7afb5e60265c277729bb517d5 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Thu, 14 Nov 2024 14:00:30 -0800
Subject: [PATCH 02/13] Document types and exclude IAM=true from identity
 center connections

---
 dbt/adapters/redshift/connections.py | 55 +++++++++++++++++-----------
 tests/unit/test_auth_method.py       |  8 ++--
 2 files changed, 38 insertions(+), 25 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index ac12adf2a..224adef88 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -49,6 +49,10 @@ class RedshiftConnectionMethod(StrEnum):
     IAM_IDENTITY_CENTER_BROWSER = "iam_idc_browser"
     IAM_IDENTITY_CENTER_TOKEN = "iam_idc_token"
 
+    @staticmethod
+    def uses_identity_center(method: str):
+        return "_idc_" in method
+
 
 class UserSSLMode(StrEnum):
     disable = "disable"
@@ -192,18 +196,20 @@ def get_connection_method(
     #
     # Helper Methods
     #
-    def __assert_required_fields(credentials, required_fields, method_name):
-        missing_fields = [
+    def __assert_required_fields(method_name: str, required_fields: Tuple[str, ...]):
+        missing_fields: List[str] = [
             field for field in required_fields if getattr(credentials, field, None) is None
         ]
         if missing_fields:
-            fields_str = "', '".join(missing_fields)
+            fields_str: str = "', '".join(missing_fields)
             raise FailedToConnectError(
                 f"'{fields_str}' field(s) are required for '{method_name}' credentials method"
             )
 
     def __base_kwargs(credentials) -> Dict[str, Any]:
-        redshift_ssl_config = RedshiftSSLConfig.parse(credentials.sslmode).to_dict()
+        redshift_ssl_config: Dict[str, Any] = RedshiftSSLConfig.parse(
+            credentials.sslmode
+        ).to_dict()
         return {
             "host": credentials.host,
             "port": int(credentials.port) if credentials.port else 5439,
@@ -217,7 +223,13 @@ def __base_kwargs(credentials) -> Dict[str, Any]:
 
     def __iam_kwargs(credentials) -> Dict[str, Any]:
 
-        if "serverless" in credentials.host:
+        # iam True except for identity center methods
+        iam: bool = not RedshiftConnectionMethod.uses_identity_center(credentials.method)
+
+        cluster_identifier: Optional[str]
+        if "serverless" in credentials.host or RedshiftConnectionMethod.uses_identity_center(
+            credentials.method
+        ):
             cluster_identifier = None
         elif credentials.cluster_id:
             cluster_identifier = credentials.cluster_id
@@ -228,8 +240,8 @@ def __iam_kwargs(credentials) -> Dict[str, Any]:
                 " 'host' must be provided for serverless endpoint"
             )
 
-        iam_specific_kwargs = {
-            "iam": True,
+        iam_specific_kwargs: Dict[str, Any] = {
+            "iam": iam,
             "user": "",
             "password": "",
             "cluster_identifier": cluster_identifier,
@@ -240,9 +252,9 @@ def __iam_kwargs(credentials) -> Dict[str, Any]:
     def __database_kwargs(credentials) -> Dict[str, Any]:
         logger.debug("Connecting to Redshift with 'database' credentials method")
 
-        __assert_required_fields(credentials, ["user", "password"], "database")
+        __assert_required_fields("database", ("user", "password"))
 
-        db_credentials = {
+        db_credentials: Dict[str, Any] = {
             "user": credentials.user,
             "password": credentials.password,
         }
@@ -252,6 +264,7 @@ def __database_kwargs(credentials) -> Dict[str, Any]:
     def __iam_user_kwargs(credentials) -> Dict[str, Any]:
         logger.debug("Connecting to Redshift with 'iam' credentials method")
 
+        iam_credentials: Dict[str, Any]
         if credentials.access_key_id and credentials.secret_access_key:
             iam_credentials = {
                 "access_key_id": credentials.access_key_id,
@@ -264,7 +277,7 @@ def __iam_user_kwargs(credentials) -> Dict[str, Any]:
         else:
             iam_credentials = {"profile": credentials.iam_profile}
 
-        __assert_required_fields(credentials, ["user"], "iam")
+        __assert_required_fields("iam", ("user",))
         iam_credentials["db_user"] = credentials.user
 
         return __iam_kwargs(credentials) | iam_credentials
@@ -283,7 +296,7 @@ def __iam_role_kwargs(credentials) -> Dict[str, Any]:
 
     def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
         logger.debug("Connecting to Redshift with 'iam_idc_browser' credentials method")
-        identity_center_method_name = "BrowserIdcAuthPlugin"
+        identity_center_method_name: str = "BrowserIdcAuthPlugin"
 
         if credentials.credentials_provider != identity_center_method_name:
             raise FailedToConnectError(
@@ -291,13 +304,13 @@ def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
             )
 
         __assert_required_fields(
-            credentials, ["credentials_provider", "idc_region", "issuer_url"], "iam_idc_browser"
+            "iam_idc_browser", ("credentials_provider", "idc_region", "issuer_url")
         )
 
-        idc_kwargs = {
+        idc_kwargs: Dict[str, Any] = {
             "credentials_provider": identity_center_method_name,
-            "idc_region": credentials.idc_region,
             "issuer_url": credentials.issuer_url,
+            "idc_region": credentials.idc_region,
             "idc_client_display_name": credentials.idc_client_display_name,
             "idp_response_timeout": credentials.idp_response_timeout,
         }
@@ -306,16 +319,14 @@ def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
 
     def __iam_idc_token_kwargs(credentials) -> Dict[str, Any]:
         logger.debug("Connecting to Redshift with 'iam_idc_token' credentials method")
-        identity_center_method_name = "IdpTokenAuthPlugin"
+        identity_center_method_name: str = "IdpTokenAuthPlugin"
 
         if credentials.credentials_provider != identity_center_method_name:
             raise FailedToConnectError(
                 f"'credentials_provider' must be set to '{identity_center_method_name}'"
             )
 
-        __assert_required_fields(
-            credentials, ["credentials_provider", "token", "token_type"], "iam_idc_token"
-        )
+        __assert_required_fields("iam_idc_token", ("credentials_provider", "token", "token_type"))
 
         try:
             _ = IdentityCenterTokenType(credentials.token_type)
@@ -324,7 +335,7 @@ def __iam_idc_token_kwargs(credentials) -> Dict[str, Any]:
                 f"'token_type' must be set to one of {[token.value for token in iter(IdentityCenterTokenType)]}"
             )
 
-        idc_token_kwargs = {
+        idc_token_kwargs: Dict[str, Any] = {
             "credentials_provider": identity_center_method_name,
             "token": credentials.token,
             "token_type": credentials.token_type,
@@ -346,11 +357,13 @@ def __iam_idc_token_kwargs(credentials) -> Dict[str, Any]:
     }
 
     try:
-        kwargs_function = method_to_kwargs_function[credentials.method]
+        kwargs_function: Callable[[RedshiftCredentials], Dict[str, Any]] = (
+            method_to_kwargs_function[credentials.method]
+        )
     except KeyError:
         raise FailedToConnectError(f"Invalid 'method' in profile: '{credentials.method}'")
 
-    kwargs = kwargs_function(credentials)
+    kwargs: Dict[str, Any] = kwargs_function(credentials)
 
     def connect() -> redshift_connector.Connection:
         c = redshift_connector.connect(**kwargs)
diff --git a/tests/unit/test_auth_method.py b/tests/unit/test_auth_method.py
index bc2876672..a1f0d3d3f 100644
--- a/tests/unit/test_auth_method.py
+++ b/tests/unit/test_auth_method.py
@@ -79,7 +79,7 @@ def test_missing_region_failure(self):
             connection = self.adapter.acquire_connection("dummy")
             connection.handle
             redshift_connector.connect.assert_called_once_with(
-                iam=True,
+                iam=False,
                 host="doesnotexist.1233_no_region",
                 database="redshift",
                 cluster_identifier=None,
@@ -590,7 +590,7 @@ def test_profile_idc_browser_all_fields(self):
         connection = self.adapter.acquire_connection("dummy")
         connection.handle
         redshift_connector.connect.assert_called_once_with(
-            iam=True,
+            iam=False,
             host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
             database="redshift",
             cluster_identifier=None,
@@ -621,7 +621,7 @@ def test_profile_idc_browser_required_fields_only(self):
         connection = self.adapter.acquire_connection("dummy")
         connection.handle
         redshift_connector.connect.assert_called_once_with(
-            iam=True,
+            iam=False,
             host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
             database="redshift",
             cluster_identifier=None,
@@ -681,7 +681,7 @@ def test_profile_idc_token_all_required_fields(self):
         connection = self.adapter.acquire_connection("dummy")
         connection.handle
         redshift_connector.connect.assert_called_once_with(
-            iam=True,
+            iam=False,
             host="doesnotexist.1235.us-east-2.redshift-serverless.amazonaws.com",
             database="redshift",
             cluster_identifier=None,

From 7ba565747ae4c67e8543e89210c5eb0271f22662 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Thu, 14 Nov 2024 14:00:53 -0800
Subject: [PATCH 03/13] Update redshift connector version to accept new auth
 method.

---
 setup.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/setup.py b/setup.py
index 675b8588e..1ef723c47 100644
--- a/setup.py
+++ b/setup.py
@@ -53,7 +53,7 @@ def _plugin_version() -> str:
         "dbt-postgres>=1.8,<1.10",
         # dbt-redshift depends deeply on this package. it does not follow SemVer, therefore there have been breaking changes in previous patch releases
         # Pin to the patch or minor version, and bump in each new minor version of dbt-redshift.
-        "redshift-connector<2.1.1,>=2.0.913,!=2.0.914",
+        "redshift-connector==2.1.3",
         # add dbt-core to ensure backwards compatibility of installation, this is not a functional dependency
         "dbt-core>=1.8.0b3",
         # installed via dbt-core but referenced directly; don't pin to avoid version conflicts with dbt-core

From cd97f7bf24a702be623e74be084ce90c45d4b41a Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 14:10:55 -0800
Subject: [PATCH 04/13] Pare this pull request down to Browser based
 authentication for now

---
 dbt/adapters/redshift/connections.py | 28 -------------
 tests/unit/test_auth_method.py       | 62 ----------------------------
 2 files changed, 90 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 224adef88..9461d847a 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -47,7 +47,6 @@ class RedshiftConnectionMethod(StrEnum):
     IAM = "iam"
     IAM_ROLE = "iam_role"
     IAM_IDENTITY_CENTER_BROWSER = "iam_idc_browser"
-    IAM_IDENTITY_CENTER_TOKEN = "iam_idc_token"
 
     @staticmethod
     def uses_identity_center(method: str):
@@ -317,32 +316,6 @@ def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
 
         return __iam_kwargs(credentials) | idc_kwargs
 
-    def __iam_idc_token_kwargs(credentials) -> Dict[str, Any]:
-        logger.debug("Connecting to Redshift with 'iam_idc_token' credentials method")
-        identity_center_method_name: str = "IdpTokenAuthPlugin"
-
-        if credentials.credentials_provider != identity_center_method_name:
-            raise FailedToConnectError(
-                f"'credentials_provider' must be set to '{identity_center_method_name}'"
-            )
-
-        __assert_required_fields("iam_idc_token", ("credentials_provider", "token", "token_type"))
-
-        try:
-            _ = IdentityCenterTokenType(credentials.token_type)
-        except ValueError:
-            raise FailedToConnectError(
-                f"'token_type' must be set to one of {[token.value for token in iter(IdentityCenterTokenType)]}"
-            )
-
-        idc_token_kwargs: Dict[str, Any] = {
-            "credentials_provider": identity_center_method_name,
-            "token": credentials.token,
-            "token_type": credentials.token_type,
-        }
-
-        return __iam_kwargs(credentials) | idc_token_kwargs
-
     #
     # Head of function execution
     #
@@ -353,7 +326,6 @@ def __iam_idc_token_kwargs(credentials) -> Dict[str, Any]:
         RedshiftConnectionMethod.IAM: __iam_user_kwargs,
         RedshiftConnectionMethod.IAM_ROLE: __iam_role_kwargs,
         RedshiftConnectionMethod.IAM_IDENTITY_CENTER_BROWSER: __iam_idc_browser_kwargs,
-        RedshiftConnectionMethod.IAM_IDENTITY_CENTER_TOKEN: __iam_idc_token_kwargs,
     }
 
     try:
diff --git a/tests/unit/test_auth_method.py b/tests/unit/test_auth_method.py
index a1f0d3d3f..4ac0586a2 100644
--- a/tests/unit/test_auth_method.py
+++ b/tests/unit/test_auth_method.py
@@ -665,65 +665,3 @@ def test_invalid_adapter_missing_fields(self):
             "'idc_region', 'issuer_url' field(s) are required for 'iam_idc_browser' credentials method"
             in context.exception.msg
         )
-
-
-class TestIAMIdcToken(AuthMethod):
-    @mock.patch("redshift_connector.connect", MagicMock())
-    def test_profile_idc_token_all_required_fields(self):
-        """Same as all possible fields"""
-        self.config.credentials = self.config.credentials.replace(
-            method="iam_idc_token",
-            credentials_provider="IdpTokenAuthPlugin",
-            token="token",
-            token_type="ACCESS_TOKEN",
-            host="doesnotexist.1235.us-east-2.redshift-serverless.amazonaws.com",
-        )
-        connection = self.adapter.acquire_connection("dummy")
-        connection.handle
-        redshift_connector.connect.assert_called_once_with(
-            iam=False,
-            host="doesnotexist.1235.us-east-2.redshift-serverless.amazonaws.com",
-            database="redshift",
-            cluster_identifier=None,
-            region=None,
-            auto_create=False,
-            db_groups=[],
-            password="",
-            user="",
-            timeout=None,
-            port=5439,
-            **DEFAULT_SSL_CONFIG,
-            credentials_provider="IdpTokenAuthPlugin",
-            token="token",
-            token_type="ACCESS_TOKEN",
-        )
-
-    def test_invalid_plugin_for_idc_token_auth_method(self):
-        self.config.credentials = self.config.credentials.replace(
-            method="iam_idc_token",
-            token="token",
-            token_type="ACCESS_TOKEN",
-            credentials_provider="BrowserIdcAuthPlugin",
-        )
-        with self.assertRaises(FailedToConnectError) as context:
-            connection = self.adapter.acquire_connection("dummy")
-            connection.handle
-
-        assert "IdpTokenAuthPlugin" in context.exception.msg
-
-    @mock.patch("redshift_connector.connect", MagicMock())
-    def test_invalid_idc_token_missing_field(self):
-        # Successful test
-        self.config.credentials = self.config.credentials.replace(
-            method="iam_idc_token",
-            credentials_provider="IdpTokenAuthPlugin",
-            token_type="ACCESS_TOKEN",
-            host="doesnotexist.1235.us-east-2.redshift-serverless.amazonaws.com",
-        )
-        with self.assertRaises(FailedToConnectError) as context:
-            connection = self.adapter.acquire_connection("dummy")
-            connection.handle
-        assert (
-            "'token' field(s) are required for 'iam_idc_token' credentials method"
-            in context.exception.msg
-        )

From f7389d3a3e3529aedf29eabe8ffbdd9a034b8c2f Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 14:32:56 -0800
Subject: [PATCH 05/13] Complete the browser based authentication method

---
 dbt/adapters/redshift/connections.py | 24 ++++++------------------
 tests/unit/test_auth_method.py       | 24 +++++-------------------
 2 files changed, 11 insertions(+), 37 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 9461d847a..495bedf49 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -46,11 +46,11 @@ class RedshiftConnectionMethod(StrEnum):
     DATABASE = "database"
     IAM = "iam"
     IAM_ROLE = "iam_role"
-    IAM_IDENTITY_CENTER_BROWSER = "iam_idc_browser"
+    IAM_IDENTITY_CENTER_BROWSER = "browser_identity_center"
 
     @staticmethod
     def uses_identity_center(method: str):
-        return "_idc_" in method
+        return method.endswith("identity_center")
 
 
 class UserSSLMode(StrEnum):
@@ -143,17 +143,13 @@ class RedshiftCredentials(Credentials):
     #
 
     # browser
-    credentials_provider: Optional[str] = None
+    authenticator: Optional[str] = None
     idc_region: Optional[str] = None
     issuer_url: Optional[str] = None
     listen_port: int = 7890
     idc_client_display_name: Optional[str] = "Amazon Redshift driver"
     idp_response_timeout: int = 60
 
-    # token
-    token: Optional[str] = None
-    token_type: Optional[str] = None
-
     _ALIASES = {"dbname": "database", "pass": "password"}
 
     @property
@@ -294,20 +290,12 @@ def __iam_role_kwargs(credentials) -> Dict[str, Any]:
         return __iam_kwargs(credentials) | role_kwargs
 
     def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
-        logger.debug("Connecting to Redshift with 'iam_idc_browser' credentials method")
-        identity_center_method_name: str = "BrowserIdcAuthPlugin"
+        logger.debug("Connecting to Redshift with '{credentials.method}' credentials method")
 
-        if credentials.credentials_provider != identity_center_method_name:
-            raise FailedToConnectError(
-                f"'credentials_provider' must be set to '{identity_center_method_name}'"
-            )
-
-        __assert_required_fields(
-            "iam_idc_browser", ("credentials_provider", "idc_region", "issuer_url")
-        )
+        __assert_required_fields("browser_identity_center", ("method", "idc_region", "issuer_url"))
 
         idc_kwargs: Dict[str, Any] = {
-            "credentials_provider": identity_center_method_name,
+            "credentials_provider": "BrowserIdcAuthPlugin",
             "issuer_url": credentials.issuer_url,
             "idc_region": credentials.idc_region,
             "idc_client_display_name": credentials.idc_client_display_name,
diff --git a/tests/unit/test_auth_method.py b/tests/unit/test_auth_method.py
index 4ac0586a2..64b98457c 100644
--- a/tests/unit/test_auth_method.py
+++ b/tests/unit/test_auth_method.py
@@ -579,8 +579,7 @@ class TestIAMIdcBrowser(AuthMethod):
     @mock.patch("redshift_connector.connect", MagicMock())
     def test_profile_idc_browser_all_fields(self):
         self.config.credentials = self.config.credentials.replace(
-            method="iam_idc_browser",
-            credentials_provider="BrowserIdcAuthPlugin",
+            method="browser_identity_center",
             idc_region="us-east-1",
             issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
             idc_client_display_name="display name",
@@ -612,8 +611,7 @@ def test_profile_idc_browser_all_fields(self):
     @mock.patch("redshift_connector.connect", MagicMock())
     def test_profile_idc_browser_required_fields_only(self):
         self.config.credentials = self.config.credentials.replace(
-            method="iam_idc_browser",
-            credentials_provider="BrowserIdcAuthPlugin",
+            method="browser_identity_center",
             idc_region="us-east-1",
             issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
             host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
@@ -633,28 +631,16 @@ def test_profile_idc_browser_required_fields_only(self):
             timeout=None,
             port=5439,
             **DEFAULT_SSL_CONFIG,
+            credentials_provider="BrowserIdcAuthPlugin",
             idp_response_timeout=60,
             idc_client_display_name="Amazon Redshift driver",
-            credentials_provider="BrowserIdcAuthPlugin",
             idc_region="us-east-1",
             issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
         )
 
-    def test_invalid_plugin_for_idc_browser_auth_method(self):
-        self.config.credentials = self.config.credentials.replace(
-            method="iam_idc_browser",
-            credentials_provider="IdpTokenAuthPlugin",
-        )
-        with self.assertRaises(FailedToConnectError) as context:
-            connection = self.adapter.acquire_connection("dummy")
-            connection.handle
-
-        assert "BrowserIdcAuthPlugin" in context.exception.msg
-
     def test_invalid_adapter_missing_fields(self):
         self.config.credentials = self.config.credentials.replace(
-            method="iam_idc_browser",
-            credentials_provider="BrowserIdcAuthPlugin",
+            method="browser_identity_center",
             idc_client_display_name="my display",
         )
         with self.assertRaises(FailedToConnectError) as context:
@@ -662,6 +648,6 @@ def test_invalid_adapter_missing_fields(self):
             connection.handle
 
         assert (
-            "'idc_region', 'issuer_url' field(s) are required for 'iam_idc_browser' credentials method"
+            "'idc_region', 'issuer_url' field(s) are required for 'browser_identity_center' credentials method"
             in context.exception.msg
         )

From aa0d07fe44e110e134eff73b46a71d5475f30d70 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 14:33:34 -0800
Subject: [PATCH 06/13] Add changelog.

---
 .changes/unreleased/Features-20241122-143326.yaml | 6 ++++++
 1 file changed, 6 insertions(+)
 create mode 100644 .changes/unreleased/Features-20241122-143326.yaml

diff --git a/.changes/unreleased/Features-20241122-143326.yaml b/.changes/unreleased/Features-20241122-143326.yaml
new file mode 100644
index 000000000..a4b8a7089
--- /dev/null
+++ b/.changes/unreleased/Features-20241122-143326.yaml
@@ -0,0 +1,6 @@
+kind: Features
+body: Add browser identity center authentication method.
+time: 2024-11-22T14:33:26.549878-08:00
+custom:
+    Author: versusfacit
+    Issue: "898"

From 6b69f2a7e4eda70d2a9647338c1bff002d13b9c2 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 14:49:44 -0800
Subject: [PATCH 07/13] Handle optional timeout field correctly

---
 dbt/adapters/redshift/connections.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 495bedf49..eb5a62cf1 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -148,7 +148,7 @@ class RedshiftCredentials(Credentials):
     issuer_url: Optional[str] = None
     listen_port: int = 7890
     idc_client_display_name: Optional[str] = "Amazon Redshift driver"
-    idp_response_timeout: int = 60
+    idp_response_timeout: Optional[int] = None
 
     _ALIASES = {"dbname": "database", "pass": "password"}
 
@@ -294,12 +294,16 @@ def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
 
         __assert_required_fields("browser_identity_center", ("method", "idc_region", "issuer_url"))
 
+        idp_timeout: int = (
+            timeout if (timeout := credentials.idp_response_timeout) or timeout == 0 else 60
+        )
+
         idc_kwargs: Dict[str, Any] = {
             "credentials_provider": "BrowserIdcAuthPlugin",
             "issuer_url": credentials.issuer_url,
             "idc_region": credentials.idc_region,
             "idc_client_display_name": credentials.idc_client_display_name,
-            "idp_response_timeout": credentials.idp_response_timeout,
+            "idp_response_timeout": idp_timeout,
         }
 
         return __iam_kwargs(credentials) | idc_kwargs

From 164a7640b9c2330b86464554fce9bbc8d1da5a65 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 14:56:43 -0800
Subject: [PATCH 08/13] revert undesired change.

---
 tests/unit/test_auth_method.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/unit/test_auth_method.py b/tests/unit/test_auth_method.py
index 64b98457c..2ba343664 100644
--- a/tests/unit/test_auth_method.py
+++ b/tests/unit/test_auth_method.py
@@ -79,7 +79,7 @@ def test_missing_region_failure(self):
             connection = self.adapter.acquire_connection("dummy")
             connection.handle
             redshift_connector.connect.assert_called_once_with(
-                iam=False,
+                iam=True,
                 host="doesnotexist.1233_no_region",
                 database="redshift",
                 cluster_identifier=None,

From a662254ce9264be692af976491756760ab2a21b1 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 14:57:50 -0800
Subject: [PATCH 09/13] Remove the authenticator param

---
 dbt/adapters/redshift/connections.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index eb5a62cf1..8b41ed738 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -143,7 +143,6 @@ class RedshiftCredentials(Credentials):
     #
 
     # browser
-    authenticator: Optional[str] = None
     idc_region: Optional[str] = None
     issuer_url: Optional[str] = None
     listen_port: int = 7890

From 4e4ec15be86c916197886c724b7b2eac297caae4 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 15:12:43 -0800
Subject: [PATCH 10/13] Handle other parms with default values better.

---
 dbt/adapters/redshift/connections.py | 14 ++++++++++++--
 tests/unit/test_auth_method.py       | 22 ++++++++++++++++++++++
 2 files changed, 34 insertions(+), 2 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 8b41ed738..18a484bdc 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -145,7 +145,7 @@ class RedshiftCredentials(Credentials):
     # browser
     idc_region: Optional[str] = None
     issuer_url: Optional[str] = None
-    listen_port: int = 7890
+    idp_listen_port: Optional[int] = 7890
     idc_client_display_name: Optional[str] = "Amazon Redshift driver"
     idp_response_timeout: Optional[int] = None
 
@@ -291,15 +291,25 @@ def __iam_role_kwargs(credentials) -> Dict[str, Any]:
     def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
         logger.debug("Connecting to Redshift with '{credentials.method}' credentials method")
 
+        __IDP_TIMEOUT: int = 60
+        __LISTEN_PORT_DEFAULT: int = 7890
+
         __assert_required_fields("browser_identity_center", ("method", "idc_region", "issuer_url"))
 
         idp_timeout: int = (
-            timeout if (timeout := credentials.idp_response_timeout) or timeout == 0 else 60
+            timeout
+            if (timeout := credentials.idp_response_timeout) or timeout == 0
+            else __IDP_TIMEOUT
+        )
+
+        idp_listen_port: int = (
+            port if (port := credentials.idp_listen_port) else __LISTEN_PORT_DEFAULT
         )
 
         idc_kwargs: Dict[str, Any] = {
             "credentials_provider": "BrowserIdcAuthPlugin",
             "issuer_url": credentials.issuer_url,
+            "listen_port": idp_listen_port,
             "idc_region": credentials.idc_region,
             "idc_client_display_name": credentials.idc_client_display_name,
             "idp_response_timeout": idp_timeout,
diff --git a/tests/unit/test_auth_method.py b/tests/unit/test_auth_method.py
index 2ba343664..16d13268f 100644
--- a/tests/unit/test_auth_method.py
+++ b/tests/unit/test_auth_method.py
@@ -585,6 +585,7 @@ def test_profile_idc_browser_all_fields(self):
             idc_client_display_name="display name",
             idp_response_timeout=0,
             host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+            idp_listen_port=1111,
         )
         connection = self.adapter.acquire_connection("dummy")
         connection.handle
@@ -606,6 +607,7 @@ def test_profile_idc_browser_all_fields(self):
             credentials_provider="BrowserIdcAuthPlugin",
             idc_region="us-east-1",
             issuer_url="https://identitycenter.amazonaws.com/ssoins-randomchars",
+            listen_port=1111,
         )
 
     @mock.patch("redshift_connector.connect", MagicMock())
@@ -632,6 +634,7 @@ def test_profile_idc_browser_required_fields_only(self):
             port=5439,
             **DEFAULT_SSL_CONFIG,
             credentials_provider="BrowserIdcAuthPlugin",
+            listen_port=7890,
             idp_response_timeout=60,
             idc_client_display_name="Amazon Redshift driver",
             idc_region="us-east-1",
@@ -641,11 +644,30 @@ def test_profile_idc_browser_required_fields_only(self):
     def test_invalid_adapter_missing_fields(self):
         self.config.credentials = self.config.credentials.replace(
             method="browser_identity_center",
+            idp_listen_port=1111,
             idc_client_display_name="my display",
         )
         with self.assertRaises(FailedToConnectError) as context:
             connection = self.adapter.acquire_connection("dummy")
             connection.handle
+            redshift_connector.connect.assert_called_once_with(
+                iam=False,
+                host="doesnotexist.1233.us-east-2.redshift-serverless.amazonaws.com",
+                database="redshift",
+                cluster_identifier=None,
+                region=None,
+                auto_create=False,
+                db_groups=[],
+                password="",
+                user="",
+                timeout=None,
+                port=5439,
+                **DEFAULT_SSL_CONFIG,
+                credentials_provider="BrowserIdcAuthPlugin",
+                listen_port=1111,
+                idp_response_timeout=60,
+                idc_client_display_name="my display",
+            )
 
         assert (
             "'idc_region', 'issuer_url' field(s) are required for 'browser_identity_center' credentials method"

From 0b93c078d2fdc894e64f0afc3affeb276bd31a1c Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 15:30:37 -0800
Subject: [PATCH 11/13] Code review comments.

---
 dbt/adapters/redshift/connections.py | 18 ++++++++++++------
 1 file changed, 12 insertions(+), 6 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 18a484bdc..068c8cc00 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -49,9 +49,13 @@ class RedshiftConnectionMethod(StrEnum):
     IAM_IDENTITY_CENTER_BROWSER = "browser_identity_center"
 
     @staticmethod
-    def uses_identity_center(method: str):
+    def uses_identity_center(method: str) -> bool:
         return method.endswith("identity_center")
 
+    @staticmethod
+    def is_iam(method: str) -> bool:
+        return not RedshiftConnectionMethod.uses_identity_center(method)
+
 
 class UserSSLMode(StrEnum):
     disable = "disable"
@@ -190,7 +194,7 @@ def get_connection_method(
     #
     # Helper Methods
     #
-    def __assert_required_fields(method_name: str, required_fields: Tuple[str, ...]):
+    def __validate_required_fields(method_name: str, required_fields: Tuple[str, ...]):
         missing_fields: List[str] = [
             field for field in required_fields if getattr(credentials, field, None) is None
         ]
@@ -218,7 +222,7 @@ def __base_kwargs(credentials) -> Dict[str, Any]:
     def __iam_kwargs(credentials) -> Dict[str, Any]:
 
         # iam True except for identity center methods
-        iam: bool = not RedshiftConnectionMethod.uses_identity_center(credentials.method)
+        iam: bool = RedshiftConnectionMethod.is_iam(credentials.method)
 
         cluster_identifier: Optional[str]
         if "serverless" in credentials.host or RedshiftConnectionMethod.uses_identity_center(
@@ -246,7 +250,7 @@ def __iam_kwargs(credentials) -> Dict[str, Any]:
     def __database_kwargs(credentials) -> Dict[str, Any]:
         logger.debug("Connecting to Redshift with 'database' credentials method")
 
-        __assert_required_fields("database", ("user", "password"))
+        __validate_required_fields("database", ("user", "password"))
 
         db_credentials: Dict[str, Any] = {
             "user": credentials.user,
@@ -271,7 +275,7 @@ def __iam_user_kwargs(credentials) -> Dict[str, Any]:
         else:
             iam_credentials = {"profile": credentials.iam_profile}
 
-        __assert_required_fields("iam", ("user",))
+        __validate_required_fields("iam", ("user",))
         iam_credentials["db_user"] = credentials.user
 
         return __iam_kwargs(credentials) | iam_credentials
@@ -294,7 +298,9 @@ def __iam_idc_browser_kwargs(credentials) -> Dict[str, Any]:
         __IDP_TIMEOUT: int = 60
         __LISTEN_PORT_DEFAULT: int = 7890
 
-        __assert_required_fields("browser_identity_center", ("method", "idc_region", "issuer_url"))
+        __validate_required_fields(
+            "browser_identity_center", ("method", "idc_region", "issuer_url")
+        )
 
         idp_timeout: int = (
             timeout

From dc237371b69fd80eadb33dad43b3503b270609cf Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 16:08:23 -0800
Subject: [PATCH 12/13] Pin connector from above.

---
 setup.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/setup.py b/setup.py
index 1ef723c47..fb1530524 100644
--- a/setup.py
+++ b/setup.py
@@ -53,7 +53,7 @@ def _plugin_version() -> str:
         "dbt-postgres>=1.8,<1.10",
         # dbt-redshift depends deeply on this package. it does not follow SemVer, therefore there have been breaking changes in previous patch releases
         # Pin to the patch or minor version, and bump in each new minor version of dbt-redshift.
-        "redshift-connector==2.1.3",
+        "redshift-connector>=2.1.3,<2.2",
         # add dbt-core to ensure backwards compatibility of installation, this is not a functional dependency
         "dbt-core>=1.8.0b3",
         # installed via dbt-core but referenced directly; don't pin to avoid version conflicts with dbt-core

From 653515d1497b23080573b9a1bb288b2222978780 Mon Sep 17 00:00:00 2001
From: VersusFacit <67295367+VersusFacit@users.noreply.github.com>
Date: Fri, 22 Nov 2024 16:16:26 -0800
Subject: [PATCH 13/13] Make the enum have class methods.

---
 dbt/adapters/redshift/connections.py | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/dbt/adapters/redshift/connections.py b/dbt/adapters/redshift/connections.py
index 068c8cc00..d93847634 100644
--- a/dbt/adapters/redshift/connections.py
+++ b/dbt/adapters/redshift/connections.py
@@ -48,13 +48,13 @@ class RedshiftConnectionMethod(StrEnum):
     IAM_ROLE = "iam_role"
     IAM_IDENTITY_CENTER_BROWSER = "browser_identity_center"
 
-    @staticmethod
-    def uses_identity_center(method: str) -> bool:
-        return method.endswith("identity_center")
+    @classmethod
+    def uses_identity_center(cls, method: str) -> bool:
+        return method in (cls.IAM_IDENTITY_CENTER_BROWSER,)
 
-    @staticmethod
-    def is_iam(method: str) -> bool:
-        return not RedshiftConnectionMethod.uses_identity_center(method)
+    @classmethod
+    def is_iam(cls, method: str) -> bool:
+        return not cls.uses_identity_center(method)
 
 
 class UserSSLMode(StrEnum):