API Security Best Practices

Protect your APIs from common vulnerabilities and attacks with industry-standard security practices

Why API Security Matters

APIs are the backbone of modern applications, but they're also prime targets for attackers. A single vulnerability can expose sensitive data, compromise user accounts, or bring down entire systems.

Real-world impact: The 2021 T-Mobile data breach exposed 76 million records through API vulnerabilities. The 2019 Capital One breach affected 100 million customers due to a misconfigured API. Equifax lost 147 million records in 2017 through unpatched API security flaws.

In this lesson, we'll cover the OWASP API Security Top 10 and learn practical techniques to protect your APIs from common attacks. You'll implement input validation, prevent injection attacks, configure security headers, manage secrets, and build comprehensive security monitoring.

OWASP API Security Top 10

The Open Web Application Security Project (OWASP) maintains a list of the most critical API security risks. Understanding these vulnerabilities is the first step to protecting your APIs.

Vulnerability	Description	Impact
API1:2023 Broken Object Level Authorization	APIs expose endpoints that handle object identifiers, creating a wide attack surface for authorization issues (IDOR attacks)	Critical
API2:2023 Broken Authentication	Authentication mechanisms are often implemented incorrectly, allowing attackers to compromise tokens or exploit flaws	Critical
API3:2023 Broken Object Property Level Authorization	Lack of validation on object properties allows unauthorized access or modification (mass assignment, data exposure)	Critical
API4:2023 Unrestricted Resource Consumption	APIs don't restrict resource consumption, leading to DoS attacks, data scraping, or excessive costs	High
API5:2023 Broken Function Level Authorization	APIs with complex user hierarchies fail to properly enforce function-level permissions	High
API6:2023 Unrestricted Access to Sensitive Business Flows	APIs expose business flows without compensating for automation risks (ticket buying bots, fake accounts)	High
API7:2023 Server Side Request Forgery	APIs fetch remote resources without validating user-supplied URIs, allowing attackers to access internal systems	High
API8:2023 Security Misconfiguration	Missing security hardening, unnecessary features enabled, verbose error messages exposing system details	High
API9:2023 Improper Inventory Management	Outdated API versions, unpatched endpoints, exposed debug endpoints	Medium
API10:2023 Unsafe Consumption of APIs	Developers trust third-party APIs without proper validation, leading to vulnerabilities	Medium

How This Lesson Addresses OWASP Top 10

Section 2-3: Input validation and injection prevention (API7, API8)
Section 4: XSS and CSRF protection (API8)
Section 5: Authorization vulnerabilities and mass assignment (API1, API3, API5)
Section 6: Security headers and CORS (API8)
Section 7: HTTPS, secrets, configuration management (API8)
Section 8: Security monitoring and audit logging (API8, API9)

Input Validation and Sanitization

Never trust user input. Every piece of data from clients should be validated, sanitized, and constrained. Input validation is your first line of defense against many attacks.

Validation Strategy: Whitelist Over Blacklist

Whitelist approach (recommended): Define what is allowed and reject everything else.

Blacklist approach (avoid): Try to filter out known bad patterns. Attackers will find bypasses.

Pydantic Validation in FastAPI

FastAPI uses Pydantic models for automatic validation. Pydantic enforces type checking, constraints, and custom validators before your endpoint code even runs.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, field_validator, EmailStr
from typing import Optional
import re

app = FastAPI()


class UserCreate(BaseModel):
    username: str = Field(
        ...,
        min_length=3,
        max_length=20,
        pattern="^[a-zA-Z0-9_-]+$"  # Only alphanumeric, underscore, hyphen
    )
    email: EmailStr  # Validates email format automatically
    age: int = Field(..., ge=13, le=120)  # Greater than or equal to 13, less than or equal to 120
    bio: Optional[str] = Field(None, max_length=500)
    website: Optional[str] = None

    @field_validator('username')
    @classmethod
    def username_not_reserved(cls, v):
        """Custom validator: prevent reserved usernames"""
        reserved = {'admin', 'root', 'system', 'moderator'}
        if v.lower() in reserved:
            raise ValueError('This username is reserved')
        return v

    @field_validator('website')
    @classmethod
    def validate_website(cls, v):
        """Custom validator: ensure website is HTTPS"""
        if v is None:
            return v
        if not v.startswith('https://'):
            raise ValueError('Website must use HTTPS')
        # Additional URL validation
        url_pattern = r'^https://[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}(/.*)?$'
        if not re.match(url_pattern, v):
            raise ValueError('Invalid website URL')
        return v

    @field_validator('bio')
    @classmethod
    def sanitize_bio(cls, v):
        """Sanitize HTML tags from bio"""
        if v is None:
            return v
        # Remove HTML tags
        v = re.sub(r'<[^>]+>', '', v)
        # Remove potentially dangerous characters
        v = re.sub(r'[<>]', '', v)
        return v.strip()


@app.post("/api/users")
async def create_user(user: UserCreate):
    """
    Pydantic automatically validates:
    - username: 3-20 chars, alphanumeric only, not reserved
    - email: valid email format
    - age: between 13 and 120
    - bio: max 500 chars, sanitized
    - website: HTTPS only, valid URL format
    """
    return {
        "message": "User created successfully",
        "user": user.model_dump()
    }

Result

Valid request: Username "john_doe", email "john@example.com", age 25, bio "Software engineer", website "https://example.com" → ✓ Accepted

Invalid username: "admin" → ✗ Rejected: "This username is reserved"

Invalid email: "not-an-email" → ✗ Rejected: "value is not a valid email address"

Invalid website: "http://insecure.com" → ✗ Rejected: "Website must use HTTPS"

Common Validation Patterns

String Validation

Length constraints (min/max)
Regex patterns (alphanumeric, email, phone)
Whitelist of allowed values
Trim whitespace
Remove HTML/script tags

Numeric Validation

Range constraints (ge, le, gt, lt)
Integer vs float vs decimal
Positive/negative only
Multiple of X (e.g., price in cents)

Date/Time Validation

ISO 8601 format enforcement
Future/past date constraints
Date range validation
Timezone awareness

File Upload Validation

File size limits (prevent DoS)
MIME type whitelist
File extension validation
Magic byte verification
Virus scanning

Common Validation Mistakes

