Fix IDOR and missing authorization in Python and FastAPI

1. Find every path operation that takes an object ID (path, query or body) and every query it runs: db.get(Model, id), select(...).where(Model.id == id) and bulk queries over a list of IDs.
2. Write one current_user dependency that reads the authenticated identity (a signed session, a verified token) and loads tenant and role from the database, and declare it with Annotated[User, Depends(current_user)] on every route; answer 401 when there is no user.
3. Write the visibility rule once, as a function that returns a SQLAlchemy condition for the user (visible_to(user)), and add it to the where() of the detail, list and export queries.
4. Return 404 for objects outside the caller's scope, the same answer as for missing ones, and refuse a bulk request entirely when any ID is out of scope.
5. In tests, replace only the authentication dependency with app.dependency_overrides, keep the real current_user, and test with users in two tenants, including an admin of the other tenant.

The three implementations share one API and one test suite; only the code below changes. The tests load the one that IMPL names.

def get_invoice(db: Session, user: User, invoice_id: int) -> dict | None:
    invoice = db.get(Invoice, invoice_id)
    return None if invoice is None else invoice.as_dict()


def list_invoices(db: Session, user: User) -> list[dict]:
    stmt = select(Invoice).where(Invoice.tenant_id == user.tenant_id).order_by(Invoice.id)
    return [invoice.as_dict() for invoice in db.scalars(stmt)]


def export_invoices(db: Session, user: User, ids: list[int]) -> list[dict]:
    stmt = select(Invoice).where(Invoice.id.in_(ids)).order_by(Invoice.id)
    return [invoice.as_dict() for invoice in db.scalars(stmt)]

# Members read their own invoices; billing admins read every invoice in their own
# tenant. The tenant condition applies to both, so it binds admins too.
def visible_to(user: User) -> ColumnElement[bool]:
    same_tenant = Invoice.tenant_id == user.tenant_id
    if user.role == "billing_admin":
        return same_tenant
    return and_(same_tenant, Invoice.owner_id == user.id)


def get_invoice(db: Session, user: User, invoice_id: int) -> dict | None:
    stmt = select(Invoice).where(Invoice.id == invoice_id, visible_to(user))
    invoice = db.scalars(stmt).one_or_none()
    return None if invoice is None else invoice.as_dict()


def list_invoices(db: Session, user: User) -> list[dict]:
    stmt = select(Invoice).where(visible_to(user)).order_by(Invoice.id)
    return [invoice.as_dict() for invoice in db.scalars(stmt)]


def export_invoices(db: Session, user: User, ids: list[int]) -> list[dict]:
    stmt = select(Invoice).where(Invoice.id.in_(ids), visible_to(user)).order_by(Invoice.id)
    return [invoice.as_dict() for invoice in db.scalars(stmt)]

def _can_read(user: User, invoice: Invoice) -> bool:
    return invoice.owner_id == user.id or user.role == "billing_admin"


def get_invoice(db: Session, user: User, invoice_id: int) -> dict | None:
    invoice = db.get(Invoice, invoice_id)
    return invoice.as_dict() if invoice is not None and _can_read(user, invoice) else None


def list_invoices(db: Session, user: User) -> list[dict]:
    stmt = select(Invoice).order_by(Invoice.id)
    return [invoice.as_dict() for invoice in db.scalars(stmt) if _can_read(user, invoice)]


def export_invoices(db: Session, user: User, ids: list[int]) -> list[dict]:
    stmt = select(Invoice).where(Invoice.id.in_(ids)).order_by(Invoice.id)
    return [invoice.as_dict() for invoice in db.scalars(stmt) if _can_read(user, invoice)]

def current_user(db: DbDep, user_id: Annotated[str | None, Depends(session_user_id)]) -> User:
    # Tenant and role come from the users table on every request, never from the
    # request itself, so a changed role or a removed user takes effect at once.
    record = None if user_id is None else db.get(UserRecord, user_id)
    if record is None:
        raise HTTPException(status_code=401, detail="sign in first")
    return User(id=record.id, tenant_id=record.tenant_id, role=record.role)

