Learning Hub

How to Stop Mass Assignment and Excessive Data Exposure in a FastAPI App

Build a small FastAPI profile API, attack it with eight property-level requests, see why hiding one field and reusing one model fall short, then fix it with per-audience schemas, strict input models,...

October 14, 2026 38 Min Read
6

What you will build, and why property-level checks matter

By the end of this tutorial you will have a small FastAPI profile API in which every caller sees only the properties they are entitled to see and can change only the properties they are entitled to change. You will also have a probe that tries to write every known property and reports which ones stick, and a pytest suite that fails when a property leaks or when a new one becomes writable. You will get there the way these bugs appear in real projects: build the first draft, attack it as an ordinary customer, try the two fixes that are easiest to reach for first, watch both fall short, and then fix it properly.

Table Of Content

The flaw has two halves, and OWASP files them under one name. Entry API3:2023 of its API Security Top 10 is Broken Object Property Level Authorization. It says an API endpoint is vulnerable when it “exposes properties of an object that are considered sensitive and should not be read by the user”, and also when it “allows a user to change, add/or delete the value of a sensitive object’s property which the user should not be able to access”. The 2023 edition merged two older entries into this one: Excessive Data Exposure for the first half and Mass Assignment for the second. Our previous tutorial on broken object level authorization asked whose record a caller may touch. This one asks the next question: once a caller may touch a record, which of its fields may they read, and which may they change?

It is worth an hour because the consequences are concrete. In March 2012 a GitHub user exploited a security vulnerability in the public key update form to add their own public key to the Rails organization. GitHub’s own write-up, by Tom Preston-Werner, says: “The root cause of the vulnerability was a failure to properly check incoming form parameters, a problem known as the mass-assignment vulnerability.” OWASP lists the impact as data disclosure, data loss or data corruption, and adds: “Under certain circumstances, unauthorized access to object properties can lead to privilege escalation or partial/full account takeover.”

The flaw also hides well in code review. The shortest path to a working PATCH endpoint is to hand the request body straight to an update helper, and the shortest path to a working GET is to return the database row. Both shortcuts appear in the first draft you will build in Step 2, and both are easy to wave through, whether a person or an assistant wrote them. Our guide to reviewing AI-generated Python code covers the wider habit; this tutorial gives you the specific checks for this one flaw.

Key terms in plain language

Property. One named field of an object: a key in a JSON document or a column in a database row. A user in this tutorial has eleven properties, from id to preferences. The last one holds a small JSON object, so the API really deals with thirteen property paths, such as preferences.flags.billing_exempt.

Audience. The relationship between the caller and the record: the owner looking at their own profile, another member looking at it, a support agent, or an admin. One record has a different set of readable properties for each audience.

Excessive data exposure. A response that carries properties the audience should not see. The server made the right decision about the object and then sent the whole object. MITRE describes the weakness behind it as CWE-200: “The product exposes sensitive information to an actor that is not explicitly authorized to have access to that information.”

Mass assignment. The server copies client-supplied properties into an object or a database row without deciding which ones the client may set. It is CWE-915, which describes input “that specifies multiple attributes, properties, or fields that are to be initialized or updated in an object, but it does not properly control which attributes can be modified.”

Allowlist and blocklist. An allowlist names what is permitted and refuses everything else. A blocklist names what is forbidden and permits everything else. OWASP’s Mass Assignment Cheat Sheet puts the difference in one sentence: “Do not rely on a block-list as the sole protection: a new or overlooked sensitive field remains bindable.”

Schema, or Pydantic model. A Python class that describes the shape of some data. FastAPI uses Pydantic models to validate what clients send and to filter what the API returns, so most of this tutorial is about deciding which models to write.

Prerequisites

  • Python 3.10 or newer. The code uses the str | None type syntax. Everything here ran on Python 3.13.14 on Windows 11; macOS and Linux work the same apart from the command that activates the virtual environment.
  • A terminal. Step 2 uses two terminal windows and curl, which is included with Windows 10 and 11 and with macOS, and is available on Linux.
  • Basic Python (functions, dictionaries, classes) and a feel for HTTP: GET, PATCH and POST, status codes, JSON. You do not need to know FastAPI or Pydantic; each piece is explained when it first appears.
  • About an hour. The code listings are long, but each one is short enough to read in a sitting, and every listing carries its file name in the first line. Save each one under that name in a single folder.

Step 1: Set up the project, the data and the rules

Create the environment

Make a folder, create a virtual environment inside it, and install five packages: FastAPI for the API, Pydantic for the schemas, uvicorn to serve it, requests to call it from scripts, and pytest for the tests. The versions below are the ones this tutorial was run with.

mkdir bopla-lab
cd bopla-lab
python -m venv .venv
.venv\Scripts\activate
python -m pip install fastapi==0.143.0 pydantic==2.14.1 uvicorn==0.54.0 requests==2.34.2 pytest==9.1.1

In PowerShell, activate with .venv\Scripts\Activate.ps1 instead. On macOS or Linux, use source .venv/bin/activate. Save the next listing as step00_check.py. It imports the data layer you write below, so run it with python step00_check.py once store.py exists. It prints the versions in use and the four users the lab starts with.

# step00_check.py - confirm the environment and the data layer work
import sys
from importlib.metadata import version

from store import Store

print("python", sys.version.split()[0])
for name in ("fastapi", "pydantic", "starlette", "uvicorn", "requests", "pytest"):
    print(name, version(name))
store = Store()
print("users:", ", ".join(f"{u['id']}={u['display_name']} ({u['role']})" for u in store.all()))
python 3.13.14
fastapi 0.143.0
pydantic 2.14.1
starlette 1.8.0
uvicorn 0.54.0
requests 2.34.2
pytest 9.1.1
users: 1=Alice Nguyen (member), 2=Bob Okafor (member), 3=Carol Diaz (support), 4=Dave Smith (admin)

Your Python and package versions can be newer. If a later step behaves differently from what this page shows, install the pinned versions above first.

Write the rules down first

Before any code, decide who may see and change what. The lab has four users: Alice and Bob are ordinary members, Carol is a support agent, and Dave is an admin. Here is what each audience may read. A property with no “yes” in any column, such as password_hash, may not be read by anyone through this API.

Property Another member The owner Support Admin
id, display_name, bio yes yes yes yes
email, role, email_verified, credit_limit_cents no yes yes yes
internal_notes no no yes yes
preferences.theme, preferences.newsletter no yes no yes
preferences.flags.billing_exempt no no no yes
password_hash, api_key no no no no

And here is what each caller may write, by endpoint. PATCH /me edits the caller’s own profile. PATCH /users/{id} is for staff and edits someone else’s. POST /signup creates an account and accepts an email, a display name and a password, nothing more.

Property Owner, PATCH /me Support, PATCH /users/{id} Admin, PATCH /users/{id}
email, display_name, bio, preferences.theme, preferences.newsletter yes no no
internal_notes no yes yes
role, email_verified, credit_limit_cents, preferences.flags.billing_exempt no no yes
id, password_hash, api_key no no no

Now write the same rules as data. Save this as policy.py. It is the specification that the tests and the probe check the API against, and it deliberately imports nothing from the application. Writing the rules in a file that a person who understands the business can review, apart from the endpoint code, is a habit worth keeping beyond this tutorial.

# policy.py - the specification the tests check, written down before any code
IDS = {"alice": 1, "bob": 2, "carol": 3, "dave": 4}
ROLES = {"alice": "member", "bob": "member", "carol": "support", "dave": "admin"}

ALL_PATHS = [
    "id", "email", "display_name", "bio", "role", "email_verified", "credit_limit_cents",
    "password_hash", "api_key", "internal_notes",
    "preferences.theme", "preferences.newsletter", "preferences.flags.billing_exempt",
]

# Which properties a caller may READ, by audience.
READ = {
    "public": {"id", "display_name", "bio"},
    "self": {"id", "email", "display_name", "bio", "role", "email_verified", "credit_limit_cents",
             "preferences.theme", "preferences.newsletter"},
    "support": {"id", "email", "display_name", "bio", "role", "email_verified", "credit_limit_cents",
                "internal_notes"},
}
READ["admin"] = READ["support"] | {"preferences.theme", "preferences.newsletter",
                                   "preferences.flags.billing_exempt"}

# Which properties a caller may WRITE, by the endpoint they use.
WRITE = {
    "self": {"email", "display_name", "bio", "preferences.theme", "preferences.newsletter"},
    "support": {"internal_notes"},
    "admin": {"role", "email_verified", "credit_limit_cents", "internal_notes",
              "preferences.flags.billing_exempt"},
}
# POST /signup accepts an email, a display name and a password, and nothing else.


