token validation

.github/workflows/python-publish.yml

@@ -1,16 +1,8 @@
-# This workflow will upload a Python Package using Twine when a release is created
-# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python#publishing-to-package-registries
-
-# This workflow uses actions that are not certified by GitHub.
-# They are provided by a third-party and are governed by
-# separate terms of service, privacy policy, and support
-# documentation.
-
 name: Upload Python Package
 
 on:
   push:
-    branches: [ "development_1.x" ]
+    branches: [ "development_2.1" ]
 
 permissions:
   contents: read
@@ -25,6 +17,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v3
       with:
+        fetch-depth: 0
         python-version: '3.x'
     - name: Install dependencies
       run: |
@@ -37,8 +30,8 @@ jobs:
       env:
         GITHUB_TOKEN: ${{ secrets.THIS_GITHUB_TOKEN }}
         DEFAULT_BRANCH: ${{ github.ref_name }}
-        DEFAULT_BUMP: minor
-        WITH_V: false
+        DEFAULT_BUMP: patch
+        WITH_V: true
     - name: Build and publish
       env:
         TWINE_USERNAME: __token__

grisera/auth/auth_bearer.py

@@ -1,7 +1,7 @@
 from fastapi import Request, HTTPException
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 
-from grisera.auth.auth_handler import decodeJWT, verify_jwt
+from grisera.auth.auth_handler import decode_jwt, verify_jwt
 
 
 class JWTBearer(HTTPBearer):
@@ -15,6 +15,6 @@ async def __call__(self, request: Request):
                 raise HTTPException(status_code=403, detail="Invalid authentication scheme.")
             if not verify_jwt(credentials.credentials):
                 raise HTTPException(status_code=403, detail="Invalid token or expired token.")
-            return decodeJWT(credentials.credentials)
+            return decode_jwt(credentials.credentials)
         else:
             raise HTTPException(status_code=403, detail="Invalid authorization code.")

grisera/auth/auth_config.py

@@ -1,4 +1,10 @@
 import os
 
-JWT_SECRET = os.environ.get("JWT_SECRET") or "jwtsecret"
-JWT_ALGORITHM = os.environ.get("JWT_ALGORITHM") or "HS256"
+KEYCLOAK_SERVER = os.environ.get("KEYCLOAK_SERVER") or "http://localhost:8090"
+REALM = os.environ.get("REALM") or "grisera"
+JWKS_URL = os.environ.get("JWKS_URL") or f"{KEYCLOAK_SERVER}/realms/{REALM}/protocol/openid-connect/certs"
+JWT_ALGORITHM = os.environ.get("JWT_ALGORITHM") or "RS256"
+CLIENT_ID = os.environ.get("CLIENT_ID") or "grisera-api"
+CLIENT_SECRET = os.environ.get("CLIENT_SECRET") or "6UkCrp7UqFy78vh5TVhkaYP0OuVagNTd"
+PERMISSIONS_ENDPOINT = os.environ.get("PERMISSIONS_ENDPOINT") or "http://localhost:8085/api/permissions"
+ISSUER = os.environ.get("ISSUER")

grisera/auth/auth_handler.py

@@ -1,25 +1,69 @@
 import time
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
 
 import jwt
 
-from grisera.auth.auth_config import JWT_SECRET, JWT_ALGORITHM
+from grisera.auth.auth_config import JWT_ALGORITHM, JWKS_URL, ISSUER
+import requests
 
 
-def verify_jwt(jwtoken: str) -> bool:
-    is_token_valid: bool = False
+def get_jwks():
+    response = requests.get(JWKS_URL)
+    response.raise_for_status()
+    return response.json()["keys"]
+
+
+def get_key_by_kid(kid: str):
+    jwks = get_jwks()
+    for key in jwks:
+        if key["kid"] == kid:
+            return key
+    raise ValueError("Key not found")
+
 
+def verify_jwt(jwtoken: str) -> bool:
     try:
-        payload = decodeJWT(jwtoken)
-    except:
-        payload = None
-    if payload:
-        is_token_valid = True
-    return is_token_valid
+        payload = decode_jwt(jwtoken)
+        return payload is not None
+    except Exception as e:
+        return False
+
+
+def construct_pem_key(key):
+    if key["kty"] != "RSA":
+        raise ValueError("Only RSA keys are supported")
 
+    exponent = int.from_bytes(jwt.utils.base64url_decode(key["e"]), "big")
+    modulus = int.from_bytes(jwt.utils.base64url_decode(key["n"]), "big")
 
-def decodeJWT(token: str) -> dict:
+    public_key = rsa.RSAPublicNumbers(exponent, modulus).public_key()
+    return public_key.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo)
+
+
+def decode_jwt(token: str) -> dict:
     try:
-        decoded_token = jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
-        return decoded_token if decoded_token["exp"] >= time.time() else None
+        unverified_header = jwt.get_unverified_header(token)
+        kid = unverified_header.get("kid")
+        if not kid:
+            raise ValueError("Token header missing 'kid'")
+
+        key = get_key_by_kid(kid)
+
+        public_key = construct_pem_key(key)
+
+        decoded_token = jwt.decode(
+            token,
+            public_key,
+            algorithms=[JWT_ALGORITHM],
+            verify_iss=ISSUER is not None,
+            issuer=ISSUER,
+        )
+
+        if decoded_token["exp"] < time.time():
+            raise ValueError("Token has expired")
+
+        return decoded_token
     except Exception as e:
