Build and Secure an API in Python with FastAPI | Okta Developer

A “professional” attack¶

Of course, the attackers would not try all this by hand, they would write a program to do it, possibly with thousands or millions of tests per second. And would get just one extra correct letter at a time.

Authenticate¶

Click the “Authorize” button.

Use the credentials:

Building a new fastapi project

In this section, you’ll create a new FastAPI project and add a single, unprotected endpoint to your API.

Before you get started, make sure your computer has Python 3.6 installed. FastAPI uses the typing and asynchronous features in Python, so earlier versions of the language won’t run it.

Configuring our auth service

As the FastAPI docs state, security is often a complex and “difficult” issue. Readers who are unfamiliar with standard authentication practices are highly encouraged to read the FastAPI docs on security, as they cover a large amount of jargon that will be useful in understanding the parts to come.

Documentation

In-depth documentation can be found at fastapi-login.readthedocs.io
Some examples can be found here

Github repo

All code up to this point can be found here:

Special thanks to Joao Ant for correcting errors in the original code.

Install fastapi

Start by creating a new Python project and using pip to install FastAPI and Uvicorn as your asynchronous server:

The dependencies will be added to your requirements.txt file.

Installation

$ pip install fastapi-login

Json web tokens

Readers who unfamiliar with token-based authentication are highly encouraged to read JWT.io’s introductory article found here. For the initiated, the basics are as follows:

A JWT is simply an encoded string containing three parts separated by periods. The example provided in the previously mentioned article states that they follow this general structure: xxxxxx.yyyyyy.zzzzz. These three parts are called the header, payload, and signature, respectively.

Learn more about python and rest apis

In this post, you’ve seen how to use FastAPI to build a REST API endpoint that uses an external authorization server to generate and validate access tokens. You’ve seen some of the key features of FastAPI in action, including dependency injection, the OpenAPI documentation, type hinting, and OAuth implementation.

FastAPI is a great option for building secure and performant backend systems. While there’s much more to building a robust production API, including testing, handling POST and PUT endpoints, and connecting to a database for persistence, I hope this tutorial helps you get started.

If you want to learn more about FastAPI, I suggest the following resources:

To learn more about building an app that can communicate with FastAPI, I suggest using one of these tutorials:

Oauth2passwordrequestform¶

First, import OAuth2PasswordRequestForm, and use it as a dependency with Depends in the path operation for /token:

fromtypingimportUnionfromfastapiimportDepends,FastAPI,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormfrompydanticimportBaseModelfake_users_db={"johndoe":{"username":"johndoe","full_name":"John Doe","email":"[email protected]","hashed_password":"fakehashedsecret","disabled":False,},"alice":{"username":"alice","full_name":"Alice Wonderson","email":"[email protected]","hashed_password":"fakehashedsecret2","disabled":True,},}app=FastAPI()deffake_hash_password(password:str):return"fakehashed" passwordoauth2_scheme=OAuth2PasswordBearer(tokenUrl="token")classUser(BaseModel):username:stremail:Union[str,None]=Nonefull_name:Union[str,None]=Nonedisabled:Union[bool,None]=NoneclassUserInDB(User):hashed_password:strdefget_user(db,username:str):ifusernameindb:user_dict=db[username]returnUserInDB(**user_dict)deffake_decode_token(token):# This doesn't provide any security at all# Check the next versionuser=get_user(fake_users_db,token)returnuserasyncdefget_current_user(token:str=Depends(oauth2_scheme)):user=fake_decode_token(token)ifnotuser:raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Invalid authentication credentials",headers={"WWW-Authenticate":"Bearer"},)returnuserasyncdefget_current_active_user(current_user:User=Depends(get_current_user)):ifcurrent_user.disabled:raiseHTTPException(status_code=400,detail="Inactive user")returncurrent_user@app.post("/token")asyncdeflogin(form_data:OAuth2PasswordRequestForm=Depends()):user_dict=fake_users_db.get(form_data.username)ifnotuser_dict:raiseHTTPException(status_code=400,detail="Incorrect username or password")user=UserInDB(**user_dict)hashed_password=fake_hash_password(form_data.password)ifnothashed_password==user.hashed_password:raiseHTTPException(status_code=400,detail="Incorrect username or password")return{"access_token":user.username,"token_type":"bearer"}@app.get("/users/me")asyncdefread_users_me(current_user:User=Depends(get_current_active_user)):returncurrent_user

