OAuth Part III – Generating Access & Refresh tokens, encrypting and saving to the database

OK, so someone has clicked on your big OAuth button and allowed access to their Google Data – you now need to process the creds, encrypt and save them to your database ready to make some API calls.

We handled the callback in the previous post, we’ve now passed the data from the endpoint into the service and handle the encrypting and saving.

The code below takes the access code from the callback URL and created params which we can pass into the post request to get an offline refresh token to make all subsequent API calls.
The google_client id, google_client_secret and google_auth_redirect are the ones we obtained when setting up the app in the first post

httpx is a python library for handling request, we use it to make a post request.

def _get_refresh_token(access_code):
    params = (
        f"code={access_code}"
        f"&client_id={config.get_settings().google_client_id}"
        f"&client_secret={config.get_settings().google_client_secret}"
        f"&redirect_uri={config.get_settings().google_auth_redirect}"
        f"&grant_type=authorization_code"
    )
    authorization_url = "https://oauth2.googleapis.com/token"

    re = httpx.post(authorization_url, 
                    params=params, 
                    headers={"content-type": "application/x-www-form-urlencoded"})

return re.json()

Once we have the refresh token we don’t need to worry about the access_code anymore – we can now process and save the refresh token. The code below handles this flow

def save_access_credentials(self, code):

    # Get the access & refresh token
    refresh_token = self._get_refresh_token(code)

    # Dump the data to a dict
    jdata = json.dumps(refresh_token)

    # Encrypt the creds
    bdata = EncryptDecrypt().encrypt(bytes(jdata, 'utf-8'))

    # Create the creds profile and save to db
    data = {"client_id": self.client_id,
            "access_credentials": bdata,
            "status": "approved"}

    # If the token exists for the client update it, if not create it
    insert_stmt = insert(GoogleAuthCredentialsModel).values(data)
    upsert_stmt = insert_stmt.on_conflict_do_update(
                               index_elements=[GoogleAuthCredentialsModel.client_id],
                               set_=data)
    self.database.execute(upsert_stmt)
    self.database.commit()
    self.database.flush()
    return self._get_creds_profiles_by_client_id().scalar()

I’m using SQLALchemy here to interact with the database (more on this later) and creating an upsert statement – this will create a record if one doesn’t exist else it’ll update the current record in the database.
All the methods sit within a GoogleAuthClass as below

class GoogleAuth:

    def __init__(self,
                 database: Session,
                 client_id: int = None,
                 scope: str = None):
        self.database = database
        self.client_id = client_id
        self.scope = scope
        self.application_creds = None

The only other piece to the puzzle here is the encrypting of the refresh token, this is handled by an security class using Fernet

from cryptography.fernet import Fernet, MultiFernet
from core import config


class EncryptDecrypt:

    def __init__(self):
        self.fernet = Fernet(str.encode(config.get_settings().fernet_key))

    def encrypt(self, token):
        if type(token) != bytes:
            token = bytes(token)
        return self.fernet.encrypt(token)

    def decrypt(self, token):
        return (self.fernet.decrypt(bytes(token))).decode()

That’s it for this post – in the next post we’ll use the refresh token to generate an access token we can then use to make an API call.