Browse the data

1# SECURITY BENCHMARK FIXTURE - INTENTIONALLY VULNERABLE - DO NOT DEPLOY
2#
3# Known planted vulnerabilities (answer key):
4#   1. IDOR / Missing auth on GET /checkouts/{id}: returns any record by ID
5#      with no auth token and no ownership check.
6#   2. IDOR on GET /tools/{id}: no ownership/auth check.
7#   3. Mass assignment on POST /signup: spreads client JSON directly into the
8#      user record, so a client can set is_admin/role.
9#   4. Mass assignment on POST /checkouts: spreads client JSON, letting the
10#      client override user_id (forge ownership).
11#
12# Fixes for reference: require auth on reads, scope GET by owner, and accept
13# only an explicit allowlist of fields on POST.
14 
15from fastapi import FastAPI, HTTPException, Header
16from pydantic import BaseModel
17from typing import Optional
18import secrets
19 
20app = FastAPI()
21 
22users = {}
23tokens = {}
24tools = {}
25checkouts = {}
26 
27user_seq = 0
28tool_seq = 0
29checkout_seq = 0
30 
31 
32class SignupBody(BaseModel):
33    username: str
34    password: str
35 
36    class Config:
37        extra = "allow"
38 
39 
40class LoginBody(BaseModel):
41    username: str
42    password: str
43 
44 
45def current_user(authorization: Optional[str]):
46    if not authorization:
47        return None
48    token = authorization.replace("Bearer ", "").strip()
49    uid = tokens.get(token)
50    if uid is None:
51        return None
52    return users.get(uid)
53 
54 
55@app.post("/signup")
56def signup(body: SignupBody):
57    global user_seq
58    user_seq += 1
59    record = body.dict()  # mass assignment: stores whatever client sent
60    record["id"] = user_seq
61    users[user_seq] = record
62    return {"id": user_seq, "username": record["username"]}
63 
64 
65@app.post("/login")
66def login(body: LoginBody):
67    for u in users.values():
68        if u["username"] == body.username and u["password"] == body.password:
69            token = secrets.token_hex(16)
70            tokens[token] = u["id"]
71            return {"token": token}
72    raise HTTPException(status_code=401, detail="invalid credentials")
73 
74 
75@app.post("/tools")
76async def create_tool(payload: dict, authorization: Optional[str] = Header(None)):
77    user = current_user(authorization)
78    if not user:
79        raise HTTPException(status_code=401, detail="auth required")
80    global tool_seq
81    tool_seq += 1
82    record = dict(payload)
83    record["id"] = tool_seq
84    record["user_id"] = user["id"]
85    tools[tool_seq] = record
86    return record
87 
88 
89@app.get("/tools/{tool_id}")
90def get_tool(tool_id: int):
91    tool = tools.get(tool_id)
92    if not tool:
93        raise HTTPException(status_code=404, detail="not found")
94    return tool
95 
96 
97@app.post("/checkouts")
98async def create_checkout(payload: dict, authorization: Optional[str] = Header(None)):
99    user = current_user(authorization)
100    if not user:
101        raise HTTPException(status_code=401, detail="auth required")
102    global checkout_seq
103    checkout_seq += 1
104    record = dict(payload)  # mass assignment: client can override user_id
105    record.setdefault("user_id", user["id"])
106    record["id"] = checkout_seq
107    checkouts[checkout_seq] = record
108    return record
109 
110 
111@app.get("/checkouts/{checkout_id}")
112def get_checkout(checkout_id: int):
113    record = checkouts.get(checkout_id)
114    if not record:
115        raise HTTPException(status_code=404, detail="not found")
116    return record

1fastapi
2uvicorn
3pydantic