10 — Whose data is it?

By the end of chapter 9, the server no longer runs input as code and no longer fetches whatever URL it is handed. Every request it handles now does what the code says. That leaves a quieter question the code has never asked: should this user be allowed to do this to this record?

Authentication answers who is calling. Authorization answers what they may touch. The app has a solid answer to the first: a random session ID in an HttpOnly cookie, stored server-side as a SHA-256 hash, with passwords hashed by scrypt. It has almost no answer to the second. That is not unusual. Broken access control is A01 in the OWASP Top 10:2025, the first category in the list, and the attacks in this chapter need no tooling at all. A logged-in user changes a number in a URL, adds a field to a JSON body, or calls an endpoint the page never links to.

The setup for every run below: Flask 3.1.3 and PostgreSQL 16.2, four seeded users (admin, alice, bob, carol) and six notes. Alice owns notes 1, 2 and 5 and has shared note 5 (her trip plan) with bob. Bob owns note 3. Carol owns note 4, and the admin owns note 6. The app takes a FIX environment variable naming which fixes are switched on, so each attack can be run against the same code before and after.

Changing the number in the URL

The JSON API serves a note by its ID:

@app.get("/api/notes/<int:note_id>")
@login_required
def api_note(note_id):
    n = db().execute("SELECT * FROM notes WHERE id = %s", (note_id,)).fetchone() or abort(404)
    return jsonify(id=n["id"], owner_id=n["owner_id"], title=n["title"], body=n["body"])

The query is parameterized, so chapter 8 has nothing to say about it. @login_required checks there is a user. Nothing checks that the user owns the note. Bob logs in and walks the IDs:

bob = requests.Session()
bob.post(B + "/login", data={"username": "bob", "password": "bob-also-long-passphrase"})
for i in range(1, 10):
    r = bob.get(f"{B}/api/notes/{i}")
    ...
r = bob.put(B + "/api/notes/2", json={"title": "Doctor", "body": "Appointment cancelled"})

  /api/notes/1  200  owner=2  'Groceries': 'eggs, coffee'
  /api/notes/2  200  owner=2  'Doctor': 'Appointment Tuesday 9am, ref 4471'
  /api/notes/3  200  owner=3  "Bob's todo": 'fix the bike'
  /api/notes/4  200  owner=4  'Passwords': 'bank: sunshine1 (I know)'
  /api/notes/5  200  owner=2  'Trip plan': 'Lisbon in May'
  /api/notes/6  200  owner=1  'Ops': 'rotate the db password on the 1st'
  /api/notes/7  404
  /api/notes/8  404
  /api/notes/9  404
PUT alice's note 2: 200

Every note in the database, including the admin’s and carol’s, and a successful write to alice’s appointment. This is an insecure direct object reference (IDOR). API people call the same bug broken object level authorization (BOLA), and it is the first entry in the OWASP API Security Top 10 2023. The CWE name describes it exactly: CWE-639, authorization bypass through user-controlled key. The client supplies the key, and the server trusts that a client holding a key is entitled to the row behind it.

Sequential integers make the walk trivial, and it is tempting to fix that by switching to UUIDs. Random IDs do make guessing impractical, but IDs leak: they appear in shared links, browser history, logs, referrers, and in other API responses (this one returns owner_id). An unguessable ID is a reasonable second layer. It is not an access check.

Putting the owner in the query

The fix is to make “may this user see this note” part of loading it. One function does the lookup for every route, and it asks PostgreSQL for the note and the permission in a single query:

def load_note(note_id, want="read"):
    """The note only comes back if this user may see it."""
    if want == "read":
        sql = """SELECT n.* FROM notes n
                 WHERE n.id = %s AND (n.owner_id = %s OR EXISTS (
                   SELECT 1 FROM note_shares s WHERE s.note_id = n.id AND s.user_id = %s))"""
        args = (note_id, g.user["id"], g.user["id"])
    else:  # write/delete/share: owner only
        sql = "SELECT * FROM notes WHERE id = %s AND owner_id = %s"
        args = (note_id, g.user["id"])
    note = db().execute(sql, args).fetchone()
    if note is None:
        abort(404)          # same answer for "doesn't exist" and "not yours"
    return note

