Delete server

server/Dockerfile

@@ -1,49 +0,0 @@
-ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
-FROM ${BASE_IMAGE} AS builder
-
-WORKDIR /app
-
-COPY envs/code_security_auditor_env /app/env
-WORKDIR /app/env
-
-RUN if ! command -v uv >/dev/null 2>&1; then \
-        curl -LsSf https://astral.sh/uv/install.sh | sh && \
-        mv /root/.local/bin/uv /usr/local/bin/uv && \
-        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
-    fi
-
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    git \
-    curl \
-    ca-certificates \
-    && rm -rf /var/lib/apt/lists/*
-
-RUN --mount=type=cache,target=/root/.cache/uv \
-    if [ -f uv.lock ]; then \
-        uv sync --frozen --no-install-project --no-editable; \
-    else \
-        uv sync --no-install-project --no-editable; \
-    fi
-
-RUN --mount=type=cache,target=/root/.cache/uv \
-    if [ -f uv.lock ]; then \
-        uv sync --frozen --no-editable; \
-    else \
-        uv sync --no-editable; \
-    fi
-
-FROM ${BASE_IMAGE}
-
-WORKDIR /app
-COPY --from=builder /app/env/.venv /app/.venv
-COPY --from=builder /app/env /app/env
-
-ENV PATH="/app/.venv/bin:$PATH"
-ENV PYTHONPATH="/app/env:$PYTHONPATH"
-ENV PYTHONUNBUFFERED=1
-ENV ENABLE_WEB_INTERFACE=true
-
-HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
-    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1
-
-CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]

server/__init__.py

@@ -1 +0,0 @@
-"""Server package for Code Security Auditor environment."""

server/app.py

@@ -1,33 +0,0 @@
-from __future__ import annotations
-
-try:
-    from core.env_server.http_server import create_app
-except ImportError:
-    try:
-        from openenv.core.env_server.http_server import create_app
-    except ImportError:
-        from openenv_core.env_server.http_server import create_app
-
-try:
-    from ..models import CodeSecurityAction, CodeSecurityObservation
-    from .security_environment import CodeSecurityAuditorEnvironment
-except ImportError:
-    from models import CodeSecurityAction, CodeSecurityObservation
-    from server.security_environment import CodeSecurityAuditorEnvironment
-
-app = create_app(
-    CodeSecurityAuditorEnvironment,
-    CodeSecurityAction,
-    CodeSecurityObservation,
-    env_name="code_security_auditor_env",
-)
-
-
-def main() -> None:
-    import uvicorn
-
-    uvicorn.run(app, host="0.0.0.0", port=8000)
-
-
-if __name__ == "__main__":
-    main()

server/grader.py

@@ -1,181 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Iterable, Optional
-
-from .tasks import SEVERITY_WEIGHTS, TARGET_CONFIDENCE, TaskSpec, VulnerabilitySpec
-
-
-@dataclass(frozen=True)
-class FindingEvaluation:
-    component_score: float
-    matched_vulnerability_id: Optional[str]
-    is_confirmed_match: bool
-    feedback: str
-    confidence_calibration: float
-
-
-def _line_overlap_score(submitted_start: int, submitted_end: int, target_line: int) -> float:
-    if submitted_start <= target_line <= submitted_end:
-        return 1.0
-    min_distance = min(abs(target_line - submitted_start), abs(target_line - submitted_end))
-    if min_distance <= 2:
-        return 0.6
-    if min_distance <= 5:
-        return 0.3
-    return 0.0
-
-
-def _best_candidate(
-    task: TaskSpec,
-    filename: str,
-    vuln_type: str,
-    severity: str,
-    line_start: int,
-    line_end: int,
-) -> tuple[Optional[VulnerabilitySpec], float, float, float, float]:
-    best_target = None
-    best_score = -1.0
-    best_type_match = 0.0
-    best_line_match = 0.0
-    best_severity_match = 0.0
-
-    for target in task.vulnerabilities:
-        file_match = 1.0 if target.filename == filename else 0.0
-        type_match = 1.0 if target.vuln_type == vuln_type else 0.0
-        severity_match = 1.0 if target.severity == severity else 0.0
-        line_match = _line_overlap_score(line_start, line_end, target.line)
-
-        candidate_score = (
-            0.35 * file_match
-            + 0.30 * type_match
-            + 0.20 * line_match
-            + 0.15 * severity_match
-        )
-
-        if candidate_score > best_score:
-            best_score = candidate_score
-            best_target = target
-            best_type_match = type_match
-            best_line_match = line_match
-            best_severity_match = severity_match
-
-    return best_target, max(best_score, 0.0), best_type_match, best_line_match, best_severity_match
-
-
-def evaluate_finding(
-    *,
-    task: TaskSpec,
-    filename: str,
-    vuln_type: str,
-    severity: str,
-    line_start: int,
-    line_end: int,
-    confidence: float,
-    matched_already: Iterable[str],
-) -> FindingEvaluation:
-    target, structure_score, type_match, line_match, severity_match = _best_candidate(
-        task,
-        filename,
-        vuln_type,
-        severity,
-        line_start,
-        line_end,
-    )
-
-    if target is None:
-        return FindingEvaluation(
-            component_score=0.0,
-            matched_vulnerability_id=None,
-            is_confirmed_match=False,
-            feedback="No plausible vulnerability match for this finding.",
-            confidence_calibration=0.0,
-        )
-
-    target_conf = TARGET_CONFIDENCE[target.severity]
-    calibration = max(0.0, 1.0 - abs(confidence - target_conf))
-
-    component_score = 0.8 * structure_score + 0.2 * calibration
-    component_score = max(0.0, min(1.0, component_score))
-
-    confirmed = (
-        target.filename == filename
-        and type_match == 1.0
-        and line_match >= 0.6
-        and severity_match == 1.0
-    )
-
-    if target.id in set(matched_already) and confirmed:
-        return FindingEvaluation(
-            component_score=0.25 * component_score,
-            matched_vulnerability_id=target.id,
-            is_confirmed_match=False,
-            feedback="Duplicate of a previously confirmed vulnerability.",
-            confidence_calibration=calibration,
-        )
-
-    if confirmed:
-        return FindingEvaluation(
-            component_score=component_score,
-            matched_vulnerability_id=target.id,
-            is_confirmed_match=True,
-            feedback="Confirmed vulnerability: file/type/line/severity align with ground truth.",
-            confidence_calibration=calibration,
-        )
-
-    if target.filename != filename:
-        hint = "Wrong file."
-    elif type_match == 0.0:
-        hint = "Correct file, vulnerability type mismatch."
-    elif line_match < 0.6:
-        hint = "Correct file/type, but location is off."
-    elif severity_match == 0.0:
-        hint = "Severity mismatch."
-    else:
-        hint = "Partial match, refine details."
-
-    return FindingEvaluation(
-        component_score=component_score,
-        matched_vulnerability_id=None,
-        is_confirmed_match=False,
-        feedback=hint,
-        confidence_calibration=calibration,
-    )
-
-
-def final_grade(
-    *,
-    task: TaskSpec,
-    confirmed_vulnerability_ids: Iterable[str],
-    findings_count: int,
-    false_positive_count: int,
-    duplicate_count: int,
-    avg_component_score: float,
-    avg_confidence_calibration: float,
-) -> float:
-    confirmed_ids = set(confirmed_vulnerability_ids)
-
-    total_weight = sum(SEVERITY_WEIGHTS[v.severity] for v in task.vulnerabilities)
-    covered_weight = sum(
-        SEVERITY_WEIGHTS[v.severity] for v in task.vulnerabilities if v.id in confirmed_ids
-    )
-    weighted_recall = (covered_weight / total_weight) if total_weight > 0 else 0.0
-
-    precision = (len(confirmed_ids) / findings_count) if findings_count > 0 else 0.0
-
-    fp_penalty = min(0.5, 0.08 * false_positive_count)
-    dup_penalty = min(0.2, 0.05 * duplicate_count)
-    volume_penalty = 0.0
-    optimal_findings = len(task.vulnerabilities) + 1
-    if findings_count > optimal_findings:
-        volume_penalty = min(0.2, 0.03 * (findings_count - optimal_findings))
-
-    score = (
-        0.55 * weighted_recall
-        + 0.20 * precision
-        + 0.15 * avg_component_score
-        + 0.10 * avg_confidence_calibration
-    )
-    score -= fp_penalty + dup_penalty + volume_penalty
-
-    return max(0.0, min(1.0, score))

server/security_environment.py

@@ -1,386 +0,0 @@
-from __future__ import annotations
-
-import random
-import uuid
-from typing import Any, Optional
-
-try:
-    from core.env_server.interfaces import Environment
-except ImportError:
-    try:
-        from openenv.core.env_server.interfaces import Environment
-    except ImportError:
-        from openenv_core.env_server.interfaces import Environment
-
-try:
-    from ..models import (
-        CodeSecurityAction,
-        CodeSecurityObservation,
-        CodeSecurityState,
-        FindingRecord,
-    )
-    from .grader import evaluate_finding, final_grade
-    from .tasks import TaskSpec, get_task, list_task_ids
-except ImportError:
-    from models import (
-        CodeSecurityAction,
-        CodeSecurityObservation,
-        CodeSecurityState,
-        FindingRecord,
-    )
-    from server.grader import evaluate_finding, final_grade
-    from server.tasks import TaskSpec, get_task, list_task_ids
-
-
-class CodeSecurityAuditorEnvironment(
-    Environment[CodeSecurityAction, CodeSecurityObservation, CodeSecurityState]
-):
-    """Real-world code security auditing simulator with deterministic graders."""
-
-    SUPPORTS_CONCURRENT_SESSIONS = True
-
-    def __init__(self, default_task_id: str = "easy"):
-        self._default_task_id = default_task_id
-        self._task_cursor = 0
-        self._task: Optional[TaskSpec] = None
-        self._state = CodeSecurityState()
-
-    def reset(
-        self,
-        seed: Optional[int] = None,
-        episode_id: Optional[str] = None,
-        **kwargs: Any,
-    ) -> CodeSecurityObservation:
-        requested_task = kwargs.get("task_id") or kwargs.get("task")
-
-        if requested_task is not None:
-            task = get_task(str(requested_task))
-        elif seed is not None:
-            rng = random.Random(seed)
-            task = get_task(rng.choice(list_task_ids()))
-        elif self._default_task_id:
-            task = get_task(self._default_task_id)
-        else:
-            task_order = list_task_ids()
-            task = get_task(task_order[self._task_cursor % len(task_order)])
-            self._task_cursor += 1
-
-        self._task = task
-        self._state = CodeSecurityState(
-            episode_id=episode_id or str(uuid.uuid4()),
-            step_count=0,
-            task_id=task.id,
-            task_title=task.title,
-            difficulty=task.difficulty,
-            objective=task.objective,
-            max_steps=task.max_steps,
-            inspected_files=[],
-            findings_submitted=[],
-            matched_vulnerability_ids=[],
-            false_positive_count=0,
-            duplicate_submission_count=0,
-            quality_multiplier=1.0,
-            final_score=None,
-        )
-
-        return self._build_observation(
-            reward=0.0,
-            done=False,
-            feedback=(
-                "Audit started. Use inspect_file before submit_finding. "
-                "Finish with submit_final_report."
-            ),
-            focused_file=None,
-            excerpt="",
-            extra_metadata={
-                "available_task_ids": list_task_ids(),
-                "task_id": task.id,
-            },
-        )
-
-    def step(
-        self,
-        action: CodeSecurityAction,
-        timeout_s: Optional[float] = None,
-        **kwargs: Any,
-    ) -> CodeSecurityObservation:
-        del timeout_s, kwargs
-
-        task = self._require_task()
-
-        if self._state.final_score is not None:
-            return self._build_observation(
-                reward=0.0,
-                done=True,
-                feedback="Episode already terminated. Call reset() to start a new task.",
-                focused_file=None,
-                excerpt="",
-            )
-
-        self._state.step_count += 1
-        feedback = ""
-        reward = 0.0
-        focused_file = None
-        excerpt = ""
-
-        if action.action_type == "inspect_file":
-            reward, feedback, focused_file, excerpt = self._handle_inspect_file(action, task)
-        elif action.action_type == "submit_finding":
-            reward, feedback = self._handle_submit_finding(action, task)
-        elif action.action_type == "submit_final_report":
-            reward, feedback = self._handle_submit_final_report()
-        else:
-            feedback = f"Unsupported action_type={action.action_type}."
-            self._degrade_quality(0.03)
-
-        done = self._state.final_score is not None
-
-        if not done and self._state.step_count >= self._state.max_steps:
-            score = self._compute_final_score(task)
-            self._state.final_score = score
-            done = True
-            reward = score
-            feedback = (
-                f"Max steps reached. Auto-finalized audit score={score:.3f}. "
-                "Use fewer but higher-quality findings to improve precision."
-            )
-
-        return self._build_observation(
-            reward=reward,
-            done=done,
-            feedback=feedback,
-            focused_file=focused_file,
-            excerpt=excerpt,
-            extra_metadata={
-                "last_action_error": None,
-            },
-        )
-
-    @property
-    def state(self) -> CodeSecurityState:
-        return self._state
-
-    def _require_task(self) -> TaskSpec:
-        if self._task is None:
-            raise RuntimeError("Environment has no active task. Call reset() first.")
-        return self._task
-
-    def _degrade_quality(self, amount: float) -> None:
-        self._state.quality_multiplier = max(0.2, self._state.quality_multiplier - amount)
-
-    def _format_file(self, content: str) -> str:
-        lines = content.splitlines()
-        numbered = [f"{idx + 1:>3}: {line}" for idx, line in enumerate(lines)]
-        return "\n".join(numbered)
-
-    def _handle_inspect_file(
-        self,
-        action: CodeSecurityAction,
-        task: TaskSpec,
-    ) -> tuple[float, str, Optional[str], str]:
-        filename = action.filename or ""
-        if filename not in task.repository:
-            self._degrade_quality(0.04)
-            return 0.0, f"Unknown file '{filename}'.", filename or None, ""
-
-        first_time = filename not in self._state.inspected_files
-        if first_time:
-            self._state.inspected_files.append(filename)
-
-        excerpt = self._format_file(task.repository[filename])
-
-        unmatched_in_file = any(
-            vuln.filename == filename and vuln.id not in self._state.matched_vulnerability_ids
-            for vuln in task.vulnerabilities
-        )
-
-        if first_time and unmatched_in_file:
-            reward = 0.04
-            feedback = "Useful inspection: this file likely contains unresolved security issues."
-        elif first_time:
-            reward = 0.02
-            feedback = "Inspection noted. No strong security signal yet."
-        else:
-            reward = 0.0
-            feedback = "File already inspected; repeated reads do not improve score."
-            self._degrade_quality(0.01)
-
-        return reward, feedback, filename, excerpt
-
-    def _handle_submit_finding(
-        self,
-        action: CodeSecurityAction,
-        task: TaskSpec,
-    ) -> tuple[float, str]:
-        required_missing = []
-        if not action.filename:
-            required_missing.append("filename")
-        if action.line_start is None:
-            required_missing.append("line_start")
-        if not action.vuln_type:
-            required_missing.append("vuln_type")
-        if not action.severity:
-            required_missing.append("severity")
-
-        if required_missing:
-            self._degrade_quality(0.05)
-            missing = ", ".join(required_missing)
-            return 0.0, f"Incomplete finding. Missing fields: {missing}."
-
-        line_end = action.line_end if action.line_end is not None else action.line_start
-
-        evaluation = evaluate_finding(
-            task=task,
-            filename=action.filename,
-            vuln_type=action.vuln_type,
-            severity=action.severity,
-            line_start=action.line_start,
-            line_end=line_end,
-            confidence=action.confidence,
-            matched_already=self._state.matched_vulnerability_ids,
-        )
-
-        finding_id = f"finding-{len(self._state.findings_submitted) + 1}"
-        finding_record = FindingRecord(
-            finding_id=finding_id,
-            filename=action.filename,
-            line_start=action.line_start,
-            line_end=line_end,
-            vuln_type=action.vuln_type,
-            severity=action.severity,
-            confidence=action.confidence,
-            evidence=(action.evidence or "").strip(),
-            summary=(action.summary or "").strip(),
-            matched_vulnerability_id=evaluation.matched_vulnerability_id,
-            component_score=evaluation.component_score,
-        )
-        self._state.findings_submitted.append(finding_record)
-
-        if evaluation.is_confirmed_match and evaluation.matched_vulnerability_id is not None:
-            self._state.matched_vulnerability_ids.append(evaluation.matched_vulnerability_id)
-            reward = min(1.0, (0.25 + 0.75 * evaluation.component_score) * self._state.quality_multiplier)
-            feedback = (
-                f"{evaluation.feedback} "
-                f"Confirmed={len(self._state.matched_vulnerability_ids)}/{len(task.vulnerabilities)}."
-            )
-            return reward, feedback
-
-        if (
-            evaluation.matched_vulnerability_id is not None
-            and evaluation.matched_vulnerability_id in self._state.matched_vulnerability_ids
-        ):
-            self._state.duplicate_submission_count += 1
-            self._degrade_quality(0.04)
-            return 0.01, evaluation.feedback
-
-        if evaluation.component_score >= 0.45:
-            self._degrade_quality(0.01)
-            reward = min(0.2, 0.2 * evaluation.component_score * self._state.quality_multiplier)
-            return reward, f"Partial progress: {evaluation.feedback}"
-
-        self._state.false_positive_count += 1
-        self._degrade_quality(0.05)
-        return 0.0, f"Likely false positive: {evaluation.feedback}"
-
-    def _handle_submit_final_report(self) -> tuple[float, str]:
-        task = self._require_task()
-        score = self._compute_final_score(task)
-        self._state.final_score = score
-        feedback = (
-            f"Audit finalized. Final deterministic score={score:.3f}. "
-            f"Confirmed {len(self._state.matched_vulnerability_ids)} of {len(task.vulnerabilities)} vulnerabilities."
-        )
-        return score, feedback
-
-    def _compute_final_score(self, task: TaskSpec) -> float:
-        if self._state.findings_submitted:
-            avg_component = sum(f.component_score for f in self._state.findings_submitted) / len(
-                self._state.findings_submitted
-            )
-        else:
-            avg_component = 0.0
-
-        if self._state.findings_submitted:
-            avg_calibration = sum(
-                max(0.0, 1.0 - abs(f.confidence - 0.75)) for f in self._state.findings_submitted
-            ) / len(self._state.findings_submitted)
-        else:
-            avg_calibration = 0.0
-
-        score = final_grade(
-            task=task,
-            confirmed_vulnerability_ids=self._state.matched_vulnerability_ids,
-            findings_count=len(self._state.findings_submitted),
-            false_positive_count=self._state.false_positive_count,
-            duplicate_count=self._state.duplicate_submission_count,
-            avg_component_score=avg_component,
-            avg_confidence_calibration=avg_calibration,
-        )
-
-        # This quality factor makes spam and random guesses strictly dominated,
-        # limiting reward hacking while preserving partial-credit gradients.
-        score *= self._state.quality_multiplier
-        return max(0.0, min(1.0, score))
-
-    def _build_observation(
-        self,
-        *,
-        reward: float,
-        done: bool,
-        feedback: str,
-        focused_file: Optional[str],
-        excerpt: str,
-        extra_metadata: Optional[dict[str, Any]] = None,
-    ) -> CodeSecurityObservation:
-        task = self._require_task()
-
-        findings_public = [
-            {
-                "finding_id": f.finding_id,
-                "filename": f.filename,
-                "line_start": f.line_start,
-                "line_end": f.line_end,
-                "vuln_type": f.vuln_type,
-                "severity": f.severity,
-                "confidence": f.confidence,
-                "component_score": round(f.component_score, 3),
-            }
-            for f in self._state.findings_submitted
-        ]
-
-        score_hint = len(self._state.matched_vulnerability_ids) / max(1, len(task.vulnerabilities))
-
-        metadata = {
-            "quality_multiplier": round(self._state.quality_multiplier, 4),
-            "false_positive_count": self._state.false_positive_count,
-            "duplicate_submission_count": self._state.duplicate_submission_count,
-            "confirmed_vulnerabilities": len(self._state.matched_vulnerability_ids),
-            "total_vulnerabilities": len(task.vulnerabilities),
-            "task_id": task.id,
-            "difficulty": task.difficulty,
-            "available_task_ids": list_task_ids(),
-            "last_action_error": None,
-        }
-        if extra_metadata:
-            metadata.update(extra_metadata)
-
-        return CodeSecurityObservation(
-            done=done,
-            reward=max(0.0, min(1.0, reward)),
-            metadata=metadata,
-            task_id=task.id,
-            task_title=task.title,
-            difficulty=task.difficulty,
-            objective=task.objective,
-            instructions=(
-                "Valid actions: inspect_file, submit_finding, submit_final_report. "
-                "For submit_finding include filename, line_start/line_end, vuln_type, severity, confidence."
-            ),
-            available_files=sorted(task.repository.keys()),
-            focused_file=focused_file,
-            file_excerpt=excerpt,
-            findings_so_far=findings_public,
-            steps_remaining=max(0, self._state.max_steps - self._state.step_count),
-            last_feedback=feedback,
-            score_hint=max(0.0, min(1.0, score_hint)),
-        )

server/tasks.py

@@ -1,208 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Dict, List, Literal
-
-Difficulty = Literal["easy", "medium", "hard"]
-Severity = Literal["low", "medium", "high", "critical"]
-
-
-@dataclass(frozen=True)
-class VulnerabilitySpec:
-    id: str
-    filename: str
-    line: int
-    vuln_type: str
-    severity: Severity
-    title: str
-    rationale: str
-
-
-@dataclass(frozen=True)
-class TaskSpec:
-    id: str
-    title: str
-    difficulty: Difficulty
-    objective: str
-    max_steps: int
-    repository: Dict[str, str]
-    vulnerabilities: List[VulnerabilitySpec]
-
-
-TASKS: Dict[str, TaskSpec] = {
-    "easy": TaskSpec(
-        id="easy",
-        title="Legacy Flask Patch Review",
-        difficulty="easy",
-        objective=(
-            "Audit a small Flask service before release. Identify concrete "
-            "security vulnerabilities with accurate location, type, and severity."
-        ),
-        max_steps=12,
-        repository={
-            "app/routes.py": """from flask import Blueprint, request, jsonify\nfrom app.db import db\n\nbp = Blueprint('api', __name__)\n\n@bp.get('/user')\ndef get_user():\n    user_id = request.args.get('id', '')\n    query = f\"SELECT id, email, role FROM users WHERE id = '{user_id}'\"\n    row = db.execute(query).fetchone()\n    return jsonify(dict(row) if row else {})\n\n@bp.post('/login')\ndef login():\n    payload = request.json or {}\n    if payload.get('token') == 'letmein':\n        return jsonify({'ok': True})\n    return jsonify({'ok': False}), 401\n""",
-            "app/config.py": """import os\n\nclass Config:\n    DEBUG = os.getenv('DEBUG', '0') == '1'\n    SECRET_KEY = 'prod-secret-2026'\n    DB_URL = os.getenv('DB_URL', 'postgresql://localhost/app')\n""",
-            "app/db.py": """import sqlite3\n\n_conn = sqlite3.connect(':memory:', check_same_thread=False)\n_conn.execute('CREATE TABLE IF NOT EXISTS users (id TEXT, email TEXT, role TEXT)')\n\ndef execute(query: str):\n    return _conn.execute(query)\n\nclass DB:\n    execute = staticmethod(execute)\n\ndb = DB()\n""",
-        },
-        vulnerabilities=[
-            VulnerabilitySpec(
-                id="E-01",
-                filename="app/routes.py",
-                line=8,
-                vuln_type="sql_injection",
-                severity="high",
-                title="Unsanitized SQL query with user-controlled id",
-                rationale="Direct string interpolation builds SQL using request input.",
-            ),
-            VulnerabilitySpec(
-                id="E-02",
-                filename="app/config.py",
-                line=5,
-                vuln_type="hardcoded_secret",
-                severity="high",
-                title="Hardcoded production secret key",
-                rationale="Embedding secrets in code leaks credentials and breaks rotation.",
-            ),
-            VulnerabilitySpec(
-                id="E-03",
-                filename="app/routes.py",
-                line=15,
-                vuln_type="weak_authentication",
-                severity="medium",
-                title="Backdoor static token for login",
-                rationale="Fixed token bypasses proper authentication controls.",
-            ),
-        ],
-    ),
-    "medium": TaskSpec(
-        id="medium",
-        title="Payment Webhook Service",
-        difficulty="medium",
-        objective=(
-            "Review a webhook and export pipeline used by operations. Catch logic and "
-            "input-handling flaws that can be exploited in production."
-        ),
-        max_steps=14,
-        repository={
-            "service/webhook.py": """import hmac\nimport hashlib\nfrom flask import request, jsonify\n\nSHARED_SECRET = 'whsec_test_123'\n\ndef verify_signature(body: bytes, signature: str) -> bool:\n    digest = hmac.new(SHARED_SECRET.encode(), body, hashlib.sha256).hexdigest()\n    return digest == signature\n\ndef process_event(event: dict):\n    return {'processed': event.get('id')}\n\ndef webhook_handler():\n    body = request.data\n    signature = request.headers.get('X-Signature', '')\n    if not verify_signature(body, signature):\n        return jsonify({'error': 'invalid signature'}), 401\n\n    event = request.json or {}\n    if event.get('debug'):\n        # temporary bypass while partner migrates\n        return jsonify({'ok': True, 'bypass': True})\n\n    return jsonify(process_event(event))\n""",
-            "service/export.py": """import os\n\ndef build_export_cmd(job_name: str, output_dir: str) -> str:\n    return f\"tar -czf {output_dir}/{job_name}.tgz /srv/data/{job_name}\"\n\ndef run_export(job_name: str, output_dir: str):\n    cmd = build_export_cmd(job_name, output_dir)\n    os.system(cmd)\n    return {'status': 'ok', 'cmd': cmd}\n""",
-            "service/serializers.py": """import yaml\n\ndef load_template(raw: str):\n    return yaml.load(raw, Loader=yaml.Loader)\n""",
-        },
-        vulnerabilities=[
-            VulnerabilitySpec(
-                id="M-01",
-                filename="service/webhook.py",
-                line=10,
-                vuln_type="weak_authentication",
-                severity="medium",
-                title="Timing-unsafe signature comparison",
-                rationale="Using == for secrets can leak comparison timing signal.",
-            ),
-            VulnerabilitySpec(
-                id="M-02",
-                filename="service/webhook.py",
-                line=22,
-                vuln_type="weak_authentication",
-                severity="high",
-                title="Debug flag bypasses signature verification outcome",
-                rationale="Attacker-controlled debug field returns success without processing guards.",
-            ),
-            VulnerabilitySpec(
-                id="M-03",
-                filename="service/export.py",
-                line=8,
-                vuln_type="command_injection",
-                severity="critical",
-                title="Unsanitized shell command execution",
-                rationale="User-controlled job_name/output_dir flow into os.system command string.",
-            ),
-            VulnerabilitySpec(
-                id="M-04",
-                filename="service/serializers.py",
-                line=4,
-                vuln_type="insecure_deserialization",
-                severity="high",
-                title="Unsafe YAML loader",
-                rationale="yaml.Loader can construct arbitrary Python objects from attacker input.",
-            ),
-        ],
-    ),
-    "hard": TaskSpec(
-        id="hard",
-        title="Enterprise Multi-Tenant API",
-        difficulty="hard",
-        objective=(
-            "Audit an API gateway handling tenants, files, and callback fetches. "
-            "Find high-impact vulnerabilities without flooding false positives."
-        ),
-        max_steps=16,
-        repository={
-            "api/auth.py": """import base64\nimport json\nimport jwt\n\ndef issue_token(user_id: str, tenant_id: str):\n    payload = {'sub': user_id, 'tenant': tenant_id, 'role': 'member'}\n    return jwt.encode(payload, 'dev-key', algorithm='HS256')\n\ndef parse_token(token: str):\n    header_b64 = token.split('.')[0] + '=='\n    header = json.loads(base64.urlsafe_b64decode(header_b64).decode())\n    if header.get('alg') == 'none':\n        return json.loads(base64.urlsafe_b64decode(token.split('.')[1] + '==').decode())\n    return jwt.decode(token, 'dev-key', algorithms=['HS256'])\n""",
-            "api/files.py": """from flask import request, jsonify\n\nFILES = {\n    'tenant-a': {'1': 'a-private-doc'},\n    'tenant-b': {'2': 'b-private-doc'},\n}\n\ndef get_file(user):\n    file_id = request.args.get('file_id')\n    tenant = request.args.get('tenant')\n    data = FILES.get(tenant, {}).get(file_id)\n    if not data:\n        return jsonify({'error': 'not found'}), 404\n    return jsonify({'file': data, 'tenant': tenant, 'user': user['sub']})\n""",
-            "api/fetcher.py": """import requests\n\ndef fetch_preview(url: str):\n    response = requests.get(url, timeout=3)\n    return {'status': response.status_code, 'body': response.text[:120]}\n""",
-            "api/storage.py": """from pathlib import Path\n\nBASE = Path('/srv/uploads')\n\ndef read_attachment(path_fragment: str) -> bytes:\n    final_path = BASE / path_fragment\n    return final_path.read_bytes()\n""",
-        },
-        vulnerabilities=[
-            VulnerabilitySpec(
-                id="H-01",
-                filename="api/auth.py",
-                line=12,
-                vuln_type="weak_authentication",
-                severity="critical",
-                title="Accepts unsigned JWT tokens when alg=none",
-                rationale="Token parser trusts attacker-controlled header and bypasses signature checks.",
-            ),
-            VulnerabilitySpec(
-                id="H-02",
-                filename="api/files.py",
-                line=11,
-                vuln_type="weak_authentication",
-                severity="high",
-                title="Tenant access controlled by request parameter",
-                rationale="Requester can switch tenant query parameter and read cross-tenant data (IDOR).",
-            ),
-            VulnerabilitySpec(
-                id="H-03",
-                filename="api/fetcher.py",
-                line=4,
-                vuln_type="ssrf",
-                severity="high",
-                title="Server-side fetch of arbitrary URL",
-                rationale="Attacker can query internal metadata endpoints through backend network path.",
-            ),
-            VulnerabilitySpec(
-                id="H-04",
-                filename="api/storage.py",
-                line=6,
-                vuln_type="path_traversal",
-                severity="critical",
-                title="Unvalidated path join for file reads",
-                rationale="Path fragments containing .. can escape upload directory.",
-            ),
-        ],
-    ),
-}
-
-SEVERITY_WEIGHTS = {
-    "low": 1.0,
-    "medium": 2.0,
-    "high": 3.0,
-    "critical": 4.0,
-}
-
-TARGET_CONFIDENCE = {
-    "low": 0.55,
-    "medium": 0.65,
-    "high": 0.8,
-    "critical": 0.9,
-}
-
-
-def get_task(task_id: str) -> TaskSpec:
-    if task_id not in TASKS:
-        raise KeyError(f"Unknown task_id '{task_id}'. Available: {', '.join(sorted(TASKS))}")
-    return TASKS[task_id]
-
-
-def list_task_ids() -> List[str]:
-    return sorted(TASKS.keys())