| EASY_SAMPLE_1 = { |
| "filename": "user_service.py", |
| "language": "python", |
| "content": """\ |
| import hashlib |
| from typing import List |
| |
| |
| class UserService: |
| def __init__(self, users: dict): |
| self.users = users |
| |
| def get_user_name(self, user_id: int) -> str: |
| # BUG-1 (line 10): KeyError not handled |
| return self.users[user_id]["name"] |
| |
| def get_active_users(self, users: List[dict]) -> List[dict]: |
| result = [] |
| # BUG-2 (line 15): off-by-one, skips last user |
| for i in range(len(users) - 1): |
| if users[i].get("active"): |
| result.append(users[i]) |
| return result |
| |
| def compute_checksum(self, data: str) -> str: |
| # BUG-3 (line 22): MD5 is cryptographically broken |
| return hashlib.md5(data.encode()).hexdigest() |
| |
| def update_email(self, user_id: int, new_email: str) -> bool: |
| user = self.users.get(user_id) |
| if user: |
| user["email"] == new_email # BUG-4 (line 27): == instead of = |
| return True |
| return False |
| |
| def calculate_average_age(self, user_ids: List[int]) -> float: |
| total = 0 |
| for uid in user_ids: |
| total += self.users[uid]["age"] |
| # BUG-5 (line 34): ZeroDivisionError if user_ids is empty |
| return total / len(user_ids) |
| """, |
| "ground_truth_bugs": [ |
| {"line": 10, "type": "missing_error_handling", "severity": "high", |
| "description": "KeyError not caught; user_id may not exist"}, |
| {"line": 15, "type": "off_by_one", "severity": "medium", |
| "description": "range(len(users)-1) skips the last element"}, |
| {"line": 22, "type": "weak_cryptography", "severity": "high", |
| "description": "MD5 is broken; use hashlib.sha256"}, |
| {"line": 27, "type": "assignment_operator", "severity": "high", |
| "description": "== used instead of =; update silently ignored"}, |
| {"line": 34, "type": "zero_division", "severity": "medium", |
| "description": "ZeroDivisionError when user_ids is empty"}, |
| ], |
| } |
|
|
| EASY_SAMPLE_2 = { |
| "filename": "order_processor.py", |
| "language": "python", |
| "content": """\ |
| from datetime import datetime |
| from typing import List, Dict, Any |
| |
| DISCOUNT_RATES = {"gold": 0.20, "silver": 0.10, "bronze": 0.05} |
| |
| |
| def apply_discount(price: float, tier: str) -> float: |
| # BUG-1 (line 8): KeyError if tier not in DISCOUNT_RATES |
| discount = DISCOUNT_RATES[tier] |
| return price * (1 - discount) |
| |
| |
| def process_orders(orders: List[Dict[str, Any]]) -> List[Dict]: |
| results = [] |
| for order in orders: |
| # BUG-2 (line 15): mutates original dict in-place |
| order["processed"] = True |
| order["timestamp"] = datetime.now().isoformat() |
| results.append(order) |
| return results |
| |
| |
| def find_duplicate_orders(orders: List[Dict]) -> List[str]: |
| seen = [] |
| duplicates = [] |
| for order in orders: |
| oid = order["order_id"] |
| # BUG-3 (line 25): O(n^2) list scan; should be a set |
| if oid in seen: |
| duplicates.append(oid) |
| seen.append(oid) |
| return duplicates |
| |
| |
| def calculate_total(items: List[Dict]) -> float: |
| # BUG-4 (line 32): no guard for empty list |
| return sum(i["price"] * i["quantity"] for i in items) / len(items) |
| |
| |
| def format_currency(amount: float) -> str: |
| # BUG-5 (line 36): float imprecision for financial data; use Decimal |
| return f"${amount:.2f}" |
| """, |
| "ground_truth_bugs": [ |
| {"line": 8, "type": "missing_error_handling", "severity": "high", |
| "description": "KeyError if tier not in DISCOUNT_RATES"}, |
| {"line": 15, "type": "mutation_side_effect", "severity": "medium", |
| "description": "Mutates original dict instead of copying"}, |
| {"line": 25, "type": "performance", "severity": "low", |
| "description": "O(n^2) list scan; use a set for O(1) lookup"}, |
| {"line": 32, "type": "zero_division", "severity": "medium", |
| "description": "ZeroDivisionError when items is empty"}, |
| {"line": 36, "type": "financial_precision", "severity": "medium", |
| "description": "Float imprecision for financial data; use Decimal"}, |
| ], |
| } |
|
|
| MEDIUM_SAMPLE_1 = { |
| "filename": "auth_api.py", |
| "language": "python", |
| "content": """\ |
| import sqlite3 |
| from flask import Flask, request, jsonify, session |
| |
| app = Flask(__name__) |
| # VULN-1 (line 5): hardcoded secret key |
| app.secret_key = "super_secret_key_123" |
| |
| # VULN-2 (line 8): hardcoded DB password |
| DB_PASSWORD = "admin123" |
| |
| |
| def get_db(): |
| return sqlite3.connect("users.db") |
| |
| |
| @app.route("/login", methods=["POST"]) |
| def login(): |
| username = request.form.get("username") |
| password = request.form.get("password") |
| db = get_db() |
| # VULN-3 (line 20): SQL Injection via f-string |
| query = f"SELECT * FROM users WHERE username='{username}' AND password='{password}'" |
| user = db.execute(query).fetchone() |
| if user: |
| session["user_id"] = user[0] |
| return jsonify({"status": "ok"}) |
| return jsonify({"status": "invalid"}), 401 |
| |
| |
| @app.route("/search", methods=["GET"]) |
| def search(): |
| term = request.args.get("q", "") |
| # VULN-4 (line 30): Reflected XSS |
| return f"<html><body><h1>Results for: {term}</h1></body></html>" |
| |
| |
| @app.route("/profile", methods=["GET"]) |
| def profile(): |
| user_id = request.args.get("user_id") |
| # VULN-5 (line 36): IDOR - no authorization check |
| db = get_db() |
| user = db.execute(f"SELECT * FROM users WHERE id={user_id}").fetchone() |
| return jsonify({"user": user}) |
| |
| |
| @app.route("/upload", methods=["POST"]) |
| def upload(): |
| file = request.files["file"] |
| # VULN-6 (line 43): Unrestricted file upload |
| file.save(f"/uploads/{file.filename}") |
| return jsonify({"status": "uploaded"}) |
| |
| |
| if __name__ == "__main__": |
| # VULN-7 (line 48): Debug mode in production |
| app.run(debug=True, host="0.0.0.0") |
| """, |
| "ground_truth_vulnerabilities": [ |
| {"line": 5, "category": "A07_Identification_Failures", |
| "owasp": "A07:2021", "severity": "critical", |
| "description": "Hardcoded secret key in source"}, |
| {"line": 8, "category": "A02_Cryptographic_Failures", |
| "owasp": "A02:2021", "severity": "critical", |
| "description": "Hardcoded DB password"}, |
| {"line": 20, "category": "A03_Injection", |
| "owasp": "A03:2021", "severity": "critical", |
| "description": "SQL Injection via f-string interpolation"}, |
| {"line": 30, "category": "A03_Injection", |
| "owasp": "A03:2021", "severity": "high", |
| "description": "Reflected XSS — unescaped user input in HTML"}, |
| {"line": 36, "category": "A01_Broken_Access_Control", |
| "owasp": "A01:2021", "severity": "high", |
| "description": "IDOR — no authorization before profile access"}, |
| {"line": 43, "category": "A04_Insecure_Design", |
| "owasp": "A04:2021", "severity": "high", |
| "description": "Unrestricted file upload, no type check"}, |
| {"line": 48, "category": "A05_Security_Misconfiguration", |
| "owasp": "A05:2021", "severity": "medium", |
| "description": "Flask debug mode exposes Werkzeug debugger"}, |
| ], |
| } |
|
|
| HARD_SAMPLE_1 = { |
| "files": [ |
| { |
| "filename": "services/payment_service.py", |
| "language": "python", |
| "content": """\ |
| import requests |
| |
| PAYMENT_API_URL = "https://api.payments.io/v1" |
| API_KEY = "pk_live_abc123xyz" # hardcoded production key |
| |
| |
| def charge_card(card_token: str, amount: int, currency: str = "USD"): |
| # No retry, no timeout |
| response = requests.post( |
| f"{PAYMENT_API_URL}/charges", |
| json={"token": card_token, "amount": amount, "currency": currency}, |
| headers={"Authorization": f"Bearer {API_KEY}"}, |
| ) |
| if response.status_code == 200: |
| return response.json() |
| raise Exception(f"Payment failed: {response.text}") |
| |
| |
| def refund_charge(charge_id: str, amount: int): |
| # No idempotency key |
| response = requests.post( |
| f"{PAYMENT_API_URL}/refunds", |
| json={"charge_id": charge_id, "amount": amount}, |
| headers={"Authorization": f"Bearer {API_KEY}"}, |
| ) |
| return response.json() |
| """, |
| }, |
| { |
| "filename": "services/inventory_service.py", |
| "language": "python", |
| "content": """\ |
| from typing import Dict, List |
| |
| # In-memory store — lost on restart, not thread-safe |
| _inventory: Dict[str, int] = {} |
| |
| |
| def reserve_item(item_id: str, quantity: int) -> bool: |
| # Race condition: check-then-act not atomic |
| current = _inventory.get(item_id, 0) |
| if current < quantity: |
| return False |
| _inventory[item_id] = current - quantity |
| return True |
| |
| |
| def restock(item_id: str, quantity: int) -> None: |
| # No validation — accepts negative quantity |
| _inventory[item_id] = _inventory.get(item_id, 0) + quantity |
| |
| |
| def get_low_stock(threshold: int = 10) -> List[str]: |
| # No pagination — could return millions of items |
| return [k for k, v in _inventory.items() if v < threshold] |
| |
| |
| def bulk_reserve(items: List[Dict]) -> Dict[str, bool]: |
| # Not atomic — partial reservation possible |
| results = {} |
| for item in items: |
| results[item["id"]] = reserve_item(item["id"], item["qty"]) |
| return results |
| """, |
| }, |
| { |
| "filename": "api/router.py", |
| "language": "python", |
| "content": """\ |
| from fastapi import FastAPI, HTTPException |
| from services import payment_service, inventory_service |
| |
| app = FastAPI() |
| print("Starting API server") # mixing print with logger |
| |
| |
| @app.post("/orders") |
| def create_order(order: dict): # no Pydantic schema |
| item_id = order["item_id"] |
| quantity = order["quantity"] |
| card_token = order["card_token"] |
| amount = order["amount"] |
| |
| reserved = inventory_service.reserve_item(item_id, quantity) |
| if not reserved: |
| raise HTTPException(status_code=409, detail="Out of stock") |
| |
| try: |
| result = payment_service.charge_card(card_token, amount) |
| return {"status": "ok", "charge": result} |
| except Exception as e: |
| # Inventory NOT released on payment failure — leaked reservation |
| # Raw exception message returned to client |
| raise HTTPException(status_code=500, detail=str(e)) |
| |
| |
| @app.get("/health") |
| def health(): |
| # No dependency checks |
| return {"status": "ok"} |
| """, |
| }, |
| { |
| "filename": "config/settings.py", |
| "language": "python", |
| "content": """\ |
| # All config hardcoded — no env vars |
| DATABASE_URL = "postgresql://admin:password123@localhost:5432/appdb" |
| REDIS_URL = "redis://localhost:6379" |
| JWT_SECRET = "my_jwt_secret_do_not_share" |
| LOG_LEVEL = "DEBUG" # leaks sensitive data in production |
| PAYMENT_API_KEY = "pk_live_abc123xyz" |
| |
| # No pydantic BaseSettings, no dev/staging/prod distinction |
| """, |
| }, |
| ], |
| "ground_truth_issues": { |
| "security": [ |
| "Hardcoded production API key in payment_service.py", |
| "Hardcoded database credentials in settings.py", |
| "JWT secret committed to source code", |
| "DEBUG log level leaks sensitive data in production", |
| ], |
| "reliability": [ |
| "No retry logic on payment API calls", |
| "No request timeout on external HTTP calls", |
| "Race condition in inventory reserve_item", |
| "Partial reservation in bulk_reserve — no atomicity", |
| "Leaked inventory reservation on payment failure", |
| ], |
| "api_design": [ |
| "Generic dict input in create_order — no Pydantic schema", |
| "Raw exception message returned to client", |
| "Health endpoint has no dependency checks", |
| "No idempotency key in refund_charge", |
| ], |
| "observability": [ |
| "print() mixed with logger — inconsistent logging", |
| "No distributed tracing or correlation IDs", |
| "No structured JSON log format", |
| ], |
| "scalability": [ |
| "In-memory inventory store — not distributed-safe", |
| "get_low_stock has no pagination", |
| "No connection pooling configured", |
| ], |
| }, |
| } |
|
|
| EASY_SAMPLES = [EASY_SAMPLE_1, EASY_SAMPLE_2] |
| MEDIUM_SAMPLES = [MEDIUM_SAMPLE_1] |
| HARD_SAMPLES = [HARD_SAMPLE_1] |
|
|
