Advanced FastAPI Authentication
OAuth2 Implementation
Securing Endpoints with OAuth2
To secure API endpoints in FastAPI, we can use the OAuth2 password flow. This method is common for first-party applications where a user directly enters their username and password to get an access token. FastAPI provides a helper class, OAuth2PasswordBearer, to manage this process. It handles extracting the token from the Authorization header of incoming requests.
First, you need to instantiate this class. The only required argument is tokenUrl, which is the relative URL where clients will send their username and password to request a token. We'll create this endpoint next.
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
Creating the Token Endpoint
The tokenUrl you specified points to an endpoint you must create. This endpoint's job is to receive a username and password, verify them, and if they are correct, return an access token. The client will send this data as form fields, not JSON.
FastAPI makes handling this easy with the OAuth2PasswordRequestForm dependency. It automatically parses the incoming form data, expecting username and password fields.
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
# --- This part is for a complete example ---
# In a real app, you'd have a database and more robust user handling.
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# --- Token Creation Helper ---
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# --- Token Endpoint Implementation ---
@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
# In a real app, you'd look up the user in a database
# and verify the password.
user = get_user(form_data.username)
if not user or not pwd_context.verify(form_data.password, user.hashed_password):
raise HTTPException(
status_code=401,
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"}
With the token endpoint ready, clients can exchange credentials for a JWT (JSON Web Token). The next step is to protect other endpoints, requiring this token for access. You do this by adding oauth2_scheme as a dependency. FastAPI checks for a valid Authorization: Bearer <token> header and raises an error if it's missing or invalid.
Managing Scopes
Sometimes you need more granular control over what a user can do. For example, some users might only have permission to read data, while others can write it. This is where scopes come in. Scopes are strings that represent specific permissions, like items:read or items:write.
You can define the available scopes when you create the OAuth2PasswordBearer instance.
from fastapi import Depends, Security
scopes_dict = {
"me:read": "Read information about the current user.",
"items:read": "Read items.",
"items:write": "Write items.",
}
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", scopes=scopes_dict)
To protect an endpoint with specific scopes, you use Security instead of Depends. You pass it a dependency function that decodes the token and verifies the user's permissions, along with a list of required scopes for that endpoint. If the token provided by the client doesn't contain the required scopes, FastAPI will return an error.
async def get_current_user(token: str = Depends(oauth2_scheme)):
# Decode the token and get user data
# (Implementation omitted for brevity)
pass
async def get_current_active_user(current_user: User = Security(get_current_user, scopes=["me:read"])):
# You can add checks here, like if the user is active
return current_user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
@app.get("/items/")
async def read_items(current_user: User = Security(get_current_user, scopes=["items:read"])):
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
In the example above, a user needs a token with the items:read scope to access the /items/ endpoint. However, they only need the me:read scope to access their own information at /users/me.
Time to check your understanding.
What is the primary role of the OAuth2PasswordBearer class in FastAPI?
When a client sends a request to the tokenUrl endpoint to get an access token, in what format should the username and password be sent?
By combining OAuth2PasswordBearer with dependencies and scopes, you can build a robust and secure authentication system for your FastAPI applications.