Authorization

This document covers authorization concepts including OAuth2PasswordBearer, get_current_user dependency, protecting endpoints with Dependency Injection, and Role-Based Access Control (RBAC).

---

📚 Table of Contents

1. Part 1: Authentication vs Authorization
2. Part 2: OAuth2PasswordBearer
3. Part 3: get_current_user Dependency
4. Part 4: Protecting Endpoints
5. Part 5: Role-Based Access Control (RBAC)
6. Part 6: CORS Configuration
7. Part 7: Testing Authentication

---

Part 1: Authentication vs Authorization

Concept	Description	Example
Authentication	Verifying WHO the user is	Login with username/password
Authorization	Verifying WHAT the user can access	Admin vs regular user permissions

Key Difference

• Authentication = "Who are you?" → Identity verification
• Authorization = "What can you do?" → Permission checking

---

Part 2: OAuth2PasswordBearer

OAuth2PasswordBearer is FastAPI's security dependency that extracts tokens from the Authorization header.

How It Works

1. Client sends request with Authorization: Bearer <token> header
2. OAuth2PasswordBearer extracts the token
3. Your dependency validates the token and returns user info

Basic Setup

from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

# Define the token URL (where users get their token)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    """This endpoint requires a valid Bearer token."""
    return {"token": token}

How the Token is Extracted

The OAuth2PasswordBearer class:

• Looks for the Authorization header
• Expects format: Bearer <token>
• Extracts and returns the token string
• Automatically documents in Swagger UI (Authorize button)

---

Part 3: get_current_user Dependency

The get_current_user pattern decodes the JWT and retrieves user information.

Implementation

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from typing import Optional

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"

async def get_current_user(token: str = Depends(oauth2_scheme)):
    """Decode JWT and return the current user."""
    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
        return {"username": username, "roles": payload.get("roles", [])}
    except JWTError:
        raise credentials_exception

Using Type Hints

from typing import Annotated

CurrentUser = Annotated[dict, Depends(get_current_user)]

@app.get("/users/me")
async def read_users_me(current_user: CurrentUser):
    return current_user

With Pydantic Models

from pydantic import BaseModel
from typing import List, Optional

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

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

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    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, roles=payload.get("roles", []))
    except JWTError:
        raise credentials_exception

    # Get user from database
    user = get_user_from_db(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)
) -> User:
    """Ensure user is not disabled."""
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

---

Part 4: Protecting Endpoints

Add Depends(get_current_user) to protect any endpoint:

Basic Protection

@app.get("/protected")
async def protected_route(current_user: dict = Depends(get_current_user)):
    """This endpoint requires authentication."""
    return {"message": f"Hello, {current_user['username']}!"}

Multiple Dependencies

from typing import Annotated

CurrentUser = Annotated[dict, Depends(get_current_user)]
DBSession = Annotated[AsyncSession, Depends(get_db)]

@app.get("/users/me/items")
async def read_own_items(
    current_user: CurrentUser,
    db: DBSession
):
    """Get items for the current user."""
    items = await db.execute(
        select(Item).where(Item.owner_id == current_user["id"])
    )
    return items.scalars().all()

Handling Token Errors

@app.get("/items/")
async def read_items(current_user: CurrentUser):
    # If token is invalid, 401 is raised automatically
    # If token is expired, 401 is raised automatically
    return {"owner": current_user["username"]}

---

Part 5: Role-Based Access Control (RBAC)

RBAC allows you to control access based on user roles (e.g., admin, user, moderator).

Storing Roles

Roles can be stored in:

1. JWT Claims: Include roles in the token payload
2. Database: Store roles in a users table

JWT with Roles:

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    to_encode["roles"] = ["user", "admin"]  # Include roles
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode["exp"] = expire
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

Role Checker Dependency

from functools import wraps
from typing import List

def require_role(allowed_roles: List[str]):
    """Dependency factory that checks user roles."""
    async def role_checker(current_user: dict = Depends(get_current_user)):
        user_roles = current_user.get("roles", [])
        if not any(role in allowed_roles for role in user_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Not enough permissions"
            )
        return current_user
    return role_checker

# Usage
@app.get("/admin/dashboard")
async def admin_dashboard(
    current_user: dict = Depends(require_role(["admin"]))
):
    return {"message": "Welcome to admin dashboard"}

@app.get("/moderator/actions")
async def moderator_actions(
    current_user: dict = Depends(require_role(["admin", "moderator"]))
):
    return {"message": "Moderator actions"}

Alternative: Class-Based Checker

class RoleChecker:
    def __init__(self, allowed_roles: List[str]):
        self.allowed_roles = allowed_roles

    async def __call__(self, current_user: dict = Depends(get_current_user)):
        user_roles = current_user.get("roles", [])
        if not any(role in self.allowed_roles for role in user_roles):
            raise HTTPException(
                status_code=403,
                detail="Operation not permitted"
            )
        return current_user

# Usage
allow_admin = RoleChecker(["admin"])
allow_moderator = RoleChecker(["admin", "moderator"])

@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    current_user: dict = Depends(allow_admin)
):
    return {"message": f"User {user_id} deleted by {current_user['username']}"}

