tests/eval/benchmark_dataset.py

"""
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": [],
    },
]