FastAPI OAuth2 Password Flow

Introduction

OAuth2 is an industry-standard protocol for authorization that enables applications to secure access to user data without exposing credentials. The Password Flow (also known as Resource Owner Password Credentials Grant) is one of the OAuth2 authorization flows that allows users to directly provide their username and password to obtain an access token.

In this tutorial, we'll learn how to implement OAuth2 Password Flow authentication in a FastAPI application. This approach is particularly useful when you need to authenticate users directly against your application's user database rather than through a third-party service.

Prerequisites

Before we begin, make sure you have:

• Basic knowledge of Python
• Understanding of FastAPI fundamentals
• Python 3.7+ installed
• FastAPI and its dependencies installed

If you haven't installed FastAPI yet, you can do so with pip:

bash

pip install fastapi uvicorn[standard] python-multipart python-jose[cryptography] passlib[bcrypt]

Understanding OAuth2 Password Flow

The OAuth2 Password Flow works as follows:

1. The user provides their username and password to your application
2. Your application verifies the credentials against your user database
3. If valid, your application issues an access token
4. The client includes this token in subsequent requests to protected resources
5. Your application validates the token before processing these requests

This flow is suitable when:

• Your application is the one that registered the users
• You have a trusted client (like your own frontend application)
• You need a simpler authentication mechanism than the full OAuth2 authorization code flow

Step-by-Step Implementation

Let's implement OAuth2 Password Flow in FastAPI:

Step 1: Project Structure

Start by organizing your project:

├── app
│   ├── __init__.py
│   ├── main.py
│   ├── auth.py
│   ├── models.py
│   └── database.py
└── requirements.txt

Step 2: Setting Up User Models and Database

First, let's create a simple user model and mock database in models.py:

python

from pydantic import BaseModel
from typing import Optional

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

Now, let's set up a mock database in database.py for demonstration:

python

from models import UserInDB

# This is just a mock database for demonstration
# In a real application, you would use a proper database
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

The hashed password above corresponds to "secret". In a real application, you would create passwords using the password hashing functions we'll define next.

Step 3: Creating Authentication Utilities

Let's create auth.py to handle our authentication logic:

python

from datetime import datetime, timedelta
from typing import Optional

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 .models import User, UserInDB
from .database import fake_users_db

# Secret key for signing JWT tokens - use a proper secret key in production
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

# OAuth2 scheme for token endpoints
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

class TokenData(BaseModel):
    username: Optional[str] = None

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(db, username: str, password: str):
    user = get_user(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: Optional[timedelta] = 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
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.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

Step 4: Building the FastAPI Application

Now, let's create our main application in main.py:

python

from datetime import timedelta
from typing import List

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

from .auth import (
    ACCESS_TOKEN_EXPIRE_MINUTES, Token, User,
    authenticate_user, create_access_token,
    get_current_active_user
)
from .database import fake_users_db

app = FastAPI()

@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

@app.get("/users/me/items")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    return [{"item_id": "Item1", "owner": current_user.username}]

Step 5: Running the Application

Run your application with Uvicorn:

bash

uvicorn app.main:app --reload

Testing Your Authentication System

Now that your application is running, you can test the authentication system:

Getting an Access Token

To get a token, make a POST request to the /token endpoint:

bash

curl -X POST "http://localhost:8000/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "username=johndoe&password=secret"

You'll receive a response like:

json

{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "bearer"
}

Accessing Protected Endpoints

Use the token to access protected endpoints:

bash

curl -X GET "http://localhost:8000/users/me" \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."

You should see your user information:

json

{
  "username": "johndoe",
  "email": "johndoe@example.com",
  "full_name": "John Doe",
  "disabled": false
}

Real-World Implementation Considerations

When implementing OAuth2 Password Flow in a production environment, consider the following:

1. Secure Password Storage:

   • Always hash passwords with a strong algorithm like bcrypt
   • Never store plain text passwords
   • Use a proper salt for each password

2. Token Management:

   • Generate a secure random secret key
   • Set appropriate token expiration times
   • Consider implementing refresh tokens for better security

3. Database Integration:

   • Replace the mock database with a real database (SQL or NoSQL)
   • Implement proper user management functionality

4. Environmental Security:

   • Store secrets in environment variables or secure vaults
   • Use HTTPS in production
   • Consider rate limiting to prevent brute force attacks

Here's an example of how to generate a proper secret key:

python

import secrets
print(secrets.token_hex(32))  # Generates a 64-character hex string

Advanced Features

To enhance your authentication system, you might want to implement:

Refresh Tokens

python

# Add to your Token model
class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str

# Add a refresh token endpoint
@app.post("/refresh", response_model=Token)
async def refresh_token(refresh_token: str):
    # Validate the refresh token
    # Issue a new access token
    # ...

Password Reset Functionality

python

@app.post("/password-reset")
async def request_password_reset(email: str):
    # Generate a password reset token
    # Send email with reset link
    # ...

@app.post("/password-reset/confirm")
async def confirm_password_reset(token: str, new_password: str):
    # Validate the reset token
    # Update the user's password
    # ...

Role-Based Access Control

python

class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    GUEST = "guest"

class UserInDB(User):
    hashed_password: str
    role: UserRole = UserRole.USER

def get_current_admin_user(current_user: User = Depends(get_current_active_user)):
    if current_user.role != UserRole.ADMIN:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Not enough permissions"
        )
    return current_user

@app.get("/admin-only")
async def admin_endpoint(current_user: User = Depends(get_current_admin_user)):
    return {"message": "You are an admin!"}

Summary

In this tutorial, we've learned how to implement OAuth2 Password Flow in a FastAPI application:

1. Setting up user models and a database
2. Creating authentication utilities for password hashing and token generation
3. Implementing token generation and user verification endpoints
4. Securing routes with dependency injection
5. Testing the authentication flow

OAuth2 Password Flow provides a secure way to authenticate users in your FastAPI applications, especially when you need to validate credentials against your own user database.

Additional Resources

• FastAPI Security Documentation
• OAuth2 Specification
• JWT (JSON Web Tokens) Official Site
• Passlib Documentation

Exercises

1. Implement a user registration endpoint that securely hashes passwords
2. Add email verification to the registration process
3. Implement refresh tokens to enhance security
4. Create a logout mechanism to invalidate tokens
5. Add role-based access control to protect certain endpoints

By completing these exercises, you'll gain a deeper understanding of FastAPI security mechanisms and be well-equipped to implement secure authentication in your applications.