DiffDojo

2026-09-05 · 8 min read

Python code review exercise: an authenticated export leaks tenant data

In January 2021, archivists scraped 70 terabytes of data from Parler through its API. The endpoints were live, the servers were running, and nobody needed to break in because the API simply handed out records to anyone who asked for them in the right order. No stolen credentials, no SQL injection, no zero-day. Just an API that checked whether a request existed and never checked whether it was allowed to see what it was asking for. That pattern has a name: Broken Object Level Authorization, the #1 entry on the OWASP API Security Top 10. And the version of it you will see in this exercise is subtler than Parler's, because the endpoint does authenticate the caller. It just never asks which org's data they're supposed to see.

This is a Python code review exercise for anyone who reviews FastAPI or Django pull requests in a multi-tenant codebase. The PR adds a CSV export endpoint. It imports the right auth dependency, it streams the response correctly, it even handles an optional query parameter. The bug is that it authenticates the user without scoping the query to their organization, which means any logged-in user downloads every tenant's invoices. The exercise teaches the difference between authentication ("who are you?") and authorization ("what are you allowed to see?"), which is the single most common security flaw in API code and the one most likely to survive both automated tests and a casual review.

What this PR is supposed to do

The requirements are short:

  1. Add GET /exports/invoices.csv: a signed-in user downloads their org's invoices (number, issue date, total, status) as a CSV file, optionally filtered by status.
  2. Every endpoint that returns org-owned data must scope its query to the caller's org through the documented tenant-isolation dependency (current_org_member) and filter on member.org_id. Authentication alone is not scoping.
  3. current_org_member is the tenant-isolation boundary: an org-scoped query that does not filter on member.org_id returns other tenants' rows and is a defect even if the caller is authenticated.
  4. The export streams a real CSV (header row + one row per invoice) and reuses the same invoice fields the list endpoint exposes.

The codebase is a multi-tenant SaaS. Users can belong to several orgs, and every request carries an X-Org-Id header. current_org_member in app/deps.py is the audited boundary: it maps an authenticated user plus that header to a Membership, and the returned member.org_id is what every org-scoped query filters on. get_current_user only proves the request is authenticated, not which org's data it may see. Invoice rows from all orgs live in one invoices table keyed by org_id.

The PR

Three files in the diff. Two are unchanged context; the third is the new endpoint. Read all three before you decide anything, because the bug is the gap between what the context files establish and what the new file does.

app/deps.py (unchanged)

app/deps.py unchanged
1 from fastapi import Depends, Header, HTTPException, status
2 from sqlalchemy.orm import Session
3
4 from app.database import get_db
5 from app.models import Membership, User
6 from app.security import get_current_user
7
8
9 def current_org_member(
10 user: User = Depends(get_current_user),
11 x_org_id: int = Header(..., alias="X-Org-Id"),
12 db: Session = Depends(get_db),
13 ) -> Membership:
14 """Resolve the caller's membership in the requested org.
15
16 TENANT-ISOLATION BOUNDARY. This is the one dependency that binds a request
17 to exactly one org: it verifies the authenticated user really belongs to
18 `X-Org-Id` and returns their Membership. The returned `member.org_id` is
19 the ONLY value an org-scoped query may filter on.
20
21 Every endpoint that reads or writes org-owned data MUST depend on this and
22 MUST filter by `member.org_id`. Authentication alone is not enough -- a
23 query that skips this scope returns other tenants' rows.
24 """
25 membership = (
26 db.query(Membership)
27 .filter(
28 Membership.user_id == user.id,
29 Membership.org_id == x_org_id,
30 )
31 .first()
32 )
33 if membership is None:
34 raise HTTPException(
35 status_code=status.HTTP_403_FORBIDDEN,
36 detail="Not a member of this org",
37 )
38 return membership

app/routers/invoices.py (unchanged)

app/routers/invoices.py unchanged
1 from fastapi import APIRouter, Depends
2 from sqlalchemy.orm import Session
3
4 from app.database import get_db
5 from app.deps import current_org_member
6 from app.models import Invoice, Membership
7 from app.schemas import InvoiceOut
8
9 router = APIRouter(prefix="/invoices", tags=["invoices"])
10
11
12 @router.get("", response_model=list[InvoiceOut])
13 async def list_invoices(
14 status: str | None = None,
15 member: Membership = Depends(current_org_member),
16 db: Session = Depends(get_db),
17 ) -> list[Invoice]:
18 """List the caller's org invoices, most recent first."""
19 query = db.query(Invoice).filter(Invoice.org_id == member.org_id)
20 if status is not None:
21 query = query.filter(Invoice.status == status)
22 return query.order_by(Invoice.issued_at.desc()).all()

