Baip 315/jwt token does not contain any accessible tenants

fleet_management_api/api_impl/constants.py

@@ -1,3 +1,7 @@
+"""Constants used in the Fleet Management API implementation, namely by the tenant-related modules."""
+
 AUTHORIZATION_HEADER_NAME = "Authorization"
 AUTHORIZATION_ENVIRONMENT_NAME = "HTTP_AUTHORIZATION"
-PAYLOAD_FIELD_NAME = "payload"
+TENANT_PAYLOAD_ITEM = (
+    "group"  # The name of the field in the JWT payload that contains the tenant information.
+)

fleet_management_api/api_impl/tenants.py

@@ -1,5 +1,4 @@
 from __future__ import annotations
-import json
 import dataclasses
 
 import jwt
@@ -9,7 +8,6 @@
 from fleet_management_api.api_impl.auth_controller import get_public_key
 from fleet_management_api.api_impl.constants import (
     AUTHORIZATION_HEADER_NAME as _AUTHORIZATION_HEADER_NAME,
-    PAYLOAD_FIELD_NAME as _PAYLOAD_FIELD_NAME,
 )
 
 
@@ -170,10 +168,10 @@ def get_accessible_tenants(
             status_code=200,
             tenants=tenants,
         )
-    except NoAccessibleTenants:
+    except NoAccessibleTenants as e:
         # If the JWT token does not contain any tenants, return an empty tenant object
         return LoadedAccessibleTenants(
-            msg="JWT token does not contain any accessible tenants.",
+            msg=f"JWT token does not contain any accessible tenants. Error: {str(e)}",
             status_code=401,
             tenants=NO_TENANTS,
         )
@@ -232,38 +230,12 @@ def _get_accessible_tenants_from_auth_headers(
         raise Unauthorized("No valid JWT token or API key provided.")
     if not key.strip():
         raise MissingRSAKey("RSA public key is not set.")
-    decoded_key = jwt.decode(bearer, key, [_ALGORITHM], audience=audience)
-
-    try:
-        payload = dict(json.loads(decoded_key[_PAYLOAD_FIELD_NAME]))
-    except KeyError:
-        raise NoAccessibleTenants(
-            "No tenants could be extracted from the token. Token is missing the payload."
-        )
-    except Exception as e:
-        raise Unauthorized("Invalid JWT token.") from e
-
-    group: list[str] = payload.get("group", [])
+    decoded_payload = jwt.decode(bearer, key, [_ALGORITHM], audience=audience)
+    if "group" not in decoded_payload:
+        raise NoAccessibleTenants("No item 'group' in token. Token does not contain tenants.")
+    group: list[str] = decoded_payload.get("group", [])
     tenants = [item.split("/")[-1] for item in group if item.startswith("/customers/")]
     tenants = [tenant for tenant in tenants if tenant]
     if not tenants:
-        raise NoAccessibleTenants("No item group in token. Token does not contain tenants.")
+        raise NoAccessibleTenants("No accessible tenants found in the token.")
     return tenants
-
-
-def encode_jwt_token(payload: dict, key: str) -> str:
-    """Encode a JWT token using the provided key."""
-    try:
-        return jwt.encode(payload, key, algorithm=_ALGORITHM)
-    except Exception as e:
-        raise Unauthorized("Failed to encode JWT token.") from e
-
-
-def decode_jwt_token(token: str, key: str) -> dict:
-    """Decode a JWT token using the provided key."""
-    try:
-        return jwt.decode(token, key, algorithms=[_ALGORITHM])
-    except jwt.ExpiredSignatureError:
-        raise Unauthorized("JWT token has expired.")
-    except jwt.InvalidTokenError:
-        raise Unauthorized("Invalid JWT token.")

fleet_management_api/app.py

@@ -21,7 +21,6 @@
 from fleet_management_api.api_impl.tenants import MissingRSAKey as _MissingRSAKey
 from fleet_management_api.api_impl.constants import (
     AUTHORIZATION_HEADER_NAME as _AUTHORIZATION_HEADER_NAME,
-    PAYLOAD_FIELD_NAME as _PAYLOAD_FIELD_NAME,
 )
 
 
@@ -33,9 +32,10 @@
 
 
 def get_token(*tenants: str) -> str:
-    tenant_list = ",".join(f'"/customers/{name}"' for name in tenants)
+    tenant_list = [f"/customers/{name}" for name in tenants]
     payload = {
-        _PAYLOAD_FIELD_NAME: '{{"group": [{tenant_list}]}}'.format(tenant_list=tenant_list),
+        "group": tenant_list,
+        "iss": "test",
         "aud": "account",
         "allowed-origins": ["test_client"],
     }
@@ -45,7 +45,7 @@ def get_token(*tenants: str) -> str:
     try:
         encoded = jwt.encode(payload, private, algorithm="RS256")
     except jwt.PyJWTError as e:
-        _log_error(f"Failed to encode JWT token: {str(e)}")
+        _log_error(f"Failed to encode JWsT token: {str(e)}")
         return ""
     return encoded
 

fleet_management_api/openapi/openapi.yaml

@@ -9,7 +9,7 @@ info:
     name: AGPLv3
     url: https://www.gnu.org/licenses/agpl-3.0.en.html
   title: BringAuto Fleet Management v2 API
-  version: 4.1.0
+  version: 4.1.1
 servers:
 - url: /v2/management
 security:

openapi/openapi.yaml

@@ -2,7 +2,7 @@ openapi: 3.0.0
 info:
   title: BringAuto Fleet Management v2 API
   description: Specification for BringAuto fleet backend HTTP API
-  version: 4.1.0
+  version: 4.1.1
   contact:
     name: BringAuto s.r.o
     url: https://bringauto.com

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fleet_management_api"
-version = "4.1.0"
+version = "4.1.1"
 
 [tool.setuptools.packages.find]
 include = ["fleet_management_api", "openapi", "config"]

tests/controllers/car/test_car_controller.py

@@ -1,7 +1,7 @@
 import unittest
+
 from fleet_management_api.models import Car, PlatformHW, Order, MobilePhone, Tenant
 import fleet_management_api.app as _app
-from fleet_management_api.api_impl.tenants import decode_jwt_token
 from fleet_management_api.database.db_access import delete
 from fleet_management_api.database.db_models import CarStateDB
 from fleet_management_api.logs import LOGGER_NAME
@@ -11,14 +11,15 @@
     clear_auth_params,
     clear_test_keys,
     get_test_public_key,
-    get_public_key,
 )
-
 from tests._utils.constants import TEST_TENANT_NAME
-from tests._utils.setup_utils import create_stops, create_platform_hws, create_route
-
+from tests._utils.setup_utils import (
+    create_stops,
+    create_platform_hws,
+    create_route,
+    TenantFromTokenMock,
+)
 import tests._utils.api_test as api_test
-from tests._utils.setup_utils import TenantFromTokenMock
 
 
 PHONE = MobilePhone(phone="123456789")

tests/security/test_tenants_from_jwt.py

@@ -17,8 +17,6 @@
 )
 from fleet_management_api.app import get_token, get_test_app
 from fleet_management_api.models import Tenant
-import fleet_management_api.database.db_access as _db_access
-from fleet_management_api.database.db_models import TenantDB
 from fleet_management_api.models import PlatformHW
 from fleet_management_api.api_impl.load_request import _LoadedRequestEmpty as _RequestEmpty
 import tests._utils.api_test as api_test