feat: expose refresh token in iam authenticator

Author: rmkeezer

ibm_cloud_sdk_core/iam_token_manager.py

@@ -17,6 +17,7 @@
 from typing import Dict, Optional
 from .jwt_token_manager import JWTTokenManager
 
+# pylint: disable=too-many-instance-attributes
 class IAMTokenManager(JWTTokenManager):
     """The IAMTokenManager takes an api key and performs the necessary interactions with
     the IAM token service to obtain and store a suitable bearer token. Additionally, the IAMTokenManager
@@ -61,6 +62,7 @@ class IAMTokenManager(JWTTokenManager):
     """
     DEFAULT_IAM_URL = 'https://iam.cloud.ibm.com/identity/token'
     CONTENT_TYPE = 'application/x-www-form-urlencoded'
+    OPERATION_PATH = "/identity/token"
     REQUEST_TOKEN_GRANT_TYPE = 'urn:ibm:params:oauth:grant-type:apikey'
     REQUEST_TOKEN_RESPONSE_TYPE = 'cloud_iam'
     TOKEN_NAME = 'access_token'
@@ -77,10 +79,15 @@ def __init__(self,
                  proxies: Optional[Dict[str, str]] = None,
                  scope: Optional[str] = None) -> None:
         self.apikey = apikey
-        self.url = url if url else self.DEFAULT_IAM_URL
+        self.url = url
+        if url == "":
+            self.url = self.DEFAULT_IAM_URL
+        elif url.endswith(self.OPERATION_PATH):
+            self.url = url[:-len(self.OPERATION_PATH)]
         self.client_id = client_id
         self.client_secret = client_secret
         self.headers = headers
+        self.refresh_token = None
         self.proxies = proxies
         self.scope = scope
         super().__init__(
@@ -118,7 +125,7 @@ def request_token(self) -> dict:
 
         response = self._request(
             method='POST',
-            url=self.url,
+            url=self.url + self.OPERATION_PATH,
             headers=headers,
             data=data,
             auth_tuple=auth_tuple,
@@ -146,6 +153,11 @@ def set_headers(self, headers: Dict[str, str]) -> None:
         else:
             raise TypeError('headers must be a dictionary')
 
+    def _save_token_info(self, token_response: dict) -> None:
+        super()._save_token_info(token_response)
+
+        self.refresh_token = token_response.get("refresh_token")
+
     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.
 

test/test_iam_token_manager.py

@@ -256,3 +256,89 @@ def test_request_token_auth_in_setter_scope():
     assert responses.calls[0].request.headers.get('Authorization') is None
     assert responses.calls[0].response.text == response
     assert 'scope=john+snow' in responses.calls[0].response.request.body
+
+@responses.activate
+def test_get_refresh_token():
+    iam_url = "https://iam.cloud.ibm.com/identity/token"
+    access_token_str = get_access_token()
+    response = """{
+        "access_token": "%s",
+        "token_type": "Bearer",
+        "expires_in": 3600,
+        "expiration": 1524167011,
+        "refresh_token": "jy4gl91BQ"
+    }""" % (access_token_str)
+    responses.add(responses.POST, url=iam_url, body=response, status=200)
+
+    token_manager = IAMTokenManager("iam_apikey")
+    token_manager.get_token()
+
+    assert len(responses.calls) == 2
+    assert token_manager.refresh_token == "jy4gl91BQ"
+
+# In order to test with a live IAM server, create file "iamtest.env" in the project root.
+# It should look like this:
+
+# IAMTEST1_AUTH_URL=<url> e.g. https://iam.cloud.ibm.com
+# IAMTEST1_AUTH_TYPE=iam
+# IAMTEST1_APIKEY=<apikey>
+
+# IAMTEST2_AUTH_URL=<url> e.g. https://iam.test.cloud.ibm.com/identity/token
+# IAMTEST2_AUTH_TYPE=iam
+# IAMTEST2_APIKEY=<apikey>
+# IAMTEST2_CLIENT_ID=<client id>
+# IAMTEST2_CLIENT_SECRET=<client secret>
+
+# Then uncomment the function below and run this command:
+# python3 -m pytest test -k "test_iam_live_token_server"
+
+# import os
+# from ibm_cloud_sdk_core import get_authenticator_from_environment
+
+# def test_iam_live_token_server():
+#     # Get two iam authenticators from the environment.
+#     # "iamtest1" uses username/password
+#     # "iamtest2" uses username/apikey
+#     os.environ['IBM_CREDENTIALS_FILE'] = "iamtest.env"
+
+
+#     # Test "iamtest1" service
+#     auth1 = get_authenticator_from_environment("iamtest1")
+#     assert auth1 is not None
+#     assert auth1.token_manager is not None
+#     assert auth1.token_manager.url is not None
+
+#     request = {'method': "GET"}
+#     request["url"] = ""
+#     request["headers"] = {}
+
+#     assert auth1.token_manager.refresh_token is None
+
+#     auth1.authenticate(request)
+
+#     assert auth1.token_manager.refresh_token is not None
+
+#     assert request.get("headers") is not None
+#     assert request["headers"].get("Authorization") is not None
+#     assert "Bearer " in request["headers"].get("Authorization")
+
+
+#     # Test "iamtest2" service
+#     auth2 = get_authenticator_from_environment("iamtest2")
+#     assert auth2 is not None
+#     assert auth2.token_manager is not None
+#     assert auth2.token_manager.url is not None
+
+#     request = {'method': "GET"}
+#     request["url"] = ""
+#     request["headers"] = {}
+
+#     assert auth2.token_manager.refresh_token is None
+
+#     auth2.authenticate(request)
+
+#     assert auth2.token_manager.refresh_token is not None
+
+#     assert request.get("headers") is not None
+#     assert request["headers"].get("Authorization") is not None
+#     assert "Bearer " in request["headers"].get("Authorization")