Hello everyone, In this beginner-friendly tutorial, we will go through the steps of implementing [JWT](https://jwt.io/) Authentication in [FastAPI](https://fastapi.tiangolo.com/) using [fastapi-another-jwt-auth](https://pypi.org/project/fastapi-another-jwt-auth/). We will be using [Pydantic](https://pydantic-docs.helpmanual.io/) for data validation and [Passlib](https://passlib.readthedocs.io/en/stable/) for password hashing. We are going to write some code, so let's jump right into it. Photo by [Markus Spiske](https://unsplash.com/@markusspiske?utm_content=creditCopyText&utm_medium=referral&utm_source=unsplash) on [Unsplash](https://unsplash.com/photos/text-6pflEeSzGUo?utm_content=creditCopyText&utm_medium=referral&utm_source=unsplash) ## Introduction to JWT JWT (JSON Web Token) is basically an encoded JSON object with which we can securely transfer data between two parties. In our case these two parties are the client and the server. ### How does it work? We call JWTs "self-contained" because they contain all the information needed to authenticate a user. This means that we don't need to store any information about the user on the server-side. This is a huge advantage because it allows us to scale our application easily. A simple JWT would contain a `header`, `payload` and a `signature` separated by dots. #### header It will contain the algorithm used to generate the signature and the type of the token. ```json { "alg": "HS256", "typ": "JWT" } ``` #### payload It will contain the data we want to transfer. [Some of the fields are reserved](https://auth0.com/docs/secure/tokens/json-web-tokens/json-web-token-claims#registered-claims) and have special meanings. An example to a payload would be: ```json { "sub": "f1677c52-0f8b-4f31-baaf-cc0b444a5b99", // subject "iat": 1700776495, // issued at "nbf": 1700776495, // not before "jti": "fe71746e-397d-4b81-8732-76f8cd042fa3", // JWT ID "exp": 1700777695, // expiration time "type": "access" // type of the token } ``` #### signature The signature is generated by combining the header and the payload with a secret key using the algorithm specified in the header. This signature is going to be used to verify that the data has not been changed. ## Setting up the project In this tutorial, we are going to utilize the access and refresh token method. The access token will be used to access the protected routes and the refresh token will be used to get a new access token once the access token expires. [For further reading on this topic.](https://stackoverflow.com/questions/3487991/why-does-oauth-v2-have-both-access-and-refresh-tokens) ### Directory structure ```ansi |___ .env |___ src | |___ __init__.py | |___ main.py | |___ database.py | |___ config.py | |___ oauth.py | |___ auth | | |___ __init__.py | | |___ dependencies.py | | |___ services.py | | |___ schemas.py | | |___ models.py | | |___ router.py | | |___ utils.py | |___ dashboard | |___ __init__.py | |___ router.py ``` ### Initialize FastAPI We initialize our FastAPI application in `main.py` with some default development settings. #### main.py ```python from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from src.database import Base, engine from src.auth.models import * from src.auth.router import router as auth_router from src.dashboard.router import router as dashboard_router origins = ["http://localhost:3000"] app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) app.include_router(auth_router, tags=["auth"], prefix="/api/auth") app.include_router(dashboard_router, tags=["dashboard"], prefix="/api/dashboard") ``` #### .env Now let's take a look at what we need in our `.env` file. ```env ... JWT_PRIVATE_KEY=... JWT_PUBLIC_KEY=... JWT_ALGORITHM=RS256 REFRESH_TOKEN_EXPIRE_MINUTES=60 ACCESS_TOKEN_EXPIRE_MINUTES=15 ... ``` #### config.py We are going to use the [asymmetric RSA algorithm](https://glitchcorp.github.io/fastapi-another-jwt-auth/advanced-usage/asymmetric/) for our JWTs. That means we need to generate a private and a public key. We can do that using this website: [cryptotool.net](https://cryptotools.net/rsagen) We add our environment variables to `config.py` ```python {12-14, 16,17} from pydantic import BaseSettings, PostgresDsn class Settings(BaseSettings): app_name: str = "A simple JWT Authentication Tutorial" POSTGRES_USER: str POSTGRES_PASSWORD: str POSTGRES_SERVER: str POSTGRES_PORT: str POSTGRES_DB: str JWT_PRIVATE_KEY: str JWT_PUBLIC_KEY: str JWT_ALGORITHM: str REFRESH_TOKEN_EXPIRE_MINUTES: int ACCESS_TOKEN_EXPIRE_MINUTES: int CLIENT_ORIGIN: str database_url: PostgresDsn = "postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_SERVER}:{POSTGRES_PORT}/{POSTGRES_DB}" class Config: env_file = ".env" settings = Settings() ``` #### oauth.py Now let's take a look at how this JWT package works. ```python showLineNumbers {24-26} import base64 from typing import List from pydantic import BaseModel from sqlalchemy.orm import Session from fastapi import Depends, HTTPException, status from fastapi_another_jwt_auth import AuthJWT from fastapi_another_jwt_auth.exceptions import InvalidHeaderError, MissingTokenError from src.config import settings from src.auth.models import User from src.database import get_db class Settings(BaseModel): authjwt_algorithm: str = settings.JWT_ALGORITHM authjwt_decode_algorithms: List[str] = [settings.JWT_ALGORITHM] authjwt_token_location: set = {"cookies", "headers"} authjwt_access_cookie_key: str = "access_token" authjwt_refresh_cookie_key: str = "refresh_token" authjwt_cookie_csrf_protect: bool = False authjwt_public_key: str = base64.b64decode(settings.JWT_PUBLIC_KEY).decode("utf-8") authjwt_private_key: str = base64.b64decode(settings.JWT_PRIVATE_KEY).decode("utf-8") @AuthJWT.load_config def get_config(): return Settings() ``` We start with defining our settings. The settings include the algorithm used to generate the signature, the location of the token, the name of the cookies, the public and the private key. Finally we load the configuration. From now on, we can use the `AuthJWT` instance as dependency in our routes to deal with authentication and authorization. ```python showLineNumbers{30} class UserNotFound(Exception): pass def require_user(db: Session = Depends(get_db), Authorize: AuthJWT = Depends()): user_id = None try: Authorize.jwt_required() user_id = Authorize.get_jwt_subject() user = db.query(User).filter(User.id == user_id).first() if not user: raise UserNotFound("User no longer exists") except MissingTokenError as e: if "access" in str(e.message): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="You must refresh", ) except InvalidHeaderError as e: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Invalid Header Error", ) except Exception as e: error = e.__class__.__name__ if error == "UserNotFound": raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return user_id ``` We define a dependency function to be used in our routes. This function will be used to get the user id from the JWT. If the user is not found, we raise an exception. If the token is missing, we raise a 401 error. If the header is invalid, we raise a 403 error. ## Login route #### schemas.py Right before we start writing our authentication routes, let's create some basic schemas using pydantic to be used in the routes. ```python from pydantic import BaseModel, EmailStr, constr class UserSignin(BaseModel): email: EmailStr password: constr(min_length=8) class UserSignup(BaseModel): first_name: str last_name: str email: str password: constr(min_length=8) passwordConfirm: str ``` #### router.py This is the login flow we are going to take a look at now. ```python showLineNumbers /authenticate_user/ /Depends/ /AuthJWT/ ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES REFRESH_TOKEN_EXPIRE_MINUTES = settings.REFRESH_TOKEN_EXPIRE_MINUTES router = APIRouter() @router.post("/login", response_model=User) async def login( payload: UserSignin, response: Response, db: Session = Depends(get_db), Authorize: AuthJWT = Depends(), ): user = await authenticate_user(db, payload.email, payload.password) if not user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) ``` Let's break down what is going on so far. We are using the `Depends{:.fn}` function to get the database session and the `AuthJWT` instance. We are using the `authenticate_user{:.fn}` function to check if the user exists and the password is correct, which we are going to create now. ```python /get_user_by_email/ /verify_password/ from src.auth.services import get_user_by_email from src.auth.utils import verify_password async def authenticate_user(db: Session, email: str, password: str) -> bool | User: user = await get_user_by_email(db, email) if user is None: return False if not verify_password(password, user.hashed_password): return False return user ``` We are using the `get_user_by_email{:.fn}` function to get the user from the database. We verify the password using the `verify_password{:.fn}` function which comes from our `utils.py`. ```python /verify_password/ from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(raw_password, hashed_password): return pwd_context.verify(raw_password, hashed_password) ``` `verify_password{:.fn}` function takes the raw password and the hashed password which we store in our database and returns a boolean value. ```python showLineNumbers{20} /set_cookie/ access_token = create_access_token(Authorize=Authorize, user=user) refresh_token = create_refresh_token(Authorize=Authorize, user=user) response.set_cookie( "access_token", access_token, ACCESS_TOKEN_EXPIRE_MINUTES * 60, ACCESS_TOKEN_EXPIRE_MINUTES * 60, "/", None, False, True, "lax", ) response.set_cookie( "refresh_token", refresh_token, REFRESH_TOKEN_EXPIRE_MINUTES * 60, REFRESH_TOKEN_EXPIRE_MINUTES * 60, "/", None, False, True, "lax", ) return user ``` We go on with creating the access and the refresh tokens using the `create_access_token{:.fn}` and the `create_refresh_token{:.fn}` functions which we are going to create now. Then we set the cookies using the `set_cookie` method of the `response` object from fastAPI. Here is how the `set_cookie{:.fn}` method of the `response` object from fastAPI works. ```python def set_cookie( key: str, value: str = "", max_age: int | None = None, expires: datetime | str | int | None = None, path: str = "/", domain: str | None = None, secure: bool = False, httponly: bool = False, samesite: Literal['lax', 'strict', 'none'] | None = "lax" ) -> None: ``` ```python ... def create_access_token(user: User, Authorize: AuthJWT = Depends()): access_token = Authorize.create_access_token( subject=str(user.id), expires_time=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES), ) return access_token def create_refresh_token(user: User, Authorize: AuthJWT = Depends()): refresh_token = Authorize.create_refresh_token( subject=str(user.id), expires_time=timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES), ) return refresh_token ``` Notice that we are using the `user_id` as the subject of the JWT. Later we will use this subject to get the user from the database. This is it for the login route. Now let's take a look at the signup route which consists of two steps, creating the user and logging the user in. So it should get easier from here. ## Signup route ```python showLineNumbers{59} /check_user/ /create_user/ @router.post("/register", status_code=status.HTTP_201_CREATED, response_model=User) async def signup_user( payload: UserSignup, response: Response, db: Session = Depends(get_db), Authorize: AuthJWT = Depends(), ): checked_user = await check_user(db=db, email=payload.email) if checked_user: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="An account by that email already registered!", ) if payload.password != payload.passwordConfirm: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Passwords do not match!", ) new_user = await create_user(db=db, payload=payload) access_token = create_access_token(Authorize=Authorize, user=new_user) refresh_token = create_refresh_token(Authorize=Authorize, user=new_user) response.set_cookie( "access_token", access_token, ACCESS_TOKEN_EXPIRE_MINUTES * 60, ACCESS_TOKEN_EXPIRE_MINUTES * 60, "/", None, False, True, "lax", ) response.set_cookie( "refresh_token", refresh_token, REFRESH_TOKEN_EXPIRE_MINUTES * 60, REFRESH_TOKEN_EXPIRE_MINUTES * 60, "/", None, False, True, "lax", ) return user ``` We begin with checking if the user already exists using the `check_user{:.fn}` function. It is a simple function that checks if the user exists in the database using SQLAlchemy. If the user exists, we raise a 409 error. If the passwords do not match, we raise a 400 error. If everything is fine, we create the user using the `create_user{:.fn}` function. The rest is the same as the login route; we create the access and the refresh tokens and set the cookies. ## Refresh route Now that the user is logged in, we can take a look at the refresh route. As we talked about earlier, the access token expires earlier than the refresh token. So, we need a router to renew the access token. ```python showLineNumbers{106} {8,9} @router.get("/refresh", response_model=User) def refresh_token( response: Response, Authorize: AuthJWT = Depends(), db: Session = Depends(get_db), ): try: Authorize.jwt_refresh_token_required() user_id = Authorize.get_jwt_subject() if not user_id: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Could not identify token", ) user = get_user_by_id(db, user_id) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="The user belonging to this token does not exist.", ) access_token = Authorize.create_access_token( subject=str(user.id), expires_time=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES), ) except MissingTokenError as e: if "refresh" in e.message: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Missing refresh token!", ) raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=e) response.set_cookie( "access_token", access_token, ACCESS_TOKEN_EXPIRE_MINUTES * 60, ACCESS_TOKEN_EXPIRE_MINUTES * 60, "/", None, False, True, "lax", ) return user ``` We start with checking if the refresh token is missing. If it is, we raise a 403 error. If the refresh token is present and we can extract the `user_id` from the JWT, we create a new access token and set the cookie. ## Dashboard route Now we are going to take a look at once the user is logged in and tries to access a protected route. ```python router = APIRouter() @router.get("/dashboard") async def get_dashboard( user_id: str = Depends(require_user), db: Session = Depends(get_db) ): if not user_id: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found", ) # do something with the user return {"message": "Hello from the dashboard!"} ``` Note how we use the `require_user{:.fn}` function as a dependency to get the user id from the JWT. If the user is not found, we raise a 404 error. It is that simple to protecte a route from unauthorized users. ## Logout route ```python showLineNumbers{150} @router.post("/logout", status_code=status.HTTP_200_OK) async def logout( response: Response, Authorize: AuthJWT = Depends(), user_id: str = Depends(require_user), db: Session = Depends(get_db), ): if user_id is not None: try: Authorize.unset_jwt_cookies() except Exception as e: return {"status": "error", "message": str(e)} return {"status": "success"} raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found!") ``` We use the `unset_jwt_cookies{:.fn}` method of the `AuthJWT` instance to unset the cookies. That's it for the logout route. ## Conclusion That's a wrap! We have successfully implemented JWT Authentication in FastAPI using `fastapi-another-jwt-auth`. I hope you enjoyed this tutorial and learned something new. Feel free to reach out to me if you have any questions or suggestions. Cheers ✌🏼

Blog Posts

A Simple JWT Authentication Tutorial