How-To Fernet Encrypt / Store / Decrypt OAuth Tokens in DynamoDB (Python)

Python developers trying to implement the Square OAuth token exchange in their AWS Lambda functions must encrypt the access/refresh tokens returned from Square. AWS provides a token management service, but those opting for their application-level DynamoDB (for ease of implementing refresh, lookup and revoke) face a challenge. Using Fernet to encrypt the Strings, the resulting Python type is <class β€˜bytes’>, which can be stored in DynamoDB as type Binary. However, when the data is read FROM DynamoDB the data type is now <class β€˜boto3.dynamodb.types.Binary’>, which when fed back into Fernet.decrypt() does not regenerate the original token.

The following code is how the Leedz uses Fernet to SUCCESSFULLY encrypt() and decrypt() token data for storage in DDB. The seed for generating the Fernet key is up to you.
(apologies for the post formatting β€” it’s the best I could do)

import os

from cryptography.fernet import Fernet
import base64
import hashlib

import boto3
from boto3.dynamodb.types import Binary
from boto3.dynamodb.conditions import Key

β€˜β€™β€™ take a string as input and converts it into a string of 32 bytes suitable for cryptographic use β€˜β€™β€™

def convert(src):

hash_object = hashlib.sha256(src.encode())
return hash_object.digest()

β€˜β€™β€™ create a Fernet encryption key from a plain text source β€˜β€™β€™

def generateFernetKey(txt_src):

  seed = convert(txt_src)
  return base64.urlsafe_b64encode(seed)

β€˜β€™β€™ encrypt the plaintext token β€˜β€™β€™

def encryptToken( key_txt, token_txt):

try:
        fernet_key = generateFernetKey(key_txt)
        f = Fernet(fernet_key)
        encrypted_token = f.encrypt(token_txt.encode('ASCII'))
        return encrypted_token
except Exception as err:
        raise err   

β€˜β€™β€™ decrypt the DDB binary type back to plaintext β€˜β€™β€™

def decryptToken(key_txt, encrypted_token):

try:
    fernet_key = generateFernetKey(key_txt)
    f = Fernet(fernet_key)

    # Convert the encrypted_token to bytes
    bytes_object = encrypted_token.value

    decrypted_token = f.decrypt(bytes(bytes_object))

    return decrypted_token.decode('ASCII')
except Exception as err:
    print("Error in decryptToken: " + str(err))
    raise err

β€˜β€™β€™
β€˜β€™β€™ Basic LAMBDA function to test encrypt and decrypt
β€˜β€™β€™
β€˜β€™β€™ encrypt text β†’ DB
β€˜β€™β€™ decrypt DB β†’ text
β€˜β€™β€™
def lambda_handler(event, context) :

try:
    start_token = "plain_text_secret_access_token"
    fernet_seed = "unique_seed"

    dynamodb_client = boto3.resource("dynamodb")
    table = dynamodb_client.Table('YOUR_DB_NAME')
    
    #
    # ENCRYPT
    #
    crypt_token = encryptToken(fernet_seed, start_token)

    print(f"ENCRYPT SUCCESS! TOKEN={crypt_token}")
    the_type = str(type(crypt_token))
    print("TYPE=" + the_type)
    
    
    #
    # STORE ITEM IN DDB
    #
    response = table.put_item(
    Item={
        'pk': 'user',
        'sk': 'test_crypt',
        'sq_at': crypt_token
        },
        ReturnValues='ALL_OLD',
    )

    print("DDB PUT_ITEM SUCCESS !!")
    print( str(response) )


    response = table.get_item(
        Key={'pk': 'user', 'sk': 'test_crypt'}
    )
        
    print("DDB GET_ITEM SUCCESS !!")
    print( str(response) )
                          
    #
    # DECRYPT
    # 
    the_user = response['Item']
    the_type = str(type(the_user['sq_at']))
    print("TYPE=" + the_type)
    end_token = decryptToken(fernet_seed, the_user['sq_at'])
            
    print(f"DECRYPT SUCCESS! TOKEN={end_token}")
        
    return end_token
    

except Exception as error:
    print( str(error) )
    raise

Thanks for sharing your findings! :slightly_smiling_face:

1 Like

you bet, dude. I could not have gotten this far without you.