"""
Web Exploitation generator (Red Team) — produces diverse web-attack scenarios
with realistic HTTP request/response pairs, vulnerability descriptions, and
a flag-based answer.
The generator cycles through 10 attack types (XSS, SQLi, CSRF, SSRF, IDOR,
LFI, XXE, CMDi, Auth Bypass, File Upload) using creative seeds and AI calls.
"""
## Generators AGENTS.md contract:
## POOL_TARGET / POOL_THRESHOLD / POOL_BATCH (module-level)
## async get_pool_count(team_role) -> int
## async refill_pool(team_role, count) -> int
## async start_pool_watcher(team_role) # long-running asyncio Task
import asyncio
import hashlib
import json
import logging
import random
import time
from dataclasses import dataclass
from typing import Optional
import httpx
from app.core.config import (
SUPABASE_URL, SUPABASE_ANON_KEY,
CLOUDFLARE_API_TOKEN, CLOUDFLARE_ACCOUNT_ID, CLOUDFLARE_URL,
GROQ_API_KEY, GROQ_API_URL,
NVIDIA_API_KEY, NVIDIA_MODEL, NVIDIA_URL,
MISTRAL_API_KEY, MISTRAL_MODEL, MISTRAL_API_URL,
)
# ── Pool tuning ───────────────────────────────────────────────────────
POOL_TARGET = 5
POOL_THRESHOLD = 2
POOL_BATCH = 3
# ── AI backoff ─────────────────────────────────────────────────────────
_AI_BACKOFF_UNTIL: dict[str, float] = {}
# Concurrency control
_POOL_LOCKS: dict[str, asyncio.Lock] = {}
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]
# ── Topic rotation ─────────────────────────────────────────────────────
WEB_EXPLOIT_TOPICS = [
"xss", "sqli", "csrf", "ssrf", "idor",
"lfi", "xxe", "cmdi", "auth", "upload",
]
_TOPIC_INDEX: dict[str, int] = {} # team_role -> last index
# Cloudflare model fallback chain (same as code_fixing.py)
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",
]
_CURATED_SEEDS: list[dict] = []
# ── Data model ─────────────────────────────────────────────────────────
@dataclass
class WebExploitChallenge:
title: str
story: str
task_outline: str
vulnerability_type: str
vulnerability_class: str
vulnerability_description: str
http_request: str
http_response: str
flag_preview: str
flag_hash: str
hints: list[dict]
difficulty: str
xp_reward: int = 200
topic: str = ""
team_role: str = "red"
module: str = "web-exploitation"
# ── Helpers ────────────────────────────────────────────────────────────
def _sb_headers(content_type: bool = False) -> dict:
h = {
"apikey": SUPABASE_ANON_KEY,
"Authorization": f"Bearer {SUPABASE_ANON_KEY}",
}
if content_type:
h["Content-Type"] = "application/json"
h["Prefer"] = "return=representation"
return h
async def _supabase_select(table: str, query: str) -> list[dict]:
url = f"{SUPABASE_URL}/rest/v1/{table}?{query}"
async with httpx.AsyncClient(timeout=20) as client:
r = await client.get(url, headers=_sb_headers())
if r.status_code == 200:
return r.json()
return []
async def _supabase_insert(table: str, row: dict) -> Optional[dict]:
url = f"{SUPABASE_URL}/rest/v1/{table}"
async with httpx.AsyncClient(timeout=20) as client:
r = await client.post(url, json=row, headers=_sb_headers(content_type=True))
if r.status_code in (200, 201):
rows = r.json()
return rows[0] if rows else None
logging.error(f"[web_exploit] insert error {r.status_code}: {r.text[:300]}")
return None
def _compute_flag_hash(flag: str) -> str:
return hashlib.sha256(flag.encode("utf-8")).hexdigest()
def _generate_flag() -> str:
"""Generate a unique CyberArena{...} flag."""
rand_hex = hashlib.sha256(str(random.getrandbits(256)).encode()).hexdigest()[:16]
return f"CyberArena{{{rand_hex}}}"
def _extract_json(text: str) -> Optional[dict]:
"""Robust JSON extraction from LLM response."""
import re as _re
text = text.strip()
# Try direct parse
try:
return json.loads(text)
except json.JSONDecodeError:
pass
# Try to find JSON block
m = _re.search(r"\{[\s\S]*\}", text)
if m:
try:
return json.loads(m.group())
except json.JSONDecodeError:
pass
return None
# ── Curated Seeds (6 diverse scenarios, one per method) ───────────────
def _build_seeds() -> list[dict]:
seeds = []
# 1. SQLi
flag1 = _generate_flag()
seeds.append({
"title": "ثغرة حقن SQL في لوحة تحكم المخزون",
"story": "أثناء اختبار اختراق لموقع متجر إلكتروني، اكتشفت أن صفحة عرض المنتجات تستخدم معامل `id` في عنوان URL بشكل مباشر في استعلام SQL دون تنقية. يمكنك استغلال هذه الثغرة لسحب جميع المنتجات المخفية والبيانات الحساسة.",
"task_outline": "حاول استغلال ثغرة SQL Injection في معامل id بالرابط /api/products?id=1. استخدم تقنية UNION لعرض جميع المنتجات المخفية واستخرج العَلم.",
"vulnerability_type": "sql-injection",
"vulnerability_class": "Injection",
"vulnerability_description": "ثغرة SQL Injection تسمح للمهاجم بحقن استعلامات SQL خبيثة عبر مدخلات المستخدم، مما قد يؤدي إلى تسرب البيانات أو تدمير قاعدة البيانات.",
"http_request": "GET /api/products?id=1 UNION SELECT 1,2,3,4,5,6-- HTTP/1.1\nHost: target.apex-train.com\nUser-Agent: Mozilla/5.0\nAccept: application/json",
"http_response": "HTTP/1.1 200 OK\nContent-Type: application/json\n\n[{\"id\":1,\"name\":\"لابتوب\",\"price\":2500},{\"id\":2,\"name\":\"هاتف\",\"price\":1200},{\"id\":999,\"name\":\"FLAG_PLACEHOLDER\",\"price\":0}]",
"flag_preview": flag1,
"flag_hash": _compute_flag_hash(flag1),
"hints": [
{"level": 1, "text": "الحقن موجود في معامل id. جرب إضافة علامات اقتباس بسيطة لتكسير الاستعلام.", "xp_cost": 10},
{"level": 2, "text": "استخدم UNION SELECT لتوحيد نتائج استعلامك مع الاستعلام الأصلي. جرب: 1 UNION SELECT 1,2,3,4,5,6--", "xp_cost": 20},
{"level": 3, "text": "قد تحتاج إلى تحديد عدد الأعمدة أولاً باستخدام ORDER BY. العلم موجود في أحد الصفوف المخفية.", "xp_cost": 30},
],
"difficulty": "متوسط",
})
# 2. XSS
flag2 = _generate_flag()
seeds.append({
"title": "ثغرة XSS في منصة التعليقات",
"story": "تتضمن منتديات النقاش العامة ثغرة من نوع Cross-Site Scripting حيث يتم عرض محتوى التعليقات دون تعقيم. يمكنك حقن حمولة XSS لسرقة جلسة المسؤول.",
"task_outline": "أرسل تعليقاً يحتوي على حمولة XSS لتنبيه نافذة تحتوي على العَلم. استخدم \"}",
"http_response": "HTTP/1.1 201 Created\nContent-Type: application/json\n\n{\"status\":\"ok\",\"comment\":\"\"}",
"flag_preview": flag2,
"flag_hash": _compute_flag_hash(flag2),
"hints": [
{"level": 1, "text": "ابحث عن حقل الإدخال الذي يرسل النص دون تنقية. جرب .", "xp_cost": 10},
{"level": 2, "text": "إذا كان