app/routers/exports.py (new file)

developer wants to merge · app/routers/exports.py +32
1+ import csv
2+ import io
3+
4+ from fastapi import APIRouter, Depends
5+ from fastapi.responses import StreamingResponse
6+ from sqlalchemy.orm import Session
7+
8+ from app.database import get_db
9+ from app.models import Invoice, User
10+ from app.security import get_current_user
11+
12+ router = APIRouter(prefix="/exports", tags=["exports"])
13+
14+
15+ @router.get("/invoices.csv")
16+ async def export_invoices_csv(
17+ status: str | None = None,
18+ user: User = Depends(get_current_user),
19+ db: Session = Depends(get_db),
20+ ) -> StreamingResponse:
21+ """Stream the caller's invoices as a CSV download, optionally filtered by status."""
22+ query = db.query(Invoice)
23+ if status is not None:
24+ query = query.filter(Invoice.status == status)
25+ invoices = query.all()
26+
27+ buffer = io.StringIO()
28+ writer = csv.writer(buffer)
29+ writer.writerow(["number", "issued_at", "total_cents", "status"])
30+ for invoice in invoices:
31+ writer.writerow([
32+ invoice.number,
33+ invoice.issued_at.isoformat(),
34+ invoice.total_cents,
35+ invoice.status,
36+ ])
37+
38+ buffer.seek(0)
39+ return StreamingResponse(
40+ buffer,
41+ media_type="text/csv",
42+ headers={"Content-Disposition": "attachment; filename=invoices.csv"},
43+ )

Thirty-two lines, clean imports, correct use of StreamingResponse and csv.writer. The docstring even says "the caller's invoices." If you skim this diff between two meetings, it reads like a finished feature. Now look more carefully.

Try it yourself first

The same PR is a hands-on exercise on DiffDojo.

Leave your comments, pick your verdict, and see the canonical review after you submit. No signup required.

Review this PR cold →

Walkthrough starts below. If you want to form your own opinion first, stop here and try the exercise above.

The docstring says "the caller's." The query says "everyone's."

Look at line 22 of the new file:

18+ user: User = Depends(get_current_user),
19+ db: Session = Depends(get_db),
20+ ) -> StreamingResponse:
21+ """Stream the caller's invoices as a CSV download, optionally filtered by status."""
22+ query = db.query(Invoice)

That's db.query(Invoice) with no filter. It selects every invoice in the database, across every organization. The optional status filter on line 24 narrows by payment status, not by tenant. The user variable from get_current_user is injected on line 18, which proves the caller is logged in, but nobody ever uses it to restrict which rows come back. The user object does not even carry an org_id in this codebase; the org binding lives in current_org_member, which this endpoint does not call.

Now compare with the existing list_invoices in app/routers/invoices.py, lines 15 and 19:

15 member: Membership = Depends(current_org_member),
16 db: Session = Depends(get_db),
17 ) -> list[Invoice]:
18 """List the caller's org invoices, most recent first."""
19 query = db.query(Invoice).filter(Invoice.org_id == member.org_id)

That endpoint depends on current_org_member, which resolves the caller's membership and gives back member.org_id. The very first thing the query does is .filter(Invoice.org_id == member.org_id). The new export endpoint skips both of those steps. It has authentication (who are you?) but not authorization (what org's data may you see?).

What actually happens on Monday morning

A user from Org A, logged in with a perfectly valid session, hits GET /exports/invoices.csv. The endpoint checks their session token, confirms they are a real user, and then runs db.query(Invoice).all(). The CSV that comes back contains every invoice from every organization in the system: Org A's, Org B's, Org C's, everyone's. Invoice numbers, amounts, dates, payment statuses. The user opens it in a spreadsheet and sees customers they have never heard of, because those customers belong to a different tenant.

Nothing errors. Nothing 401s. The response is a perfectly formatted CSV. If the user only has a few invoices of their own, they might not even notice the extra rows at first, or they might assume the export is "just showing everything" and report it as a cosmetic bug. Meanwhile, the data has left the building.

This is not a theoretical risk. The status filter makes it worse: GET /exports/invoices.csv?status=overdue returns every overdue invoice from every org. An attacker who knows this endpoint exists does not need to guess IDs or manipulate headers. They log in to any account, call the endpoint, and receive a full export of every tenant's billing data.

