166 lines
6.5 KiB
Python
166 lines
6.5 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Friday, 13th Dec., 2024
|
|
|
|
OBJECTIVE:
|
|
|
|
To validate if a token id belongs to a user based on the session token sent by the user.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# To use Quart:
|
|
from quart import current_app
|
|
|
|
# My utils:
|
|
from utils_v2.database.async_mongo_v2 import AsyncMongo
|
|
|
|
# To work with MongoDB:
|
|
from bson.objectid import ObjectId
|
|
|
|
# To work with datatypes:
|
|
from typing import List
|
|
|
|
# The data model:
|
|
from models.core.user import CoreUserInfoModel
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
async def is_authorized(
|
|
mongo_conn: AsyncMongo,
|
|
user_info: CoreUserInfoModel | dict,
|
|
token_ids: ObjectId | str | List[ObjectId | str]
|
|
) -> bool:
|
|
|
|
"""
|
|
To verify if the token id is owned by the user (who will be initially identified from his session token).
|
|
:param mongo_conn: The instance of the MongoDB connector to use to perform this action.
|
|
:param user_info: The user trying to access some service.
|
|
:param token_ids: One or more token ids that the user is claiming to own.
|
|
:return: True if the user is authorized to use this token, else False.
|
|
"""
|
|
|
|
# If the user info was never available:
|
|
if user_info is None: return False
|
|
|
|
# Start by assuming the access is authorized,
|
|
# and pre-process the inputs:
|
|
authorized = True
|
|
if not isinstance(token_ids, list): token_ids = [token_ids]
|
|
if isinstance(user_info, dict): user_info = CoreUserInfoModel(**user_info)
|
|
|
|
# Fetch the auth tokens from the database:
|
|
# Retrieve the document of the token id:
|
|
auth_tokens = await current_app.core_auth_token_controller.get_tokens(
|
|
mongo_conn = mongo_conn,
|
|
token_ids = [ObjectId(t) for t in token_ids]
|
|
)
|
|
|
|
# Check all the token ids:
|
|
for auth_token in auth_tokens:
|
|
if user_info.billingAccountId != auth_token.user.billingAccountId:
|
|
authorized = False
|
|
break
|
|
|
|
# Done here:
|
|
return authorized
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
async def is_not_authorized(
|
|
mongo_conn: AsyncMongo,
|
|
user_info: CoreUserInfoModel,
|
|
token_ids: ObjectId | str | List[ObjectId | str]
|
|
) -> bool:
|
|
|
|
"""
|
|
Just a wrapper around the above function to improve readability.
|
|
:param mongo_conn: The instance of the MongoDB connector to use to perform this action.
|
|
:param user_info: The user trying to access some service.
|
|
:param token_ids: One or more token ids that the user is claiming to own.
|
|
:return: True if the user is authorized to use this token, else False.
|
|
"""
|
|
|
|
authorized = await is_authorized(
|
|
mongo_conn = mongo_conn,
|
|
user_info = user_info,
|
|
token_ids = token_ids
|
|
)
|
|
|
|
return not authorized
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
pass
|