def audience(viewer_role, viewer_id, row_id):
    if viewer_role == "admin":
        return "admin"
    if viewer_role == "support":
        return "support"
    return "self" if viewer_id == row_id else "public"

The data layer

Save the next two listings as store.py, the only application file that touches the database. Users live in one SQLite table with eleven columns. The first listing holds the schema and four seeded users with made-up hashes and keys.

# store.py
import json
import sqlite3

COLUMNS = [
    "id", "email", "display_name", "bio", "role", "email_verified", "credit_limit_cents",
    "password_hash", "api_key", "internal_notes", "preferences",
]
WRITABLE = set(COLUMNS) - {"id"}

SCHEMA = """
CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    email TEXT NOT NULL UNIQUE,
    display_name TEXT NOT NULL,
    bio TEXT NOT NULL DEFAULT '',
    role TEXT NOT NULL DEFAULT 'member',
    email_verified INTEGER NOT NULL DEFAULT 0,
    credit_limit_cents INTEGER NOT NULL DEFAULT 0,
    password_hash TEXT NOT NULL,
    api_key TEXT NOT NULL,
    internal_notes TEXT NOT NULL DEFAULT '',
    preferences TEXT NOT NULL DEFAULT '{"theme": "light", "newsletter": true, "flags": {"billing_exempt": false}}'
);
"""


def prefs(theme):
    return {"theme": theme, "newsletter": True, "flags": {"billing_exempt": False}}


# id, email, display_name, bio, role, email_verified, credit_limit_cents,
# password_hash, api_key, internal_notes, preferences
SEED = [
    (1, "[email protected]", "Alice Nguyen", "Backend engineer", "member", 0, 50000,
     "pbkdf2-sha256$1000$a1b2c3d4e5f60718$3f9a1c0d77e2b4a85c6d19e0f2a3b4c5d6e7f8091a2b3c4d5e6f708192a3b4c5",
     "key-alice-7d41c2", "", prefs("dark")),
    (2, "[email protected]", "Bob Okafor", "Product designer", "member", 0, 120000,
     "pbkdf2-sha256$1000$0f1e2d3c4b5a6978$8c7b6a59483726150413f2e1d0c9b8a7968574635241302f1e0d9c8b7a695847",
     "key-bob-31c9aa", "Two chargebacks in 2025. Watch for refund abuse.", prefs("light")),
    (3, "[email protected]", "Carol Diaz", "Support lead", "support", 1, 0,
     "pbkdf2-sha256$1000$aa11bb22cc33dd44$c1d2e3f4a5b60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f01",
     "key-carol-88ab10", "", prefs("light")),
    (4, "[email protected]", "Dave Smith", "Platform admin", "admin", 1, 0,
     "pbkdf2-sha256$1000$55ee66ff77008811$d4c3b2a190807f6e5d4c3b2a190807162a3b4c5d6e7f8091a2b3c4d5e6f70819",
     "key-dave-0b5e9d", "", prefs("dark")),
]

Two details matter later. preferences is stored as JSON text and mixes settings the owner edits (theme, newsletter) with an internal flag, billing_exempt, that only an admin should touch. A settings column that started with two harmless keys can pick up internal ones as a product grows; this one already has. Second, bob carries an internal note about chargebacks, which is exactly the kind of text that should not reach another member.

The second listing is the class that reads and writes the table. It offers two kinds of helper on purpose.

# store.py (continued)
class Store:
    """A tiny user database. Nothing in here knows who is asking."""

    def __init__(self, path=":memory:"):
        self.db = sqlite3.connect(path, check_same_thread=False)
        self.db.row_factory = sqlite3.Row
        self.reset()

    def reset(self):
        self.db.executescript("DROP TABLE IF EXISTS users;" + SCHEMA)
        rows = [(*row[:-1], json.dumps(row[-1])) for row in SEED]
        self.db.executemany("INSERT INTO users VALUES (?,?,?,?,?,?,?,?,?,?,?)", rows)
        self.db.commit()

    @staticmethod
    def _decode(row):
        if row is None:
            return None
        data = dict(row)
        data["preferences"] = json.loads(data["preferences"])
        return data

    @staticmethod
    def _bind(value):
        return json.dumps(value) if isinstance(value, (dict, list)) else value

    def get(self, user_id):
        return self._decode(self.db.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone())

    def by_email(self, email):
        return self._decode(self.db.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone())

    def all(self):
        return [self._decode(row) for row in self.db.execute("SELECT * FROM users ORDER BY id")]

    # First-draft helpers: they build SQL from whatever keys they are handed.
    def create_any(self, fields):
        names = ", ".join(fields)
        marks = ", ".join("?" for _ in fields)
        cursor = self.db.execute(
            f"INSERT INTO users ({names}) VALUES ({marks})", [self._bind(v) for v in fields.values()]
        )
        self.db.commit()
        return self.get(cursor.lastrowid)

    def update_any(self, user_id, fields):
        sets = ", ".join(f"{name} = ?" for name in fields)
        self.db.execute(
            f"UPDATE users SET {sets} WHERE id = ?", [*[self._bind(v) for v in fields.values()], user_id]
        )
        self.db.commit()
        return self.get(user_id)

    # Checked helpers: column names come from the WRITABLE allowlist, never from the client.
    def update(self, user_id, fields):
        unknown = set(fields) - WRITABLE
        if unknown:
            raise ValueError(f"unknown or read-only columns: {sorted(unknown)}")
        return self.update_any(user_id, fields)

    def create(self, email, display_name, password_hash, api_key):
        cursor = self.db.execute(
            "INSERT INTO users (email, display_name, password_hash, api_key) VALUES (?, ?, ?, ?)",
            (email, display_name, password_hash, api_key),
        )
        self.db.commit()
        return self.get(cursor.lastrowid)

create_any and update_any are the first-draft helpers: they build SQL from whatever keys they are handed. update and create are the checked versions. update refuses any key outside the WRITABLE list, and create inserts a fixed set of columns. reset() rebuilds the table, which lets every demo and test start from the same four users. Plain def endpoints run in worker threads, so the single lab connection is opened with check_same_thread=False; a real service would use one connection per request or a pool.

Authentication and a test harness

Save this as auth.py. It turns a bearer token into the caller’s database row and answers 401 if the token is missing or unknown. That is authentication only: it says who is calling. The rest of this tutorial is about authorization, the separate question of what that caller may see and do. The password hashing here is lab-grade so the demo has something realistic to protect; for real accounts use the approach in our tutorial on hashing passwords with bcrypt and Argon2.

# auth.py
import hashlib
import secrets

from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

TOKENS = {"tok-alice": 1, "tok-bob": 2, "tok-carol": 3, "tok-dave": 4}

bearer = HTTPBearer(auto_error=False)


def hash_password(password: str) -> str:
    # Lab-grade hashing, so the demo has something realistic to protect. Use Argon2id in production.
    salt = secrets.token_hex(8)
    digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 1000).hex()
    return f"pbkdf2-sha256$1000${salt}${digest}"


def new_api_key() -> str:
    return "key-" + secrets.token_hex(6)


def make_current_user(store):
    """Authentication only: turn the bearer token into the caller's database row."""

    def current_user(creds: HTTPAuthorizationCredentials | None = Depends(bearer)) -> dict:
        user_id = TOKENS.get(creds.credentials) if creds else None
        user = store.get(user_id) if user_id else None
        if user is None:
            raise HTTPException(
                status_code=401,
                detail="missing or invalid token",
                headers={"WWW-Authenticate": "Bearer"},
            )
        return user

    return current_user

Last, save harness.py. LiveServer runs a FastAPI app on localhost in a background thread. Api pairs one app version with its own database, so a demo can reset the data and look inside after a request. flatten and nest convert between a nested JSON object and the dotted property paths used in the tables above.

# harness.py - run one version of the API on localhost and talk to it as a named user
import importlib
import threading
import time

import requests
import uvicorn

from policy import IDS
from store import Store

TOKENS = {name: f"tok-{name}" for name in IDS}


class LiveServer:
    """Run a FastAPI app on 127.0.0.1 in a background thread."""

    def __init__(self, app, port):
        config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="critical")
        self.server = uvicorn.Server(config)
        self.thread = threading.Thread(target=self.server.run, daemon=True)
        self.url = f"http://127.0.0.1:{port}"

    def __enter__(self):
        self.thread.start()
        while not self.server.started:
            if not self.thread.is_alive():
                raise RuntimeError("the server did not start (is the port already in use?)")
            time.sleep(0.02)
        return self

    def __exit__(self, *exc):
        self.server.should_exit = True
        self.thread.join(timeout=5)