Client-side only validation: Never rely on frontend validation alone. Always validate server-side.
Inconsistent validation: Apply the same rules across all endpoints (create, update, search).
Verbose error messages: Don't reveal system internals. "Invalid input" is better than "Database column username failed constraint check".
Validation after processing: Validate BEFORE any business logic, database queries, or external API calls.

Injection Attacks Prevention

Injection attacks occur when untrusted data is sent to an interpreter as part of a command or query. The attacker tricks the interpreter into executing unintended commands or accessing unauthorized data.

SQL Injection (SQLi)

SQL injection is one of the most dangerous vulnerabilities. Attackers can read, modify, or delete database data, bypass authentication, or even execute operating system commands.

Vulnerable Code (String Concatenation)

from fastapi import FastAPI, HTTPException
import sqlite3

app = FastAPI()

@app.get("/api/users/{username}")
async def get_user(username: str):
    conn = sqlite3.connect("database.db")
    cursor = conn.cursor()

    # VULNERABLE: String concatenation allows SQL injection
    query = f"SELECT * FROM users WHERE username = '{username}'"
    cursor.execute(query)

    user = cursor.fetchone()
    conn.close()
    return {"user": user}


# Attack: GET /api/users/admin' OR '1'='1
# Executed query: SELECT * FROM users WHERE username = 'admin' OR '1'='1'
# Result: Returns all users (always true condition)

Secure Code (Parameterized Queries)

from fastapi import FastAPI, HTTPException
import sqlite3

app = FastAPI()

@app.get("/api/users/{username}")
async def get_user(username: str):
    conn = sqlite3.connect("database.db")
    cursor = conn.cursor()

    # SECURE: Parameterized query prevents SQL injection
    query = "SELECT * FROM users WHERE username = ?"
    cursor.execute(query, (username,))

    user = cursor.fetchone()
    conn.close()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return {"user": user}


# Attack attempt: GET /api/users/admin' OR '1'='1
# Executed query: SELECT * FROM users WHERE username = 'admin'' OR ''1''=''1'
# Result: No user found (literal string match)

The placeholder ? ensures the input is treated as data, not code. The database driver automatically escapes special characters.

SQL Injection with ORMs (SQLAlchemy)

Using an ORM like SQLAlchemy significantly reduces SQL injection risk, but you must still use it correctly.

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session

DATABASE_URL = "sqlite:///./database.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

class Base(DeclarativeBase):
    pass

app = FastAPI()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True, index=True)
    email = Column(String)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/api/users/{username}")
async def get_user(username: str, db: Session = Depends(get_db)):
    # SECURE: SQLAlchemy automatically parameterizes queries
    user = db.query(User).filter(User.username == username).first()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return {
        "id": user.id,
        "username": user.username,
        "email": user.email
    }


# ⚠️ DANGEROUS: Raw SQL with ORM
@app.get("/api/users/search")
async def search_users_dangerous(query: str, db: Session = Depends(get_db)):
    # VULNERABLE: String formatting with raw SQL
    sql = f"SELECT * FROM users WHERE username LIKE '%{query}%'"
    result = db.execute(sql)
    return {"users": result.fetchall()}


# ✓ SAFE: Raw SQL with parameters
@app.get("/api/users/search")
async def search_users_safe(query: str, db: Session = Depends(get_db)):
    # SECURE: Parameterized raw SQL
    from sqlalchemy import text
    sql = text("SELECT * FROM users WHERE username LIKE :pattern")
    result = db.execute(sql, {"pattern": f"%{query}%"})
    return {"users": result.fetchall()}

Key Takeaway

Always use parameterized queries or ORM query builders. Never concatenate user input into SQL strings, even with ORMs. When using raw SQL, use named parameters (:param or ?).

NoSQL Injection (MongoDB, Redis)

NoSQL databases are also vulnerable to injection attacks when queries are constructed from user input.

Vulnerable MongoDB Query

from fastapi import FastAPI
from pymongo import MongoClient

app = FastAPI()
client = MongoClient("mongodb://localhost:27017/")
db = client.mydb

@app.post("/api/login")
async def login(username: str, password: str):
    # VULNERABLE: Allows NoSQL injection via MongoDB operators
    user = db.users.find_one({
        "username": username,
        "password": password
    })

    if user:
        return {"message": "Login successful"}
    return {"message": "Invalid credentials"}


# Attack: POST /api/login
# Body: {"username": "admin", "password": {"$ne": null}}
# MongoDB query: {username: "admin", password: {$ne: null}}
# Result: Login successful (password is not null = always true)

Secure MongoDB Query

from fastapi import FastAPI, HTTPException
from pymongo import MongoClient
from pydantic import BaseModel, field_validator
import bcrypt

app = FastAPI()
client = MongoClient("mongodb://localhost:27017/")
db = client.mydb


class LoginRequest(BaseModel):
    username: str
    password: str

    @field_validator('username', 'password')
    @classmethod
    def validate_string(cls, v):
        """Ensure inputs are strings, not objects"""
        if not isinstance(v, str):
            raise ValueError('Must be a string')
        if len(v) < 1 or len(v) > 100:
            raise ValueError('Invalid length')
        return v


@app.post("/api/login")
async def login(credentials: LoginRequest):
    # SECURE: Pydantic ensures username and password are strings
    # MongoDB operators like $ne are rejected during validation
    user = db.users.find_one({"username": credentials.username})

    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Verify hashed password
    if not bcrypt.checkpw(
        credentials.password.encode('utf-8'),
        user['password'].encode('utf-8')
    ):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    return {"message": "Login successful", "user_id": str(user['_id'])}


# Attack attempt: POST /api/login
# Body: {"username": "admin", "password": {"$ne": null}}
# Result: 422 Validation Error - password must be a string

Defense: Validate that inputs are the expected primitive types (string, number). Reject objects that could contain MongoDB operators like $ne, $gt, $regex.

Command Injection

Command injection occurs when user input is passed to system shell commands. Attackers can execute arbitrary operating system commands.

Vulnerable Code (Shell=True)

from fastapi import FastAPI
import subprocess

app = FastAPI()

@app.get("/api/ping")
async def ping(host: str):
    # VULNERABLE: Allows command injection via shell
    result = subprocess.run(
        f"ping -c 4 {host}",
        shell=True,  # Dangerous!
        capture_output=True
    )
    return {"output": result.stdout.decode()}


