| TASKS = [ |
| { |
| "task_id": "easy_off_by_one", |
| "difficulty": "easy", |
| "code": "def sum_list(arr): return sum(arr[:-1]) # missing last", |
| "intent": "Sum all elements", |
| "constraints": ["O(n) time"], |
| "expected_issues": ["off-by-one error"], |
| "expected_risk": "low", |
| "canonical_fix": "return sum(arr)" |
| }, |
| { |
| "task_id": "medium_sql_injection", |
| "difficulty": "medium", |
| "code": "def get_user(user_id): query = f'SELECT * FROM users WHERE id={user_id}'; cursor.execute(query); return cursor.fetchone()", |
| "intent": "Fetch user by ID safely", |
| "constraints": ["SQL injection prevention", "parameterized queries"], |
| "expected_issues": ["SQL injection"], |
| "expected_risk": "high", |
| "canonical_fix": "cursor.execute('SELECT * FROM users WHERE id=%s', (user_id,))" |
| }, |
| { |
| "task_id": "hard_async_race", |
| "difficulty": "hard", |
| "code": """import asyncio |
| from openai import OpenAI |
| |
| async def fetch(url, client): |
| try: |
| response = await client.chat.completions.create( |
| model="gpt-3.5-turbo", |
| messages=[{"role": "user", "content": f"Fetch data from: {url}"}], |
| max_tokens=100 |
| ) |
| return {"data": response.choices[0].message.content} |
| except Exception as e: |
| return {"error": str(e)} |
| |
| class DataCache: |
| def __init__(self): |
| self.cache = {} # mutable shared state |
| |
| async def update(self, key, value): |
| self.cache[key] = value # race condition |
| await asyncio.sleep(0.1) |
| return self.cache[key]""", |
| "intent": "Fetch data asynchronously with thread-safe caching", |
| "constraints": ["async safety", "no blocking calls", "thread safety"], |
| "expected_issues": ["missing await", "race condition", "blocking I/O"], |
| "expected_risk": "high", |
| "canonical_fix": """async def fetch(url): |
| async with aiohttp.ClientSession() as session: |
| async with session.get(url) as response: |
| return await response.json() |
| |
| class DataCache: |
| def __init__(self): |
| self.cache = {} |
| self._lock = asyncio.Lock() |
| |
| async def update(self, key, value): |
| async with self._lock: |
| self.cache[key] = value |
| await asyncio.sleep(0.1) |
| return self.cache[key]""" |
| }, |
| { |
| "task_id": "medium_xss_vulnerability", |
| "difficulty": "medium", |
| "code": "def render_user_input(user_input): return f'<div>{user_input}</div>'", |
| "intent": "Safely render user input in HTML", |
| "constraints": ["XSS prevention", "input sanitization"], |
| "expected_issues": ["XSS vulnerability", "unsanitized HTML"], |
| "expected_risk": "high", |
| "canonical_fix": "import html; return f'<div>{html.escape(user_input)}</div>'" |
| }, |
| { |
| "task_id": "easy_none_handling", |
| "difficulty": "easy", |
| "code": "def get_user_name(user): return user.name.upper() if user else ''", |
| "intent": "Get user name safely", |
| "constraints": ["null safety"], |
| "expected_issues": ["None handling", "attribute error"], |
| "expected_risk": "medium", |
| "canonical_fix": "return user.name.upper() if user and hasattr(user, 'name') else ''" |
| }, |
| { |
| "task_id": "medium_logic_bug", |
| "difficulty": "medium", |
| "code": "def find_first_negative(numbers): for i in range(len(numbers)): if numbers[i] < 0: return i; return -1", |
| "intent": "Find index of first negative number", |
| "constraints": ["O(n) time", "correct logic"], |
| "expected_issues": ["logic error", "off-by-one"], |
| "expected_risk": "medium", |
| "canonical_fix": "return next((i for i, num in enumerate(numbers) if num < 0), -1)" |
| } |
| ] |