class Api:
    """One app version plus its database, so a demo can reset and inspect the data."""

    def __init__(self, module, port):
        if isinstance(module, str):
            module = importlib.import_module(module)
        self.store = Store()
        self._server = LiveServer(module.make_app(self.store), port)
        self.url = self._server.url

    def __enter__(self):
        self._server.__enter__()
        return self

    def __exit__(self, *exc):
        self._server.__exit__(*exc)

    def call(self, caller, method, path, json=None):
        headers = {"Authorization": f"Bearer {TOKENS[caller]}"} if caller else {}
        return requests.request(method, self.url + path, json=json, headers=headers, timeout=10)


def flatten(obj, prefix=""):
    # {"preferences": {"theme": "dark"}} becomes {"preferences.theme": "dark"}
    flat = {}
    for key, value in obj.items():
        path = prefix + key
        if isinstance(value, dict):
            flat.update(flatten(value, path + "."))
        else:
            flat[path] = value
    return flat


def nest(path, value):
    # ("preferences.theme", "dark") becomes {"preferences": {"theme": "dark"}}
    for key in reversed(path.split(".")):
        value = {key: value}
    return value

Step 2: Build the first draft and look at what it returns

This is the version that gets written first when the goal is a working endpoint. It works, it is short, and it trusts everything the client sends. Save it as app_v1.py.

# app_v1.py - first draft: it works, it is short, and it trusts everything the client sends
from fastapi import Depends, FastAPI, HTTPException

from auth import hash_password, make_current_user, new_api_key


def make_app(store):
    app = FastAPI(title="Profiles API v1")
    current_user = make_current_user(store)

    @app.post("/signup", status_code=201)
    def signup(body: dict):
        data = dict(body)
        data["password_hash"] = hash_password(data.pop("password"))
        data.setdefault("api_key", new_api_key())
        return store.create_any(data)

    @app.get("/me")
    def me(user: dict = Depends(current_user)):
        return user

    @app.get("/users")
    def list_users(user: dict = Depends(current_user)):
        return store.all()

    @app.get("/users/{user_id}")
    def get_user(user_id: int, user: dict = Depends(current_user)):
        row = store.get(user_id)
        if row is None:
            raise HTTPException(404, "no such user")
        return row

    @app.patch("/me")
    def update_me(body: dict, user: dict = Depends(current_user)):
        return store.update_any(user["id"], body)

    @app.patch("/users/{user_id}")
    def update_user(user_id: int, body: dict, user: dict = Depends(current_user)):
        if user["role"] not in ("support", "admin"):
            raise HTTPException(403, "staff only")
        return store.update_any(user_id, body)

    return app

Three habits in this file cause the attacks that follow. The read endpoints return the database row as a plain dictionary, so there is nothing for FastAPI to filter. The write endpoints take body: dict, which accepts any JSON object, and pass it to update_any. And the only authorization is the check that PATCH /users/{id} is for staff. Notice what the file gets right: a missing token gets a 401, and a member calling the staff endpoint gets a 403. Those are real checks. They are just not the ones this flaw slips through.

Save serve.py, which runs any version on a real port.

# serve.py - run one version of the API for curl: python serve.py v1 8000
import sys
from importlib import import_module

import uvicorn

from store import Store

version, port = sys.argv[1], int(sys.argv[2])
app = import_module(f"app_{version}").make_app(Store())
uvicorn.run(app, host="127.0.0.1", port=port, log_level="warning")

Open two terminals in the project folder. Start the server in the first one:

python serve.py v1 8000

In the second, ask for Bob’s profile while logged in as Alice. The pipe into python -m json.tool only pretty-prints the answer. In Windows PowerShell 5.1, type curl.exe instead of curl, because there curl is an alias for a different command.

curl -s -H "Authorization: Bearer tok-alice" http://127.0.0.1:8000/users/2 | python -m json.tool
{
    "id": 2,
    "email": "[email protected]",
    "display_name": "Bob Okafor",
    "bio": "Product designer",
    "role": "member",
    "email_verified": 0,
    "credit_limit_cents": 120000,
    "password_hash": "pbkdf2-sha256$1000$0f1e2d3c4b5a6978$8c7b6a59483726150413f2e1d0c9b8a7968574635241302f1e0d9c8b7a695847",
    "api_key": "key-bob-31c9aa",
    "internal_notes": "Two chargebacks in 2025. Watch for refund abuse.",
    "preferences": {
        "theme": "light",
        "newsletter": true,
        "flags": {
            "billing_exempt": false
        }
    }
}

Alice is a member, and the rules say a member may see Bob’s name and bio. Alice also received Bob’s email address, credit limit, password hash and API key, and the support team’s private note about chargebacks. The authorization check on the object did its job, since a logged-in member may view profiles, and then the server serialized the entire row. Stop the server with Ctrl+C before you continue.

Step 3: Attack the first draft as an ordinary user

Reading is only half of it. Save attack.py, a script that runs eight attacks against one version of the API, each as a caller who has no special tools, and then checks the database to see whether the attack worked. It runs the server in-process, so it can reset the data between attacks and inspect the result.

# attack.py - attack one version of the API the way an ordinary logged-in user could
import sys

from harness import Api

VERSION = sys.argv[1] if len(sys.argv) > 1 else "v1"
PORTS = {"v1": 8801, "v2": 8802, "v3": 8803}
SENSITIVE = ["email", "credit_limit_cents", "internal_notes", "api_key", "password_hash"]
MALLORY = {"email": "[email protected]", "display_name": "Mallory", "password": "lab-password"}


def check_profile(api, response):
    seen = [name for name in SENSITIVE if name in response.json()]
    return bool(seen), "sees " + (", ".join(seen) if seen else "no sensitive property")


def check_list(api, response):
    rows = response.json()
    leaked = sum(1 for row in rows if "api_key" in row)
    return leaked > 0, f"{leaked} of {len(rows)} rows include an api_key"


def check_role(user_id):
    def check(api, response):
        role = api.store.get(user_id)["role"]
        return role == "admin", f"role is {role!r}"
    return check


def check_money(api, response):
    user = api.store.get(1)
    changed = user["credit_limit_cents"] != 50000 or bool(user["email_verified"])
    return changed, f"credit limit {user['credit_limit_cents']}, email_verified {bool(user['email_verified'])}"


def check_flag(api, response):
    exempt = api.store.get(1)["preferences"].get("flags", {}).get("billing_exempt")
    return bool(exempt), f"billing_exempt is {exempt}"


def check_signup(api, response):
    new = api.store.by_email(MALLORY["email"])
    if new is None:
        return False, "no account created"
    return new["role"] == "admin", f"new account has role {new['role']!r}"


def check_key_injection(api, response):
    user = api.store.get(1)
    changed = user["role"] == "admin" or user["display_name"] == "pwned"
    return changed, f"display_name {user['display_name']!r}, role {user['role']!r}"


ATTACKS = [
    ("alice reads bob's profile", "alice", "GET", "/users/2", None, check_profile),
    ("alice lists every user", "alice", "GET", "/users", None, check_list),
    ("alice PATCHes her own role", "alice", "PATCH", "/me", {"role": "admin"}, check_role(1)),
    ("alice PATCHes her credit limit and verified flag", "alice", "PATCH", "/me",
     {"credit_limit_cents": 10_000_000, "email_verified": True}, check_money),
    ("alice sets a flag inside her preferences", "alice", "PATCH", "/me",
     {"preferences": {"theme": "dark", "newsletter": True, "flags": {"billing_exempt": True}}}, check_flag),
    ("carol (support) PATCHes her own role", "carol", "PATCH", "/users/3", {"role": "admin"}, check_role(3)),
    ("a stranger signs up with role admin", None, "POST", "/signup", {**MALLORY, "role": "admin"}, check_signup),
    ("alice sends a column name as a key", "alice", "PATCH", "/me",
     {"display_name = 'pwned', role": "admin"}, check_key_injection),
]

with Api(f"app_{VERSION}", PORTS[VERSION]) as api:
    wins = 0
    for number, (label, caller, method, path, body, check) in enumerate(ATTACKS, 1):
        api.store.reset()
        response = api.call(caller, method, path, json=body)
        succeeded, detail = check(api, response)
        wins += succeeded
        verdict = "SUCCEEDED" if succeeded else "blocked"
        print(f"{number}. {label:<49} {verdict:<9} HTTP {response.status_code}  {detail}")
    print(f"{VERSION}: {wins} of {len(ATTACKS)} attacks succeeded")

