feat(icp4d): Add support for icp4d

Author: ehdsouza

ibm_cloud_sdk_core/__init__.py

@@ -16,5 +16,7 @@
 from .base_service import BaseService
 from .detailed_response import DetailedResponse
 from .iam_token_manager import IAMTokenManager
+from .jwt_token_manager import JWTTokenManager
+from .icp_token_manager import ICPTokenManager
 from .api_exception import ApiException
 from .utils import datetime_to_string, string_to_datetime

ibm_cloud_sdk_core/base_service.py

@@ -24,6 +24,7 @@
 from .version import __version__
 from .utils import has_bad_first_or_last_char, remove_null_values, cleanup_values
 from .iam_token_manager import IAMTokenManager
+from .icp_token_manager import ICPTokenManager
 from .detailed_response import DetailedResponse
 from .api_exception import ApiException
 
@@ -55,9 +56,9 @@ class BaseService(object):
     SDK_NAME = 'ibm-python-sdk-core'
 
     def __init__(self, vcap_services_name, url, username=None, password=None,
-                 use_vcap_services=True, api_key=None,
-                 iam_apikey=None, iam_access_token=None, iam_url=None, iam_client_id=None, iam_client_secret=None,
-                 display_name=None):
+                 use_vcap_services=True, api_key=None, iam_apikey=None, iam_url=None,
+                 iam_access_token=None, iam_client_id=None, iam_client_secret=None,
+                 display_name=None, icp_access_token=None, authentication_type=None):
         """
         It loads credentials with the following preference:
         1) Credentials explicitly set in the request
@@ -66,41 +67,50 @@ def __init__(self, vcap_services_name, url, username=None, password=None,
         """
         self.url = url
         self.http_config = {}
+        self.authentication_type = authentication_type.lower() if authentication_type else None
         self.jar = None
-        self.api_key = None
-        self.username = None
-        self.password = None
-        self.default_headers = None
-        self.iam_apikey = None
-        self.iam_access_token = None
-        self.iam_url = None
-        self.iam_client_id = None
-        self.iam_client_secret = None
+        self.api_key = api_key
+        self.username = username
+        self.password = password
+        self.iam_apikey = iam_apikey
+        self.iam_access_token = iam_access_token
+        self.iam_url = iam_url
+        self.iam_client_id = iam_client_id
+        self.iam_client_secret = iam_client_secret
+        self.icp_access_token = icp_access_token
         self.token_manager = None
+        self.default_headers = None
         self.verify = None # Indicates whether to ignore verifying the SSL certification
 
-        if has_bad_first_or_last_char(self.url):
-            raise ValueError('The URL shouldn\'t start or end with curly brackets or quotes. '
-                             'Be sure to remove any {} and \" characters surrounding your URL')
+        self._check_credentials()
 
         self.set_user_agent_header(self.build_user_agent())
 
         # 1. Credentials are passed in constructor
-        if api_key is not None:
-            if api_key.startswith(self.ICP_PREFIX):
-                self.set_username_and_password(self.APIKEY, api_key)
-            else:
-                self.set_token_manager(api_key, iam_access_token, iam_url, iam_client_id, iam_client_secret)
-        elif username is not None and password is not None:
-            if username is self.APIKEY and not password.startswith(self.ICP_PREFIX):
-                self.set_token_manager(password, iam_access_token, iam_url, iam_client_id, iam_client_secret)
-            else:
-                self.set_username_and_password(username, password)
-        elif iam_access_token is not None or iam_apikey is not None:
-            if iam_apikey and iam_apikey.startswith(self.ICP_PREFIX):
-                self.set_username_and_password(self.APIKEY, iam_apikey)
-            else:
-                self.set_token_manager(iam_apikey, iam_access_token, iam_url, iam_client_id, iam_client_secret)
+        if self.authentication_type == 'iam' or self._has_iam_credentials(self.iam_apikey, self.iam_access_token) or self._has_iam_credentials(self.api_key, self.iam_access_token):
+            self.token_manager = IAMTokenManager(self.iam_apikey or self.api_key or self.password,
+                                                 self.iam_access_token,
+                                                 self.iam_url,
+                                                 self.iam_client_id,
+                                                 self.iam_client_secret)
+            self.iam_apikey = self.iam_apikey or self.api_key or self.password
+        elif self._uses_basic_for_iam(self.username, self.password):
+            self.token_manager = IAMTokenManager(self.password,
+                                                 self.iam_access_token,
+                                                 self.iam_url,
+                                                 self.iam_client_id,
+                                                 self.iam_client_secret)
+            self.iam_apikey = self.password
+            self.username = None
+            self.password = None
+        elif self._is_for_icp4d(self.authentication_type, self.icp_access_token):
+            self.token_manager = ICPTokenManager(self.url,
+                                                 self.username,
+                                                 self.password,
+                                                 self.icp_access_token)
+        elif self._is_for_icp(self.api_key) or self._is_for_icp(self.iam_apikey):
+            self.username = self.APIKEY
+            self.password = self.api_key or self.iam_apikey
 
         # 2. Credentials from credential file
         if display_name and not self.username and not self.token_manager:
@@ -183,6 +193,41 @@ def _load_from_vcap_services(self, service_name):
         else:
             return None
 
+    def _is_for_icp(self, credential=None):
+        return credential and credential.startswith(self.ICP_PREFIX)
+
+    def _is_for_icp4d(self, authentication_type, icp_access_token=None):
+        return authentication_type == 'icp4d' or icp_access_token
+
+    def _has_basic_credentials(self, username, password):
+        return username and password and not self._uses_basic_for_iam(username, password)
+
+    def _has_iam_credentials(self, iam_apikey, iam_access_token):
+        return (iam_apikey or iam_access_token) and not self._is_for_icp(iam_apikey)
+
+    def _uses_basic_for_iam(self, username, password):
+        """
+        Returns true if the user provides basic auth creds with the intention
+        of using IAM auth
+        """
+        return username and password and username == self.APIKEY and not self._is_for_icp(password)
+
+    def _has_bad_first_or_last_char(self, str):
+        return str is not None and (str.startswith('{') or str.startswith('"') or str.endswith('}') or str.endswith('"'))
+
+    def _check_credentials(self):
+        credentials_to_check = {
+            'URL': self.url,
+            'username': self.username,
+            'password': self.password,
+            'credentials': self.iam_apikey
+        }
+
+        for key in credentials_to_check:
+            if self._has_bad_first_or_last_char(credentials_to_check.get(key)):
+                raise ValueError('The ' + key + ' shouldn\'t start or end with curly brackets or quotes. '
+                                 'Be sure to remove any {} and \" characters surrounding your ' + key)
+
     def disable_SSL_verification(self):
         self.verify = False
 

ibm_cloud_sdk_core/iam_token_manager.py

@@ -14,69 +14,23 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import requests
-import time
-from .api_exception import ApiException
+from .jwt_token_manager import JWTTokenManager
 
-class IAMTokenManager(object):
+class IAMTokenManager(JWTTokenManager):
     DEFAULT_IAM_URL = 'https://iam.cloud.ibm.com/identity/token'
     CONTENT_TYPE = 'application/x-www-form-urlencoded'
     REQUEST_TOKEN_GRANT_TYPE = 'urn:ibm:params:oauth:grant-type:apikey'
     REQUEST_TOKEN_RESPONSE_TYPE = 'cloud_iam'
-    REFRESH_TOKEN_GRANT_TYPE = 'refresh_token'
 
     def __init__(self, iam_apikey=None, iam_access_token=None, iam_url=None,
                  iam_client_id=None, iam_client_secret=None):
         self.iam_apikey = iam_apikey
-        self.user_access_token = iam_access_token
         self.iam_url = iam_url if iam_url else self.DEFAULT_IAM_URL
         self.iam_client_id = iam_client_id
         self.iam_client_secret = iam_client_secret
-        self.token_info = {
-            'access_token': None,
-            'refresh_token': None,
-            'token_type': None,
-            'expires_in': None,
-            'expiration': None,
-        }
-
-    def request(self, method, url, headers=None, params=None, data=None, **kwargs):
-        auth_tuple = ('bx', 'bx')
-        if self.iam_client_id and self.iam_client_secret:
-            auth_tuple = (self.iam_client_id, self.iam_client_secret)
-        response = requests.request(method=method, url=url,
-                                    headers=headers, params=params,
-                                    data=data, auth=auth_tuple, **kwargs)
-        if 200 <= response.status_code <= 299:
-            return response.json()
-        else:
-            raise ApiException(response.status_code, http_response=response)
+        super(IAMTokenManager, self).__init__(self.iam_url, iam_access_token)
 
