hello_server_protected.py

from flask import Flask, jsonify, request, abort, g, make_response

# @link https://github.com./jpadilla/pyjwt/
import jwt
import base64
import hashlib

# @link https://github.com./theskumar/python-dotenv
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv(), override=True)
from os import getenv

import logging
log = logging.getLogger(__name__)

api = Flask(__name__)

# Token secret value obtained with the Approov CLI tool:
#  - approov secret -get
approov_base64_secret = getenv('APPROOV_BASE64_SECRET')

if approov_base64_secret == None:
    raise ValueError("Missing the value for environment variable: APPROOV_BASE64_SECRET")

APPROOV_SECRET = base64.b64decode(approov_base64_secret)

@api.before_request
def _verifyApproovToken():
    approov_token = request.headers.get("Approov-Token")

    # If we didn't find a token, then reject the request.
    if approov_token is None or approov_token == "":
        # You may want to add some logging here.
        log.error('Approov Token Check: The Approov Token is empty...')
        return abort(make_response({}, 401))

    try:
        # Decode the Approov token explicitly with the HS256 algorithm to
        # avoid the algorithm None attack.
        g.approov_token_claims = jwt.decode(approov_token, APPROOV_SECRET, algorithms=['HS256'])

        # When doesn't occur an exception we have a valid Aproov Token

    except jwt.ExpiredSignatureError as e:
        # You may want to add some logging here.
        log.error(e)
        return abort(make_response({}, 401))

    except jwt.InvalidTokenError as e:
        # You may want to add some logging here.
        log.error(e)
        return abort(make_response({}, 401))

    except:
        log.error('Approov Token check failed with unknown error...')
        return abort(make_response({}, 401))

@api.before_request
def _verifyApproovTokenBinding():
    # We use the Authorization token, but feel free to use another header in
    # the request. Bear in mind that it needs to be the same header used in
    # the mobile app to bind the request with the Approov token.
    token_binding_header = request.headers.get("Authorization")

    if not token_binding_header:
        # You may want to add some logging here.
        log.error('Approov Token Binding: Missing the Approov token binding header...')
        return abort(make_response({}, 401))

    if not 'approov_token_claims' in g:
        # You may want to add some logging here.
        log.error('Approov Token Binding: Missing the Approov token claims...')
        return abort(make_response({}, 401))

    if not 'pay' in g.approov_token_claims:
        # You may want to add some logging here.
        log.error('Approov Token Binding: Missing the `pay` in the Approov token claims...')
        return abort(make_response({}, 401))

    # 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 = hashlib.sha256(token_binding_header.encode('utf-8')).digest()
    token_binding_header_encoded = base64.b64encode(token_binding_header_hash).decode('utf-8')

    if g.approov_token_claims['pay'] != token_binding_header_encoded:
        log.error('Approov Token binding check failed...')
        return abort(make_response({}, 401))

@api.route("/")
def hello():
    return jsonify({"message": "Hello, World!"})