Run it against the first draft with python attack.py v1.

1. alice reads bob's profile                         SUCCEEDED HTTP 200  sees email, credit_limit_cents, internal_notes, api_key, password_hash
2. alice lists every user                            SUCCEEDED HTTP 200  4 of 4 rows include an api_key
3. alice PATCHes her own role                        SUCCEEDED HTTP 200  role is 'admin'
4. alice PATCHes her credit limit and verified flag  SUCCEEDED HTTP 200  credit limit 10000000, email_verified True
5. alice sets a flag inside her preferences          SUCCEEDED HTTP 200  billing_exempt is True
6. carol (support) PATCHes her own role              SUCCEEDED HTTP 200  role is 'admin'
7. a stranger signs up with role admin               SUCCEEDED HTTP 201  new account has role 'admin'
8. alice sends a column name as a key                SUCCEEDED HTTP 200  display_name 'pwned', role 'admin'
v1: 8 of 8 attacks succeeded

All eight attacks succeeded. They fall into four groups.

  1. Attacks 1 and 2 are excessive data exposure. One request returned a colleague’s secrets, and the list endpoint returned all four users’ API keys at once.
  2. Attacks 3, 4 and 5 are mass assignment through PATCH /me. Alice became an admin, raised the credit limit on Alice’s own account, marked the email address as verified, and switched on an internal billing exemption hidden inside the preferences blob. No one guessed a password or found a bug in the login; the only change was an extra key in a JSON request. Attack 5 shows why nested JSON matters: replacing preferences replaced the internal flag along with the settings.
  3. Attack 6 slips past a real check. Carol passed the staff-only test, so Carol may call the endpoint, and nothing decided which properties a support agent may change. The result was a self-promotion to admin. OWASP tracks endpoint-level checks separately as API5:2023, Broken Function Level Authorization; this is the property-level gap that remains after such a check.
  4. Attacks 7 and 8 are the same weakness at the edges. A stranger with no account created an admin account by adding role to the signup body. The last attack sent a column name as a key: the update helper pastes keys into the SQL, so the key display_name = 'pwned', role became the fragment SET display_name = 'pwned', role = ?. SQL placeholders stand in for values, not for column names, so keys from the client must not be pasted into a statement. Our tutorial on preventing SQL injection with parameterized queries covers the wider rule.

Step 4: Try the fix that is easiest to reach for first

After a customer or a scanner reports that password_hash is in the response, one tempting reaction is to add Pydantic models and hide that one field. Save this as app_v2.py.

# app_v2.py - the reactive fix: Pydantic models, and one hidden property
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel

from auth import hash_password, make_current_user, new_api_key


class UserOut(BaseModel):
    # Mirrors the users table minus password_hash, the one property the bug report named.
    id: int
    email: str
    display_name: str
    bio: str = ""
    role: str = "member"
    email_verified: bool = False
    credit_limit_cents: int = 0
    api_key: str
    internal_notes: str = ""
    preferences: dict = {}


class SignupIn(BaseModel):
    email: str
    display_name: str
    password: str
    role: str = "member"
    credit_limit_cents: int = 0


class UserPatch(BaseModel):
    # Every column except id, password_hash and api_key, all optional so PATCH can change any of them.
    email: str | None = None
    display_name: str | None = None
    bio: str | None = None
    role: str | None = None
    email_verified: bool | None = None
    credit_limit_cents: int | None = None
    internal_notes: str | None = None
    preferences: dict | None = None


def make_app(store):
    app = FastAPI(title="Profiles API v2")
    current_user = make_current_user(store)

    @app.post("/signup", status_code=201, response_model=UserOut)
    def signup(body: SignupIn):
        fields = body.model_dump(exclude={"password"})
        fields["password_hash"] = hash_password(body.password)
        fields["api_key"] = new_api_key()
        return store.create_any(fields)

    @app.get("/me", response_model=UserOut)
    def me(user: dict = Depends(current_user)):
        return user

    @app.get("/users", response_model=list[UserOut])
    def list_users(user: dict = Depends(current_user)):
        return store.all()

    @app.get("/users/{user_id}", response_model=UserOut)
    def get_user(user_id: int, user: dict = Depends(current_user)):
        row = store.get(user_id)
        if row is None:
            raise HTTPException(404, "no such user")
        return row

    @app.patch("/me", response_model=UserOut)
    def update_me(body: UserPatch, user: dict = Depends(current_user)):
        fields = body.model_dump(exclude_unset=True)
        if fields:
            store.update(user["id"], fields)
        return store.get(user["id"])

    @app.patch("/users/{user_id}", response_model=UserOut)
    def update_user(user_id: int, body: UserPatch, user: dict = Depends(current_user)):
        if user["role"] not in ("support", "admin"):
            raise HTTPException(403, "staff only")
        if store.get(user_id) is None:
            raise HTTPException(404, "no such user")
        fields = body.model_dump(exclude_unset=True)
        if fields:
            store.update(user_id, fields)
        return store.get(user_id)

    return app

The changes are reasonable on their face. UserOut is the users table minus the hash, and every read route declares it as response_model. FastAPI’s documentation says this does what it sounds like: “It will limit and filter the output data to what is defined in the return type.” UserPatch lists every column except id, password_hash and api_key, all optional so that a PATCH can change any of them. SignupIn accepts role and credit_limit_cents so that someone, someday, can create staff accounts. Run python attack.py v2.

1. alice reads bob's profile                         SUCCEEDED HTTP 200  sees email, credit_limit_cents, internal_notes, api_key
2. alice lists every user                            SUCCEEDED HTTP 200  4 of 4 rows include an api_key
3. alice PATCHes her own role                        SUCCEEDED HTTP 200  role is 'admin'
4. alice PATCHes her credit limit and verified flag  SUCCEEDED HTTP 200  credit limit 10000000, email_verified True
5. alice sets a flag inside her preferences          SUCCEEDED HTTP 200  billing_exempt is True
6. carol (support) PATCHes her own role              SUCCEEDED HTTP 200  role is 'admin'
7. a stranger signs up with role admin               SUCCEEDED HTTP 201  new account has role 'admin'
8. alice sends a column name as a key                blocked   HTTP 200  display_name 'Alice Nguyen', role 'member'
v2: 7 of 8 attacks succeeded

Seven of eight attacks still work. What improved is small and real: attack 1 no longer finds the password hash, and attack 8 stopped working. The rest tells you how these fixes fail.

A response model is only as strict as the model you give it. FastAPI filtered exactly as documented, but UserOut still declares api_key and internal_notes, so they still go out. It also types preferences as a plain dict, which has no inner structure to filter, so the internal billing flag goes out with the rest of the settings.

An input model is only as narrow as the properties it declares. UserPatch declares role, email_verified and credit_limit_cents, so Alice may set them, and it declares preferences as a dict, so anything inside it may be set. The same goes for support staff: attack 6 still works.

Attack 8 was blocked by accident. Look at its line: HTTP 200, and nothing changed. Pydantic’s documentation explains why: “By default, Pydantic models won’t error when you provide extra data, and these values will simply be ignored.” The hostile key was not declared, so it was dropped, and the client was told everything went fine. That quiet success is a problem of its own, which Step 7 measures.

Step 5: Fix it properly with allowlist schemas

The real fix has four moves. Give each audience its own read model, written as an explicit list of properties. Give each endpoint and role its own strict write model. Make sure every response passes through one of those read models, either declared on the route or picked by one small function. And keep a second allowlist in the data layer. OWASP’s API3 guidance says the same things in different words: “cherry-pick specific object properties you specifically want to return”, “Allow changes only to the object’s properties that should be updated by the client”, and “Implement a schema-based response validation mechanism as an extra layer of security.” Save the next four listings together as app_v3.py.

Read models, one per audience

Each class below is a list that a person wrote on purpose. None is derived from the table, so adding a column changes nothing. Strict is a base class for what clients send, covered in the next section.

# app_v3.py - allowlist schemas per audience, strict input models, one place that picks the view
import sqlite3
from typing import Literal

from fastapi import Body, Depends, FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict, Field, ValidationError

from auth import hash_password, make_current_user, new_api_key


class Strict(BaseModel):
    # Base class for everything a client sends: an unknown property is an error, not a shrug.
    model_config = ConfigDict(extra="forbid")


# ---- What each audience may READ: explicit allowlists, never derived from the table.
class PublicView(BaseModel):
    id: int
    display_name: str
    bio: str