-    def get_token(self):
-        """
-        The source of the token is determined by the following logic:
-        1. If user provides their own managed access token, assume it is valid and send it
-        2. If this class is managing tokens and does not yet have one, make a request for one
-        3. If this class is managing tokens and the token has expired refresh it. In case the refresh token is expired, get a new one
-        If this class is managing tokens and has a valid token stored, send it
-        """
-        if self.user_access_token:
-            return self.user_access_token
-        elif not self.token_info.get('access_token'):
-            token_info = self._request_token()
-            self._save_token_info(token_info)
-            return self.token_info.get('access_token')
-        elif self._is_token_expired():
-            if self._is_refresh_token_expired():
-                token_info = self._request_token()
-            else:
-                token_info = self._refresh_token()
-            self._save_token_info(token_info)
-            return self.token_info.get('access_token')
-        else:
-            return self.token_info.get('access_token')
-
-    def _request_token(self):
+    def request_token(self):
         """
         Request an IAM token using an API key
         """
@@ -89,39 +43,22 @@ def _request_token(self):
             'apikey': self.iam_apikey,
             'response_type': self.REQUEST_TOKEN_RESPONSE_TYPE
         }
-        response = self.request(
-            method='POST',
-            url=self.iam_url,
-            headers=headers,
-            data=data)
-        return response
 
-    def _refresh_token(self):
-        """
-        Refresh an IAM token using a refresh token
-        """
-        headers = {
-            'Content-type': self.CONTENT_TYPE,
-            'Accept': 'application/json'
-        }
-        data = {
-            'grant_type': self.REFRESH_TOKEN_GRANT_TYPE,
-            'refresh_token': self.token_info.get('refresh_token')
-        }
-        response = self.request(
+        # Use bx:bx as default auth header creds
+        auth_tuple = ('bx', 'bx')
+
+        # If both the clientId and secret were specified by the user, then use them
+        if self.iam_client_id and self.iam_client_secret:
+            auth_tuple = (self.iam_client_id, self.iam_client_secret)
+
+        response = self._request(
             method='POST',
-            url=self.iam_url,
+            url=self.url,
             headers=headers,
-            data=data)
+            data=data,
+            auth_tuple=auth_tuple)
         return response
 
-    def set_access_token(self, iam_access_token):
-        """
-        Set a self-managed IAM access token.
-        The access token should be valid and not yet expired.
-        """
-        self.user_access_token = iam_access_token
-
     def set_iam_apikey(self, iam_apikey):
         """
         Set the IAM api key
@@ -145,40 +82,3 @@ def set_iam_authorization_info(self, iam_client_id, iam_client_secret):
         """
         self.iam_client_id = iam_client_id
         self.iam_client_secret = iam_client_secret
-
-    def _is_token_expired(self):
-        """
-        Check if currently stored token is expired.
-
-        Using a buffer to prevent the edge case of the
-        oken expiring before the request could be made.
-
-        The buffer will be a fraction of the total TTL. Using 80%.
-        """
-        fraction_of_ttl = 0.8
-        time_to_live = self.token_info.get('expires_in')
-        expire_time = self.token_info.get('expiration')
-        refresh_time = expire_time - (time_to_live * (1.0 - fraction_of_ttl))
-        current_time = int(time.time())
-        return refresh_time < current_time
-
-    def _is_refresh_token_expired(self):
-        """
-        Used as a fail-safe to prevent the condition of a refresh token expiring,
-        which could happen after around 30 days. This function will return true
-        if it has been at least 7 days and 1 hour since the last token was set
-        """
-        if self.token_info.get('expiration') is None:
-            return True
-
-        seven_days = 7 * 24 * 3600
-        current_time = int(time.time())
-        new_token_time = self.token_info.get('expiration') + seven_days
-        return new_token_time < current_time
-
-    def _save_token_info(self, token_info):
-        """
-        Save the response from the IAM service request to the object's state.
-        """
-        self.token_info = token_info
-        

ibm_cloud_sdk_core/icp_token_manager.py

@@ -0,0 +1,33 @@
+# 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 .jwt_token_manager import JWTTokenManager
+
+class ICPTokenManager(JWTTokenManager):
+    def __init__(self, url, username=None, password=None, access_token=None):
+        url = url + '/v1/preauth/validateAuth'
+        self.username = username
+        self.password = password
+        super(ICPTokenManager, self).__init__(url, access_token)
+
+    def request_token(self):
+        auth_tuple = (self.username, self.password)
+
+        response = self._request(
+            method='GET',
+            url=self.url,
+            auth_tuple=auth_tuple)
+        return response