MGMT-19026: Add authentication support for refresh token of ocm-cli

Makefile

@@ -64,6 +64,7 @@ ROUTE53_SECRET := $(or $(ROUTE53_SECRET), "")
 OFFLINE_TOKEN := $(or $(OFFLINE_TOKEN), "")
 SERVICE_ACCOUNT_CLIENT_ID := $(or $(SERVICE_ACCOUNT_CLIENT_ID), "")
 SERVICE_ACCOUNT_CLIENT_SECRET := $(or $(SERVICE_ACCOUNT_CLIENT_SECRET), "")
+OCM_CLI_REFRESH_TOKEN := $(or $(OCM_CLI_REFRESH_TOKEN), "")
 
 # deploy
 DEPLOY_TAG := $(or $(DEPLOY_TAG), "")

skipper.env

@@ -193,6 +193,9 @@ IMAGES_FLAVOR
 SERVICE_ACCOUNT_CLIENT_ID
 SERVICE_ACCOUNT_CLIENT_SECRET
 OCM_SELF_TOKEN
+OCM_CLI_REFRESH_TOKEN
+# SECOND_OCM_CLI_REFRESH_TOKEN is required for QE automated tests
+SECOND_OCM_CLI_REFRESH_TOKEN
 OS_IMAGES_PATH
 RELEASE_IMAGES_PATH
 

src/assisted_test_infra/download_logs/__main__.py

@@ -51,6 +51,7 @@ def main():
         service_account=ServiceAccount(
             client_id=get_env("SERVICE_ACCOUNT_CLIENT_ID"), client_secret=get_env("SERVICE_ACCOUNT_CLIENT_SECRET")
         ),
+        refresh_token=get_env("OCM_CLI_REFRESH_TOKEN"),
         timeout=CONNECTION_TIMEOUT,
     )
     if args.cluster_id:

src/manage/manage.py

@@ -21,9 +21,11 @@
 
 
 class Manage:
-    def __init__(self, inventory_url: str, type: str, offline_token: str, service_account: ServiceAccount):
+    def __init__(
+        self, inventory_url: str, type: str, offline_token: str, service_account: ServiceAccount, refresh_token: str
+    ):
         self.client = ClientFactory.create_client(
-            url=inventory_url, offline_token=offline_token, service_account=service_account
+            url=inventory_url, offline_token=offline_token, service_account=service_account, refresh_token=refresh_token
         )
 
         with open("src/manage/manageable_options.yaml", "r") as f:
@@ -104,6 +106,9 @@ def handle_arguments():
         help="client secret of the service account used to authenticate against assisted-service",
         type=str,
     )
+    parser.add_argument(
+        "--ocm-cli-refresh-token", help="ocm cli refresh token to authenticate against assisted-service", type=str
+    )
     parser.add_argument("--type", help="Type of managing process to commit", type=str)
 
     return parser.parse_args()
@@ -118,6 +123,7 @@ def main():
         service_account=ServiceAccount(
             client_id=args.service_account_client_id, client_secret=args.service_account_client_secret
         ),
+        refresh_token=args.ocm_cli_refresh_token,
     )
 
 

src/service_client/assisted_service_api.py

@@ -40,6 +40,7 @@ def is_provided(self) -> bool:
 class AuthenticationMethod(enum.Enum):
     OFFLINE_TOKEN = "OFFLINE_TOKEN"
     SERVICE_ACCOUNT = "SERVICE_ACCOUNT"
+    REFRESH_TOKEN = "REFRESH_TOKEN"
 
 
 class InventoryClient(object):
@@ -48,13 +49,16 @@ def __init__(
         inventory_url: str,
         offline_token: Optional[str],
         service_account: Optional[ServiceAccount],
+        refresh_token: Optional[str],
         pull_secret: str,
     ):
         self.inventory_url = inventory_url
         configs = Configuration()
         configs.host = self.get_host(configs)
         configs.verify_ssl = False
-        self.set_config_auth(c=configs, offline_token=offline_token, service_account=service_account)
+        self.set_config_auth(
+            c=configs, offline_token=offline_token, service_account=service_account, refresh_token=refresh_token
+        )
         self._set_x_secret_key(configs, pull_secret)
 
         self.api = ApiClient(configuration=configs)
