feat: add keycloak

.gitignore

@@ -1,6 +1,6 @@
 # Mac OSX hidden files
 .DS_Store
-
+venv
 # Byte-compiled / optimized / DLL files
 *.py[cod]
 __pycache__/

genotype_api/clients/authentication/keycloak_client.py

@@ -0,0 +1,35 @@
+from keycloak import KeycloakOpenID
+from keycloak.exceptions import KeycloakConnectionError
+
+
+class KeycloakClient:
+    def __init__(self, server_url, client_id, realm_name, client_secret_key, redirect_uri):
+        self.client_id = client_id
+        self.client_secret_key = client_secret_key
+        self.realm_name = realm_name
+        self.redirect_uri = redirect_uri
+        self.server_url = server_url
+        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 decode_token(self, access_token: str) -> dict:
+        """Decode an access token.
+        Args:
+            access_token: An jwt access token.
+        """
+        client: KeycloakOpenID = self.get_client()
+        return client.decode_token(token=access_token)

genotype_api/config.py

@@ -2,6 +2,8 @@
 
 from pydantic_settings import BaseSettings
 
+from genotype_api.clients.authentication.keycloak_client import KeycloakClient
+
 GENOTYPE_PACKAGE = Path(__file__).parent
 PACKAGE_ROOT: Path = GENOTYPE_PACKAGE.parent
 ENV_FILE: Path = PACKAGE_ROOT / ".env"
@@ -25,9 +27,11 @@ class Config:
 class SecuritySettings(BaseSettings):
     """Settings for serving the genotype-api app"""
 
-    client_id: str = ""
-    algorithm: str = ""
-    jwks_uri: str = "https://www.googleapis.com/oauth2/v3/certs"
+    keycloak_client_id: str = "client_id"
+    keycloak_client_secret: str = "client_secret"
+    keycloak_server_url: str = "server_url"
+    keycloak_realm_name: str = "realm_name"
+    keycloak_redirect_uri: str = "redirect_uri"
     api_root_path: str = "/"
 
     class Config:
@@ -37,3 +41,11 @@ class Config:
 security_settings = SecuritySettings()
 
 settings = DBSettings()
+
+keycloak_client = KeycloakClient(
+    server_url=security_settings.keycloak_server_url,
+    client_id=security_settings.keycloak_client_id,
+    client_secret_key=security_settings.keycloak_client_secret,
+    realm_name=security_settings.keycloak_realm_name,
+    redirect_uri=security_settings.keycloak_redirect_uri,
+)

genotype_api/exceptions.py

@@ -29,6 +29,10 @@ class UserNotFoundError(Exception):
     pass
 
 
+class UserRoleError(Exception):
+    pass
+
+
 class UserArchiveError(Exception):
     pass
 
@@ -55,3 +59,7 @@ class SNPExistsError(Exception):
 
 class PlateExistsError(Exception):
     pass
+
+
+class AuthenticationError(Exception):
+    pass

genotype_api/models.py

@@ -52,3 +52,29 @@ class MatchCounts(BaseModel):
 class MatchResult(BaseModel):
     sample_id: str
     match_results: MatchCounts | None = None
+
+
+class RealmAccess(BaseModel):
+    roles: list[str]
+
+
+class DecodingResponse(BaseModel):
+    exp: int
+    iat: int
+    auth_time: int
+    jti: str
+    iss: str
+    sub: str
+    typ: str
+    azp: str
+    sid: str
+    acr: str
+    allowed_origins: list[str] | None = None
+    realm_access: RealmAccess
+    scope: str
+    email_verified: bool
+    name: str
+    preferred_username: str
+    given_name: str
+    family_name: str
+    email: str

genotype_api/security.py

@@ -1,35 +1,22 @@
-import requests
 from fastapi import Depends, HTTPException, Security
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from jose import jwt
 from starlette import status
 from starlette.requests import Request
 
-from genotype_api.config import security_settings
+from genotype_api.config import security_settings, keycloak_client
 from genotype_api.database.models import User
 from genotype_api.database.store import Store, get_store
 from genotype_api.dto.user import CurrentUser
 
-
-def decode_id_token(token: str):
-    try:
-        payload = jwt.decode(
-            token,
-            key=requests.get(security_settings.jwks_uri).json(),
-            algorithms=[security_settings.algorithm],
-            audience=security_settings.client_id,
-            options={
-                "verify_at_hash": False,
-            },
-        )
-        return payload
-    except jwt.JWTError:
-        return None
+from genotype_api.exceptions import AuthenticationError
+from genotype_api.services.authentication.service import AuthenticationService
 
 
 class JWTBearer(HTTPBearer):
-    def __init__(self, auto_error: bool = True):
+    def __init__(self, auth_service: AuthenticationService, auto_error: bool = True):
         super(JWTBearer, self).__init__(auto_error=auto_error)
+        self.auth_service = auth_service
 
     async def __call__(self, request: Request):
         credentials: HTTPAuthorizationCredentials = await super(JWTBearer, self).__call__(request)
@@ -48,19 +35,23 @@ async def __call__(self, request: Request):
 
     def verify_jwt(self, jwtoken: str) -> dict | None:
         try:
-            payload = decode_id_token(jwtoken)
+            payload: dict = self.auth_service.verify_token(jwtoken).model_dump()
             if payload and "email" in payload:
                 return {"email": payload["email"]}
             else:
                 return None
-        except jwt.JWTError:
+        except AuthenticationError as error:
             raise HTTPException(
                 status_code=status.HTTP_403_FORBIDDEN,
-                detail="Invalid token or expired token.",
+                detail=f"{error}",
             )
 
 
-jwt_scheme = JWTBearer()
+auth_service = AuthenticationService(
+    redirect_uri=security_settings.keycloak_redirect_uri,
+    keycloak_client=keycloak_client,
+)
+jwt_scheme = JWTBearer(auth_service=auth_service)
 
 
 async def get_active_user(

genotype_api/services/authentication/service.py

@@ -0,0 +1,51 @@
+from genotype_api.clients.authentication.keycloak_client import KeycloakClient
+from genotype_api.exceptions import AuthenticationError, UserRoleError
+from genotype_api.models import DecodingResponse
+
+
+class AuthenticationService:
+    """Authentication service to verify tokens against Keycloak and return user information."""
+
+    def __init__(
+        self,
+        redirect_uri: str,
+        keycloak_client: KeycloakClient,
+    ):
+        """Initialize the AuthenticationService
+
+        Args:
+            redirect_uri: Redirect uri for keycloak
+            keycloak_client: KeycloakOpenID client.
+        """
+        self.redirect_uri = redirect_uri
+        self.client = keycloak_client
+
+    def verify_token(self, jwt_token: str) -> DecodingResponse:
+        """Verify the token and user role.
+        Args:
+            token (str): The token to verify.
+
+        Returns:
+            Decoded token payload
+
+        Raises:
+            AuthenticationError: if an error occured the authentication
+        """
+        try:
+            decoded_token = DecodingResponse(**self.client.decode_token(jwt_token))
+            self.check_role(decoded_token.realm_access.roles)
+            return decoded_token
+        except Exception as error:
+            raise AuthenticationError(f"An error occured during authorisation: {error}")
+
+    @staticmethod
+    def check_role(roles: list[str]) -> None:
+        """Check the user roles.
+        Currently set to a single permissable role, expand if needed.
+        Args:
+            roles (list[str]): The user roles received from the RealmAccess.
+        Raises:
+            UserRoleError: if required role not present
+        """
+        if not "cg-employee" in roles:
+            raise UserRoleError("The user does not have the required role to access this service.")