• 131
• 255
• 14

How to code Oauth in a Cloud Function using Python

by bernt & torsten

A RESTful API is an architectural design for an application program interface (API) that uses HTTP requests to access and use data. OAuth is an authentication protocol that allows you to approve one application interacting with another on your behalf without giving away your password.

OAuth is a delegated authorization framework that enables apps to obtain limited access (scopes) to a user’s data without giving away a user’s password. It decouples authentication from authorization and supports multiple use cases addressing different device capabilities.

The basic Authentication for RestFul APIs looks something like the code example below, the code is created with Python and uses Google Secrets Manager to hide the secrets.

Basic Authentication

    # Install Google Libary
    from google.cloud import secretmanager
    
    # Get the secret for Token
    token = _getsecrets("token")

    # Get the secret for Company ID
    companyid = _getsecrets("company_id")

    # Create the encoded authentication parameters
    auth = companyid + ':' + token
    encoded_u = base64.b64encode(auth.encode()).decode()

    # Create the Header and url
    headers = {
        "Authorization" : "Basic %s" % encoded_u
    }

The code snippet reads the secrets and from the secrets create and encode the authentication that is used in the header. Setting up the URL and call the function _requestandstore with URL and header.

    url = 'https://api.test.com/v1/users'
    _requestandstore(url, headers)        

The function that makes the request, the URL request is straightforward by including the details of the headers for the request.get

def _requestandstore(url, header):
    try:
        # Make a request
        response = requests.get(url, params={'page': 1, 'per_page': 10}, headers=header)

The following code snippet functions are for getting the secrets from the Google Cloud Secrets manager.

def _getsecrets(secret_name):
    resource_name = f"projects/{project_id}/secrets/{secret_name}/versions/latest"
    try:
        response = client.access_secret_version(request={"name": resource_name})
        return response.payload.data.decode('UTF-8')
    except:
        return False

Oauth Authentication

OAuth (Open Authorization) is an open standard for access to an application for users of that application to grant websites or applications access to their information on other websites but without giving them the passwords. When a user is using Oauth on a website, a consent screen popups up to allow the users to give their permission.

A cloud function that runs on a serverless server does not have a UI that allows the users to give that consent, in the next part here I will explain how we use OAuth in a Cloud Function.

Before I go further I like to mention that the following code works well in custom RestFul API Oauth, if you are going to use a Google API – you may want to read Datapipeline youtube analytics to BigQuery with Airbyte in which I wrote on how you can get it done with Google OAuth.

Fake Oauth Login

For us to be able to use OAuth in a Cloud Function, we need to do a fake OAuth login, so we can get the access token, this is done by making an invalid request. If the redirect_uri is invalid, the browser will stop the redirect and show the authorization code.

Here is the code – first we have to get our client id, client secret and token_url, this information is provided by the service provider you trying to connect to.

The next step is to set the grant_type and make a RestFUL API post request to get the access token.

    # Get the secret for Client ID
    client_id = _getsecrets("client_id")

    # Get the secret for Client Secret
    client_secret = _getsecrets("client_secret")

    # Get the secret for Token URL
    token_url = _getsecrets("token_url")

    # Fake Oauth login to get an access token by making an invalid request, If the redirect_uri is invalid,
    # the browser will stop the redirect and show the authorization code.
    data = {'grant_type': 'client_credentials'}
    access_token_response = requests.post(token_url, data=data, verify=False, allow_redirects=False, auth=(client_id, client_secret))
    tokens = json.loads(access_token_response.text)

    # Create the Header and url
    headers = {
        'Authorization': 'Bearer ' + tokens['access_token']
    }

When you have the access token, you set your RestFUL API URL with the endpoint and make a RestFul API call.

    url = 'https://api.test.com/api/v2/users'
    _requestandstore(filename, url, headers)

def _requestandstore(filename, url, header):
    try:
        # Make a request to API
        response = requests.get(url, params={'page': 1, 'perPage': 400}, headers=header, verify=False)
 

That is it, this should work in most scenarios – and advice is to talk to the service provider you try to connect with to work out the details.

Use Cases

To mention there are several use cases where you want to use a Cloud Function to get access to a Service Providers Application. I will mention a few, you may have more so please comment below.

• Service Provider extraction of data to Data Warehosue to be used for Analytics
• Social Media extraction of data to Data Warehosue to be used for Analytics
• Online Media extraction of data to Data Warehosue to be used for Analytics