""" AUTHOR: Khushal P Soonderji DATE: Friday, 22nd Nov., 2024 OBJECTIVE: To handle OAuth2.0 activities for Google's services. REFERENCES: 1. https://developers.google.com/calendar/api/quickstart/python 2. https://developers.google.com/identity/protocols/oauth2/web-server#python 3. https://www.youtube.com/watch?v=vQQEaSnQ_bs&t=940s&pp=ygUVb2F1dGgyIHB5dGhvbiB5b3V0dWJl DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import io # My utils: from utils_v2.string import json from utils_v2.date_time import date_time # The base model: from utils_v2.oauth.base import OAuthBase # Related to Google: from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow # For asynchronous activities: import asyncio # To make deep-copies: import copy # To work with date and time: import datetime # For working with datatypes: from typing import List # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # Google Scopes: SCOPES_GMAIL_MAIL_MANAGEMENT = [ r"https://www.googleapis.com/auth/gmail.modify", r"https://www.googleapis.com/auth/gmail.labels" ] SCOPES_GMAIL_FULL = [r"https://mail.google.com/"] # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class GoogleOAuth(OAuthBase): # Class variables: service_name = "google" __flow = None # ┏┓┓ ┳┳┓ ┓ ┓ # ┣┫┣┓┏╋┏┓┏┓┏╋ ┃┃┃┏┓╋┣┓┏┓┏┫┏ # ┛┗┗┛┛┗┛ ┗┻┗┗ ┛ ┗┗ ┗┛┗┗┛┗┻┛ async def initialize( self, scopes: List, raise_exception = False ) -> bool: """ Initialize the Google OAuth mechanism by creating an app-flow. This defines the app that you are trying to deploy. You must create this app in Google's Cloud Platform's console. :param scopes: The scopes (permissions) needed by this app. :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any exception that occurs will be suppressed. :return: True if the initialization succeeded, False if it failed. """ # Start by assuming success: success = True try: # Initialize your Google App: self._printer("Initializing flow.") self.__flow = InstalledAppFlow.from_client_config( self._config, scopes = scopes, redirect_uri = self._redirect_url ) # If something goes wrong: except Exception as exception: self._printer(exception) if raise_exception: raise success = False # Done here: return success async def get_authorization_url( self, raise_exception = False, **kwargs, ) -> str | None: """ To generate an authorization URL that can be sent to the front end. When the :param raise_exception: If set to True, any exception that occurs will be propagated. If set to False, any exception that occurs will be suppressed. :param kwargs: Any no. of keyword args that you might want to give to this specific service. :return: """ # Start by assuming failure: authorization_url = None try: # Request a URL that will be sent to the user to request # permissions to access their account: authorization_url, _ = self.__flow.authorization_url( access_type = kwargs.get("access_type", "offline"), approval_prompt = kwargs.get("approval_prompt", "force"), include_granted_scopes = kwargs.get("include_granted_scopes", "true"), login_hint = kwargs.get("email"), state = kwargs.get("user_id") ) # Done here: return authorization_url # If something goes wrong: except Exception as exception: self._printer(exception) if raise_exception: raise authorization_url = None # Done here: return authorization_url async def get_tokens( self, raise_exception = False, **kwargs ) -> dict | None: """ To get the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires every-so-often, and one refresh token that grants you a new access token. In Google's case, their servers hit your callback URL with some query params. You must furnish this whole URL right here under the kwarg 'redirect_url'. Fortunately, this URL is readily available in Quart and Flask by calling 'request.url'. :param kwargs: The identifiers of the user who wants to use your service and any other service-specific options. :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any exception that occurs will be suppressed. :return: The tokens for the service if successful, or None if failed. """ # Start by assuming failure: tokens = None try: # Fetch the tokens: credentials = self.__flow.fetch_token(authorization_response = kwargs["redirect_url"]) tokens = { "access_token": credentials.get("access_token"), "refresh_token": credentials.get("refresh_token"), "expires_in": (ttl := credentials["expires_in"] - 60), "expires_at": date_time.get_current_utc_date_time() + datetime.timedelta(seconds = ttl), "scopes": credentials.get("scope"), } # If something goes wrong: except Exception as exception: self._printer(exception) if raise_exception: raise tokens = None # Done here: return tokens async def refresh_tokens( self, old_tokens: dict, force_refresh = False, raise_exception = False ) -> dict | None: """ To refresh the tokens of a user. Plural 'tokens' because OAuth typically has one access token that expires every-so-often, and one refresh token that grants you a new access token. :param old_tokens: The current set of tokens as obtained by 'get_tokens'. :param force_refresh: To force a refresh request even if the tokens haven't yet expired. :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any exception that occurs will be suppressed. :return: The same tokens if they haven't expired, refreshed tokens if the tokens have expired and were successfully refreshed, None if the tokens have expired but could not be refreshed. """ # Start by assuming failure: tokens = None try: # If the tokens haven't expired, just return the existing tokens back: tokens_expired = True if date_time.get_current_utc_date_time() >= old_tokens["expires_at"] else False if not tokens_expired and not force_refresh: return old_tokens # Construct the credentials and request a refresh: credentials = await self.credentials_from_tokens(old_tokens) if credentials.refresh_token: credentials.refresh(Request()) tokens = { "access_token": credentials.token, "refresh_token": credentials.refresh_token, "expires_at": (exp_at := date_time.as_if_timezone( credentials.expiry, timezone = date_time.TIMEZONE_UTC )), "expires_in": (exp_at - date_time.get_current_utc_date_time()).total_seconds(), "scopes": credentials.scopes, } # If something goes wrong: except Exception as exception: self._printer(exception) if raise_exception: raise tokens = None # Done here: return tokens # ┏┓ • ┏┓ •┏• # ┗┓┏┓┏┓┓┏┓┏┏┓ ┗┓┏┓┏┓┏┓╋┓┏ # ┗┛┗ ┛ ┗┛┗┗┗ ┗┛┣┛┗ ┗┗┛┗┗ # ┛ @staticmethod async def credentials_have_expired(credentials: Credentials) -> bool: """ Checks is a 'Credentials' object has expired or not. The native mechanism has some complication with timezone considerations. This method was created to overcome that bug. :param credentials: The 'Credentials' object of the user. :return: True if the credentials have expired, False if they're yet valid. """ # Take the timezone into account: expires_at = date_time.as_if_timezone( credentials.expiry, timezone = date_time.TIMEZONE_UTC ) # Return the assessment based on the evaluation with the timezone considered: return True if date_time.get_current_utc_date_time() >= expires_at else False async def credentials_from_tokens( self, tokens: dict, raise_exception = False ) -> Credentials | None: """ This is a very service-specific thing. When using Google's APIs, you will need to use their native 'Credentials' object. This method allows you to quickly convert the output of 'get_tokens' and 'refresh_tokens' to that object for easy use. :param tokens: The output of either 'get_tokens' or 'refresh_tokens'. :param raise_exception: If set to True, any exception that occurs will be propagated. If set to false, any exception that occurs will be suppressed. :return: Google's API's native 'Credentials' object if successful, else None. """ # Start by assuming failure: credentials = None try: # Make a deep-copy and add some fields from the config: tokens_copy = copy.deepcopy(tokens) first_key = list(self._config.keys())[0] tokens_copy["client_id"] = self._config.get(first_key, {}).get("client_id") tokens_copy["client_secret"] = self._config.get(first_key, {}).get("client_secret") # Create the credentials: credentials = Credentials.from_authorized_user_info(info = tokens_copy) # If something goes wrong: except Exception as exception: self._printer(exception) if raise_exception: raise credentials = None # Done here: return credentials # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": import dateparser secrets_file = r"../../../creds/google_tcaoff_test_oauth_20241125.json" secrets_dict = json.from_file(secrets_file) my_goog = GoogleOAuth( config = secrets_dict, redirect_url = r"https://api.thecaoffice.com/converse/mail/callback/gmail", debug = True, debug_prefix = "OAuth (Goog) | " ) async def main(): await my_goog.initialize(scopes = SCOPES_GMAIL_MAIL_MANAGEMENT) # await asyncio.sleep(1.0) # print("Service Name:", my_goog.service_name) # print("AUTH URL 0:", await my_goog.get_authorization_url( # user_id = "BHOPLI", # # email = "pskhushal@gmail.com" # )) # redirect_url = input("Paste the redirect URL here: ") # tokens = await my_goog.get_tokens(redirect_url = redirect_url) # print("TOKENS:", json.to_string(tokens, default = str)) # To test refreshing: tokens = { "access_token": "ya29.a0AeDClZAYoo85BXRId_n-hwo_amKshzi46c33GaJcsZZvGB7A7OGU2RFYcWBM_BleNBfAFUSJP2NHAvmd7Nsp_U5Kg68hXSy0iO99PNTm3pvKrJSzbkA-rXsVLsCnBIfPUMyNt2nOOVJmGwm17DNN0jAELkm1fPNTju7SZzmuaCgYKAZwSARMSFQHGX2Mis8TZui2rZT1gKySVds-N0w0175", "refresh_token": "1//0gnqzjMf9YT19CgYIARAAGBASNgF-L9Ir3rcY37nGrV45XyOUBRllEH7Txui7T1JbwevlmDoNw7PuMu149cCWQSwsScuKaZusUQ", "expires_in": 3539, "expires_at": dateparser.parse("2024-11-25 10:40:40.833699+00:00"), "scopes": [ "https://www.googleapis.com/auth/gmail.labels", "https://www.googleapis.com/auth/gmail.modify" ] } new_tok = await my_goog.refresh_tokens(old_tokens = tokens, force_refresh = False) print("REFRESHED TOKENS:", json.to_string(new_tok, default = str)) asyncio.run(main())