class PrefsView(BaseModel):
    theme: str
    newsletter: bool


class SelfView(BaseModel):
    id: int
    email: str
    display_name: str
    bio: str
    role: str
    email_verified: bool
    credit_limit_cents: int
    preferences: PrefsView


class StaffView(BaseModel):
    id: int
    email: str
    display_name: str
    bio: str
    role: str
    email_verified: bool
    credit_limit_cents: int
    internal_notes: str


class FlagsView(BaseModel):
    billing_exempt: bool


class AdminPrefsView(PrefsView):
    flags: FlagsView


class AdminView(StaffView):
    preferences: AdminPrefsView

PublicView lists three properties, SelfView nine property paths (preferences counts as two), StaffView eight, and AdminView eleven (preferences counts as three), matching the read table in Step 1. Nested JSON gets typed models too, PrefsView and AdminPrefsView, so the internal flag appears only in the admin’s version. The password hash and the API key appear in none of them.

Write models, one per endpoint and role

The write side repeats the idea. Strict sets extra="forbid". The Pydantic configuration reference describes that value this way: “Providing extra data is not permitted, and a ValidationError will be raised if this is the case.” Combined with models that declare only what the caller may change, a request that includes role or api_key is refused instead of ignored. Each nested model is strict too, so the attack that smuggled a flag inside preferences fails at the inner level.

# app_v3.py (continued)
# ---- What each caller may WRITE: one strict model per endpoint and role.
class SignupIn(Strict):
    email: str
    display_name: str
    password: str


class PrefsPatch(Strict):
    theme: str | None = None
    newsletter: bool | None = None


class SelfUpdate(Strict):
    email: str | None = None
    display_name: str | None = None
    bio: str | None = None
    preferences: PrefsPatch | None = None


class SupportUpdate(Strict):
    internal_notes: str | None = None


class FlagsPatch(Strict):
    billing_exempt: bool | None = None


class AdminPrefsPatch(Strict):
    flags: FlagsPatch | None = None


class AdminUpdate(Strict):
    role: Literal["member", "support", "admin"] | None = None
    email_verified: bool | None = None
    credit_limit_cents: int | None = Field(default=None, ge=0, le=1_000_000)
    internal_notes: str | None = None
    preferences: AdminPrefsPatch | None = None

Two details deserve a second look. role is a Literal of the three real roles, so a typo or an invented role is a validation error. And credit_limit_cents carries ge=0 and le=1_000_000, because a property an admin may write is still not a property anyone may set to any number.

One function that picks the view

These helpers do the choosing. audience decides how the caller relates to the row. render picks the read model for the routes whose answer depends on the caller, and lets Pydantic drop everything the model does not list. parse validates a request body against a write model and, on failure, answers 422 with the errors but without echoing the submitted values. deep_merge lets a PATCH change one nested key without wiping its siblings.

# app_v3.py (continued)
VIEWS = {"public": PublicView, "self": SelfView, "support": StaffView, "admin": AdminView}


def audience(viewer: dict, row: dict) -> str:
    if viewer["role"] == "admin":
        return "admin"
    if viewer["role"] == "support":
        return "support"
    return "self" if viewer["id"] == row["id"] else "public"


def render(viewer: dict, row: dict) -> dict:
    # The only place where a database row becomes a response.
    return VIEWS[audience(viewer, row)].model_validate(row).model_dump()


def parse(model: type[BaseModel], payload: dict) -> dict:
    try:
        return model.model_validate(payload).model_dump(exclude_none=True)
    except ValidationError as exc:
        detail = exc.errors(include_url=False, include_context=False, include_input=False)
        raise HTTPException(422, detail=detail)


def deep_merge(base: dict, patch: dict) -> dict:
    merged = dict(base)
    for key, value in patch.items():
        merged[key] = deep_merge(merged.get(key, {}), value) if isinstance(value, dict) else value
    return merged

The routes

The last listing wires it together.

# app_v3.py (continued)
def make_app(store):
    app = FastAPI(title="Profiles API v3")
    current_user = make_current_user(store)

    def staff_only(user: dict = Depends(current_user)) -> dict:
        if user["role"] not in ("support", "admin"):
            raise HTTPException(403, "staff only")
        return user

    def save(user_id: int, changes: dict) -> None:
        if "preferences" in changes:
            changes["preferences"] = deep_merge(store.get(user_id)["preferences"], changes["preferences"])
        try:
            store.update(user_id, changes)
        except sqlite3.IntegrityError:
            raise HTTPException(409, "email already registered")

    @app.post("/signup", status_code=201, response_model=SelfView)
    def signup(body: SignupIn):
        try:
            return store.create(body.email, body.display_name, hash_password(body.password), new_api_key())
        except sqlite3.IntegrityError:
            raise HTTPException(409, "email already registered")

    @app.get("/me", response_model=SelfView)
    def me(user: dict = Depends(current_user)):
        return user

    @app.get("/users", response_model=None)
    def list_users(viewer: dict = Depends(current_user)):
        return [render(viewer, row) for row in store.all()]

    @app.get("/users/{user_id}", response_model=None)
    def get_user(user_id: int, viewer: dict = Depends(current_user)):
        row = store.get(user_id)
        if row is None:
            raise HTTPException(404, "no such user")
        return render(viewer, row)

    @app.patch("/me", response_model=SelfView)
    def update_me(body: SelfUpdate, user: dict = Depends(current_user)):
        changes = body.model_dump(exclude_none=True)
        if changes.get("email", user["email"]) != user["email"]:
            changes["email_verified"] = False  # a new address has not been verified yet
        if changes:
            save(user["id"], changes)
        return store.get(user["id"])

    @app.patch("/users/{user_id}", response_model=None)
    def update_user(user_id: int, body: dict = Body(...), staff: dict = Depends(staff_only)):
        if store.get(user_id) is None:
            raise HTTPException(404, "no such user")
        model = AdminUpdate if staff["role"] == "admin" else SupportUpdate
        changes = parse(model, body)
        if changes:
            save(user_id, changes)
        return render(staff, store.get(user_id))

    return app

Here is how the routes apply the four moves. /me always shows the owner’s own profile, so it declares response_model=SelfView, the plain FastAPI way. /users and /users/{user_id} depend on who is asking, so they call render and declare response_model=None, which tells FastAPI not to filter a second time. PATCH /me takes SelfUpdate as its body, so FastAPI validates it, and model_dump(exclude_none=True) keeps only the properties the client sent with a value.

PATCH /users/{user_id} cannot declare one body model, because the right model depends on the caller’s role. It takes a plain dictionary and runs it through parse with AdminUpdate or SupportUpdate. Two small rules are worth copying. Changing the email address resets email_verified, because otherwise an owner could attach a verified flag to an address nobody has verified, which would make a property they may not write change as a side effect of one they may. And save merges nested preferences, then calls the data layer’s checked update, which is a second allowlist behind the schemas. A bug in a model should not be able to put a column name into SQL. OWASP’s cheat sheet adds a caution that applies here too: allowlists “restrict writable properties; they do not replace authorization for the requested object and operation.” The object-level checks from the previous tutorial still need to be there.

Run the attacks again

Run python attack.py v3.

1. alice reads bob's profile                         blocked   HTTP 200  sees no sensitive property
2. alice lists every user                            blocked   HTTP 200  0 of 4 rows include an api_key
3. alice PATCHes her own role                        blocked   HTTP 422  role is 'member'
4. alice PATCHes her credit limit and verified flag  blocked   HTTP 422  credit limit 50000, email_verified False
5. alice sets a flag inside her preferences          blocked   HTTP 422  billing_exempt is False
6. carol (support) PATCHes her own role              blocked   HTTP 422  role is 'support'
7. a stranger signs up with role admin               blocked   HTTP 422  no account created
8. alice sends a column name as a key                blocked   HTTP 422  display_name 'Alice Nguyen', role 'member'
v3: 0 of 8 attacks succeeded

None of the eight attacks worked. Attacks 1 and 2 got answers, as they should, but the answers held only the properties the audience may see, and the other six were refused with 422. Compare this with the 7 of 8 that survived the first fix.

Step 6: Change the product and see which version leaks

Passing the attacks once is not the same as staying safe. The next risk is quieter than a missed field: it is what happens when the product changes. Imagine the next feature request. The admin screen needs the last four digits of each customer’s social security number. A developer adds a column, adds one line to the model that mirrors the table, and ships. Save step_drift.py. It applies exactly that one-line edit to a copy of app_v2.py, adds the column to every version’s database, and asks each version who can now see it.

# step_drift.py - what a new column or a one-line model change does to each version
import difflib
import types
from pathlib import Path

