Fixed bug when using multiple cloud native authentication plugins for

connections.

Author: anthony-tuininga

doc/src/release_notes.rst

@@ -31,6 +31,10 @@ Common Changes
 #)  Added Session Token-based authentication support when using
     :ref:`OCI Cloud Native Authentication <cloudnativeauthoci>`
     (`issue 527 <https://github.com/oracle/python-oracledb/issues/527>`__).
+#)  Fixed bug when using multiple
+    :ref:`cloud native authentication <tokenauth>` plugins for connections.
+    Note that an invalid ``auth_type`` parameter will no longer raise an
+    exception but will simply be ignored.
 #)  Updated the `Jupyter notebook samples <https://github.com/oracle/
     python-oracledb/tree/main/samples/notebooks>`__ to cover recent
     python-oracledb features.

src/oracledb/plugins/azure_tokens.py

@@ -28,22 +28,14 @@
 # Methods that generates an OAuth2 access token using the MSAL SDK
 # -----------------------------------------------------------------------------
 
+import enum
+
 import msal
 import oracledb
 
 
-def generate_token(token_auth_config, refresh=False):
-    """
-    Generates an Azure access token based on provided credentials.
-    """
-    user_auth_type = token_auth_config.get("auth_type") or ""
-    auth_type = user_auth_type.lower()
-    if auth_type == "azureserviceprincipal":
-        return _service_principal_credentials(token_auth_config)
-    else:
-        raise ValueError(
-            f"Unrecognized auth_type authentication method: {user_auth_type}"
-        )
+class AuthType(str, enum.Enum):
+    AzureServicePrincipal = "AzureServicePrincipal".lower()
 
 
 def _service_principal_credentials(token_auth_config):
@@ -65,11 +57,30 @@ def _service_principal_credentials(token_auth_config):
         return auth_response["access_token"]
 
 
+def generate_token(token_auth_config, refresh=False):
+    """
+    Generates an Azure access token based on provided credentials.
+    """
+    auth_type = token_auth_config["auth_type"].lower()
+    if auth_type == AuthType.AzureServicePrincipal:
+        return _service_principal_credentials(token_auth_config)
+
+
+def has_azure_auth_type(extra_auth_params):
+    """
+    Validates that extra_auth_params contains a valid 'auth_type'
+    """
+    if extra_auth_params is None:
+        return False
+    auth_type = extra_auth_params.get("auth_type")
+    return auth_type is not None and auth_type.lower() in AuthType
+
+
 def azure_token_hook(params: oracledb.ConnectParams):
     """
     Azure-specific hook for generating a token.
     """
-    if params.extra_auth_params is not None:
+    if has_azure_auth_type(params.extra_auth_params):
 
         def token_callback(refresh):
             return generate_token(params.extra_auth_params, refresh)

src/oracledb/plugins/oci_tokens.py

@@ -190,8 +190,7 @@ def generate_token(token_auth_config, refresh=False):
     """
     Generates an OCI access token based on provided credentials.
     """
-    user_auth_type = token_auth_config.get("auth_type") or ""
-    auth_type = user_auth_type.lower()
+    auth_type = token_auth_config["auth_type"].lower()
     if auth_type == AuthType.ConfigFileAuthentication:
         return _config_file_based_authentication(token_auth_config)
     elif auth_type == AuthType.InstancePrincipal:
@@ -202,17 +201,23 @@ def generate_token(token_auth_config, refresh=False):
         return _security_token_simple_authentication(token_auth_config)
     elif auth_type == AuthType.SimpleAuthentication:
         return _simple_authentication(token_auth_config)
-    else:
-        raise ValueError(
-            f"Unrecognized auth_type authentication method {user_auth_type}"
-        )
+
+
+def has_oci_auth_type(extra_auth_params):
+    """
+    Validates that extra_auth_params contains a valid 'auth_type'
+    """
+    if extra_auth_params is None:
+        return False
+    auth_type = extra_auth_params.get("auth_type")
+    return auth_type is not None and auth_type.lower() in AuthType
 
 
 def oci_token_hook(params: oracledb.ConnectParams):
     """
     OCI-specific hook for generating a token.
     """
-    if params.extra_auth_params is not None:
+    if has_oci_auth_type(params.extra_auth_params):
 
         def token_callback(refresh):
             return generate_token(params.extra_auth_params, refresh)