Authentication and Authorization

Implement secure authentication flows and fine-grained authorization for your APIs

Securing Your API

Authentication verifies who you are, while authorization determines what you can do. This lesson covers API keys, JWT tokens, OAuth2 flows, and role-based access control, giving you the tools to build secure, production-ready APIs that protect user data and enforce permissions properly.

Authentication vs Authorization

These terms are often confused, but they serve different purposes in API security. Understanding the distinction is fundamental to building secure systems.

🔐 Authentication

"Who are you?" - Verifying the identity of a user or system.

Proves you are who you claim to be
Uses credentials (password, token, certificate)
Happens first, before authorization
Typically one-time per session

Examples:

Logging in with username/password
Presenting an API key
Verifying a JWT token
Scanning a fingerprint

🛡️ Authorization

"What can you do?" - Determining permissions and access rights.

Controls what resources you can access
Defines what actions you can perform
Happens after authentication
Checked on every request/action

Examples:

Admin can delete users
User can only edit own profile
Read-only access to reports
Premium tier features access

Real-World Analogy

Authentication is showing your ID at the airport security, proving you are who your ticket says you are. Authorization is your ticket itself, it determines which gate you can access, whether you board first, and if you can enter the lounge. You need both: the ID proves identity, the ticket grants permissions.

API Keys - Simplest Authentication

API keys are the simplest form of authentication. A long, random string identifies and authenticates the client application. While easy to implement, they have significant limitations.

How API Keys Work

from fastapi import FastAPI, Header, HTTPException, Security
from fastapi.security import APIKeyHeader
from typing import Optional

app = FastAPI()

# In production, store these in a database
VALID_API_KEYS = {
    "sk_live_51abc123xyz": {"user": "client_1", "tier": "premium"},
    "sk_test_49def456uvw": {"user": "client_2", "tier": "free"}
}

# Define API key header scheme
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

async def verify_api_key(api_key: str = Security(api_key_header)) -> dict:
    """Verify API key and return associated user info"""
    if not api_key:
        raise HTTPException(
            status_code=401,
            detail="API key is missing",
            headers={"WWW-Authenticate": "ApiKey"}
        )

    user_info = VALID_API_KEYS.get(api_key)
    if not user_info:
        raise HTTPException(
            status_code=403,
            detail="Invalid API key"
        )

    return user_info

@app.get("/api/data")
async def get_data(user_info: dict = Security(verify_api_key)):
    return {
        "data": "Sensitive information",
        "user": user_info["user"],
        "tier": user_info["tier"]
    }

# Client request:
# GET /api/data
# X-API-Key: sk_live_51abc123xyz

# Response: 200 OK
# {
#   "data": "Sensitive information",
#   "user": "client_1",
#   "tier": "premium"
# }

Result: The API key in the header identifies the client. FastAPI's Security dependency automatically validates the key before the endpoint runs.

API Key Best Practices

DO This

Use long, random keys (32+ characters)
Send via headers, not URL parameters
Use HTTPS to encrypt transmission
Allow key rotation without downtime
Store hashed versions in database
Implement rate limiting per key

DON'T Do This

Put keys in URL query strings
Commit keys to version control
Share keys between multiple apps
Use keys for user authentication
Send keys over unencrypted HTTP
Let keys never expire

Limitations of API Keys

No expiration: Keys live forever unless manually revoked
No fine-grained permissions: Typically all-or-nothing access
No user context: Identifies the application, not the user
Hard to rotate: Updating keys requires client changes
Not suitable for end users: Meant for server-to-server communication

JWT (JSON Web Tokens)

JWTs are compact, self-contained tokens that securely transmit information between parties. They're the modern standard for stateless authentication in APIs.

JWT Structure

A JWT consists of three Base64-encoded parts separated by dots:header.payload.signature

1. Header (Algorithm & Token Type)

{
  "alg": "HS256",  // Signing algorithm
  "typ": "JWT"     // Token type
}

2. Payload (Claims/Data)

{
  "sub": "user123",           // Subject (user ID)
  "name": "John Doe",         // User name
  "email": "john@example.com",
  "role": "admin",            // Custom claim
  "iat": 1704067200,          // Issued at (timestamp)
  "exp": 1704153600           // Expiration time
}

3. Signature (Verification)

HMACSHA256(
  base64UrlEncode(header) + "." + base64UrlEncode(payload),
  secret_key
)

The signature ensures the token hasn't been tampered with.

Implementing JWT in FastAPI

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
import jwt
from jwt import InvalidTokenError
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from typing import Optional
from pydantic import BaseModel

app = FastAPI()

# Configuration
SECRET_KEY = "your-secret-key-keep-this-safe-use-env-vars"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

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

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