from harness import Api

VIEWS = [("alice", 2, "alice views bob"), ("alice", 1, "alice views self"),
         ("carol", 2, "carol views bob"), ("dave", 2, "dave views bob")]
OLD = '    internal_notes: str = ""\n    preferences: dict = {}\n'
NEW = '    internal_notes: str = ""\n    ssn_last4: str = ""\n    preferences: dict = {}\n'

source = Path("app_v2.py").read_text(encoding="utf-8")
assert source.count(OLD) == 1
edited_source = source.replace(OLD, NEW)
print("\n".join(difflib.unified_diff(source.splitlines(), edited_source.splitlines(),
                                     "app_v2.py", "app_v2.py (edited)", n=1, lineterm="")))
print()

edited = types.ModuleType("app_v2_edited")
exec(compile(edited_source, "app_v2_edited.py", "exec"), edited.__dict__)

VARIANTS = [("app_v1", "app_v1", 8811), ("app_v2", "app_v2", 8812),
            ("app_v2 plus the one line", edited, 8813), ("app_v3", "app_v3", 8814)]
print(f"{'variant':<28}" + "".join(f"{label:<19}" for _, _, label in VIEWS))
for name, module, port in VARIANTS:
    with Api(module, port) as api:
        api.store.db.execute("ALTER TABLE users ADD COLUMN ssn_last4 TEXT NOT NULL DEFAULT '6789'")
        cells = []
        for caller, target, _ in VIEWS:
            body = api.call(caller, "GET", f"/users/{target}").json()
            cells.append("LEAKS" if "ssn_last4" in body else "hidden")
    print(f"{name:<28}" + "".join(f"{cell:<19}" for cell in cells))

Run it with python step_drift.py.

--- app_v2.py
+++ app_v2.py (edited)
@@ -18,2 +18,3 @@
     internal_notes: str = ""
+    ssn_last4: str = ""
     preferences: dict = {}
variant                     alice views bob    alice views self   carol views bob    dave views bob     
app_v1                      LEAKS              LEAKS              LEAKS              LEAKS              
app_v2                      hidden             hidden             hidden             hidden             
app_v2 plus the one line    LEAKS              LEAKS              LEAKS              LEAKS              
app_v3                      hidden             hidden             hidden             hidden

Read the table row by row; the last row is the fix from Step 5. The first draft leaks the new column the moment it exists in the database, with no code change at all, because it returns whole rows. app_v2 does not leak it, because its model does not declare the column. But the version with the one added line leaks it to another member, to the owner, to support and to the admin alike, because there is only one model and it serves every audience. The shape of this failure is the point. In a design that lists what to hide, forgetting something exposes it. In a design that lists what to show, forgetting something produces a missing field, which is a visible bug rather than a silent leak. FastAPI’s documentation calls reducing code duplication “one of the core ideas in FastAPI”, and that advice is sound for properties every audience may see; it becomes a hazard when one shared model also carries properties that only some audiences may see.

Step 7: Prove it with a property probe

Eight hand-picked attacks are a demonstration, not evidence. The honest question is which of all the properties can be written by each caller through each endpoint. OWASP’s description of how these flaws are found points at the method: “Whether they can be changed is a matter of crafting an API request and analyzing the response. Side-effect analysis may be required if the target property is not returned in the API response.” The probe automates that. For each caller and endpoint it sends one property at a time, with a canary value that differs from the current value, and then reads the database to see whether the canary was stored. It sends one property per request because a strict model rejects a whole body that contains one unknown key, so a body with every property would only tell you that something was wrong.

Write the probe

The property names come from policy.py here. In a real project, build the list from your ORM models, your migrations, your OpenAPI document and any response you can capture, because a probe can only test the names it is given. Save probe.py in three parts. The first holds the canary values and the function that probes PATCH endpoints.

# probe.py - try to write every known property, one at a time, and see which ones stick
import sys
import textwrap

from harness import Api, flatten, nest
from policy import ALL_PATHS, IDS, WRITE

# Canary values, in order of preference. pick() uses the first one that differs from the current value.
CANDIDATES = {
    "id": [99, 98],
    "email": ["[email protected]", "[email protected]"],
    "role": ["admin", "support"],
    "email_verified": [True, False],
    "credit_limit_cents": [777700, 777701],
    "newsletter": [False, True],
    "billing_exempt": [True, False],
}
NEW_ACCOUNT = {
    "role": "member", "email_verified": False, "credit_limit_cents": 0, "bio": "", "internal_notes": "",
    "preferences.theme": "light", "preferences.newsletter": True, "preferences.flags.billing_exempt": False,
}
SIGNUP_FIELDS = {"email", "display_name"}


def pick(path, current):
    leaf = path.split(".")[-1]
    for value in CANDIDATES.get(leaf, [f"canary-{leaf}", f"canary2-{leaf}"]):
        if value != current:
            return value


def probe_write(api, caller, path_template, target_id):
    """PATCH one property at a time as `caller`. Returns {path: (status, stuck)}."""
    outcome = {}
    for path in ALL_PATHS:
        api.store.reset()
        current = flatten(api.store.get(target_id)).get(path)
        value = pick(path, current)
        response = api.call(caller, "PATCH", path_template.format(target=target_id), json=nest(path, value))
        after = api.store.get(target_id)
        if path == "id":
            stuck = after is None and api.store.get(value) is not None
        else:
            stuck = after is not None and flatten(after).get(path) == value
        outcome[path] = (response.status_code, stuck)
    return outcome

The second probes signup, which has no logged-in user, and treats the new account as the object under test.

# probe.py (continued)
def probe_signup(api):
    """POST /signup with one extra property at a time. Returns {path: (status, stuck)}."""
    outcome = {}
    for path in ALL_PATHS:
        if path in SIGNUP_FIELDS:
            continue
        api.store.reset()
        value = pick(path, NEW_ACCOUNT.get(path))
        body = {"email": "[email protected]", "display_name": "Mallory", "password": "lab-password"}
        body.update(nest(path, value))
        response = api.call(None, "POST", "/signup", json=body)
        new = api.store.by_email("[email protected]")
        if new is None:
            stuck = False
        elif path == "id":
            stuck = new["id"] == value
        else:
            stuck = flatten(new).get(path) == value
        outcome[path] = (response.status_code, stuck)
    return outcome

The third turns the results into one row per caller and endpoint.

# probe.py (continued)
def summarize(outcome, allowed):
    landed = {path for path, (status, stuck) in outcome.items() if stuck}
    return {
        "tried": len(outcome),
        "allowed": len(allowed),
        "landed": len(landed),
        "unauthorized": sorted(landed - allowed),
        "missing": sorted(allowed - landed),
        "quiet": sum(1 for status, stuck in outcome.values() if status < 300 and not stuck),
        "rejected": sum(1 for status, stuck in outcome.values() if status >= 400),
    }


def main(version):
    ports = {"v1": 8805, "v2": 8806, "v3": 8807}
    rows = []
    with Api(f"app_{version}", ports[version]) as api:
        for caller in ("alice", "carol", "dave"):
            rows.append((caller, "PATCH /me", probe_write(api, caller, "/me", IDS[caller]), WRITE["self"]))
        for caller, allowed in (("alice", set()), ("carol", WRITE["support"]), ("dave", WRITE["admin"])):
            rows.append((caller, "PATCH /users/2", probe_write(api, caller, "/users/{target}", 2), allowed))
        rows.append(("anonymous", "POST /signup", probe_signup(api), set()))
    print(f"{'caller':<10}{'request':<16}{'tried':>6}{'allowed':>9}{'landed':>8}{'UNAUTH':>8}{'quiet':>7}{'rejected':>10}")
    total = 0
    for caller, request, outcome, allowed in rows:
        s = summarize(outcome, allowed)
        total += len(s["unauthorized"])
        print(f"{caller:<10}{request:<16}{s['tried']:>6}{s['allowed']:>9}{s['landed']:>8}"
              f"{len(s['unauthorized']):>8}{s['quiet']:>7}{s['rejected']:>10}")
        if s["unauthorized"]:
            print(textwrap.fill(", ".join(s["unauthorized"]), width=110, initial_indent="    unauthorized: ",
                                subsequent_indent="                  "))
        if s["missing"]:
            print("    allowed but did not stick: " + ", ".join(s["missing"]))
    print(f"{version}: {total} unauthorized property writes stuck")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "v1")

Run it on all three versions

The first draft

Run python probe.py v1. Each probe resets the database first, so the numbers do not depend on the order. A full run is 89 write requests.

