Validate the Access token with FastAPI

This tutorial covers how to validate that the Access JWT is on requests made to FastAPI apps.

Time to complete: 15 minutes

Prerequisites

• A self-hosted Access application for your FastAPI app
• The AUD tag for your Access application

1. Create a validation function

1. In your FastAPI project, create a new file called cloudflare.py that contains the following code:

from fastapi import Request, HTTPException

# The Application Audience (AUD) tag for your application
POLICY_AUD = "XXXXX"

# Your CF Access team domain
TEAM_DOMAIN = "https://<your-team-name>.cloudflareaccess.com"
CERTS_URL = "{}/cdn-cgi/access/certs".format(TEAM_DOMAIN)

async def validate_cloudflare(request: Request):
    """
    Validate that the request is authenticated by Cloudflare Access.
    """
    if verify_token(request) != True:
        raise HTTPException(status_code=400, detail="Not authenticated properly!")


def _get_public_keys():
    """
    Returns:
        List of RSA public keys usable by PyJWT.
    """
    r = requests.get(CERTS_URL)
    public_keys = []
    jwk_set = r.json()
    for key_dict in jwk_set["keys"]:
        public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(key_dict))
        public_keys.append(public_key)
    return public_keys


def verify_token(request):
    """
    Verify the token in the request.
    """
    token = ""

    if "CF_Authorization" in request.cookies:
        token = request.cookies["CF_Authorization"]
    else:
        raise HTTPException(status_code=400, detail="missing required cf authorization token")

    keys = _get_public_keys()

    # Loop through the keys since we can't pass the key set to the decoder
    valid_token = False
    for key in keys:
        try:
            # decode returns the claims that has the email when needed
            jwt.decode(token, key=key, audience=POLICY_AUD, algorithms=["RS256"])
            valid_token = True
            break
        except:
            raise HTTPException(status_code=400, detail="Error decoding token")
    if not valid_token:
        raise HTTPException(status_code=400, detail="Invalid token")

    return True

2. Use the validation function in your app

You can now add the validation function as a dependency in your FastAPI app. One way to do this is by creating an APIRouter instance ↗. The following example executes the validation function on each request made to paths that start with /admin:

from fastapi import APIRouter, Depends, HTTPException
from cloudflare import validate_cloudflare

router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(validate_cloudflare)]
    responses={404: {"description": "Not found"}},
)

@router.get("/")
async def root():
    return {"message": "Hello World"}