UserDep = Annotated[User, Depends(current_user)]

@app.get("/invoices")
def list_invoices(db: DbDep, user: UserDep) -> list[dict]:
    return impl.list_invoices(db, user)

# Declared before /invoices/{invoice_id}: FastAPI matches path operations in order,
# and "export" would otherwise fail the int conversion with a 422.
@app.get("/invoices/export")
def export_invoices(db: DbDep, user: UserDep, ids: str | None = None) -> Response:
    wanted = parse_ids(ids)
    if wanted is None:
        raise HTTPException(status_code=400, detail=f"ids must be 1 to {MAX_EXPORT_IDS} invoice IDs, comma-separated")
    invoices = impl.export_invoices(db, user, wanted)
    # All or nothing: one missing or unreadable ID refuses the whole export, so a
    # partial file is never mistaken for the full one.
    if len(invoices) != len(wanted):
        raise HTTPException(status_code=404, detail="not found")
    return Response(to_csv(invoices), media_type="text/csv")

@app.get("/invoices/{invoice_id}")
def get_invoice(db: DbDep, user: UserDep, invoice_id: Annotated[int, Path(ge=1, le=999_999_999)]) -> dict:
    invoice = impl.get_invoice(db, user, invoice_id)
    # Missing and not yours get the same answer, so IDs cannot be probed.
    if invoice is None:
        raise HTTPException(status_code=404, detail="not found")
    return invoice

The exploit tests send the attack and check that it fails; functional tests next to them keep ordinary requests working, so a fix that rejects everything fails too.

# Exploit: other users, other tenants and other tenants' admins must get nothing.

def assert_hidden(signed_in, user: str, path: str, number: str) -> None:
    """Denied looks exactly like missing: 404 and none of the invoice's data."""
    res = signed_in(user).get(path)
    body = res.text
    assert res.status_code == 404, f"expected 404, got {res.status_code}: {body}"
    assert number not in body, f"response leaked {number}"


def test_exploit_member_cannot_read_a_colleagues_invoice_in_the_same_tenant(signed_in):
    assert_hidden(signed_in, "alice", "/invoices/502", "INV-502")


def test_exploit_member_of_another_tenant_cannot_read_the_invoice(signed_in):
    assert_hidden(signed_in, "erin", "/invoices/501", "INV-501")


def test_exploit_billing_admin_cannot_read_an_invoice_in_another_tenant(signed_in):
    assert_hidden(signed_in, "cara", "/invoices/601", "INV-601")


def test_exploit_members_list_leaves_out_colleagues_invoices(signed_in):
    assert listed_ids(signed_in, "alice") == [501, 503]


def test_exploit_billing_admins_list_stays_inside_their_tenant(signed_in):
    assert listed_ids(signed_in, "cara") == [501, 502, 503]


def test_exploit_export_that_includes_a_colleagues_invoice_is_refused_whole(signed_in):
    assert_hidden(signed_in, "alice", "/invoices/export?ids=501,502", "INV-502")


def test_exploit_billing_admins_export_into_another_tenant_is_refused_whole(signed_in):
    assert_hidden(signed_in, "cara", "/invoices/export?ids=502,601", "INV-601")

• invoice.owner_id == user.id or user.role == 'billing_admin' lets a billing admin of any tenant read every tenant's invoices. That is this card's wrong implementation, and every test inside one tenant passes against it.
• session.get(Invoice, invoice_id) and db.get() select by primary key alone. Use select(Invoice).where(Invoice.id == invoice_id, visible_to(user)) in every query; with_loader_criteria() can add the tenant condition to every SELECT of the entity as a second layer.
• Filtering rows in Python after an unscoped SELECT loads other tenants' data and breaks LIMIT, OFFSET and counts. Put the scope in the WHERE clause.
• Overriding current_user itself in tests hands every route a ready-made User and skips the code that loads tenant and role. Override only the authentication step, as this card's tests do.
• A route-level dependency such as dependencies=[Depends(current_user)] proves the caller is signed in, not that they may read this object. The object check belongs in the query.
• FastAPI matches path operations in the order they are declared: /invoices/export after /invoices/{invoice_id} is answered with a 422 for the int conversion. Declare the fixed path first.
• Starlette's SessionMiddleware stores the session as base64 JSON in a cookie signed with itsdangerous: the client can read it but not change it. It sets HttpOnly and SameSite=lax by default but Secure only with https_only=True, and keeps sessions for 14 days unless max_age says otherwise.
• A tenant ID from a header, a subdomain or the URL is only a selector. If you accept one, check that the signed-in user belongs to that tenant before using it.

