refactor: use shared code for IAM based authenticators

ibm_cloud_sdk_core/__init__.py

@@ -38,9 +38,9 @@
 
 from .base_service import BaseService
 from .detailed_response import DetailedResponse
-from .iam_token_manager import IAMTokenManager
-from .jwt_token_manager import JWTTokenManager
-from .cp4d_token_manager import CP4DTokenManager
+from .token_managers.iam_token_manager import IAMTokenManager
+from .token_managers.jwt_token_manager import JWTTokenManager
+from .token_managers.cp4d_token_manager import CP4DTokenManager
 from .api_exception import ApiException
 from .utils import datetime_to_string, string_to_datetime, read_external_sources
 from .utils import datetime_to_string_list, string_to_datetime_list

ibm_cloud_sdk_core/api_exception.py

@@ -13,8 +13,10 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from typing import Optional
+
 from http import HTTPStatus
+from typing import Optional
+
 from requests import Response
 
 

ibm_cloud_sdk_core/authenticators/authenticator.py

@@ -16,6 +16,7 @@
 
 from abc import ABC, abstractmethod
 
+
 class Authenticator(ABC):
     """This interface defines the common methods and constants associated with an Authenticator implementation."""
     @abstractmethod

ibm_cloud_sdk_core/authenticators/bearer_token_authenticator.py

@@ -18,6 +18,7 @@
 
 from .authenticator import Authenticator
 
+
 class BearerTokenAuthenticator(Authenticator):
     """The BearerTokenAuthenticator will add a user-supplied bearer token
     to requests.

ibm_cloud_sdk_core/authenticators/cp4d_authenticator.py

@@ -19,7 +19,7 @@
 from requests import Request
 
 from .authenticator import Authenticator
-from ..cp4d_token_manager import CP4DTokenManager
+from ..token_managers.cp4d_token_manager import CP4DTokenManager
 from ..utils import has_bad_first_or_last_char
 
 

ibm_cloud_sdk_core/authenticators/iam_authenticator.py

@@ -16,13 +16,12 @@
 
 from typing import Dict, Optional
 
-from requests import Request
-
-from .authenticator import Authenticator
-from ..iam_token_manager import IAMTokenManager
+from .iam_request_based_authenticator import IAMRequestBasedAuthenticator
+from ..token_managers.iam_token_manager import IAMTokenManager
 from ..utils import has_bad_first_or_last_char
 
-class IAMAuthenticator(Authenticator):
+
+class IAMAuthenticator(IAMRequestBasedAuthenticator):
     """The IAMAuthenticator utilizes an apikey, or client_id and client_secret pair to
     obtain a suitable bearer token, and adds it to requests.
 
