H

app/api/training.py

@@ -18,6 +18,7 @@ from app.services.challenge_loader import (
     map_code_fixing_row_to_training,
     map_log_analysis_row_to_training,
     map_vuln_hunter_row_to_training,
+    map_steganography_row_to_training,
 )
 from app.services.evaluator import (
     evaluate_training,
@@ -75,7 +76,7 @@ async def list_challenges(team_role: str = "blue", difficulty: Optional[str] = N
     if team_role == "blue":
         candidates = ["code-fixing", "log-analysis", "vulnerability-hunter"]
     else:
-        candidates = ["crypto"]
+        candidates = ["crypto", "steganography"]
 
     items: list[dict] = []
     per_type_cap = max(1, limit // max(1, len(candidates)))
@@ -189,6 +190,7 @@ _TYPE_TO_MAPPER = {
     "code-fixing":          "map_code_fixing_row_to_training",
     "log-analysis":         "map_log_analysis_row_to_training",
     "vulnerability-hunter": "map_vuln_hunter_row_to_training",
+    "steganography":        "map_steganography_row_to_training",
 }
 
 
@@ -235,6 +237,7 @@ _CANONICAL_CHALLENGE_TYPES: dict[str, str] = {
     "log_analysis": "log-analysis",
     "vulnerability-hunter": "vulnerability-hunter",
     "vulnerability_hunter": "vulnerability-hunter",
+    "steganography": "steganography",
 }
 
 
@@ -258,6 +261,7 @@ def _tables_for_team(team_role: str) -> list[str]:
     if team_role == "red":
         return [
             "encryption_challenges",
+            "steganography_challenges",
         ]
     return [
         "code_fixing_challenges",
@@ -368,4 +372,5 @@ __all__ = [
     "map_code_fixing_row_to_training",
     "map_log_analysis_row_to_training",
     "map_vuln_hunter_row_to_training",
+    "map_steganography_row_to_training",
 ]

app/core/config.py

@@ -71,6 +71,7 @@ TABLE_CANONICAL_MODULE: dict[str, str] = {
     "code_fixing_challenges":          "code-fixing",
     "log_analysis_challenges":         "log-analysis",
     "vulnerability_hunter_challenges": "vulnerability-hunter",
+    "steganography_challenges":        "steganography",
 }
 
 

app/core/module_router.py

@@ -16,4 +16,6 @@ def challenge_type_for_module(module: str) -> str:
     """
     if module in WEB_EXPLOIT_MODULES:
         return "web"
+    if module == "steganography":
+        return "steganography"
     return "crypto"

app/generators/__init__.py

@@ -16,6 +16,7 @@ from . import crypto
 from . import code_fixing
 from . import log_analysis
 from . import vulnerability_hunter
+from . import steganography
 
 # The orchestrator in app.main iterates this registry. A new challenge
 # type is added by appending a tuple below (and a matching table in
@@ -26,6 +27,7 @@ REGISTRY = [
     ("code-fixing",          code_fixing,           ["blue"]),
     ("log-analysis",         log_analysis,          ["blue"]),
     ("vulnerability-hunter", vulnerability_hunter,  ["blue"]),
+    ("steganography",        steganography,         ["red"]),
 ]
 
 __all__ = [
@@ -33,5 +35,6 @@ __all__ = [
     "code_fixing",
     "log_analysis",
     "vulnerability_hunter",
+    "steganography",
     "REGISTRY",
 ]

app/generators/code_fixing.py

@@ -772,6 +772,13 @@ CRITICAL REQUIREMENTS:
 7. difficulty must be EXACTLY one of these three Arabic words: "مبتدئ", "متوسط", or "قوي"
 8. hints must be a JSON array of exactly 3 objects: [{{"level": 1, "text": "...", "xp_cost": 20}}, {{"level": 2, "text": "...", "xp_cost": 30}}, {{"level": 3, "text": "...", "xp_cost": 50}}]. The first hint should be vague, the second more specific, the third should hint at the actual fix approach.
 
+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" or "IoT controller").
+- NEVER reuse variable names, function names, or class 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.
+
 Return ONLY this exact JSON structure (no other text):
 {{"title": "عنوان التحدي بالعربية", "story": "قصة التحدي بالعربية", "task_outline": "المطلوب من المتدرب بالعربية", "vulnerable_code": "الكود الكامل المصاب بالثغرة", "vulnerability_description": "شرح الثغرة بالعربية", "difficulty": "متوسط", "hints": [{{"level": 1, "text": "تلميح أول", "xp_cost": 20}}, {{"level": 2, "text": "تلميح ثاني", "xp_cost": 30}}, {{"level": 3, "text": "تلميح ثالث", "xp_cost": 50}}]}}
 """
@@ -1117,6 +1124,18 @@ async def _insert_to_db(row: dict) -> bool:
     # like 'web-security' or 'systems-security'.
     from app.core.config import normalize_row_module
     row = normalize_row_module(TABLE_NAME, row)
+    # --- Deduplication: skip if a similar challenge already exists ---
+    try:
+        from app.services.dedup import is_duplicate_code_fixing
+        if is_duplicate_code_fixing(
+            row.get("title", ""),
+            row.get("vulnerable_code", ""),
+            role_filter=row.get("team_role"),
+        ):
+            print(f"  [dedup] Skipped duplicate: {row.get('title', '?')[:60]}")
+            return False
+    except Exception as _dedup_err:
+        print(f"  [dedup] Check failed (non-fatal): {_dedup_err}")
     url = f"{SUPABASE_URL}/rest/v1/{TABLE_NAME}"
     headers = supabase_headers(content_type=True)
     headers["Prefer"] = "return=representation"

app/generators/crypto.py

@@ -803,6 +803,12 @@ CRYPTO_SCENARIO_PROMPT = """أنت مصمم تحديات أمن سيبراني
 - اسم الملف الذي يحتوي المحتوى (مثال: intercept.txt، leak.bin، firmware.bin، contract.txt، payload.json): ضعه في `extra.filename` واستعمل نفس الاسم داخل story و task_outline بالضبط.
 - الـ hints: 3 تلميحات سقراطية قصيرة (جملة واحدة لكل تلميح) — توجّه الطالب للتفكير بدون إعطاء الجواب.
 
+禁止 التكرار — قواعد صارمة ضد التشابه:
+- لا تستخدم نفس السياق (embassy, leak, ransomware, military, hospital) أكثر من مرة.
+- غيّر الإعداد الجغرافي والزمني والمهني بالكامل عن أي تحدي سابق.
+- العنوان والقصة يجب أن يكونا مختلفين جذرياً — لا تكتفي بتغيير كلمة واحدة.
+- الـ plaintext يجب أن يكون محتوى مختلف تماماً (不同类型 من الملفات، محتوى مختلف).
+
 أمثلة على عناوين مميزة (استلهم منها، لا تنسخها حرفياً):
 - "اعتراض بث مشفر من ضابط ألماني في الـ Afrika Korps"
 - "تسريب رسالة بين CFO و broker في تحقيق SEC"
@@ -1146,6 +1152,18 @@ def insert_to_db(challenge: Challenge, supabase_url: str = SUPABASE_URL,
     # 'encryption-basics', 'hash-cracking' or 'rsa-aes'.
     from app.core.config import normalize_row_module
     payload = normalize_row_module("encryption_challenges", payload)
+    # --- Deduplication: skip if a similar challenge already exists ---
+    try:
+        from app.services.dedup import is_duplicate_encryption
+        if is_duplicate_encryption(
+            payload.get("title", ""),
+            payload.get("story", ""),
+            role_filter=payload.get("team_role"),
+        ):
+            print(f"  [dedup] Skipped duplicate: {payload.get('title', '?')[:60]}")
+            return False
+    except Exception as _dedup_err:
+        print(f"  [dedup] Check failed (non-fatal): {_dedup_err}")
     last_err: str = ""
     for attempt in range(1, 4):
         try:

app/generators/log_analysis.py

@@ -692,6 +692,13 @@ CRITICAL REQUIREMENTS:
 8. difficulty must be EXACTLY "مبتدئ" or "متوسط" or "قوي"
 9. hints must be 3 objects: [{{"level":1,"text":"hint 1","xp_cost":20}}, {{"level":2,"text":"hint 2","xp_cost":30}}, {{"level":3,"text":"hint 3","xp_cost":50}}]
 
+UNIQUENESS RULES — ZERO TOLERANCE FOR DUPLICATES:
+- Every challenge MUST be completely unique in title, story, log content, and IP addresses.
+- NEVER reuse the same server type or application domain. Vary between: web server, database, mail server, DNS, FTP, SSH, API gateway, container orchestrator, IoT device, etc.
+- Use DIFFERENT attacker IPs, different timestamps, different log formats each time.
+- The log_content MUST contain different benign traffic patterns and different attack payloads each time.
+- The title and story MUST describe a completely different investigation scenario.
+
 Return ONLY this JSON:
 {{"title":"...","story":"...","task_outline":"...","log_content":"...","vulnerability_description":"...","expected_attack_type":"...","expected_attacker_ip":"...","expected_timestamp":"...","expected_ioc":"...","difficulty":"متوسط","hints":[{{"level":1,"text":"...","xp_cost":20}},{{"level":2,"text":"...","xp_cost":30}},{{"level":3,"text":"...","xp_cost":50}}]}}
 """
@@ -1036,6 +1043,18 @@ async def _insert_to_db(row: dict) -> bool:
     # for consistency with the other 4 generators.
     from app.core.config import normalize_row_module
     row = normalize_row_module(TABLE_NAME, row)
+    # --- Deduplication: skip if a similar challenge already exists ---
+    try:
+        from app.services.dedup import is_duplicate_log_analysis
+        if is_duplicate_log_analysis(
+            row.get("title", ""),
+            row.get("log_content", ""),
+            role_filter=row.get("team_role"),
+        ):
+            print(f"  [dedup] Skipped duplicate: {row.get('title', '?')[:60]}")
+            return False
+    except Exception as _dedup_err:
+        print(f"  [dedup] Check failed (non-fatal): {_dedup_err}")
     url = f"{SUPABASE_URL}/rest/v1/{TABLE_NAME}"
     headers = supabase_headers(content_type=True)
     headers["Prefer"] = "return=representation"

app/generators/steganography.py

@@ -0,0 +1,660 @@
+"""
+steganography_generator.py
+==========================
+Generates dynamic steganography (information hiding) challenges for the CyberArena pool.
+
+Hiding methods:
+    1. EOF: Append the flag bytes to the end of the image file.
+    2. EXIF: Inject the flag into the JPEG Comment (COM) segment manually.
+    3. ZIP: Append a zip archive containing flag.txt to the image file.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import base64
+import hashlib
+import json
+import os
+import random
+import sys
+import uuid
+import zipfile
+import io
+from dataclasses import dataclass
+from typing import Optional, Literal
+
+import httpx
+
+# Load .env early
+try:
+    from app._env import load_app_env
+    load_app_env()
+except Exception:
+    pass
+
+from app.generators.crypto import _repair_json
+
+# --------------------------------------------------------------------------- #
+# 0. Constants
+# --------------------------------------------------------------------------- #
+
+ALLOWED_DIFFICULTIES = ("مبتدئ", "متوسط", "قوي")
+ALLOWED_TEAMS = ("red",)
+
+POOL_TARGET = 5
+POOL_THRESHOLD = 2
+POOL_BATCH = 3
+IDLE_POLL_SECS = 60
+WATCHER_POLL_SECS = 60
+
+_AI_BACKOFF_UNTIL: dict[str, float] = {}
+_POOL_LOCKS: dict[str, asyncio.Lock] = {}
+_WATCHER_STARTED: set[tuple[str, str]] = set()
+
+SUPABASE_URL = os.environ.get("SUPABASE_URL", "https://yevtnyokixocpihpdwqu.supabase.co")
+SUPABASE_ANON_KEY = os.environ.get("SUPABASE_ANON_KEY", "")
+
+# AI Configurations
+MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY", "")
+MISTRAL_MODEL = os.environ.get("MISTRAL_MODEL", "mistral-large-latest")
+MISTRAL_API_URL = os.environ.get("MISTRAL_API_URL", "https://api.mistral.ai/v1/chat/completions")
+
+CLOUDFLARE_API_TOKEN = os.environ.get("CLOUDFLARE_API_TOKEN", "")
+CLOUDFLARE_ACCOUNT_ID = os.environ.get("CLOUDFLARE_ACCOUNT_ID", "")
+CLOUDFLARE_MODEL = os.environ.get("CLOUDFLARE_MODEL", "@cf/meta/llama-3.1-8b-instruct")
+
+GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "")
+GROQ_MODEL = os.environ.get("GROQ_MODEL", "llama-3.1-8b-instant")
+GROQ_API_URL = "https://api.groq.com/openai/v1/chat/completions"
+
+NVIDIA_API_KEY = os.environ.get("NVIDIA_API_KEY", "")
+NVIDIA_MODEL = os.environ.get("NVIDIA_MODEL", "deepseek-ai/deepseek-v4-pro")
+NVIDIA_URL = "https://integrate.api.nvidia.com/v1/chat/completions"
+
+# Curated Unsplash images for cyber themes
+UNSPLASH_THEME_URLS = [
+    "https://images.unsplash.com/photo-1526374965328-7f61d4dc18c5?w=600&auto=format&fit=crop", # matrix/cyber
+    "https://images.unsplash.com/photo-1550751827-4bd374c3f58b?w=600&auto=format&fit=crop", # cybersecurity/servers
+    "https://images.unsplash.com/photo-1563986768609-322da13575f3?w=600&auto=format&fit=crop", # technology/abstract
+    "https://images.unsplash.com/photo-1510511459019-5dda7724fd87?w=600&auto=format&fit=crop", # hacker/code
+    "https://images.unsplash.com/photo-1601597111158-2fceff292cdc?w=600&auto=format&fit=crop", # network switch
+    "https://images.unsplash.com/photo-1526374865147-73cf412a95c5?w=600&auto=format&fit=crop", # cyber circuit
+    "https://images.unsplash.com/photo-1515879218367-8466d910aaa4?w=600&auto=format&fit=crop", # hacking terminal
+    "https://images.unsplash.com/photo-1517694712202-14dd9538aa97?w=600&auto=format&fit=crop", # laptop code
+]
+
+XP_BY_DIFFICULTY = {"مبتدئ": 100, "متوسط": 150, "قوي": 200}
+
+# --------------------------------------------------------------------------- #
+# 1. Hiding Algorithms
+# --------------------------------------------------------------------------- #
+
+def generate_fallback_image_bytes(text: str = "APEX CyberArena") -> bytes:
+    """Generate a valid dark theme JPEG image using PIL (Pillow) or a tiny binary fallback."""
+    try:
+        from PIL import Image, ImageDraw
+        img = Image.new("RGB", (500, 500), color=(11, 11, 18))
+        draw = ImageDraw.Draw(img)
+        # Grid lines
+        for i in range(0, 500, 50):
+            draw.line([(i, 0), (i, 500)], fill=(0, 212, 170, 30), width=1)
+            draw.line([(0, i), (500, i)], fill=(255, 0, 85, 30), width=1)
+        draw.ellipse([(150, 150), (350, 350)], outline=(0, 212, 170), width=2)
+        draw.text((20, 20), text, fill=(255, 255, 255))
+        draw.text((20, 450), "Apex training artifact", fill=(120, 120, 120))
+        
+        buf = io.BytesIO()
+        img.save(buf, format="JPEG")
+        return buf.getvalue()
+    except Exception:
+        # Minimal 1x1 valid JPEG bytes fallback
+        return (
+            b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x01\x00`\x00`\x00\x00"
+            b"\xff\xdb\x00C\x00\x08\x06\x06\x07\x06\x05\x08\x07\x07\x07\t\t\x08"
+            b"\n\x0c\x14\r\x0c\x0b\x0b\x0c\x19\x12\x13\x0f\x14\x1d\x1a\x1f\x1e"
+            b"\x1d\x1a\x1c\x1c $.' \",#\x1c\x1c(7),01444\x1f'9=82<.342"
+            b"\xff\xc0\x00\x0b\x08\x00\x01\x00\x01\x01\x01\x11\x00\xff\xc4\x00"
+            b"\x1f\x00\x00\x01\x05\x01\x01\x01\x01\x01\x01\x00\x00\x00\x00\x00"
+            b"\x00\x00\x00\x01\x02\x03\x04\x05\x06\x07\x08\t\n\x0b\xff\xc4\x00"
+            b"\xb5\x10\x00\x02\x01\x03\x03\x02\x04\x03\x05\x05\x04\x04\x00\x00"
+            b"\x00\x00\x01\x02\x03\x00\x04\x11\x05\x12!1A\x06\x13"
+            b"Qa\x07\"q\x142\x81\x91\xa1\x08#B\xb1\xc1\x15R\xd1\xf0"
+            b"$3br\x82\x16\x17\x18\x19\x1a%&'()*456789:CDEFGHIJSTUVWXYZ"
+            b"cdefghijstuvwxyz\x83\x84\x85\x86\x87\x88\x89\x8a\x92\x93\x94\x95"
+            b"\x96\x97\x98\x99\x9a\xa2\xa3\xa4\xa5\xa6\xa7\xa8\xa9\xaa\xb2\xb3"
+            b"\xb4\xb5\xb6\xb7\xb8\xb9\xba\xc2\xc3\xc4\xc5\xc6\xc7\xc8\xc9\xca"
+            b"\xd2\xd3\xd4\xd5\xd6\xd7\xd8\xd9\xda\xe2\xe3\xe4\xe5\xe6\xe7\xe8"
+            b"\xe9\xea\xf2\xf3\xf4\xf5\xf6\xf7\xf8\xf9\xfa\xff\xda\x00\x0c\x03"
+            b"\x01\x00\x02\x11\x03\x11\x00?\x00\xed\xfc\xff\xd9"
+        )
+
+async def fetch_random_unsplash_image() -> bytes:
+    """Fetch an image from Unsplash. Falls back to generating one if offline."""
+    url = random.choice(UNSPLASH_THEME_URLS)
+    try:
+        async with httpx.AsyncClient(timeout=10) as client:
+            r = await client.get(url)
+            if r.status_code == 200:
+                return r.content
+    except Exception as e:
+        print(f"[steganography] Unsplash fetch failed: {e}")
+    return generate_fallback_image_bytes()
+
+def hide_flag_eof(image_bytes: bytes, flag: str) -> bytes:
+    """Hiding method 1: Append flag directly to EOF."""
+    return image_bytes + b"\n" + flag.encode("utf-8") + b"\n"
+
+def hide_flag_exif(image_bytes: bytes, flag: str) -> bytes:
+    """Hiding method 2: Inject flag into standard JPEG comment segment manually."""
+    if image_bytes.startswith(b"\xff\xd8"):
+        comment_bytes = flag.encode("utf-8")
+        length = len(comment_bytes) + 2
+        length_bytes = length.to_bytes(2, byteorder='big')
+        # FF FE is the JPEG COM marker
+        com_segment = b"\xff\xfe" + length_bytes + comment_bytes
+        # Insert segment right after the SOI marker (FF D8)
+        return image_bytes[0:2] + com_segment + image_bytes[2:]
+    return hide_flag_eof(image_bytes, flag)
+
+def hide_flag_zip(image_bytes: bytes, flag: str) -> bytes:
+    """Hiding method 3: Append a zip containing flag.txt to the end of the image."""
+    zip_buf = io.BytesIO()
+    with zipfile.ZipFile(zip_buf, 'w', zipfile.ZIP_DEFLATED) as zip_file:
+        zip_file.writestr("flag.txt", flag + "\n")
+    return image_bytes + zip_buf.getvalue()
+
+def hide_flag(image_bytes: bytes, method: str, flag: str) -> bytes:
+    if method == "EOF":
+        return hide_flag_eof(image_bytes, flag)
+    elif method == "EXIF":
+        return hide_flag_exif(image_bytes, flag)
+    elif method == "ZIP":
+        return hide_flag_zip(image_bytes, flag)
+    raise ValueError(f"Unknown hiding method: {method}")
+
+# --------------------------------------------------------------------------- #
+# 2. Challenge Dataclasses
+# --------------------------------------------------------------------------- #
+
+@dataclass
+class ScenarioSpec:
+    team_role: str
+    module: str
+    difficulty: str
+    title: str
+    story: str
+    task_outline: str
+    hiding_method: str
+    hints: list
+
+@dataclass
+class Challenge:
+    team_role: str
+    module: str
+    title: str
+    story: str
+    task_outline: str
+    files: dict
+    file_metadata: dict
+    command_outputs: dict
+    hints: list
+    tools_whitelist: list
+    flag_hash: str
+    flag_preview: str
+    difficulty: str
+    xp_reward: int
+    hiding_method: str
+    image_url: str
+    topic: str
+
+    def to_db_row(self) -> dict:
+        return {
+            "team_role": self.team_role,
+            "module": self.module,
+            "title": self.title,
+            "story": self.story,
+            "task_outline": self.task_outline,
+            "files": self.files,
+            "file_metadata": self.file_metadata,
+            "command_outputs": self.command_outputs,
+            "hints": self.hints,
+            "tools_whitelist": self.tools_whitelist,
+            "flag_hash": self.flag_hash,
+            "flag_preview": self.flag_preview,
+            "difficulty": self.difficulty,
+            "xp_reward": self.xp_reward,
+            "hiding_method": self.hiding_method,
+            "image_url": self.image_url,
+            "topic": self.topic,
+        }
+
+# --------------------------------------------------------------------------- #
+# 3. Curated Seeds
+# --------------------------------------------------------------------------- #
+
+CURATED_SEEDS = [
+    {
+        "difficulty": "مبتدئ",
+        "hiding_method": "EOF",
+        "title": "شفرة الملفات المتروكة",
+        "story": "اعترض فريق المراقبة الأمنية صورة مشبوهة باسم suspect.jpg تم رفعها إلى خادم غير مصرح به. يُعتقد أن المهاجم استغل ثغرة وقام بلصق العلم في نهاية ملف الصورة دون تعديل واصفاتها.",
+        "task_outline": "قم بفحص الملف suspect.jpg المتاح في ساندبوكس التدريب. ابحث عن نصوص نصية معلقة في نهاية الصورة. العلم مخفي في نهاية ملف الصورة.",
+        "hints": [
+            "الصورة هي ملف ثنائي، ولكن يمكنك قراءة النصوص المكتوبة بداخلها باستخدام أداة strings.",
+            "قم بتنفيذ strings suspect.jpg في الطرفية لمشاهدة جميع النصوص القابلة للقراءة."
+        ]
+    },
+    {
+        "difficulty": "متوسط",
+        "hiding_method": "ZIP",
+        "title": "الحقيبة المزدوجة",
+        "story": "عثر فريق الاستجابة للحوادث على خادم ويب مصاب. المهاجم قام بدمج أرشيف كامل مضغوط يحوي معلومات الاختراق داخل صورة التحدي challenge.jpg كتقنية للتمويه والتخفي.",
+        "task_outline": "افحص الملف challenge.jpg في ساندبوكس التدريب لتحديد ما إذا كان يحوي أرشيفاً مضغوطاً مدمجاً. استخرج الأرشيف لتحصل على ملف flag.txt.",
+        "hints": [
+            "أداة file يمكنها كشف البنية الحقيقية للملف وتحديد إذا ما كان هناك أرشيف مضغوط appended.",
+            "جرب استخدام أداة فك الضغط unzip challenge.jpg مباشرة لاستخراج الملفات المخفية."
+        ]
+    },
+    {
+        "difficulty": "قوي",
+        "hiding_method": "EXIF",
+        "title": "واصفات العميل الوصفية",
+        "story": "أرسل عميل سري صورة طبيعية باسم agent.jpg إلى مركز العمليات. تحليل واصفات الملف أظهر أن الصورة لا تحتوي على أي تعديل في البيانات الثنائية الملحقة، ولكن تم استغلال حقل التعليقات (Comment) في EXIF Metadata لنقل رسالة سرية.",
+        "task_outline": "افحص بيانات واصفات الملف agent.jpg واستخرج العلم المخفي داخل حقل التعليق (COM/EXIF metadata).",
+        "hints": [
+            "البيانات الوصفية للصور (Metadata) تخزن معلومات إضافية مثل الكاميرا أو تعليقات المبرمج.",
+            "أداة strings يمكنها طباعة التعليق من داخل البنية الرأسية للملف مباشرة، أو يمكنك قراءة واصفات الملف."
+        ]
+    }
+]
+
+# --------------------------------------------------------------------------- #
+# 4. Challenge Builder
+# --------------------------------------------------------------------------- #
+
+class ChallengeBuilder:
+    async def build(self, spec: ScenarioSpec, image_bytes: bytes) -> Challenge:
+        flag_preview = f"CyberArena{{{uuid.uuid4().hex[:16]}}}"
+        flag_hash = hashlib.sha256(flag_preview.encode()).hexdigest()
+
+        # Filename based on method
+        filename = "suspect.jpg"
+        if spec.hiding_method == "ZIP":
+            filename = "challenge.jpg"
+        elif spec.hiding_method == "EXIF":
+            filename = "agent.jpg"
+
+        file_bytes = hide_flag(image_bytes, spec.hiding_method, flag_preview)
+        files = {
+            filename: base64.b64encode(file_bytes).decode("utf-8")
+        }
+
+        # Command outputs metadata for UI cheat sheets / dashboard previews
+        command_outputs = {
+            "ls": {"stdout": filename, "stderr": "", "exit_code": 0},
+            f"file {filename}": {
+                "stdout": f"{filename}: JPEG image data, JFIF standard 1.01" + (
+                    " (with appended Zip archive data)" if spec.hiding_method == "ZIP" else ""
+                ),
+                "stderr": "",
+                "exit_code": 0
+            }
+        }
+
+        return Challenge(
+            team_role=spec.team_role,
+            module="steganography",
+            title=spec.title,
+            story=spec.story,
+            task_outline=spec.task_outline,
+            files=files,
+            file_metadata={
+                filename: {"encoding": "binary", "size": len(file_bytes)}
+            },
+            command_outputs=command_outputs,
+            hints=spec.hints,
+            tools_whitelist=["cat", "ls", "file", "strings", "unzip"],
+            flag_hash=flag_hash,
+            flag_preview=flag_preview,
+            difficulty=spec.difficulty,
+            xp_reward=XP_BY_DIFFICULTY.get(spec.difficulty, 150),
+            hiding_method=spec.hiding_method,
+            image_url="",
+            topic="steganography"
+        )
+
+# --------------------------------------------------------------------------- #
+# 5. AI Generation Prompts
+# --------------------------------------------------------------------------- #
+
+STEGANOGRAPHY_PROMPT = """أنت كبير مصممي سيناريوهات الأمن السيبراني في منصة APEX.
+مهمتك: تصميم سيناريو تدريبي للتحقيق الجنائي الرقمي وإخفاء المعلومات (Steganography) للفريق الأحمر.
+
+الموضوع: إخفاء المعلومات في الوسائط (Steganography)
+مستوى الصعوبة: {difficulty}
+الطريقة المحددة للإخفاء: {hiding_method}
+
+السيناريو يجب أن يكون بلغة عربية فصحى احترافية ومشوقة.
+1. العنوان: عنوان غامض ومبتكر باللغة العربية (مثل: "تسريب الظلال"، "الرسالة المشفرة للعميل"، "الملف الملتحم").
+2. القصة: قصة سياقية واقعية (3-4 جمل) عن حادثة أمنية، تسريب، أو عميل يقوم بنقل رسائل سرية مخفية في صورة.
+3. المهمة: اصف للمستخدم ما يجب القيام به باللغة العربية. تجنب ذكر الطريقة المباشرة (مثل "افتح unzip" أو "شغل strings") ولكن أشر لها بأسلوب استكشافي مشوق.
+
+أرجع JSON فقط بدون أي نص قبله أو بعده وبدون كود التنسيق ```json:
+{{
+  "title": "عنوان التحدي باللغة العربية",
+  "story": "قصة السياق الكاملة باللغة العربية الفصحى",
+  "task_outline": "تفاصيل المهمة باللغة العربية الفصحى",
+  "hints": [
+    "تلميح ذكي أول يساعد اللاعب خطوة بخطوة",
+    "تلميح ذكي ثان يوجه اللاعب للأداة المناسبة"
+  ]
+}}"""
+
+async def ai_generate_scenario_via_mistral(team_role: str, difficulty: str, hiding_method: str) -> ScenarioSpec:
+    if not MISTRAL_API_KEY:
+        raise RuntimeError("Mistral API Key missing")
+    
+    prompt = STEGANOGRAPHY_PROMPT.format(difficulty=difficulty, hiding_method=hiding_method)
+    body = {
+        "model": MISTRAL_MODEL,
+        "messages": [
+            {"role": "system", "content": "You output valid JSON only. No prose, no markdown fences."},
+            {"role": "user", "content": prompt},
+        ],
+        "temperature": 0.7,
+        "response_format": {"type": "json_object"}
+    }
+    async with httpx.AsyncClient(timeout=30) as client:
+        r = await client.post(MISTRAL_API_URL, headers={"Authorization": f"Bearer {MISTRAL_API_KEY}"}, json=body)
+    
+    if r.status_code != 200:
+        raise RuntimeError(f"Mistral HTTP {r.status_code}")
+    
+    data = _repair_json(r.json()["choices"][0]["message"]["content"])
+    return ScenarioSpec(
+        team_role=team_role,
+        module="steganography",
+        difficulty=difficulty,
+        title=data["title"],
+        story=data["story"],
+        task_outline=data["task_outline"],
+        hiding_method=hiding_method,
+        hints=data.get("hints") or []
+    )
+
+async def ai_generate_scenario_via_cloudflare(team_role: str, difficulty: str, hiding_method: str) -> ScenarioSpec:
+    if not CLOUDFLARE_API_TOKEN or not CLOUDFLARE_ACCOUNT_ID:
+        raise RuntimeError("Cloudflare settings missing")
+    
+    url = f"https://api.cloudflare.com/client/v4/accounts/{CLOUDFLARE_ACCOUNT_ID}/ai/run/{CLOUDFLARE_MODEL}"
+    prompt = STEGANOGRAPHY_PROMPT.format(difficulty=difficulty, hiding_method=hiding_method)
+    
+    body = {
+        "messages": [
+            {"role": "system", "content": "You are a cybersecurity training scenario generator. Output JSON only."},
+            {"role": "user", "content": prompt}
+        ]
+    }
+    
+    async with httpx.AsyncClient(timeout=30) as client:
+        r = await client.post(url, headers={"Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}"}, json=body)
+        
+    if r.status_code != 200:
+        raise RuntimeError(f"Cloudflare HTTP {r.status_code}")
+        
+    data = _repair_json(r.json()["result"]["response"])
+    return ScenarioSpec(
+        team_role=team_role,
+        module="steganography",
+        difficulty=difficulty,
+        title=data["title"],
+        story=data["story"],
+        task_outline=data["task_outline"],
+        hiding_method=hiding_method,
+        hints=data.get("hints") or []
+    )
+
+async def ai_generate_scenario_via_groq(team_role: str, difficulty: str, hiding_method: str) -> ScenarioSpec:
+    if not GROQ_API_KEY:
+        raise RuntimeError("Groq key missing")
+        
+    prompt = STEGANOGRAPHY_PROMPT.format(difficulty=difficulty, hiding_method=hiding_method)
+    body = {
+        "model": GROQ_MODEL,
+        "messages": [
+            {"role": "system", "content": "You output valid JSON only."},
+            {"role": "user", "content": prompt}
+        ],
+        "temperature": 0.7,
+        "response_format": {"type": "json_object"}
+    }
+    
+    async with httpx.AsyncClient(timeout=30) as client:
+        r = await client.post(GROQ_API_URL, headers={"Authorization": f"Bearer {GROQ_API_KEY}"}, json=body)
+        
+    if r.status_code != 200:
+        raise RuntimeError(f"Groq HTTP {r.status_code}")
+        
+    data = _repair_json(r.json()["choices"][0]["message"]["content"])
+    return ScenarioSpec(
+        team_role=team_role,
+        module="steganography",
+        difficulty=difficulty,
+        title=data["title"],
+        story=data["story"],
+        task_outline=data["task_outline"],
+        hiding_method=hiding_method,
+        hints=data.get("hints") or []
+    )
+
+# --------------------------------------------------------------------------- #
+# 6. DB operations
+# --------------------------------------------------------------------------- #
+
+def get_pool_count(team_role: str, module: Optional[str] = None) -> int:
+    if not SUPABASE_ANON_KEY or not SUPABASE_URL:
+        return 0
+    url = f"{SUPABASE_URL}/rest/v1/steganography_challenges?select=id&team_role=eq.{team_role}"
+    try:
+        r = httpx.get(url, headers={
+            "apikey": SUPABASE_ANON_KEY,
+            "Authorization": f"Bearer {SUPABASE_ANON_KEY}"
+        }, timeout=10)
+        if r.status_code == 200:
+            return len(r.json())
+    except Exception as e:
+        print(f"[steganography] get_pool_count error: {e}")
+    return 0
+
+def insert_to_db(challenge: Challenge) -> bool:
+    if not SUPABASE_ANON_KEY or not SUPABASE_URL:
+        return False
+    
+    url = f"{SUPABASE_URL}/rest/v1/steganography_challenges"
+    headers = {
+        "apikey": SUPABASE_ANON_KEY,
+        "Authorization": f"Bearer {SUPABASE_ANON_KEY}",
+        "Content-Type": "application/json",
+        "Prefer": "return=minimal"
+    }
+    payload = challenge.to_db_row()
+    
+    from app.core.config import normalize_row_module
+    payload = normalize_row_module("steganography_challenges", payload)
+    
+    # Deduplication check
+    try:
+        from app.services.dedup import is_duplicate_steganography
+        if is_duplicate_steganography(payload.get("title", ""), payload.get("story", ""), payload.get("team_role")):
+            print(f"  [dedup] Skipped duplicate: {payload.get('title')[:50]}")
+            return False
+    except Exception as e:
+        print(f"  [dedup] Check error (non-fatal): {e}")
+
+    for attempt in range(3):
+        try:
+            with httpx.Client(timeout=20) as client:
+                r = client.post(url, headers=headers, json=payload)
+                if r.status_code in (200, 201, 204):
+                    print(f"✓ Inserted stego: {challenge.title[:50]}")
+                    return True
+                print(f"✗ Insert stego failed [{r.status_code}]: {r.text[:200]}")
+                return False
+        except Exception as e:
+            print(f"Attempt {attempt+1} failed: {e}")
+            time.sleep(2 ** attempt)
+    return False
+
+# --------------------------------------------------------------------------- #
+# 7. Refill and Watcher API
+# --------------------------------------------------------------------------- #
+
+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]
+
+async def refill_pool(team_role: str, count: int = POOL_BATCH) -> int:
+    lock = _get_pool_lock(team_role)
+    async with lock:
+        current = get_pool_count(team_role)
+        gap = max(0, POOL_TARGET - current)
+        if gap == 0:
+            return 0
+        if count > gap:
+            count = gap
+            
+        inserted = 0
+        methods = ["EOF", "ZIP", "EXIF"]
+        difficulties = ["مبتدئ", "متوسط", "قوي"]
+        
+        for i in range(count):
+            method = methods[i % len(methods)]
+            difficulty = difficulties[i % len(difficulties)]
+            spec = None
+            source = "ai"
+            
+            # AI Refill Loop
+            providers = [
+                ai_generate_scenario_via_mistral,
+                ai_generate_scenario_via_cloudflare,
+                ai_generate_scenario_via_groq
+            ]
+            for p in providers:
+                try:
+                    spec = await p(team_role, difficulty, method)
+                    print(f"[steganography] AI {p.__name__} OK: title={spec.title[:40]}")
+                    break
+                except Exception as e:
+                    print(f"[steganography] AI {p.__name__} failed: {e}")
+            
+            # Fallback Seed
+            if spec is None:
+                source = "seed"
+                seed = random.choice([s for s in CURATED_SEEDS if s["hiding_method"] == method] or CURATED_SEEDS)
+                spec = ScenarioSpec(
+                    team_role=team_role,
+                    module="steganography",
+                    difficulty=seed["difficulty"],
+                    title=seed["title"] + f" #{random.randint(10, 99)}",
+                    story=seed["story"],
+                    task_outline=seed["task_outline"],
+                    hiding_method=seed["hiding_method"],
+                    hints=seed["hints"]
+                )
+            
+            # Fetch stock photo and build
+            try:
+                img_bytes = await fetch_random_unsplash_image()
+                ch = await ChallengeBuilder().build(spec, img_bytes)
+                if insert_to_db(ch):
+                    inserted += 1
+            except Exception as e:
+                print(f"[steganography] Build/insert failed: {e}")
+                
+            await asyncio.sleep(0.3)
+            
+        return inserted
+
+async def start_pool_watcher(team_role: str) -> None:
+    key = ("steganography", team_role)
+    if key in _WATCHER_STARTED:
+        print(f"[steganography:{team_role}] watcher already running — skipping duplicate start.")
+        return
+    _WATCHER_STARTED.add(key)
+    try:
+        label = f"[steganography:{team_role}]"
+        print(f"{label} watcher started (target={POOL_TARGET}).")
+        while True:
+            try:
+                sleep_secs = IDLE_POLL_SECS
+                async with _get_pool_lock(team_role):
+                    count = get_pool_count(team_role)
+                    if count < POOL_TARGET:
+                        needed = POOL_TARGET - count
+                        print(f"{label} pool below target ({count}/{POOL_TARGET}) — refilling {needed}…")
+                        added = await refill_pool(team_role, needed)
+                        new_count = count + added
+                        print(f"{label} refilled: {count} → {new_count} (target {POOL_TARGET}).")
+                        sleep_secs = 5
+                await asyncio.sleep(sleep_secs)
+            except Exception as e:
+                import traceback
+                print(f"{label} watcher error: {e}")
+                traceback.print_exc()
+                await asyncio.sleep(10)
+    finally:
+        _WATCHER_STARTED.discard(key)
+
+# --------------------------------------------------------------------------- #
+# 8. CLI Parser
+# --------------------------------------------------------------------------- #
+
+def main():
+    try:
+        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
+    except Exception:
+        pass
+    p = argparse.ArgumentParser(description="Steganography challenge generator (RED TEAM ONLY)")
+    p.add_argument("--team", choices=("red",), default="red")
+    p.add_argument("--count", type=int, default=5)
+    p.add_argument("--refill", choices=("red",), help="Refill the team-wide pool")
+    p.add_argument("--dry-run", action="store_true", help="Build but don't insert")
+    args = p.parse_args()
+
+    if args.refill:
+        added = asyncio.run(refill_pool(args.refill, args.count))
+        print(f"Refilled {args.refill}: +{added}")
+        return
+
+    print(f"Building {args.count} challenges for RED team...")
+    
+    async def test_generation():
+        methods = ["EOF", "ZIP", "EXIF"]
+        for i in range(args.count):
+            method = methods[i % len(methods)]
+            seed = CURATED_SEEDS[i % len(CURATED_SEEDS)]
+            spec = ScenarioSpec(
+                team_role="red",
+                module="steganography",
+                difficulty=seed["difficulty"],
+                title=seed["title"] + " (Test)",
+                story=seed["story"],
+                task_outline=seed["task_outline"],
+                hiding_method=method,
+                hints=seed["hints"]
+            )
+            img = generate_fallback_image_bytes()
+            ch = await ChallengeBuilder().build(spec, img)
+            print(f"  • {ch.title} | {ch.difficulty} | hiding={ch.hiding_method} | flag={ch.flag_preview[:40]}...")
+            if not args.dry_run:
+                insert_to_db(ch)
+                
+    asyncio.run(test_generation())
+
+if __name__ == "__main__":
+    import time
+    main()

app/generators/vulnerability_hunter.py

@@ -820,6 +820,14 @@ ABSOLUTE RULES — VIOLATIONS WILL BE REJECTED:
    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.
+
 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}}]}}
 """
@@ -1199,6 +1207,18 @@ async def _insert_to_db(row: dict) -> bool:
     # values like 'web-security', 'secure-coding' or 'cryptography'.
     from app.core.config import normalize_row_module
     row = normalize_row_module(TABLE_NAME, row)
+    # --- Deduplication: skip if a similar challenge already exists ---
+    try:
+        from app.services.dedup import is_duplicate_vuln_hunter
+        if is_duplicate_vuln_hunter(
+            row.get("title", ""),
+            row.get("vulnerable_code", ""),
+            role_filter=row.get("team_role"),
+        ):
+            print(f"  [dedup] Skipped duplicate: {row.get('title', '?')[:60]}")
+            return False
+    except Exception as _dedup_err:
+        print(f"  [dedup] Check failed (non-fatal): {_dedup_err}")
     url = f"{SUPABASE_URL}/rest/v1/{TABLE_NAME}"
     headers = supabase_headers(content_type=True)
     headers["Prefer"] = "return=representation"

app/sandbox/terminal.py

@@ -60,7 +60,8 @@ async def run_terminal_command(team_role: str, challenge_id: str, command: str)
 
     # Built-ins
     if tool in {"cat", "ls", "pwd", "echo", "whoami", "clear", "help",
-                "sha256sum", "md5sum", "sha1sum", "base64", "xxd", "tr"}:
+                "sha256sum", "md5sum", "sha1sum", "base64", "xxd", "tr",
+                "file", "strings", "unzip", "unzip-file"}:
         result = shell_builtins(parts, workdir)
         if result is not None:
             return result

app/sandbox/tools.py

@@ -110,6 +110,81 @@ def shell_builtins(args: list[str], workdir: str) -> Optional[dict]:
                     src = src.replace(a, b)
                 return {"stdout": src, "stderr": "", "exitCode": 0}
             return {"stdout": "", "stderr": "tr: no input", "exitCode": 1}
+
+        if tool == "file":
+            if not rest:
+                return {"stdout": "", "stderr": "file: missing filename", "exitCode": 1}
+            fname = rest[0]
+            p = os.path.join(workdir, fname.lstrip("/\\").replace("..", "_"))
+            if not os.path.isfile(p):
+                return {"stdout": "", "stderr": f"file: {fname}: No such file", "exitCode": 1}
+            with open(p, "rb") as f:
+                header = f.read(8)
+            if header.startswith(b"\xff\xd8\xff"):
+                return {"stdout": f"{fname}: JPEG image data, JFIF standard 1.01", "stderr": "", "exitCode": 0}
+            if header.startswith(b"\x89PNG\r\n\x1a\n"):
+                return {"stdout": f"{fname}: PNG image data, 8-bit/color RGBA, non-interlaced", "stderr": "", "exitCode": 0}
+            if header.startswith(b"PK\x03\x04"):
+                return {"stdout": f"{fname}: Zip archive data, at least v2.0 to extract", "stderr": "", "exitCode": 0}
+            
+            with open(p, "rb") as f:
+                content = f.read()
+            if b"PK\x03\x04" in content:
+                if content.startswith(b"\xff\xd8\xff"):
+                    return {"stdout": f"{fname}: JPEG image data (with appended Zip archive data)", "stderr": "", "exitCode": 0}
+                return {"stdout": f"{fname}: Zip archive data (appended)", "stderr": "", "exitCode": 0}
+            
+            try:
+                content.decode("utf-8")
+                return {"stdout": f"{fname}: ASCII text", "stderr": "", "exitCode": 0}
+            except UnicodeDecodeError:
+                return {"stdout": f"{fname}: data", "stderr": "", "exitCode": 0}
+
+        if tool == "strings":
+            if not rest:
+                return {"stdout": "", "stderr": "strings: missing filename", "exitCode": 1}
+            fname = rest[0]
+            p = os.path.join(workdir, fname.lstrip("/\\").replace("..", "_"))
+            if not os.path.isfile(p):
+                return {"stdout": "", "stderr": f"strings: {fname}: No such file", "exitCode": 1}
+            with open(p, "rb") as f:
+                data = f.read()
+            out = []
+            curr = []
+            for b in data:
+                if 32 <= b <= 126 or b == 10 or b == 13:
+                    curr.append(chr(b))
+                else:
+                    if len(curr) >= 4:
+                        out.append("".join(curr).strip())
+                    curr = []
+            if len(curr) >= 4:
+                out.append("".join(curr).strip())
+            out = [line for line in out if line]
+            return {"stdout": "\n".join(out[:1000]), "stderr": "", "exitCode": 0}
+
+        if tool in ("unzip", "unzip-file"):
+            if not rest:
+                return {"stdout": "", "stderr": "unzip: missing filename", "exitCode": 1}
+            fname = rest[0]
+            args_clean = [a for a in rest if not a.startswith("-")]
+            if not args_clean:
+                 return {"stdout": "", "stderr": "unzip: missing filename", "exitCode": 1}
+            fname = args_clean[0]
+            p = os.path.join(workdir, fname.lstrip("/\\").replace("..", "_"))
+            if not os.path.isfile(p):
+                return {"stdout": "", "stderr": f"unzip: {fname}: No such file", "exitCode": 1}
+            import zipfile
+            try:
+                out_lines = [f"Archive: {fname}"]
+                with zipfile.ZipFile(p, 'r') as zip_ref:
+                    zip_ref.extractall(workdir)
+                    for info in zip_ref.infolist():
+                        out_lines.append(f"  extracting: {info.filename}")
+                return {"stdout": "\n".join(out_lines), "stderr": "", "exitCode": 0}
+            except Exception as e:
+                return {"stdout": "", "stderr": f"unzip: error: {e}", "exitCode": 1}
+
     except Exception as e:
         return {"stdout": "", "stderr": f"{tool}: {e}", "exitCode": 1}
     return None  # not a builtin

app/services/challenge_loader.py

@@ -141,3 +141,45 @@ def map_vuln_hunter_row_to_training(row: dict, team_role: str) -> dict:
         "xpReward": row.get("xp_reward", 150),
         "hints": row.get("hints") or [],
     }
+
+
+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": row.get("hints") or [],
+        "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",
+    }
+

app/services/dedup.py

@@ -0,0 +1,211 @@
+"""
+Challenge deduplication utilities.
+
+Before inserting a new challenge into any of the four tables,
+we check whether a sufficiently similar challenge already exists
+to prevent the pool from filling with near-duplicates.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from difflib import SequenceMatcher
+from typing import Dict, List, Optional, Set
+
+import httpx
+
+SUPABASE_URL = os.getenv("SUPABASE_URL", "")
+SUPABASE_ANON_KEY = os.getenv("SUPABASE_ANON_KEY", "")
+
+# ---------------------------------------------------------------------------
+# Text normalisation helpers
+# ---------------------------------------------------------------------------
+
+_AR_STOP_WORDS: Set[str] = {
+    "في", "من", "على", "إلى", "عن", "مع", "هذا", "هذه", "التي", "الذي",
+    "أن", "إن", "لا", "ما", "لا", "كل", "بها", "به", "فيها", "فيه",
+    "أو", "ثم", "لكن", "بل", "حتى", "قد", "لقد", "كان", "يجب", "يتم",
+    "يمكن", "حيث", "كما", "أيضاً", "ذلك", "تلك", "اللذين", "اللتين",
+}
+
+
+def _normalise(text: str) -> str:
+    """Lower-case, strip punctuation, collapse whitespace."""
+    text = text.lower()
+    text = re.sub(r"[^\w\s]", " ", text)
+    text = re.sub(r"\s+", " ", text).strip()
+    return text
+
+
+def _token_set(text: str) -> Set[str]:
+    """Return de-duplicated token set, Arabic stop-words removed."""
+    tokens = _normalise(text).split()
+    return {t for t in tokens if t not in _AR_STOP_WORDS and len(t) > 1}
+
+
+def _title_similarity(a: str, b: str) -> float:
+    """
+    Title similarity = max of:
+      - token overlap (Jaccard-ish)
+      - SequenceMatcher ratio on normalised strings
+    """
+    sa, sb = _normalise(a), _normalise(b)
+    if not sa or not sb:
+        return 0.0
+    # SequenceMatcher on normalised strings
+    seq = SequenceMatcher(None, sa, sb).ratio()
+    # Token overlap
+    ta, tb = _token_set(a), _token_set(b)
+    if not ta or not tb:
+        return seq
+    jaccard = len(ta & tb) / max(len(ta | tb), 1)
+    return max(seq, jaccard)
+
+
+def _code_similarity(a: str, b: str) -> float:
+    """
+    For code: use SequenceMatcher on stripped lines.
+    Code tends to have boilerplate; line-level matching works well.
+    """
+    def _strip(lines: str) -> List[str]:
+        return [ln.strip() for ln in lines.splitlines() if ln.strip()]
+    la, lb = _strip(a), _strip(b)
+    if not la or not lb:
+        return 0.0
+    return SequenceMatcher(None, la, lb).ratio()
+
+
+# ---------------------------------------------------------------------------
+# DB fetch helpers
+# ---------------------------------------------------------------------------
+
+_HEADERS = {
+    "apikey": SUPABASE_ANON_KEY,
+    "Authorization": f"Bearer {SUPABASE_ANON_KEY}",
+    "Content-Type": "application/json",
+}
+
+
+def _fetch_existing_titles(table: str, role_filter: Optional[str] = None,
+                           limit: int = 200) -> List[str]:
+    """Fetch recent titles from a challenge table."""
+    url = f"{SUPABASE_URL}/rest/v1/{table}"
+    params: Dict[str, str] = {
+        "select": "title",
+        "order": "created_at.desc",
+        "limit": str(limit),
+    }
+    if role_filter:
+        params["team_role"] = f"eq.{role_filter}"
+    try:
+        r = httpx.get(url, headers=_HEADERS, params=params, timeout=10)
+        r.raise_for_status()
+        return [row.get("title", "") for row in r.json()]
+    except Exception:
+        return []
+
+
+def _fetch_existing_codes(table: str, code_col: str,
+                          role_filter: Optional[str] = None,
+                          limit: int = 100) -> List[str]:
+    """Fetch recent code/story/log_content from a challenge table."""
+    url = f"{SUPABASE_URL}/rest/v1/{table}"
+    params: Dict[str, str] = {
+        "select": f"title,{code_col}",
+        "order": "created_at.desc",
+        "limit": str(limit),
+    }
+    if role_filter:
+        params["team_role"] = f"eq.{role_filter}"
+    try:
+        r = httpx.get(url, headers=_HEADERS, params=params, timeout=10)
+        r.raise_for_status()
+        return [row.get(code_col, "") for row in r.json() if row.get(code_col)]
+    except Exception:
+        return []
+
+
+# ---------------------------------------------------------------------------
+# Public API — one function per generator
+# ---------------------------------------------------------------------------
+
+def is_duplicate_encryption(title: str, story: str,
+                            role_filter: Optional[str] = None) -> bool:
+    """Check if an encryption challenge is too similar to existing ones."""
+    existing_titles = _fetch_existing_titles("encryption_challenges", role_filter)
+    for t in existing_titles:
+        if _title_similarity(title, t) > 0.55:
+            return True
+
+    existing_stories = _fetch_existing_codes(
+        "encryption_challenges", "story", role_filter)
+    for s in existing_stories:
+        if _title_similarity(story, s) > 0.55:
+            return True
+    return False
+
+
+def is_duplicate_code_fixing(title: str, vulnerable_code: str,
+                             role_filter: Optional[str] = None) -> bool:
+    """Check if a code-fixing challenge is too similar to existing ones."""
+    existing_titles = _fetch_existing_titles("code_fixing_challenges", role_filter)
+    for t in existing_titles:
+        if _title_similarity(title, t) > 0.55:
+            return True
+
+    existing_codes = _fetch_existing_codes(
+        "code_fixing_challenges", "vulnerable_code", role_filter)
+    for c in existing_codes:
+        if _code_similarity(vulnerable_code, c) > 0.60:
+            return True
+    return False
+
+
+def is_duplicate_log_analysis(title: str, log_content: str,
+                              role_filter: Optional[str] = None) -> bool:
+    """Check if a log-analysis challenge is too similar to existing ones."""
+    existing_titles = _fetch_existing_titles("log_analysis_challenges", role_filter)
+    for t in existing_titles:
+        if _title_similarity(title, t) > 0.55:
+            return True
+
+    existing_logs = _fetch_existing_codes(
+        "log_analysis_challenges", "log_content", role_filter)
+    for lc in existing_logs:
+        if _code_similarity(log_content, lc) > 0.55:
+            return True
+    return False
+
+
+def is_duplicate_vuln_hunter(title: str, vulnerable_code: str,
+                             role_filter: Optional[str] = None) -> bool:
+    """Check if a vulnerability-hunter challenge is too similar to existing ones."""
+    existing_titles = _fetch_existing_titles(
+        "vulnerability_hunter_challenges", role_filter)
+    for t in existing_titles:
+        if _title_similarity(title, t) > 0.55:
+            return True
+
+    existing_codes = _fetch_existing_codes(
+        "vulnerability_hunter_challenges", "vulnerable_code", role_filter)
+    for c in existing_codes:
+        if _code_similarity(vulnerable_code, c) > 0.60:
+            return True
+    return False
+
+
+def is_duplicate_steganography(title: str, story: str,
+                              role_filter: Optional[str] = None) -> bool:
+    """Check if a steganography challenge is too similar to existing ones."""
+    existing_titles = _fetch_existing_titles("steganography_challenges", role_filter)
+    for t in existing_titles:
+        if _title_similarity(title, t) > 0.55:
+            return True
+
+    existing_stories = _fetch_existing_codes(
+        "steganography_challenges", "story", role_filter)
+    for s in existing_stories:
+        if _title_similarity(story, s) > 0.55:
+            return True
+    return False

app/services/scenario_service.py

@@ -363,6 +363,8 @@ def map_scenario_to_list_item(scenario: dict) -> dict:
         module_out = "web"
     elif challenge_type == "crypto":
         module_out = "crypto"
+    elif challenge_type == "steganography":
+        module_out = "steganography"
     else:
         module_out = mod
 
@@ -377,6 +379,8 @@ def map_scenario_to_list_item(scenario: dict) -> dict:
         category = "Web Security"
     elif challenge_type == "crypto":
         category = "Cryptography"
+    elif challenge_type == "steganography":
+        category = "Steganography"
 
     return {
         "id": scenario["id"],

app/services/supabase_service.py

@@ -19,13 +19,15 @@ from app.core.config import SUPABASE_URL, SUPABASE_ANON_KEY
 def scenario_table(team_role: str = "", challenge_type: str = "") -> str:
     """Pick the right per-type table for a challenge lookup.
 
-    ``challenge_type`` can be one of: ``"crypto"``,
+    ``challenge_type`` can be one of: ``"crypto"``, ``"steganography"``,
     ``"code-fixing"``, ``"log-analysis"``, ``"vulnerability-hunter"``.
     If omitted we fall back to ``"encryption_challenges"`` for legacy
     behavior (Blue/Red crypto).
     """
     if challenge_type == "crypto":
         return "encryption_challenges"
+    if challenge_type == "steganography":
+        return "steganography_challenges"
     if challenge_type == "code-fixing":
         return "code_fixing_challenges"
     if challenge_type == "log-analysis":
@@ -74,17 +76,47 @@ async def fetch_scenario_by_id(
 ) -> Optional[dict]:
     if not SUPABASE_ANON_KEY or not SUPABASE_URL:
         return None
-    table = scenario_table(team_role, challenge_type)
-    url = f"{SUPABASE_URL}/rest/v1/{table}?id=eq.{scenario_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:
-                rows = resp.json()
-                if rows:
-                    return rows[0]
-    except Exception as e:
-        print(f"Error fetching scenario {scenario_id}: {e}")
+
+    async def _fetch(tbl: str) -> Optional[dict]:
+        url = f"{SUPABASE_URL}/rest/v1/{tbl}?id=eq.{scenario_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:
+                    rows = resp.json()
+                    if rows:
+                        return rows[0]
+        except Exception as e:
+            print(f"Error fetching scenario {scenario_id} from {tbl}: {e}")
+        return None
+
+    if challenge_type:
+        tbl = scenario_table(team_role, challenge_type)
+        return await _fetch(tbl)
+
+    # Search all tables for this team role
+    tables = []
+    if team_role == "red":
+        tables = ["encryption_challenges", "steganography_challenges"]
+    elif team_role == "blue":
+        tables = [
+            "code_fixing_challenges",
+            "log_analysis_challenges",
+            "vulnerability_hunter_challenges",
+        ]
+    else:
+        tables = [
+            "encryption_challenges",
+            "steganography_challenges",
+            "code_fixing_challenges",
+            "log_analysis_challenges",
+            "vulnerability_hunter_challenges",
+        ]
+
+    for tbl in tables:
+        row = await _fetch(tbl)
+        if row:
+            return row
     return None