From d16fa61ec3450a33261ac8689a6d0d31ab4274d5 Mon Sep 17 00:00:00 2001 From: khushal Date: Thu, 20 Feb 2025 16:26:43 +0530 Subject: [PATCH] (20250220) --- api/blueprints/common/otp/timed_otp.py | 340 +++++++++++++++---------- api/main.py | 14 +- controllers_v2/common/otp/timed_otp.py | 68 ++--- controllers_v2/core/auth_token.py | 35 +++ models/api/common/otp/__init__.py | 0 models/api/common/otp/timed_otp.py | 209 +++++++++++++++ models/common/otp/timed_otp.py | 4 +- 7 files changed, 475 insertions(+), 195 deletions(-) create mode 100644 models/api/common/otp/__init__.py create mode 100644 models/api/common/otp/timed_otp.py diff --git a/api/blueprints/common/otp/timed_otp.py b/api/blueprints/common/otp/timed_otp.py index 679d10b..5c7e6d0 100644 --- a/api/blueprints/common/otp/timed_otp.py +++ b/api/blueprints/common/otp/timed_otp.py @@ -26,8 +26,7 @@ N/A """ - - +import copy # ***************************************************************************************************************** # ***** **** # *** IMPORT *** @@ -53,15 +52,31 @@ 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, + set_api_version, read_input, - limit_rate, validate_input + get_session_info, + log_request_to_mongo, + log_chain_to_mongo, + should_not_be_under_maintenance, + only_whitelisted_ips, + limit_rate, + validate_input, + handle_cancelled_request, + messages_from_pydantic_exception ) # Data models: +from models.core.user import CoreUserInfoModel +from models.api.common.otp.timed_otp import ( + TimedOTPRequestHeaders, + SendTimedOTPSMSFromSavvyBulkSMSKenyaRequestData, + VerifyTimedOTPRequestData +) from models.common.otp.timed_otp import GenerateTimedOTP, VerifyTimedOTP +from models.message.sms.send import SavvyBulkSMSKenyaMessage + +# Helpers: +from api.helpers.user import token_check # For asynchronous activities: import asyncio @@ -76,7 +91,6 @@ import asyncio # Related to Quart: timed_otp_bp = Blueprint("timed_otp", __name__) -api_version = "2.0.0" # ***************************************************************************************************************** @@ -107,177 +121,231 @@ def init(blueprint_setup_state): # --------------------------------------------------------------------------------------------------------------------- -@timed_otp_bp.route("/generate", methods = ["POST", "GET"]) +@timed_otp_bp.route("/send/sms/kenya/savvybulksms", methods = ["POST"]) +@set_api_version(api_version = "1.0.0") @read_input(sanitize_headers = True, sanitize_data = True) +@get_session_info(key = "X-Session-Token", session_coro = "get_session") @log_request_to_mongo( - attr_name = "mongo", - log_type = "otp", - operation = "generateOtp", - api_version = api_version, + attr_name = "logs_mongo", + project = constants.PROJECT_NAME, + log_type = constants.MODULE_NAME, + operation = "otpSmsSendApi", log_input = True, - log_output = True + log_output = True, + sensitive_keys = ["sessionToken", "X-Session-Token", "tokenKey"] ) +@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 = "redis_cache", -# rate_limit = 1, -# seconds = 2, -# data_keys = ["id"], -# allow_if_exception = False -# ) +@validate_input( + header_validator = lambda x: TimedOTPRequestHeaders(**x).model_dump(), + data_validator = lambda x: SendTimedOTPSMSFromSavvyBulkSMSKenyaRequestData(**x) +) +@handle_cancelled_request() async def generate_otp( - inbound_headers: dict = None, - inbound_data: dict | GenerateTimedOTP = None, + inbound_headers: dict | TimedOTPRequestHeaders = None, + inbound_data: dict | SendTimedOTPSMSFromSavvyBulkSMSKenyaRequestData = 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'. + :param inbound_headers: auto-extracted by the decorators. + :param inbound_data: auto-extracted by the decorators. + :param inbound_files: auto-extracted by the decorators. + :param kwargs: Any number of extra inputs supplied by the decorators. + :return: A standard response structure. """ - try: + # ┏┓ ┓ ┏┓┓ ┓ + # ┣┫┓┏╋┣┓ ┃ ┣┓┏┓┏┃┏ + # ┛┗┗┻┗┛┗ ┗┛┛┗┗ ┗┛┗ - # Generate the OTP: - otp_key = current_app.redis_cache.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.redis_cache.delete(key = otp_key) - - # Store the OTP in Redis: - otp_stored = await current_app.redis_cache.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: + # Either the session must be valid, or + # the IP address requesting the service must be whitelisted: + if ( + kwargs.get("session_info") is None and + inbound_headers["Remote-IP"] not in current_app.whitelisted_ips + ): return ResponseModel( - api_version = api_version, - status_code = StatusCodes.OK if otp_stored else StatusCodes.FAILED, - data = {"otp": otp} if otp_stored else None + status_code = StatusCodes.FAILED, + http_code = HttpCodes.UNAUTHORIZED, + message = "Invalid session and/or bad IP addr." ) - # In case the client terminates the connection prematurely: - except asyncio.CancelledError as exception: - current_app.printer(exception) - return ResponseModel(api_version = api_version, status_code = StatusCodes.CLIENT_CLOSED_REQUEST) + # Get the user's info: + user_info = CoreUserInfoModel(**kwargs["session_info"]) + + # ┏┓ ┏┓┳┳┓┏┓ ┏┓ + # ┃┓┏┓╋ ┏┓┏┓ ┗┓┃┃┃┗┓ ┣┫┏┏┏┓┓┏┏┓╋ + # ┗┛┗ ┗ ┗┻┛┗ ┗┛┛ ┗┗┛ ┛┗┗┗┗┛┗┻┛┗┗ + + sms_auth_token = None + + # If a token key has been given, we retrieve that account: + if inbound_data.tokenKey is not None: + sms_auth_token = await current_app.savvy_bulk_sms_kenya_controller.get_token_from_key( + mongo_data_conn = current_app.data_mongo, + token_key = inbound_data.tokenKey, + must_be_active = True, + additional_filter = token_check.get_authorization_filter(user_info = user_info) + ) + + # Otherwise, we fetch the first SMS client account: + else: + sms_auth_token = await current_app.savvy_bulk_sms_kenya_controller.get_first_token( + mongo_data_conn = current_app.data_mongo, + must_be_active = True, + additional_filter = token_check.get_authorization_filter(user_info = user_info) + ) + + # If we found no account: + if not sms_auth_token: return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.NOT_FOUND, + message = "No SMS integration account found." + ) + + # ┏┓ ┓ ┓ ┏┓┏┳┓┏┓ + # ┗┓┏┓┏┓┏┫ ╋┣┓┏┓ ┃┃ ┃ ┃┃ + # ┗┛┗ ┛┗┗┻ ┗┛┗┗ ┗┛ ┻ ┣┛ + + # Add the user's info to the 'id' map: + otp_id = copy.deepcopy(inbound_data.id) + otp_id["user"] = kwargs["session_info"] + + # Generate an OTP: + otp_response = await current_app.timed_otp_controller.generate( + redis_cache = current_app.module_cache, + inbound_data = GenerateTimedOTP( + id = otp_id, + attempts = inbound_data.attempts, + seconds = inbound_data.seconds + ) + ) + + # If the OTP generation failed: + if not otp_response.success: return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.INTERNAL_SERVER_ERROR, + message = otp_response.message + ) + + # Format the message: + sms_message = SavvyBulkSMSKenyaMessage( + recipientNo = inbound_data.recipientNo, + text = inbound_data.smsText.replace(inbound_data.smsReplace, str(otp_response.data)) + ) + + # Send the message: + sms_results = await current_app.savvy_bulk_sms_kenya_controller.send_many_sms( + mongo_data_conn = current_app.data_mongo, + auth_token = sms_auth_token, + messages = [sms_message], + tags = inbound_data.tags or [] + ) + sms_success = True if sms_results.successCount == sms_results.totalCount else False + + # ┳┓ + # ┣┫┏┓┏┏┓┏┓┏┓┏┏┓ + # ┛┗┗ ┛┣┛┗┛┛┗┛┗ + # ┛ + + # Done here: + return ResponseModel( + status_code = StatusCodes.OK if sms_success else StatusCodes.FAILED, + http_code = HttpCodes.SUCCESS if sms_success else HttpCodes.INTERNAL_SERVER_ERROR, + message = "OTP sent successfully." if sms_success else "OTP NOT sent." + ) # --------------------------------------------------------------------------------------------------------------------- @timed_otp_bp.route("/verify", methods = ["POST", "GET"]) +@set_api_version(api_version = "1.0.0") @read_input(sanitize_headers = True, sanitize_data = True) +@get_session_info(key = "X-Session-Token", session_coro = "get_session") @log_request_to_mongo( - attr_name = "mongo", - log_type = "otp", - operation = "verifyOtp", - api_version = api_version, + attr_name = "logs_mongo", + project = constants.PROJECT_NAME, + log_type = constants.MODULE_NAME, + operation = "otpVerifyApi", log_input = True, - log_output = True + log_output = True, + sensitive_keys = ["sessionToken", "X-Session-Token", "tokenKey"] ) +@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 = "redis_cache", -# rate_limit = 1, -# seconds = 2, -# data_keys = ["id"], -# allow_if_exception = False, -# count_for_http_codes = [200] -# ) +@validate_input( + header_validator = lambda x: TimedOTPRequestHeaders(**x).model_dump(), + data_validator = lambda x: VerifyTimedOTPRequestData(**x) +) +@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'. + :param inbound_headers: auto-extracted by the decorators. + :param inbound_data: auto-extracted by the decorators. + :param inbound_files: auto-extracted by the decorators. + :param kwargs: Any number of extra inputs supplied by the decorators. + :return: A standard response structure. """ - try: + # ┏┓ ┓ ┏┓┓ ┓ + # ┣┫┓┏╋┣┓ ┃ ┣┓┏┓┏┃┏ + # ┛┗┗┻┗┛┗ ┗┛┛┗┗ ┗┛┗ - # Start by assuming failure: - is_valid = False - attempts_left = None - - # Fetch the OTP from Redis: - otp_key = current_app.redis_cache.make_key(str(inbound_data.id)) - stored_otp = await current_app.redis_cache.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.redis_cache.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.redis_cache.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: + # Either the session must be valid, or + # the IP address requesting the service must be whitelisted: + if ( + kwargs.get("session_info") is None and + inbound_headers["Remote-IP"] not in current_app.whitelisted_ips + ): return ResponseModel( - api_version = api_version, - status_code = StatusCodes.OK if is_valid else StatusCodes.FAILED, - data = { - "isValid": is_valid, - "attemptsLeft": attempts_left - }, - http_code = HttpCodes.UNAUTHORIZED + status_code = StatusCodes.FAILED, + http_code = HttpCodes.UNAUTHORIZED, + message = "Invalid session and/or bad IP addr." ) - # In case the client terminates the connection prematurely: - except asyncio.CancelledError as exception: - current_app.printer(exception) - return ResponseModel(api_version = api_version, status_code = StatusCodes.CLIENT_CLOSED_REQUEST) + # Get the user's info: + user_info = CoreUserInfoModel(**kwargs["session_info"]) + + # ┓┏ •┏ ┓ ┏┓┏┳┓┏┓ + # ┃┃┏┓┏┓┓╋┓┏ ╋┣┓┏┓ ┃┃ ┃ ┃┃ + # ┗┛┗ ┛ ┗┛┗┫ ┗┛┗┗ ┗┛ ┻ ┣┛ + # ┛ + + # Add the user's info to the 'id' map: + otp_id = copy.deepcopy(inbound_data.id) + otp_id["user"] = kwargs["session_info"] + + # Perform actual verification: + otp_response = await current_app.timed_otp_controller.verify( + redis_cache = current_app.module_cache, + inbound_data = VerifyTimedOTP( + id = otp_id, + otp = inbound_data.otp + ) + ) + + # ┳┓ + # ┣┫┏┓┏┏┓┏┓┏┓┏┏┓ + # ┛┗┗ ┛┣┛┗┛┛┗┛┗ + # ┛ + + # Done here: + return ResponseModel( + status_code = StatusCodes.OK if otp_response.success else StatusCodes.FAILED, + http_code = HttpCodes.SUCCESS if otp_response.success else HttpCodes.UNAUTHORIZED, + message = otp_response.message + ) # ***************************************************************************************************************** diff --git a/api/main.py b/api/main.py index f68a077..5492bef 100644 --- a/api/main.py +++ b/api/main.py @@ -101,6 +101,8 @@ from controllers_v2.finstitutions.payments.safaricom_mpesa_express import Safari from controllers_v2.software.mikrotik.all_mikrotik import AllMikroTikController from controllers_v2.software.mikrotik.mikrotik_pppoe_1000 import MikroTikPPPoE1000Controller from controllers_v2.software.mikrotik.mikrotik_hostpot_1000 import MikroTikHotspot1000Controller +# --- +from controllers_v2.common.otp.timed_otp import TimedOTPController # To make REST API calls: import httpx @@ -152,8 +154,7 @@ from api.blueprints.ai.llm.invoke import llm_invoke_bp # Common Blueprints: from api.blueprints.common.disable import auth_token_disable_bp from api.blueprints.common.session_token import session_token_bp -from api.blueprints.common.isp.user_login import isp_user_login_bp -from api.blueprints.common.isp.user_signup import isp_user_signup_bp +from api.blueprints.common.otp.timed_otp import timed_otp_bp # Tech and Testing Blueprints: from api.blueprints.tech.chat_alerts import tech_chat_alert_bp @@ -229,8 +230,7 @@ app.register_blueprint(trading_symbols_list_bp, url_prefix = f"/{MODULE_BASE}/fi # Common Blueprints: app.register_blueprint(auth_token_disable_bp, url_prefix = f"/{MODULE_BASE}") app.register_blueprint(session_token_bp, url_prefix = f"/{MODULE_BASE}") -app.register_blueprint(isp_user_login_bp, url_prefix = f"/{MODULE_BASE}/isp/user") -app.register_blueprint(isp_user_signup_bp, url_prefix = f"/{MODULE_BASE}/isp/user") +app.register_blueprint(timed_otp_bp, url_prefix = f"/{MODULE_BASE}/otp/timed") # AI Blueprints: app.register_blueprint(llm_invoke_bp, url_prefix = f"/{MODULE_BASE}/ai") @@ -592,6 +592,12 @@ async def app_startup(**kwargs): ) current_app.printer("Software/MikroTik (C) ready.") + # Common / OTP: + current_app.timed_otp_controller = TimedOTPController( + debug = enable_debugging + ) + current_app.printer("Common/OTP (C) ready.") + # ┏┓ ┓ ┏┓┓• # ┃ ┏┓┏┓┏┓┏┓┏╋┏┓┏┓┏ ┏┓┏┓┏┫ ┃ ┃┓┏┓┏┓╋┏ # ┗┛┗┛┛┗┛┗┗ ┗┗┗┛┛ ┛ ┗┻┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛ diff --git a/controllers_v2/common/otp/timed_otp.py b/controllers_v2/common/otp/timed_otp.py index 5cc0fe8..9fac95a 100644 --- a/controllers_v2/common/otp/timed_otp.py +++ b/controllers_v2/common/otp/timed_otp.py @@ -146,15 +146,18 @@ class TimedOTPController: self, redis_cache: AsyncRedisCache, inbound_data: GenerateTimedOTP - ) -> str | None: + ) -> FunctionCallResponse: """ To generate one OTP and store it in cache. :param redis_cache: The instance of the cache client to use to hold the OTP. - :param inbound_data: The - :return: + :param inbound_data: The data from which the OTP will be generated. + :return: A structured response to describe what happened in the process. """ + # Start by assuming failure: + response = FunctionCallResponse() + # Generate the OTP: otp_key = self.KEY_PREFIX + redis_cache.make_key(str(inbound_data.id)) otp_client = HashedOTP(secret = HashedOTP.generate_secret()) @@ -172,53 +175,12 @@ class TimedOTPController: expiry = inbound_data.seconds ) - # Done here: - return otp if otp_stored else None - - async def generate_and_send_sms( - self, - mongo_data_conn: AsyncMongo, - redis_cache: AsyncRedisCache, - auth_token: CoreAuthTokenModel, - sms_client: SMSController, - sms_message: NimbusSMSIndiaMessage | SavvyBulkSMSKenyaMessage, - inbound_data: GenerateTimedOTP, - tags: List[str] = None - ) -> FunctionCallResponse: - - """ - Generate an OTP and send it via SMS. - :param mongo_data_conn: The client to use to connect to the database. - :param redis_cache: The client to use to connect to the cache. - :param auth_token: The credentials - :param sms_client: - :param sms_message: - :param inbound_data: - :param tags: - :return: - """ - - # Start by assuming failure: - response = FunctionCallResponse() - - # Generate the OTP: - otp = await self.generate(redis_cache, inbound_data) - if not otp: - response.message = "Failed to generate OTP." - return response - - # Send the SMS: - sms_response = await sms_client.send_many_sms( - mongo_data_conn = mongo_data_conn, - auth_token = auth_token, - messages = [sms_message], - tags = tags - ) - - # Analyze the actions: - success = True if sms_response.successCount == 1 else False - response.success = success - response.message = "OTP sent successfully." if success else "OTP could NOT be sent." + # Analyze the response: + if otp_stored: + response.success = True + response.message = "OTP generated successfully." + response.data = otp + else: response.message = "Failed to generate an OTP." # Done here: return response @@ -235,9 +197,9 @@ class TimedOTPController: """ To verify if an OTP is valid, or not. - :param redis_cache: - :param inbound_data: - :return: + :param redis_cache: The instance of the cache client to use to hold the OTP. + :param inbound_data: The data from which the OTP will be verified. + :return: A structured response to describe what happened in the process. """ # Start by assuming failure: diff --git a/controllers_v2/core/auth_token.py b/controllers_v2/core/auth_token.py index 97b2e91..d733b36 100644 --- a/controllers_v2/core/auth_token.py +++ b/controllers_v2/core/auth_token.py @@ -630,6 +630,41 @@ class CoreAuthTokenController(CoreBaseModel): # Done here: return [CoreAuthTokenModel(**token) for token in tokens] + async def get_first_token( + self, + mongo_data_conn: AsyncMongo, + must_be_active: bool = True, + additional_filter: dict = None + ) -> CoreAuthTokenModel | None: + + """ + To fetch the first auth-token that matches the given conditions. Good when you have been asked to, say, send an + SMS without being specific about which client to send it from. In such a case, tune your conditions such that + you fetch the first auth token for an SMS client and use that for the next steps. + :param mongo_data_conn: The database connection (MongoDB) to use to perform the action. + :param must_be_active: Set this to False if you want to allow pending and disabled accounts to be retrieved. + :param additional_filter: Any addition filters to use. + :return: The retrieved record that has the token, and information about the service and client if found, else + None when there is no matching record. + """ + + # Prepare the filter: + filter_json = {} + if self._base_filter: + for k, v in self._base_filter.items(): filter_json[k] = v + if additional_filter: + for k, v in additional_filter.items(): filter_json[k] = v + if must_be_active: filter_json["status"] = "active" + + # If there is some filtering possible, we fetch the token: + token = await mongo_data_conn.find_one( + collection = self.AUTH_COLLECTION, + filter = mongo_data_conn.dict_to_dot_notation(filter_json) + ) + + # Done here: + return None if token is None else CoreAuthTokenModel(**token) + async def get_batches_to_sync( self, mongo_data_conn: AsyncMongo, diff --git a/models/api/common/otp/__init__.py b/models/api/common/otp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/models/api/common/otp/timed_otp.py b/models/api/common/otp/timed_otp.py new file mode 100644 index 0000000..11d66b8 --- /dev/null +++ b/models/api/common/otp/timed_otp.py @@ -0,0 +1,209 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Tuesday, 10th Sept., 2024. + + OBJECTIVE: + + To provide a data structure for the JSON received in the API calls to generate and verify timed OTPs. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For making data models: +from pydantic import BaseModel, Field, field_validator, Extra +from typing import Optional, Any, Dict, List +from typing_extensions import Annotated + +# My utils: +from utils_v2.string import regex + +# To work with MongoDB: +from bson.objectid import ObjectId + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# RegEx Patterns: +REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +class TimedOTPRequestHeaders(BaseModel): + + sessionToken: str | None = Field( + description = "the session token of the user who is requesting the service", + pattern = REGEX_SESSION_TOKEN, + frozen = True, + default = None, + alias = "X-Session-Token" + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "allow" + + def model_dump(self, *args, **kwargs): + return super().model_dump(*args, by_alias = True, **kwargs) + + +# --------------------------------------------------------------------------------------------------------------------- + + +class SendTimedOTPSMSFromSavvyBulkSMSKenyaRequestData(BaseModel): + + tokenKey: ObjectId | None = Field( + description = ( + "If you wish to use a specific account, you may send that account's token key, else the first SMS client " + "account will be picked." + ), + frozen = True, + default = None + ) + + id: dict = Field( + description = "Any identifier to associate the OTP with. Cannot be blank.", + frozen = True + ) + + attempts: int = Field( + description = "How many times the user can try to verify the generated OTP.", + frozen = True, + gt = 0, + lt = 11, + default = 3 + ) + + seconds: int | float = Field( + description = "The no. of seconds until which the OTP will remain valid.", + frozen = True, + gt = 29.9, + lt = 300.1, + default = 3 + ) + + recipientNo: str = Field( + description = "The phone no. of the target recipient of the OTP.", + frozen = True + ) + + smsText: str = Field( + description = "The templet of the message to send.", + frozen = True, + default = "Hello! Please use the following OTP ##OTP##" + ) + + smsReplace: str = Field( + description = "The substring in the 'smsText' template to replace with the actual OTP." + ) + + tags: List[str] | None = Field( + description = "A set of tags to associate with the message.", + frozen = True, + default = None + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + arbitrary_types_allowed = True + + def get(self, key: str, default = None): + return getattr(self, key, default) + + # ┓┏ ┓• ┓ • + # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ + # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ + + @field_validator("tokenKey", mode = "before") + def parse_oid(cls, value): + try: + if value is not None: value = ObjectId(value) + except: pass + return value + + @field_validator("id", mode = "after") + def validate_attempts(cls, value): + if not value: raise ValueError("The 'id' field cannot be empty!") + return value + + +# --------------------------------------------------------------------------------------------------------------------- + + +class VerifyTimedOTPRequestData(BaseModel): + + id: str | Dict | List + otp: str + + class Config: + extra = "forbid" + + def get(self, key: str, default = None): + return getattr(self, key, default) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/models/common/otp/timed_otp.py b/models/common/otp/timed_otp.py index 65312b2..7e68d20 100644 --- a/models/common/otp/timed_otp.py +++ b/models/common/otp/timed_otp.py @@ -73,7 +73,7 @@ from utils_v2.string import regex class GenerateTimedOTP(BaseModel): - id: str | Dict | List + id: dict attempts: Optional[int] = 3 seconds: Optional[int | float] = 30.0 @@ -99,7 +99,7 @@ class GenerateTimedOTP(BaseModel): class VerifyTimedOTP(BaseModel): - id: str | Dict | List + id: dict otp: str class Config: