Use IBM Cloud access token to call runtime API (#2102)

* wip use cloud access token

* Support staging & add reno

* Fallback to apikey

* Add integration test

* refresh token after expiration

* update docstring

* address comments

* cameron's diff + lint & test fixes

Co-authored-by: Cameron Kurotori <[email protected]>

* lint

* lint again

---------

Co-authored-by: Cameron Kurotori <[email protected]>

Author: kt474

qiskit_ibm_runtime/accounts/account.py

@@ -272,7 +272,7 @@ def __init__(
 
     def get_auth_handler(self) -> AuthBase:
         """Returns the Cloud authentication handler."""
-        return CloudAuth(api_key=self.token, crn=self.instance)
+        return CloudAuth(api_key=self.token, crn=self.instance, private=self.private_endpoint)
 
     def resolve_crn(self) -> None:
         """Resolves the corresponding unique Cloud Resource Name (CRN) for the given non-unique service

qiskit_ibm_runtime/api/auth.py

@@ -14,17 +14,28 @@
 
 from typing import Dict
 
-
+import warnings
 from requests import PreparedRequest
 from requests.auth import AuthBase
 
+from ibm_cloud_sdk_core import IAMTokenManager
+from ..utils.utils import cname_from_crn
+
+CLOUD_IAM_URL = "iam.cloud.ibm.com"
+STAGING_CLOUD_IAM_URL = "iam.test.cloud.ibm.com"
+
 
 class CloudAuth(AuthBase):
     """Attaches IBM Cloud Authentication to the given Request object."""
 
-    def __init__(self, api_key: str, crn: str):
-        self.api_key = api_key
+    def __init__(self, api_key: str, crn: str, private: bool = False):
         self.crn = crn
+        self.api_key = api_key
+        iam_url = (
+            f"https://{'private.' if private else ''}"
+            f"{STAGING_CLOUD_IAM_URL if cname_from_crn(crn) == 'staging' else CLOUD_IAM_URL}"
+        )
+        self.tm = IAMTokenManager(api_key, url=iam_url)
 
     def __eq__(self, other: object) -> bool:
         if isinstance(other, CloudAuth):
@@ -42,7 +53,14 @@ def __call__(self, r: PreparedRequest) -> PreparedRequest:
 
     def get_headers(self) -> Dict:
         """Return authorization information to be stored in header."""
-        return {"Service-CRN": self.crn, "Authorization": f"apikey {self.api_key}"}
+        try:
+            access_token = self.tm.get_token()
+            return {"Service-CRN": self.crn, "Authorization": f"Bearer {access_token}"}
+        except Exception as ex:  # pylint: disable=broad-except
+            warnings.warn(
+                f"Unable to retrieve IBM Cloud access token. API Key will be used instead. {ex}"
+            )
+            return {"Service-CRN": self.crn, "Authorization": f"apikey {self.api_key}"}
 
 
 class QuantumAuth(AuthBase):

qiskit_ibm_runtime/api/client_parameters.py

@@ -62,7 +62,7 @@ def __init__(
     def get_auth_handler(self) -> Union[CloudAuth, QuantumAuth]:
         """Returns the respective authentication handler."""
         if self.channel == "ibm_cloud":
-            return CloudAuth(api_key=self.token, crn=self.instance)
+            return CloudAuth(api_key=self.token, crn=self.instance, private=self.private_endpoint)
 
         return QuantumAuth(access_token=self.token)
 

qiskit_ibm_runtime/utils/utils.py

@@ -356,6 +356,21 @@ def _location_from_crn(crn: str) -> str:
     return re.search(pattern, crn).group(6)
 
 
+def cname_from_crn(crn: str) -> str:
+    """Computes the CNAME ('bluemix' or 'staging') from a given CRN.
+
+    Args:
+        crn: A CRN (format: https://cloud.ibm.com/docs/account?topic=account-crn#format-crn)
+
+    Returns:
+        The location.
+    """
+    if is_crn(crn):
+        pattern = "(.*?):(.*?):(.*?):(.*?):(.*?):(.*?):.*"
+        return re.search(pattern, crn).group(3)
+    return None
+
+
 def to_python_identifier(name: str) -> str:
     """Convert a name to a valid Python identifier.
 

release-notes/unreleased/2102.other.rst

@@ -0,0 +1,2 @@
+IBM Cloud accounts will now use an access token to call the Qiskit Runtime API instead of the 
+token provided by the user. 

test/integration/test_auth_client.py

@@ -31,6 +31,19 @@ def test_valid_login(self, dependencies: IntegrationTestDependencies) -> None:
         client = self._init_auth_client(dependencies.token, dependencies.url)
         self.assertTrue(client.access_token)
 
+    @integration_test_setup(supported_channel=["ibm_cloud"], init_service=False)
+    def test_cloud_access_token(self, dependencies: IntegrationTestDependencies) -> None:
+        """Test valid cloud authentication."""
+        params = ClientParameters(
+            channel="ibm_cloud",
+            token=dependencies.token,
+            url=dependencies.url,
+            instance=dependencies.instance,
+        )
+        cloud_auth = params.get_auth_handler()
+        self.assertTrue(cloud_auth.tm)
+        self.assertTrue(cloud_auth.tm.access_token())
+
     @integration_test_setup(supported_channel=["ibm_quantum"], init_service=False)
     def test_url_404(self, dependencies: IntegrationTestDependencies) -> None:
         """Test login against a 404 URL"""