PILOT-3002: Add support for API Key

.editorconfig

@@ -0,0 +1,18 @@
+root = true
+
+[*]
+charset = utf-8
+tab_width = 4
+indent_style = space
+
+[*.py]
+end_of_line = lf
+indent_size = 4
+trim_trailing_whitespace = true
+insert_final_newline = true
+
+[*.{yml,yaml}]
+end_of_line = lf
+indent_size = 2
+trim_trailing_whitespace = true
+insert_final_newline = true

.github/workflows/run-tests.yml

@@ -47,7 +47,7 @@ jobs:
         uses: pre-commit/action@v3.0.0
 
       - name: Run tests
-        run: poetry run pytest -vvv --exitfirst --cov=app --cov-report=term --cov-report=xml --cov-fail-under=23
+        run: poetry run pytest -vvv --exitfirst --cov=app --cov-report=term --cov-report=xml --cov-fail-under=54
 
       - name: Coverage report comment
         uses: mishakav/pytest-coverage-comment@v1.1.42

app/commands/user.py

@@ -2,11 +2,15 @@
 #
 # Contact Indoc Research for any questions regarding the use of this source code.
 
+from typing import Union
+
 import click
 
 import app.services.output_manager.help_page as user_help
 import app.services.output_manager.message_handler as mhandler
+from app.models.enums import LoginMethod
 from app.services.user_authentication.decorator import require_login_session
+from app.services.user_authentication.user_login_logout import login_using_api_key
 from app.services.user_authentication.user_login_logout import user_device_id_login
 from app.services.user_authentication.user_login_logout import user_logout
 from app.services.user_authentication.user_login_logout import validate_user_device_login
@@ -20,22 +24,36 @@ def cli():
 
 
 @click.command()
+@click.option(
+    '--api-key',
+    envvar='PILOT_API_KEY',
+    help=(user_help.user_help_page(user_help.UserHELP.USER_LOGIN_API_KEY)),
+)
 @doc(user_help.user_help_page(user_help.UserHELP.USER_LOGIN))
-def login():
-    device_login = user_device_id_login()
-    if device_login:
-        mhandler.SrvOutPutHandler.login_input_device_code(device_login['verification_uri_complete'])
-        mhandler.SrvOutPutHandler.login_device_code_qrcode(device_login['verification_uri_complete'])
+def login(api_key: Union[str, None]):
+    if api_key:
+        mhandler.SrvOutPutHandler.login_using_method(LoginMethod.API_KEY)
+        is_valid = login_using_api_key(api_key)
+        if is_valid:
+            mhandler.SrvOutPutHandler.login_success()
+        else:
+            mhandler.SrvOutPutHandler.login_using_api_key_failed_error()
     else:
-        mhandler.SrvOutPutHandler.login_input_device_error()
+        mhandler.SrvOutPutHandler.login_using_method(LoginMethod.DEVICE_CODE)
+        device_login = user_device_id_login()
+        if device_login:
+            mhandler.SrvOutPutHandler.login_input_device_code(device_login['verification_uri_complete'])
+            mhandler.SrvOutPutHandler.login_device_code_qrcode(device_login['verification_uri_complete'])
+        else:
+            mhandler.SrvOutPutHandler.login_input_device_error()
 
-    is_validated = validate_user_device_login(
-        device_login['device_code'], device_login['expires'], device_login['interval']
-    )
-    if is_validated:
-        mhandler.SrvOutPutHandler.login_success()
-    else:
-        mhandler.SrvOutPutHandler.validation_login_input_device_error()
+        is_validated = validate_user_device_login(
+            device_login['device_code'], device_login['expires'], device_login['interval']
+        )
+        if is_validated:
+            mhandler.SrvOutPutHandler.login_success()
+        else:
+            mhandler.SrvOutPutHandler.validation_login_input_device_error()
 
 
 @click.command()

app/configs/app_config.py

@@ -5,8 +5,8 @@
 from env import ConfigClass
 
 
-class AppConfig(object):
-    class Env(object):
+class AppConfig:
+    class Env:
         section = 'environment'
         project = ConfigClass.project
         user_config_path = ConfigClass.config_path
@@ -34,8 +34,9 @@ class Env(object):
         greenroom_bucket_prefix = 'gr'
 
         keycloak_device_client_id = ConfigClass.keycloak_device_client_id
+        keycloak_api_key_audience = ConfigClass.keycloak_api_key_audience
 
-    class Connections(object):
+    class Connections:
         section = 'connections'
         url_harbor = ConfigClass.url_harbor
         url_authn = ConfigClass.url_authn
@@ -53,6 +54,7 @@ class Connections(object):
         url_validation = ConfigClass.url_validation
         url_keycloak = ConfigClass.url_keycloak
         url_keycloak_token = f'{ConfigClass.url_keycloak}/token'
