From 2d27620675984ec357e6d0bc651536f73d043027 Mon Sep 17 00:00:00 2001 From: yatmesh Date: Mon, 24 Mar 2025 14:21:42 +0530 Subject: [PATCH] (2025-03-24)- Added Client Update with note into the time based otp test py file. --- api/blueprints/common/otp/timed_otp_test.py | 433 ++++++++++++++++++++ api/main.py | 4 +- 2 files changed, 436 insertions(+), 1 deletion(-) create mode 100644 api/blueprints/common/otp/timed_otp_test.py diff --git a/api/blueprints/common/otp/timed_otp_test.py b/api/blueprints/common/otp/timed_otp_test.py new file mode 100644 index 0000000..80e6bf4 --- /dev/null +++ b/api/blueprints/common/otp/timed_otp_test.py @@ -0,0 +1,433 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Tuesday, 10th Sept., 2024 + + OBJECTIVE: + + To be able to send SMS messages and get reports and balance through Nimbus IT's SMS sending service. + The core logic is implemented elsewhere, this module aims to build an API layer around it. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + + NOTES: + + N/A + +""" +import copy +# ***************************************************************************************************************** +# ***** **** +# *** 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.string import regex +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 ( + set_api_version, + read_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 + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# Related to Quart: +timed_otp_bp_test = Blueprint("timed_otp_test", __name__) + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +@timed_otp_bp_test.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 + + +# --------------------------------------------------------------------------------------------------------------------- + + +@timed_otp_bp_test.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 = "logs_mongo", + project = constants.PROJECT_NAME, + log_type = constants.MODULE_NAME, + operation = "otpSmsSendApi", + log_input = 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") +@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 | TimedOTPRequestHeaders = None, + inbound_data: dict | SendTimedOTPSMSFromSavvyBulkSMSKenyaRequestData = None, + inbound_files: dict = None, + **kwargs +): + + """ + Generates one OTP for the amount of time specified in the request. + :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. + """ + + # ┏┓ ┓ ┏┓┓ ┓ + # ┣┫┓┏╋┣┓ ┃ ┣┓┏┓┏┃┏ + # ┛┗┗┻┗┛┗ ┗┛┛┗┗ ┗┛┗ + + # Either the session must be valid, or + # the IP address requesting the service must be whitelisted: + if kwargs.get("session_info") is None: + return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.UNAUTHORIZED, + message = "Invalid session." + ) + + # 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: + phone_no = regex.replace( + text = inbound_data.recipientNo, + pattern = r"[^\d]", + substitute_text = "" + ) + if ( + not phone_no.startswith("0") and + not phone_no.startswith("254") + ): phone_no = "0" + phone_no + sms_message = SavvyBulkSMSKenyaMessage( + recipientNo = phone_no, + 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_test.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 = "logs_mongo", + project = constants.PROJECT_NAME, + log_type = constants.MODULE_NAME, + operation = "otpVerifyApi", + log_input = 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") +@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, + **kwargs +): + + """ + Verifies the claimed OTP against the stored OTP. + :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. + """ + + # ┏┓ ┓ ┏┓┓ ┓ + # ┣┫┓┏╋┣┓ ┃ ┣┓┏┓┏┃┏ + # ┛┗┗┻┗┛┗ ┗┛┛┗┗ ┗┛┗ + + # Either the session must be valid, or + # the IP address requesting the service must be whitelisted: + if kwargs.get("session_info") is None: + return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.UNAUTHORIZED, + message = "Invalid session." + ) + + # 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 + ) + ) + + # If the OTP failed: + if not otp_response.success: return ResponseModel( + status_code = StatusCodes.FAILED, + http_code = HttpCodes.UNAUTHORIZED, + message = otp_response.message + ) + + # ┏┓ ┓ ┓ ┏┓┓• ┓ ┏• ┓ ┳┓ + # ┣┫┏┫┏┫ ┃ ┃┓┏┓┏┓╋ ┃┃┃┓╋┣┓ ┃┃┏┓╋┏┓┏ + # ┛┗┗┻┗┻ ┗┛┗┗┗ ┛┗┗ ┗┻┛┗┗┛┗ ┛┗┗┛┗┗ ┛ + + # Fix the phone no. to have the country code: + phone_no = regex.replace( + text = inbound_data.phoneNo, + pattern = r"[^\d]", + substitute_text = "" + ) + if not phone_no.startswith("254"): phone_no = "254" + phone_no + phone_no = "+" + phone_no + + # if client id is not none then hit update the client add with notes else - add client with notes + # 24/03/2025 - + client_id = inbound_data.id.get("idClient") + if client_id is not None: + # hit Omkar's API: + api_response = await current_app.http_client.post( + url="https://api.thecaoffice.com/client/update/with/notes", + headers={"X-Session-Token": inbound_headers["X-Session-Token"]}, + json={ + "email": inbound_data.id.get("email"), + "clientName": inbound_data.id.get("clientName"), + "address": inbound_data.id.get("address"), + "phoneNo": inbound_data.id.get("phoneNo"), + "city": inbound_data.id.get("city"), + "pincode": inbound_data.id.get("pincode"), + "panCard": inbound_data.id.get("panCard"), + "gst": inbound_data.id.get("gst"), + "entity": inbound_data.id.get("entity"), + "country": inbound_data.id.get("country"), + "startDate": inbound_data.id.get("startDate"), + "period": inbound_data.id.get("period"), + "amount": inbound_data.id.get("amount"), + "idClient": inbound_data.id.get("idClient"), + "username": inbound_data.id.get("username"), + "password": inbound_data.id.get("password") + } + ) + else: + # hit Omkar's API: + api_response = await current_app.http_client.post( + url = "https://api.thecaoffice.com/client/add/with/notes", + headers = {"X-Session-Token": inbound_headers["X-Session-Token"]}, + json = { + "email": inbound_data.email, + "clientName": inbound_data.username, + "address": None, + "phoneNo": phone_no, + "city": None, + "pincode": None, + "panCard": None, + "gst": None, + "entity": None, + "country": None, + "startDate": None, + "period": None, + "amount": None + } + ) + + # ┳┓ + # ┣┫┏┓┏┏┓┏┓┏┓┏┏┓ + # ┛┗┗ ┛┣┛┗┛┛┗┛┗ + # ┛ + + # Done here: + success = api_response.is_success + return ResponseModel( + status_code = StatusCodes.OK if success else StatusCodes.FAILED, + http_code = HttpCodes.SUCCESS if success else HttpCodes.UNAUTHORIZED + ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/api/main.py b/api/main.py index 5492bef..d4e4a52 100644 --- a/api/main.py +++ b/api/main.py @@ -155,6 +155,7 @@ from api.blueprints.ai.llm.invoke import llm_invoke_bp from api.blueprints.common.disable import auth_token_disable_bp from api.blueprints.common.session_token import session_token_bp from api.blueprints.common.otp.timed_otp import timed_otp_bp +from api.blueprints.common.otp.timed_otp_test import timed_otp_bp_test # Tech and Testing Blueprints: from api.blueprints.tech.chat_alerts import tech_chat_alert_bp @@ -230,7 +231,8 @@ 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(timed_otp_bp, url_prefix = f"/{MODULE_BASE}/otp/timed") +# app.register_blueprint(timed_otp_bp, url_prefix = f"/{MODULE_BASE}/otp/timed") +app.register_blueprint(timed_otp_bp_test, url_prefix = f"/{MODULE_BASE}/otp/timed") # AI Blueprints: app.register_blueprint(llm_invoke_bp, url_prefix = f"/{MODULE_BASE}/ai")