Implementing OAuth2 Authentication in FastAPI

FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.7+ based on standard Python type hints. Authentication is a crucial aspect of most web applications, and OAuth2 is a widely - adopted standard for authorization that provides a secure way to grant access to resources. In this blog post, we’ll explore how to implement OAuth2 authentication in FastAPI, covering fundamental concepts, usage methods, common practices, and best practices.

Table of Contents

  1. Fundamental Concepts of OAuth2 and FastAPI
  2. Setting Up a FastAPI Project
  3. Implementing OAuth2 Password Flow in FastAPI
  4. Common Practices
  5. Best Practices
  6. Conclusion
  7. References

Fundamental Concepts of OAuth2 and FastAPI

OAuth2 Basics

OAuth2 is an authorization framework that enables applications to obtain limited access to user accounts on an HTTP service. It provides different grant types such as authorization code, implicit, resource owner password credentials, and client credentials.

FastAPI and Security

FastAPI has built - in support for security features, including OAuth2. It uses the fastapi.security module to handle authentication and authorization easily. The OAuth2PasswordBearer class is a key component for implementing OAuth2 password flow.

Setting Up a FastAPI Project

First, make sure you have Python 3.7+ installed. Then, create a virtual environment and install FastAPI and other necessary libraries:

python -m venv venv
source venv/bin/activate  # On Windows use `venv\Scripts\activate`
pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt]

Implementing OAuth2 Password Flow in FastAPI

1. Import necessary modules

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from datetime import datetime, timedelta

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Pydantic models
class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


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


# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()


# Mock user database
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "[email protected]",
        "hashed_password": pwd_context.hash("secret"),
        "disabled": False,
    }
}


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW - Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW - Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

2. Run the application

uvicorn main:app --reload

3. Test the authentication

  • Use a tool like curl or Postman to send a POST request to /token with the username and password.
curl -X POST "http://127.0.0.1:8000/token" -H "accept: application/json" -H "Content-Type: application/x-www-form-urlencoded" -d "grant_type=&username=johndoe&password=secret&scope=&client_id=&client_secret="
  • Use the received access token to access the protected endpoint /users/me/.
curl -X GET "http://127.0.0.1:8000/users/me/" -H "accept: application/json" -H "Authorization: Bearer <ACCESS_TOKEN>"

Common Practices

  • Use HTTPS: Always use HTTPS in production to protect the tokens from being intercepted.
  • Token Expiration: Set appropriate expiration times for access tokens. Short - lived access tokens reduce the risk of a compromised token being misused.
  • Error Handling: Provide clear and appropriate error messages for authentication and authorization failures.

Best Practices

  • Use a Real Database: In a production environment, use a real database like PostgreSQL or MySQL instead of a mock database.
  • Refresh Tokens: Implement refresh tokens to avoid frequent re - authentication by the user.
  • Security Headers: Set security headers such as Content - Security - Policy, X - Frame - Options, and X - Content - Type - Options to enhance security.

Conclusion

Implementing OAuth2 authentication in FastAPI is relatively straightforward thanks to its built - in security features. By following the fundamental concepts, common practices, and best practices, you can build secure and reliable APIs. Remember to always prioritize security when dealing with user authentication and authorization.

References