caller    request          tried  allowed  landed  UNAUTH  quiet  rejected
alice     PATCH /me           13        5      13       8      0         0
    unauthorized: api_key, credit_limit_cents, email_verified, id, internal_notes, password_hash,
                  preferences.flags.billing_exempt, role
carol     PATCH /me           13        5      13       8      0         0
    unauthorized: api_key, credit_limit_cents, email_verified, id, internal_notes, password_hash,
                  preferences.flags.billing_exempt, role
dave      PATCH /me           13        5      13       8      0         0
    unauthorized: api_key, credit_limit_cents, email_verified, id, internal_notes, password_hash,
                  preferences.flags.billing_exempt, role
alice     PATCH /users/2      13        0       0       0      0        13
carol     PATCH /users/2      13        1      13      12      0         0
    unauthorized: api_key, bio, credit_limit_cents, display_name, email, email_verified, id, password_hash,
                  preferences.flags.billing_exempt, preferences.newsletter, preferences.theme, role
dave      PATCH /users/2      13        5      13       8      0         0
    unauthorized: api_key, bio, display_name, email, id, password_hash, preferences.newsletter,
                  preferences.theme
anonymous POST /signup        11        0      10      10      1         0
    unauthorized: api_key, bio, credit_limit_cents, email_verified, id, internal_notes,
                  preferences.flags.billing_exempt, preferences.newsletter, preferences.theme, role
v1: 54 unauthorized property writes stuck

Read the columns like this. tried is the number of properties attempted, and allowed is how many policy.py permits for that caller and endpoint. landed is how many were actually stored. UNAUTH is the difference, the number of writes that should have been refused, and it is the number to drive to zero. quiet counts requests answered with a success status where nothing was stored, and rejected counts requests answered with an error status.

The first draft lets 54 unauthorized property writes stick. Alice can write 13 of the 13 properties through PATCH /me where the policy allows 5, including id, password_hash and api_key. Carol, a support agent, can write 12 properties on someone else’s profile that the policy does not allow. Even the admin can overwrite another user’s password hash and API key, which the policy does not allow through this endpoint. Alice calling the staff endpoint is rejected 13 times, which confirms the function-level check works and also shows how little it covers. On signup, 10 of 11 properties stick. The single exception, password_hash, was protected by accident: the handler computes it after copying the body.

The first fix

Now the first fix, with python probe.py v2.

caller    request          tried  allowed  landed  UNAUTH  quiet  rejected
alice     PATCH /me           13        5      10       5      3         0
    unauthorized: credit_limit_cents, email_verified, internal_notes, preferences.flags.billing_exempt, role
carol     PATCH /me           13        5      10       5      3         0
    unauthorized: credit_limit_cents, email_verified, internal_notes, preferences.flags.billing_exempt, role
dave      PATCH /me           13        5      10       5      3         0
    unauthorized: credit_limit_cents, email_verified, internal_notes, preferences.flags.billing_exempt, role
alice     PATCH /users/2      13        0       0       0      0        13
carol     PATCH /users/2      13        1      10       9      3         0
    unauthorized: bio, credit_limit_cents, display_name, email, email_verified,
                  preferences.flags.billing_exempt, preferences.newsletter, preferences.theme, role
dave      PATCH /users/2      13        5      10       5      3         0
    unauthorized: bio, display_name, email, preferences.newsletter, preferences.theme
anonymous POST /signup        11        0       2       2      9         0
    unauthorized: credit_limit_cents, role
v2: 31 unauthorized property writes stuck

The count drops from 54 to 31, which is real progress and still a problem. Look at the quiet column: on every PATCH /me, three properties (id, password_hash and api_key, which UserPatch does not declare) were dropped while the server answered with a success status. A client that sent them would believe they were saved. On signup, role and credit_limit_cents still stick and the other nine are dropped without an error.

The real fix

Finally the fix from Step 5, with python probe.py v3.

caller    request          tried  allowed  landed  UNAUTH  quiet  rejected
alice     PATCH /me           13        5       5       0      0         8
carol     PATCH /me           13        5       5       0      0         8
dave      PATCH /me           13        5       5       0      0         8
alice     PATCH /users/2      13        0       0       0      0        13
carol     PATCH /users/2      13        1       1       0      0        12
dave      PATCH /users/2      13        5       5       0      0         8
anonymous POST /signup        11        0       0       0      0        11
v3: 0 unauthorized property writes stuck

Zero. Every row shows landed equal to allowed, which also shows the allowed writes still work, and every other property is rejected with an error. The signup row shows all 11 extra properties refused.

Step 8: Lock it in with a test suite

A probe you run once is a lesson; a probe in the test suite is a guard. Save conftest.py, which adds an --app option so the same tests can run against any version, starts one server for the session, and resets the database before every test.

# conftest.py
import pytest

from harness import Api


def pytest_addoption(parser):
    parser.addoption("--app", default="app_v3", help="module that provides make_app(store)")


@pytest.fixture(scope="session")
def api(request):
    with Api(request.config.getoption("--app"), 8830) as running:
        yield running


@pytest.fixture(autouse=True)
def fresh(api):
    api.store.reset()

Then save test_properties.py. It checks 18 things against policy.py: the exact property set each audience receives from /users/{id}, /me and /users, the exact set each caller can write through each endpoint, signup, and one test for the drift problem from Step 6. The exact-set comparison matters. Checking only that nothing extra appears would pass an API that refuses everything, so the tests also fail if an allowed property goes missing.

# test_properties.py - the property matrix: who may read and write which property
import pytest

from harness import flatten
from policy import IDS, READ, ROLES, WRITE, audience
from probe import probe_signup, probe_write

VIEWS = [("alice", 2, "public"), ("alice", 1, "self"), ("carol", 2, "support"), ("dave", 2, "admin")]


def same(got, want):
    extra, missing = sorted(got - want), sorted(want - got)
    problems = []
    if extra:
        problems.append("extra: " + ", ".join(extra))
    if missing:
        problems.append("missing: " + ", ".join(missing))
    assert not problems, "; ".join(problems)


def property_names(response):
    assert response.status_code == 200, response.text
    return set(flatten(response.json()))


def stuck(outcome):
    return {path for path, (status, landed) in outcome.items() if landed}


@pytest.mark.parametrize("caller, target, kind", VIEWS)
def test_user_view(api, caller, target, kind):
    same(property_names(api.call(caller, "GET", f"/users/{target}")), READ[kind])


@pytest.mark.parametrize("caller", ["alice", "carol", "dave"])
def test_me_view(api, caller):
    same(property_names(api.call(caller, "GET", "/me")), READ["self"])


@pytest.mark.parametrize("caller", ["alice", "carol", "dave"])
def test_list_view(api, caller):
    for row in api.call(caller, "GET", "/users").json():
        same(set(flatten(row)), READ[audience(ROLES[caller], IDS[caller], row["id"])])


@pytest.mark.parametrize("caller", ["alice", "carol", "dave"])
def test_patch_me(api, caller):
    same(stuck(probe_write(api, caller, "/me", IDS[caller])), WRITE["self"])


@pytest.mark.parametrize(
    "caller, allowed",
    [("alice", set()), ("carol", WRITE["support"]), ("dave", WRITE["admin"])],
    ids=["alice", "carol", "dave"],
)
def test_patch_user(api, caller, allowed):
    same(stuck(probe_write(api, caller, "/users/{target}", 2)), allowed)


def test_signup_accepts_nothing_extra(api):
    same(stuck(probe_signup(api)), set())


def test_new_column_stays_hidden(api):
    before = {(caller, target): property_names(api.call(caller, "GET", f"/users/{target}")) for caller, target, _ in VIEWS}
    api.store.db.execute("ALTER TABLE users ADD COLUMN ssn_last4 TEXT NOT NULL DEFAULT '6789'")
    for caller, target, _ in VIEWS:
        same(property_names(api.call(caller, "GET", f"/users/{target}")), before[(caller, target)])

Run the suite against the first draft with python -m pytest -q --tb=no -rf --app=app_v1 test_properties.py. The listing below keeps a few of the failing lines; the real run prints all 17.