# Attack: GET /api/ping?host=example.com; rm -rf /
# Executed command: ping -c 4 example.com; rm -rf /
# Result: Pings example.com, then deletes entire filesystem

Secure Code (Whitelist + No Shell)

from fastapi import FastAPI, HTTPException
import subprocess
import re

app = FastAPI()

@app.get("/api/ping")
async def ping(host: str):
    # Validate: Only allow hostname or IP address
    if not re.match(r'^[a-zA-Z0-9.-]+$', host):
        raise HTTPException(status_code=400, detail="Invalid host format")

    # Additional check: Prevent command injection characters
    dangerous_chars = [';', '&', '|', '$', '\n', '\r', '(', ')', '<', '>']
    if any(char in host for char in dangerous_chars):
        raise HTTPException(status_code=400, detail="Invalid characters in host")

    try:
        # SECURE: Pass arguments as list, shell=False
        result = subprocess.run(
            ["ping", "-c", "4", host],
            shell=False,  # Never use shell=True with user input
            capture_output=True,
            timeout=10  # Prevent hanging
        )
        return {"output": result.stdout.decode()}
    except subprocess.TimeoutExpired:
        raise HTTPException(status_code=408, detail="Request timeout")


# Attack attempt: GET /api/ping?host=example.com; rm -rf /
# Result: 400 Bad Request - Invalid characters in host

Best practices: (1) Never use shell=True with user input, (2) Pass arguments as a list, not a string, (3) Whitelist allowed characters, (4) Use built-in libraries instead of shell commands when possible.

Injection Prevention Checklist

SQL: Use parameterized queries or ORM query builders. Never concatenate user input into SQL.
NoSQL: Validate that inputs are primitive types. Reject objects containing operators.
Commands: Avoid shell execution. If necessary, use argument arrays and whitelist input.
LDAP/XPath: Use parameterized interfaces. Escape special characters for the query language.
General: Principle of least privilege - run services with minimal permissions needed.

Cross-Site Scripting (XSS) and CSRF Protection

While APIs primarily serve data (not HTML), understanding XSS and CSRF is crucial when your API is consumed by web frontends or serves any HTML content.

Cross-Site Scripting (XSS) in APIs

XSS occurs when an application includes untrusted data in a web page without proper validation or escaping. While JSON APIs are less vulnerable, XSS risks exist when:

API returns HTML content (error messages, rich text fields)
API serves a web interface (API docs, admin panels)
Frontend renders API data without sanitization
API sets cookies or localStorage that frontend uses unsafely

Vulnerable: Unescaped User Content

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.get("/api/profile/{username}", response_class=HTMLResponse)
async def get_profile(username: str):
    # VULNERABLE: User input directly in HTML
    return f"""
    <html>
        <body>
            <h1>Profile: {username}</h1>
        </body>
    </html>
    """

# Attack: GET /api/profile/<script>alert('XSS')</script>
# Result: JavaScript executes in victim's browser

Secure: Escaped Output + Content-Type

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import html

app = FastAPI()

@app.get("/api/profile/{username}", response_class=HTMLResponse)
async def get_profile(username: str):
    # SECURE: Escape HTML entities
    safe_username = html.escape(username)

    return f"""
    <html>
        <body>
            <h1>Profile: {safe_username}</h1>
        </body>
    </html>
    """

# Attack attempt: GET /api/profile/<script>alert('XSS')</script>
# Rendered: <h1>Profile: &lt;script&gt;alert('XSS')&lt;/script&gt;</h1>
# Result: Displays as text, does not execute


# BEST PRACTICE: Return JSON, let frontend handle rendering
@app.get("/api/profile/{username}")
async def get_profile_json(username: str):
    # No XSS risk - JSON is automatically escaped
    return {"username": username}  # FastAPI returns application/json

Best practice: Return pure JSON from APIs. Let frontend frameworks (React, Vue, Angular) handle rendering - they escape content by default.

Cross-Site Request Forgery (CSRF)

CSRF attacks trick authenticated users into performing unwanted actions. An attacker creates a malicious website that sends requests to your API using the victim's credentials (cookies).

When CSRF Protection is Needed

CSRF Protection Required

Cookie-based authentication
Session tokens in cookies
HTTP Basic Authentication
Any browser-stored credentials

CSRF Protection NOT Needed

JWT in Authorization header
API keys in headers
OAuth tokens (not in cookies)
Custom header authentication

Why JWTs in Headers Are Safe from CSRF

When using Authorization: Bearer <token> headers, CSRF is not a risk because:

Browsers do not automatically send custom headers with cross-origin requests
JavaScript can only add headers to requests it explicitly creates
An attacker's website cannot make your browser send an Authorization header to your API
Cookies are automatically sent by browsers - that's why they're vulnerable to CSRF

from fastapi import FastAPI, HTTPException, Request, Response, Depends
from fastapi.responses import JSONResponse
import secrets
import hmac
import hashlib

app = FastAPI()

# Store CSRF tokens (in production, use Redis or session store)
csrf_tokens = {}


def generate_csrf_token(session_id: str) -> str:
    """Generate a CSRF token tied to user session"""
    token = secrets.token_urlsafe(32)
    csrf_tokens[session_id] = token
    return token


def verify_csrf_token(request: Request) -> bool:
    """Verify CSRF token from request header matches session"""
    session_id = request.cookies.get("session_id")
    if not session_id:
        return False

    # Get token from header
    token = request.headers.get("X-CSRF-Token")
    if not token:
        return False

    # Verify token matches session
    expected_token = csrf_tokens.get(session_id)
    if not expected_token:
        return False

    # Use constant-time comparison to prevent timing attacks
    return hmac.compare_digest(token, expected_token)


@app.post("/api/login")
async def login(username: str, password: str, response: Response):
    """Login endpoint - sets session cookie and CSRF token"""
    # Verify credentials (simplified)
    if username != "admin" or password != "secret":
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Create session
    session_id = secrets.token_urlsafe(32)
    response.set_cookie(
        key="session_id",
        value=session_id,
        httponly=True,  # Prevent JavaScript access
        secure=True,    # HTTPS only
        samesite="strict"  # Additional CSRF protection
    )

    # Generate CSRF token
    csrf_token = generate_csrf_token(session_id)

    return {
        "message": "Login successful",
        "csrf_token": csrf_token  # Frontend stores this, sends in X-CSRF-Token header
    }


