Authorisation and Safety Checkpoint
26 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)Fix the mistakes
The slips that cost marks in this module. Spot each one, then check the fix.
A safe API stacks several small checks: a rate limit, a login, a role, an owner filter and a strict input model. This checkpoint builds that stack on an orders API and tests the whole module.
Stack the gates in the right order
Each check has one job and its own status code. Cheap checks that protect the server go first: the rate limit (429). Then identity (401), then the role (403), then the record (404) and finally the input model (422). A visitor who fails an early gate learns nothing about the later ones.
app = FastAPI(dependencies=[Depends(rate_limit)])
@app.delete("/orders/{order_id}")def remove(order_id: int, user=Depends(admin_only)): ...Example 1 · Which gate answers first?
import sqlite3from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Requestfrom fastapi.security import OAuth2PasswordBearerfrom fastapi.testclient import TestClientfrom pydantic import BaseModel, Field
class FakeClock: def __init__(self): self.now = 0.0
def __call__(self): return self.now
clock = FakeClock()hits = {}
def rate_limit(request: Request): key = request.client.host window = int(clock() // 10) start, count = hits.get(key, (window, 0)) if start != window: start, count = window, 0 if count >= 5: raise HTTPException(429, "Too many requests", headers={"Retry-After": "10"}) hits[key] = (start, count + 1)
db = sqlite3.connect(":memory:", check_same_thread=False)db.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY AUTOINCREMENT, owner_id INTEGER, item TEXT, qty INTEGER)")db.execute("INSERT INTO orders (owner_id, item, qty) VALUES (2, 'lamp', 1)")
TOKENS = {"t-asha": {"id": 1, "role": "customer"}, "t-meena": {"id": 3, "role": "admin"}}app = FastAPI(dependencies=[Depends(rate_limit)])oauth2 = OAuth2PasswordBearer(tokenUrl="token")
def current_user(token: Annotated[str, Depends(oauth2)]): user = TOKENS.get(token) if user is None: raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"}) return user
def admin_only(user: Annotated[dict, Depends(current_user)]): if user["role"] != "admin": raise HTTPException(403, "Admins only") return user
class NewOrder(BaseModel): item: str = Field(min_length=1, max_length=40) qty: int = Field(ge=1, le=10)
@app.post("/orders", status_code=201)def create(body: NewOrder, user: Annotated[dict, Depends(current_user)]): cur = db.execute("INSERT INTO orders (owner_id, item, qty) VALUES (?, ?, ?)", (user["id"], body.item, body.qty)) return {"id": cur.lastrowid, "item": body.item, "qty": body.qty}
@app.get("/orders/{order_id}")def read(order_id: int, user: Annotated[dict, Depends(current_user)]): row = db.execute("SELECT id, item, qty FROM orders WHERE id = ? AND owner_id = ?", (order_id, user["id"])).fetchone() if row is None: raise HTTPException(404, "Order not found") return {"id": row[0], "item": row[1], "qty": row[2]}
@app.delete("/orders/{order_id}")def remove(order_id: int, user: Annotated[dict, Depends(admin_only)]): cur = db.execute("DELETE FROM orders WHERE id = ?", (order_id,)) return {"deleted": cur.rowcount}
client = TestClient(app)def h(name): return {"Authorization": "Bearer " + name}
print(client.post("/orders", json={"item": "pen", "qty": 99}, headers=h("t-asha")).status_code)print(client.get("/orders/1").status_code)for _ in range(5): r = client.get("/orders/1")print(r.status_code, r.headers.get("retry-after"))clock.now = 10.0print(client.get("/orders/1", headers=h("t-asha")).status_code)Output
422 401 429 10 404
What this code does
- An out-of-range quantity gives 422. A missing token gives 401.
- After five requests in one window, the next one is refused with 429 before any login check. In the next time slot the request passes the gates again and reaches the owner check, which answers 404.
A dependency on the whole app runs for every route. The role dependency itself depends on current_user, so 401 comes before 403.
Own the record, name the fields
Two flaws remain even after login: reading another user's record, and sending fields you should not control. The owner comes from the token and goes into the query. The fields come from a small model, never from the raw body.
class NewOrder(BaseModel): item: str = Field(min_length=1, max_length=40) qty: int = Field(ge=1, le=10)
row = db.execute("SELECT id FROM orders WHERE id = ? AND owner_id = ?", (order_id, user["id"])).fetchone()Example 2 · An orders API with four gates
import sqlite3from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Requestfrom fastapi.security import OAuth2PasswordBearerfrom fastapi.testclient import TestClientfrom pydantic import BaseModel, Field
class FakeClock: def __init__(self): self.now = 0.0
def __call__(self): return self.now
clock = FakeClock()hits = {}
def rate_limit(request: Request): key = request.client.host window = int(clock() // 10) start, count = hits.get(key, (window, 0)) if start != window: start, count = window, 0 if count >= 5: raise HTTPException(429, "Too many requests", headers={"Retry-After": "10"}) hits[key] = (start, count + 1)
db = sqlite3.connect(":memory:", check_same_thread=False)db.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY AUTOINCREMENT, owner_id INTEGER, item TEXT, qty INTEGER)")db.execute("INSERT INTO orders (owner_id, item, qty) VALUES (2, 'lamp', 1)")
TOKENS = {"t-asha": {"id": 1, "role": "customer"}, "t-meena": {"id": 3, "role": "admin"}}app = FastAPI(dependencies=[Depends(rate_limit)])oauth2 = OAuth2PasswordBearer(tokenUrl="token")
def current_user(token: Annotated[str, Depends(oauth2)]): user = TOKENS.get(token) if user is None: raise HTTPException(401, "Not authenticated", headers={"WWW-Authenticate": "Bearer"}) return user
def admin_only(user: Annotated[dict, Depends(current_user)]): if user["role"] != "admin": raise HTTPException(403, "Admins only") return user
class NewOrder(BaseModel): item: str = Field(min_length=1, max_length=40) qty: int = Field(ge=1, le=10)
@app.post("/orders", status_code=201)def create(body: NewOrder, user: Annotated[dict, Depends(current_user)]): cur = db.execute("INSERT INTO orders (owner_id, item, qty) VALUES (?, ?, ?)", (user["id"], body.item, body.qty)) return {"id": cur.lastrowid, "item": body.item, "qty": body.qty}
@app.get("/orders/{order_id}")def read(order_id: int, user: Annotated[dict, Depends(current_user)]): row = db.execute("SELECT id, item, qty FROM orders WHERE id = ? AND owner_id = ?", (order_id, user["id"])).fetchone() if row is None: raise HTTPException(404, "Order not found") return {"id": row[0], "item": row[1], "qty": row[2]}
@app.delete("/orders/{order_id}")def remove(order_id: int, user: Annotated[dict, Depends(admin_only)]): cur = db.execute("DELETE FROM orders WHERE id = ?", (order_id,)) return {"deleted": cur.rowcount}
client = TestClient(app)def h(name): return {"Authorization": "Bearer " + name}
r = client.post("/orders", json={"item": "pen", "qty": 2, "owner_id": 2}, headers=h("t-asha"))print(r.status_code, r.json())print(client.get("/orders/1", headers=h("t-asha")).status_code)print(client.get("/orders/2", headers=h("t-asha")).status_code)print(client.delete("/orders/2", headers=h("t-asha")).status_code)print(client.delete("/orders/2", headers=h("t-meena")).json())Output
201 {'id': 2, 'item': 'pen', 'qty': 2}
404
200
403
{'deleted': 1}What this code does
- The extra
owner_idin the body is ignored: the order is created for the caller. Order 1 belongs to someone else, so asha gets 404 on it, and 200 on her own order 2. - A customer cannot delete (403). The admin can, and the response shows one deleted row.
A request that tries to set owner_id has that field ignored. The new order always belongs to the caller. Limits such as ge=1, le=10 stop absurd values.
Review with a checklist
Read code the way a reviewer reads it. Look for SQL built from text, user text printed as HTML, keys written in code, whole request bodies copied into records, ids used without an owner check, CORS with a wildcard and credentials, and login routes without a limit.
origins = os.environ["ALLOWED_ORIGINS"].split(",")app.add_middleware(CORSMiddleware, allow_origins=origins, allow_credentials=True)Example 3 · A tiny code review script
import re
code_lines = [ 'db.execute(f"SELECT * FROM users WHERE name = \'{name}\'")', 'db.execute("SELECT * FROM users WHERE name = ?", (name,))', 'SECRET_KEY = "abc123"', 'SECRET_KEY = os.environ["APP_SECRET_KEY"]', 'user.update(**request_body)',]
checks = [ ("SQL built with an f-string", re.compile(r"execute\(f[\"']")), ("secret in code", re.compile(r"(SECRET|PASSWORD|TOKEN)\w*\s*=\s*[\"']")), ("mass assignment", re.compile(r"\*\*\w*body")),]
for number, line in enumerate(code_lines, start=1): for label, pattern in checks: if pattern.search(line): print(number, label)Output
1 SQL built with an f-string 3 secret in code 5 mass assignment
What this code does
- Three simple patterns find three of the mistakes in this list. Safe lines print nothing.
- Real teams use linters and scanners such as bandit for the same job, and a human review for the rest.
This setting is safe: a fixed list from the environment, with credentials. The same line with ["*"] would not be.
Finish up
Mark this lesson complete
This is the last lesson in this section.