updated last web service

ServicesApi/Extensions/Middlewares/token_middleware.py

@@ -0,0 +1,19 @@
+from fastapi import Request, status
+from fastapi.responses import JSONResponse
+from config import api_config
+
+from endpoints.routes import get_safe_endpoint_urls
+
+
+async def token_middleware(request: Request, call_next):
+    base_url = request.url.path
+    safe_endpoints = [_[0] for _ in get_safe_endpoint_urls()]
+    if base_url in safe_endpoints:
+        return await call_next(request)
+
+    token = request.headers.get(api_config.ACCESS_TOKEN_TAG, None)
+    if not token:
+        return JSONResponse(content={"error": "EYS_0002"}, status_code=status.HTTP_401_UNAUTHORIZED)
+
+    response = await call_next(request)
+    return response

ServicesApi/Extensions/Middlewares/token_provider.py

@@ -0,0 +1,96 @@
+import enum
+
+from typing import Optional, Union, Dict, Any, List
+from pydantic import BaseModel
+
+from api_controllers.redis.database import RedisActions
+from api_validations.token.validations import (
+    TokenDictType,
+    OccupantTokenObject,
+    EmployeeTokenObject,
+    UserType,
+)
+
+class TokenProvider:
+
+    AUTH_TOKEN: str = "AUTH_TOKEN"
+
+    @classmethod
+    def convert_redis_object_to_token(cls, redis_object: Dict[str, Any]) -> TokenDictType:
+        """
+        Process Redis object and return appropriate token object.
+        """
+        if redis_object.get("user_type") == UserType.employee.value:
+            return EmployeeTokenObject(**redis_object)
+        elif redis_object.get("user_type") == UserType.occupant.value:
+            return OccupantTokenObject(**redis_object)
+        raise ValueError("Invalid user type")
+
+
+    @classmethod
+    def get_login_token_from_redis(
+        cls, token: Optional[str] = None, user_uu_id: Optional[str] = None
+    ) -> Union[TokenDictType, List[TokenDictType]]:
+        """
+        Retrieve token object from Redis using token and user_uu_id
+        """
+        token_to_use, user_uu_id_to_use = token or "*", user_uu_id or "*"
+        list_of_token_dict, auth_key_list = [], [cls.AUTH_TOKEN, token_to_use, user_uu_id_to_use]
+        if token:
+            result = RedisActions.get_json(list_keys=auth_key_list, limit=1)
+            if first_record := result.first:
+                return cls.convert_redis_object_to_token(first_record)
+        elif user_uu_id:
+            result = RedisActions.get_json(list_keys=auth_key_list)
+            if all_records := result.all:
+                for all_record in all_records:
+                    list_of_token_dict.append(cls.convert_redis_object_to_token(all_record))
+                return list_of_token_dict
+        raise ValueError("Token not found in Redis. Please check the token or user_uu_id.")
+
+    @classmethod
+    def get_dict_from_redis(
+        cls, token: Optional[str] = None, user_uu_id: Optional[str] = None
+    ) -> Union[TokenDictType, List[TokenDictType]]:
+        """
+        Retrieve token object from Redis using token and user_uu_id
+        """
+        token_to_use, user_uu_id_to_use = token or "*", user_uu_id or "*"
+        list_of_token_dict, auth_key_list = [], [cls.AUTH_TOKEN, token_to_use, user_uu_id_to_use, "*"]
+        if token:
+            result = RedisActions.get_json(list_keys=auth_key_list, limit=1)
+            if first_record := result.first:
+                return cls.convert_redis_object_to_token(first_record)
+        elif user_uu_id:
+            result = RedisActions.get_json(list_keys=auth_key_list)
+            if all_records := result.all:
+                for all_record in all_records:
+                    list_of_token_dict.append(cls.convert_redis_object_to_token(all_record))
+                return list_of_token_dict
+        raise ValueError("Token not found in Redis. Please check the token or user_uu_id.")
+
+    @classmethod
+    def retrieve_application_codes(cls, page_url: str, token: TokenDictType):
+        """
+        Retrieve application code from the token object or list of token objects.
+        """
+        if isinstance(token, EmployeeTokenObject):
+            if application_codes := token.selected_company.reachable_app_codes.get(page_url, None):
+                return application_codes
+        elif isinstance(token, OccupantTokenObject):
+            if application_codes := token.selected_occupant.reachable_app_codes.get(page_url, None):
+                return application_codes
+        raise ValueError("Invalid token type or no application code found.")
+
+    @classmethod
+    def retrieve_event_codes(cls, endpoint_code: str, token: TokenDictType) -> str:
+        """
+        Retrieve event code from the token object or list of token objects.
+        """
+        if isinstance(token, EmployeeTokenObject):
+            if event_codes := token.selected_company.reachable_event_codes.get(endpoint_code, None):
+                return event_codes
+        elif isinstance(token, OccupantTokenObject):
+            if event_codes := token.selected_occupant.reachable_event_codes.get(endpoint_code, None):
+                return event_codes
+        raise ValueError("Invalid token type or no event code found.")

ServicesApi/Extensions/Token/config.py

@@ -0,0 +1,15 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Configs(BaseSettings):
+    """
+    ApiTemplate configuration settings.
+    """
+
+    ACCESS_TOKEN_LENGTH: int = 90
+    REFRESHER_TOKEN_LENGTH: int = 144
+
+    model_config = SettingsConfigDict(env_prefix="API_")
+
+
+token_config = Configs()

ServicesApi/Extensions/Token/password_module.py

@@ -0,0 +1,40 @@
+import hashlib
+import uuid
+import secrets
+import random
+
+from .config import token_config
+
+
+class PasswordModule:
+
+    @staticmethod
+    def generate_random_uu_id(str_std: bool = True):
+        return str(uuid.uuid4()) if str_std else uuid.uuid4()
+
+    @staticmethod
+    def generate_token(length=32) -> str:
+        letters = "abcdefghijklmnopqrstuvwxyz"
+        merged_letters = [letter for letter in letters] + [letter.upper() for letter in letters]
+        token_generated = secrets.token_urlsafe(length)
+        for i in str(token_generated):
+            if i not in merged_letters:
+                token_generated = token_generated.replace(i, random.choice(merged_letters), 1)
+            return token_generated
+        raise ValueError("EYS_0004")
+
+    @classmethod
+    def generate_access_token(cls) -> str:
+        return cls.generate_token(int(token_config.ACCESS_TOKEN_LENGTH))
+
+    @classmethod
+    def generate_refresher_token(cls) -> str:
+        return cls.generate_token(int(token_config.REFRESHER_TOKEN_LENGTH))
+
+    @staticmethod
+    def create_hashed_password(domain: str, id_: str, password: str) -> str:
+        return hashlib.sha256(f"{domain}:{id_}:{password}".encode("utf-8")).hexdigest()
+
+    @classmethod
+    def check_password(cls, domain, id_, password, password_hashed) -> bool:
+        return cls.create_hashed_password(domain, id_, password) == password_hashed