[DOP-23149] Replace python-jose with pyjwt

data_rentgen/server/settings/auth/jwt.py

@@ -32,7 +32,7 @@ class JWTSettings(BaseModel):
             """
             Algorithm used for signing JWT tokens.
 
-            See `python-jose <https://python-jose.readthedocs.io/en/latest/jws/index.html#supported-algorithms>`_
+            See `pyjwt <https://pyjwt.readthedocs.io/en/latest/algorithms.html>`_
             documentation.
             """,
         ),

data_rentgen/server/utils/jwt.py

@@ -1,28 +1,25 @@
 # SPDX-FileCopyrightText: 2024 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
-from jose import ExpiredSignatureError, JWTError, jwt
+import jwt
 
 from data_rentgen.exceptions.auth import AuthorizationError
 
 
 def sign_jwt(payload: dict, secret_key: str, security_algorithm: str) -> str:
     return jwt.encode(
-        payload,
-        secret_key,
+        payload=payload,
+        key=secret_key,
         algorithm=security_algorithm,
     )
 
 
 def decode_jwt(token: str, secret_key: str, security_algorithm: str) -> dict:
     try:
-        result = jwt.decode(
-            token,
-            secret_key,
-            algorithms=[security_algorithm],
-        )
-        if "exp" not in result:
-            raise ExpiredSignatureError("Missing expiration time in token")
+        claims = jwt.decode(jwt=token, key=secret_key, algorithms=[security_algorithm])
 
-        return result
-    except JWTError as e:
+        if "exp" not in claims:
+            raise jwt.exceptions.ExpiredSignatureError("Missing expiration time in token")
+
+        return claims
+    except jwt.exceptions.InvalidTokenError as e:
         raise AuthorizationError("Invalid token") from e

pyproject.toml

@@ -68,11 +68,10 @@ faststream = {extras = ["kafka", "cli"], version = "^0.5.33", optional = true}
 # used by aiokafka for LZ4, Snappy, ZSTD compression algorithms
 cramjam = {version = "^2.9.1", optional = true}
 packaging = {version = "^24.2", optional = true}
-python-jose = {version = "^3.3.0", optional = true}
 itsdangerous = {version = "^2.2.0", optional = true}
 python-multipart = {version = "^0.0.20", optional = true}
-types-python-jose = {version = "^3.3.4.20240106", optional = true}
 python-keycloak = {version = "^5.1.1", optional = true}
+pyjwt = {version = "^2.10.1 ", optional = true}
 
 [tool.poetry.extras]
 server = [
@@ -92,10 +91,9 @@ server = [
   "uuid6",
   "python-dateutil",
   "packaging",
-  "python-jose",
+  "pyjwt",
   "itsdangerous",
   "python-multipart",
-  "types-python-jose",
   "python-keycloak",
 ]
 consumer = [

tests/test_server/fixtures/keycloak.py

@@ -8,7 +8,7 @@
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
 from itsdangerous import TimestampSigner
-from jose import jwt
+import jwt
 
 
 @pytest.fixture(scope="session")
@@ -60,7 +60,11 @@ def _create_session_cookie(user, expire_in_msec=5000) -> str:
             "exp": int(time.time()) + (expire_in_msec / 1000),
         }
 
-        access_token = jwt.encode(payload, private_pem, algorithm="RS256")
+        access_token = jwt.encode(
+            payload=payload,
+            key=private_pem,
+            algorithm="RS256",
+        )
         refresh_token = "mock_refresh_token"
 
         session_data = {
@@ -127,11 +131,10 @@ def mock_keycloak_token_refresh(user, server_app_settings, rsa_keys):
     server_url = server_app_settings.auth.dict()["keycloak"]["server_url"]
     realm_name = server_app_settings.auth.dict()["keycloak"]["client_id"]
     token_url = f"{server_url}/realms/{realm_name}/protocol/openid-connect/token"
-    private_pem = rsa_keys["private_pem"]
 
     # generate new access and refresh tokens
-    expires_in = int(time.time()) + 1000
-
+    expires_in = int(time.time()) + 5000
+    private_pem = rsa_keys["private_pem"]
     payload = {
         "sub": str(user.id),
         "preferred_username": user.name,
@@ -142,7 +145,11 @@ def mock_keycloak_token_refresh(user, server_app_settings, rsa_keys):
         "exp": expires_in,
     }
 
-    new_access_token = jwt.encode(payload, private_pem, algorithm="RS256")
+    new_access_token = jwt.encode(
+        payload=payload,
+        key=private_pem,
+        algorithm="RS256",
+    )
     new_refresh_token = "mock_new_refresh_token"
 
     responses.add(