Authorisation and broken access control

from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()

# Demo data and logins (in a real app: a database and real authentication)
TOKENS = {
    "token-asha": {"id": 1, "role": "customer"},
    "token-ravi": {"id": 2, "role": "customer"},
    "token-staff": {"id": 99, "role": "staff"},
}
ORDERS = {
    41: {"id": 41, "user_id": 1, "phone": "9876500001", "status": "received", "total": 330},
    42: {"id": 42, "user_id": 2, "phone": "9876500002", "status": "ready", "total": 180},
}
STATUSES = {"received", "preparing", "ready", "picked_up", "cancelled"}


def current_user(authorization: str = Header(default="")):
    token = authorization.removeprefix("Bearer ").strip()
    user = TOKENS.get(token)
    if user is None:
        raise HTTPException(status_code=401, detail="Not logged in")
    return user


def require_role(*roles: str):
    def checker(user: dict = Depends(current_user)):
        if user["role"] not in roles:
            raise HTTPException(status_code=403, detail="Not allowed")
        return user
    return checker


# ---- VULNERABLE: any logged-in user can read any order ----
@app.get("/unsafe/orders/{order_id}")
def get_order_unsafe(order_id: int, user: dict = Depends(current_user)):
    order = ORDERS.get(order_id)
    if order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    return order


# ---- FIXED: owner or staff only ----
@app.get("/orders/{order_id}")
def get_order(order_id: int, user: dict = Depends(current_user)):
    order = ORDERS.get(order_id)
    if order is None or (user["role"] != "staff" and order["user_id"] != user["id"]):
        # Same answer for "missing" and "not yours": do not reveal which ids exist
        raise HTTPException(status_code=404, detail="Order not found")
    return order


@app.get("/orders")
def list_my_orders(user: dict = Depends(current_user)):
    if user["role"] == "staff":
        return list(ORDERS.values())
    return [o for o in ORDERS.values() if o["user_id"] == user["id"]]


@app.patch("/orders/{order_id}/status")
def update_status(order_id: int, body: dict, user: dict = Depends(require_role("staff"))):
    order = ORDERS.get(order_id)
    if order is None:
        raise HTTPException(status_code=404, detail="Order not found")
    new_status = body.get("status")
    if new_status not in STATUSES:
        raise HTTPException(status_code=422, detail="Unknown status")
    order["status"] = new_status
    return order

curl -H "Authorization: Bearer token-asha" http://127.0.0.1:8000/unsafe/orders/42

curl -i -H "Authorization: Bearer token-asha" http://127.0.0.1:8000/orders/42
curl -i -H "Authorization: Bearer token-asha" http://127.0.0.1:8000/orders/41
curl -i -H "Authorization: Bearer token-staff" http://127.0.0.1:8000/orders/42

# SQLAlchemy 2.x style sketch
from sqlalchemy import select

stmt = select(Order).where(Order.id == order_id, Order.user_id == user.id)
order = session.execute(stmt).scalar_one_or_none()
if order is None:
    raise HTTPException(status_code=404, detail="Order not found")

curl -i -X PATCH -H "Authorization: Bearer token-asha" -H "Content-Type: application/json" -d "{\"status\":\"picked_up\"}" http://127.0.0.1:8000/orders/41/status

curl -i -X PATCH -H "Authorization: Bearer token-staff" -H "Content-Type: application/json" -d "{\"status\":\"preparing\"}" http://127.0.0.1:8000/orders/41/status

from fastapi.testclient import TestClient
from app import app

client = TestClient(app)


def auth(token):
    return {"Authorization": f"Bearer {token}"}


def test_anonymous_gets_401():
    assert client.get("/orders/41").status_code == 401


def test_owner_can_read_own_order():
    assert client.get("/orders/41", headers=auth("token-asha")).status_code == 200


def test_customer_cannot_read_someone_elses_order():
    assert client.get("/orders/42", headers=auth("token-asha")).status_code == 404


def test_staff_can_read_any_order():
    assert client.get("/orders/42", headers=auth("token-staff")).status_code == 200


def test_customer_cannot_change_status():
    r = client.patch("/orders/41/status", json={"status": "picked_up"}, headers=auth("token-asha"))
    assert r.status_code == 403


def test_staff_can_change_status():
    r = client.patch("/orders/41/status", json={"status": "preparing"}, headers=auth("token-staff"))
    assert r.status_code == 200

from pydantic import BaseModel


class ProfileUpdate(BaseModel):
    # Allow-list: only these fields can ever be changed through this endpoint
    name: str | None = None
    phone: str | None = None
    # No 'role', 'is_staff' or 'user_id' here, so a client cannot send them

You are an application security reviewer. Review these API routes for broken access control: missing authentication, missing role checks (function level), missing ownership checks (object level), mass assignment, and trust in client-supplied ids or roles. For each finding give the route, why it is a problem, and a fixed version of the code. Then suggest pytest tests that log in as different users and check the forbidden cases.

Framework: [e.g. FastAPI / Express]

Routes:
[paste your routes here]