feat(icp4d): Add support for icp4d url

Author: ehdsouza

README.md

@@ -20,6 +20,19 @@ or
 easy_install --upgrade ibm-cloud-sdk-core
 ```
 
+## Authentication Types
+There are several flavors of authentication supported in this package. To specify the intended authentication pattern to use, the user can pass in the parameter `authentication_type`. This parameter is optional, but it may become required in a future major release. The options for this parameter are `basic`, `iam`, and `icp4d`.
+
+### basic
+This indicates Basic Auth is to be used. Users will pass in a `username` and `password` and the SDK will generate a Basic Auth header to send with requests to the service.
+
+### iam
+This indicates that IAM token authentication is to be used. Users can pass in an `iam_apikey` or an `iam_access_token`. If an API key is used, the SDK will manage the token for the user. In either case, the SDK will generate a Bearer Auth header to send with requests to the service.
+
+### icp4d
+This indicates that the service is an instance of ICP4D, which has its own version of token authentication. Users can pass in a `username` and `password`, or an `icp4d_access_token`. If a username and password is given, the SDK will manage the token for the user.
+A `icp4d_url` is **required** for this type. In order to use an SDK-managed token with ICP4D authentication, this option **must** be passed in.
+
 ## Issues
 
 If you encounter an issue with this project, you are welcome to submit a [bug report](https://github.com/IBM/python-sdk-core/issues).

ibm_cloud_sdk_core/__init__.py

@@ -17,6 +17,6 @@
 from .detailed_response import DetailedResponse
 from .iam_token_manager import IAMTokenManager
 from .jwt_token_manager import JWTTokenManager
-from .icp_token_manager import ICPTokenManager
+from .icp4d_token_manager import ICP4DTokenManager
 from .api_exception import ApiException
 from .utils import datetime_to_string, string_to_datetime

ibm_cloud_sdk_core/base_service.py

@@ -24,7 +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 .icp4d_token_manager import ICP4DTokenManager
 from .detailed_response import DetailedResponse
 from .api_exception import ApiException
 
@@ -46,20 +46,21 @@ class BaseService(object):
     ICP_PREFIX = 'icp-'
     APIKEY = 'apikey'
     IAM_ACCESS_TOKEN = 'iam_access_token'
-    ICP_ACCESS_TOKEN = 'icp_access_token'
+    ICP4D_ACCESS_TOKEN = 'icp4d_access_token'
     URL = 'url'
     USERNAME = 'username'
     PASSWORD = 'password'
     IAM_APIKEY = 'iam_apikey'
     IAM_URL = 'iam_url'
+    ICP4D_URL = 'icp4d_url'
     APIKEY_DEPRECATION_MESSAGE = 'Authenticating with apikey is deprecated. Move to using Identity and Access Management (IAM) authentication.'
     DEFAULT_CREDENTIALS_FILE_NAME = 'ibm-credentials.env'
     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_url=None,
                  iam_access_token=None, iam_client_id=None, iam_client_secret=None,
-                 display_name=None, icp_access_token=None, authentication_type=None):
+                 display_name=None, icp4d_access_token=None, icp4d_url=None, authentication_type=None):
         """
         It loads credentials with the following preference:
         1) Credentials explicitly set in the request
@@ -78,7 +79,8 @@ def __init__(self, vcap_services_name, url, username=None, password=None,
         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.icp4d_access_token = icp4d_access_token
+        self.icp4d_url = icp4d_url
         self.token_manager = None
         self.default_headers = None
         self.verify = None # Indicates whether to ignore verifying the SSL certification
@@ -104,11 +106,13 @@ def __init__(self, vcap_services_name, url, username=None, password=None,
             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_icp4d(self.authentication_type, self.icp4d_access_token):
+            if self.icp4d_url is None:
+                raise Exception('The icp4d_url is mandatory for ICP4D.')
+            self.token_manager = ICP4DTokenManager(self.icp4d_url,
+                                                   self.username,
+                                                   self.password,
+                                                   self.icp4d_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
@@ -138,8 +142,10 @@ def __init__(self, vcap_services_name, url, username=None, password=None,
                     self.set_iam_apikey(self.vcap_service_credentials.get(self.IAM_APIKEY))
                 if self.IAM_ACCESS_TOKEN in self.vcap_service_credentials:
                     self.set_iam_access_token(self.vcap_service_credentials.get(self.IAM_ACCESS_TOKEN))
-                if self.ICP_ACCESS_TOKEN in self.vcap_service_credentials:
-                    self.set_icp_access_token(self.vcap_service_credentials.get(self.ICP_ACCESS_TOKEN))
+                if self.ICP4D_URL in self.vcap_service_credentials:
+                    self.icp4d_url = self.vcap_service_credentials.get(self.ICP4D_URL)
+                if self.ICP4D_ACCESS_TOKEN in self.vcap_service_credentials:
+                    self.set_icp4d_access_token(self.vcap_service_credentials.get(self.ICP4D_ACCESS_TOKEN))
 
         if (self.username is None or self.password is None) and self.token_manager is None:
             raise ValueError(
@@ -202,8 +208,8 @@ def _load_from_vcap_services(self, service_name):
     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 _is_for_icp4d(self, authentication_type, icp4d_access_token=None):
+        return authentication_type == 'icp4d' or icp4d_access_token
 
     def _has_basic_credentials(self, username, password):
         return username and password and not self._uses_basic_for_iam(username, password)
@@ -259,12 +265,14 @@ def set_iam_access_token(self, iam_access_token):
         self.iam_access_token = iam_access_token
         self.jar = CookieJar()
 
-    def set_icp_access_token(self, icp_access_token):
+    def set_icp4d_access_token(self, icp4d_access_token):
         if self.token_manager:
-            self.token_manager.set_access_token(icp_access_token)
+            self.token_manager.set_access_token(icp4d_access_token)
         else:
-            self.token_manager = ICPTokenManager(self.url, access_token=icp_access_token)
-        self.icp_access_token = icp_access_token
+            if self.icp4d_url is None:
+                raise Exception('The icp4d_url is mandatory for ICP4D.')
+            self.token_manager = ICP4DTokenManager(self.icp4d_url, access_token=icp4d_access_token)
+        self.icp4d_access_token = icp4d_access_token
         self.jar = CookieJar()
 
     def set_iam_url(self, iam_url):

ibm_cloud_sdk_core/iam_token_manager.py

@@ -21,14 +21,15 @@ class IAMTokenManager(JWTTokenManager):
     CONTENT_TYPE = 'application/x-www-form-urlencoded'
     REQUEST_TOKEN_GRANT_TYPE = 'urn:ibm:params:oauth:grant-type:apikey'
     REQUEST_TOKEN_RESPONSE_TYPE = 'cloud_iam'
+    TOKEN_NAME = 'access_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.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
-        super(IAMTokenManager, self).__init__(self.iam_url, iam_access_token)
+        super(IAMTokenManager, self).__init__(self.iam_url, iam_access_token, self.TOKEN_NAME)
 
     def request_token(self):
         """

ibm_cloud_sdk_core/icp_token_manager.py renamed to ibm_cloud_sdk_core/icp4d_token_manager.py

@@ -16,12 +16,13 @@
 
 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'
+class ICP4DTokenManager(JWTTokenManager):
+    TOKEN_NAME = 'accessToken'
+    def __init__(self, icp4d_url, username=None, password=None, access_token=None):
+        url = icp4d_url + '/v1/preauth/validateAuth'
         self.username = username
         self.password = password
-        super(ICPTokenManager, self).__init__(url, access_token)
+        super(ICP4DTokenManager, self).__init__(url, access_token, self.TOKEN_NAME)
 
     def request_token(self):
         auth_tuple = (self.username, self.password)

ibm_cloud_sdk_core/jwt_token_manager.py

@@ -20,9 +20,7 @@
 from .api_exception import ApiException
 
 class JWTTokenManager(object):
-    TOKEN_NAME = 'access_token'
-
-    def __init__(self, url, access_token=None):
+    def __init__(self, url, access_token=None, token_name=None):
         """
         Parameters
         ----------
@@ -37,6 +35,7 @@ def __init__(self, url, access_token=None):
         self.time_to_live = None
         self.expire_time = None
         self.verify = None # to enable/ disable SSL verification
+        self.token_name = token_name
 
     def get_token(self):
         """
@@ -57,7 +56,7 @@ def get_token(self):
             token_response = self.request_token()
             self._save_token_info(token_response)
 
-        return self.token_info.get(self.TOKEN_NAME)
+        return self.token_info.get(self.token_name)
 
     def set_access_token(self, access_token):
         """
@@ -117,7 +116,7 @@ def _save_token_info(self, token_response):
         token_response : str
             Response from token service
         """
-        access_token = token_response.get(self.TOKEN_NAME)
+        access_token = token_response.get(self.token_name)
 
         # The time of expiration is found by decoding the JWT access token
         decoded_response = jwt.decode(access_token, verify=False)

test/test_base_service.py

@@ -7,7 +7,7 @@
 import jwt
 from ibm_cloud_sdk_core import BaseService
 from ibm_cloud_sdk_core import ApiException
-from ibm_cloud_sdk_core import ICPTokenManager
+from ibm_cloud_sdk_core import ICP4DTokenManager
 
 class AnyServiceV1(BaseService):
     default_url = 'https://gateway.watsonplatform.net/test/api'
@@ -17,7 +17,8 @@ def __init__(self, version, url=default_url, username=None, password=None,
                  iam_apikey=None,
                  iam_access_token=None,
                  iam_url=None,
-                 icp_access_token=None,
+                 icp4d_access_token=None,
+                 icp4d_url=None,
                  authentication_type=None
                 ):
         BaseService.__init__(
@@ -32,7 +33,8 @@ def __init__(self, version, url=default_url, username=None, password=None,
             iam_access_token=iam_access_token,
             iam_url=iam_url,
             display_name='Watson',
-            icp_access_token=icp_access_token,
+            icp4d_access_token=icp4d_access_token,
+            icp4d_url=icp4d_url,
             authentication_type=authentication_type)
         self.version = version
 
@@ -223,29 +225,29 @@ def test_for_icp():
     assert service6.password is not None
 
 def test_for_icp4d():
-    service1 = AnyServiceV1('2017-07-07', username='hello', password='world', url='service_url', authentication_type='icp4d')
+    service1 = AnyServiceV1('2017-07-07', username='hello', password='world', icp4d_url='service_url', authentication_type='icp4d')
     assert service1.token_manager is not None
     assert service1.iam_apikey is None
     assert service1.username is not None
     assert service1.password is not None
-    assert service1.url == 'service_url'
-    assert isinstance(service1.token_manager, ICPTokenManager)
+    assert service1.icp4d_url == 'service_url'
+    assert isinstance(service1.token_manager, ICP4DTokenManager)
 
-    service2 = AnyServiceV1('2017-07-07', icp_access_token='icp_access_token')
+    service2 = AnyServiceV1('2017-07-07', icp4d_access_token='icp4d_access_token', icp4d_url='service_url')
     assert service2.token_manager is not None
     assert service2.iam_apikey is None
     assert service2.username is None
     assert service2.password is None
-    assert isinstance(service2.token_manager, ICPTokenManager)
+    assert isinstance(service2.token_manager, ICP4DTokenManager)
 
-    service3 = AnyServiceV1('2019-06-03', username='hello', password='world')
+    service3 = AnyServiceV1('2019-06-03', username='hello', password='world', icp4d_url='icp4d_url')
     assert service3.username is not None
     assert service3.password is not None
     assert service3.token_manager is None
 
-    service3.set_icp_access_token('icp_access_token')
+    service3.set_icp4d_access_token('icp4d_access_token')
     assert service3.token_manager is not None
-    assert isinstance(service3.token_manager, ICPTokenManager)
+    assert isinstance(service3.token_manager, ICP4DTokenManager)
 
 def test_disable_SSL_verification():
     service1 = AnyServiceV1('2017-07-07', username='apikey', password='icp-xxxx', url='service_url')
@@ -254,7 +256,7 @@ def test_disable_SSL_verification():
     service1.disable_SSL_verification()
     assert service1.verify is False
 
-    service2 = AnyServiceV1('2017-07-07', username='hello', password='world', authentication_type='icp4d')
+    service2 = AnyServiceV1('2017-07-07', username='hello', password='world', authentication_type='icp4d', icp4d_url='icp4d_url')
     assert service2.verify is None
     service2.disable_SSL_verification()
     assert service2.token_manager.verify is not None
@@ -354,11 +356,12 @@ def test_vcap_credentials():
 
     vcap_services = '{"test":[{"credentials":{ \
         "url":"https://gateway.watsonplatform.net/compare-comply/api",\
-        "icp_access_token":"bogus icp_access_token"}}]}'
+        "icp4d_url":"https://test/",\
+        "icp4d_access_token":"bogus icp4d_access_token"}}]}'
     os.environ['VCAP_SERVICES'] = vcap_services
     service = AnyServiceV1('2018-11-20')
     assert service.token_manager is not None
-    assert service.token_manager.user_access_token == 'bogus icp_access_token'
+    assert service.token_manager.user_access_token == 'bogus icp4d_access_token'
     del os.environ['VCAP_SERVICES']
 
 @responses.activate

test/test_icp_token_manager.py renamed to test/test_icp4d_token_manager.py

@@ -1,7 +1,7 @@
 import responses
 import jwt
 import json
-from ibm_cloud_sdk_core import ICPTokenManager
+from ibm_cloud_sdk_core import ICP4DTokenManager
 
 @responses.activate
 def test_request_token():
@@ -25,15 +25,15 @@ def test_request_token():
                               'secret', algorithm='HS256',
                               headers={'kid': '230498151c214b788dd97f22b85410a5'}).decode('utf-8')
     response = {
-        "access_token": access_token,
+        "accessToken": access_token,
         "token_type": "Bearer",
         "expires_in": 3600,
         "expiration": 1524167011,
         "refresh_token": "jy4gl91BQ"
     }
     responses.add(responses.GET, url + '/v1/preauth/validateAuth', body=json.dumps(response), status=200)
 
-    token_manager = ICPTokenManager(url, "username", "password")
+    token_manager = ICP4DTokenManager(url, "username", "password")
     token_manager.disable_SSL_verification(True)
     token = token_manager.get_token()
 

test/test_jwt_token_manager.py

@@ -8,7 +8,7 @@ def __init__(self, url=None, access_token=None):
         self.url = url
         self.access_token = access_token
         self.request_count = 0 # just for tests to see how  many times request was called
-        super(JWTTokenManagerMockImpl, self).__init__(url, access_token)
+        super(JWTTokenManagerMockImpl, self).__init__(url, access_token, 'access_token')
 
     def request_token(self):
         self.request_count += 1