class User(BaseModel):
    username: str
    email: Optional[str] = None
    role: str = "user"

# Fake user database (use real database in production)
fake_users_db = {
    "john": {
        "username": "john",
        "email": "john@example.com",
        "hashed_password": pwd_context.hash("secret123"),
        "role": "admin"
    }
}

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

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """Generate JWT token with expiration"""
    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

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    """Decode and validate JWT token"""
    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 InvalidTokenError:
        raise credentials_exception

    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception

    return User(**user)

@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    """Login endpoint - returns JWT token"""
    user = fake_users_db.get(form_data.username)

    if not user or not verify_password(form_data.password, user["hashed_password"]):
        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"], "role": user["role"]},
        expires_delta=access_token_expires
    )

    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    """Protected endpoint - requires valid JWT"""
    return current_user

Result: Users authenticate once via /token, receive a JWT, then include it in the Authorization: Bearer <token> header for subsequent requests. The server validates the token without database lookups.

JWT Usage Flow

# Step 1: Login to get token
POST /token
Content-Type: application/x-www-form-urlencoded

username=john&password=secret123

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

# Step 2: Use token for authenticated requests
GET /users/me
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

# Response:
{
  "username": "john",
  "email": "john@example.com",
  "role": "admin"
}

✅ JWT Advantages

Stateless (no server storage)
Self-contained (includes user data)
Can include custom claims
Works across domains
Scalable for microservices

❌ JWT Disadvantages

Can't revoke until expiration
Larger than session IDs
Token theft is problematic
Data in token can be read
Requires careful expiration

🔒 Security Tips

Keep expiration short (15-60 min)
Use refresh tokens for renewal
Never store sensitive data in JWT
Always use HTTPS
Validate signature on every request

OAuth 2.0 Flows

OAuth 2.0 is an authorization framework that allows third-party applications to access user data without exposing passwords. It's used by Google, Facebook, GitHub, and most social login systems.

OAuth 2.0 Terminology

Term	Description	Example
Resource Owner	The user who owns the data	You, wanting to share Google Drive files
Client	The application requesting access	A photo printing app
Authorization Server	Server that issues tokens	Google's OAuth server
Resource Server	Server hosting the protected resources	Google Drive API
Access Token	Credential for accessing resources	Token granting read access to Drive
Scope	What permissions are requested	`drive.readonly photos.read`

Authorization Code Flow (Most Common)

Used for web applications where the client can keep a secret. Most secure OAuth flow.

Step 1: Authorization Request

GET https://oauth-provider.com/authorize?
  response_type=code
  &client_id=your_client_id
  &redirect_uri=https://yourapp.com/callback
  &scope=read_user+read_repos
  &state=random_state_value

User is redirected to OAuth provider to grant permission

Step 2: User Grants PermissionUser sees "MyApp wants to access your data" and clicks "Allow"

Step 3: Authorization Code Returned

GET https://yourapp.com/callback?
  code=AUTH_CODE_HERE
  &state=random_state_value

Provider redirects back with a temporary authorization code

Step 4: Exchange Code for Access Token

POST https://oauth-provider.com/token
Content-Type: application/x-www-form-urlencoded

grant_type=authorization_code
&code=AUTH_CODE_HERE
&client_id=your_client_id
&client_secret=your_client_secret
&redirect_uri=https://yourapp.com/callback

Response:
{
  "access_token": "abc123xyz...",
  "token_type": "Bearer",
  "expires_in": 3600,
  "refresh_token": "refresh_abc...",
  "scope": "read_user read_repos"
}

Other OAuth 2.0 Flows

Use case: Machine-to-machine authentication (no user involved)

When to use: Backend services, cron jobs, server-to-server communication

POST /token
grant_type=client_credentials
&client_id=service_id
&client_secret=service_secret

Use case: First-party apps where you trust the client completely

⚠️ Not recommended: Exposes user credentials to the client

POST /token
grant_type=password
&username=user@example.com
&password=secret123
&client_id=app_id

Previously used for: Single-page applications (SPAs)

⚠️ Deprecated: Use Authorization Code Flow with PKCE instead

Returns token directly in URL fragment, which is less secure than using authorization codes.

Use case: Mobile apps, SPAs, any public client

Benefit: Prevents authorization code interception attacks

# Step 1: Generate code verifier and challenge
code_verifier = random_string(43-128 chars)
code_challenge = base64url(sha256(code_verifier))

# Step 2: Authorization request with challenge
GET /authorize?...&code_challenge=ABC...&code_challenge_method=S256

# Step 3: Token request with verifier
POST /token
code=AUTH_CODE&code_verifier=XYZ...

