Approov Integration in a Python Flask API

This walk-though will show us how simple it is to integrate Approov in a current API server using Python and the Flask framework.

We will see the requirements, dependencies and a step by step walk-through over the code necessary to implement Approov in a Python Flask API.

Before we tackle the integration of Approov we need first to know how Approov validation is processed in the server and how to setup the environment to follow this walk-through.

Note that this article assumes a basic understanding of the Approov mechanics. If you need an overview of that, please read first the Approov Product page.

Approov Token Validation Process

We will understand at a high level how the Approov token itself is validated and also how to validate the optional Approov token binding.

The Approov Token

API calls protected by Approov will typically include a header holding an Approov JWT token. This token must be checked to ensure it has not expired and that it is properly signed with the secret shared between the back-end and the Approov cloud service.

We will use a Python package to help us in the validation of the Approov JWT token. Just to be sure that we are on the same page, a JWT token has 3 parts which are separated by dots and represented as a string in the format of header.payload.signature. Read more about JWT tokens here.

The Approov Token Binding

When an Approov token contains the key pay, its value is a base64 encoded sha256 hash of some unique identifier in the request, that we may want to bind with the Approov token, in order to enhance the security on that request, like an Authorization token.

{
    "exp": 123456789, # required - the timestamp for when the token expires.
    "pay":"f3U2fniBJVE04Tdecj0d6orV9qT9t52TjfHxdUqDBgY=" # optional - a sha256 hash of the token binding, encoded with base64.
}

The token binding in an Approov token is the one in the pay key:

"pay":"f3U2fniBJVE04Tdecj0d6orV9qT9t52TjfHxdUqDBgY="

ALERT:

Please bear in mind that the token binding is not meant to pass application data to the API server.

System Clock

In order to correctly check for the expiration times of the Approov tokens it is very important that the Python Flask server is automatically synchronizing the system clock over the network with an authoritative time source. In Linux this is usually done with a NTP server.

Requirements

The example code in this walk-through is based on the Approov Shapes API server that uses Python 3 and Flask. The full source code can be found in this Github repository.

Docker is required only if you want to use the docker environment provided by the stack bash script, which is a wrapper around docker commands.

Postman is the tool we recommend to be used when simulating the queries against the API, but feel free to use any other tool of your choice.

The Docker Stack

We recommend the use of the included Docker stack to play with this Approov integration.

For details in how to use it you need to follow the setup instructions in the Approov Shapes API Server walk-through, but feel free to use your local environment to play with this Approov integration.

The Postman Collection

Import this Postman collection which contains all the API endpoints for the Approov Shapes API Server and we strongly recommend you follow this walk-through after finishing the Approov integration that we are about to start.

The Approov tokens used in the headers of this Postman collection were generated by the Approov CLI Tool and they cover all necessary scenarios, but feel free to generate some more valid and invalid tokens, with different expire times and with token bindings. Some examples of using it can be found here.

Install Dependencies

If not already using the packages pyjwt and python-dotenv in your Python Flask API project, please add them:

pip3 install pyjwt python-dotenv

Original Server

Let’s use the original-server.py as an example for a current server where we want to add Approov to protect some or all the endpoints and we will add to it only the necessary code to integrate Approov. The end result can be seen in the approov-protected-server.py.

How to Integrate

We will learn how to go from the original-server.py to the approov-protected-server.py and how to configure the server.

In order to be able to check the Approov token the PyJWT library needs to know the secret used by the Approov cloud service to sign it. A secure way to do this is by passing it as an environment variable, as it can be seen here.

Next we need to define two core methods to be used during the Approov token check process. The method _decodeApproovToken() is to decode and simultaneously check the token with the library PyJWT and _checkApproovTokenBinding() is to check the optional token binding in the Approov Token. We also define some other methods to help with the Approov integration and these are probably the ones you may want to customize for your use case.

The token binding in the Approov Token payload is optional, but when present it needs to be a base64 encoded string from a hash of some value you want to tie up with the Approov token. A good example is to bind the user authentication token with the Approov token, but your needs and requirements may be different.