FFFFFFFFFFFFF.FFFF [100%]
=== short test summary info ===
FAILED test_properties.py::test_user_view[alice-2-public] - AssertionError: extra: api_key, credit_limit_cents, email, email_verified, internal_notes, password_hash, p...
FAILED test_properties.py::test_user_view[alice-1-self] - AssertionError: extra: api_key, internal_notes, password_hash, preferences.flags.billing_exempt
FAILED test_properties.py::test_user_view[carol-2-support] - AssertionError: extra: api_key, password_hash, preferences.flags.billing_exempt, preferences.newsletter, p...
FAILED test_properties.py::test_user_view[dave-2-admin] - AssertionError: extra: api_key, password_hash
FAILED test_properties.py::test_patch_me[alice] - AssertionError: extra: api_key, credit_limit_cents, email_verified, id, internal_notes, password_hash, preferences.fl...
FAILED test_properties.py::test_patch_me[carol] - AssertionError: extra: api_key, credit_limit_cents, email_verified, id, internal_notes, password_hash, preferences.fl...
FAILED test_properties.py::test_patch_me[dave] - AssertionError: extra: api_key, credit_limit_cents, email_verified, id, internal_notes, password_hash, preferences.fla...
FAILED test_properties.py::test_signup_accepts_nothing_extra - AssertionError: extra: api_key, bio, credit_limit_cents, email_verified, id, internal_notes, preferences...
FAILED test_properties.py::test_new_column_stays_hidden - AssertionError: extra: ssn_last4
17 failed, 1 passed in 0.60s

Each failing line names the test and lists the extra properties it found, so a regression tells you which properties leaked or became writable. The one test that passes is Alice calling the staff endpoint, the only check the first draft had. Now the first fix, --app=app_v2, again trimmed.

FFFFFFFFFFFFF.FFF. [100%]
=== short test summary info ===
FAILED test_properties.py::test_user_view[alice-2-public] - AssertionError: extra: api_key, credit_limit_cents, email, email_verified, internal_notes, preferences.flag...
FAILED test_properties.py::test_user_view[alice-1-self] - AssertionError: extra: api_key, internal_notes, preferences.flags.billing_exempt
FAILED test_properties.py::test_user_view[carol-2-support] - AssertionError: extra: api_key, preferences.flags.billing_exempt, preferences.newsletter, preferences.theme
FAILED test_properties.py::test_user_view[dave-2-admin] - AssertionError: extra: api_key
FAILED test_properties.py::test_patch_me[alice] - AssertionError: extra: credit_limit_cents, email_verified, internal_notes, preferences.flags.billing_exempt, role
FAILED test_properties.py::test_patch_me[carol] - AssertionError: extra: credit_limit_cents, email_verified, internal_notes, preferences.flags.billing_exempt, role
FAILED test_properties.py::test_patch_me[dave] - AssertionError: extra: credit_limit_cents, email_verified, internal_notes, preferences.flags.billing_exempt, role
FAILED test_properties.py::test_signup_accepts_nothing_extra - AssertionError: extra: credit_limit_cents, role
16 failed, 2 passed in 0.70s

Sixteen tests fail and two pass. The drift test passes for app_v2, which matches Step 6: a model that mirrors the table is stable until someone edits it. Finally the fix, --app=app_v3.

.................. [100%]
18 passed in 0.59s

All 18 pass in under a second on this machine. From now on, adding a column to the table changes nothing in any response, and adding a property to a response or a request model fails a test until someone edits policy.py to say who may use it. That is the behavior you want: a security decision that has to be written down, in a file that is reviewed.

Common mistakes and traps

Save step_pitfalls.py. It reproduces three traps that are easy to hit even after you have adopted schemas.

# step_pitfalls.py - three small traps around the schema layer
import requests
from fastapi import FastAPI, Request
from pydantic import BaseModel, ConfigDict

from harness import Api, LiveServer
from store import Store


def raw_body_app(store, extra):
    class Update(BaseModel):
        model_config = ConfigDict(extra=extra)
        display_name: str = ""

    app = FastAPI()

    @app.patch("/me")
    async def update_me(body: Update, request: Request):
        payload = await request.json()  # validated `body`, then used the raw payload
        store.update_any(1, payload)
        return {"ok": True}

    return app


for extra in ("ignore", "forbid"):
    store = Store()
    with LiveServer(raw_body_app(store, extra), 8821) as server:
        response = requests.patch(server.url + "/me", json={"display_name": "Alice N.", "role": "admin"}, timeout=10)
    print(f"raw body re-read with extra={extra!r:<9}: HTTP {response.status_code}, "
          f"role in the database is {store.get(1)['role']!r}")

print()
for version, port in (("app_v2", 8822), ("app_v3", 8823)):
    with Api(version, port) as api:
        response = api.call("alice", "PATCH", "/me", json={"bio": None})
        print(f"{version}: PATCH /me with bio set to null -> HTTP {response.status_code}")
        api.store.reset()
        response = api.call("alice", "PATCH", "/me", json={"is_admin": True})
        message = response.json()["detail"][0]["msg"] if response.status_code == 422 else "accepted and ignored"
        print(f"{version}: PATCH /me with an unknown property is_admin -> HTTP {response.status_code} ({message})")

Run it with python step_pitfalls.py.

raw body re-read with extra='ignore' : HTTP 200, role in the database is 'admin'
raw body re-read with extra='forbid' : HTTP 422, role in the database is 'member'

app_v2: PATCH /me with bio set to null -> HTTP 500
app_v2: PATCH /me with an unknown property is_admin -> HTTP 200 (accepted and ignored)
app_v3: PATCH /me with bio set to null -> HTTP 200
app_v3: PATCH /me with an unknown property is_admin -> HTTP 422 (Extra inputs are not permitted)

What each trap teaches

Validating a body and then using the raw one

The first two lines come from a handler that declares a Pydantic model, which makes the request look validated, and then calls await request.json() and passes the raw payload to the update helper. With Pydantic’s default of ignoring extra keys, the validation passes and the raw payload carries role into the database. With extra="forbid" the same sloppy handler is refused with 422, so strict input models protect you even on the day someone writes the wrong line. This is a strong argument for forbid over silently ignoring.

Explicit nulls in a PATCH

FastAPI’s documentation recommends partial updates with Pydantic’s exclude_unset: “If you want to receive partial updates, it’s very useful to use the parameter exclude_unset in Pydantic’s model’s .model_dump().” But a client that sends "bio": null has set the property to None, so exclude_unset keeps it and the database refuses it. In the first fix that produced a 500 error. Version 3 uses exclude_none=True, which treats an explicit null as not sent and answers 200. The cost is that a client cannot clear a property by sending null; decide that per property.

Quiet success versus a loud refusal

The last two lines show an unknown property, is_admin, which exists nowhere in this schema. The first fix accepts the request and ignores the key. Version 3 answers 422 with “Extra inputs are not permitted”. Refusing is better for an API you control, because a client that sends a property you do not accept is either buggy or probing, and you want to know about both. The trade-off is compatibility: an old client that still sends a removed property will start failing, so for a public API with many clients, plan how you retire properties. One more detail: FastAPI’s own 422 body for /me includes an input key that repeats the submitted value, which is why the staff route’s parse helper leaves it out; whether that matters depends on what your logs and proxies keep.

A short checklist

A few smaller traps are worth a checklist. Typing a nested object as dict opts its contents out of property-level filtering and validation, so give nested JSON its own model. Do not build SQL from client keys, even after a schema check; keep the allowlist in the data layer as well. Do not let a single table-shaped model serve every audience. And keep the tests honest: the probe and the matrix only know the properties and endpoints you gave them, so when you add a route, add it to the matrix. The route-inventory idea from the previous tutorial applies here unchanged.

Check the whole thing end to end, then keep going

You are done when these three commands print the results shown. The first two exercise the fixed API from the outside, and the third is the permanent guard.

python attack.py v3
python probe.py v3
python -m pytest -q --app=app_v3 test_properties.py

Expect v3: 0 of 8 attacks succeeded, v3: 0 unauthorized property writes stuck, and 18 passed. Then run the same three commands with v1 and app_v1 to confirm that the tools can fail, which is how you know a green result means something.

To keep learning, try these in order. Add a property, for example phone, to the table and watch which test demands a decision. Add a new route and extend the matrix, then add an automatic route inventory so a forgotten route fails the build. Write tests for function-level authorization, the API5:2023 entry that Attack 6 touched. Pin the response schema in your OpenAPI document so that a change to a response model shows up in review. Replace the lab’s token table with real authentication. And once you can state the property policy for your own API in a table like the ones in Step 1, run the probe against it. Our property-based testing tutorial shows how to generate such inputs automatically, and the earlier tutorials on CSRF, SSRF and path traversal cover neighboring weaknesses on the same OWASP list.

Sources

Tags:

API SecurityApplication SecurityFastAPIOWASPPydanticPython

Share

No Comment! Be the first one.

Leave a Reply

Your email address will not be published. Required fields are marked *

Related Posts