"""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", }