Module 5: Common Error Patterns

Exercise: Identify Error Patterns

Exercise: Identify Error Patterns

Capsule overview

This capsule is your practice field. In capsules 02, 03, and 04 you learned to recognize three categories of error patterns: incorrect naming and abstractions, unhandled edge cases, and security holes. Now you're going to apply it all together in a realistic scenario.

Below you'll find a complete FastAPI application — a notes management API with users. The code partially works: if you run it, some endpoints respond correctly. But it has 5 embedded error patterns that represent the three categories you studied. Your job is to:

  1. Identify each error pattern
  2. Explain why it's a problem (it's not enough to point it out — justify)
  3. Provide the fix

The errors aren't obvious. This code could pass a superficial code review. The errors are in code that "looks good" — exactly like the code AI generates in practice.


Instructions

How to approach the exercise

Step 1: Read the complete code once without looking for errors
        → Understand what the application does

Step 2: Reread applying the module's patterns
        → For each function, ask yourself:
          - Does the name reflect what it actually does?
          - What happens with empty/null/extreme inputs?
          - Is there any security vulnerability?

Step 3: Document each error you find
        → For each one: location, category, impact, fix

Step 4: Compare with the solutions
        → Did you find all 5? Are your fixes correct?

Success criteria

5 of 5 errors found → Excellent. Solid pattern recognition.
4 of 5 errors found → Very good. Review which one slipped and why.
3 of 5 errors found → Good. Reread capsules 02-04.
2 or fewer           → You need more practice. Review the patterns.

The format of your answers

For each error you identify, use this format:

Error #N:
- Location: [line or function]
- Category: [naming | edge case | security]
- Description: [what's wrong]
- Impact: [what can happen]
- Fix: [corrected code]

The Application: NotesAPI

This is a FastAPI application for managing personal notes with users. Read the complete code before looking for errors.

"""
NotesAPI — a personal notes management API.
Features: user registration, login, notes CRUD, search.
"""

from fastapi import FastAPI, HTTPException, Query, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field
from datetime import datetime, timedelta
from jose import jwt, JWTError
import sqlite3
import hashlib

app = FastAPI(title="NotesAPI", version="1.0.0")
security = HTTPBearer()

# --- Configuration ---
JWT_SECRET = "notes-api-secret-key-2024-production"
JWT_ALGORITHM = "HS256"
DB_PATH = "notes.db"


# --- Models ---
class UserRegister(BaseModel):
    username: str = Field(min_length=3, max_length=50)
    password: str = Field(min_length=6)
    email: str


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


class NoteCreate(BaseModel):
    title: str = Field(max_length=200)
    content: str
    tags: list[str] = []


class NoteUpdate(BaseModel):
    title: str | None = None
    content: str | None = None
    tags: list[str] | None = None


# --- Database ---
def get_db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db() -> None:
    conn = get_db()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT UNIQUE NOT NULL,
            password_hash TEXT NOT NULL,
            email TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
        CREATE TABLE IF NOT EXISTS notes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            title TEXT NOT NULL,
            content TEXT NOT NULL,
            tags TEXT DEFAULT '',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (user_id) REFERENCES users(id)
        );
    """)
    conn.commit()
    conn.close()


init_db()


# --- Utilities ---
def hash_password(password: str) -> str:
    return hashlib.md5(password.encode()).hexdigest()


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


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

    conn = get_db()
    user = conn.execute(
        "SELECT id, username, email FROM users WHERE id = ?", (user_id,)
    ).fetchone()
    conn.close()

    if user is None:
        raise HTTPException(status_code=401, detail="User not found")

    return dict(user)


# --- Authentication endpoints ---
@app.post("/auth/register")
async def register(user: UserRegister) -> dict:
    conn = get_db()

    existing = conn.execute(
        "SELECT id FROM users WHERE username = ?", (user.username,)
    ).fetchone()

    if existing:
        conn.close()
        raise HTTPException(status_code=409, detail="Username already exists")

    password_hash = hash_password(user.password)
    cursor = conn.execute(
        "INSERT INTO users (username, password_hash, email) VALUES (?, ?, ?)",
        (user.username, password_hash, user.email),
    )
    conn.commit()
    user_id = cursor.lastrowid
    conn.close()

    token = create_token(user_id, user.username)
    return {"user_id": user_id, "token": token}


@app.post("/auth/login")
async def login(credentials: UserLogin) -> dict:
    conn = get_db()
    user = conn.execute(
        "SELECT id, username, password_hash FROM users WHERE username = ?",
        (credentials.username,),
    ).fetchone()
    conn.close()

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

    if user["password_hash"] != hash_password(credentials.password):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    token = create_token(user["id"], user["username"])
    return {"user_id": user["id"], "token": token}


# --- Notes endpoints ---
@app.post("/notes")
async def create_note(
    note: NoteCreate,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    tags_str = ",".join(note.tags)
    cursor = conn.execute(
        "INSERT INTO notes (user_id, title, content, tags) VALUES (?, ?, ?, ?)",
        (current_user["id"], note.title, note.content, tags_str),
    )
    conn.commit()
    note_id = cursor.lastrowid
    conn.close()

    return {
        "id": note_id,
        "title": note.title,
        "content": note.content,
        "tags": note.tags,
        "created_at": datetime.now().isoformat(),
    }


@app.get("/notes")
async def get_user_notes(
    current_user: dict = Depends(get_current_user),
    page: int = Query(default=1),
    size: int = Query(default=10),
) -> dict:
    conn = get_db()
    total = conn.execute(
        "SELECT COUNT(*) as count FROM notes WHERE user_id = ?",
        (current_user["id"],),
    ).fetchone()["count"]

    offset = (page - 1) * size
    total_pages = total // size
    notes = conn.execute(
        "SELECT id, title, content, tags, created_at, updated_at FROM notes WHERE user_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?",
        (current_user["id"], size, offset),
    ).fetchall()
    conn.close()

    return {
        "notes": [dict(n) for n in notes],
        "page": page,
        "total_pages": total_pages,
        "total_items": total,
    }


@app.get("/notes/{note_id}")
async def get_note(
    note_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    note = conn.execute(
        "SELECT id, user_id, title, content, tags, created_at, updated_at FROM notes WHERE id = ?",
        (note_id,),
    ).fetchone()
    conn.close()

    if not note:
        raise HTTPException(status_code=404, detail="Note not found")

    return dict(note)


@app.put("/notes/{note_id}")
async def update_note(
    note_id: int,
    update: NoteUpdate,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    note = conn.execute(
        "SELECT * FROM notes WHERE id = ?", (note_id,)
    ).fetchone()

    if not note:
        conn.close()
        raise HTTPException(status_code=404, detail="Note not found")

    updates = {}
    if update.title is not None:
        updates["title"] = update.title
    if update.content is not None:
        updates["content"] = update.content
    if update.tags is not None:
        updates["tags"] = ",".join(update.tags)

    if updates:
        updates["updated_at"] = datetime.now().isoformat()
        set_clause = ", ".join(f"{k} = ?" for k in updates)
        values = list(updates.values()) + [note_id]
        conn.execute(f"UPDATE notes SET {set_clause} WHERE id = ?", values)
        conn.commit()

    updated_note = conn.execute(
        "SELECT * FROM notes WHERE id = ?", (note_id,)
    ).fetchone()
    conn.close()

    return dict(updated_note)


@app.delete("/notes/{note_id}")
async def delete_note(
    note_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    note = conn.execute(
        "SELECT * FROM notes WHERE id = ?", (note_id,)
    ).fetchone()

    if not note:
        conn.close()
        raise HTTPException(status_code=404, detail="Note not found")

    conn.execute("DELETE FROM notes WHERE id = ?", (note_id,))
    conn.commit()
    conn.close()

    return {"deleted": note_id, "message": "Note deleted successfully"}


@app.get("/notes/search")
async def search_notes(
    q: str = Query(..., min_length=1),
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    query = f"SELECT id, title, content, tags FROM notes WHERE user_id = {current_user['id']} AND (title LIKE '%{q}%' OR content LIKE '%{q}%')"
    results = conn.execute(query).fetchall()
    conn.close()

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

Your Turn: Find the 5 Errors

Before looking at the solutions, try to find the 5 errors. Use the format indicated above to document each one.

Hints by category (if you need them):

Categories of the 5 errors:
├── 1 naming/abstractions error
├── 2 edge case errors
└── 2 security errors

Additional Hints (Only If You're Stuck)

If after 20-30 minutes you haven't found them all, these hints point you in the right direction without giving the answer:

Hint for Error #1 (Security)

Look in the configuration and utilities section. How are passwords stored? Investigate whether the method used is appropriate for password hashing in 2024.

Hint for Error #2 (Security)

Look at the search endpoint. How is the SQL query built? Compare it with how the queries are built in the other endpoints.

Hint for Error #3 (Edge Case)

Look at the pagination in get_user_notes. What arithmetic operator is used to calculate total_pages? What happens with the last group of items if it doesn't fill a complete page?

Hint for Error #4 (Naming/Abstraction)

Look at the get_note, update_note, and delete_note endpoints. They all look up the note by note_id. Do they verify that the note belongs to the current user? What does this imply about the name get_current_user — does it give you a false sense of security?

Hint for Error #5 (Edge Case/Security)

Look at the constants at the start of the file. Is there something that should be in environment variables? Is there something in the configuration that violates the security practices you studied in capsule 04?


Detailed Solutions

Error #1: MD5 for Password Hashing (Security)

Location: The hash_password function

def hash_password(password: str) -> str:
    return hashlib.md5(password.encode()).hexdigest()

Category: Security

Why it looks good at first glance:

  • The function has a descriptive name
  • It uses hashlib, a standard library
  • The password is hashed before being stored (not plaintext)
  • The API is simple and clean

The problem: MD5 is a general hashing algorithm, not designed for passwords. Its weaknesses:

  1. Speed: MD5 is extremely fast — an attacker can try billions of combinations per second with GPUs
  2. No salt: Two users with the same password have the same hash. Precomputed rainbow tables crack MD5 in seconds
  3. Known collisions: MD5 has demonstrated collisions — different inputs can produce the same hash
  4. Deprecated: The industry abandoned MD5 for passwords over a decade ago
# Demonstration of the problem:
import hashlib
# Same password → same hash (no salt)
hashlib.md5("password123".encode()).hexdigest()
# → '482c811da5d5b4bc6d497ffa98491e38'
# This hash is in EVERY rainbow table in the world
# An attacker cracks it in < 1 second

Impact: If the database is leaked (breach), all passwords are cracked in minutes.

See fix
from passlib.context import CryptContext

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


def hash_password(password: str) -> str:
    """Hashes a password with bcrypt (automatic salt, configurable cost factor)."""
    return pwd_context.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verifies a password against a bcrypt hash."""
    return pwd_context.verify(plain_password, hashed_password)