fromfastapiimportDepends,FastAPI,HTTPException,statusfromfastapi.securityimportOAuth2PasswordBearer,OAuth2PasswordRequestFormfrompydanticimportBaseModelfake_users_db={"johndoe":{"username":"johndoe","full_name":"John Doe","email":"[email protected]","hashed_password":"fakehashedsecret","disabled":False,},"alice":{"username":"alice","full_name":"Alice Wonderson","email":"[email protected]","hashed_password":"fakehashedsecret2","disabled":True,},}app=FastAPI()deffake_hash_password(password:str):return"fakehashed" passwordoauth2_scheme=OAuth2PasswordBearer(tokenUrl="token")classUser(BaseModel):username:stremail:str|None=Nonefull_name:str|None=Nonedisabled:bool|None=NoneclassUserInDB(User):hashed_password:strdefget_user(db,username:str):ifusernameindb:user_dict=db[username]returnUserInDB(**user_dict)deffake_decode_token(token):# This doesn't provide any security at all# Check the next versionuser=get_user(fake_users_db,token)returnuserasyncdefget_current_user(token:str=Depends(oauth2_scheme)):user=fake_decode_token(token)ifnotuser:raiseHTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Invalid authentication credentials",headers={"WWW-Authenticate":"Bearer"},)returnuserasyncdefget_current_active_user(current_user:User=Depends(get_current_user)):ifcurrent_user.disabled:raiseHTTPException(status_code=400,detail="Inactive user")returncurrent_user@app.post("/token")asyncdeflogin(form_data:OAuth2PasswordRequestForm=Depends()):user_dict=fake_users_db.get(form_data.username)ifnotuser_dict:raiseHTTPException(status_code=400,detail="Incorrect username or password")user=UserInDB(**user_dict)hashed_password=fake_hash_password(form_data.password)ifnothashed_password==user.hashed_password:raiseHTTPException(status_code=400,detail="Incorrect username or password")return{"access_token":user.username,"token_type":"bearer"}@app.get("/users/me")asyncdefread_users_me(current_user:User=Depends(get_current_active_user)):returncurrent_user

OAuth2PasswordRequestForm is a class dependency that declares a form body with:

• An optional client_id (we don’t need it for our example).
• An optional client_secret (we don’t need it for our example).

P.s. исходники

Вот собственно и все, репозиторий с исходниками из поста можно посмотреть на GitHub.

Password hashing¶

“Hashing” means: converting some content (a password in this case) into a sequence of bytes (just a string) that looks like gibberish.

Whenever you pass exactly the same content (exactly the same password) you get exactly the same gibberish.

But you cannot convert from the gibberish back to the password.

Scope¶

The spec also says that the client can send another form field “scope”.

The form field name is scope (in singular), but it is actually a long string with “scopes” separated by spaces.

Each “scope” is just a string (without spaces).

They are normally used to declare specific security permissions, for example:

Securing fastapi with okta

To demonstrate some of the more advanced features of the FastAPI framework, I’ll show you how to build a protected endpoint that uses the client credentials flow to authorize access. By the end of this tutorial, you will be able to enter your Okta client ID and secret into FastAPI’s interactive docs to get an access token.

If you’d like to run the final application, the code is available on GitHub, or you can follow along for step-by-step instructions.

Security and new packages

Before we get going, we’ll need to install a few new packages and rebuild our docker container. Let’s get that out of the way now.

requirements.txt

We’re installing two new packages here:

Go ahead and build the docker container with the new packages.

While we wait for our container to build, let’s chat about security.

Setting up a new application in okta

Before you create any endpoints in your FastAPI application, you’ll need to create a new application in Okta and get your Authorization Server’s issuer URL and audience. As you go through these steps, add the Okta environment variables to a new file in your application called .env. You’ll see how to use these variables later in the tutorial.

To create a new server application, log in to your Okta account and go to Applications and click the Add Application button in the top left. Select Service, Machine-to-Machine, then click Next.

Enter a name for your application and click Next again.

Copy the Client ID and Client Secret from this page and add them to your FastAPI application’s .env file as OKTA_CLIENT_ID and OKTA_CLIENT_SECRET respectively.

Your .env file should look like the example below, with your OKTA_CLIENT_ID and OKTA_CLIENT_SECRET values filled out:

Next, go to API > Authorization Servers. Copy the Issuer URI and Audience, and add them as the OKTA_ISSUER and OKTA_AUDIENCE environment variables in your .env file.

Your .env file should now look something like this:

Now that you have your environment variables, you need to create a custom scope for your authorization server. Your FastAPI application will request a token with this scope. Click the pencil icon to edit the authorization server.

Click the Scopes tab and then the Add Scopes button.

Timing attacks¶

But what’s a “timing attack”?

Usage

To begin we have to setup our FastAPI app:

fromfastapiimportFastAPISECRET='your-secret-key'app=FastAPI()

To obtain a suitable secret key you can run import os; print(os.urandom(24).hex()).

Now we can import and setup the LoginManager, which will handle the process of
encoding and decoding our Json Web Tokens.

fromfastapi_loginimportLoginManagermanager=LoginManager(SECRET,token_url='/auth/token')

Validating access tokens locally

Because access tokens are generally short-lived (an hour by default), you might prefer to validate the tokens locally. This method is slightly less secure because you can’t be sure that the access token hasn’t been revoked remotely, but on the other hand, you don’t have to use your Okta client secret to validate the token locally.

To validate the access token locally, install the Okta JWT Python package:

Next, import the package’s validate_token function and update the validate function in your main.py file:

When you call the /items endpoint, the API will decode the JWT and validate it locally. The decoded JWTs are cached, so subsequent requests will be faster than the first one.

Validating access tokens remotely

There are two ways to validate JWT access tokens generated by Okta. The first method you’ll see uses the Okta authorization server’s /inspect endpoint to check the token. The advantage of this method is that you will know if the token has been revoked; the downside is that it’s slower than validating the JWT locally.

To validate access tokens remotely, update your validate function, and add the following validate_remotely function:

defvalidate_remotely(token,issuer,clientId,clientSecret):headers={'accept':'application/json','cache-control':'no-cache','content-type':'application/x-www-form-urlencoded',}data={'client_id':clientId,'client_secret':clientSecret,'token':token,}url=issuer '/v1/introspect'response=httpx.post(url,headers=headers,data=data)returnresponse.status_code==httpx.codes.OKandresponse.json()['active']defvalidate(token:str=Depends(oauth2_scheme)):res=validate_remotely(token,config('OKTA_ISSUER'),config('OKTA_CLIENT_ID'),config('OKTA_CLIENT_SECRET'))ifres:returnTrueelse:raiseHTTPException(status_code=400)

Now, refresh the docs, generate an access token by entering your Okta client ID and secret again, and call the /items endpoint. You should see a 200 Success response again with the items. This endpoint is now calling the Okta authorization server on every request.

Wrapping up and resources

There’s a lot packed into this one post, and we’re only just getting started. In the next post, we’ll dive a little further into security and handle login, along with some authorization dependencies.

Аутентификация и контроль доступа

Теперь, когда в нашей базе данных есть пользователи, все готово для того чтобы настроить аутентификацию приложения. Добавим эндпоинт, который принимает имя пользователя и пароль и возвращает токен. Обновим файл

Запускаем приложение и подключаем бд

Создадим файл

main.py

Схема базы данных и миграции

Прежде всего, с помощью SQLAlchemy Expression Language, опишем схему базы данных. Создадим файл

Тестирование

Тесты мы будем писать на pytest:

$ pip install pytest

Для тестирования эндпоинтов FastAPI предоставляет специальный инструмент TestClient.

Напишем тест для эндпоинта, который не требует подключения к базе данных:

Create a new endpoint

For this tutorial, you’ll use a single main.py file that contains all your routes. You can break this file up as your application grows, but since you’ll just be adding a couple of endpoints, you don’t need to worry about that now.

Create the main.py file in your project’s root directory and add the following:

Creating a protected endpoint

Now that you have an endpoint that generates a token, you are ready to create a new endpoint that checks the token before granting access.

Before you worry about token validation, create the new endpoint and validate function. You’ll add the logic to this function in the next step, but for testing purposes, you can simply return True.

The new /items endpoint includes a response_model definition. This allows FastAPI to generate documentation for your endpoint with a sample response.

It uses FastAPI’s dependency injection pattern to call the validate function. In turn, that function injects the oauth2_scheme, which extracts the access token for you.

Agenda

• Setup
• FastAPI app skeleton Auth logic

Update the dependencies¶

https://www.youtube.com/watch?v=nBghs4eTmik

Now we are going to update our dependencies.