""" AUTHOR: Khushal P Soonderji DATE: Wednesday, 19th Jun., 2025 OBJECTIVE: To generate and verify OTPs that expire. REFERENCES: N/A DOWNLOADS: N/A NOTES: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # For using Quart: from quart import Blueprint, current_app # Common: from shared import constants # My utils: from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.security.otp import HashedOTP from utils_v2.api.response import ResponseModel from utils_v2.api.codes import StatusCodes, HttpCodes from utils_v2.api.async_quart import ( log_request_to_mongo, only_whitelisted_ips, should_not_be_under_maintenance, read_input, limit_rate, validate_input, set_api_version, log_chain_to_mongo, handle_cancelled_request ) # Data models: from models.otp.timed_otp import GenerateTimedOTP, VerifyTimedOTP # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Related to Quart: otp_bp = Blueprint("otp", __name__) # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** @otp_bp.record_once def init(blueprint_setup_state): # This gets called when the blueprint is registered. # Consider this to be a one-time setup for the whole blueprint: pass # --------------------------------------------------------------------------------------------------------------------- @otp_bp.route("/timed/generate", methods = ["POST", "GET"]) @set_api_version(api_version = "1.0.0") @read_input(sanitize_headers = True, sanitize_data = True) @log_request_to_mongo( attr_name = "logs_mongo", project = constants.PROJECT_NAME, log_type = "timedOTP", operation = "generate", log_input = True, log_output = True, sensitive_keys = None ) @log_chain_to_mongo(attr_name = "logs_mongo") @should_not_be_under_maintenance(attr_name = "is_under_maintenance") # @only_whitelisted_ips(attr_name = "whitelisted_ips") @validate_input(data_validator = lambda x: GenerateTimedOTP(**x)) # @limit_rate( # attr_name = "otp_redis", # rate_limit = 1, # seconds = 2, # data_keys = ["id"], # allow_if_exception = False # ) @handle_cancelled_request() async def generate_otp( inbound_headers: dict = None, inbound_data: dict | GenerateTimedOTP = None, inbound_files: dict = None, log_id: str = None, **kwargs ): """ Generates one OTP for the amount of time specified in the request. :param inbound_headers: auto-extracted by the decorators from 'async_quart.py'. :param inbound_data: auto-extracted by the decorators from 'async_quart.py'. :param inbound_files: auto-extracted by the decorators from 'async_quart.py'. :param log_id: An identifier for the logs (if logging is enabled). :return: A standard response structure from the function in 'async_quart.py'. """ # Generate the OTP: otp_key = current_app.otp_redis.make_key(str(inbound_data.id)) otp_client = HashedOTP(secret = HashedOTP.generate_secret()) otp = otp_client.generate_otp(count = 0) # Delete any existing OTP with the same identifiers: await current_app.otp_redis.delete(key = otp_key) # Store the OTP in Redis: otp_stored = await current_app.otp_redis.set( key = otp_key, value = { "otp": otp, "att": inbound_data.attempts, "iat": date_time.get_current_utc_date_time().timestamp(), "sec": inbound_data.seconds }, expiry = inbound_data.seconds ) # Done here: return ResponseModel( status_code = StatusCodes.OK if otp_stored else StatusCodes.FAILED, data = {"otp": otp} if otp_stored else None ) # --------------------------------------------------------------------------------------------------------------------- @otp_bp.route("/timed/verify", methods = ["POST", "GET"]) @set_api_version(api_version = "1.0.0") @read_input(sanitize_headers = True, sanitize_data = True) @log_request_to_mongo( attr_name = "logs_mongo", project = constants.PROJECT_NAME, log_type = "timedOTP", operation = "verify", log_input = True, log_output = True, sensitive_keys = None ) @log_chain_to_mongo(attr_name = "logs_mongo") @should_not_be_under_maintenance(attr_name = "is_under_maintenance") # @only_whitelisted_ips(attr_name = "whitelisted_ips") @validate_input(data_validator = lambda x: VerifyTimedOTP(**x)) # @limit_rate( # attr_name = "otp_redis", # rate_limit = 1, # seconds = 2, # data_keys = ["id"], # allow_if_exception = False, # count_for_http_codes = [200] # ) @handle_cancelled_request() async def verify_otp( inbound_headers: dict = None, inbound_data: dict | VerifyTimedOTP = None, inbound_files: dict = None, log_id: str = None, **kwargs ): """ Verifies the claimed OTP against the stored OTP. :param inbound_headers: auto-extracted by the decorators from 'async_quart.py'. :param inbound_data: auto-extracted by the decorators from 'async_quart.py'. :param inbound_files: auto-extracted by the decorators from 'async_quart.py'. :param log_id: An identifier for the logs (if logging is enabled). :return: A standard response structure from the function in 'async_quart.py'. """ # Start by assuming failure: is_valid = False attempts_left = None # Fetch the OTP from Redis: otp_key = current_app.otp_redis.make_key(str(inbound_data.id)) stored_otp = await current_app.otp_redis.get(key = otp_key) # Test the validity of the OTP: if stored_otp is not None: # Delete the record from the cache.: await current_app.otp_redis.delete(key = otp_key) # Make note of the attempts left. # We reduce the count by one straightaway: attempts_left = stored_otp["att"] - 1 # If the stored OTP matches the claimed OTP, # we note down the acceptance and delete the record from the cache: if inbound_data.otp == stored_otp["otp"]: is_valid = True attempts_left = 0 # If the stored OTP and the claimed OTP don't match, # we reduce the attempt count and : elif attempts_left > 0: # Update the OTP info in the cache: otp_updated = await current_app.otp_redis.set( key = otp_key, value = { "otp": stored_otp["otp"], "att": attempts_left, "iat": stored_otp["iat"], "sec": stored_otp["sec"] }, expiry = (stored_otp["iat"] + stored_otp["sec"]) - date_time.get_current_utc_date_time().timestamp() ) # If the update failed: if not otp_updated: attempts_left = None # Done here: return ResponseModel( status_code = StatusCodes.OK if is_valid else StatusCodes.FAILED, data = { "isValid": is_valid, "attemptsLeft": attempts_left }, http_code = HttpCodes.UNAUTHORIZED ) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass