212 lines
8.4 KiB
Python
212 lines
8.4 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Friday, 22nd Nov., 2024
|
|
|
|
OBJECTIVE:
|
|
|
|
To provide a structured way to handle OAuth2.0 behaviour for various services.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# System-level activities:
|
|
import io
|
|
|
|
# For defining the class's structure:
|
|
from abc import ABC, abstractmethod
|
|
|
|
# For working with datatypes:
|
|
from typing import List
|
|
|
|
# For debugging:
|
|
from icecream import IceCreamDebugger
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class OAuthBase(ABC):
|
|
|
|
def __init__(
|
|
self,
|
|
config: dict,
|
|
redirect_url: str,
|
|
debug = True,
|
|
debug_prefix = "OAuth | ",
|
|
debug_only_errors = True
|
|
):
|
|
|
|
# Prepare the debugging utility:
|
|
self._debug_prefix = debug_prefix
|
|
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
|
|
if not debug: self._printer.disable()
|
|
self._debug_only_errors = debug_only_errors
|
|
|
|
# Accept the input config:
|
|
self._config = config
|
|
self._redirect_url = redirect_url
|
|
|
|
def enable_debug(self):
|
|
self._printer.enable()
|
|
|
|
def disable_debug(self):
|
|
self._printer.disable()
|
|
|
|
def debug_only_errors(self):
|
|
self._debug_only_errors = True
|
|
|
|
def debug_everything(self):
|
|
self._debug_only_errors = False
|
|
|
|
# ┏┓┓ ┳┳┓ ┓ ┓
|
|
# ┣┫┣┓┏╋┏┓┏┓┏╋ ┃┃┃┏┓╋┣┓┏┓┏┫┏
|
|
# ┛┗┗┛┛┗┛ ┗┻┗┗ ┛ ┗┗ ┗┛┗┗┛┗┻┛
|
|
|
|
@abstractmethod
|
|
async def initialize(
|
|
self,
|
|
scopes: List,
|
|
raise_exception = False
|
|
) -> bool:
|
|
|
|
"""
|
|
To initialize the service-specific OAuth2.0 class. For example, in Google's case, we need to initialize an app
|
|
flow that was created for an app through its Cloud Console panel.
|
|
:param scopes: The list of permissions being requested. The word 'scopes' has been borrowed from Google's OAuth
|
|
documentation (which was implemented first).
|
|
: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.
|
|
"""
|
|
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def get_authorization_url(
|
|
self,
|
|
raise_exception = False,
|
|
**kwargs,
|
|
) -> str | None:
|
|
|
|
"""
|
|
To create an authorization URL which will be then sent to the front-end for the user to click and grant/decline
|
|
various permissions.
|
|
: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 authorization URL if successful, or None if failed.
|
|
"""
|
|
|
|
pass
|
|
|
|
@abstractmethod
|
|
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.
|
|
: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.
|
|
"""
|
|
|
|
pass
|
|
|
|
@abstractmethod
|
|
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.
|
|
: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.
|
|
"""
|
|
|
|
pass
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
pass
|