Spaces:
Sleeping
Sleeping
File size: 9,205 Bytes
80a4a65 bef7f19 80a4a65 bef7f19 80a4a65 bef7f19 80a4a65 bef7f19 80a4a65 bef7f19 80a4a65 bef7f19 80a4a65 4871da9 cb16781 bef7f19 cb16781 4871da9 bef7f19 4871da9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 | """Map raw Supabase rows to the ``TrainingData`` shape the frontend expects.
Each ``_map_*_row_to_training(row, team_role) -> dict`` function knows
the columns of one challenge table and returns a consistent object so
the frontend editor can render any challenge type.
"""
import base64 as _b64
import json
_CODE_LANG_BY_EXT = {
"html": "html", "js": "javascript", "ts": "typescript",
"py": "python", "json": "json", "csv": "text",
"pem": "text", "log": "text", "txt": "text", "bin": "binary",
}
def _parse_hints(raw) -> list:
if isinstance(raw, list):
return raw
if isinstance(raw, str):
try:
parsed = json.loads(raw)
return parsed if isinstance(parsed, list) else []
except (json.JSONDecodeError, TypeError):
return []
return []
def _infer_code_language(filename, team_role: str) -> str:
if not filename:
return "text"
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
return _CODE_LANG_BY_EXT.get(ext, "text")
def map_encryption_row_to_training(row: dict, team_role: str) -> dict:
"""Map a row from ``encryption_challenges`` to TrainingData."""
files = row.get("files") or {}
file_meta = row.get("file_metadata") or {}
first_filename = next(iter(files), None)
code = ""
if first_filename:
try:
code = _b64.b64decode(files[first_filename]).decode("utf-8", errors="replace")
except Exception:
code = ""
return {
"id": row.get("id"),
"scenarioId": row.get("id"),
"title": row.get("title", ""),
"story": row.get("story", ""),
# Challenge type is always 'crypto' for rows coming from
# encryption_challenges — the ``module`` column is no longer
# authoritative after the 011 normalization migration.
"type": "crypto",
"topic": row.get("topic") or row.get("module", "encryption-basics"),
"task": row.get("task_outline", ""),
"code": code,
"codeLanguage": _infer_code_language(first_filename, team_role),
"htmlPreview": code if (first_filename or "").endswith(".html") else None,
"logData": code if (first_filename or "").endswith((".log", ".txt")) else None,
"configData": code if (first_filename or "").endswith((".json", ".csv", ".pem")) else None,
"vulnerabilityLocation": None,
"hints": _parse_hints(row.get("hints")),
"expectedAnswer": row.get("flag_preview", ""), # CyberArena{...}
"expectedAnswerHash": row.get("flag_hash", ""), # server-side check
"explanation": "العلم يظهر في مخرجات الطرفية بعد تنفيذ الأمر الصحيح.",
"xpReward": row.get("xp_reward", 100),
"difficulty": row.get("difficulty", "متوسط"),
"files": files,
"fileMetadata": file_meta,
"commandOutputs": row.get("command_outputs") or {},
"toolsWhitelist": row.get("tools_whitelist") or [],
"challengeType": "crypto",
}
def map_code_fixing_row_to_training(row: dict, team_role: str) -> dict:
"""Map a row from ``code_fixing_challenges`` to TrainingData."""
return {
"id": row.get("id"),
"scenarioId": row.get("id"),
"title": row.get("title", ""),
"story": row.get("story", ""),
"type": "code-fixing",
"topic": row.get("topic") or row.get("module", "code-fixing"),
"task": row.get("task_outline", ""),
# Additional fields for code-fixing
"language": row.get("language", "PYTHON"),
"vulnerable_code": row.get("vulnerable_code", ""),
"vulnerability_type": row.get("vulnerability_type", ""),
"vulnerability_description": row.get("vulnerability_description", ""),
"difficulty": row.get("difficulty", "متوسط"),
"xpReward": row.get("xp_reward", 150),
"hints": _parse_hints(row.get("hints")),
}
def map_log_analysis_row_to_training(row: dict, team_role: str) -> dict:
"""Map a row from ``log_analysis_challenges`` to TrainingData."""
from app.core.config import SUPABASE_URL
storage_path = row.get("storage_path", "")
is_inline = storage_path.startswith("inline://")
actual_path = storage_path.replace("inline://", "") if is_inline else storage_path
log_url = ""
if not is_inline and SUPABASE_URL and actual_path:
log_url = f"{SUPABASE_URL}/storage/v1/object/public/log-analysis-files/{actual_path}"
return {
"id": row.get("id"),
"scenarioId": row.get("id"),
"title": row.get("title", ""),
"story": row.get("story", ""),
"type": "log-analysis",
"topic": row.get("topic") or row.get("module", "log-analysis"),
"task": row.get("task_outline", ""),
# Log-analysis specific
"log_type": row.get("log_type", "auth"),
"storage_path": storage_path,
"log_url": log_url,
"is_inline": is_inline,
"file_size_bytes": row.get("file_size_bytes", 0),
"vulnerability_description": row.get("vulnerability_description", ""),
"difficulty": row.get("difficulty", "متوسط"),
"xpReward": row.get("xp_reward", 150),
"hints": _parse_hints(row.get("hints")),
}
def map_vuln_hunter_row_to_training(row: dict, team_role: str) -> dict:
"""Map a row from ``vulnerability_hunter_challenges`` to TrainingData.
Consumed by the Cyberpunk-themed VulnerabilityHunterEditor.
"""
return {
"id": row.get("id"),
"scenarioId": row.get("id"),
"title": row.get("title", ""),
"story": row.get("story", ""),
"type": "vulnerability-hunter",
"topic": row.get("topic") or row.get("module", "vulnerability-hunter"),
"task": row.get("task_outline", ""),
"language": row.get("language", "PYTHON"),
"vulnerable_code": row.get("vulnerable_code", ""),
"vulnerability_type": row.get("vulnerability_type", ""),
"vulnerability_class": row.get("vulnerability_class", ""),
"vulnerability_description": row.get("vulnerability_description", ""),
"difficulty": row.get("difficulty", "متوسط"),
"xpReward": row.get("xp_reward", 150),
"hints": _parse_hints(row.get("hints")),
}
def map_web_exploit_row_to_training(row: dict, team_role: str) -> dict:
"""Map a row from ``web_exploitation_challenges`` to TrainingData."""
return {
"id": row.get("id"),
"scenarioId": row.get("id"),
"title": row.get("title", ""),
"story": row.get("story", ""),
"type": "web-exploitation",
"topic": row.get("topic") or row.get("module", "web-exploitation"),
"task": row.get("task_outline", ""),
"vulnerability_type": row.get("vulnerability_type", ""),
"vulnerability_class": row.get("vulnerability_class", ""),
"vulnerability_description": row.get("vulnerability_description", ""),
"http_request": row.get("http_request", ""),
"http_response": row.get("http_response", ""),
"flag_preview": row.get("flag_preview", ""),
"flag_hash": row.get("flag_hash", ""),
"difficulty": row.get("difficulty", "متوسط"),
"xpReward": row.get("xp_reward", 200),
"hints": _parse_hints(row.get("hints")),
"challengeType": "web-exploitation",
}
def map_steganography_row_to_training(row: dict, team_role: str) -> dict:
"""Map a row from ``steganography_challenges`` to TrainingData."""
files = row.get("files") or {}
file_meta = row.get("file_metadata") or {}
first_filename = next(iter(files), None)
code = ""
if first_filename:
try:
# Safely attempt to decode as UTF-8, but fall back or truncate for binaries
code = _b64.b64decode(files[first_filename]).decode("utf-8", errors="ignore")[:300]
except Exception:
code = ""
return {
"id": row.get("id"),
"scenarioId": row.get("id"),
"title": row.get("title", ""),
"story": row.get("story", ""),
"type": "steganography",
"topic": row.get("topic") or row.get("module", "steganography"),
"task": row.get("task_outline", ""),
"code": code,
"codeLanguage": _infer_code_language(first_filename, team_role),
"htmlPreview": None,
"logData": None,
"configData": None,
"vulnerabilityLocation": None,
"hints": _parse_hints(row.get("hints")),
"expectedAnswer": row.get("flag_preview", ""), # CyberArena{...}
"expectedAnswerHash": row.get("flag_hash", ""), # server-side check
"explanation": "العلم مخفي داخل ملف الصورة المعطى. استخدم أدوات التحقيق مثل file أو strings أو unzip.",
"xpReward": row.get("xp_reward", 120),
"difficulty": row.get("difficulty", "متوسط"),
"files": files,
"fileMetadata": file_meta,
"commandOutputs": row.get("command_outputs") or {},
"toolsWhitelist": row.get("tools_whitelist") or [],
"challengeType": "steganography",
}
|