@app.post("/api/transfer")
async def transfer_money(amount: int, to_account: str, request: Request):
    """Protected endpoint - requires valid CSRF token"""
    # Verify CSRF token
    if not verify_csrf_token(request):
        raise HTTPException(status_code=403, detail="Invalid CSRF token")

    # Verify session (simplified)
    session_id = request.cookies.get("session_id")
    if not session_id:
        raise HTTPException(status_code=401, detail="Not authenticated")

    # Process transfer
    return {
        "message": f"Transferred ${amount} to {to_account}",
        "success": True
    }


# Frontend usage:
# 1. After login, store csrf_token in JavaScript variable (NOT localStorage - XSS risk)
# 2. Include token in all state-changing requests:
#    fetch('/api/transfer', {
#        method: 'POST',
#        headers: {
#            'X-CSRF-Token': csrfToken,
#            'Content-Type': 'application/json'
#        },
#        credentials: 'include',  // Include cookies
#        body: JSON.stringify({amount: 100, to_account: 'attacker'})
#    })

How CSRF Protection Works

Login: Server generates CSRF token, returns it to frontend, stores it tied to session
Storage: Frontend stores token in JavaScript variable (NOT cookie)
Request: Frontend includes token in custom header (X-CSRF-Token) for state-changing operations
Verification: Server validates token matches session. Attacker's site cannot read or send custom headers.

Additional CSRF Defenses

SameSite Cookie Attribute

Set SameSite=Strict or SameSite=Lax on cookies:

response.set_cookie(
    key="session_id",
    value=session_id,
    samesite="strict"  # Prevents cross-site sending
)

Double-Submit Cookie Pattern

Send CSRF token in both cookie and header/body:

Set CSRF token as cookie
Require token in header
Verify cookie matches header
Attacker can't read cookies

Authorization Vulnerabilities (BOLA, IDOR, Mass Assignment)

Broken Object Level Authorization (BOLA) is the #1 API security risk according to OWASP. It occurs when APIs fail to verify that the authenticated user has permission to access the requested resource.

Broken Object Level Authorization (BOLA / IDOR)

Also known as Insecure Direct Object Reference (IDOR), this vulnerability allows users to access or modify resources they shouldn't by manipulating object identifiers.

Vulnerable Code (No Authorization Check)

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session

app = FastAPI()

@app.get("/api/documents/{document_id}")
async def get_document(document_id: int, db: Session = Depends(get_db)):
    """VULNERABLE: No check if user owns the document"""
    document = db.query(Document).filter(Document.id == document_id).first()

    if not document:
        raise HTTPException(status_code=404, detail="Document not found")

    # ❌ PROBLEM: Returns document regardless of who owns it
    return {"document": document}


# User A (authenticated) requests: GET /api/documents/123
# Server returns document 123 even if it belongs to User B
# Attack: User A can enumerate IDs (1, 2, 3...) to access all documents

Secure Code (Ownership Verification)

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session

app = FastAPI()

async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
    """Get authenticated user from token"""
    # Verify JWT and return user (see Lesson 4)
    ...