The comment you would actually leave

16+ async def export_invoices_csv(
17+ status: str | None = None,
18+ user: User = Depends(get_current_user),
Yyou commented

This depends on get_current_user (authentication) but not current_org_member (tenant scoping). The query on line 22 is db.query(Invoice) with no org_id filter, so it returns invoices from every org in the system. The existing list_invoices endpoint does this correctly: it takes Depends(current_org_member) and filters on member.org_id. This endpoint needs the same pattern. As written, any authenticated user can download every tenant's billing data.

19+ db: Session = Depends(get_db),

Verdict: request changes

This is a blocker. The endpoint leaks cross-tenant data to any authenticated user. The CSV streaming, the status filter, the response headers are all fine. The only problem is that the query is unscoped, and the fix is to use the same dependency and filter that the list endpoint already uses.

The fix

Replace get_current_user with current_org_member and add the org filter to the query:

9- from app.models import Invoice, User
10- from app.security import get_current_user
9+ from app.deps import current_org_member
10+ from app.models import Invoice, Membership
11
12 router = APIRouter(prefix="/exports", tags=["exports"])
13
14
15 @router.get("/invoices.csv")
16 async def export_invoices_csv(
17 status: str | None = None,
18- user: User = Depends(get_current_user),
18+ member: Membership = Depends(current_org_member),
19 db: Session = Depends(get_db),
20 ) -> StreamingResponse:
21 """Stream the caller's invoices as a CSV download, optionally filtered by status."""
22- query = db.query(Invoice)
22+ query = db.query(Invoice).filter(Invoice.org_id == member.org_id)

Four lines change. The import swaps User and get_current_user for Membership and current_org_member. The function signature takes member instead of user. The query starts with .filter(Invoice.org_id == member.org_id). Everything else stays the same.

The test that catches it

Set up two orgs with their own invoices. Authenticate as a user in Org A. Call GET /exports/invoices.csv. Parse the CSV. Assert that every invoice in the response belongs to Org A and none belong to Org B. Then test with the status filter: GET /exports/invoices.csv?status=paid should return only Org A's paid invoices, not Org B's.

In pytest, with two fixtures for two orgs:

1+ def test_export_scoped_to_org(client_org_a, invoices_org_a, invoices_org_b):
2+ resp = client_org_a.get("/exports/invoices.csv")
3+ assert resp.status_code == 200
4+ rows = resp.text.strip().split("\n")
5+ # header + org A invoices only
6+ assert len(rows) == 1 + len(invoices_org_a)
7+ numbers = {r.split(",")[0] for r in rows[1:]}
8+ org_a_numbers = {i.number for i in invoices_org_a}
9+ assert numbers == org_a_numbers
10+
11+ def test_export_with_status_filter_scoped(client_org_a, invoices_org_a, invoices_org_b):
12+ resp = client_org_a.get("/exports/invoices.csv?status=paid")
13+ rows = resp.text.strip().split("\n")
14+ org_a_paid = [i for i in invoices_org_a if i.status == "paid"]
15+ assert len(rows) == 1 + len(org_a_paid)

On the buggy code, the first test fails: len(rows) includes Org B's invoices, so the count is wrong and the set of invoice numbers contains entries that don't belong to Org A. On the fixed code, both tests pass. The key is that the test has data from two orgs and asserts that only one org's rows appear. A test with a single org cannot catch this bug, because db.query(Invoice).all() and db.query(Invoice).filter(Invoice.org_id == member.org_id).all() return the same rows when there is only one org in the database.

The habit: trace the query to the tenant boundary

Every time a PR adds or modifies an endpoint that returns tenant-owned data, ask one question: what binds this query to the caller's tenant? Then trace backwards from the WHERE clause (or .filter()) to the value it filters on, and from that value to wherever it was resolved. If the chain goes: HTTP header or token, through an audited dependency that verifies membership, into the query filter, you are looking at a scoped endpoint. If the chain stops at "the user is logged in" and never reaches the tenant, you are looking at a cross-tenant data leak that will pass every test anyone writes with a single-tenant fixture.

Authentication answers "who are you?" Authorization answers "what are you allowed to see?" The first is never a substitute for the second, and the PR that mixes them up will always look clean in review, because the auth import is right there in the signature and the endpoint does not crash.

Read next Python code review exercise: a search query becomes SQL
← All posts