Three decisions are packed in here. Reading is allowed to the owner and to anyone the note is shared with; writing, deleting and re-sharing are owner-only, because a share grants reading, not control. The check lives in the data access, not in each view, so a new route that calls load_note cannot forget it. And a note that exists but isn’t yours returns the same 404 as a note that doesn’t exist, so the API doesn’t confirm which IDs are real.

With FIX=idor, the same script:

  /api/notes/1  404
  /api/notes/2  404
  /api/notes/3  200  owner=3  "Bob's todo": 'fix the bike'
  /api/notes/4  404
  /api/notes/5  200  owner=2  'Trip plan': 'Lisbon in May'
  /api/notes/6  404
  ...
PUT alice's note 2: 404

Bob sees his own note and the one alice shared with him, and nothing else. The write to note 2 fails at the lookup, before the UPDATE runs.

PostgreSQL can enforce the same rule one level lower with row-level security: ALTER TABLE notes ENABLE ROW LEVEL SECURITY and a policy such as USING (owner_id = current_setting('app.user_id')::int OR ...), with the app setting app.user_id per transaction. That catches a query someone writes without the ownership clause. It also adds moving parts (the app’s database role must not own the table or bypass RLS, and connection pools must reset the setting), so it works best as a backstop behind a lookup function like the one above, not instead of it.

One extra field in the JSON

The profile endpoint lets a user change their display name and bio. The first version was written to be convenient: take whatever keys the client sent, keep the ones that are real columns, and update them.

@app.patch("/api/me")
@login_required
def update_me():
    data = request.get_json()
    cols = {r["column_name"] for r in db().execute(
        "SELECT column_name FROM information_schema.columns WHERE table_name = 'users'")}
    fields = {k: v for k, v in data.items() if k in cols and k != "id"}
    sets = ", ".join(f"{k} = %s" for k in fields)
    db().execute(f"UPDATE users SET {sets} WHERE id = %s", (*fields.values(), g.user["id"]))

Checking keys against information_schema stops SQL injection through the column names. It does nothing about which columns a user should be allowed to set, and the users table has an is_admin column. The profile page never sends it. Bob does:

print("GET /admin before:", bob.get(B + "/admin").status_code)
r = bob.patch(B + "/api/me", json={"display_name": "Bob", "is_admin": True})
print("PATCH /api/me:", r.status_code, r.json())
print("GET /admin after:", bob.get(B + "/admin").status_code)

GET /admin before: 403
PATCH /api/me: 200 {'bio': None, 'display_name': 'Bob', 'id': 3, 'is_admin': True, 'username': 'bob'}
GET /admin after: 200 <h1>Admin</h1><ul><li>admin admin@notes.test</li><li>alice alice@notes.test</li>...

This is mass assignment (CWE-915, improperly controlled modification of dynamically determined object attributes). Frameworks that bind request bodies straight onto model objects make it easy to write by accident; ORMs with Model(**request.json) or obj.update(**data) do the same thing as the code above. OWASP’s API list folds it into API3, broken object property level authorization, which also covers the reverse problem: returning properties the caller shouldn’t see. The note API above returns owner_id, which is harmless here; returning a whole user row would include password_hash.

The fix is an explicit allowlist of the fields this endpoint edits, and a hard failure on anything else:

PROFILE_FIELDS = {"display_name", "bio"}

    unknown = set(data) - PROFILE_FIELDS
    if unknown:
        return jsonify(error="fields not allowed", fields=sorted(unknown)), 400

GET /admin before: 403
PATCH /api/me: 400 {'error': 'fields not allowed', 'fields': ['is_admin']}
GET /admin after: 403