Let’s breakdown the implementation of the approov-protected-server.py to make it easier to adapt to your current project.

Import Dependencies

We require the dependencies we installed before, plus some more system dependencies:

# System packages
from base64 import b64decode, b64encode
from os import getenv
from hashlib import sha256

# Third part packages
import jwt

Setup Environment

If you don’t have already an .env file, then you need to create one in the root of your project by using this .env.example as your starting point.

The .env file must contain these four variables:

APPROOV_BASE64_SECRET=h+CX0tOzdAAR9l15bWAqvq7w9olk66daIH+Xk+IAHhVVHszjDzeGobzNnqyRze3lw/WVyWrc2gZfh3XXfBOmww==
APPROOV_LOGGING_ENABLED=true
APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN=true
APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN_BINDING=true

Now we can read them from our code, as is done here:

APPROOV_BASE64_SECRET = getenv('APPROOV_BASE64_SECRET')

APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN = True
_approov_enabled = getenv('APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN', 'True').lower()
if _approov_enabled == 'false':
    APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN = False

APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN_BINDING = True
_abort_on_invalid_token_binding = getenv('APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN_BINDING', 'True').lower()
if _abort_on_invalid_token_binding == 'false':
    APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN_BINDING = False

APPROOV_LOGGING_ENABLED = True
_approov_logging_enabled = getenv('APPROOV_LOGGING_ENABLED', 'True').lower()
if _approov_logging_enabled == 'false':
    APPROOV_LOGGING_ENABLED = False

Methods

Let’s start by adding this method to enable logging for Approov specific occurrences;

def _logApproov(message):
    if APPROOV_LOGGING_ENABLED is True:
    log.info(message)

Next we need to add this method to decode the Approov token;

def _decodeApproovToken(approov_token):
    try:
        # Decode the approov token, allowing only the HS256 algorithm and using
        # the approov base64 encoded SECRET
        approov_token_decoded = jwt.decode(approov_token, b64decode(APPROOV_BASE64_SECRET), algorithms=['HS256'])

        return approov_token_decoded

    except jwt.InvalidSignatureError as e:
        _logApproov('APPROOV JWT TOKEN INVALID SIGNATURE: %s' % e)
        return None
    except jwt.ExpiredSignatureError as e:
        _logApproov('APPROOV JWT TOKEN EXPIRED: %s' % e)
        return None
    except jwt.InvalidTokenError as e:
        _logApproov('APPROOV JWT TOKEN INVALID: %s' % e)
        return None

Now we need to add this method to get the Approov token and validate it at each endpoint we want to protect;

def _getApproovToken():

    message = 'REQUEST WITH APPROOV TOKEN HEADER EMPTY OR MISSING'

    approov_token = _getHeader('approov-token')

    if _isEmpty(approov_token):

        if APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN is True:
            _logApproov('REJECTED ' + message)
            abort(make_response(jsonify(BAD_REQUEST_RESPONSE), 400))

        _logApproov(message)

        return None

    approov_token_decoded = _decodeApproovToken(approov_token)

    if _isEmpty(approov_token_decoded):
        return None

    return approov_token_decoded

We also need to add this method to handle requests with an invalid Approov token;

def _handleApproovProtectedRequest(approov_token_decoded):

    message = 'REQUEST WITH VALID APPROOV TOKEN'

    if not approov_token_decoded:
        message = 'REQUEST WITH INVALID APPROOV TOKEN'

    if APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN is True and not approov_token_decoded:
        _logApproov('REJECTED ' + message)
        abort(make_response(jsonify(BAD_REQUEST_RESPONSE), 401))

    _logApproov('ACCEPTED ' + message)

Then we need this method to validate the token binding in the Approov token;

def _checkApproovTokenBinding(approov_token_decoded, token_binding_header):
    if _isEmpty(approov_token_decoded):
        return False

    if 'pay' in approov_token_decoded:

        # We need to hash and base64 encode the token binding header, because that's how it was included in the Approov
        # token on the mobile app.
        token_binding_header_hash = sha256(token_binding_header.encode('utf-8')).digest()
        token_binding_header_encoded = b64encode(token_binding_header_hash).decode('utf-8')

        return approov_token_decoded['pay'] == token_binding_header_encoded

    return False

