feat(keycloak authentication)

cg/apps/tb/api.py

@@ -3,13 +3,10 @@
 import datetime
 import logging
 from typing import Any
-
-from google.auth.transport.requests import Request
-from google.oauth2 import service_account
-
 from cg.apps.tb.dto.create_job_request import CreateJobRequest
 from cg.apps.tb.dto.summary_response import AnalysisSummary, SummariesResponse
 from cg.apps.tb.models import AnalysesResponse, TrailblazerAnalysis
+from cg.clients.authentication import keycloak_client
 from cg.constants import Workflow
 from cg.constants.constants import APIMethods, FileFormat, JobType, WorkflowManager
 from cg.constants.priority import TrailblazerPriority
@@ -20,7 +17,8 @@
     TrailblazerAPIHTTPError,
 )
 from cg.io.controller import APIRequest, ReadStream
-
+from cg.services.authentication.models import TokenResponseModel
+from cg.clients.authentication.keycloak_client import KeycloakClient
 
 LOG = logging.getLogger(__name__)
 
@@ -43,19 +41,22 @@ class TrailblazerAPI:
         AnalysisStatus.QC,
     ]
 
-    def __init__(self, config: dict):
-        self.service_account = config["trailblazer"]["service_account"]
-        self.service_account_auth_file = config["trailblazer"]["service_account_auth_file"]
+    def __init__(self, config: dict, keycloak_client: KeycloakClient):
+        self.keycloak_client: KeycloakClient = keycloak_client
+        self.keycloak_backend_user = config["trailblazer"]["keycloak_backend_user"]
+        self.keycloak_backend_user_password = config["trailblazer"][
+            "keycloak_backend_user_password"
+        ]
         self.host = config["trailblazer"]["host"]
 
     @property
     def auth_header(self) -> dict:
-        credentials = service_account.IDTokenCredentials.from_service_account_file(
-            filename=self.service_account_auth_file,
-            target_audience="trailblazer",
+        token = TokenResponseModel(
+            **self.keycloak_client.get_token_by_user_password(
+                user_name=self.keycloak_backend_user, password=self.keycloak_backend_user_password
+            )
         )
-        credentials.refresh(Request())
-        return {"Authorization": f"Bearer {credentials.token}"}
+        return {"Authorization": f"Bearer {token.access_token}"}
 
     def query_trailblazer(
         self, command: str, request_body: dict, method: str = APIMethods.POST

cg/clients/authentication/keycloak_client.py

@@ -0,0 +1,73 @@
+from keycloak import KeycloakOpenID
+from keycloak import KeycloakConnectionError
+
+
+class KeycloakClient:
+    def __init__(self, server_url, client_id, realm_name, client_secret_key, redirect_uri):
+        self.server_url = server_url
+        self.client_id = client_id
+        self.realm_name = realm_name
+        self.client_secret_key = client_secret_key
+        self.redirect_uri = redirect_uri
+        self._client_instance: KeycloakOpenID | None = None
+
+    def get_client(self) -> KeycloakOpenID:
+        if self._client_instance is None:
+            try:
+                self._client_instance = KeycloakOpenID(
+                    server_url=self.server_url,
+                    client_id=self.client_id,
+                    realm_name=self.realm_name,
+                    client_secret_key=self.client_secret_key,
+                )
+            except KeycloakConnectionError as error:
+                raise KeycloakConnectionError(f"Failed to connect to Keycloak: {error}")
+            except Exception as error:
+                raise Exception(f"An error occurred while creating Keycloak client: {error}")
+        return self._client_instance
+
+    def get_auth_url(self, scope: str = "openid profile email") -> str:
+        """Get the authorization URL for user login."""
+        client = self.get_client()
+        return client.auth_url(redirect_uri=self.redirect_uri, scope=scope)
+
+    def logout_user(self, refresh_token: str) -> None:
+        """
+        Logout a user.
+        Args:
+            refresh_token: the refresh token stored in the session
+        """
+        client: KeycloakOpenID = self.get_client()
+        client.logout(refresh_token)
+
+    def get_token_by_authorisation_code(self, code: str) -> dict:
+        """
+        Get a token using the authorisation code.
+        Args:
+            code: code retrieved request
+        """
+        client: KeycloakOpenID = self.get_client()
+        return client.token(
+            grant_type="authorization_code", code=code, redirect_uri=self.redirect_uri
+        )
+
+    def get_token_by_user_password(self, user_name: str, password: str) -> dict:
+        """
+        Get a token using a username and password.
+        Args:
+            code: code retrieved request
+        """
+        client: KeycloakOpenID = self.get_client()
+        return client.token(grant_type="password", username=user_name, password=pa)
+
+    def get_user_info(self, access_token: str) -> dict:
+        """Get the user info for a provided access token.
+
+        Args:
+            access_token: access token given by keycloak
+
+        Returns:
+            dict: with the user information
+        """
+        client: KeycloakOpenID = self.get_client()
+        return client.userinfo(access_token)

cg/models/cg_config.py

@@ -20,6 +20,7 @@
 from cg.apps.scout.scoutapi import ScoutAPI
 from cg.apps.tb import TrailblazerAPI
 from cg.clients.arnold.api import ArnoldAPIClient
+from cg.clients.authentication.keycloak_client import KeycloakClient
 from cg.clients.chanjo2.client import Chanjo2APIClient
 from cg.clients.janus.api import JanusAPIClient
 from cg.constants.observations import LoqusdbInstance
@@ -125,8 +126,8 @@ class ClientConfig(BaseModel):
 
 
 class TrailblazerConfig(BaseModel):
-    service_account: str
-    service_account_auth_file: str
+    keycloak_backend_user: str
+    keycloak_backend_user_password: str
     host: str
 
 
@@ -387,6 +388,14 @@ class PostProcessingServices(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
 
+class KeyCloakConfig(BaseModel):
+    server_url: str
+    client_id: str
+    realm_name: str
+    client_secret_key: str
+    redirect_uri: str
+
+
 class CGConfig(BaseModel):
     data_input: DataInput | None = None
     database: str
@@ -456,6 +465,8 @@ class CGConfig(BaseModel):
     tar: CommonAppConfig | None = None
     trailblazer: TrailblazerConfig = None
     trailblazer_api_: TrailblazerAPI = None
+    keycloak: KeyCloakConfig = None
+    keycloak_client_: KeycloakClient | None = None
 
     # Meta APIs that will use the apps from CGConfig
     balsamic: BalsamicConfig | None = None
@@ -737,7 +748,7 @@ def trailblazer_api(self) -> TrailblazerAPI:
         api = self.__dict__.get("trailblazer_api_")
         if api is None:
             LOG.debug("Instantiating trailblazer api")
-            api = TrailblazerAPI(config=self.dict())
+            api = TrailblazerAPI(config=self.dict(), keycloak_client=self.keycloak_client)
             self.trailblazer_api_ = api
         return api
 
@@ -792,3 +803,18 @@ def delivery_service_factory(self) -> DeliveryServiceFactory:
             )
             self.delivery_service_factory_ = factory
         return factory
+
+    @property
+    def keycloak_client(self) -> KeycloakClient:
+        client = self.keycloak_client_
+        if client is None:
+            LOG.debug("Instantiating keycloak client")
+            client = KeycloakClient(
+                server_url=self.keycloak.server_url,
+                client_id=self.keycloak.client_id,
+                client_secret_key=self.keycloak.client_secret_key,
+                realm_name=self.keycloak.realm_name,
+                redirect_uri=self.keycloak.redirect_uri,
+            )
+            self.keycloak_client_ = client
+        return client