Secure a FastAPI Server with Asgardeo

FastAPI is a modern, fast, web framework for building APIs with Python 3.6+. FastAPI uses a web server called ASGI (Asynchronous Server Gateway) Interface, it makes faster performance in your applications as it doesn’t have to wait for the other ones to complete before they are processed. You can find out more details from their official documentation.

From this article I’ll be explaining you on how to secure a FastAPI endpoint using a Opaque token of Asgardeo.

Step 1 — Register an OpenID application in Asgardeo.

1. Login to Asgardeo Console.
2. Go to Develop section and then select Applications.
3. Click New Application.
4. Choose Standard-Based Application.
5. Provide any application name and Choose OpenID Connect click register.

6. Then go to Protocol tab and obtain a Client ID and Client Secret.

7. Then update the allowed grant types by selecting the Client Credentials and Password. And also make sure Access Token type is selected as Opaque as well.

Step 2 — Create a customer user in Asgardeo.

1. Go to Manage section and then select Users.
2. Click New User and select Customer.
3. Enter the customer Email, First Name, Last Name and select Set a temporary password for user and enter a password and then click Finish.

Step 3 — Getting started with FastAPI.

Now we have the necessary configuration from the Asgardeo side. Lets Start coding.

1. First You need to install the FastAPI using the pip.

pip install fastapi

2. Then we need install uvicorn to work as the server:

pip3 install "uvicorn[standard]"

3. Lets create the public endpoint first which is accessible without an access token.

# main.py
from fastapi import FastAPI
from fastapi_versioning import VersionedFastAPI, version
app = FastAPI()
@app.get("/")
@version(1)
def public():
     result = {
         "status": "Success",
         "msg": ("Hello from a public endpoint.")
      }
     return result
app = VersionedFastAPI(app)

Start the server using uvicorn main:app --relaod and try out the endpoint using the below command.

curl --location --request GET 'http://127.0.0.1:8000/v1_0/api/public'

Step 4 — Create a secured endpoint with Asgardeo.

Opaque: These types of tokens are plain text tokens. If a resource server wants to know information related to opaque token, it has to query introspection endpoint and get the information related to tokens.

As we are using the Opaque token, we have to have a method implemented in our code to query the introspection endpoint to validate the access token.

import requests
import os
org = os.getenv("ORG")
internal_client_id = os.getenv("CLIENT_ID")
internal_client_secret = os.getenv("CLIENT_SECRET")
def introspect_token(token_string):
   url = 'https://api.asgardeo.io/t/' + org + '/oauth2/introspect'
   data = {'token': token_string, 'token_type_hint': 'access_token'}
   auth = (internal_client_id, internal_client_secret)
   resp = requests.post(url, data=data, auth=auth)
   resp.raise_for_status()
   active_status = resp.json().get("active")
   return active_status

When calling the introspect_token method with the access token, It will send the post request to the introspection endpoint of your organization. Then we’ll be using the active key of the response json to validate the token. More information regarding the token validation can be found at the Asgardeo docs.

Lets create the private endpoint just like the public endpoint. Here we’ll be using the FastAPIs HTTPBearer schema to get the authorization header. This is available under the fastapi.security.

from fastapi.security import HTTPBearer
# Scheme for the Authorization header                             token_auth_scheme = HTTPBearer()
@app.get("/api/private")
def private(response: Response, token: str = Depends(token_auth_scheme)): 
   
   """A valid access token is required to access this route"""
   authorized = introspect_token(token.credentials)
   result = {
      "status": "Success",
      "msg": ("Hello from a private endpoint protected with Asgardeo opaque token.")
   }
   if authorized == True:
      return result
   else:
      return unauthorizedResponse()
def unauthorizedResponse():
   result = {
      "status": "Unauthorized",
      "msg": "Token is invalid or expired."
   }
   return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content=result)

Great…!!

Here we have created our first endpoint secured with the Asgardeo Opaque token.

Let try out the endpoint with a valid Opaque token. Use the following command to get a valid access token for the user you have created in the Step 2.

curl --location --request POST 'https://api.asgardeo.io/t/<ORG>/oauth2/token?grant_type=password&username=<USERNAME>&password=<PASSWORD>' \
--header 'Content-Type: application/json' \
--header 'accept: application/json' \
--header 'Authorization: Basic BASE46_ENCODING<client_id:client_secret>'

Then try out the endpoint with the token obtained from the above.

curl --location --request GET 'http://127.0.0.1:8000/v1_0/api/private' \
--header 'Authorization: Bearer <OPAQUE_TOKEN>'

If the token you have entered is valid, you will get the below response

{
    "status": "Success",
    "msg": "Hello from a private endpoint protected with Asgardeo opaque token."
}

You can find the full code from below.

If you want to tryout the sample please use the following command.

docker run -p 8000:80 -e ORG="<ORG>" -e CLIENT_ID="<CLIENT_ID>" -e CLIENT_SECRET="<CLIENT_SECRET>" wathsara/asgardeo-fastapi-sample:1.0.0

References

Home

FastAPI

Thank you for reading my article. Have a Good Day!!