feat(keycloak authentication)

cg/server/app_config.py

@@ -25,6 +25,10 @@ class AppConfig(BaseSettings):
     freshdesk_api_key: str = "freshdesk_api_key"
     freshdesk_order_email_id: int = 10
     freshdesk_environment: str = "Stage"
+    keycloak_client_url: str = "http://localhost:8081"
+    keycloak_realm_name = "orderportal"
+    keycloak_client_id = "cg-backend-client"
+    keycloak_client_secret_key = "cg-very-secret-password"
 
 
 app_config = AppConfig()

cg/server/endpoints/utils.py

@@ -3,26 +3,22 @@
 from http import HTTPStatus
 
 import cachecontrol
+from keycloak import KeycloakError
 import requests
 from flask import abort, current_app, g, jsonify, make_response, request
-from google.auth import exceptions
-from google.auth.transport import requests as google_requests
-from google.oauth2 import id_token
 
-from cg.server.ext import db
-from cg.store.models import User
+from cg.server.ext import auth_service
+from cg.services.authentication.models import AuthenticatedUser
+
+
+
 
 LOG = logging.getLogger(__name__)
 
 session = requests.session()
 cached_session = cachecontrol.CacheControl(session)
 
 
-def verify_google_token(token):
-    request = google_requests.Request(session=cached_session)
-    return id_token.verify_oauth2_token(id_token=token, request=request)
-
-
 def is_public(route_function):
     @wraps(route_function)
     def public_endpoint(*args, **kwargs):
@@ -54,17 +50,19 @@ def before_request():
 
     jwt_token = auth_header.split("Bearer ")[-1]
     try:
-        user_data = verify_google_token(jwt_token)
-    except (exceptions.OAuthError, ValueError) as e:
-        LOG.error(f"Error {e} occurred while decoding JWT token: {jwt_token}")
+        user: AuthenticatedUser = auth_service.verify_token(jwt_token)
+
+    except ValueError as error:
         return abort(
-            make_response(jsonify(message="outdated login certificate"), HTTPStatus.UNAUTHORIZED)
+            make_response(jsonify(message=str(error)), HTTPStatus.FORBIDDEN)
+        )
+    except KeycloakError as error:
+        return abort(
+            make_response(jsonify(message=str(error)), HTTPStatus.UNAUTHORIZED)
+        )
+    except Exception as error:
+        return abort(
+            make_response(jsonify(message=str(error)), HTTPStatus.INTERNAL_SERVER_ERROR)
         )
-
-    user: User = db.get_user_by_email(user_data["email"])
-    if user is None or not user.order_portal_login:
-        message = f"{user_data['email']} doesn't have access"
-        LOG.error(message)
-        return abort(make_response(jsonify(message=message), HTTPStatus.FORBIDDEN))
 
     g.current_user = user

cg/server/ext.py

@@ -4,12 +4,14 @@
 from flask_admin import Admin
 from flask_cors import CORS
 from flask_wtf.csrf import CSRFProtect
+from keycloak import KeycloakOpenID
 
 from cg.apps.lims import LimsAPI
 from cg.apps.tb.api import TrailblazerAPI
 from cg.clients.freshdesk.freshdesk_client import FreshdeskClient
 from cg.server.app_config import app_config
 from cg.services.application.service import ApplicationsWebService
+from cg.services.authentication.service import AuthenticationService
 from cg.services.delivery_message.delivery_message_service import DeliveryMessageService
 from cg.services.orders.order_service.order_service import OrderService
 from cg.services.orders.order_summary_service.order_summary_service import OrderSummaryService
@@ -108,3 +110,11 @@ def init_app(self, app):
     system_email_id=app_config.freshdesk_order_email_id,
     env=app_config.freshdesk_environment,
 )
+
+auth_service = AuthenticationService(
+    store=db,
+    server_url=app_config.keycloak_client_url,
+    client_id=app_config.keycloak_client_id,
+    client_secret=app_config.keycloak_client_secret_key,
+    realm_name=app_config.keycloak_realm_name,
+)

cg/services/authentication/constants.py

@@ -0,0 +1,2 @@
+ADMIN: str = "admin"
+CUSTOMER: str = "customer"

cg/services/authentication/models.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class AuthenticatedUser(BaseModel):
+    id: int
+    username: str
+    email: str
+    role: str

cg/services/authentication/service.py

@@ -0,0 +1,110 @@
+from cycler import V
+from cg.services.authentication.constants import ADMIN, CUSTOMER
+from cg.store.models import User
+from cg.store.store import Store
+from keycloak import KeycloakError, KeycloakOpenID
+from cg.services.authentication.models import AuthenticatedUser
+
+
+class AuthenticationService:
+    """Authentication service user to verify tokens against keycloak and return user information."""
+
+    def __init__(
+        self, store: Store, server_url: str, client_id: str, client_secret: str, realm_name: str
+    ):
+        """_summary_
+
+        Args:
+            store (Store): Connection to statusDB
+            server_url (str): server url to the keycloak server or container
+            realm_name (str): the keycloak realm to connect to (can be found in keycloak)
+            client_id (str): the client id to use in keycloak realm (can be found in keycloak)
+            client_secret (str): the client secret to use in keycloak realm (can be found in keycloak)  
+        """
+        self.store: Store = store
+        self.server_url: str = server_url
+        self.client_id: str = client_id
+        self.client_secret: str = client_secret
+        self.realm_name: str = realm_name
+        self.client: KeycloakOpenID = self._get_client()
+
+    def _get_client(self):
+        """Set the KeycloakOpenID client.
+        """
+        keycloak_openid_client = KeycloakOpenID(
+            server_url=self.server_url,
+            client_id=self.client_id,
+            realm_name=self.realm_name,
+            client_secret_key=self.client_secret,
+        )
+        return keycloak_openid_client
+
+    def verify_token(self, token: str) -> AuthenticatedUser:
+        """Verify the token and return the user.
+        args:
+            token: str
+        returns:
+            AuthenticatedUser
+        raises:
+            ValueError: if the token is not active
+        """
+        token_info = self.client.introspect(token)
+
+        if not token_info['active']:
+            raise ValueError('Token is not active')
+        verified_token = self.client.decode_token(token)
+
+        user_email = verified_token["email"]
+        return self._get_user_reponse(user_email)
+
+
+
+    def _get_user_by_email(self, email: str):
+        """
+        Get user by email.
+        args:
+            email: str
+        returns:
+            User
+        raises:
+            ValueError: if the user is not found
+        """
+        user: User | None = self.store.get_user_by_email(email)
+        if not user:
+            raise ValueError(f"User with email {email} not found")
+        return user
+
+    def _check_role(self, user: User) -> str:
+        """
+        Check if user has the role.
+        args:
+            user: User
+        returns:
+            str
+        raises:
+            ValueError: if the user does not have access
+        """
+        if user.is_admin:
+            return ADMIN
+        if user.order_portal_login:
+            return CUSTOMER
+        raise ValueError('User does not have access')  
+
+    def _get_user_reponse(self, email: str) -> AuthenticatedUser:
+        """
+        Get user response.
+        args:
+            email: str
+        returns:
+            AuthenticatedUser
+        """
+        user = self._get_user_by_email(email)
+        role: str = self._check_role(user)
+        return AuthenticatedUser(
+            id=user.id,
+            username=user.name,
+            email=user.email,
+            role=role,
+        )
+
+