+        url_keycloak_realm = ConfigClass.url_keycloak.rstrip('/').replace('/protocol/openid-connect', '')
         url_bff = ConfigClass.url_bff
         # add url_base to check if value exist
         url_base = ConfigClass.base_url

app/configs/user_config.py

@@ -6,13 +6,12 @@
 import os
 import time
 
+from app.configs.app_config import AppConfig
 from app.models.singleton import Singleton
 from app.services.crypto.crypto import decryption
 from app.services.crypto.crypto import encryption
 from app.services.crypto.crypto import generate_secret
 
-from .app_config import AppConfig
-
 
 class UserConfig(metaclass=Singleton):
     """The class to maintain the user access/fresh token Note here: the base class is Singleton, meaning no matter how
@@ -32,6 +31,7 @@ def __init__(self):
             self.config['USER'] = {
                 'username': '',
                 'password': '',
+                'api_key': '',
                 'access_token': '',
                 'refresh_token': '',
                 'secret': generate_secret(),
@@ -49,6 +49,7 @@ def clear(self):
         self.config['USER'] = {
             'username': '',
             'password': '',
+            'api_key': '',
             'access_token': '',
             'refresh_token': '',
             'hpc_token': '',
@@ -59,7 +60,7 @@ def clear(self):
         self.save()
 
     def is_logged_in(self) -> bool:
-        return bool(self.access_token and self.refresh_token)
+        return bool(self.api_key or (self.access_token and self.refresh_token))
 
     @property
     def username(self):
@@ -77,6 +78,14 @@ def password(self):
     def password(self, val):
         self.config['USER']['password'] = encryption(val, self.secret)
 
+    @property
+    def api_key(self):
+        return decryption(self.config['USER']['api_key'], self.secret)
+
+    @api_key.setter
+    def api_key(self, val):
+        self.config['USER']['api_key'] = encryption(val, self.secret)
+
     @property
     def access_token(self):
         return decryption(self.config['USER']['access_token'], self.secret)

app/models/enums.py

@@ -0,0 +1,12 @@
+# Copyright (C) 2023 Indoc Research
+#
+# Contact Indoc Research for any questions regarding the use of this source code.
+
+from enum import Enum
+
+
+class LoginMethod(str, Enum):
+    """Available login methods."""
+
+    API_KEY = 'api-key'
+    DEVICE_CODE = 'device-code'

app/resources/custom_help.py

@@ -27,6 +27,7 @@ class HelpPage:
             ),
             'USER_LOGIN_USERNAME': 'Specify username for login.',
             'USER_LOGIN_PASSWORD': 'Specify password for login.',
+            'USER_LOGIN_API_KEY': 'Specify API Key for login.',
         },
         'file': {
             'FILE_ATTRIBUTE_LIST': 'List attribute templates of a given Project.',

app/services/output_manager/help_page.py

@@ -53,6 +53,7 @@ class UserHELP(enum.Enum):
     USER_LOGOUT_CONFIRM = 'USER_LOGOUT_CONFIRM'
     USER_LOGIN_USERNAME = 'USER_LOGIN_USERNAME'
     USER_LOGIN_PASSWORD = 'USER_LOGIN_PASSWORD'
+    USER_LOGIN_API_KEY = 'USER_LOGIN_API_KEY'
 
 
 def user_help_page(UserHELP: UserHELP):

app/services/output_manager/message_handler.py

@@ -7,10 +7,23 @@
 import qrcode
 
 import app.services.logger_services.log_functions as logger
+from app.models.enums import LoginMethod
 from app.models.service_meta_class import MetaService
 
 
 class SrvOutPutHandler(metaclass=MetaService):
+    @staticmethod
+    def login_using_method(method: LoginMethod):
+        """Selected login method message."""
+        return logger.info(f'Trying to log in using "{method.value}" method.')
+
+    @staticmethod
+    def login_using_api_key_failed_error():
+        """Error when logging in with the API Key!"""
+        return logger.error(
+            f'Failed to log in using "{LoginMethod.API_KEY.value}" method, please make sure you are using a valid key!'
+        )
+
     @staticmethod
     def login_device_code_qrcode(url: str):
         """Print QRCode with login url!"""

app/services/user_authentication/token_manager.py

@@ -9,8 +9,10 @@
 
 from app.configs.app_config import AppConfig
 from app.configs.user_config import UserConfig
+from app.models.enums import LoginMethod
 from app.models.service_meta_class import MetaService
 from app.services.output_manager.error_handler import SrvErrorHandler
+from app.services.user_authentication.user_login_logout import exchange_api_key
 
 
 class SrvTokenManager(metaclass=MetaService):
@@ -37,6 +39,13 @@ def decode_refresh_token(self):
         tokens = self.get_token()
         return jwt.decode(tokens[1], verify=False)
 
+    def is_api_key(self) -> bool:
+        token = self.decode_access_token()
+        audience = token['aud']
+        if isinstance(audience, str):
+            audience = [audience]
+        return AppConfig.Env.keycloak_api_key_audience.issubset(set(audience))
+
     def check_valid(self, required_azp):
         """
         check token validation