@@ -81,81 +80,12 @@ def validate(self) -> None:
         Raises:
             ValueError: The apikey, client_id, and/or client_secret are not valid for IAM token requests.
         """
+        super().validate()
+
         if self.token_manager.apikey is None:
             raise ValueError('The apikey shouldn\'t be None.')
 
         if has_bad_first_or_last_char(self.token_manager.apikey):
             raise ValueError(
                 'The apikey shouldn\'t start or end with curly brackets or quotes. '
                 'Please remove any surrounding {, }, or \" characters.')
-
-        if (self.token_manager.client_id and
-                not self.token_manager.client_secret) or (
-                    not self.token_manager.client_id and
-                    self.token_manager.client_secret):
-            raise ValueError(
-                'Both client_id and client_secret should be initialized.')
-
-    def authenticate(self, req: Request) -> None:
-        """Adds IAM authentication information to the request.
-
-        The IAM bearer token will be added to the request's headers in the form:
-
-            Authorization: Bearer <bearer-token>
-
-        Args:
-            req: The request to add IAM authentication information too. Must contain a key to a dictionary
-            called headers.
-        """
-        headers = req.get('headers')
-        bearer_token = self.token_manager.get_token()
-        headers['Authorization'] = 'Bearer {0}'.format(bearer_token)
-
-    def set_client_id_and_secret(self, client_id: str, client_secret: str) -> None:
-        """Set the client_id and client_secret pair the token manager will use for IAM token requests.
-
-        Args:
-            client_id: The client id to be used in basic auth.
-            client_secret: The client secret to be used in basic auth.
-
-        Raises:
-            ValueError: The apikey, client_id, and/or client_secret are not valid for IAM token requests.
-        """
-        self.token_manager.set_client_id_and_secret(client_id, client_secret)
-        self.validate()
-
-    def set_disable_ssl_verification(self, status: bool = False) -> None:
-        """Set the flag that indicates whether verification of the server's SSL certificate should be
-        disabled or not. Defaults to False.
-
-        Keyword Arguments:
-            status: Headers to be sent with every IAM token request. Defaults to None.
-        """
-        self.token_manager.set_disable_ssl_verification(status)
-
-    def set_headers(self, headers: Dict[str, str]) -> None:
-        """Headers to be sent with every IAM token request.
-
-        Args:
-            headers: Headers to be sent with every IAM token request.
-        """
-        self.token_manager.set_headers(headers)
-
-    def set_proxies(self, proxies: Dict[str, str]) -> None:
-        """Sets the proxies the token manager will use to communicate with IAM on behalf of the host.
-
-        Args:
-            proxies: Dictionary for mapping request protocol to proxy URL.
-            proxies.http (optional): The proxy endpoint to use for HTTP requests.
-            proxies.https (optional): The proxy endpoint to use for HTTPS requests.
-        """
-        self.token_manager.set_proxies(proxies)
-
-    def set_scope(self, value: str) -> None:
-        """Sets the "scope" parameter to use when fetching the bearer token from the IAM token server.
-        This can be used to obtain an access token with a specific scope.
-
-        Args:
-            value: A space seperated string that makes up the scope parameter.
-        """
-        self.token_manager.set_scope(value)

ibm_cloud_sdk_core/authenticators/iam_request_based_authenticator.py

@@ -0,0 +1,113 @@
+# coding: utf-8
+
+# Copyright 2019 IBM All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Dict
+
+from requests import Request
+
+from .authenticator import Authenticator
+
+
+class IAMRequestBasedAuthenticator(Authenticator):
+    """The IAMRequestBasedAuthenticator class contains code that is common to all authenticators
+    that need to interact with the IAM tokens service to obtain an access token.
+
+    The bearer token will be sent as an Authorization header in the form:
+
+        Authorization: Bearer <bearer-token>
+
+    Attributes:
+        token_manager (TokenManager): Retrives and manages IAM tokens from the endpoint specified by the url.
+    """
+
+    def validate(self) -> None:
+        """Validates the client_id, and client_secret for IAM token requests.
+
+        Ensure both the client_id and client_secret are set if either of them are defined.
+
+        Raises:
+            ValueError: The client_id, and/or client_secret are not valid for IAM token requests.
+        """
+        if (self.token_manager.client_id and
+                not self.token_manager.client_secret) or (
+                    not self.token_manager.client_id and
+                    self.token_manager.client_secret):
+            raise ValueError(
+                'Both client_id and client_secret should be initialized.')
+
+    def authenticate(self, req: Request) -> None:
+        """Adds IAM authentication information to the request.
+
+        The IAM bearer token will be added to the request's headers in the form:
+
+            Authorization: Bearer <bearer-token>
+
+        Args:
+            req: The request to add IAM authentication information too. Must contain a key to a dictionary
+            called headers.
+        """
+        headers = req.get('headers')
+        bearer_token = self.token_manager.get_token()
+        headers['Authorization'] = 'Bearer {0}'.format(bearer_token)
+
+    def set_client_id_and_secret(self, client_id: str, client_secret: str) -> None:
+        """Set the client_id and client_secret pair the token manager will use for IAM token requests.
+
+        Args:
+            client_id: The client id to be used in basic auth.
+            client_secret: The client secret to be used in basic auth.
+
+        Raises:
+            ValueError: The apikey, client_id, and/or client_secret are not valid for IAM token requests.
+        """
+        self.token_manager.set_client_id_and_secret(client_id, client_secret)
+        self.validate()
+
+    def set_disable_ssl_verification(self, status: bool = False) -> None:
+        """Set the flag that indicates whether verification of the server's SSL certificate should be
+        disabled or not. Defaults to False.
+
+        Keyword Arguments:
+            status: Headers to be sent with every IAM token request. Defaults to None.
+        """
+        self.token_manager.set_disable_ssl_verification(status)
+
+    def set_headers(self, headers: Dict[str, str]) -> None:
+        """Headers to be sent with every IAM token request.
+
+        Args:
+            headers: Headers to be sent with every IAM token request.
+        """
+        self.token_manager.set_headers(headers)
+
+    def set_proxies(self, proxies: Dict[str, str]) -> None:
+        """Sets the proxies the token manager will use to communicate with IAM on behalf of the host.
+
+        Args:
+            proxies: Dictionary for mapping request protocol to proxy URL.
+            proxies.http (optional): The proxy endpoint to use for HTTP requests.
+            proxies.https (optional): The proxy endpoint to use for HTTPS requests.
+        """
+        self.token_manager.set_proxies(proxies)
+
+    def set_scope(self, value: str) -> None:
+        """Sets the "scope" parameter to use when fetching the bearer token from the IAM token server.
+        This can be used to obtain an access token with a specific scope.
+
+        Args:
+            value: A space seperated string that makes up the scope parameter.
+        """
+        self.token_manager.set_scope(value)

ibm_cloud_sdk_core/authenticators/no_auth_authenticator.py

@@ -16,6 +16,7 @@
 
 from .authenticator import Authenticator
 
+
 class NoAuthAuthenticator(Authenticator):
     """Performs no authentication."""
     authentication_type = 'noAuth'

ibm_cloud_sdk_core/base_service.py

@@ -14,26 +14,27 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-from http.cookiejar import CookieJar
+import gzip
 import json as json_import
-from os.path import basename
+import logging
 import platform
 import sys
-import gzip
+from http.cookiejar import CookieJar
+from os.path import basename
 from typing import Dict, List, Optional, Tuple, Union
+from urllib3.util.retry import Retry
 
 import requests
 from requests.adapters import HTTPAdapter
 from requests.structures import CaseInsensitiveDict
-from urllib3.util.retry import Retry
+
 from ibm_cloud_sdk_core.authenticators import Authenticator
-from .version import __version__
+from .api_exception import ApiException
+from .detailed_response import DetailedResponse
+from .token_managers.token_manager import TokenManager
 from .utils import (has_bad_first_or_last_char, remove_null_values,
                     cleanup_values, read_external_sources, strip_extra_slashes)
-from .detailed_response import DetailedResponse
-from .api_exception import ApiException
-from .token_manager import TokenManager
+from .version import __version__
 
 # Uncomment this to enable http debugging
 # import http.client as http_client

ibm_cloud_sdk_core/detailed_response.py

@@ -14,11 +14,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 from typing import Dict, Optional
 
-import json
 import requests
 
+
 class DetailedResponse:
     """Custom class for detailed response returned from APIs.
 

ibm_cloud_sdk_core/get_authenticator.py

@@ -18,6 +18,7 @@
                              CloudPakForDataAuthenticator, IAMAuthenticator, NoAuthAuthenticator)
 from .utils import read_external_sources
 
+
 def get_authenticator_from_environment(service_name: str) -> Authenticator:
     """Look for external configuration of authenticator.
 
@@ -38,6 +39,7 @@ def get_authenticator_from_environment(service_name: str) -> Authenticator:
         authenticator = __construct_authenticator(config)
     return authenticator
 
+
 def __construct_authenticator(config: dict) -> Authenticator:
     auth_type = config.get('AUTH_TYPE').lower() if config.get('AUTH_TYPE') else 'iam'
     authenticator = None

ibm_cloud_sdk_core/token_managers/__init__.py

@@ -0,0 +1,15 @@
+# coding: utf-8
+
+# Copyright 2021 IBM All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

ibm_cloud_sdk_core/cp4d_token_manager.py → ibm_cloud_sdk_core/token_managers/cp4d_token_manager.py

@@ -16,6 +16,7 @@
 
 import json
 from typing import Dict, Optional
+
 from .jwt_token_manager import JWTTokenManager