Which OAuth Flow to Use?

Web apps with backend: Authorization Code Flow
Mobile/SPA apps: Authorization Code Flow + PKCE
Server-to-server: Client Credentials Flow
First-party trusted apps only: Password Flow (use sparingly)
Never use: Implicit Flow (deprecated)

Role-Based Access Control (RBAC)

RBAC assigns permissions to roles rather than individual users. Users are assigned roles, and roles define what actions can be performed. This simplifies permission management at scale.

Implementing RBAC in FastAPI

from fastapi import FastAPI, Depends, HTTPException
from enum import Enum
from typing import List

app = FastAPI()

# Define roles
class Role(str, Enum):
    ADMIN = "admin"
    MODERATOR = "moderator"
    USER = "user"
    GUEST = "guest"

# Define permissions for each role
ROLE_PERMISSIONS = {
    Role.ADMIN: ["read", "write", "delete", "manage_users"],
    Role.MODERATOR: ["read", "write", "delete"],
    Role.USER: ["read", "write"],
    Role.GUEST: ["read"]
}

def require_role(allowed_roles: List[Role]):
    """Dependency that checks if user has required role"""
    async def role_checker(current_user: User = Depends(get_current_user)):
        if current_user.role not in allowed_roles:
            raise HTTPException(
                status_code=403,
                detail=f"Access denied. Requires one of: {[r.value for r in allowed_roles]}"
            )
        return current_user
    return role_checker

def require_permission(required_permission: str):
    """Dependency that checks if user has specific permission"""
    async def permission_checker(current_user: User = Depends(get_current_user)):
        user_permissions = ROLE_PERMISSIONS.get(current_user.role, [])
        if required_permission not in user_permissions:
            raise HTTPException(
                status_code=403,
                detail=f"Permission denied. Requires: {required_permission}"
            )
        return current_user
    return permission_checker

# Public endpoint - no authentication required
@app.get("/public/data")
async def public_data():
    return {"message": "This is public"}

# Authenticated endpoint - any logged-in user
@app.get("/protected/data")
async def protected_data(user: User = Depends(get_current_user)):
    return {"message": f"Hello {user.username}"}

# Role-based endpoint - only admins
@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    current_user: User = Depends(require_role([Role.ADMIN]))
):
    # Delete user logic
    return {"message": f"User {user_id} deleted by {current_user.username}"}

# Permission-based endpoint
@app.post("/articles")
async def create_article(
    article: dict,
    current_user: User = Depends(require_permission("write"))
):
    # Users, moderators, and admins can write
    return {"message": "Article created", "author": current_user.username}

# Multiple roles allowed
@app.put("/comments/{comment_id}")
async def moderate_comment(
    comment_id: int,
    current_user: User = Depends(require_role([Role.ADMIN, Role.MODERATOR]))
):
    # Both admins and moderators can moderate
    return {"message": f"Comment {comment_id} moderated"}

Result: FastAPI dependencies automatically check roles/permissions before endpoint execution. If the user lacks required access, a 403 Forbidden error is returned.

RBAC vs ABAC

Role-Based (RBAC)

Permissions based on user's role

Simple to implement and understand
Easy to audit ("who has admin?")
Good for most applications

if user.role == "admin": allow()

Attribute-Based (ABAC)

Permissions based on attributes/context

More flexible and fine-grained
Complex to implement and debug
Good for complex requirements

if user.dept == resource.dept: allow()

Resource-Level Permissions