@@ -49,20 +58,26 @@ def check_valid(self, required_azp):
         now = time.time()
         diff = expiry_at - now
 
-        # TODO: check why here will need enforce the token refresh when
-        # azp is not `kong``
-        # ``kong`` is hardcoded in the decorator definition as default value.
-        azp_token_condition = decoded_access_token['azp'] not in [required_azp, AppConfig.Env.keycloak_device_client_id]
+        if not self.is_api_key():
+            # TODO: check why here will need enforce the token refresh when
+            # azp is not `kong``
+            # ``kong`` is hardcoded in the decorator definition as default value.
+            azp_token_condition = decoded_access_token['azp'] not in [
+                required_azp,
+                AppConfig.Env.keycloak_device_client_id,
+            ]
+
+            if azp_token_condition or expiry_at <= now:
+                return 2
 
-        if azp_token_condition or expiry_at <= now:
-            return 2
-        # print(expiry_at, now)
-        # print(diff, AppConfig.Env.token_warn_need_refresh)
         if diff <= AppConfig.Env.token_warn_need_refresh:
             return 1
         return 0
 
-    def refresh(self, azp: str):
+    def refresh(self, azp: str) -> None:
+        if self.is_api_key():
+            return self.refresh_api_key()
+
         url = AppConfig.Connections.url_keycloak_token
         payload = {
             'grant_type': 'refresh_token',
@@ -79,4 +94,12 @@ def refresh(self, azp: str):
             self.update_token(response.json()['access_token'], response.json()['refresh_token'])
         else:
             SrvErrorHandler.default_handle(response.content)
-        return response.json()
+
+    def refresh_api_key(self) -> None:
+        access_token = exchange_api_key(self.config.api_key)
+        if access_token is None:
+            return SrvErrorHandler.default_handle(
+                f'Unable to get access token using "{LoginMethod.API_KEY.value}" method. Unable to proceed.', True
+            )
+
+        self.update_token(access_token, '')

app/services/user_authentication/user_login_logout.py

@@ -5,10 +5,12 @@
 import time
 from typing import Any
 from typing import Dict
+from typing import Union
 from uuid import uuid4
 
 import jwt
 import requests
+from requests import RequestException
 
 from app.configs.app_config import AppConfig
 from app.configs.user_config import UserConfig
@@ -17,6 +19,42 @@
 from app.services.output_manager.message_handler import SrvOutPutHandler
 
 
+def exchange_api_key(api_key: str) -> Union[str, None]:
+    """Exchange API Key with JWT token using Keycloak."""
+
+    url = f'{AppConfig.Connections.url_keycloak_realm}/api-key/{api_key}'
+    try:
+        response = requests.get(url, timeout=5)
+        response.raise_for_status()
+    except RequestException:
+        return None
+
+    return response.json()['access_token']
+
+
+def login_using_api_key(api_key: str) -> bool:
+    """Try to log in using API Key and store results in user config."""
+
+    access_token = exchange_api_key(api_key)
+    if access_token is None:
+        return False
+
+    decoded_token = jwt.decode(access_token, verify=False)
+    username = decoded_token['preferred_username']
+
+    user_config = UserConfig()
+    user_config.api_key = api_key
+    user_config.access_token = access_token
+    user_config.refresh_token = ''
+    user_config.username = username
+    user_config.last_active = str(int(time.time()))
+    user_config.hpc_token = ''
+    user_config.session_id = 'cli-' + str(uuid4())
+    user_config.save()
+
+    return True
+
+
 def user_device_id_login() -> Dict[str, Any]:
     """Get device code URL for user login."""
 
@@ -64,6 +102,7 @@ def validate_user_device_login(device_code: str, expires: int, interval: int) ->
     resp_dict = resp.json()
     decode_token = jwt.decode(resp_dict['access_token'], verify=False)
     user_config = UserConfig()
+    user_config.api_key = ''
     user_config.access_token = resp_dict['access_token']
     user_config.refresh_token = resp_dict['refresh_token']
     user_config.username = decode_token['preferred_username']

env.py

@@ -3,6 +3,7 @@
 # Contact Indoc Research for any questions regarding the use of this source code.
 
 import os
+from typing import Set
 
 from dotenv import load_dotenv
 from pydantic import BaseSettings
@@ -25,6 +26,7 @@ class Settings(BaseSettings):
     url_keycloak: str = ''
 
     keycloak_device_client_id: str = 'cli_test2'
+    keycloak_api_key_audience: Set[str] = {'api-key'}
 
     VM_INFO: str = ''