Rejecting with a 400 rather than silently dropping the field is a choice. Dropping is friendlier to clients that send whole objects back; rejecting makes a probing attempt visible in logs and catches client bugs early. Either way the rule is the same: the server decides which fields a request may change, per endpoint, and a blocklist of “dangerous” fields is the wrong shape, because the next column someone adds (email_verified, plan, org_id) won’t be on it. In larger apps this becomes a separate input schema per operation (a Pydantic model or a marshmallow schema with only the editable fields), never the database model itself.

The admin page checked; its API didn’t

The admin page is protected:

@admin_bp.get("/admin")
@admin_required
def admin_page(): ...

The page’s JavaScript calls two JSON endpoints, and whoever wrote those copied the wrong decorator:

@admin_bp.get("/api/admin/users")
@login_required                          # the page above checks is_admin; this one forgot
def api_admin_users(): ...

@admin_bp.delete("/api/admin/users/<int:user_id>")
@login_required
def api_admin_delete_user(user_id): ...

Bob has never seen the admin page, but the endpoint names are in its JavaScript, and admin APIs follow predictable patterns anyway:

$ curl -b bob.cookies http://127.0.0.1:5401/api/admin/users
[{"email":"admin@notes.test","id":1,"is_admin":true,"username":"admin"},
 {"email":"alice@notes.test","id":2,"is_admin":false,"username":"alice"}, ...]
$ curl -b bob.cookies -X DELETE http://127.0.0.1:5401/api/admin/users/4
{"ok":true}

Carol’s account is gone. This is missing function-level authorization, API5 in the OWASP API list. Hiding a link in the UI is not a control; the server answers whoever asks.

Testing every route as every role

Bugs like this hide because each route is reviewed alone, and the decorator looks plausible in isolation. A more reliable check is to ask every route the same question from every role and look at the grid. Flask knows all its routes, so the test can enumerate them instead of relying on someone to keep a list:

from app import app
who = {"anon": requests.Session(),
       "bob": session("bob", "bob-also-long-passphrase"),
       "admin": session("admin", "Wq7-correct-horse-admin")}
for rule in sorted(app.url_map.iter_rules(), key=lambda r: r.rule):
    if "GET" not in rule.methods or rule.endpoint == "static":
        continue
    path = re.sub(r"<int:\w+>", "3", re.sub(r"<(?:string:)?\w+>", "x", rule.rule))
    codes = [who[k].get(B + path, allow_redirects=False).status_code for k in who]
    print(f"{path:28} {codes[0]:>5} {codes[1]:>5} {codes[2]:>5}")

route                         anon   bob admin
/admin                         401   403   200
/api/admin/users               401   200   200
/api/notes/3                   401   200   200
/me                            401   200   200
/notes/3                       401   200   200
/reset/x                       200   200   200

The hole is one cell: bob gets 200 on /api/admin/users, a route under /admin where the row above it says 403. In a real test suite the grid becomes an expected matrix checked in alongside the code, and a new route with no entry fails the build. The same idea extends to write methods and to object-level checks (bob against alice’s note IDs, alice against bob’s).

The fix moves the check from each route to the blueprint, so every route registered on admin_bp, including ones not yet written, passes through it:

@admin_bp.before_request
def require_admin():                 # runs before every route in this blueprint
    if not g.user: abort(401)
    if not g.user["is_admin"]: abort(403)

route                         anon   bob admin
/admin                         401   403   200
/api/admin/users               401   403   200
...

Bob’s DELETE now returns 403 too. The principle is deny by default: protection attached to a group of routes, so that forgetting is safe, rather than to each route, where forgetting is a breach. The same thinking applies across services. If an internal admin service trusts any caller on the network, chapter 9 showed how a server can be made to be that caller. Authorization models (roles, attributes, relationships) and where checks belong in a larger system are covered in the System Design security chapter.