1. Run the tests in all three modes: IMPL=fixed must pass; IMPL=vulnerable and IMPL=wrong must fail (node ../../run-all.mjs missing-authorization/python-fastapi does this for you).
2. Against your own API, sign in as a member and request a colleague's invoice ID, then as an admin of one tenant request an invoice of another tenant; both must answer 404 without the invoice number.
3. Scanners do not help here: Semgrep's packs and Bandit report nothing on any implementation, so the regression tests are the control.

What each rule reported when it was run on this card's code, or what its source shows it covers where it was not run. A rule that exists for the weakness does not always catch it.

No rule targets object-level authorization in FastAPI. semgrep-rules has one FastAPI rule (python/fastapi/security/wildcard-cors), and in github/codeql main at b5b8165 the only Python query tagged CWE-285, CWE-639, CWE-862 or CWE-863 is PamAuthorization.ql (PAM account checks).

Download the project, install its dependencies once (this step needs the network), then run the tests against each implementation from the missing-authorization/python-fastapifolder.

python3 -m venv .venv
.venv/bin/python -m pip install -r requirements.txt

# Expected: every test passes
IMPL=fixed .venv/bin/python -m pytest

# Expected: an exploit test fails
IMPL=vulnerable .venv/bin/python -m pytest

# Expected: a test fails
IMPL=wrong .venv/bin/python -m pytest

Each run writes a JUnit report to reports/junit.xml. From the top folder of the download, node run-all.mjs missing-authorization/python-fastapi runs all three modes and checks the expected results.

Download the project ZIP, 20 KB

Verified

Machine

macOS 27.0 (Darwin 27.0.0, arm64)

Toolchain

• python CPython 3.14.8
• fastapi 0.142.2
• starlette 1.7.0
• sqlalchemy 2.0.54
• itsdangerous 2.2.0
• pytest 9.1.1
• semgrep 1.180.0
• bandit 1.9.4

Matrix

fixed: 15 of 15 tests pass; vulnerable: 6 of 15 tests fail (6 exploit tests); wrong: 3 of 15 tests fail (3 exploit tests)

Test results

fixed 15/15 pass; vulnerable 6/15 fail (6 exploit tests); wrong 3/15 fail (3 exploit tests: the other tenant's admin), offline after install

Dependency freshness

Same pins as the FastAPI SQL injection card (FastAPI 0.142.2 from 2026-09-30, Starlette 1.7.0 from 2026-09-23, Pydantic 2.13.5, SQLAlchemy 2.0.54, httpx2 2.13.1), plus itsdangerous 2.2.0 from 2024-04-16. The test-only iniconfig is pinned at 2.3.0 because 2.3.1 is from 2026-10-06. OSV reported no known vulnerabilities for any pinned package.

Versions and results hold for the toolchain above. The card's tests are the check: run them with your own versions before relying on the fix.

References

1. OWASP API Security Top 10 2023: API1 Broken Object Level Authorization
2. OWASP Insecure Direct Object Reference Prevention Cheat Sheet
3. OWASP Authorization Cheat Sheet
4. OWASP Multi-Tenant Security Cheat Sheet
5. FastAPI: get the current user (security dependencies)
6. FastAPI: testing dependencies with overrides
7. FastAPI: path parameters, order matters
8. Starlette: SessionMiddleware
9. SQLAlchemy 2.0: Session.get() selects by primary key
10. SQLAlchemy 2.0: with_loader_criteria (adding criteria to every load of an entity)