client frontend tested
This commit is contained in:
@@ -20,7 +20,7 @@ from endpoints.index import endpoints_index
|
||||
|
||||
from api_validations.defaults.validations import CommonHeaders
|
||||
from api_middlewares.token_provider import TokenProvider
|
||||
|
||||
from events.auth.events import LoginHandler
|
||||
|
||||
auth_route = APIRouter(prefix="/authentication", tags=["Authentication Cluster"])
|
||||
|
||||
@@ -124,8 +124,12 @@ auth_route_check_token = "AuthCheckToken"
|
||||
)
|
||||
def check_token(headers: CommonHeaders = Depends(CommonHeaders.as_dependency)):
|
||||
"""Check if token is valid"""
|
||||
token_object = TokenProvider.get_dict_from_redis(token=headers.token)
|
||||
return None
|
||||
try:
|
||||
if token_object := LoginHandler.authentication_check_token_valid(access_token=headers.token, domain=headers.domain):
|
||||
return JSONResponse(status_code=status.HTTP_200_OK, content={"success": True})
|
||||
except Exception as e:
|
||||
print(e)
|
||||
return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={"success": False})
|
||||
|
||||
|
||||
auth_route_refresh_token = "AuthRefreshToken"
|
||||
|
||||
@@ -402,14 +402,7 @@ class LoginHandler:
|
||||
def authentication_check_token_valid(cls, domain, access_token: str) -> bool:
|
||||
redis_handler = RedisHandlers()
|
||||
if auth_token := redis_handler.get_object_from_redis(access_token=access_token):
|
||||
if auth_token.is_employee:
|
||||
if domain not in auth_token.domain_list:
|
||||
raise ValueError("EYS_00112")
|
||||
return True
|
||||
elif auth_token.is_occupant:
|
||||
if domain not in auth_token.domain_list:
|
||||
raise ValueError("EYS_00113")
|
||||
return True
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
|
||||
@@ -26,10 +26,32 @@ class TokenProvider:
|
||||
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]]:
|
||||
) -> Union[TokenDictType, List[TokenDictType]]:
|
||||
"""
|
||||
Retrieve token object from Redis using token and user_uu_id
|
||||
"""
|
||||
|
||||
Reference in New Issue
Block a user