@@ -77,13 +81,20 @@ def get_host(self, configs: Configuration) -> str:
 
     @classmethod
     def set_config_auth(
-        cls, c: Configuration, offline_token: Optional[str], service_account: Optional[ServiceAccount]
+        cls,
+        c: Configuration,
+        offline_token: Optional[str],
+        service_account: Optional[ServiceAccount],
+        refresh_token: Optional[str],
     ) -> None:
         if service_account is not None and service_account.is_provided():
             authentication_method = AuthenticationMethod.SERVICE_ACCOUNT
             log.info("authenticating to assisted service using service account")
+        elif refresh_token is not None:
+            authentication_method = AuthenticationMethod.REFRESH_TOKEN
+            log.info("authenticating to assisted service using ocm cli refresh token")
         else:
-            log.info("service account was not provided, trying offline token")
+            log.info("service account or refresh token was not provided, trying offline token")
             if offline_token is None:
                 log.info("offline token wasn't provided as well, skipping authentication headers")
                 return
@@ -112,6 +123,12 @@ def refresh_api_key(config: Configuration) -> None:
                     "client_id": service_account.client_id,
                     "client_secret": service_account.client_secret,
                 }
+            elif authentication_method == AuthenticationMethod.REFRESH_TOKEN:
+                params = {
+                    "client_id": "ocm-cli",
+                    "grant_type": "refresh_token",
+                    "refresh_token": refresh_token,
+                }
             else:
                 params = {
                     "client_id": "cloud-services",

src/service_client/client_factory.py

@@ -18,13 +18,18 @@ def create_client(
         url: str,
         offline_token: str,
         service_account: ServiceAccount,
+        refresh_token: str,
         pull_secret: Optional[str] = "",
         wait_for_api: Optional[bool] = True,
         timeout: Optional[int] = consts.WAIT_FOR_BM_API,
     ) -> InventoryClient:
         log.info("Creating assisted-service client for url: %s", url)
         c = InventoryClient(
-            inventory_url=url, offline_token=offline_token, service_account=service_account, pull_secret=pull_secret
+            inventory_url=url,
+            offline_token=offline_token,
+            service_account=service_account,
+            refresh_token=refresh_token,
+            pull_secret=pull_secret,
         )
         if wait_for_api:
             c.wait_for_api_readiness(timeout)

src/tests/global_variables/default_variables.py

@@ -44,16 +44,21 @@ def get_env(self, item: str) -> EnvVar:
         return _EnvVariables.__getattribute__(self, item)
 
     def get_api_client(
-        self, offline_token: Optional[str] = None, service_account: Optional[ServiceAccount] = None, **kwargs
+        self,
+        offline_token: Optional[str] = None,
+        service_account: Optional[ServiceAccount] = None,
+        refresh_token: Optional[str] = None,
+        **kwargs,
     ) -> InventoryClient:
         url = self.remote_service_url
 
         offline_token = offline_token or self.offline_token
         service_account = service_account or ServiceAccount(
             client_id=self.service_account_client_id, client_secret=self.service_account_client_secret
         )
+        refresh_token = refresh_token or self.refresh_token
 
         if not url:
             url = utils.get_local_assisted_service_url(self.namespace, "assisted-service", self.deploy_target)
 
-        return ClientFactory.create_client(url, offline_token, service_account, **kwargs)
+        return ClientFactory.create_client(url, offline_token, service_account, refresh_token, **kwargs)

src/tests/global_variables/env_variables_defaults.py

@@ -21,6 +21,7 @@ class _EnvVariables(DataPool, ABC):
     offline_token: EnvVar = EnvVar(["OFFLINE_TOKEN"])
     service_account_client_id: EnvVar = EnvVar(["SERVICE_ACCOUNT_CLIENT_ID"])
     service_account_client_secret: EnvVar = EnvVar(["SERVICE_ACCOUNT_CLIENT_SECRET"])
+    refresh_token: EnvVar = EnvVar(["OCM_CLI_REFRESH_TOKEN"])
     kernel_arguments: EnvVar = EnvVar(["KERNEL_ARGUMENTS"], loader=json.loads)
     host_installer_args: EnvVar = EnvVar(["HOST_INSTALLER_ARGS"], loader=json.loads)
     openshift_version: EnvVar = EnvVar(["OPENSHIFT_VERSION"], default=consts.OpenshiftVersion.DEFAULT.value)