RBAC Summary Table

Role	Dashboard	Moderator Actions	Delete Users
user	❌	❌	❌
moderator	❌	✅	❌
admin	✅	✅	✅

---

Part 6: CORS Configuration

CORS (Cross-Origin Resource Sharing) is essential for frontend-backend communication.

Why CORS Matters

When your frontend (e.g., http://localhost:3000) makes requests to your API (e.g., http://localhost:8000), browsers block cross-origin requests by default.

Configuration

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Configure CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",  # React dev server
        "https://myapp.com",      # Production frontend
    ],
    allow_credentials=True,  # Allow cookies
    allow_methods=["*"],     # Allow all HTTP methods
    allow_headers=["*"],     # Allow all headers
)

Development vs Production

Development (permissive):

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allow all origins
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Production (restrictive):

app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "https://myapp.com",
        "https://www.myapp.com",
    ],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

---

Part 7: Testing Authentication

Using TestClient

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

def test_protected_route_without_token():
    response = client.get("/protected")
    assert response.status_code == 401

def test_protected_route_with_token():
    # First, login to get token
    login_response = client.post(
        "/token",
        data={"username": "testuser", "password": "testpass"}
    )
    token = login_response.json()["access_token"]

    # Access protected route
    response = client.get(
        "/protected",
        headers={"Authorization": f"Bearer {token}"}
    )
    assert response.status_code == 200

Dependency Overrides for Testing

def get_test_user():
    """Mock user for testing."""
    return {"username": "testuser", "roles": ["admin"]}

# Override the dependency
app.dependency_overrides[get_current_user] = get_test_user

def test_admin_route():
    response = client.get("/admin/dashboard")
    assert response.status_code == 200

# Reset overrides after tests
app.dependency_overrides = {}

Async Testing with httpx

import pytest
import httpx
from main import app

@pytest.mark.asyncio
async def test_async_protected_route():
    async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
        # Login
        login_response = await ac.post(
            "/token",
            data={"username": "testuser", "password": "testpass"}
        )
        token = login_response.json()["access_token"]

        # Access protected route
        response = await ac.get(
            "/protected",
            headers={"Authorization": f"Bearer {token}"}
        )
        assert response.status_code == 200

---

📖 References

Resource	Link	Duration
FastAPI OAuth2PasswordBearer	fastapi.tiangolo.com	~15 min
FastAPI Dependencies	fastapi.tiangolo.com	~20 min
FastAPI Security & DI	fastapi.tiangolo.com	~15 min
Auth0 RBAC Concepts	auth0.com	~15 min

Learning Checklist

• Understand what OAuth2PasswordBearer does and how it extracts tokens
• Implement a get_current_user dependency that decodes JWT
• Secure endpoints with Depends(get_current_user) and verify 401 for unauthorized
• Create roles (admin, user) and store in JWT claims or DB
• Implement RBAC logic with role checking dependencies
• Test endpoints with different roles and confirm access control
• Handle token expiration and invalid token scenarios
• Document security scheme in OpenAPI (Swagger Authorize button)
• Configure CORS for frontend-backend communication
• Validate DI keeps code modular and reusable

---

Complete Example: Full Auth Flow

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

app = FastAPI()

# Security configuration
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Fake user database
fake_users_db = {
    "admin": {
        "username": "admin",
        "hashed_password": pwd_context.hash("admin123"),
        "roles": ["admin", "user"],
        "disabled": False,
    },
    "user": {
        "username": "user",
        "hashed_password": pwd_context.hash("user123"),
        "roles": ["user"],
        "disabled": False,
    },
}

# Type aliases
CurrentUser = Annotated[dict, Depends(lambda token=Depends(oauth2_scheme): get_current_user(token))]

async def get_current_user(token: str) -> dict:
    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 or username not in fake_users_db:
            raise credentials_exception
        user = fake_users_db[username]
        if user["disabled"]:
            raise HTTPException(status_code=400, detail="Inactive user")
        return user
    except JWTError:
        raise credentials_exception

def require_role(allowed_roles: List[str]):
    async def role_checker(token: str = Depends(oauth2_scheme)):
        user = await get_current_user(token)
        if not any(role in allowed_roles for role in user["roles"]):
            raise HTTPException(status_code=403, detail="Not enough permissions")
        return user
    return role_checker

# Endpoints
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(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")

    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    token = jwt.encode(
        {"sub": user["username"], "roles": user["roles"], "exp": expire},
        SECRET_KEY,
        algorithm=ALGORITHM
    )
    return {"access_token": token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(require_role(["user"]))):
    return {"username": current_user["username"], "roles": current_user["roles"]}

@app.get("/admin/dashboard")
async def admin_dashboard(current_user: dict = Depends(require_role(["admin"]))):
    return {"message": f"Welcome Admin {current_user['username']}!"}

---

Summary

This document covers:

1. OAuth2PasswordBearer: Extracting tokens from Authorization header
2. get_current_user: Decoding JWT and retrieving user info
3. Protecting Endpoints: Using Depends() for authentication
4. RBAC: Role-based access control with custom dependencies
5. CORS: Configuring cross-origin requests
6. Testing: Testing authenticated endpoints with TestClient