-        return {}
+        print(f"Token verification failed: {e}")
+        return None

grisera/dataset/dataset_router.py

@@ -10,7 +10,7 @@
 from grisera.channel.channel_model import Types as channel_types
 from grisera.dataset.dataset_model import DatasetOut, DatasetsOut, DatasetIn
 from grisera.helpers.hateoas import get_links
-from grisera.helpers.helpers import check_dataset_permission
+from grisera.helpers.helpers import check_dataset_permission, get_permissions
 from grisera.life_activity.life_activity_model import LifeActivity as life_activity_types
 from grisera.measure_name.measure_name_model import MeasureName as measure_name_types, MeasureNameIn
 from grisera.measure.measure_model import Measure as measure_type, MeasureIn
@@ -103,7 +103,9 @@ async def get_datasets(self, response: Response, token=Depends(JWTBearer())):
         Get all datasets
         """
         dataset_ids = []
-        for permission in token['permissions']:
+        user_id = token['sub']
+        permissions = get_permissions(user_id)
+        for permission in permissions:
             dataset_ids.append(str(permission['datasetId']))
         get_response = self.dataset_service.get_datasets(dataset_ids)
         if get_response.errors is not None:

grisera/helpers/helpers.py

@@ -1,9 +1,12 @@
+import requests
 from fastapi import Request, Depends, HTTPException
 from typing import Union
 
 from grisera.auth.auth_bearer import JWTBearer
 from grisera.auth.auth_module import Roles
 
+from grisera.auth.auth_config import PERMISSIONS_ENDPOINT, KEYCLOAK_SERVER, REALM, CLIENT_ID, CLIENT_SECRET
+
 
 def create_stub_from_response(response, id_key='id', properties=None):
     if properties is None:
@@ -21,10 +24,50 @@ def create_stub_from_response(response, id_key='id', properties=None):
 
 
 def check_dataset_permission(request: Request, dataset_id: Union[int, str], token=Depends(JWTBearer())):
-    for permission in token['permissions']:
+    user_id = token['sub']
+    permissions = get_permissions(user_id)
+    for permission in permissions:
         if str(permission['datasetId']) == str(dataset_id):
             if (not request.method == "GET") and (str(permission['role']) == Roles.reader):
                 raise HTTPException(status_code=403, detail="Invalid permission level to dataset")
             return dataset_id
 
     raise HTTPException(status_code=403, detail="Invalid authentication to dataset")
+
+
+def get_permissions(user_id: Union[int, str]):
+    access_token = _getAccessToken()
+    return _request_permissions(access_token, user_id)
+
+
+def _getAccessToken():
+    url = f"{KEYCLOAK_SERVER}/realms/{REALM}/protocol/openid-connect/token"
+
+    payload = f'grant_type=client_credentials&client_id={CLIENT_ID}&client_secret={CLIENT_SECRET}'
+    headers = {
+        'Content-Type': 'application/x-www-form-urlencoded',
+    }
+
+    response = requests.request("POST", url, headers=headers, data=payload)
+
+    try:
+        response.raise_for_status()
+    except requests.exceptions.HTTPError as http_err:
+        print(f"Token request failure: {response.status_code}, {response.json()}")
+        raise http_err
+
+    return response.json()['access_token']
+
+
+def _request_permissions(access_token, user_id):
+    headers = {
+        "Authorization": f"Bearer {access_token}"
+    }
+    try:
+        response = requests.get(f'{PERMISSIONS_ENDPOINT}/{user_id}', headers=headers)
+        response.raise_for_status()
+        return response.json()
+    except requests.exceptions.HTTPError as http_err:
+        response = http_err.response
+        print(f"Request failure: {response.status_code}, {response.json()}")
+        return response.json()

requirements.txt

@@ -4,4 +4,6 @@ requests~=2.28.2
 fastapi-utils
 pydantic~=1.10.6
 starlette~=0.26.1
-pyjwt
+pyjwt~=2.10.1
+cryptography~=44.0.0
+setuptools~=75.6.0

setup.py

@@ -1,12 +1,12 @@
 from setuptools import setup, find_packages
 
-VERSION = '0.0.38.30'
+VERSION = '0.0.39.6'
 DESCRIPTION = 'Grisera-api package'
 LONG_DESCRIPTION = 'Graph Representation Integrating Signals for Emotion Recognition and Analysis (GRISERA) framework provides a persistent model for storing integrated signals and methods for its creation.'
 
 # Setting up
 setup(
-    name="grisera",
+    name="grisera2",
     version=VERSION,
     author="",
     author_email="",
@@ -21,7 +21,9 @@
         'fastapi-utils',
         'pydantic~=1.10.6',
         'starlette~=0.26.1',
-        'pyjwt'
+        'pyjwt',
+        'cryptography~=44.0.0',
+        'setuptools~=75.6.0'
     ],
     classifiers=[
         "Development Status :: 1 - Planning",