Is there an account called alice?

The login endpoint was written to be helpful:

if not user:
    return "No account with that username.\n", 401
if not check_pw(password, user["password_hash"]):
    return "Wrong password.\n", 401

The password reset endpoint did the same with “No account with that email.” (404). Both tell an anonymous caller which accounts exist. That matters more than it seems: a confirmed list of usernames or emails turns password guessing from a two-dimensional search into a one-dimensional one, and a confirmed email at a sensitive service is useful for targeted phishing on its own.

Fixing the messages is easy. The timing is the part that’s usually missed. For a username that doesn’t exist, the code returns before running scrypt. For one that does, it hashes the guess. Twenty attempts each with a wrong password:

for u in ["alice", "zelda"]:
    for _ in range(20):
        a = time.perf_counter()
        r = requests.post(B + "/login", data={"username": u, "password": "wrong-guess"})
        xs.append((time.perf_counter() - a) * 1000)

alice  401 'Wrong password.'                        median  34.2 ms
zelda  401 'No account with that username.'         median   3.4 ms
reset alice@notes.test   200 'If that address has an account, a reset link is on its way.'
reset zelda@notes.test   404 'No account with that email.'

A tenfold gap (the first writer measured 32.8 ms against 6.8 ms on the same code; the shape is the same). Even with identical messages, a response that comes back in 3 ms means “no such user”. The password hash is deliberately slow, and that slowness is the signal.

The fix gives both branches the same message and the same work. When the user doesn’t exist, the code verifies the guess against a dummy hash computed at startup with the same scrypt parameters:

DUMMY_HASH = hash_pw(secrets.token_hex(16))

user = db().execute("SELECT * FROM users WHERE username = %s", (username,)).fetchone()
ok = check_pw(password, user["password_hash"] if user else DUMMY_HASH) and user is not None
if not ok:
    return "Invalid username or password.\n", 401

alice  401 'Invalid username or password.'          median  33.6 ms
zelda  401 'Invalid username or password.'          median  33.5 ms
reset alice@notes.test   200 'If that address has an account, a reset link is on its way.'
reset zelda@notes.test   200 'If that address has an account, a reset link is on its way.'

The reset endpoint now gives one answer for every address. Its timing still differs slightly, because a real address means a database insert and a mail write: medians of 6.2 ms against 4.5 ms over 20 requests each, after the fix. The robust fix is to put the mail on a queue and return immediately for both cases, so the request does the same work either way.

Sign-up is the hardest leak to close: “that username is taken” is the answer by design. The usual mitigations are to sign up by email and send “you already have an account” to the address instead of saying so on screen, and to rate-limit the sign-up check the same way as login.

Login throttling, briefly

The first writer of this chapter also measured how the login endpoint behaves under repeated guessing. Those runs aren’t extended here; this section only summarizes what was observed and what the fixes trade off.

Three things were seen. Carol’s password, sunshine1, fell to a short list of common passwords with no limit in place: a few hundred requests is nothing for an endpoint that answers in tens of milliseconds. A per-IP limit was weak when the app was configured to read the client address from X-Forwarded-For without a proxy in front, because the client writes that header and can put anything in it. And the throttle that did stop the guessing also returned 429 to carol herself, typing the right password, until the window passed. That last point is the core trade-off: any limit keyed on the account is also a way for someone else to lock that account.

The design that holds up combines several pieces, none of which is enough alone:

• Per-account limits with growing delays rather than hard lockout. The app’s version allows five failures, then makes each further attempt wait 1, 2, 4 seconds and so on, capped at 15 minutes, in a login_failures table. A delay slows an online guesser to a crawl while letting the real user in after a short wait; a permanent lockout hands anyone who knows a username a denial-of-service button. NIST SP 800-63B (section 3.2.2) caps consecutive failed attempts on one account at 100 and explicitly allows lower limits plus measures such as increasing delays and bot challenges.
• Per-source limits as well, because password spraying tries one common password against many accounts and never trips a per-account counter. The app counts failures per address over ten minutes.
• Knowing which address is the client. Behind a reverse proxy, request.remote_addr is the proxy, and the real client is in X-Forwarded-For. Werkzeug’s ProxyFix reads it, and its x_for count says how many proxies to trust from the right-hand end of the header. Set it to exactly the number of proxies that are really there, and only when they are; with no proxy, the header is just attacker input.
• Not relying on the password alone. A breached-password check at sign-up and password change (SP 800-63B, section 3.1.1.2, requires comparing new passwords against a blocklist of common and compromised passwords) would have refused sunshine1 in the first place. A second factor makes a guessed password insufficient. Telling the user, by email, about failed attempts and new sign-ins turns a silent attack into one the owner can react to.

How a limiter is built and kept consistent across many app servers is the subject of the rate limiter chapter; the login case is one of its clients, with the added twist that the key (an account) can be chosen by the attacker.

Password reset

A reset flow is a second way to log in, and it is often built with less care than the first. The app’s first version has three problems in the code. The token-entropy and expiry fixes below are described from the code and the docs rather than attacked; the Host-header problem was run.

The token

token = hashlib.md5(f"{email}{int(time.time())}".encode()).hexdigest()
db().execute("INSERT INTO reset_tokens (token, user_id) VALUES (%s, %s)", (token, user["id"]))

It looks random (32 hex characters), but it is a function of two things an attacker knows: the email address and roughly when the request was made. A check against the token that landed in the outbox confirmed it, offline: md5("alice@notes.test" + "1794286688") reproduced it exactly. Someone who requests a reset for alice and notes the time has a handful of candidates per second of uncertainty. The app’s own logs show the same weakness from the other side: two reset requests for the same address within one second produced the same token, and the second insert failed with duplicate key value violates unique constraint "reset_tokens_pkey", a 500 error.

The token is also stored in plaintext, never expires and is never deleted. A database read (a backup, a SQL injection like the one in chapter 8, a support tool) yields working reset links for every user who ever asked for one.

The fixed version:

token = secrets.token_urlsafe(32)                        # 256 bits from the OS CSPRNG
db().execute("DELETE FROM reset_tokens WHERE user_id = %s", (user["id"],))
db().execute("""INSERT INTO reset_tokens (token, user_id, expires_at)
                VALUES (%s, %s, now() + interval '30 minutes')""",
             (hashlib.sha256(token.encode()).hexdigest(), user["id"]))

and on use:

row = db().execute("""DELETE FROM reset_tokens WHERE token = %s AND expires_at > now()
                      RETURNING *""", (hashlib.sha256(token.encode()).hexdigest(),)).fetchone()
...
db().execute("DELETE FROM sessions WHERE user_id = %s", (row["user_id"],))

The token comes from secrets, which uses the operating system’s CSPRNG. Only its SHA-256 is stored, the same way the app stores session IDs; a fast hash is fine here because the token has 256 bits of entropy and can’t be guessed. Issuing a new token deletes older ones. The lookup is a DELETE ... RETURNING with the expiry in the WHERE clause, so a token works once and only within 30 minutes, and two concurrent uses can’t both succeed. After the password changes, every existing session for that user is deleted, which logs out an attacker who was already in. The single-use behavior was checked:

use 1: 200 Password changed.
use 2: 400 Invalid or expired link.

The link’s host

The unfixed code builds the link with url_for(..., _external=True). Inside a request, Flask takes the host for an external URL from the request itself, which means from the Host header the client sent. A reset request with a different Host:

for h in "127.0.0.1:5401" "evil.test" "127.0.0.1:5401@evil.test"; do
  curl -s -o /dev/null -w '%{http_code}\n' -H "Host: $h" -d email=alice@notes.test http://127.0.0.1:5401/reset
done

