""" 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.logging.context import AsyncLoggerContext 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 ) # GMail-related utils: from utils_v2.goog.controllers.gmail.gmail_client import SCOPES_GMAIL_MAIL_MANAGEMENT # Data Models: from models.core.auth_token import CoreAuthTokenModel # Common: from shared import constants # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Related to Quart: mail_oauth_callback_bp = Blueprint("mail_cb", __name__) # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** @mail_oauth_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 # --------------------------------------------------------------------------------------------------------------------- @AsyncLoggerContext.log_it( api_version = "1.0.0", project = constants.PROJECT_NAME, log_type = constants.MODULE_NAME, operation = "gmailClbk", log_input = 2, log_output = 1, sensitive_keys = ["sessionToken", "X-Session-Token"] ) async def handle_gmail_callback() -> render_template: """ 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 (HTML) 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: google_tokens = await current_app.gmail_client.get_authorization_tokens( redirect_url = request.url, scopes = None ) if google_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 = google_tokens) if user_profile.success: google_tokens.email = user_profile.data["emailAddress"] google_tokens.displayName = user_profile.data["displayName"] google_tokens.displayPictureUrl = user_profile.data["displayPictureUrl"] # Here's where we do the checking of the e-mails, # if they don't match, we reject the authorization: auth_token = await current_app.mail_controller.get_token_from_key( mongo_conn = current_app.data_mongo, token_key = g.inbound_data["state"] ) if ( (not auth_token) or auth_token.clientUserId["email"] != str(google_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 '{auth_token.clientUserId['email']}', " f"but got authorization from '{google_tokens.email}' instead." ) ) # We create standard labels that we will use: labels = [ { "name": "TCAOFF", "textColor": "#434343", "backgroundColor": "#e7e7e7" }, { "name": "CA-Doc", "textColor": "#434343", "backgroundColor": "#e7e7e7" }, { "name": "CA-AI", "textColor": "#434343", "backgroundColor": "#e7e7e7" } ] tasks = [ current_app.gmail_client.create_label( tokens = google_tokens, label_name = label["name"], label_visibility = "labelShow", message_visibility = "show", label_text_color = label["textColor"], label_background_color = label["backgroundColor"] ) for label in labels ] client_responses = await asyncio.gather(*tasks) # Add the labels to the tokens data: client_response = await current_app.gmail_client.list_labels(tokens = google_tokens) google_tokens.labels = client_response.data if client_response.success else None # Now that we have passed the check, # we save the tokens to the database: auth_token.clientUserId = google_tokens.client_user_id auth_token.token = google_tokens.model_dump() auth_token.status = "active" tokens_saved = await current_app.mail_controller.set_token( db_conn = current_app.sql_writer, mongo_conn = current_app.data_mongo, session_token = g.inbound_headers.get("X-Session-Token"), token_key = g.inbound_data["state"], auth_token = auth_token ) # 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_oauth_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_auth_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 # ┓┏ ┓┓ ┏┓ ┓┓┓ ┓ # ┣┫┏┓┏┓┏┫┃┏┓ ┃ ┏┓┃┃┣┓┏┓┏┃┏┏ # ┛┗┗┻┛┗┗┻┗┗ ┗┛┗┻┗┗┗┛┗┻┗┛┗┛ try: if mail_client == "gmail": return await handle_gmail_callback() # If something goes wrong: except Exception as exception: return await render_template( "/mail/oauth/oauth_failure_v2.html", mail_client = mail_client.title(), failure_hint = ( f"An internal server error occurred. " f"Please use log-id '{g.log_id}' to check with the support team." ) ) # ┓┏ ┓┓ ┳ ┓• ┓ ┏┓┓• # ┣┫┏┓┏┓┏┫┃┏┓ ┃┏┓┓┏┏┓┃┓┏┫ ┃ ┃┓┏┓┏┓╋ # ┛┗┗┻┛┗┗┻┗┗ ┻┛┗┗┛┗┻┗┗┗┻ ┗┛┗┗┗ ┛┗┗ return await render_template( "/mail/oauth/oauth_failure_v2.html", mail_client = mail_client.title(), failure_hint = ( f"Invalid client '{mail_client}' selected. " f"Please use log-id '{g.log_id}' to check with the support team." ) ) # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass