Working with OAuth2 and JWT for authentication

In this recipe, we’ll integrate OAuth2 with JWTs for secure user authentication in your application. This approach improves security by utilizing tokens instead of credentials, aligning with modern authentication standards.

Getting ready

Since we will use a specific library to manage JWT, ensure you have the necessary dependencies installed. If you haven’t installed the packages from requirements.txt, run the following:

$ pip install python-jose[cryptography]

Also, we will use the users table used in the previous recipe, Setting up user registration. Make sure to have set it up before starting the recipe.

How to do it...

We can set up the JWT token integration through the following steps.

1. In a new module called security.py, let’s define the authentication function for the user:

   from sqlalchemy.orm import Session
   from models import User
   from email_validator import (
       validate_email,
       EmailNotValidError,
   )
   from operations import pwd_context
   def authenticate_user(
       session: Session,
       username_or_email: str,
       password: str,
   ) -> User | None:
       try:
           validate_email(username_or_email)
           query_filter = User.email
       except EmailNotValidError:
           query_filter = User.username
       user = (
           session.query(User)
           .filter(query_filter == username_or_email)
           .first()
       )
       if not user or not pwd_context.verify(
           password, user.hashed_password
       ):
           return
       return user

   The function can validate the input based on either the username or email.

2. Let’s define the functions to create and decode the access token in the same module (create_access_token and decode_access_token).

   To create the access token, we will need to specify a secret key, the algorithm used to generate it, and the expiration time, as follows:

   SECRET_KEY = "a_very_secret_key"
   ALGORITHM = "HS256"
   ACCESS_TOKEN_EXPIRE_MINUTES = 30

   Then, the create_access_token_function is as follows:

   from jose import jwt
   def create_access_token(data: dict) -> str:
       to_encode = data.copy()
       expire = datetime.utcnow() + timedelta(
           minutes=ACCESS_TOKEN_EXPIRE_MINUTES
       )
       to_encode.update({"exp": expire})
       encoded_jwt = jwt.encode(
           to_encode, SECRET_KEY, algorithm=ALGORITHM
       )
       return encoded_jwt

   To decode the access token, we can use a support function, get_user, that returns the User object by the username. You can do it on your own in the operations.py module or take it from the GitHub repository.

   The function to decode the token will be as follows:

   from jose import JWTError
   def decode_access_token(
       token: str, session: Session
   ) -> User | None:
       try:
           payload = jwt.decode(
               token, SECRET_KEY, algorithms=[ALGORITHM]
           )
           username: str = payload.get("sub")
       except JWTError:
           return
       if not username:
           return
       user = get_user(session, username)
       return user

3. We can now proceed to create the endpoint to retrieve the token in the same module, security.py, with the APIRouter class:

   from fastapi import (
       APIRouter,
       Depends,
       HTTPException,
       status,
   )
   from fastapi.security import (
       OAuth2PasswordRequestForm,
   )
   router = APIRouter()
   class Token(BaseModel):
       access_token: str
       token_type: str
   @router.post(
       "/token",
       response_model=Token,
       responses=..., # document the responses
   )
   def get_user_access_token(
       form_data: OAuth2PasswordRequestForm = Depends(),
       session: Session = Depends(get_session),
   ):
       user = authenticate_user(
           session,
           form_data.username,
           form_data.password
       )
       if not user:
           raise HTTPException(
               status_code=status.HTTP_401_UNAUTHORIZED,
               detail="Incorrect username or password",
           )
       access_token = create_access_token(
           data={"sub": user.username}
       )
       return {
           "access_token": access_token,
           "token_type": "bearer",
       }

4. Then, we can now create an OAuth2PasswordBearer object for the POST /token endpoint to obtain the access token:

   from fastapi.security import (
       OAuth2PasswordBearer,
   )
   oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

5. Finally, we can create the /users/me endpoint that returns the credentials based on the token:

   @router.get(
       "/users/me",
       responses=..., # document responses
   )
   def read_user_me(
       token: str = Depends(oauth2_scheme),
       session: Session = Depends(get_session),
   ):
       user = decode_access_token(token, session)
       if not user:
           raise HTTPException(
               status_code=status.HTTP_401_UNAUTHORIZED,
               detail="User not authorized",
           )
       return {
           "description": f"{user.username} authorized",
       }

6. Now, let’s import those endpoints into the FastAPI server in main.py. Right after defining the FastAPI object, let’s add the router, as follows:

   import security
   # rest of the code
   app.include_router(security.router)

We have just defined the authentication mechanism for our SaaS.

How it works…

Now, spin up the server by running the following code from the terminal at the project root folder level:

$ uvicorn main:app

Go to the Swagger documentation address in your browser (localhost:8000/docs) and you will see the new endpoints, POST /token and GET /users/me.

You need the token to call the second endpoint, which you can store in your browser automatically by clicking on the lock icon and filling out the form with your credentials.

You’ve made your SaaS application more secure by using OAuth2 with JWT, which help you guard your sensitive endpoints and make sure that only users who are logged in can use them. This arrangement gives you a reliable and safe way to verify users that works well for modern web applications.

See also

You can gain a better understanding of the OAuth2 framework by reading this article:

• Introduction to OAuth2: https://www.digitalocean.com/community/tutorials/an-introduction-to-oauth-2

Also, you can have a look at the protocol definition for JWTs at the following:

• JWT IETF Document: https://datatracker.ietf.org/doc/html/rfc7519