""" Benchmark dataset for bug detection evaluation. Each fixture has: - id: unique identifier - description: human-readable summary - clean: if True, no bugs expected (measures false positive rate) - code: source snippet to analyse - expected_bugs: list of keywords that MUST appear in critical/warnings for the detection to count as a true positive Fixture breakdown: Buggy (11): sql_injection, hardcoded_secret, missing_error_handling, infinite_loop_risk, race_condition, toctou_vulnerability, async_await_misuse, missing_auth_decorator, n_plus_one_query, insecure_deserialization, path_traversal Clean (4): clean_function, clean_dataclass, clean_async_handler, clean_auth_check """ BENCHMARK_FIXTURES = [ # ── buggy fixtures ────────────────────────────────────────────────────── { "id": "sql_injection", "description": "Raw string formatting in SQL query", "clean": False, "code": """\ def get_user(username: str): import sqlite3 conn = sqlite3.connect("app.db") query = f"SELECT * FROM users WHERE name = '{username}'" return conn.execute(query).fetchone() """, "expected_bugs": ["sql", "injection", "format"], }, { "id": "hardcoded_secret", "description": "API key hardcoded in source", "clean": False, "code": """\ API_KEY = "sk-abc123supersecretkey9999" def call_api(): import httpx return httpx.get("https://api.example.com", headers={"X-Key": API_KEY}) """, "expected_bugs": ["secret", "hardcoded", "api_key", "key"], }, { "id": "missing_error_handling", "description": "File open with no exception handling", "clean": False, "code": """\ def read_config(path: str) -> dict: with open(path) as f: import json return json.load(f) """, "expected_bugs": ["error", "exception", "handling"], }, { "id": "infinite_loop_risk", "description": "While loop with no guaranteed exit and no sleep", "clean": False, "code": """\ def wait_for_ready(service): while not service.is_ready(): pass # no timeout, no sleep return True """, "expected_bugs": ["loop", "timeout", "infinite", "blocking"], }, { "id": "race_condition", "description": "Shared mutable counter incremented without a lock", "clean": False, "code": """\ import threading counter = 0 def increment(): global counter for _ in range(100_000): counter += 1 # read-modify-write is not atomic threads = [threading.Thread(target=increment) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() """, "expected_bugs": ["race", "lock", "thread", "atomic", "concurrent"], }, { "id": "toctou_vulnerability", "description": "Time-of-check to time-of-use: file existence checked then opened separately", "clean": False, "code": """\ import os def read_if_exists(path: str) -> str: if os.path.exists(path): # check with open(path) as f: # use — file may have changed between check and open return f.read() return "" """, "expected_bugs": ["toctou", "race", "time-of-check", "exists", "atomic"], }, { "id": "async_await_misuse", "description": "Blocking I/O call inside an async function starves the event loop", "clean": False, "code": """\ import asyncio import time import httpx async def fetch_data(url: str) -> str: time.sleep(5) # blocks the entire event loop response = httpx.get(url) # sync HTTP call inside async context return response.text """, "expected_bugs": ["blocking", "sleep", "async", "event loop", "sync"], }, { "id": "missing_auth_decorator", "description": "Admin endpoint with no authentication or authorisation check", "clean": False, "code": """\ from fastapi import APIRouter router = APIRouter() @router.delete("/admin/users/{user_id}") async def delete_user(user_id: int): # No auth check — any caller can delete any user return {"deleted": user_id} """, "expected_bugs": ["auth", "authentication", "authorization", "permission", "security"], }, { "id": "n_plus_one_query", "description": "N+1 query: separate DB hit for every item in a loop", "clean": False, "code": """\ from app.db import session from app.models import Order, Product def get_order_products(order_ids: list[int]) -> list[dict]: results = [] for order_id in order_ids: # N iterations order = session.query(Order).get(order_id) # query 1 product = session.query(Product).get(order.product_id) # query 2 results.append({"order": order.id, "product": product.name}) return results """, "expected_bugs": ["n+1", "query", "loop", "database", "join", "eager"], }, { "id": "insecure_deserialization", "description": "pickle.loads() called on untrusted user input", "clean": False, "code": """\ import pickle from flask import request def load_session(): data = request.cookies.get("session") if data: return pickle.loads(bytes.fromhex(data)) # arbitrary code execution return {} """, "expected_bugs": ["pickle", "deserializ", "untrusted", "arbitrary", "code execution"], }, { "id": "path_traversal", "description": "User-controlled filename used in file open without sanitisation", "clean": False, "code": """\ import os from flask import request, send_file UPLOAD_DIR = "/var/uploads" def download_file(): filename = request.args.get("file") # user controlled path = os.path.join(UPLOAD_DIR, filename) # ../../../etc/passwd works return send_file(path) """, "expected_bugs": ["path", "traversal", "sanitiz", "directory", "../"], }, # ── clean fixtures ────────────────────────────────────────────────────── { "id": "clean_function", "description": "Well-typed, documented, no issues", "clean": True, "code": """\ from typing import Optional def divide(a: float, b: float) -> Optional[float]: \"\"\"Safely divide a by b, returning None on division by zero.\"\"\" if b == 0: return None return a / b """, "expected_bugs": [], }, { "id": "clean_dataclass", "description": "Clean dataclass with validation", "clean": True, "code": """\ from dataclasses import dataclass, field from typing import List @dataclass class Order: id: str items: List[str] = field(default_factory=list) total: float = 0.0 def add_item(self, item: str, price: float) -> None: self.items.append(item) self.total += price """, "expected_bugs": [], }, { "id": "clean_async_handler", "description": "Properly async handler using httpx.AsyncClient with timeout", "clean": True, "code": """\ import httpx async def fetch_data(url: str, timeout: float = 10.0) -> str: \"\"\"Fetch URL asynchronously with explicit timeout.\"\"\" async with httpx.AsyncClient(timeout=timeout) as client: response = await client.get(url) response.raise_for_status() return response.text """, "expected_bugs": [], }, { "id": "clean_auth_check", "description": "FastAPI endpoint with proper JWT auth dependency", "clean": True, "code": """\ from fastapi import APIRouter, Depends, HTTPException, status from app.auth import get_current_user, User router = APIRouter() @router.delete("/admin/users/{user_id}") async def delete_user( user_id: int, current_user: User = Depends(get_current_user), ) -> dict: \"\"\"Delete a user. Requires authenticated admin.\"\"\" if not current_user.is_admin: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admins only") return {"deleted": user_id} """, "expected_bugs": [], }, ]