Finally we need this method to handle the validation result for the token binding in the Approov token.

def _handlesApproovTokenBindingVerification(approov_token_decoded, token_binding_header):

    if not 'pay' in approov_token_decoded:
        message = 'REQUEST WITH APPROOV TOKEN MISSING THE CLAIM TO VERIFY THE TOKEN BINDING'

        if APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN_BINDING is True:
            _logApproov('REJECTED ' + message)
            abort(make_response(jsonify(BAD_REQUEST_RESPONSE), 401))
        else:
            _logApproov('ACCEPTED ' + message)
            return

    message = 'REQUEST WITH VALID TOKEN BINDING IN THE APPROOV TOKEN'

    valid_token_binding = _checkApproovTokenBinding(approov_token_decoded, token_binding_header)

    if not valid_token_binding:
        message = 'REQUEST WITH INVALID TOKEN BINDING IN THE APPROOV TOKEN'

    if not valid_token_binding and APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN_BINDING is True:
        _logApproov('REJECTED ' + message)
        abort(make_response(jsonify(BAD_REQUEST_RESPONSE), 401))

    _logApproov('ACCEPTED ' + message)

Endpoints

To protect specific endpoints in a current server we only need to add the Approov token check for each endpoint we want to protect, as we have done in the shapes endpoint;

@api.route("/v2/shapes")
def shapesV2():

    # Will get the Approov JWT token from the header, decode it and on success
    # will return it, otherwise None is returned.
    approov_token_decoded = _getApproovToken()

    # If APPROOV_ABORT_REQUEST_ON_INVALID_TOKEN is set to True it will abort the request
    # when the decoded approov token is empty.
    _handleApproovProtectedRequest(approov_token_decoded)

    return _buildShapeResponse()

or if using the token binding in the Approov token, as we have done in the forms endpoint.

Code here

Approov in Action

Let's see how to query the Python Flask API from Postman with the collection we told you to install in the requirements section.

NOTE:

For your convenience the Postman collection includes a token that only expires in the very distant future for this call "Approov Token with valid signature and expire time". For the call "Expired Approov Token with valid signature" an expired token is also included.

Postman view with an Approov token correctly signed and not expired:

Postman view with token correctly signed but this time it is expired:

Shell view with the logs for the above requests:

Request Overview:

We used an helper script to generate an Approov Token that was valid for 1 minute.

In Postman we performed 2 requests with the same token and the first one was successful, but the second request, performed 2 minutes later, failed with a 401 response because the token has already expired as we can see by the log messages in the shell view.

Play Time for the Approov Shapes API Server

If you have not done it already, now is the time to follow the Approov Shapes API Server walk-through to get a feel for how all this works.

The Approov Shapes API Server contains endpoints with and without the Approov protection. The protected endpoints differ in the sense that one uses the token binding in the Approov token.

We will demonstrate how to call each API endpoint with screen-shots from Postman and from the shell. Postman is used here as an easy way to demonstrate how you can play with the Approov integration in the API server. If you want to see a real demo of how Approov would work in production, please to request a demo here.

The Code Difference

IYou can use git to compare the original-server.py with the approov-protected-server.py:

git diff --no-index servers/shapes-api/original-server.py servers/shapes-api/approov-protected-server.py

As you can see the Approov integration in a current server is simple, easy and is done with just a few lines of code.

Production

In order to protect the communication between your mobile app and the API server is important to only communicate over a secure communication channel, also known as https.

Please bear in mind that https on its own is not enough, certificate pinning must be also used to pin the connection between the mobile app and the API server in order to prevent Man-in-the-Middle Attacks.

We do not use https and certificate pinning in this Approov integration example because we want to be able to run the Approov Shapes API Server in localhost.

However in production is mandatory to configure and implement certificate pinning, that is made easy by using the dynamic pinning feature built-in the Appoov CLI tool, that allows to update the pins without the need to release a new version of your mobile app.