feat(auth, iam): allowed use of getClusterCredentials V2 API when IdP Plugin authentication is used and group_federation=True. Previously an unsupported error was thrown.

Author: Brooke-white

redshift_connector/iam_helper.py

@@ -59,6 +59,7 @@ def can_support_v2(provider_type: "IamHelper.IAMAuthenticationType") -> bool:
                     IamHelper.IAMAuthenticationType.PROFILE,
                     IamHelper.IAMAuthenticationType.IAM_KEYS,
                     IamHelper.IAMAuthenticationType.IAM_KEYS_WITH_SESSION,
+                    IamHelper.IAMAuthenticationType.PLUGIN,
                 )
             ) and IdpAuthHelper.get_pkg_version("boto3") >= Version("1.24.5")
 
@@ -72,6 +73,11 @@ def get_cluster_credentials_api_type(
         Returns an enum representing the Python SDK method to use for getting temporary IAM credentials.
         """
         _logger.debug("Determining which Redshift API to use for retrieving temporary Redshift instance credentials")
+        FAILED_TO_USE_V2_API_ERROR_MSG: str = (
+            "Environment does not meet requirements to use {} API. "
+            "This could be due to the connection properties provided or the version of boto3 in use. "
+            "Please try updating the boto3 version or consider setting group_federation connection parameter to False."
+        )
 
         if not info._is_serverless:
             _logger.debug("Redshift provisioned")
@@ -82,7 +88,7 @@ def get_cluster_credentials_api_type(
                 _logger.debug("Provisioned cluster GetClusterCredentialsAPIType.IAM_V2")
                 return IamHelper.GetClusterCredentialsAPIType.IAM_V2
             else:
-                raise InterfaceError("Authentication with plugin is not supported for group federation")
+                raise InterfaceError(FAILED_TO_USE_V2_API_ERROR_MSG.format("GetClusterCredentials V2 API"))
         elif not info.group_federation:
             _logger.debug("Serverless cluster GetClusterCredentialsAPIType.SERVERLESS_V1")
             return IamHelper.GetClusterCredentialsAPIType.SERVERLESS_V1
@@ -93,7 +99,7 @@ def get_cluster_credentials_api_type(
                 _logger.debug("Serverless cluster GetClusterCredentialsAPIType.IAM_V2")
                 return IamHelper.GetClusterCredentialsAPIType.IAM_V2
         else:
-            raise InterfaceError("Authentication with plugin is not supported for group federation")
+            raise InterfaceError(FAILED_TO_USE_V2_API_ERROR_MSG.format("GetClusterCredentials V2 API"))
 
     @staticmethod
     def set_iam_properties(info: RedshiftProperty) -> RedshiftProperty:

test/integration/plugin/test_credentials_providers.py

@@ -207,3 +207,11 @@ def uses_db_groups_nominal(idp_arg, db_groups):
 
     with redshift_connector.connect(**idp_arg):
         pass
+
+
+@pytest.mark.parametrize("idp_arg", NON_BROWSER_IDP, indirect=True)
+def test_connect_with_group_federation(idp_arg):
+    idp_arg["group_federation"] = True
+
+    with redshift_connector.connect(**idp_arg):
+        pass

test/unit/test_iam_helper.py

@@ -781,12 +781,12 @@ def test_get_authentication_type_for_iam_with_plugin() -> None:
         (
             {"group_federation": True, "credentials_provider": "BrowserSamlCredentialsProvider"},
             IamHelper.IAMAuthenticationType.PLUGIN,
-            "Authentication with plugin is not supported for group federation",
+            IamHelper.GetClusterCredentialsAPIType.IAM_V2,
         ),
         (
             {"is_serverless": True, "group_federation": True, "credentials_provider": "BrowserSamlCredentialsProvider"},
             IamHelper.IAMAuthenticationType.PLUGIN,
-            "Authentication with plugin is not supported for group federation",
+            IamHelper.GetClusterCredentialsAPIType.IAM_V2,
         ),
         (
             {"is_serverless": True, "group_federation": True, "is_cname": True},