Verifying JWT Tokens for Secure Access

Welcome back! In our previous lessons, you learned how to set up basic user authentication using FastAPI . You also explored how to secure endpoints using Dependency Injection and understood the importance of JSON Web Tokens (JWT) for secure authentication. Today, we’re going to build on those concepts and the progress you've made. Remember how we generated JWT tokens and returned them to the user upon login? Now, let’s take the next step: verifying these tokens to secure our API endpoints. By the end of this lesson, you'll be able to implement and verify JWT tokens in FastAPI, ensuring only authenticated users can access certain parts of your application. Let's dive in!

Setup from Previous Lesson

Before we move forward, let's quickly recap what we covered in the previous lesson, where we generated JWT tokens for authentication. Here's a snippet of the essential code you learned: Pythonfrom fastapi import FastAPI, HTTPException, Depends from fastapi.security import OAuth2PasswordRequestForm from pydantic import BaseModel from jose import JWTError, jwt from datetime import datetime, timedelta app = FastAPI() # Mock database with usernames and passwords users_db = {"user1": "pass1", "user2": "pass2"} # Configuration for JWT SECRET_KEY = "your_secret_key" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 # Token class class Token(BaseModel): access_token: str token_type: str # Function to create JWT token with an expiration time def create_access_token(data: dict): to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) # Function to check if user exists in the database def authenticate_user(form_data: OAuth2PasswordRequestForm = Depends()): username = form_data.username password = form_data.password if users_db.get(username) == password: return {"username": username} raise HTTPException(status_code=401, detail="Incorrect username or password") # Login endpoint @app.post("/login", response_model=Token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): user = authenticate_user(form_data) access_token = create_access_token(data={"sub": user["username"]}) return {"access_token": access_token, "token_type": "bearer"}from fastapi import FastAPI, HTTPException, Depends from fastapi.security import OAuth2PasswordRequestForm from pydantic import BaseModel from jose import JWTError, jwt from datetime import datetime, timedelta app = FastAPI() # Mock database with usernames and passwords users_db = {"user1": "pass1", "user2": "pass2"} # Configuration for JWT SECRET_KEY = "your_secret_key" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 # Token class class Token(BaseModel): access_token: str token_type: str # Function to create JWT token with an expiration time def create_access_token(data: dict): to_encode = data.copy() expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) # Function to check if user exists in the database def authenticate_user(form_data: OAuth2PasswordRequestForm = Depends()): username = form_data.username password = form_data.password if users_db.get(username) == password: return {"username": username} raise HTTPException(status_code=401, detail="Incorrect username or password") # Login endpoint @app.post("/login", response_model=Token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()): user = authenticate_user(form_data) access_token = create_access_token(data={"sub": user["username"]}) return {"access_token": access_token, "token_type": "bearer"} This code generated JWT tokens upon a successful login, which will be crucial for securing endpoints.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Python

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel
from jose import JWTError, jwt
from datetime import datetime, timedelta

app = FastAPI()

# Mock database with usernames and passwords
users_db = {"user1": "pass1", "user2": "pass2"}

# Configuration for JWT
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Token class
class Token(BaseModel):
    access_token: str
    token_type: str

# Function to create JWT token with an expiration time
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    
# Function to check if user exists in the database
def authenticate_user(form_data: OAuth2PasswordRequestForm = Depends()):
    username = form_data.username
    password = form_data.password
    if users_db.get(username) == password:
        return {"username": username}
    raise HTTPException(status_code=401, detail="Incorrect username or password")

# Login endpoint
@app.post("/login", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data)
    access_token = create_access_token(data={"sub": user["username"]})
    return {"access_token": access_token, "token_type": "bearer"}

Python

from fastapi.security import OAuth2PasswordBearer

# Define the OAuth2 scheme to be used for token authentication
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

from fastapi.security import OAuth2PasswordBearer

# Define the OAuth2 scheme to be used for token authentication
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

Python

# Define a function to get the current user from the JWT token
async def get_current_user(token: str = Depends(oauth2_scheme)):

# Define a function to get the current user from the JWT token
async def get_current_user(token: str = Depends(oauth2_scheme)):

Python

from jose import JWTError

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        # Decode and verify the JWT token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # Extract the username from the token payload
        username: str = payload.get("sub")
        # Check if the username is present in the token payload
        if not username:
            raise HTTPException(status_code=401, detail="Invalid token")
        # Return the username if valid
        return {"username": username}
    # Handle any JWT decoding errors
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

from jose import JWTError

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        # Decode and verify the JWT token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # Extract the username from the token payload
        username: str = payload.get("sub")
        # Check if the username is present in the token payload
        if not username:
            raise HTTPException(status_code=401, detail="Invalid token")
        # Return the username if valid
        return {"username": username}
    # Handle any JWT decoding errors
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Python

# Secured endpoint which depends on decoding the token
@app.get("/secure-message")
async def read_secure_message(current_user: dict = Depends(get_current_user)):
    return {"message": "This is a secured message"}

# Secured endpoint which depends on decoding the token
@app.get("/secure-message")
async def read_secure_message(current_user: dict = Depends(get_current_user)):
    return {"message": "This is a secured message"}

JSON

{
  "detail": "Not authenticated"
}

{
  "detail": "Not authenticated"
}

Python

from fastapi.security import OAuth2PasswordBearer

# Define the OAuth2 scheme to be used for token authentication
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

Python

# Define a function to get the current user from the JWT token
async def get_current_user(token: str = Depends(oauth2_scheme)):

Python

from jose import JWTError

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        # Decode and verify the JWT token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # Extract the username from the token payload
        username: str = payload.get("sub")
        # Check if the username is present in the token payload
        if not username:
            raise HTTPException(status_code=401, detail="Invalid token")
        # Return the username if valid
        return {"username": username}
    # Handle any JWT decoding errors
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

Python

# Secured endpoint which depends on decoding the token
@app.get("/secure-message")
async def read_secure_message(current_user: dict = Depends(get_current_user)):
    return {"message": "This is a secured message"}

JSON

{
  "detail": "Not authenticated"
}

Verifying JWT Tokens for Secure Access

Setup from Previous Lesson

Join the 1M+ learners on CodeSignal

Understanding Token Verification

Set Up OAuth2 Scheme

Passing the Scheme to a Function

Verifying the Token

Securing an Endpoint

Accessing the Secured Endpoint

Unauthorized Access

Summary and Last Practices