# Check if user owns the resource
async def get_current_user_post(
    post_id: int,
    current_user: User = Depends(get_current_user)
):
    """Ensure user can only access their own posts"""
    post = await db.get_post(post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")

    # Allow if owner OR admin
    if post.author_id != current_user.id and current_user.role != Role.ADMIN:
        raise HTTPException(
            status_code=403,
            detail="You can only edit your own posts"
        )

    return post

@app.put("/posts/{post_id}")
async def update_post(
    post_id: int,
    updates: dict,
    post = Depends(get_current_user_post)  # Validates ownership
):
    # Update post logic
    return {"message": "Post updated", "post_id": post_id}

# Result: Users can only modify their own posts, but admins can modify any post

Result: Combines authentication, role checking, and resource ownership validation in a reusable dependency.

Refresh Tokens

Access tokens should be short-lived for security. Refresh tokens allow clients to obtain new access tokens without requiring users to log in again.

How Refresh Tokens Work

1. Initial Login

User logs in, receives both tokens:

{
  "access_token": "eyJhbG...",  // Short-lived (15-60 min)
  "refresh_token": "def502...", // Long-lived (days/weeks)
  "expires_in": 900              // 15 minutes
}

2. Use Access Token

Client uses access token for API requests until it expires

3. Access Token Expires

API returns 401 Unauthorized when access token expires

4. Refresh Access Token

Client uses refresh token to get a new access token:

POST /token/refresh
Content-Type: application/json

{
  "refresh_token": "def502..."
}

Response:
{
  "access_token": "eyJNew...",  // New access token
  "expires_in": 900
}

Implementing Refresh Tokens

from datetime import datetime, timedelta, timezone
import secrets

# Store refresh tokens in database
refresh_tokens_db = {}  # Use Redis or database in production

def create_refresh_token(username: str) -> str:
    """Generate a secure refresh token"""
    token = secrets.token_urlsafe(32)
    refresh_tokens_db[token] = {
        "username": username,
        "created_at": datetime.now(timezone.utc),
        "expires_at": datetime.now(timezone.utc) + timedelta(days=30)
    }
    return token

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # Validate credentials...
    user = authenticate_user(form_data.username, form_data.password)

    # Create both tokens
    access_token = create_access_token(
        data={"sub": user.username},
        expires_delta=timedelta(minutes=15)
    )
    refresh_token = create_refresh_token(user.username)

    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
        "expires_in": 900
    }

@app.post("/token/refresh")
async def refresh_access_token(refresh_token: str):
    """Get new access token using refresh token"""
    token_data = refresh_tokens_db.get(refresh_token)

    if not token_data:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    # Check if expired
    if datetime.now(timezone.utc) > token_data["expires_at"]:
        del refresh_tokens_db[refresh_token]
        raise HTTPException(status_code=401, detail="Refresh token expired")

    # Create new access token
    access_token = create_access_token(
        data={"sub": token_data["username"]},
        expires_delta=timedelta(minutes=15)
    )

    return {
        "access_token": access_token,
        "token_type": "bearer",
        "expires_in": 900
    }

@app.post("/logout")
async def logout(
    refresh_token: str,
    current_user: User = Depends(get_current_user)
):
    """Revoke refresh token on logout"""
    if refresh_token in refresh_tokens_db:
        del refresh_tokens_db[refresh_token]
    return {"message": "Logged out successfully"}

Result: Users stay logged in for 30 days without re-entering passwords. Access tokens expire quickly for security, but can be refreshed silently.

Refresh Token Security

Store securely: Use httpOnly cookies or secure storage
Rotate on use: Issue new refresh token with each refresh
Detect reuse: If old refresh token is used, revoke all user tokens
Limit count: Maximum number of devices per user
Monitor activity: Log token usage for suspicious patterns

Security Best Practices

Authentication and authorization are only as strong as their implementation. Follow these practices to build secure APIs.

Password Security

Hash passwords: Use bcrypt, argon2, or scrypt
Salt each password: Prevent rainbow table attacks
Enforce strength: Minimum length, complexity requirements
Rate limit attempts: Prevent brute force attacks
Never log passwords: Not even failed login attempts
Use password reset tokens: Time-limited, single-use

Token Security

Short expiration: 15-60 minutes for access tokens
Strong secrets: 256+ bit random keys
Use HTTPS always: Encrypt token transmission
Validate signatures: On every token use
Secure storage: Never localStorage for sensitive tokens
Include token ID: For revocation tracking

Attack Prevention

CSRF protection: Use CSRF tokens or SameSite cookies
XSS prevention: Sanitize user input, CSP headers
SQL injection: Use parameterized queries
Timing attacks: Constant-time comparisons
Session fixation: Regenerate session on login
Clickjacking: X-Frame-Options header

Monitoring & Auditing

Log auth events: Logins, failures, permission changes
Monitor anomalies: Unusual login patterns
Track token usage: Detect stolen tokens
Alert on suspicious activity: Multiple failed attempts
Regular security audits: Review permissions
Incident response plan: Know what to do if breached

Security Headers

from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()

# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourapp.com"],  # Specific origins, not "*"
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

# Trusted hosts
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["yourapi.com", "*.yourapi.com"]
)

# Security headers middleware
@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

Result: Essential security headers protect against common attacks like XSS, clickjacking, and protocol downgrade attacks.

Common Authentication Pitfalls

Bad:

"User not found"
"Incorrect password"

Reveals which emails are registered

Good:

"Invalid credentials"
(same message for both)

Prevents user enumeration attacks

Problem: Accessible to JavaScript, vulnerable to XSS attacks

Better: Use httpOnly cookies for tokens (inaccessible to JavaScript)

Best: httpOnly + Secure + SameSite cookies

Problem: Trusting token content without verification allows forged tokens

Always: Validate JWT signature on every request, verify expiration, check issuer