And in the login endpoint, change:

# Before:
if user["password_hash"] != hash_password(credentials.password):

# After:
if not verify_password(credentials.password, user["password_hash"]):

Why bcrypt is correct:

  • It includes automatic salt — the same password produces different hashes
  • Configurable cost factor — you can make the hashing intentionally slower
  • Designed specifically for passwords — resistant to GPU attacks
  • An industry standard with decades of cryptographic analysis

Error #2: SQL Injection in Search (Security)

Location: The search_notes endpoint

@app.get("/notes/search")
async def search_notes(
    q: str = Query(..., min_length=1),
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    query = f"SELECT id, title, content, tags FROM notes WHERE user_id = {current_user['id']} AND (title LIKE '%{q}%' OR content LIKE '%{q}%')"
    results = conn.execute(query).fetchall()
    conn.close()
    ...

Category: Security

Why it looks good at first glance:

  • The endpoint requires authentication (Depends(get_current_user))
  • It filters by the current user's user_id
  • The query param has min_length=1 validation
  • The other endpoints in the same file use parameterized queries correctly

The problem: This is the only endpoint that uses an f-string to build the SQL query. It's especially dangerous because it's mixed in with endpoints that DO use parameters — it goes unnoticed in a review if you don't read each query individually.

# Attack: extract other users' data
# GET /notes/search?q=' UNION SELECT id, username, password_hash, email FROM users --

# Resulting query:
# SELECT id, title, content, tags FROM notes
# WHERE user_id = 1
# AND (title LIKE '%' UNION SELECT id, username, password_hash, email FROM users --%'
# OR content LIKE '%' UNION SELECT id, username, password_hash, email FROM users --%')

# → Returns the usernames and password hashes of ALL users

Impact: An authenticated user can extract data from the entire database — including other users' passwords.

See fix
@app.get("/notes/search")
async def search_notes(
    q: str = Query(..., min_length=1, max_length=100),
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    search_term = f"%{q}%"
    results = conn.execute(
        "SELECT id, title, content, tags FROM notes WHERE user_id = ? AND (title LIKE ? OR content LIKE ?)",
        (current_user["id"], search_term, search_term),
    ).fetchall()
    conn.close()

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

The three values (user_id, and the two search_term) go as ? parameters. max_length=100 was added to prevent abusively long searches.


Error #3: Off-by-One in Pagination (Edge Case)

Location: The get_user_notes endpoint

total_pages = total // size

Category: Edge case

Why it looks good at first glance:

  • The pagination has page and size with reasonable defaults
  • The offset is calculated correctly: (page - 1) * size
  • The total is obtained from the database with COUNT(*)
  • The response includes pagination metadata

The problem: Integer division (//) truncates. If you have 25 notes with size=10:

total_pages = 25 // 10  # = 2 (should be 3)
# Page 3 has 5 notes, but total_pages says only 2 pages exist

Also, there's no validation of page or size:

# page=0 → offset = -10 → SQLite returns unexpected results
# page=-5 → offset = -60 → absurd results
# size=0 → ZeroDivisionError in total // 0
# size=-1 → LIMIT -1 in SQLite returns ALL records
# size=1000000 → dump of all the data in one request

Impact: Users lose access to the last notes (those on the partial page). Invalid values cause crashes or incorrect data.

See fix
import math

@app.get("/notes")
async def get_user_notes(
    current_user: dict = Depends(get_current_user),
    page: int = Query(default=1, ge=1),
    size: int = Query(default=10, ge=1, le=100),
) -> dict:
    conn = get_db()
    total = conn.execute(
        "SELECT COUNT(*) as count FROM notes WHERE user_id = ?",
        (current_user["id"],),
    ).fetchone()["count"]

    total_pages = math.ceil(total / size) if total > 0 else 0

    if page > total_pages and total_pages > 0:
        conn.close()
        raise HTTPException(
            status_code=404,
            detail=f"Page {page} not found. Total pages: {total_pages}",
        )

    offset = (page - 1) * size
    notes = conn.execute(
        "SELECT id, title, content, tags, created_at, updated_at FROM notes WHERE user_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?",
        (current_user["id"], size, offset),
    ).fetchall()
    conn.close()

    return {
        "notes": [dict(n) for n in notes],
        "page": page,
        "page_size": size,
        "total_pages": total_pages,
        "total_items": total,
        "has_next": page < total_pages,
        "has_previous": page > 1,
    }

Changes:

  • math.ceil() instead of // to calculate total_pages
  • ge=1 on page and size — FastAPI rejects values ≤ 0 automatically
  • le=100 on size — prevents massive data dumps
  • Validation of out-of-range pages
  • has_next and has_previous to make frontend navigation easier

Error #4: IDOR — Access to Other Users' Notes (Naming/Authorization)

Location: The get_note, update_note, and delete_note endpoints

@app.get("/notes/{note_id}")
async def get_note(
    note_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    note = conn.execute(
        "SELECT id, user_id, title, content, tags, created_at, updated_at FROM notes WHERE id = ?",
        (note_id,),
    ).fetchone()
    conn.close()

    if not note:
        raise HTTPException(status_code=404, detail="Note not found")

    return dict(note)

Category: Naming/Abstraction (IDOR — Insecure Direct Object Reference)

Why it looks good at first glance:

  • The endpoint requires authentication (Depends(get_current_user))
  • The presence of current_user gives the impression that there's authorization
  • The query uses parameterized queries (no SQL injection)
  • The code verifies that the note exists

The problem: The endpoint verifies that the user is authenticated, but it doesn't verify that the note belongs to them. current_user is obtained but never used to filter. This is a classic case where the naming misleads: having current_user as a parameter creates the illusion that there's access control, when in reality any authenticated user can read, modify, or delete any other user's notes.

# Attack (as user alice, id=1):
# GET /notes/5        → Read bob's note
# PUT /notes/5        → Modify bob's note
# DELETE /notes/5     → Delete bob's note

# You only need to be authenticated — it doesn't matter whose note it is

This applies to all three endpoints: get_note, update_note, and delete_note. All three look up the note by note_id only, without filtering by user_id.

Impact: Any authenticated user can read, modify, and delete all other users' notes. It's a total violation of privacy and data integrity.

See fix
async def get_user_note(note_id: int, user_id: int, conn: sqlite3.Connection) -> dict:
    """Gets a note, verifying that it belongs to the user."""
    note = conn.execute(
        "SELECT id, user_id, title, content, tags, created_at, updated_at FROM notes WHERE id = ? AND user_id = ?",
        (note_id, user_id),
    ).fetchone()
    if note is None:
        raise HTTPException(status_code=404, detail="Note not found")
    return dict(note)


@app.get("/notes/{note_id}")
async def get_note(
    note_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    try:
        return get_user_note(note_id, current_user["id"], conn)
    finally:
        conn.close()


@app.put("/notes/{note_id}")
async def update_note(
    note_id: int,
    update: NoteUpdate,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    try:
        note = get_user_note(note_id, current_user["id"], conn)

        updates = {}
        if update.title is not None:
            updates["title"] = update.title
        if update.content is not None:
            updates["content"] = update.content
        if update.tags is not None:
            updates["tags"] = ",".join(update.tags)

        if updates:
            updates["updated_at"] = datetime.now().isoformat()
            set_clause = ", ".join(f"{k} = ?" for k in updates)
            values = list(updates.values()) + [note_id, current_user["id"]]
            conn.execute(
                f"UPDATE notes SET {set_clause} WHERE id = ? AND user_id = ?",
                values,
            )
            conn.commit()

        return get_user_note(note_id, current_user["id"], conn)
    finally:
        conn.close()


@app.delete("/notes/{note_id}")
async def delete_note(
    note_id: int,
    current_user: dict = Depends(get_current_user),
) -> dict:
    conn = get_db()
    try:
        get_user_note(note_id, current_user["id"], conn)

        conn.execute(
            "DELETE FROM notes WHERE id = ? AND user_id = ?",
            (note_id, current_user["id"]),
        )
        conn.commit()
        return {"deleted": note_id}
    finally:
        conn.close()

Key changes:

  • The get_user_note helper filters by note_id AND user_id
  • Each endpoint uses the helper — it's impossible to access others' notes
  • The DELETE and UPDATE also filter by user_id in the query
  • try/finally guarantees the connection is closed

Error #5: Hardcoded JWT Secret (Security/Edge Case)

Location: The configuration constants

JWT_SECRET = "notes-api-secret-key-2024-production"
JWT_ALGORITHM = "HS256"
DB_PATH = "notes.db"

Category: Security

Why it looks good at first glance:

  • It's at the start of the file as a constant, following convention
  • The name JWT_SECRET is descriptive
  • The value looks like a legitimate secret (it's not "secret" or "1234")
  • It's separated from the business logic

The problem: The JWT secret is hardcoded in the source code. If this file reaches a git repository (public or private):

  1. Anyone with access to the repo can create valid JWT tokens — a total authentication bypass
  2. The secret is predictable — it contains the year and the app's name, an attacker could guess it
  3. It can't be rotated without changing the code — if you suspect the secret was compromised, you need a deploy
  4. It's the same in all environments — dev, staging, and production share the same secret
# An attacker with the secret can create tokens for any user:
from jose import jwt
fake_token = jwt.encode(
    {"sub": "1", "username": "admin", "exp": datetime.utcnow() + timedelta(hours=24)},
    "notes-api-secret-key-2024-production",
    algorithm="HS256",
)
# This token is valid — the attacker is now user 1

Impact: A complete authentication bypass. An attacker can impersonate any user.

See fix
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    jwt_secret: str
    jwt_algorithm: str = "HS256"
    database_url: str = "notes.db"

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


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

And update the functions that use the secret:

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


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    settings = get_settings()
    try:
        payload = jwt.decode(
            credentials.credentials,
            settings.jwt_secret,
            algorithms=[settings.jwt_algorithm],
        )
        ...

The .env file (never in git):

JWT_SECRET=a-secret-generated-with-openssl-rand-hex-32-here

And in .gitignore:

.env
.env.*
!.env.example

jwt_secret: str with no default — the application won't start if it's not configured. Fail-fast is better than running with an insecure secret.


Summary Table of the 5 Errors

#ErrorCategorySeverityLine/Function
1MD5 for password hashingSecurityCriticalhash_password()
2SQL injection in searchSecurityCriticalsearch_notes()
3Off-by-one in paginationEdge CaseMediumget_user_notes()
4IDOR — access to others' notesNaming/AuthCriticalget_note(), update_note(), delete_note()
5Hardcoded JWT secretSecurityCriticalConfiguration constants

Distribution by category

Security:              3 errors (#1, #2, #5)
Edge Case:             1 error (#3)
Naming/Abstraction:    1 error (#4)

Error #4 is interesting because it crosses categories: it's an authorization problem (security) that manifests as a naming/abstraction error — the presence of current_user creates the illusion that there's access control when there isn't.


Post-Exercise Reflection

What made each error hard to find?

Error #1 (MD5): It looks like legitimate hashing. MD5 generates a hash, 
the password isn't stored in plaintext. The problem is subtle
— it's the wrong algorithm, not the absence of hashing.

Error #2 (SQL injection): It's in ONE endpoint of ~10. The others 
use parameterized queries. It's easy to assume that if 9 are fine,
the 10th is too.

Error #3 (Pagination): // vs math.ceil() is a one-character difference. 
The calculation looks correct at a glance. It only shows up
when total_items isn't a multiple of size.

Error #4 (IDOR): The PRESENCE of current_user misleads. The brain
sees "there's authentication" and assumes "there's authorization." The dependency
is injected but its value is never used to filter.

Error #5 (Hardcoded JWT): It looks like a normal constant. Constants
at the start of the file are an accepted pattern in Python. The problem
is that this particular value shouldn't be a constant in code.

Patterns you should take to the capstone project

1. Verify EVERY SQL query — does it use parameters or f-strings?
2. Verify that authentication ≠ authorization — is the user_id used to filter?
3. Verify security algorithms — MD5, SHA1, or bcrypt?
4. Verify pagination — // or math.ceil()? page/size validation?
5. Verify secrets — hardcoded or in environment variables?

Connection to the Project

This exercise is your dress rehearsal for the capstone project in module 8. The differences:

This exercise (module 5):           Capstone project (module 8):
├── 1 file, ~150 lines              ├── 8-12 files, ~500-800 lines
├── 5 errors                        ├── 15-20 errors
├── 3 categories                    ├── 5 categories (+ hallucinations, logic)
├── Embedded errors                 ├── Embedded errors
└── Only find and fix               └── Find + fix + document + retrospective

If you found 4-5 errors in this exercise, you're ready for the project.


Troubleshooting

"I found additional errors that aren't in the list of 5"

Well done. The code has more minor issues than the 5 main ones — for example, the database connection doesn't use context managers, datetime.utcnow() is deprecated, or the register endpoint doesn't validate the email format. These are legitimate issues but of lower severity than the 5 main ones.

"I didn't find error #4 (IDOR). Is it really an error?"

Yes, and it's one of the most common in real applications. The confusion between authentication ("who are you?") and authorization ("do you have permission to do this?") is one of the most frequent vulnerabilities according to OWASP (Broken Access Control is #1).

"Should I fix all 5 errors or only identify them?"

Both. Identifying without fixing demonstrates that you recognize the pattern. Fixing demonstrates that you know the solution. In the capstone project in module 8, you'll need to do both.

"How do I practice more?"

Generate a FastAPI application with Claude Code and apply the 5-point checklist from the "Patterns you should take" section. Look for the same patterns. With practice, you'll detect them automatically.

"Are these errors real or invented?"

They're real. Each of these errors appears frequently in LLM-generated code. MD5 for passwords, SQL injection in a search endpoint, IDOR in CRUD — they're documented patterns in the AI code security literature.


Additional Exercises

Extra Exercise 1: Fix the entire application

Take the complete NotesAPI code and apply the 5 fixes. Verify that the application still works after each fix.

See validation criteria

Your fixed version must meet:

  • ✅ Passwords hashed with bcrypt (install passlib[bcrypt])
  • ✅ All SQL queries use parameterized queries
  • ✅ Pagination with math.ceil() and ge=1, le=100 validation
  • ✅ All notes endpoints filter by user_id
  • ✅ JWT secret loaded from an environment variable
  • ✅ The application starts and responds correctly

Extra Exercise 2: Add tests that verify the fixes

Write a test for each fix that verifies the error no longer exists.

See example tests
import pytest
from fastapi.testclient import TestClient

def test_user_cannot_access_other_users_notes(client: TestClient):
    """Verifies that IDOR is fixed."""
    # Create user 1 and a note
    r1 = client.post("/auth/register", json={"username": "alice", "password": "pass123", "email": "a@x.com"})
    token1 = r1.json()["token"]
    note = client.post(
        "/notes",
        json={"title": "Private Note", "content": "Secret"},
        headers={"Authorization": f"Bearer {token1}"},
    )
    note_id = note.json()["id"]

    # Create user 2
    r2 = client.post("/auth/register", json={"username": "bob", "password": "pass456", "email": "b@x.com"})
    token2 = r2.json()["token"]

    # User 2 tries to access user 1's note
    response = client.get(
        f"/notes/{note_id}",
        headers={"Authorization": f"Bearer {token2}"},
    )
    assert response.status_code == 404  # Should not find the note


def test_search_resists_sql_injection(client: TestClient):
    """Verifies that SQL injection is fixed."""
    r = client.post("/auth/register", json={"username": "test", "password": "pass123", "email": "t@x.com"})
    token = r.json()["token"]

    response = client.get(
        "/notes/search",
        params={"q": "' UNION SELECT id, username, password_hash, email FROM users --"},
        headers={"Authorization": f"Bearer {token}"},
    )
    assert response.status_code == 200
    # Must not return data from the users table
    for result in response.json()["results"]:
        assert "password_hash" not in result


def test_pagination_total_pages_correct(client: TestClient):
    """Verifies that total_pages uses ceil, not floor division."""
    r = client.post("/auth/register", json={"username": "pager", "password": "pass123", "email": "p@x.com"})
    token = r.json()["token"]
    headers = {"Authorization": f"Bearer {token}"}

    # Create 15 notes
    for i in range(15):
        client.post("/notes", json={"title": f"Note {i}", "content": "test"}, headers=headers)

    # With size=10, 15 notes should give 2 pages (not 1)
    response = client.get("/notes?size=10", headers=headers)
    assert response.json()["total_pages"] == 2

    # Page 2 should have 5 notes
    response = client.get("/notes?page=2&size=10", headers=headers)
    assert len(response.json()["notes"]) == 5

Extra Exercise 3: Look for errors in your own code

Generate a FastAPI application with Claude Code (any prompt you want) and apply the 5-point checklist. Document the errors you find.

See checklist for your code
5-point checklist for AI-generated code:

1. [ ] Do all SQL queries use parameterized queries?
       Look for: f"SELECT, f"INSERT, f"UPDATE, f"DELETE
       Fix: Replace with ? or %s

2. [ ] Do the endpoints verify ownership (not just authentication)?
       Look for: endpoints with Depends(get_current_user) that don't filter by user_id
       Fix: Add AND user_id = ? to the queries

3. [ ] Are the secrets in environment variables?
       Look for: strings that look like keys, passwords, or tokens in the code
       Fix: Move to pydantic-settings with .env

4. [ ] Does the pagination use math.ceil() with validation?
       Look for: // to calculate total_pages, page without ge=1
       Fix: math.ceil() + Query(ge=1, le=100)

5. [ ] Do the passwords use bcrypt (not MD5/SHA)?
       Look for: hashlib.md5, hashlib.sha1, hashlib.sha256 for passwords
       Fix: passlib with bcrypt

Summary

  • The exercise presents a realistic FastAPI application with 5 embedded errors from the 3 module categories
  • The errors are subtle: MD5 instead of bcrypt, SQL injection in a single endpoint of 10, off-by-one in pagination, IDOR hidden by the presence of current_user, and a hardcoded JWT secret
  • The difficulty is that the code works — the errors don't cause immediate crashes
  • Error #4 (IDOR) is the most instructive: it demonstrates that authentication ≠ authorization
  • The 5-point checklist is your portable tool for reviewing any AI-generated code
  • This exercise is your rehearsal for the capstone project in module 8

Additional resources

  1. OWASP Top 10 — 2021 - The standard reference for web vulnerabilities (IDOR, injection, broken access control)
  2. CWE-639: Authorization Bypass Through User-Controlled Key - The formal classification of IDOR
  3. passlib Documentation - A Python library for secure password hashing
  4. SQLite Parameterized Queries - Official Python documentation on parameterized queries
  5. FastAPI Security Best Practices - The official FastAPI security guide

Next module: Debugging with Claude Code — how to diagnose and resolve errors using Claude Code as a debugging tool.


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