@app.get("/api/documents/{document_id}")
async def get_document(
    document_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """SECURE: Verify user owns the document"""
    document = db.query(Document).filter(
        Document.id == document_id,
        Document.owner_id == current_user.id  # ✓ Authorization check
    ).first()

    if not document:
        # Same error for not found vs not authorized (prevents enumeration)
        raise HTTPException(status_code=404, detail="Document not found")

    return {"document": document}


# User A requests: GET /api/documents/123 (owned by User B)
# Result: 404 Not Found (query filters by owner_id)
# User A cannot access documents they don't own

Key principle: Always filter resources by the authenticated user's ID. Never rely on the client to "only request their own resources."

Advanced BOLA Prevention Patterns

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from enum import Enum

app = FastAPI()


class Permission(str, Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"


def verify_resource_access(
    resource_type: str,
    resource_id: int,
    required_permission: Permission,
    current_user: User,
    db: Session
) -> bool:
    """Generic authorization checker for any resource type"""

    if resource_type == "document":
        resource = db.query(Document).filter(Document.id == resource_id).first()
        if not resource:
            return False

        # Check ownership
        if resource.owner_id == current_user.id:
            return True

        # Check shared access
        access = db.query(DocumentAccess).filter(
            DocumentAccess.document_id == resource_id,
            DocumentAccess.user_id == current_user.id
        ).first()

        if access:
            # Verify permission level
            if required_permission == Permission.READ:
                return True
            elif required_permission == Permission.WRITE:
                return access.can_write
            elif required_permission == Permission.DELETE:
                return access.can_delete

        return False

    # Add other resource types...
    return False


@app.get("/api/documents/{document_id}")
async def get_document(
    document_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """Get document with permission check"""
    if not verify_resource_access(
        "document", document_id, Permission.READ, current_user, db
    ):
        raise HTTPException(status_code=404, detail="Document not found")

    document = db.query(Document).filter(Document.id == document_id).first()
    return {"document": document}


@app.delete("/api/documents/{document_id}")
async def delete_document(
    document_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """Delete document - requires DELETE permission"""
    if not verify_resource_access(
        "document", document_id, Permission.DELETE, current_user, db
    ):
        raise HTTPException(status_code=403, detail="Insufficient permissions")

    db.query(Document).filter(Document.id == document_id).delete()
    db.commit()
    return {"message": "Document deleted"}

Mass Assignment Vulnerabilities

Mass assignment occurs when an application automatically binds request parameters to internal objects, allowing attackers to modify fields they shouldn't have access to.

Vulnerable Code (Unrestricted Field Updates)

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()

class UserUpdate(BaseModel):
    """VULNERABLE: Allows updating ANY field"""
    pass  # No field restrictions

@app.patch("/api/users/{user_id}")
async def update_user(
    user_id: int,
    updates: dict,  # ❌ Accepts arbitrary fields
    db: Session = Depends(get_db)
):
    """Update user profile"""
    user = db.query(User).filter(User.id == user_id).first()

    # ❌ PROBLEM: Applies all fields from request
    for key, value in updates.items():
        setattr(user, key, value)

    db.commit()
    return {"user": user}


# Attack: PATCH /api/users/123
# Body: {"name": "John", "is_admin": true, "account_balance": 1000000}
# Result: User grants themselves admin privileges and adds money

Secure Code (Explicit Field Whitelist)

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

class UserUpdate(BaseModel):
    """SECURE: Only allows specific, safe fields"""
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    bio: Optional[str] = Field(None, max_length=500)
    avatar_url: Optional[str] = None

    # ✓ Sensitive fields NOT included:
    # - is_admin (only admins can modify)
    # - account_balance (separate endpoint)
    # - email (requires verification)
    # - password_hash (separate endpoint)


@app.patch("/api/users/{user_id}")
async def update_user(
    user_id: int,
    updates: UserUpdate,  # ✓ Pydantic model with explicit fields
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    """Update user profile - only safe fields"""
    # Verify user can only update their own profile
    if user_id != current_user.id:
        raise HTTPException(status_code=403, detail="Can only update own profile")

    user = db.query(User).filter(User.id == user_id).first()

    # ✓ Only update provided fields from whitelist
    update_data = updates.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(user, key, value)

    db.commit()
    return {"user": user}


# Attack attempt: PATCH /api/users/123
# Body: {"name": "John", "is_admin": true}
# Result: 422 Validation Error - "is_admin" is not a valid field

Defense: Define explicit Pydantic models with only the fields users should be able to modify. Never accept arbitrary dict objects.

Authorization Security Checklist

Object-level: Always verify the current user has permission to access the specific resource (not just "any resource of this type").
Function-level: Check user role/permissions for the specific action (read vs write vs delete).
Mass assignment: Use explicit Pydantic models. Never accept arbitrary dict or **kwargs for updates.
Error messages: Return same error (404) whether resource doesn't exist or user lacks permission (prevents enumeration).
Testing: Test authorization boundaries - try accessing User B's resources as User A.
UUIDs vs integers: Consider using UUIDs for resource IDs to prevent enumeration attacks (but still validate ownership!).

Security Headers and CORS Configuration

HTTP security headers provide an additional layer of defense by instructing browsers how to handle your application's content. Proper CORS configuration ensures only trusted origins can access your API.

Essential Security Headers

Header	Purpose	Recommended Value
`Strict-Transport-Security`	Force HTTPS connections, prevent downgrade attacks	`max-age=31536000; includeSubDomains`
`X-Content-Type-Options`	Prevent MIME-sniffing, enforce declared content type	`nosniff`
`X-Frame-Options`	Prevent clickjacking by disabling iframe embedding	`DENY` or `SAMEORIGIN`
`Content-Security-Policy`	Control which resources can be loaded (scripts, styles, etc.)	`default-src 'self'`
`Referrer-Policy`	Control how much referrer information is sent	`strict-origin-when-cross-origin`
`Permissions-Policy`	Control browser features (camera, microphone, geolocation)	`geolocation=(), microphone=()`

from fastapi import FastAPI, Request
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response

app = FastAPI()


class SecurityHeadersMiddleware(BaseHTTPMiddleware):
    """Add security headers to all responses"""

    async def dispatch(self, request: Request, call_next):
        response = await call_next(request)

        # Strict-Transport-Security (HSTS)
        response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"

        # X-Content-Type-Options
        response.headers["X-Content-Type-Options"] = "nosniff"

        # X-Frame-Options
        response.headers["X-Frame-Options"] = "DENY"

        # Content-Security-Policy
        response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self'; style-src 'self'"

        # Referrer-Policy
        response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"

        # Permissions-Policy
        response.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=()"

        # Remove server header (don't reveal FastAPI/Starlette version)
        response.headers.pop("Server", None)

        return response


# Add security middleware
app.add_middleware(SecurityHeadersMiddleware)

# Add trusted host middleware (prevent host header attacks)
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["api.example.com", "www.example.com"]
)

# Add gzip compression
app.add_middleware(GZipMiddleware, minimum_size=1000)


@app.get("/api/health")
async def health_check():
    return {"status": "healthy"}


# Response will include all security headers:
# Strict-Transport-Security: max-age=31536000; includeSubDomains
# X-Content-Type-Options: nosniff
# X-Frame-Options: DENY
# Content-Security-Policy: default-src 'self'; script-src 'self'; style-src 'self'
# Referrer-Policy: strict-origin-when-cross-origin
# Permissions-Policy: geolocation=(), microphone=(), camera=()

CORS (Cross-Origin Resource Sharing) Configuration

CORS controls which external domains can access your API from web browsers. Misconfigured CORS is a common security vulnerability that can expose your API to unauthorized access.

How CORS Works

Browser detects cross-origin request (e.g., https://app.example.com → https://api.example.com)
For simple requests: Browser sends request with Origin header
For complex requests: Browser sends preflight OPTIONS request first
Server responds with Access-Control-Allow-Origin header
Browser allows/blocks response based on CORS headers

Insecure CORS Configuration

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

app = FastAPI()

# ❌ DANGEROUS: Allows ALL origins
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Any website can access your API
    allow_credentials=True,  # Sends cookies/auth headers
    allow_methods=["*"],  # All HTTP methods allowed
    allow_headers=["*"],  # All headers allowed
)

# Problem: Malicious website attacker.com can make authenticated
# requests to your API using the victim's credentials (cookies)

Secure CORS Configuration

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

app = FastAPI()

# ✓ SECURE: Explicit whitelist of allowed origins
allowed_origins = [
    "https://app.example.com",  # Production frontend
    "https://www.example.com",  # Marketing site
]

# Add localhost for development (only if DEV environment)
if os.getenv("ENV") == "development":
    allowed_origins.extend([
        "http://localhost:3000",
        "http://localhost:8080"
    ])

app.add_middleware(
    CORSMiddleware,
    allow_origins=allowed_origins,  # ✓ Explicit whitelist
    allow_credentials=True,  # OK with specific origins
    allow_methods=["GET", "POST", "PUT", "DELETE"],  # ✓ Specific methods
    allow_headers=["Content-Type", "Authorization", "X-CSRF-Token"],  # ✓ Specific headers
    max_age=3600,  # Cache preflight responses for 1 hour
)


# Alternative: Dynamic origin validation
from fastapi import Request

@app.middleware("http")
async def validate_origin(request: Request, call_next):
    """Custom CORS validation with logging"""
    origin = request.headers.get("origin")

    if origin and origin not in allowed_origins:
        # Log suspicious requests
        print(f"Blocked CORS request from unauthorized origin: {origin}")

    response = await call_next(request)
    return response

Best practice: Use specific origin whitelist. Only use allow_origins=["*"]for truly public APIs that don't use credentials.

CORS Common Mistakes

Allow all origins with credentials: allow_origins=["*"] + allow_credentials=True is a critical vulnerability.
Regex origin matching: Validating origins with regex like .*\\.example\\.com can be bypassed with attacker-example.com.
Reflecting Origin header: Blindly copying the request Origin to Access-Control-Allow-Origin without validation.
Preflight caching issues: Not setting max_age causes excessive OPTIONS requests.
Development CORS in production: Leaving localhost origins in production configuration.

CORS for Different API Types

Public API (No Auth)

Weather data, public content, read-only endpoints:

allow_origins=["*"]
allow_credentials=False
allow_methods=["GET"]

Private API (Auth Required)

User data, authenticated endpoints:

allow_origins=["https://app.example.com"]
allow_credentials=True
allow_methods=["GET", "POST", "PUT", "DELETE"]

HTTPS, Secret Management, and Secure Configuration

Secure transport (HTTPS), proper secret management, and hardened configuration are foundational security practices. A single exposed secret can compromise your entire system.

HTTPS / TLS Enforcement

Always use HTTPS in production. HTTPS encrypts data in transit, preventing eavesdropping, tampering, and man-in-the-middle attacks. HTTP is only acceptable for local development.

Why HTTPS Matters

Encrypts all data (credentials, tokens, personal info)
Prevents packet sniffing on networks
Verifies server identity (prevents impersonation)
Required for modern browser features
Required for PCI compliance, GDPR

HTTP Risks

Passwords sent in plaintext
Tokens can be intercepted
Data can be modified in transit
Session hijacking on public WiFi
Browsers warn users about insecure sites

from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()


class HTTPSRedirectMiddleware(BaseHTTPMiddleware):
    """Redirect HTTP requests to HTTPS in production"""

    async def dispatch(self, request: Request, call_next):
        # Check if request is HTTP (not HTTPS)
        if request.url.scheme == "http":
            # Get environment
            import os
            if os.getenv("ENV") == "production":
                # Redirect to HTTPS
                url = request.url.replace(scheme="https")
                from starlette.responses import RedirectResponse
                return RedirectResponse(url, status_code=301)

        response = await call_next(request)
        return response


# Add HTTPS redirect middleware
app.add_middleware(HTTPSRedirectMiddleware)


# Alternative: Reject HTTP requests entirely
@app.middleware("http")
async def enforce_https(request: Request, call_next):
    """Reject HTTP requests in production"""
    import os
    if os.getenv("ENV") == "production" and request.url.scheme == "http":
        raise HTTPException(
            status_code=403,
            detail="HTTPS is required for this API"
        )

    response = await call_next(request)
    return response


# Running with HTTPS locally (for testing):
# uvicorn main:app --ssl-keyfile=key.pem --ssl-certfile=cert.pem

# Production: Use reverse proxy (nginx, Caddy) or platform-managed TLS
# Most cloud platforms (AWS, GCP, Railway, Render) handle TLS termination

HTTPS in Production (Deployment Strategies)

Cloud platforms (AWS, GCP, Azure): Use load balancer with TLS termination (ALB, Cloud Load Balancing)
PaaS (Railway, Render, Heroku): HTTPS automatically provided and managed
Self-hosted: Use reverse proxy (nginx, Caddy) with Let's Encrypt certificates
Certificate management: Automate renewal with cert-manager (Kubernetes) or Certbot

Secret Management

Never hardcode secrets in source code. Database passwords, API keys, JWT secrets, and encryption keys must be stored securely and accessed through environment variables or secret managers.

Insecure: Hardcoded Secrets

# ❌ DANGEROUS: Secrets in code
DATABASE_URL = "postgresql://admin:P@ssw0rd123@db.example.com/prod"
JWT_SECRET = "super-secret-key-12345"
STRIPE_API_KEY = "sk_live_abc123xyz789"

# Problems:
# 1. Secrets visible in source control (git)
# 2. Secrets exposed in code reviews
# 3. Secrets accessible to anyone with repo access
# 4. Can't rotate secrets without code changes
# 5. Same secrets used across environments

Secure: Environment Variables

# .env (NOT committed to git)
DATABASE_URL=postgresql://admin:P@ssw0rd123@db.example.com/prod
JWT_SECRET=super-secret-key-12345
STRIPE_API_KEY=sk_live_abc123xyz789
ENV=production

# .gitignore
.env
.env.*


# Python code - use environment variables
# pip install pydantic-settings
from fastapi import FastAPI
from pydantic_settings import BaseSettings, SettingsConfigDict
import os

class Settings(BaseSettings):
    """Application settings loaded from environment"""
    database_url: str
    jwt_secret: str
    stripe_api_key: str
    env: str = "development"

    model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)


# Create settings instance
settings = Settings()

# Use settings throughout application
app = FastAPI()

@app.get("/api/config")
async def get_config():
    """Never expose secrets in API responses"""
    return {
        "env": settings.env,
        # ❌ Don't include: database_url, jwt_secret, api_keys
    }

Benefits: Secrets not in source control, different values per environment, can rotate without code changes, supported by all deployment platforms.

Advanced Secret Management

# Using cloud secret managers (AWS Secrets Manager, GCP Secret Manager)
import boto3
from botocore.exceptions import ClientError

def get_secret(secret_name: str, region_name: str = "us-east-1") -> dict:
    """Retrieve secret from AWS Secrets Manager"""
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        response = client.get_secret_value(SecretId=secret_name)
        import json
        return json.loads(response['SecretString'])
    except ClientError as e:
        raise Exception(f"Failed to retrieve secret: {e}")


# Load secrets at startup
secrets = get_secret("api-production-secrets")
DATABASE_URL = secrets["database_url"]
JWT_SECRET = secrets["jwt_secret"]


# Using HashiCorp Vault
import hvac

def get_vault_secrets(path: str) -> dict:
    """Retrieve secrets from HashiCorp Vault"""
    client = hvac.Client(url='https://vault.example.com:8200')

    # Authenticate with Vault (use AppRole, Kubernetes, etc.)
    client.auth.approle.login(
        role_id=os.getenv('VAULT_ROLE_ID'),
        secret_id=os.getenv('VAULT_SECRET_ID')
    )

    # Read secrets
    response = client.secrets.kv.v2.read_secret_version(path=path)
    return response['data']['data']


# Load secrets from Vault
vault_secrets = get_vault_secrets('api/production')


# Rotating secrets automatically
from datetime import datetime, timedelta

class SecretCache:
    """Cache secrets with automatic refresh"""
    def __init__(self, secret_name: str, ttl_minutes: int = 60):
        self.secret_name = secret_name
        self.ttl = timedelta(minutes=ttl_minutes)
        self._cached_secret = None
        self._last_refresh = None

    def get_secret(self) -> dict:
        """Get secret, refreshing if expired"""
        now = datetime.now()

        if (self._cached_secret is None or
            self._last_refresh is None or
            now - self._last_refresh > self.ttl):
            # Refresh secret
            self._cached_secret = get_secret(self.secret_name)
            self._last_refresh = now

        return self._cached_secret


# Usage
db_secrets = SecretCache("database-credentials", ttl_minutes=30)
jwt_secrets = SecretCache("jwt-signing-keys", ttl_minutes=60)

# Secrets automatically refresh every 30-60 minutes
DATABASE_URL = db_secrets.get_secret()["url"]

Secret Management Best Practices

Never commit secrets: Add .env, .env.*, secrets.json to .gitignore
Rotate secrets regularly: Change API keys, passwords, JWT secrets every 90 days minimum
Use different secrets per environment: Dev, staging, production should have unique secrets
Principle of least privilege: Grant minimum permissions needed (read-only database user for read endpoints)
Audit secret access: Log when secrets are accessed, who accessed them, from where
Encrypt secrets at rest: Use secret managers that encrypt stored secrets (AWS Secrets Manager, Vault)
Revoke on employee departure: Rotate all secrets when team members leave

Secure Configuration Checklist

Production Settings

Debug mode: OFF (DEBUG=False)
Verbose errors: OFF
HTTPS: Required
CORS: Specific origins
Rate limiting: Enabled
Logging: Structured, no secrets

Development Settings

Debug mode: ON (safe locally)
Verbose errors: ON
HTTPS: Optional
CORS: Localhost allowed
Rate limiting: Disabled/relaxed
Logging: Verbose

Security Monitoring, Logging, and Incident Response

Security is not "set and forget." Continuous monitoring, comprehensive logging, and a clear incident response plan are essential for detecting, responding to, and learning from security events.

Security-Focused Logging

Logs are your audit trail and your first line of defense for detecting attacks. Log security-relevant events, but never log sensitive data.

from fastapi import FastAPI, Request, HTTPException, Depends
import logging
import json
from datetime import datetime, timezone
import uuid

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format='%(message)s'  # We'll use JSON format
)
logger = logging.getLogger(__name__)

app = FastAPI()


def log_security_event(
    event_type: str,
    user_id: str = None,
    ip_address: str = None,
    details: dict = None,
    severity: str = "INFO"
):
    """Log security-relevant events in structured JSON format"""
    log_entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event_id": str(uuid.uuid4()),
        "event_type": event_type,
        "severity": severity,
        "user_id": user_id,
        "ip_address": ip_address,
        "details": details or {}
    }

    # Log at appropriate level
    if severity == "CRITICAL":
        logger.critical(json.dumps(log_entry))
    elif severity == "ERROR":
        logger.error(json.dumps(log_entry))
    elif severity == "WARNING":
        logger.warning(json.dumps(log_entry))
    else:
        logger.info(json.dumps(log_entry))


@app.middleware("http")
async def security_logging_middleware(request: Request, call_next):
    """Log all requests with security context"""
    start_time = datetime.now(timezone.utc)

    # Generate request ID for tracing
    request_id = str(uuid.uuid4())

    # Get client IP (handle proxy headers)
    client_ip = request.headers.get("X-Forwarded-For", request.client.host)

    try:
        response = await call_next(request)

        # Log suspicious patterns
        if response.status_code == 401:
            log_security_event(
                "authentication_failure",
                ip_address=client_ip,
                details={
                    "path": request.url.path,
                    "method": request.method,
                    "user_agent": request.headers.get("user-agent")
                },
                severity="WARNING"
            )

        elif response.status_code == 403:
            log_security_event(
                "authorization_failure",
                ip_address=client_ip,
                details={
                    "path": request.url.path,
                    "method": request.method
                },
                severity="WARNING"
            )

        return response

    except Exception as e:
        log_security_event(
            "request_error",
            ip_address=client_ip,
            details={
                "path": request.url.path,
                "method": request.method,
                "error": str(e)
            },
            severity="ERROR"
        )
        raise


@app.post("/api/login")
async def login(username: str, password: str, request: Request):
    """Login with security event logging"""
    client_ip = request.headers.get("X-Forwarded-For", request.client.host)

    # Verify credentials (simplified)
    user = verify_credentials(username, password)

    if not user:
        log_security_event(
            "login_failed",
            user_id=username,
            ip_address=client_ip,
            details={
                "reason": "invalid_credentials",
                "user_agent": request.headers.get("user-agent")
            },
            severity="WARNING"
        )
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Check for brute force (multiple failures from same IP)
    recent_failures = count_recent_failures(client_ip)
    if recent_failures > 5:
        log_security_event(
            "brute_force_detected",
            user_id=username,
            ip_address=client_ip,
            details={
                "failure_count": recent_failures,
                "time_window": "5_minutes"
            },
            severity="CRITICAL"
        )
        raise HTTPException(status_code=429, detail="Too many login attempts")

    # Successful login
    log_security_event(
        "login_success",
        user_id=str(user.id),
        ip_address=client_ip,
        details={
            "user_agent": request.headers.get("user-agent"),
            "new_session": True
        },
        severity="INFO"
    )

    return {"token": create_token(user)}


@app.delete("/api/users/{user_id}")
async def delete_user(
    user_id: int,
    current_user: User = Depends(get_current_user),
    request: Request
):
    """Delete user with audit logging"""
    log_security_event(
        "user_deletion",
        user_id=str(current_user.id),
        ip_address=request.client.host,
        details={
            "target_user_id": str(user_id),
            "performed_by": str(current_user.id)
        },
        severity="WARNING"
    )

    # Perform deletion...
    return {"message": "User deleted"}

What to Log (Security Events)

Authentication attempts (success/failure)
Authorization failures (403 errors)
Password changes, email changes
API key creation/rotation/deletion
Admin actions (user deletion, role changes)
Suspicious patterns (rapid requests, enumeration)

Input validation failures
Rate limit violations
Data export/download events
Configuration changes
Unusual IP addresses or geolocations
Error rates and exception traces

What NOT to Log (Sensitive Data)

Passwords: Never log passwords, even hashed. Log "password_changed" event, not the password itself.
API keys and tokens: Log "token issued" but not the token value. Log last 4 characters only if needed.
Personal data: Avoid logging PII (SSN, credit cards, health info). Use user IDs instead of names/emails.
Full request bodies: Don't log entire payloads that may contain sensitive fields.
Session data: Don't log session contents, cookies, or auth headers.

Security Monitoring and Alerting

Logging is useless without monitoring. Set up alerts for suspicious patterns and anomalies.

Critical Alerts (Immediate Response)

Brute force attacks (>10 failed logins/min)
Unauthorized admin access attempts
Mass data export events
Unusual spike in 403 errors
SQL injection patterns detected
API key leaked to public repository

Warning Alerts (Review Within Hours)

Elevated error rates (>5% of requests)
New IP addresses for admin accounts
Unusual traffic patterns
Failed CORS requests (enumeration?)
Deprecated API version usage spike
Slow query performance (potential DoS)

# Monitoring with Prometheus + Grafana
from prometheus_client import Counter, Histogram, start_http_server
from fastapi import FastAPI
import time

app = FastAPI()

# Define metrics
login_attempts = Counter(
    'api_login_attempts_total',
    'Total login attempts',
    ['status']  # Labels: success, failure
)

auth_failures = Counter(
    'api_auth_failures_total',
    'Authorization failures (403)',
    ['endpoint']
)

request_duration = Histogram(
    'api_request_duration_seconds',
    'Request duration',
    ['method', 'endpoint']
)


@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    """Collect metrics for monitoring"""
    start_time = time.time()

    response = await call_next(request)

    # Record request duration
    duration = time.time() - start_time
    request_duration.labels(
        method=request.method,
        endpoint=request.url.path
    ).observe(duration)

    # Count auth failures
    if response.status_code == 403:
        auth_failures.labels(endpoint=request.url.path).inc()

    return response


@app.post("/api/login")
async def login(username: str, password: str):
    user = verify_credentials(username, password)

    if user:
        login_attempts.labels(status='success').inc()
        return {"token": create_token(user)}
    else:
        login_attempts.labels(status='failure').inc()
        raise HTTPException(status_code=401)


# Start Prometheus metrics server
start_http_server(9090)  # Metrics available at http://localhost:9090/metrics


# Grafana Alert Rules (PromQL):
# - rate(api_login_attempts_total{status="failure"}[5m]) > 10
# - rate(api_auth_failures_total[5m]) > 5
# - histogram_quantile(0.95, api_request_duration_seconds) > 2.0

Incident Response Plan

When a security incident occurs, having a documented response plan ensures quick, coordinated action.

Incident Response Steps

Detection: Alert triggers from monitoring system. Security team notified immediately.
Assessment: Determine severity, scope, and affected systems. Is data compromised? Are attackers still active?
Containment: Stop the bleeding. Revoke compromised credentials, block attacker IPs, disable affected endpoints.
Eradication: Remove attacker access. Patch vulnerabilities, rotate secrets, update dependencies.
Recovery: Restore services, verify integrity, monitor for reinfection.
Post-incident review: Document what happened, why it happened, and how to prevent recurrence. Update security measures.

# Emergency response endpoints (admin-only)
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

@app.post("/api/admin/emergency/revoke-all-tokens")
async def revoke_all_tokens(admin: User = Depends(require_admin)):
    """Emergency: Invalidate all active sessions"""
    log_security_event(
        "mass_token_revocation",
        user_id=str(admin.id),
        details={"reason": "security_incident"},
        severity="CRITICAL"
    )

    # Clear all active tokens from Redis/database
    redis_client.flushdb()  # Or delete specific token keys

    return {"message": "All tokens revoked. Users must re-authenticate"}


@app.post("/api/admin/emergency/block-ip")
async def block_ip(ip_address: str, admin: User = Depends(require_admin)):
    """Emergency: Block an IP address"""
    log_security_event(
        "ip_blocked",
        user_id=str(admin.id),
        details={"blocked_ip": ip_address},
        severity="CRITICAL"
    )

    # Add to firewall rules or rate limit blocklist
    redis_client.sadd("blocked_ips", ip_address)

    return {"message": f"IP {ip_address} blocked"}


@app.post("/api/admin/emergency/read-only-mode")
async def enable_read_only_mode(admin: User = Depends(require_admin)):
    """Emergency: Disable all write operations"""
    log_security_event(
        "read_only_mode_enabled",
        user_id=str(admin.id),
        severity="CRITICAL"
    )

    # Set global flag
    redis_client.set("read_only_mode", "true")

    return {"message": "API in read-only mode. Write operations disabled"}


# Middleware to enforce read-only mode
@app.middleware("http")
async def read_only_middleware(request: Request, call_next):
    if redis_client.get("read_only_mode") == "true":
        if request.method in ["POST", "PUT", "PATCH", "DELETE"]:
            raise HTTPException(
                status_code=503,
                detail="API in emergency read-only mode"
            )
    return await call_next(request)

Security Monitoring Checklist

Centralized logging: Use ELK stack, CloudWatch, Datadog, or similar for aggregated logs
Real-time alerts: Configure PagerDuty, Opsgenie, or Slack notifications for critical events
Anomaly detection: Use ML-based tools to detect unusual patterns (auth0 Anomaly Detection, AWS GuardDuty)
Dependency scanning: Automate vulnerability scanning with Snyk, Dependabot, or safety
Penetration testing: Conduct regular security audits and pen tests
Bug bounty program: Incentivize ethical hackers to report vulnerabilities
Incident drills: Practice incident response quarterly to ensure readiness

Back to "Authentication & Authorization"

Web APIsLesson 6

Next: "Error Handling & Validation"