Host: 127.0.0.1:5401             -> 200
Host: evil.test                  -> 200
Host: 127.0.0.1:5401@evil.test   -> 200
outbox:
{"to": "alice@notes.test", "body": "Reset your password: http://127.0.0.1:5401/reset/c71e0cd3..."}
{"to": "alice@notes.test", "body": "Reset your password: http://evil.test/reset/9c85d2e8..."}
{"to": "alice@notes.test", "body": "Reset your password: http:///reset/21969bac..."}

The second email is the attack. It is a real reset email, sent by the real app to alice’s real address, with a valid token in a link that points at the attacker’s server. If alice clicks it, or if her mail system fetches links to scan them, the token arrives in the attacker’s logs, and they use it on the real site. This is Host-header poisoning of a password reset. (The third request shows Werkzeug rejecting the malformed host value and building a link with an empty host, which is broken but not exploitable.)

Two fixes were applied together. The link is built from a configured base URL instead of the request:

PUBLIC_BASE_URL = "http://127.0.0.1:5401"
link = f"{PUBLIC_BASE_URL}/reset/{token}"

And the app refuses requests for hosts it doesn’t serve. Flask 3.1 added TRUSTED_HOSTS for this: it validates Request.host during routing and raises a SecurityError, a 400, if the host isn’t in the list (a leading dot matches subdomains; the port is ignored).

app.config["TRUSTED_HOSTS"] = ["127.0.0.1"]

Host: 127.0.0.1:5401             -> 200
Host: evil.test                  -> 400
Host: 127.0.0.1:5401@evil.test   -> 400
outbox:
{"to": "alice@notes.test", "body": "Reset your password: http://127.0.0.1:5401/reset/srwVirzG..."}

Either fix alone stops this particular email. Both together are better, because url_for(_external=True) is used for more than reset links, and the Host header also feeds caches and redirects. Behind a proxy that rewrites the host, ProxyFix with x_host has the same caveat as x_for: trust it only from a proxy that is really there. The SERVER_NAME setting, which older advice suggests for this, no longer restricts which hosts a request may use as of Flask 3.1, according to the same page.

A few more reset rules that are worth stating even though they weren’t run: the reset page should not load third-party resources while the token is in the URL (the Referer header would carry it; Referrer-Policy: no-referrer on that page is cheap insurance, and chapter 7 covers the header); the reset itself should not log the user in silently if they have a second factor; and the user should get an email after any password change, so an attacker’s reset doesn’t go unnoticed.

Tokens for API clients, OAuth flows and how sessions compare with JWTs are covered in the essay on JWTs, OAuth and Zero Trust; the System Design security chapter covers password storage and multi-factor authentication at the system level, and Passwords, second factors and passkeys covers what a login itself should ask for.

What the fixed app looks like

All the fixes in this chapter share a pattern. Each bug was a place where the server let the client decide something the server should have decided: which row (IDOR), which columns (mass assignment), which functions (the admin API), which host the links point at (reset). Each fix moved that decision back to the server, and put it somewhere it applies by default: in the one function that loads notes, in a per-endpoint field allowlist, in a blueprint-wide before_request, in configuration. The access matrix test is what keeps it that way as routes are added.

Where this leaves things

The app now decides who may touch what, on the server, with checks that apply by default and a test that shows the full grid. Enumeration is closed on login and reset, the reset flow uses strong single-use tokens and links that can’t be pointed elsewhere, and login guessing is slowed per account and per address.

All of that assumes the request reaching the app is the one the user sent, from a page that really came from the app. So far the app has run on plain HTTP on 127.0.0.1. A real deployment sits on a domain, behind TLS, with DNS records pointing at servers and services the team may have forgotten about. Chapter 11 takes up the connection and the domain: certificates and who may issue them, what a page loses when it pulls in one plain-HTTP resource, and how a stale DNS record lets someone else serve pages on the app’s own subdomain.