Spaces:
Sleeping
Sleeping
| """ | |
| vulnerability_hunter_generator.py | |
| ================================== | |
| Generates fully-realized Vulnerability Hunter challenges for the CyberArena | |
| Blue Team pool. | |
| A "Vulnerability Hunter" challenge gives the trainee a small vulnerable code | |
| sample and asks them to IDENTIFY the vulnerability (canonical key like | |
| "sql-injection", "xss", "command-injection", ...) — not to fix the code. | |
| This is the "defender triage" path of the Blue Team curriculum. | |
| Architecture mirrors `code_fixing_generator.py`: | |
| main.py watcher (polls vulnerability_hunter_challenges count) | |
| | | |
| v | |
| refill_pool(team_role, count) --> ai_generate_challenge() | |
| | | | |
| | v | |
| | AI generates vulnerable code | |
| | | | |
| | v | |
| | ChallengeBuilder produces DB row | |
| v | |
| insert_to_db() <----- Challenge row | |
| | | |
| v | |
| public.vulnerability_hunter_challenges | |
| Difficulty uses 5 levels (Beginner / Easy / Medium / Hard / Expert), mapped | |
| to Arabic values already in the platform: مبتدئ / سهل / متوسط / صعب / خبير. | |
| Public API (used by main.py): | |
| - POOL_TARGET = 5 | |
| - POOL_THRESHOLD = 2 | |
| - POOL_BATCH = 3 | |
| - get_pool_count(team_role) -> int | |
| - async refill_pool(team_role, count) -> int | |
| - async start_pool_watcher(team_role) | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import asyncio | |
| import json | |
| import os | |
| import random | |
| import re | |
| import sys | |
| import time | |
| import uuid | |
| from typing import Optional | |
| # Load .env early so the CLI works without manual export | |
| try: | |
| from dotenv import load_dotenv | |
| _env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") | |
| if os.path.exists(_env_path): | |
| load_dotenv(_env_path) | |
| except ImportError: | |
| pass | |
| import httpx | |
| # --------------------------------------------------------------------------- # | |
| # 0. Constants | |
| # --------------------------------------------------------------------------- # | |
| ALLOWED_LANGUAGES = ("C++", "JAVA", "PYTHON", "JAVASCRIPT", "PHP", "RUST", "GO", "CSHARP", "RUBY", "CSHARP") | |
| ALLOWED_DIFFICULTIES = ("مبتدئ", "سهل", "متوسط", "صعب", "خبير") | |
| ALLOWED_TEAMS = ("blue",) | |
| # Canonical vulnerability types — what the student must type. Order matters | |
| # for the rotation. Deliberately broad: covers web, crypto, memory safety, | |
| # concurrency, and language-specific pitfalls across C/C++/Java/Python/JS/PHP/ | |
| # Rust/Go/C#/Ruby. | |
| VULNERABILITY_TYPES = [ | |
| # web / app | |
| "sql-injection", "xss", "command-injection", "path-traversal", | |
| "csrf", "ssrf", "xxe", "idor", "open-redirect", | |
| "insecure-deserialization", "broken-authentication", | |
| "security-misconfiguration", "sensitive-data-exposure", | |
| "unrestricted-file-upload", "prototype-pollution", | |
| "mass-assignment", "ssr-rce", "file-inclusion", | |
| # crypto / secrets | |
| "weak-hashing", "weak-randomness", "hardcoded-credentials", | |
| "insecure-key-exchange", "ecb-mode", "padding-oracle", | |
| # memory safety (C/C++/Rust) | |
| "buffer-overflow", "use-after-free", "format-string", | |
| "integer-overflow", "null-pointer-dereference", | |
| "double-free", "off-by-one", "uninitialized-memory", | |
| # concurrency / logic | |
| "race-condition", "time-of-check-time-of-use", | |
| "unsafe-reflection", "trust-boundary-violation", | |
| ] | |
| # OWASP-family / class labels (used in hints + UI). | |
| VULNERABILITY_CLASS = { | |
| "sql-injection": "Injection", | |
| "xss": "Injection (XSS)", | |
| "command-injection": "Injection", | |
| "path-traversal": "Path Traversal", | |
| "csrf": "Cross-Site Request Forgery", | |
| "ssrf": "Server-Side Request Forgery", | |
| "xxe": "XML External Entity", | |
| "xml-external-entity": "XML External Entity", | |
| "idor": "Broken Access Control", | |
| "insecure-direct-object-reference": "Broken Access Control", | |
| "open-redirect": "Unvalidated Redirects", | |
| "ssr-rce": "Server-Side Template Injection", | |
| "file-inclusion": "File Inclusion", | |
| "insecure-deserialization": "Insecure Deserialization", | |
| "unsafe-deserialization": "Insecure Deserialization", | |
| "broken-authentication": "Broken Authentication", | |
| "security-misconfiguration": "Security Misconfiguration", | |
| "sensitive-data-exposure": "Cryptographic Failures", | |
| "unrestricted-file-upload": "File Upload", | |
| "mass-assignment": "Mass Assignment", | |
| "weak-hashing": "Cryptographic Failures", | |
| "weak-randomness": "Cryptographic Failures", | |
| "hardcoded-credentials": "Hardcoded Secrets", | |
| "insecure-key-exchange": "Cryptographic Failures", | |
| "ecb-mode": "Cryptographic Failures", | |
| "padding-oracle": "Cryptographic Failures", | |
| "buffer-overflow": "Memory Safety", | |
| "use-after-free": "Memory Safety", | |
| "format-string": "Memory Safety", | |
| "integer-overflow": "Memory Safety", | |
| "null-pointer-dereference": "Memory Safety", | |
| "double-free": "Memory Safety", | |
| "off-by-one": "Memory Safety", | |
| "uninitialized-memory": "Memory Safety", | |
| "race-condition": "Concurrency", | |
| "time-of-check-time-of-use": "Concurrency", | |
| "unsafe-reflection": "Input Validation", | |
| "trust-boundary-violation": "Input Validation", | |
| "prototype-pollution": "Prototype Pollution", | |
| } | |
| # Module names for each vulnerability (matches the existing /paths/modules). | |
| # Web flaws land under web-security; crypto under cryptography; memory-safety / | |
| # concurrency / language-specific bugs under secure-coding (a generic catch-all | |
| # module for non-web code-review exercises). | |
| MODULE_BY_VULN = { | |
| "sql-injection": "web-security", | |
| "xss": "web-security", | |
| "command-injection": "web-security", | |
| "path-traversal": "web-security", | |
| "csrf": "web-security", | |
| "ssrf": "web-security", | |
| "xxe": "web-security", | |
| "xml-external-entity": "web-security", | |
| "idor": "web-security", | |
| "insecure-direct-object-reference": "web-security", | |
| "open-redirect": "web-security", | |
| "ssr-rce": "web-security", | |
| "file-inclusion": "web-security", | |
| "insecure-deserialization": "web-security", | |
| "unsafe-deserialization": "web-security", | |
| "broken-authentication": "web-security", | |
| "security-misconfiguration": "web-security", | |
| "sensitive-data-exposure": "web-security", | |
| "unrestricted-file-upload": "web-security", | |
| "mass-assignment": "web-security", | |
| "prototype-pollution": "web-security", | |
| "weak-hashing": "cryptography", | |
| "weak-randomness": "cryptography", | |
| "hardcoded-credentials": "secure-coding", | |
| "insecure-key-exchange": "cryptography", | |
| "ecb-mode": "cryptography", | |
| "padding-oracle": "cryptography", | |
| "buffer-overflow": "secure-coding", | |
| "use-after-free": "secure-coding", | |
| "format-string": "secure-coding", | |
| "integer-overflow": "secure-coding", | |
| "null-pointer-dereference": "secure-coding", | |
| "double-free": "secure-coding", | |
| "off-by-one": "secure-coding", | |
| "uninitialized-memory": "secure-coding", | |
| "race-condition": "secure-coding", | |
| "time-of-check-time-of-use": "secure-coding", | |
| "unsafe-reflection": "secure-coding", | |
| "trust-boundary-violation": "secure-coding", | |
| } | |
| # Pick a sensible default language per vuln type (the AI is told the language, | |
| # not asked to choose). This list is what rotates through the seed pool. | |
| LANG_BY_VULN = { | |
| "sql-injection": "PYTHON", | |
| "xss": "JAVASCRIPT", | |
| "command-injection": "PYTHON", | |
| "path-traversal": "PYTHON", | |
| "csrf": "JAVASCRIPT", | |
| "ssrf": "PYTHON", | |
| "xxe": "JAVA", | |
| "xml-external-entity": "JAVA", | |
| "idor": "JAVASCRIPT", | |
| "insecure-direct-object-reference": "JAVASCRIPT", | |
| "open-redirect": "JAVASCRIPT", | |
| "ssr-rce": "JAVASCRIPT", | |
| "file-inclusion": "PHP", | |
| "insecure-deserialization": "PYTHON", | |
| "unsafe-deserialization": "PYTHON", | |
| "broken-authentication": "JAVASCRIPT", | |
| "security-misconfiguration": "PYTHON", | |
| "sensitive-data-exposure": "PYTHON", | |
| "unrestricted-file-upload": "PHP", | |
| "mass-assignment": "RUBY", | |
| "prototype-pollution": "JAVASCRIPT", | |
| "weak-hashing": "PYTHON", | |
| "weak-randomness": "JAVA", | |
| "hardcoded-credentials": "CSHARP", | |
| "insecure-key-exchange": "PYTHON", | |
| "ecb-mode": "PYTHON", | |
| "padding-oracle": "PYTHON", | |
| "buffer-overflow": "C++", | |
| "use-after-free": "C++", | |
| "format-string": "C++", | |
| "integer-overflow": "C++", | |
| "null-pointer-dereference": "JAVA", | |
| "double-free": "C++", | |
| "off-by-one": "C++", | |
| "uninitialized-memory": "C++", | |
| "race-condition": "GO", | |
| "time-of-check-time-of-use": "C++", | |
| "unsafe-reflection": "JAVA", | |
| "trust-boundary-violation": "PYTHON", | |
| } | |
| # Difficulty-level human descriptions (used inside the system prompt so the | |
| # AI actually scales the vulnerable code complexity). | |
| DIFFICULTY_PROFILE = { | |
| "مبتدئ": { | |
| "rank": 1, | |
| "label": "Beginner", | |
| "code_lines": (20, 30), | |
| "obviousness": "very high — the vulnerability is the only interesting thing in the file", | |
| }, | |
| "سهل": { | |
| "rank": 2, | |
| "label": "Easy", | |
| "code_lines": (25, 40), | |
| "obviousness": "high — one obvious flaw, minimal surrounding code", | |
| }, | |
| "متوسط": { | |
| "rank": 3, | |
| "label": "Medium", | |
| "code_lines": (35, 55), | |
| "obviousness": "medium — vulnerability is hidden among 2-3 decoy patterns, the trainee must pick the right one", | |
| }, | |
| "صعب": { | |
| "rank": 4, | |
| "label": "Hard", | |
| "code_lines": (50, 75), | |
| "obviousness": "low — multiple plausible-looking patterns, the real flaw is non-obvious and requires careful reading", | |
| }, | |
| "خبير": { | |
| "rank": 5, | |
| "label": "Expert", | |
| "code_lines": (70, 110), | |
| "obviousness": "very low — production-quality code with a subtle flaw; multiple red herrings; requires deep expertise to spot", | |
| }, | |
| } | |
| # Pool constants | |
| # Target: 5 challenges always available, refill when drops below 3 | |
| POOL_TARGET = 5 | |
| POOL_THRESHOLD = 2 | |
| POOL_BATCH = 3 | |
| from app.core.config import ( | |
| SUPABASE_URL, SUPABASE_ANON_KEY, | |
| CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_MODEL, CLOUDFLARE_URL, | |
| GROQ_API_KEY, GROQ_MODEL, GROQ_API_URL, | |
| NVIDIA_API_KEY, NVIDIA_MODEL, NVIDIA_URL, | |
| MISTRAL_API_KEY, MISTRAL_MODEL, MISTRAL_API_URL, | |
| ) | |
| CLOUDFLARE_MODEL_FALLBACKS = [ | |
| "@cf/qwen/qwen2.5-coder-32b-instruct", | |
| "@cf/meta/llama-3.3-70b-instruct-fp8-fast", | |
| "@cf/meta/llama-3.1-70b-instruct", | |
| "@cf/mistralai/mistral-small-3.1-24b-instruct", | |
| "@cf/openai/gpt-oss-120b", | |
| "@cf/openai/gpt-oss-20b", | |
| "@cf/meta/llama-3.1-8b-instruct", | |
| ] | |
| TABLE_NAME = "vulnerability_hunter_challenges" | |
| # Per-team backoff tracker | |
| _AI_BACKOFF_UNTIL: dict[str, float] = {} | |
| # Concurrency control | |
| _POOL_LOCKS: dict[str, asyncio.Lock] = {} | |
| _WATCHER_STARTED: set[str] = set() | |
| def _get_pool_lock(team_role: str) -> asyncio.Lock: | |
| if team_role not in _POOL_LOCKS: | |
| _POOL_LOCKS[team_role] = asyncio.Lock() | |
| return _POOL_LOCKS[team_role] | |
| # --------------------------------------------------------------------------- # | |
| # 1. Helpers (mirror code_fixing_generator.py) | |
| # --------------------------------------------------------------------------- # | |
| def supabase_headers(content_type: bool = False) -> dict: | |
| headers = { | |
| "apikey": SUPABASE_ANON_KEY, | |
| "Authorization": f"Bearer {SUPABASE_ANON_KEY}", | |
| } | |
| if content_type: | |
| headers["Content-Type"] = "application/json" | |
| return headers | |
| def _extract_string_value(text: str, key: str) -> Optional[str]: | |
| pattern = rf'"{re.escape(key)}"\s*:\s*"((?:[^"\\]|\\.)*)"' | |
| match = re.search(pattern, text, re.DOTALL) | |
| if match: | |
| value = match.group(1) | |
| value = value.replace('\\n', '\n').replace('\\t', '\t').replace('\\"', '"').replace('\\\\', '\\') | |
| return value | |
| return None | |
| def _find_matching_bracket(text: str, open_idx: int, open_ch: str, close_ch: str) -> Optional[int]: | |
| depth = 0 | |
| i = open_idx | |
| in_string = False | |
| escape_next = False | |
| while i < len(text): | |
| c = text[i] | |
| if escape_next: | |
| escape_next = False | |
| i += 1 | |
| continue | |
| if c == '\\' and in_string: | |
| escape_next = True | |
| i += 1 | |
| continue | |
| if c == '"': | |
| in_string = not in_string | |
| elif not in_string: | |
| if c == open_ch: | |
| depth += 1 | |
| elif c == close_ch: | |
| depth -= 1 | |
| if depth == 0: | |
| return i | |
| i += 1 | |
| return None | |
| def parse_json_safe(raw) -> dict: | |
| if raw is None or raw == "": | |
| raise ValueError("Empty response from model") | |
| if isinstance(raw, dict): | |
| return raw | |
| if not isinstance(raw, str): | |
| raise ValueError(f"Expected str or dict, got {type(raw).__name__}") | |
| cleaned = raw.strip() | |
| fence_match = re.search(r"```(?:json)?\s*([\s\S]*?)```", cleaned, re.IGNORECASE) | |
| if fence_match: | |
| cleaned = fence_match.group(1).strip() | |
| start = cleaned.find("{") | |
| if start == -1: | |
| raise ValueError(f"No JSON object found. Raw: {cleaned[:200]}") | |
| end = _find_matching_bracket(cleaned, start, "{", "}") | |
| if end is None: | |
| end = cleaned.rfind("}") | |
| if end == -1 or end <= start: | |
| raise ValueError(f"Unbalanced JSON braces. Raw: {cleaned[:200]}") | |
| cleaned = cleaned[start:end + 1] | |
| cleaned = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", cleaned) | |
| try: | |
| return json.loads(cleaned, strict=False) | |
| except json.JSONDecodeError: | |
| pass | |
| fixed = cleaned | |
| fixed = re.sub(r',\s*([}\]])', r'\1', fixed) | |
| fixed = re.sub(r'(?<!")(\b[A-Za-z_][A-Za-z0-9_]*\b)(\s*:)', r'"\1"\2', fixed) | |
| fixed = re.sub(r"'([^'\\]*(?:\\.[^'\\]*)*)'", lambda m: '"' + m.group(1).replace('"', '\\"') + '"', fixed) | |
| try: | |
| return json.loads(fixed, strict=False) | |
| except json.JSONDecodeError: | |
| pass | |
| result: dict = {} | |
| for key in ("title", "story", "task_outline", "vulnerable_code", | |
| "vulnerability_description", "difficulty", | |
| "vulnerability_type", "vulnerability_class"): | |
| val = _extract_string_value(cleaned, key) | |
| if val is not None: | |
| result[key] = val | |
| hints_key_idx = cleaned.find('"hints"') | |
| if hints_key_idx != -1: | |
| bracket_start = cleaned.find("[", hints_key_idx) | |
| if bracket_start != -1: | |
| bracket_end = _find_matching_bracket(cleaned, bracket_start, "[", "]") | |
| if bracket_end is not None: | |
| hints_str = cleaned[bracket_start:bracket_end + 1] | |
| try: | |
| result["hints"] = json.loads(hints_str, strict=False) | |
| except json.JSONDecodeError: | |
| hints_str = re.sub(r',\s*([}\]])', r'\1', hints_str) | |
| try: | |
| result["hints"] = json.loads(hints_str, strict=False) | |
| except Exception: | |
| result["hints"] = [] | |
| if not result: | |
| preview = raw[:500].replace("\n", " ") | |
| raise ValueError(f"Could not extract any fields from model output. Raw preview: {preview}") | |
| return result | |
| def _pick_rotation_slot(i: int) -> tuple[str, str, str]: | |
| """Return (language, vulnerability_type, difficulty) for slot i.""" | |
| vuln = VULNERABILITY_TYPES[i % len(VULNERABILITY_TYPES)] | |
| lang = LANG_BY_VULN.get(vuln, "PYTHON") | |
| diff = ALLOWED_DIFFICULTIES[i % len(ALLOWED_DIFFICULTIES)] | |
| return lang, vuln, diff | |
| # --------------------------------------------------------------------------- # | |
| # 2. Curated Seeds — one per (difficulty, vuln) hot spot so the pool is | |
| # usable the moment migrations are applied. | |
| # --------------------------------------------------------------------------- # | |
| def _build_seeds() -> list[dict]: | |
| """Return the curated seed challenges used as a safety net when AI is | |
| down. Coverage is intentionally broad across BOTH web and non-web | |
| vulnerabilities in C / C++ / Java / Python / JS / PHP / Go / Rust / C# / | |
| Ruby, and across the 5 difficulty levels (3 seeds per level = 15 total). | |
| Each entry MUST be a complete, runnable program with a single | |
| intentionally-planted flaw matching `vulnerability_type` (canonical | |
| kebab-case key). Titles are short, no prefix. | |
| """ | |
| seeds: list[dict] = [ | |
| # ===================== مبتدئ (Beginner) ===================== | |
| { | |
| "language": "PYTHON", | |
| "vulnerability_type": "sql-injection", | |
| "vulnerability_class": "Injection", | |
| "module": "web-security", | |
| "title": "استعلام تسجيل دخول قابل للحقن", | |
| "story": "اكتشف فريق الأمن في شركة تجارة إلكترونية أن صفحة تسجيل الدخول تتعرض لمحاولات اختراق متكررة. تم اعتراض دالة `login()` التي يستدعيها السيرفر وُجد أنها تكتب اسم المستخدم مباشرة في الاستعلام.", | |
| "task_outline": "ادرس الكود وحدد الثغرة الأمنية الوحيدة الموجودة فيه. اكتب اسم الثغرة بالإنجليزية بالضبط كما هي متعارف عليها في معايير OWASP.", | |
| "vulnerable_code": "import sqlite3\n\ndef login(username, password):\n conn = sqlite3.connect('shop.db')\n cur = conn.cursor()\n query = f\"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'\"\n cur.execute(query)\n row = cur.fetchone()\n conn.close()\n return row is not None\n", | |
| "vulnerability_description": "استخدام f-string في بناء استعلام SQL يسمح للمهاجم بحقن تعليمات SQL إضافية، مثل ' OR '1'='1 لتجاوز المصادقة.", | |
| "difficulty": "مبتدئ", | |
| "xp_reward": 100, | |
| "hints": [ | |
| {"level": 1, "text": "انتبه لكيفية دمج متغيرات المستخدم داخل الاستعلام", "xp_cost": 10}, | |
| {"level": 2, "text": "الاستعلام يُبنى عبر f-string بدلًا من parameterized query", "xp_cost": 20}, | |
| {"level": 3, "text": "اسم الثغرة: sql-injection", "xp_cost": 40}, | |
| ], | |
| }, | |
| { | |
| "language": "C++", | |
| "vulnerability_type": "buffer-overflow", | |
| "vulnerability_class": "Memory Safety", | |
| "module": "secure-coding", | |
| "title": "قراءة اسم في مخزن ثابت بحجم 32 بايت", | |
| "story": "برنامج ترحيب بسيط يطلب من المستخدم اسمه ويطبعه. لاحظ المختبر الأمني أنه عند إدخال نص طويل يتعطل البرنامج.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة. اكتب اسمها بالإنجليزية بالضبط.", | |
| "vulnerable_code": "#include <iostream>\n#include <cstring>\n\nvoid greet() {\n char name[32];\n std::cout << \"Enter your name: \";\n std::cin >> name;\n std::cout << \"Hello, \" << name << \"\\n\";\n}\n\nint main() {\n greet();\n return 0;\n}\n", | |
| "vulnerability_description": "std::cin >> مع مصفوفة ثابتة لا يحد من طول الإدخال، فيسمح بفيض في الذاكرة (Stack Buffer Overflow).", | |
| "difficulty": "مبتدئ", | |
| "xp_reward": 100, | |
| "hints": [ | |
| {"level": 1, "text": "قارن حجم المخزن المؤقت مع ما يقبله std::cin >>", "xp_cost": 10}, | |
| {"level": 2, "text": "std::cin >> لا يحد من طول النص المُدخل", "xp_cost": 20}, | |
| {"level": 3, "text": "اسم الثغرة: buffer-overflow", "xp_cost": 40}, | |
| ], | |
| }, | |
| { | |
| "language": "PYTHON", | |
| "vulnerability_type": "weak-hashing", | |
| "vulnerability_class": "Cryptographic Failures", | |
| "module": "cryptography", | |
| "title": "تجزئة كلمات المرور بـ MD5", | |
| "story": "نظام إدارة مستخدمين يحفظ كلمات المرور بعد تجزئتها باستخدام MD5. لاحظ الفريق الأمني أن المهاجم يستطيع كسر كلمات المرور بسرعة باستخدام rainbow tables.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "import hashlib\n\ndef hash_password(pwd: str) -> str:\n return hashlib.md5(pwd.encode()).hexdigest()\n\ndef store_user(username, password):\n return {\n 'user': username,\n 'pwd_hash': hash_password(password),\n }\n", | |
| "vulnerability_description": "MD5 دالة مجزّأة ضعيفة، سريعة، وبدون salt. يسهل كسرها.", | |
| "difficulty": "مبتدئ", | |
| "xp_reward": 100, | |
| "hints": [ | |
| {"level": 1, "text": "ما نوع دالة التجزئة المستخدمة؟ هل هي آمنة لكلمات المرور؟", "xp_cost": 10}, | |
| {"level": 2, "text": "MD5 دالة قديمة وسريعة بدون salt", "xp_cost": 20}, | |
| {"level": 3, "text": "اسم الثغرة: weak-hashing", "xp_cost": 40}, | |
| ], | |
| }, | |
| # ===================== سهل (Easy) ===================== | |
| { | |
| "language": "JAVA", | |
| "vulnerability_type": "null-pointer-dereference", | |
| "vulnerability_class": "Memory Safety", | |
| "module": "secure-coding", | |
| "title": "وصول لعضو قبل فحص null", | |
| "story": "كلاس `CustomerService` يستدعي `customer.getName()` في معالجة الطلبات. لاحظ فريق المراقبة انهيار التطبيق عند تسجيل دخول ضيف (guest customer = null).", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "public class CustomerService {\n\n public String greet(Customer customer) {\n return \"Welcome, \" + customer.getName();\n }\n\n public static void main(String[] args) {\n CustomerService svc = new CustomerService();\n System.out.println(svc.greet(null));\n }\n}\n", | |
| "vulnerability_description": "استدعاء customer.getName() على null قبل التحقق يرمي NullPointerException وقد يُستخدم لرفض الخدمة.", | |
| "difficulty": "سهل", | |
| "xp_reward": 120, | |
| "hints": [ | |
| {"level": 1, "text": "هل يوجد فحص customer == null قبل الوصول لعضو؟", "xp_cost": 10}, | |
| {"level": 2, "text": "الاستدعاء يقع قبل أي فحص", "xp_cost": 20}, | |
| {"level": 3, "text": "اسم الثغرة: null-pointer-dereference", "xp_cost": 40}, | |
| ], | |
| }, | |
| { | |
| "language": "JAVASCRIPT", | |
| "vulnerability_type": "prototype-pollution", | |
| "vulnerability_class": "Prototype Pollution", | |
| "module": "secure-coding", | |
| "title": "دمج عميق لكائن JSON دون تصفية", | |
| "story": "مكتبة داخلية توفر دالة merge لدمج كائنات JSON. اكتشف الفريق الأمني أن المهاجم يستطيع تعديل Object.prototype مما يؤثر على كل الكائنات في النظام.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "function deepMerge(target, source) {\n for (const key of Object.keys(source)) {\n const sv = source[key];\n if (sv && typeof sv === 'object' && !Array.isArray(sv)) {\n target[key] = deepMerge(target[key] || {}, sv);\n } else {\n target[key] = sv;\n }\n }\n return target;\n}\n", | |
| "vulnerability_description": "عدم تصفية المفاتيح __proto__ / constructor / prototype يسمح بتلويث النموذج الأولي العام.", | |
| "difficulty": "سهل", | |
| "xp_reward": 120, | |
| "hints": [ | |
| {"level": 1, "text": "ماذا لو كان key = '__proto__'؟", "xp_cost": 10}, | |
| {"level": 2, "text": "لا توجد قائمة مفاتيح ممنوعة قبل الدمج العميق", "xp_cost": 20}, | |
| {"level": 3, "text": "اسم الثغرة: prototype-pollution", "xp_cost": 40}, | |
| ], | |
| }, | |
| { | |
| "language": "CSHARP", | |
| "vulnerability_type": "hardcoded-credentials", | |
| "vulnerability_class": "Hardcoded Secrets", | |
| "module": "secure-coding", | |
| "title": "بيانات اعتماد مكتوبة في الكود", | |
| "story": "خدمة داخلية تتصل بقاعدة بيانات الشركة. في الفحص الروتيني اكتشف فريق المراجعة أن اسم المستخدم وكلمة السر مكتوبتان في الكود المصدري.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "using System.Data.SqlClient;\n\npublic class Database {\n private const string DB_USER = \"sa\";\n private const string DB_PASS = \"Adm1n!2024#\";\n private const string DB_HOST = \"db.internal.corp\";\n\n public SqlConnection Open() {\n var conn = new SqlConnection(\n $\"Server={DB_HOST};Database=app;User Id={DB_USER};Password={DB_PASS};\");\n conn.Open();\n return conn;\n }\n}\n", | |
| "vulnerability_description": "الكتابة المباشرة لكلمات السر في الكود تُعرّضها لتسريب المصدر (git, decompile, logs).", | |
| "difficulty": "سهل", | |
| "xp_reward": 120, | |
| "hints": [ | |
| {"level": 1, "text": "هل كلمة السر في متغير بيئي أو vault؟", "xp_cost": 10}, | |
| {"level": 2, "text": "القيم مكتوبة في const string داخل الكود", "xp_cost": 20}, | |
| {"level": 3, "text": "اسم الثغرة: hardcoded-credentials", "xp_cost": 40}, | |
| ], | |
| }, | |
| # ===================== متوسط (Medium) ===================== | |
| { | |
| "language": "C++", | |
| "vulnerability_type": "format-string", | |
| "vulnerability_class": "Memory Safety", | |
| "module": "secure-coding", | |
| "title": "سجل طباعة يستخدم مدخلات المستخدم كصيغة", | |
| "story": "تطبيق خادم يطبع رسائل السجل باستخدام printf(message) حيث message يأتي من المستخدم. اكتشف فريق الأمن أن المهاجم يستطيع قراءة ذاكرة البرنامج أو تنفيذ كود.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "#include <cstdio>\n#include <cstring>\n\nvoid log_user_input(const char* user_msg) {\n char buf[256];\n snprintf(buf, sizeof(buf), \"[USER] \");\n strncat(buf, user_msg, 200);\n printf(buf);\n}\n", | |
| "vulnerability_description": "printf بدون سلسلة تنسيق '%s' يسمح للمهاجم بحقن محددات مثل %x أو %n لقراءة/كتابة الذاكرة.", | |
| "difficulty": "متوسط", | |
| "xp_reward": 150, | |
| "hints": [ | |
| {"level": 1, "text": "هل سلسلة التنسيق محددة في printf؟", "xp_cost": 15}, | |
| {"level": 2, "text": "printf(buf) يفسّر أي % في buf كمحدد", "xp_cost": 30}, | |
| {"level": 3, "text": "اسم الثغرة: format-string", "xp_cost": 50}, | |
| ], | |
| }, | |
| { | |
| "language": "GO", | |
| "vulnerability_type": "race-condition", | |
| "vulnerability_class": "Concurrency", | |
| "module": "secure-coding", | |
| "title": "عدّاد مشترك بدون مزامنة في Go", | |
| "story": "خدمة تتلقى طلبات كثيرة في نفس الوقت، وكل handler يزيد عدادًا مشتركًا. يلاحظ فريق المراقبة أن العداد يعطي قيمًا أقل من المتوقع.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "package main\n\nimport (\n \"fmt\"\n \"net/http\"\n)\n\nvar counter int\n\nfunc handler(w http.ResponseWriter, r *http.Request) {\n counter++\n fmt.Fprintf(w, \"count=%d\\n\", counter)\n}\n\nfunc main() {\n http.HandleFunc(\"/hit\", handler)\n http.ListenAndServe(\":8080\", nil)\n}\n", | |
| "vulnerability_description": "counter++ ليس ذريًا، القراءة-الزيادة-الكتابة تتداخل بين الـ goroutines وتُفقد التحديثات.", | |
| "difficulty": "متوسط", | |
| "xp_reward": 150, | |
| "hints": [ | |
| {"level": 1, "text": "هل يتم استخدام mutex أو atomic؟", "xp_cost": 15}, | |
| {"level": 2, "text": "counter++ = LOAD + ADD + STORE، تتخللها goroutines أخرى", "xp_cost": 30}, | |
| {"level": 3, "text": "اسم الثغرة: race-condition", "xp_cost": 50}, | |
| ], | |
| }, | |
| { | |
| "language": "PYTHON", | |
| "vulnerability_type": "insecure-deserialization", | |
| "vulnerability_class": "Insecure Deserialization", | |
| "module": "secure-coding", | |
| "title": "تحميل pickle من ملف يتحكم به المستخدم", | |
| "story": "تطبيق يخزن الجلسات في ملفات على القرص، وعند كل طلب يتم تحميلها عبر pickle.load. المهاجم يستطيع رفع ملف جلسة مزيف لتنفيذ كود على السيرفر.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "import pickle\nimport os\n\ndef load_session(session_id):\n path = os.path.join('/tmp/sessions', session_id)\n with open(path, 'rb') as f:\n return pickle.load(f)\n", | |
| "vulnerability_description": "pickle.load على بيانات يتحكم بها المهاجم ينفذ كود Python تعسفي عند الإلغاء عبر __reduce__.", | |
| "difficulty": "متوسط", | |
| "xp_reward": 150, | |
| "hints": [ | |
| {"level": 1, "text": "هل تثق بمحتوى الملف الذي يحمّله pickle.load؟", "xp_cost": 15}, | |
| {"level": 2, "text": "pickle يستدعي __reduce__ الذي قد ينفذ أوامر", "xp_cost": 30}, | |
| {"level": 3, "text": "اسم الثغرة: insecure-deserialization", "xp_cost": 50}, | |
| ], | |
| }, | |
| # ===================== صعب (Hard) ===================== | |
| { | |
| "language": "JAVA", | |
| "vulnerability_type": "unsafe-reflection", | |
| "vulnerability_class": "Input Validation", | |
| "module": "secure-coding", | |
| "title": "استدعاء Class.forName على مدخلات المستخدم", | |
| "story": "نظام إضافة إضافات (plugins) يستدعي Class.forName باسم يحدده المستخدم، ثم يستدعي newInstance. لاحظ فريق الأمن أن المهاجم يستطيع تشغيل أي كلاس على الـ classpath.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "public class PluginLoader {\n\n public static Object load(String className) throws Exception {\n Class<?> cls = Class.forName(className);\n return cls.getDeclaredConstructor().newInstance();\n }\n\n public static void main(String[] args) throws Exception {\n Object plugin = load(args[0]);\n System.out.println(\"loaded: \" + plugin);\n }\n}\n", | |
| "vulnerability_description": "Class.forName على مدخلات المستخدم يسمح بتحميل أي كلاس متاح (RCE عبر Runtime.exec مثلًا).", | |
| "difficulty": "صعب", | |
| "xp_reward": 200, | |
| "hints": [ | |
| {"level": 1, "text": "هل className يُتحقق منه ضد whitelist؟", "xp_cost": 20}, | |
| {"level": 2, "text": "Class.forName يقبل أي اسم متاح في الـ classpath", "xp_cost": 40}, | |
| {"level": 3, "text": "اسم الثغرة: unsafe-reflection", "xp_cost": 70}, | |
| ], | |
| }, | |
| { | |
| "language": "RUBY", | |
| "vulnerability_type": "mass-assignment", | |
| "vulnerability_class": "Mass Assignment", | |
| "module": "secure-coding", | |
| "title": "تحديث نموذج Ruby on Rails مع جميع المعاملات", | |
| "story": "نموذج User في تطبيق Rails يقبل كل المعاملات من params ويحدّثها. اكتشف فريق الأمن أن المهاجم يرسل admin=true في JSON ويصبح مديرًا.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "class UsersController < ApplicationController\n def update\n user = User.find(params[:id])\n user.update(params[:user])\n render json: user\n end\nend\n", | |
| "vulnerability_description": "تمرير params[:user] مباشرةً إلى update يسمح للمهاجم بضبط أعمدة حساسة (is_admin, role) لم يكن من المفترض أن يعدّلها.", | |
| "difficulty": "صعب", | |
| "xp_reward": 200, | |
| "hints": [ | |
| {"level": 1, "text": "هل يوجد strong parameters أو attr_accessible؟", "xp_cost": 20}, | |
| {"level": 2, "text": "params[:user] قد يحوي admin: true", "xp_cost": 40}, | |
| {"level": 3, "text": "اسم الثغرة: mass-assignment", "xp_cost": 70}, | |
| ], | |
| }, | |
| { | |
| "language": "C++", | |
| "vulnerability_type": "double-free", | |
| "vulnerability_class": "Memory Safety", | |
| "module": "secure-coding", | |
| "title": "تحرير نفس المخزن مرتين في مسار الخطأ", | |
| "story": "دالة تعالج مدخلات في مخزن مؤقت. في حالة فشل المعالجة، تُستدعى free(buf) مرتين في فروع شرطية متقاربة، مما يتسبب في تعطل البرنامج أو تنفيذ كود.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "#include <cstdlib>\n#include <cstring>\n\nvoid process(const char* data, int len) {\n char* buf = (char*)malloc(len);\n memcpy(buf, data, len);\n if (len < 0) {\n free(buf);\n return;\n }\n if (len == 0) {\n free(buf);\n return;\n }\n free(buf);\n}\n", | |
| "vulnerability_description": "تحرير الذاكرة مرتين (double-free) يسمح للمهاجم بالسيطرة على allocator وتنفيذ كود.", | |
| "difficulty": "صعب", | |
| "xp_reward": 200, | |
| "hints": [ | |
| {"level": 1, "text": "هل buf = nullptr بعد كل free؟", "xp_cost": 20}, | |
| {"level": 2, "text": "إذا مرّ المسار بفروع متتالية قد يُحرر buf مرتين", "xp_cost": 40}, | |
| {"level": 3, "text": "اسم الثغرة: double-free", "xp_cost": 70}, | |
| ], | |
| }, | |
| # ===================== خبير (Expert) ===================== | |
| { | |
| "language": "C++", | |
| "vulnerability_type": "integer-overflow", | |
| "vulnerability_class": "Memory Safety", | |
| "module": "secure-coding", | |
| "title": "حساب حجم المخزن عبر ضرب int*sizeof", | |
| "story": "دالة تحجز مخزنًا بناءً على ضرب حجم العنصر في عدد العناصر. في حالة تجاوز حاصل الضرب للحد الأقصى، يُحجز مخزن صغير جدًا ويسبب كتابة خارج الحدود.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "#include <cstdlib>\n#include <cstring>\n\nvoid copy(int count, const char* data) {\n size_t total = count * sizeof(int);\n char* buf = (char*)malloc(total);\n memcpy(buf, data, total);\n}\n", | |
| "vulnerability_description": "count * sizeof(int) قد يفيض إذا كان count كبيرًا، فيُحجز مخزن أصغر من المتوقع ثم تُنسخ البيانات كاملةً إلى خارج حدوده.", | |
| "difficulty": "خبير", | |
| "xp_reward": 250, | |
| "hints": [ | |
| {"level": 1, "text": "ماذا لو كان count قريبًا من INT_MAX؟", "xp_cost": 30}, | |
| {"level": 2, "text": "الضرب count * sizeof(int) يتجاوز size_t؟ كلاهما", "xp_cost": 60}, | |
| {"level": 3, "text": "اسم الثغرة: integer-overflow", "xp_cost": 100}, | |
| ], | |
| }, | |
| { | |
| "language": "PYTHON", | |
| "vulnerability_type": "time-of-check-time-of-use", | |
| "vulnerability_class": "Concurrency", | |
| "module": "secure-coding", | |
| "title": "سباق بين فحص الصلاحية والاستخدام", | |
| "story": "خدمة تحقّق أن المستخدم يملك رصيدًا كافيًا قبل تنفيذ السحب. اكتشف فريق الأمن أن المهاجم يرسل طلبين متزامنين بنفس المعرّف فيخصم الرصيد مرتين.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "balances = {}\n\ndef withdraw(user, amount):\n cur = balances.get(user, 0)\n if cur >= amount:\n balances[user] = cur - amount\n return True\n return False\n", | |
| "vulnerability_description": "الفحص (>=) والاستخدام (-) ليسا ذريين. بين الخطوتين يقرأ طلب آخر نفس الرصيد ويخصمه أيضًا (TOCTOU).", | |
| "difficulty": "خبير", | |
| "xp_reward": 250, | |
| "hints": [ | |
| {"level": 1, "text": "هل الخطوتان داخل عملية ذرية (قفل/معاملة)؟", "xp_cost": 30}, | |
| {"level": 2, "text": "هناك فجوة زمنية بين القراءة والكتابة", "xp_cost": 60}, | |
| {"level": 3, "text": "اسم الثغرة: time-of-check-time-of-use", "xp_cost": 100}, | |
| ], | |
| }, | |
| { | |
| "language": "JAVA", | |
| "vulnerability_type": "weak-randomness", | |
| "vulnerability_class": "Cryptographic Failures", | |
| "module": "cryptography", | |
| "title": "توليد توكن باستخدام java.util.Random", | |
| "story": "خدمة تولّد توكن استعادة كلمة المرور عبر java.util.Random ثم تحوّله إلى Base64. المهاجم يلاحظ أن قيم التوكنات متسلسلة ويسطو على حسابات المستخدمين.", | |
| "task_outline": "ادرس الكود وحدد نوع الثغرة.", | |
| "vulnerable_code": "import java.util.Random;\n\npublic class TokenGen {\n private final Random rng = new Random();\n\n public String nextToken() {\n long v = rng.nextLong();\n return Long.toHexString(v);\n }\n\n public static void main(String[] args) {\n TokenGen g = new TokenGen();\n for (int i = 0; i < 5; i++) {\n System.out.println(g.nextToken());\n }\n }\n}\n", | |
| "vulnerability_description": "java.util.Random يستخدم LCG بذرة مكشوفة، فيمكن إعادة بناء البذرة من عيّنات متتالية وتوقع التوكنات القادمة.", | |
| "difficulty": "خبير", | |
| "xp_reward": 250, | |
| "hints": [ | |
| {"level": 1, "text": "هل Random آمن للاستخدامات الأمنية؟", "xp_cost": 30}, | |
| {"level": 2, "text": "java.util.Random = LCG، قابل للتوقع", "xp_cost": 60}, | |
| {"level": 3, "text": "اسم الثغرة: weak-randomness", "xp_cost": 100}, | |
| ], | |
| }, | |
| ] | |
| return seeds | |
| # --------------------------------------------------------------------------- # | |
| # 3. AI Generation | |
| # --------------------------------------------------------------------------- # | |
| SYSTEM_PROMPT_TEMPLATE = """You are a cybersecurity expert who generates realistic vulnerable code samples for a "Vulnerability Hunter" training game. | |
| ========================================================= | |
| GAME MECHANICS — READ THIS FIRST AND OBEY STRICTLY | |
| ========================================================= | |
| This is NOT a "code fixing" game. There is no code editor for the trainee. | |
| There is no patched/fixed version, no "modify this line", no "rewrite this function". | |
| The trainee's ONLY interaction with the challenge is: | |
| 1. They SEE a code snippet on screen (read-only). | |
| 2. They TYPE the canonical vulnerability name (e.g. "sql-injection", "buffer-overflow") | |
| into a single text input field. | |
| 3. They press "Submit". The backend compares their typed string against the | |
| canonical key and reports right/wrong. | |
| That is the entire game. The trainee never edits, never compiles, never runs, | |
| never refactors, never "fixes" the code. The code is a static exhibit; the only | |
| task is to name the flaw. | |
| ========================================================= | |
| WHAT YOU MUST PRODUCE | |
| ========================================================= | |
| - A complete, syntactically correct, runnable program in {language} of between | |
| {code_min} and {code_max} non-blank lines, with exactly ONE primary | |
| vulnerability of type "{vuln_type}" (OWASP class: {vuln_class}). | |
| - The code is a FORENSIC EVIDENCE — frozen, not to be modified. | |
| - An Arabic task_outline that ONLY tells the trainee: "read the code, identify | |
| the vulnerability, and type its canonical English name in the input field". | |
| DO NOT write: "fix the code", "patch the bug", "rewrite the function", | |
| "edit line N", "correct the error", or anything that implies editing. | |
| The target canonical vulnerability is: {vuln_type} | |
| The target OWASP family / class is: {vuln_class} | |
| The target language is: {language} | |
| The target difficulty is: {difficulty} ({difficulty_label}) — {difficulty_obviousness} | |
| DIFFICULTY SCALING: | |
| - Code length MUST be between {code_min} and {code_max} lines (non-blank, non-comment). | |
| - For مبتدئ / سهل: vulnerability is the obvious focal point. Minimal surrounding code. | |
| - For متوسط: 1 vulnerability hidden among 2-3 plausible-looking but harmless patterns. | |
| - For صعب / خبير: realistic production code. Multiple imports, helper functions, | |
| configuration, docstrings. The real flaw is subtle and requires careful reading; | |
| multiple red herrings. | |
| ABSOLUTE RULES — VIOLATIONS WILL BE REJECTED: | |
| 1. Respond with ONLY valid JSON. No markdown, no code fences, no explanation | |
| before or after. | |
| 2. The "vulnerable_code" field MUST be a complete, syntactically correct, | |
| fully-implemented program of EXACTLY between {code_min} and {code_max} | |
| non-blank lines. All imports, function definitions, and a main entry point | |
| must be present. No truncation, no "...", no half-written functions, no | |
| "# TODO" placeholders, no "see fix below" comments. | |
| 3. There must be exactly ONE primary vulnerability of type "{vuln_type}". | |
| No other major issues. | |
| 4. "vulnerability_description" must explain the specific flaw and how an | |
| attacker could exploit it. | |
| 5. "task_outline" MUST be a single sentence in Arabic, in EXACTLY this shape | |
| (translated, paraphrased freely but keeping the meaning): | |
| "ادرس الكود التالي جيداً، حدّد الثغرة الأمنية الوحيدة الموجودة فيه، | |
| ثم اكتب اسمها بالإنجليزية (مفتاح قصير) في حقل الإجابة." | |
| DO NOT mention: إصلاح، أصلح، صحّح، عدّل، طبّق، اكتب الكود الآمن، ضع، غيّر، | |
| rewrite, fix, patch, edit, modify, correct. These are FORBIDDEN words in | |
| task_outline. The trainee's job is IDENTIFICATION, not REMEDIATION. | |
| 6. "vulnerability_type" MUST be the canonical key "{vuln_type}" (English, | |
| kebab-case). "vulnerability_class" MUST be "{vuln_class}". | |
| 7. ALL user-facing text fields (title, story, task_outline, | |
| vulnerability_description, hints) MUST be written in Arabic ONLY. No | |
| English, no Chinese, no other languages. The "vulnerable_code" field | |
| contains code in {language} (which is correct), but all comments inside | |
| the code should be in English (//, #, etc.) since this is conventional. | |
| 8. "difficulty" must be EXACTLY one of these Arabic words: | |
| "مبتدئ", "سهل", "متوسط", "صعب", or "خبير". | |
| 9. "hints" must be a JSON array of exactly 3 objects: | |
| [{{"level": 1, "text": "تلميح أول غامض", "xp_cost": 10}}, | |
| {{"level": 2, "text": "تلميح ثانٍ أكثر تحديداً", "xp_cost": 20}}, | |
| {{"level": 3, "text": "تلميح ثالث يكشف اسم الثغرة", "xp_cost": 40}}] | |
| The first hint should be vague, the second more specific, the third | |
| should reveal the canonical vulnerability name. | |
| UNIQUENESS RULES — ZERO TOLERANCE FOR DUPLICATES: | |
| - Every challenge MUST be completely unique in title, story, code structure, and variable names. | |
| - NEVER reuse the same application domain (e.g. if previous was a "bank API", make this one a "hospital system", "IoT firmware", "game server", "scientific computing", etc.). | |
| - NEVER reuse variable names, function names, class names, or file names from common templates. Use creative, context-specific naming. | |
| - The vulnerable_code MUST use a different programming pattern, different data structures, and different business logic each time. | |
| - The title and story MUST describe a completely different scenario — not just a paraphrase. | |
| - Vary the code style: different imports, different module structure, different comments. | |
| - DONT GENERATE ANY OF THE FOLLOWING TITLES (THEY ARE ALREADY IN THE DATABASE): | |
| {blacklisted_titles} | |
| Return ONLY this exact JSON structure (no other text): | |
| {{"title": "عنوان التحدي بالعربية", "story": "قصة التحدي بالعربية", "task_outline": "ادرس الكود التالي وحدد الثغرة الأمنية الوحيدة الموجودة فيه ثم اكتب اسمها بالإنجليزية (مفتاح قصير) في حقل الإجابة.", "vulnerable_code": "الكود الكامل المصاب بالثغرة", "vulnerability_type": "{vuln_type}", "vulnerability_class": "{vuln_class}", "vulnerability_description": "شرح الثغرة بالعربية", "difficulty": "{difficulty}", "hints": [{{"level": 1, "text": "...", "xp_cost": 10}}, {{"level": 2, "text": "...", "xp_cost": 20}}, {{"level": 3, "text": "...", "xp_cost": 40}}]}} | |
| """ | |
| async def _post_with_json_fallback(client, url: str, payload: dict, headers: dict, provider_name: str) -> Optional[httpx.Response]: | |
| resp = await client.post(url, json=payload, headers=headers) | |
| if resp.status_code in (400, 404, 422) and "response_format" in payload: | |
| print(f" [{provider_name}] response_format not supported ({resp.status_code}), retrying without...") | |
| payload2 = {k: v for k, v in payload.items() if k != "response_format"} | |
| resp = await client.post(url, json=payload2, headers=headers) | |
| return resp | |
| async def _try_mistral(prompt: str, system: str, model: str) -> Optional[str]: | |
| """Try Mistral API (mistral-large-latest by default). | |
| Per the user's request, Mistral is the PRIMARY tier for challenge | |
| generation — it runs first in the orchestrator. Only on a hard | |
| failure (429 / timeout / 5xx / parse) do we fall through to | |
| Cloudflare, then Groq, then NVIDIA/DeepSeek, then seed. | |
| """ | |
| if not MISTRAL_API_KEY: | |
| return None | |
| import httpx as _httpx | |
| try: | |
| headers = { | |
| "Content-Type": "application/json", | |
| "Authorization": f"Bearer {MISTRAL_API_KEY}", | |
| } | |
| messages = [] | |
| if system: | |
| messages.append({"role": "system", "content": system}) | |
| messages.append({"role": "user", "content": prompt}) | |
| payload = { | |
| "model": model or MISTRAL_MODEL, | |
| "messages": messages, | |
| "temperature": 0.7, | |
| "max_tokens": 4096, | |
| "response_format": {"type": "json_object"}, | |
| } | |
| async with _httpx.AsyncClient(timeout=30) as client: | |
| resp = await _post_with_json_fallback(client, MISTRAL_API_URL, payload, headers, "mistral") | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| content = data.get("choices", [{}])[0].get("message", {}).get("content") | |
| if content: | |
| return content | |
| print(f" [mistral] 200 but empty content: {str(data)[:200]}") | |
| return None | |
| if resp.status_code == 429: | |
| print(f" [mistral] 429 rate-limited") | |
| else: | |
| print(f" [mistral] error {resp.status_code}: {resp.text[:200]}") | |
| return None | |
| except Exception as e: | |
| print(f" [mistral] exception: {type(e).__name__}: {e}") | |
| return None | |
| async def _try_cloudflare(prompt: str, system: str) -> Optional[str]: | |
| if not (CLOUDFLARE_API_TOKEN and CLOUDFLARE_ACCOUNT_ID): | |
| return None | |
| import httpx as _httpx | |
| model_candidates = [CLOUDFLARE_MODEL] + [m for m in CLOUDFLARE_MODEL_FALLBACKS if m != CLOUDFLARE_MODEL] | |
| messages = [] | |
| if system: | |
| messages.append({"role": "system", "content": system}) | |
| messages.append({"role": "user", "content": prompt}) | |
| payload_base = {"messages": messages, "temperature": 0.7, "max_tokens": 4096} | |
| for model_name in model_candidates: | |
| url = f"https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/ai/run/{model_name}" | |
| headers = {"Content-Type": "application/json", "Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}"} | |
| payload = {**payload_base, "response_format": {"type": "json_object"}} | |
| try: | |
| async with _httpx.AsyncClient(timeout=30) as client: | |
| resp = await _post_with_json_fallback(client, url, payload, headers, f"cloudflare/{model_name.split('/')[-1]}") | |
| if resp.status_code == 200: | |
| result = resp.json() | |
| if result.get("success") and result.get("result", {}).get("response"): | |
| return result["result"]["response"] | |
| print(f" [cloudflare/{model_name.split('/')[-1]}] 200 but no response field") | |
| continue | |
| if resp.status_code == 429: | |
| print(f" [cloudflare/{model_name.split('/')[-1]}] 429 rate-limited") | |
| continue | |
| print(f" [cloudflare/{model_name.split('/')[-1]}] error {resp.status_code}: {resp.text[:120]}") | |
| except Exception as e: | |
| print(f" [cloudflare/{model_name.split('/')[-1]}] exception: {type(e).__name__}: {e}") | |
| continue | |
| return None | |
| async def _try_groq(prompt: str, system: str, model: str) -> Optional[str]: | |
| if not GROQ_API_KEY: | |
| return None | |
| import httpx as _httpx | |
| try: | |
| headers = {"Content-Type": "application/json", "Authorization": f"Bearer {GROQ_API_KEY}"} | |
| messages = [] | |
| if system: | |
| messages.append({"role": "system", "content": system}) | |
| messages.append({"role": "user", "content": prompt}) | |
| payload = { | |
| "model": model or GROQ_MODEL, | |
| "messages": messages, | |
| "temperature": 0.7, | |
| "max_tokens": 4096, | |
| "response_format": {"type": "json_object"}, | |
| } | |
| async with _httpx.AsyncClient(timeout=90) as client: | |
| resp = await _post_with_json_fallback(client, GROQ_API_URL, payload, headers, "groq") | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| content = data.get("choices", [{}])[0].get("message", {}).get("content") | |
| if content: | |
| return content | |
| print(f" [groq] 200 but empty content: {str(data)[:200]}") | |
| return None | |
| if resp.status_code == 429: | |
| print(f" [groq] 429 rate-limited") | |
| else: | |
| print(f" [groq] error {resp.status_code}: {resp.text[:200]}") | |
| return None | |
| except Exception as e: | |
| print(f" [groq] exception: {type(e).__name__}: {e}") | |
| return None | |
| async def _try_nvidia(prompt: str, system: str, model: str) -> Optional[str]: | |
| if not NVIDIA_API_KEY: | |
| return None | |
| import httpx as _httpx | |
| try: | |
| headers = {"Content-Type": "application/json", "Authorization": f"Bearer {NVIDIA_API_KEY}"} | |
| messages = [] | |
| if system: | |
| messages.append({"role": "system", "content": system}) | |
| messages.append({"role": "user", "content": prompt}) | |
| payload = { | |
| "model": model or NVIDIA_MODEL, | |
| "messages": messages, | |
| "temperature": 0.7, | |
| "max_tokens": 4096, | |
| "response_format": {"type": "json_object"}, | |
| } | |
| async with _httpx.AsyncClient(timeout=90) as client: | |
| resp = await _post_with_json_fallback(client, NVIDIA_API_URL, payload, headers, "deepseek") | |
| if resp.status_code == 200: | |
| data = resp.json() | |
| content = data.get("choices", [{}])[0].get("message", {}).get("content") | |
| if content: | |
| return content | |
| print(f" [deepseek] 200 but empty content: {str(data)[:200]}") | |
| return None | |
| if resp.status_code == 429: | |
| print(f" [deepseek] 429 rate-limited") | |
| else: | |
| print(f" [deepseek] error {resp.status_code}: {resp.text[:200]}") | |
| return None | |
| except Exception as e: | |
| print(f" [deepseek] exception: {type(e).__name__}: {e}") | |
| return None | |
| async def _call_ai(prompt: str, system: str = "", model: str = "") -> Optional[str]: | |
| print(f" [ai] Trying Mistral...") | |
| r = await _try_mistral(prompt, system, model) | |
| if r: | |
| return r | |
| print(f" [ai] Trying Cloudflare...") | |
| r = await _try_cloudflare(prompt, system) | |
| if r: | |
| return r | |
| print(f" [ai] Trying Groq...") | |
| r = await _try_groq(prompt, system, model) | |
| if r: | |
| return r | |
| print(f" [ai] Trying NVIDIA/DeepSeek...") | |
| r = await _try_nvidia(prompt, system, model) | |
| if r: | |
| return r | |
| print(f" [ai] All 4 providers failed") | |
| return None | |
| async def ai_generate_challenge(language: str, vuln_type: str, difficulty: str, blacklisted_titles: str = "") -> Optional[dict]: | |
| vuln_class = VULNERABILITY_CLASS.get(vuln_type, "Security Misconfiguration") | |
| diff_profile = DIFFICULTY_PROFILE.get(difficulty, DIFFICULTY_PROFILE["متوسط"]) | |
| system = SYSTEM_PROMPT_TEMPLATE.format( | |
| vuln_type=vuln_type, | |
| vuln_class=vuln_class, | |
| language=language, | |
| difficulty=difficulty, | |
| difficulty_label=diff_profile["label"], | |
| difficulty_obviousness=diff_profile["obviousness"], | |
| code_min=diff_profile["code_lines"][0], | |
| code_max=diff_profile["code_lines"][1], | |
| blacklisted_titles=blacklisted_titles or "لا يوجد عناوين سابقة حالياً.", | |
| ) | |
| base_prompt = f"ولّد كود {language} بطول {diff_profile['code_lines'][0]}-{diff_profile['code_lines'][1]} سطر يحتوي على ثغرة {vuln_type} (OWASP: {vuln_class}) وفق الهيكل المطلوب. مستوى الصعوبة: {diff_profile['label']}." | |
| # Round 1 | |
| raw = await _call_ai(base_prompt, system) | |
| data = _try_parse(raw) | |
| built = _validate_and_build(data, language, vuln_type, difficulty) if data else None | |
| if built: | |
| return built | |
| # Round 2 | |
| retry_prompt = ( | |
| f"CRITICAL: Return ONLY valid JSON, no markdown, no code fences, no comments. " | |
| f"Generate a complete {diff_profile['code_lines'][0]}-{diff_profile['code_lines'][1]} line {language} program " | |
| f"with a {vuln_type} vulnerability (class: {vuln_class}). Difficulty: {diff_profile['label']}. " | |
| f"Return exactly this structure with Arabic in all string fields:\n" | |
| f'{{"title":"...","story":"...","task_outline":"...","vulnerable_code":"...","vulnerability_type":"{vuln_type}","vulnerability_class":"{vuln_class}","vulnerability_description":"...","difficulty":"{difficulty}","hints":[{{"level":1,"text":"...","xp_cost":10}},{{"level":2,"text":"...","xp_cost":20}},{{"level":3,"text":"...","xp_cost":40}}]}}' | |
| ) | |
| print(f" [ai] Round 2: retrying with stricter prompt") | |
| raw2 = await _call_ai(retry_prompt, system) | |
| data2 = _try_parse(raw2) | |
| built2 = _validate_and_build(data2, language, vuln_type, difficulty) if data2 else None | |
| if built2: | |
| return built2 | |
| # Round 3 | |
| minimal_prompt = ( | |
| f"Output JSON only. language={language}, vuln={vuln_type}, class={vuln_class}, " | |
| f"difficulty={difficulty}. Required keys: title, story, task_outline, vulnerable_code " | |
| f"(complete {diff_profile['code_lines'][0]}-{diff_profile['code_lines'][1]} line program), " | |
| f"vulnerability_type, vulnerability_class, vulnerability_description, difficulty, " | |
| f"hints (array of 3 objects with level,text,xp_cost). No markdown. No commentary." | |
| ) | |
| print(f" [ai] Round 3: minimal retry") | |
| raw3 = await _call_ai(minimal_prompt, system) | |
| data3 = _try_parse(raw3) | |
| built3 = _validate_and_build(data3, language, vuln_type, difficulty) if data3 else None | |
| if built3: | |
| return built3 | |
| print(f" [ai] All rounds exhausted for {language}/{vuln_type}/{difficulty}") | |
| return None | |
| def _try_parse(raw) -> Optional[dict]: | |
| if not raw: | |
| return None | |
| if isinstance(raw, dict): | |
| return raw | |
| try: | |
| return parse_json_safe(raw) | |
| except Exception as e: | |
| print(f" [ai] parse_json_safe failed: {e}") | |
| return None | |
| # Words that betray the AI's "code-fixing" bias. If any of these appear in | |
| # task_outline, the challenge is REJECTED — the trainee's job is identification, | |
| # not remediation. The model has to learn that this is a triage game. | |
| FORBIDDEN_TASK_WORDS_AR = [ | |
| "أصلح", "إصلاح", "صحّح", "تصحيح", "عدّل", "تعديل", "طبّق", "تطبيق", | |
| "اكتب الكود", "ضع الكود", "غيّر", "غيّر الكود", "استبدل", "أعد كتابة", | |
| "برمج", "أنشئ دالة", "أنشئ كود", "أنشئ الإصدار", "اكتب دالة", "اكتب إصدار", | |
| "طبّق", "نفّذ", "بادر", "احذف", "أزل", "تخلص من", | |
| ] | |
| FORBIDDEN_TASK_WORDS_EN = [ | |
| "fix the", "fix this", "patch the", "patch this", "rewrite the", | |
| "rewrite this", "modify the", "modify this", "edit the", "edit this", | |
| "correct the", "correct this", "replace the", "replace this", | |
| "refactor the", "refactor this", "implement a fix", "write a fix", | |
| "write a patch", "secure version", "fixed version", "fixed code", | |
| "update the code", "change the code", "apply a fix", | |
| ] | |
| def _strip_fix_bias(text: str) -> str: | |
| """Post-process the AI's task_outline to neutralize any 'fix the code' bias. | |
| The game is identification-only. If the AI slipped in repair instructions, | |
| we replace the whole sentence with the canonical identification directive. | |
| """ | |
| if not text: | |
| return text | |
| low = text.lower() | |
| for w in FORBIDDEN_TASK_WORDS_AR + FORBIDDEN_TASK_WORDS_EN: | |
| if w.lower() in low: | |
| return ("ادرس الكود التالي جيداً، حدّد الثغرة الأمنية الوحيدة الموجودة فيه، " | |
| "ثم اكتب اسمها بالإنجليزية (مفتاح قصير) في حقل الإجابة.") | |
| return text | |
| def _validate_and_build(data: dict, language: str, vuln_type: str, difficulty: str) -> Optional[dict]: | |
| required = ["title", "story", "task_outline", "vulnerable_code", | |
| "vulnerability_description", "difficulty", | |
| "vulnerability_type", "vulnerability_class"] | |
| for f in required: | |
| if not data.get(f): | |
| print(f" [ai] Missing field: {f}") | |
| return None | |
| vcode = data.get("vulnerable_code", "") | |
| line_count = len([l for l in vcode.split("\n") if l.strip() and not l.strip().startswith(("#", "//"))]) | |
| if line_count < 12: | |
| print(f" [ai] Rejected: vulnerable_code too short ({line_count} effective lines)") | |
| return None | |
| # Acceptable diffs are only the 5 levels | |
| if data["difficulty"] not in ALLOWED_DIFFICULTIES: | |
| data["difficulty"] = difficulty if difficulty in ALLOWED_DIFFICULTIES else "متوسط" | |
| # Reject mismatched vuln types | |
| if data.get("vulnerability_type") and data["vulnerability_type"] != vuln_type: | |
| # Trust the request slot — overwrite | |
| data["vulnerability_type"] = vuln_type | |
| # Strip any "fix the code" bias from the AI's task_outline | |
| raw_task = data.get("task_outline", "") | |
| sanitized_task = _strip_fix_bias(raw_task) | |
| if sanitized_task != raw_task: | |
| print(f" [ai] task_outline had fix-bias; replaced with canonical identification directive") | |
| hints = data.get("hints") or [] | |
| if isinstance(hints, str): | |
| try: | |
| hints = json.loads(hints) | |
| except Exception: | |
| hints = [] | |
| if not isinstance(hints, list): | |
| hints = [] | |
| xp = data.get("xp_reward") or {"خبير": 250, "صعب": 200, "متوسط": 150, "سهل": 120, "مبتدئ": 100}.get(data["difficulty"], 150) | |
| return { | |
| "team_role": "blue", | |
| "language": language, | |
| "module": MODULE_BY_VULN.get(vuln_type, "web-security"), | |
| "title": data["title"], | |
| "story": data["story"], | |
| "task_outline": sanitized_task, | |
| "vulnerable_code": vcode, | |
| "vulnerability_type": vuln_type, | |
| "vulnerability_class": VULNERABILITY_CLASS.get(vuln_type, data.get("vulnerability_class") or "Security Misconfiguration"), | |
| "vulnerability_description": data.get("vulnerability_description", ""), | |
| "hints": hints, | |
| "difficulty": data["difficulty"], | |
| "xp_reward": int(xp), | |
| } | |
| # --------------------------------------------------------------------------- # | |
| # 4. DB Operations | |
| # --------------------------------------------------------------------------- # | |
| async def get_pool_count(team_role: str) -> int: | |
| if not SUPABASE_ANON_KEY or not SUPABASE_URL: | |
| return 0 | |
| url = f"{SUPABASE_URL}/rest/v1/{TABLE_NAME}?select=id&team_role=eq.{team_role}" | |
| try: | |
| async with httpx.AsyncClient() as client: | |
| resp = await client.get(url, headers=supabase_headers()) | |
| if resp.status_code == 200: | |
| return len(resp.json()) | |
| except Exception as e: | |
| print(f"[vuln-hunter] Error checking pool count: {e}") | |
| return 0 | |
| async def _insert_to_db(row: dict) -> bool: | |
| if not SUPABASE_ANON_KEY or not SUPABASE_URL: | |
| return False | |
| from app.core.config import normalize_row_module | |
| row = normalize_row_module(TABLE_NAME, row) | |
| from app.services.insert_guard import atomic_insert | |
| from app.services.dedup import is_duplicate_vuln_hunter | |
| team_role = row.get("team_role", "blue") | |
| return await atomic_insert( | |
| table=TABLE_NAME, | |
| team_role=team_role, | |
| row=row, | |
| dedup_func=is_duplicate_vuln_hunter, | |
| dedup_args=[ | |
| row.get("title", ""), | |
| row.get("vulnerable_code", ""), | |
| row.get("task_outline", ""), | |
| ], | |
| dedup_kwargs={"role_filter": team_role}, | |
| ) | |
| async def _delete_challenge(challenge_id: str): | |
| if not SUPABASE_ANON_KEY or not SUPABASE_URL: | |
| return | |
| url = f"{SUPABASE_URL}/rest/v1/{TABLE_NAME}?id=eq.{challenge_id}" | |
| try: | |
| async with httpx.AsyncClient() as client: | |
| await client.delete(url, headers=supabase_headers()) | |
| except Exception as e: | |
| print(f" Delete error: {e}") | |
| # --------------------------------------------------------------------------- # | |
| # 5. Pool Refill | |
| # --------------------------------------------------------------------------- # | |
| async def _refill_with_seeds_only(team_role: str, count: int) -> int: | |
| """Insert `count` curated seeds (no AI) — used for pre-warm.""" | |
| seeds = _build_seeds() | |
| random.shuffle(seeds) | |
| inserted = 0 | |
| for seed in seeds[:count]: | |
| if await _insert_to_db(seed): | |
| inserted += 1 | |
| return inserted | |
| async def _refill_pool_inner(team_role: str, count: int) -> int: | |
| """Unlocked inner — caller must hold _get_pool_lock(team_role).""" | |
| if team_role not in ALLOWED_TEAMS: | |
| return 0 | |
| until = _AI_BACKOFF_UNTIL.get(team_role, 0) | |
| if time.time() < until: | |
| print(f" [{team_role}] Backoff active until {until:.0f}, using seeds only") | |
| return await _refill_with_seeds_only(team_role, count) | |
| base_count = await get_pool_count(team_role) | |
| needed = max(0, POOL_TARGET - base_count) | |
| target = min(count, needed) | |
| if target <= 0: | |
| return 0 | |
| recent_titles = [] | |
| try: | |
| from app.services.dedup import fetch_existing_titles | |
| recent_titles = fetch_existing_titles("vulnerability_hunter_challenges", team_role, limit=30) | |
| except Exception as e: | |
| print(f"[vuln-hunter] Failed to fetch recent titles: {e}") | |
| blacklisted_titles_str = "\n".join([f"- {t}" for t in recent_titles]) if recent_titles else "لا يوجد عناوين سابقة حالياً." | |
| async def _gen_one(i: int) -> bool: | |
| lang, vuln, diff = _pick_rotation_slot(int(time.time()) + i) | |
| print(f" [{team_role}] Generating: {lang} / {vuln} / {diff}") | |
| challenge = await ai_generate_challenge(lang, vuln, diff, blacklisted_titles=blacklisted_titles_str) | |
| if challenge: | |
| if await _insert_to_db(challenge): | |
| print(f" [+] Inserted AI: {challenge['title']}") | |
| return True | |
| seeds = _build_seeds() | |
| matching = [s for s in seeds if s["difficulty"] == diff] | |
| if not matching: | |
| matching = seeds | |
| seed = random.choice(matching) | |
| rand_tag = uuid.uuid4().hex[:6] | |
| seed = dict(seed) | |
| seed["title"] = f"{seed.get('title', 'تحدي ثغرة')} - رمز {rand_tag}" | |
| seed["story"] = seed.get("story", "") + f" [معرف التدقيق: {rand_tag}]" | |
| if await _insert_to_db(seed): | |
| print(f" [+] Inserted seed fallback ({diff})") | |
| return True | |
| return False | |
| results = [] | |
| for i in range(target): | |
| ok = await _gen_one(i) | |
| results.append(ok) | |
| await asyncio.sleep(2.0) | |
| return sum(1 for r in results if r) | |
| async def refill_pool(team_role: str, count: int) -> int: | |
| async with _get_pool_lock(team_role): | |
| return await _refill_pool_inner(team_role, count) | |
| # --------------------------------------------------------------------------- # | |
| # 6. Pool Watcher | |
| # --------------------------------------------------------------------------- # | |
| async def start_pool_watcher(team_role: str): | |
| if team_role not in ALLOWED_TEAMS: | |
| return | |
| if team_role in _WATCHER_STARTED: | |
| print(f"[vuln-hunter:{team_role}] watcher already running — skipping duplicate start.") | |
| return | |
| _WATCHER_STARTED.add(team_role) | |
| print(f"[vuln-hunter] Pool watcher started for '{team_role}' (target={POOL_TARGET}, threshold={POOL_THRESHOLD}, batch={POOL_BATCH})") | |
| try: | |
| while True: | |
| try: | |
| sleep_secs = 30 | |
| async with _get_pool_lock(team_role): | |
| count = await get_pool_count(team_role) | |
| if count <= POOL_THRESHOLD: | |
| print(f"[vuln-hunter] Pool at {count}/{POOL_TARGET} (≤ {POOL_THRESHOLD}), refilling {POOL_BATCH}…") | |
| added = await _refill_pool_inner(team_role, POOL_BATCH) | |
| print(f"[vuln-hunter] Refill done: +{added} (now {count + added}/{POOL_TARGET})") | |
| sleep_secs = 2 | |
| await asyncio.sleep(sleep_secs) | |
| except Exception as e: | |
| print(f"[vuln-hunter] Watcher error: {e}") | |
| await asyncio.sleep(10) | |
| finally: | |
| _WATCHER_STARTED.discard(team_role) | |
| # --------------------------------------------------------------------------- # | |
| # 7. CLI | |
| # --------------------------------------------------------------------------- # | |
| def main(): | |
| parser = argparse.ArgumentParser(description="Vulnerability Hunter Challenge Generator") | |
| parser.add_argument("--team", default="blue", choices=["blue"]) | |
| parser.add_argument("--ai", action="store_true", help="Use AI generation") | |
| parser.add_argument("--count", type=int, default=3, help="Number to generate") | |
| parser.add_argument("--seed-only", action="store_true", help="Insert seeds only") | |
| args = parser.parse_args() | |
| print(f"[vuln-hunter] Team: {args.team}, Count: {args.count}") | |
| if args.seed_only: | |
| seeds = _build_seeds() | |
| random.shuffle(seeds) | |
| inserted = 0 | |
| for seed in seeds[:args.count]: | |
| import asyncio as _aio | |
| if _aio.run(_insert_to_db(seed)): | |
| inserted += 1 | |
| print(f" [+] Inserted seed: {seed['title']}") | |
| print(f"[vuln-hunter] Seeded {inserted} challenges") | |
| elif args.ai: | |
| inserted = _aio.run(refill_pool(args.team, args.count)) | |
| print(f"[vuln-hunter] AI generated {inserted} challenges") | |
| else: | |
| _aio.run(start_pool_watcher(args.team)) | |
| if __name__ == "__main__": | |
| main() | |