""" AUTHOR: Khushal P Soonderji DATE: Monday, 25th Nov., 2024 OBJECTIVE: To receive callbacks (webhooks). 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, g, request, render_template # My utils: from utils_v2.string import json from utils_v2.api.codes import StatusCodes, HttpCodes from utils_v2.api.response import ResponseModel 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 ) # Common: from shared import constants # Behaviour Models: from models.behaviour.mail.oauth import MailOAuthModel # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Related to Quart: mail_callback_bp = Blueprint("mail_cb", __name__) # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** @mail_callback_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 # --------------------------------------------------------------------------------------------------------------------- async def handle_gmail_callback(): """ To handle the callbacks from GMail specifically. Refer to the individual comments to check what hap[pens at each step of the process. :return: A rendered template of the final status of the authorization. """ # Start by assuming failure: tokens_saved = False # In case the user cancelled halfway through (on Google's screen): if g.inbound_data.get("error") == "access_denied": return await render_template( "/mail/oauth/oauth_cancelled_v2.html", mail_client = g.mail_client.title() ) # Generate the tokens from the callback. Google sends all the needed params in the callback as the URL's query # params. We can simply use the exact URL that was hit to generate the tokens. In Quart (and Flask) this can be # achieved by 'request.url' like this: tokens = await current_app.gmail_client.get_authorization_tokens( redirect_url = request.url, scopes = g.inbound_data["scope"] ) if tokens: # Get the e-mail id that granted authorization. We will be comparing this to the e-mail id that had been given # to us when the authorization was initiated. We don't mind any e-mail id being used, but we need them to be the # same at both ends: user_profile = await current_app.gmail_client.get_user_profile(tokens = tokens) tokens.email = user_profile.data["emailAddress"] if user_profile.success else None # Here's where we do the checking of the e-mails, # if they don't match, we reject the authorization: placeholder_token = await current_app.mail_oauth_model.get_token( mongo_conn = current_app.data_mongo, account_identifier = g.inbound_data["state"] ) if ( (not placeholder_token) or placeholder_token["clientUserId"]["email"] != str(tokens.email) ): return await render_template( "/mail/oauth/oauth_failure_v2.html", mail_client = g.mail_client.title(), failure_hint = f"We were expecting authorization from '{placeholder_token['clientUserId']['email']}' but got authorization from '{tokens.email}' instead." ) # Now that we have passed the check, # we save the tokens to the database: tokens_saved = await current_app.mail_oauth_model.set_token( db_conn = current_app.sql_writer, mongo_conn = current_app.data_mongo, session_token = g.inbound_headers.get("X-Session-Token"), account_identifier = g.inbound_data["state"], token = tokens.model_dump() ) # Return an HTML response for success: if tokens_saved: return await render_template( "/mail/oauth/oauth_success_v2.html", mail_client = g.mail_client.title() ) # Return an HTML response for failure: else: return await render_template( "/mail/oauth/oauth_failure_v2.html", mail_client = g.mail_client.title(), failure_hint = f"Unknown error. Please use log-id '{g.log_id}' to check with the support team." ) # --------------------------------------------------------------------------------------------------------------------- @mail_callback_bp.route("/callback/", methods = ["POST", "GET"]) @set_api_version(api_version = "1.0.0") @read_input(sanitize_headers = False, sanitize_data = False) @log_request_to_mongo( attr_name = "logs_mongo", project = constants.PROJECT_NAME, log_type = constants.MODULE_NAME, operation = "gmailCllBckApi", 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") @handle_cancelled_request() async def mail_callback( mail_client: str = None, inbound_headers: dict = None, inbound_data: dict = None, inbound_files: dict = None, **kwargs ): """ Use this when authorizing access to someone's GMail account. This can be used to capture the authentication token. :param mail_client: The mail company/brand that you want the authorization from. :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. """ # ┓┏ ┓┓ ┓┏ • ┓ ┓ # ┣┫┏┓┏┓┏┫┃┏┓ ┃┃┏┓┏┓┓┏┓┣┓┃┏┓┏ # ┛┗┗┻┛┗┗┻┗┗ ┗┛┗┻┛ ┗┗┻┗┛┗┗ ┛ # Here we make various variables available in the scope of the current request through 'g': g.log_id = kwargs.get("log_id") g.inbound_headers = inbound_headers g.inbound_data = inbound_data g.mail_client = mail_client # ┓┏ ┓┓ ┏┓ ┓┓┓ ┓ # ┣┫┏┓┏┓┏┫┃┏┓ ┃ ┏┓┃┃┣┓┏┓┏┃┏┏ # ┛┗┗┻┛┗┗┻┗┗ ┗┛┗┻┗┗┗┛┗┻┗┛┗┛ if mail_client == "gmail": return await handle_gmail_callback() # ┓┏ ┓┓ ┳ ┓• ┓ ┏┓┓• # ┣┫┏┓┏┓┏┫┃┏┓ ┃┏┓┓┏┏┓┃┓┏┫ ┃ ┃┓┏┓┏┓╋ # ┛┗┗┻┛┗┗┻┗┗ ┻┛┗┗┛┗┻┗┗┗┻ ┗┛┗┗┗ ┛┗┗ return await render_template( "/mail/oauth/oauth_failure_v2.html", mail_client = mail_client.title(), failure_hint = f"Invalid client '{mail_client}' selected. Please use log-id '{g.log_id}' to check with the support team." ) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass