Initial commit: vuln-patch-env OpenEnv hackathon submission

.gitignore

@@ -0,0 +1,5 @@
+.env
+.venv/
+__pycache__/
+*.pyc
+.pytest_cache/

Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# Install dependencies first for Docker layer caching
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the application
+COPY . .
+
+# Expose the standard Hugging Face Spaces port
+EXPOSE 7860
+
+# Run the FastAPI server using the server package
+CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,429 @@
+# Vuln-Patch-Env
+
+A real-world OpenEnv environment for training and evaluating AI agents on automated code vulnerability detection and patching.
+
+---
+
+## Overview
+
+Vuln-Patch-Env simulates the task of **Static Application Security Testing (SAST) auto-remediation** — a genuine problem that security engineering teams solve daily. Given a snippet of Python code containing a known vulnerability, an agent must identify and patch the vulnerability without breaking the existing functionality.
+
+The environment exposes three tasks of increasing difficulty, covering three of the most common vulnerability classes found in real-world codebases:
+
+- **Hardcoded Secrets** (CWE-798)
+- **SQL Injection via f-strings** (CWE-89)
+- **Command Injection via os.system** (CWE-78)
+
+The environment is fully compliant with the [OpenEnv](https://huggingface.co/openenv) specification and is designed to be used as a benchmark for evaluating LLM-based security agents.
+
+---
+
+## Motivation
+
+Vulnerability remediation is a task that organizations spend significant engineering resources on. Automated patching tools exist but are largely rule-based and brittle. Training RL or LLM agents on a structured environment like this one opens the door to agents that can:
+
+- Understand code semantics, not just surface patterns
+- Generalize across variable names, table names, and command structures
+- Learn to use safe APIs (parameterized queries, subprocess with argument lists) as a matter of policy
+
+This environment fills a gap in the OpenEnv ecosystem by providing a **code security domain** benchmark with rigorous, AST-based grading that rewards only structurally correct fixes — not superficial string changes.
+
+---
+
+## Environment Specification
+
+### OpenEnv Compliance
+
+| Interface | Implementation |
+|---|---|
+| `reset()` | Returns initial `Observation`. Resets step counter, generates fresh vulnerable code. |
+| `step(action)` | Returns `(Observation, Reward, done: bool, Info)` |
+| `state()` | Returns current `Observation` without advancing the episode |
+| `close()` | No-op cleanup method satisfying the OpenEnv spec |
+| Typed models | `Observation`, `Action`, `Reward`, `Info` all defined as Pydantic models |
+| `openenv.yaml` | Present at project root with name, version, entrypoint, and task list |
+
+---
+
+## Data Models
+
+### Observation
+
+Returned by `reset()`, `step()`, and `state()`.
+
+| Field | Type | Description |
+|---|---|---|
+| `step` | `int` | Current step index within the episode (starts at 0) |
+| `vulnerability_type` | `str` | Task identifier: `"easy"`, `"medium"`, or `"hard"` |
+| `current_code` | `str` | The Python source code containing the vulnerability to be fixed |
+| `linter_output` | `str` | Output from the security scanner. `"Not run yet."` until a scan is performed |
+
+### Action
+
+Submitted by the agent on each step.
+
+| Field | Type | Required | Description |
+|---|---|---|---|
+| `action_type` | `str` | Always | Either `"run_scan"` or `"submit_patch"` |
+| `patched_code` | `str` | Only for `"submit_patch"` | The complete fixed Python code as a string |
+
+### Reward
+
+| Field | Type | Description |
+|---|---|---|
+| `value` | `float` | Scalar reward in the range `[0.0, 1.0]` |
+
+### Info
+
+| Field | Type | Description |
+|---|---|---|
+| `error` | `str` or `None` | Environment-level error message, if any |
+
+---
+
+## Action Space
+
+The agent has exactly two possible action types per step:
+
+**1. `run_scan`**
+Triggers the security linter. Returns a hint in the `linter_output` field of the next observation. Grants a small reward (`+0.1`) as a partial progress signal, encouraging the agent to gather information before patching.
+
+**2. `submit_patch`**
+Submits the agent's proposed fix via the `patched_code` field. The environment grades the patch using a hybrid AST + string analysis grader and returns a reward between `0.0` and `1.0`. The episode ends immediately upon submission.
+
+---
+
+## Observation Space
+
+The agent receives a structured JSON observation on every step containing:
+
+- The **current step index** so the agent can manage its budget
+- The **vulnerability type** so the agent knows what class of problem it is solving
+- The **current source code** — the same vulnerable snippet on every step (the environment does not modify the code between steps)
+- The **linter output** — populated only after `run_scan` is called, providing a natural-language hint about the vulnerability class
+
+---
+
+## Tasks and Difficulty
+
+### Task 1 — Easy: Hardcoded Secrets (CWE-798)
+
+**Objective:** Remove a hardcoded API key from the source code and replace it with a call to `os.getenv()` or an equivalent safe environment-variable access pattern.
+
+**Example vulnerable code generated:**
+```python
+import os
+
+def fetch_resource():
+    ACCESS_KEY = 'sk-A7FX29KQBR1NWLTZ'
+    return fetch(ACCESS_KEY)
+```
+
+**Expected fix:**
+```python
+import os
+
+def fetch_resource():
+    ACCESS_KEY = os.getenv('ACCESS_KEY')
+    return fetch(ACCESS_KEY)
+```
+
+**Grading breakdown:**
+| Condition | Reward |
+|---|---|
+| Hardcoded secret string is absent from the patched code | +0.5 |
+| Code uses a safe env-var access pattern (AST-verified) | +0.5 |
+| **Maximum total** | **1.0** |
+
+The grader accepts all of the following valid patterns via AST analysis:
+- `os.getenv("KEY")`
+- `os.environ.get("KEY")`
+- `os.environ["KEY"]`
+- Any bare reference to `os.environ`
+
+---
+
+### Task 2 — Medium: SQL Injection via f-strings (CWE-89)
+
+**Objective:** Replace an f-string SQL query with a parameterized query to prevent SQL injection.
+
+**Example vulnerable code generated:**
+```python
+import sqlite3
+
+def get_users(name):
+    query = f"SELECT * FROM users WHERE name='{name}'"
+    cursor.execute(query)
+```
+
+**Expected fix:**
+```python
+import sqlite3
+
+def get_users(name):
+    query = "SELECT * FROM users WHERE name=?"
+    cursor.execute(query, (name,))
+```
+
+**Grading breakdown:**
+| Condition | Reward |
+|---|---|
+| f-string SQL construction is absent from patched code | +0.4 |
+| `cursor.execute()` is called with a parameter argument (AST-verified) | +0.6 |
+| Fallback: patch contains `?` or `%s` placeholder (string check) | +0.4 (instead of +0.6) |
+| **Maximum total** | **1.0** |
+
+---
+
+### Task 3 — Hard: Command Injection via os.system (CWE-78)
+
+**Objective:** Replace an `os.system()` call that passes unsanitized user input with a `subprocess.run()` call using a separated argument list, which prevents shell injection.
+
+**Example vulnerable code generated:**
+```python
+import os
+import subprocess
+
+def run_util(user_arg):
+    os.system(f'ping -c 4 {user_arg}')
+```
+
+**Expected fix:**
+```python
+import subprocess
+
+def run_util(user_arg):
+    subprocess.run(["ping", "-c", "4", user_arg])
+```
+
+**Grading breakdown:**
+| Condition | Reward |
+|---|---|
+| `os.system` is absent from the patched code | +0.3 |
+| `subprocess.run()` (or equivalent) is called with a list argument and no `shell=True` (AST-verified) | +0.7 |
+| Fallback: patch contains `subprocess` with list brackets (string check) | +0.4 (instead of +0.7) |
+| **Maximum total** | **1.0** |
+
+---
+
+## Reward Function Design
+
+The reward function is designed to provide **meaningful signal over the full trajectory**, not just at episode end.
+
+| Event | Reward | Rationale |
+|---|---|---|
+| `run_scan` action | +0.1 | Encourages information gathering before patching |
+| Correct patch (full credit) | +1.0 | Agent removed the vulnerability and used the correct safe API |
+| Partial patch (fallback) | +0.4 to +0.8 | Agent improved security but did not use the ideal pattern |
+| Reaching step 5 without completing | -0.2 | Penalizes agents that loop without making progress |
+| All rewards | Clamped to `[0.0, 1.0]` | Strict compliance with OpenEnv spec |
+
+An episode ends when the agent calls `submit_patch` or when the step count reaches 5 (whichever comes first).
+
+---
+
+## Grading Methodology
+
+A key design decision in this environment is the use of **Python's Abstract Syntax Tree (AST) module** for grading, rather than simple string matching or regex. This makes the grader:
+
+- **Robust to formatting differences** — whitespace, line breaks, and quote style do not affect the grade
+- **Semantically accurate** — a patch is only credited if the correct API is actually *called* in the code, not just mentioned in a comment
+- **Resistant to false positives** — for example, `subprocess.run(..., shell=True)` is explicitly detected and rejected even though it contains `subprocess`
+
+Each grader function (`uses_os_getenv`, `uses_parameterized_query`, `uses_safe_subprocess`) parses the submitted code into an AST and walks the node tree to verify the structural correctness of the fix. All graders fall back to string-based checks if the AST check is inconclusive, ensuring partial credit is awarded for genuinely improved but imperfect patches.
+
+---
+
+## Dynamic Code Generation
+
+To prevent an LLM agent from memorizing fixed vulnerable snippets, the environment **generates code dynamically** on each `reset()` call by randomly selecting from pools of variable names, function names, table names, SQL fields, and shell commands. A fixed random seed (`seed=42`) is applied at the start of each `reset()` to ensure that baseline scores are fully reproducible across runs.
+
+---
+
+## Episode Lifecycle
+
+```
+reset(task)
+    |
+    v
+Observation (step=0, vulnerable code, linter="Not run yet.")
+    |
+    v
+Agent calls run_scan         --> reward=0.1, linter hint populated
+    |
+    v
+Agent calls submit_patch     --> grader runs, reward=0.0–1.0, done=True
+    |
+    v
+[END] logged, episode complete
+```
+
+Maximum steps per episode: **5**. If the agent does not submit a patch within 5 steps, the episode is forcibly ended with a `-0.2` penalty.
+
+---
+
+## Baseline Inference Script
+
+The `inference.py` script runs a full three-task evaluation loop using any OpenAI-compatible LLM endpoint.
+
+**Required environment variables:**
+
+| Variable | Description | Default |
+|---|---|---|
+| `HF_TOKEN` | Hugging Face API token (or any OpenAI-compatible API key) | None (required) |
+| `MODEL_NAME` | Model identifier to use for inference | `meta-llama/Llama-3.3-70B-Instruct` |
+| `API_BASE_URL` | Base URL for the OpenAI-compatible inference endpoint | `https://router.huggingface.co/v1` |
+
+**Running the baseline:**
+```bash
+export HF_TOKEN="your_hf_token_here"
+export MODEL_NAME="meta-llama/Llama-3.3-70B-Instruct"
+export API_BASE_URL="https://router.huggingface.co/v1"
+
+python inference.py
+```
+
+**Expected stdout format:**
+```
+[START] task=easy env=vuln-patch-env model=meta-llama/Llama-3.3-70B-Instruct
+[STEP] step=1 action=run_scan() reward=0.10 done=false error=null
+[STEP] step=2 action=submit_patch() reward=1.00 done=true error=null
+[END] success=true steps=2 score=1.00 rewards=0.10,1.00
+
+[START] task=medium env=vuln-patch-env model=meta-llama/Llama-3.3-70B-Instruct
+[STEP] step=1 action=submit_patch() reward=1.00 done=true error=null
+[END] success=true steps=1 score=1.00 rewards=1.00
+
+[START] task=hard env=vuln-patch-env model=meta-llama/Llama-3.3-70B-Instruct
+[STEP] step=1 action=run_scan() reward=0.10 done=false error=null
+[STEP] step=2 action=submit_patch() reward=1.00 done=true error=null
+[END] success=true steps=2 score=1.00 rewards=0.10,1.00
+```
+
+**Baseline scores (Llama-3.3-70B-Instruct, temperature=0.0):**
+
+| Task | Expected Score |
+|---|---|
+| easy | 1.00 |
+| medium | >= 0.80 |
+| hard | >= 0.80 |
+
+---
+
+## REST API Reference
+
+The environment is served as a FastAPI application. All endpoints are available once the server is running.
+
+| Method | Endpoint | Description |
+|---|---|---|
+| `GET` | `/` | Health check, returns server status |
+| `GET` | `/health` | Returns `{"status": "healthy"}` — required by `openenv validate` |
+| `GET` | `/metadata` | Returns environment name, description, version, and task list |
+| `GET` | `/schema` | Returns JSON schemas for `Action`, `Observation`, and state |
+| `POST` | `/mcp` | Minimal MCP (Model Context Protocol) endpoint — JSON-RPC 2.0 |
+| `POST` | `/reset` | Resets the environment. Accepts `{"task": "easy" | "medium" | "hard"}` |
+| `POST` | `/step` | Takes one step. Accepts an `Action` object as JSON body |
+| `GET` | `/state` | Returns the current observation without advancing the episode |
+
+**Example: Reset to the hard task**
+```bash
+curl -X POST http://localhost:7860/reset \
+  -H "Content-Type: application/json" \
+  -d '{"task": "hard"}'
+```
+
+**Example: Submit a patch**
+```bash
+curl -X POST http://localhost:7860/step \
+  -H "Content-Type: application/json" \
+  -d '{"action_type": "submit_patch", "patched_code": "import subprocess\n\ndef run_util(user_arg):\n    subprocess.run([\"ping\", \"-c\", \"4\", user_arg])"}'
+```
+
+---
+
+## Project Structure
+
+```
+vuln-patch-env/
+|
+|-- server/
+|   |-- __init__.py        # Exports the FastAPI app
+|   +-- app.py             # FastAPI server with all OpenEnv-required endpoints
+|
+|-- environment.py         # Core environment: Pydantic models, AST graders, VulnPatchEnv class
+|-- inference.py           # Baseline inference script using OpenAI-compatible client
+|-- server.py              # Root-level entry point (re-exports server/app.py)
+|-- openenv.yaml           # OpenEnv metadata declaration
+|-- pyproject.toml         # Python package configuration
+|-- requirements.txt       # Pinned dependencies
+|-- Dockerfile             # Container definition for Hugging Face Spaces deployment
+|-- README.md              # This file
++-- .gitignore
+```
+
+---
+
+## Setup and Local Usage
+
+**1. Clone the repository and install dependencies:**
+```bash
+git clone https://github.com/YOUR_USERNAME/vuln-patch-env.git
+cd vuln-patch-env
+pip install -r requirements.txt
+```
+
+**2. Start the server locally:**
+```bash
+uvicorn server.app:app --host 0.0.0.0 --port 7860
+```
+
+**3. Validate the OpenEnv spec compliance:**
+```bash
+openenv validate .
+```
+
+**4. Use the environment directly in Python:**
+```python
+from environment import VulnPatchEnv, Action
+
+# Run the easy task
+env = VulnPatchEnv(task="easy")
+obs = env.reset()
+print(obs.current_code)
+
+# Get a hint
+obs, reward, done, info = env.step(Action(action_type="run_scan"))
+print(obs.linter_output)
+
+# Submit a patch
+patched = obs.current_code.replace("ACCESS_KEY = 'sk-...'", "ACCESS_KEY = os.getenv('ACCESS_KEY')")
+obs, reward, done, info = env.step(Action(action_type="submit_patch", patched_code=patched))
+print(f"Reward: {reward.value}")
+
+env.close()
+```
+
+---
+
+## Docker
+
+**Build and run locally:**
+```bash
+docker build -t vuln-patch-env .
+docker run -p 7860:7860 vuln-patch-env
+```
+
+The server will be available at `http://localhost:7860`.
+
+---
+
+## Dependencies
+
+| Package | Version | Purpose |
+|---|---|---|
+| `openai` | 2.30.0 | OpenAI-compatible client for inference script |
+| `pydantic` | 2.12.5 | Typed data models for Observation, Action, Reward, Info |
+| `fastapi` | 0.135.3 | REST API server |
+| `uvicorn` | 0.44.0 | ASGI server for FastAPI |
+| `openenv-core` | 0.2.3 | OpenEnv spec utilities and validation |
+| `python-dotenv` | 1.2.2 | Environment variable loading from `.env` files

environment.py

@@ -0,0 +1,226 @@
+import ast
+import random
+import string
+from typing import Optional
+
+from pydantic import BaseModel
+
+
+# 1. OpenEnv Typed Models
+class Observation(BaseModel):
+    step: int
+    vulnerability_type: str
+    current_code: str
+    linter_output: str
+
+
+class Action(BaseModel):
+    action_type: str  # e.g., "run_scan" or "submit_patch"
+    patched_code: Optional[str] = ""
+
+
+class Reward(BaseModel):
+    value: float
+
+
+class Info(BaseModel):
+    error: Optional[str] = None
+
+
+# AST Checkers for Robust Grading
+def uses_os_getenv(code: str) -> bool:
+    """
+    Returns True if the code uses any safe environment-variable access pattern:
+      - os.getenv("KEY")          -> ast.Call with Attribute attr='getenv'
+      - os.environ.get("KEY")     -> ast.Call with chained Attribute: environ -> get
+      - os.environ["KEY"]         -> ast.Subscript on os.environ
+      - bare os.environ reference -> ast.Attribute with attr='environ'
+    """
+    try:
+        tree = ast.parse(code)
+        for node in ast.walk(tree):
+            # Pattern 1: os.getenv(...) or getenv(...)
+            if isinstance(node, ast.Call):
+                func = node.func
+                if isinstance(func, ast.Attribute) and func.attr == "getenv":
+                    return True
+                if isinstance(func, ast.Name) and func.id == "getenv":
+                    return True
+                # Pattern 2: os.environ.get(...) — chained call
+                if (
+                    isinstance(func, ast.Attribute)
+                    and func.attr == "get"
+                    and isinstance(func.value, ast.Attribute)
+                    and func.value.attr == "environ"
+                ):
+                    return True
+            # Pattern 3: os.environ["KEY"] — subscript access
+            if isinstance(node, ast.Subscript):
+                val = node.value
+                if isinstance(val, ast.Attribute) and val.attr == "environ":
+                    return True
+            # Pattern 4: bare os.environ reference (e.g. env = os.environ)
+            if isinstance(node, ast.Attribute) and node.attr == "environ":
+                return True
+    except SyntaxError:
+        pass
+    return False
+
+
+def uses_parameterized_query(code: str) -> bool:
+    try:
+        tree = ast.parse(code)
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call):
+                # Simple heuristic mapping for typical execution paths
+                if hasattr(node.func, "attr") and node.func.attr == "execute":
+                    # length of args > 1 indicates query + params, or uses named keywords like parameters=...
+                    if len(node.args) > 1 or (
+                        node.keywords
+                        and any(
+                            k.arg in ("parameters", "params") for k in node.keywords
+                        )
+                    ):
+                        return True
+                    # Check if any arg is explicitly a tuple or dictionary, common in SQLite parameterization
+                    for arg in node.args:
+                        if isinstance(arg, ast.Tuple) or isinstance(arg, ast.Dict):
+                            return True
+    except SyntaxError:
+        pass
+    return False
+
+
+def uses_safe_subprocess(code: str) -> bool:
+    try:
+        tree = ast.parse(code)
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call):
+                if hasattr(node.func, "attr") and node.func.attr in (
+                    "run",
+                    "Popen",
+                    "call",
+                    "check_call",
+                    "check_output",
+                ):
+                    # Ensure shell=True is NOT passed
+                    is_shell = False
+                    for kw in node.keywords:
+                        if (
+                            kw.arg == "shell"
+                            and isinstance(kw.value, ast.Constant)
+                            and kw.value.value is True
+                        ):
+                            is_shell = True
+                    if not is_shell:
+                        # Ensure first argument is a list (args array)
+                        if node.args and isinstance(node.args[0], ast.List):
+                            return True
+    except SyntaxError:
+        pass
+    return False
+
+
+# 2. Environment Implementation
+class VulnPatchEnv:
+    def __init__(self, task="easy"):
+        self.task = task
+        self.step_count = 0
+        self.done = False
+        self.code = ""
+        self.linter = ""
+        self.target_secret = ""
+        self.reset()
+
+    def reset(self) -> Observation:
+        random.seed(42)  # Fixed seed for reproducible baseline scores
+        self.step_count = 0
+        self.done = False
+        self.linter = "Not run yet."
+
+        # Dynamic variable generation prevents simple string memorization by the LLM
+        if self.task == "easy":
+            key_val = "sk-" + "".join(
+                random.choices(string.ascii_uppercase + string.digits, k=16)
+            )
+            var_name = random.choice(
+                ["API_KEY", "SECRET_KEY", "AUTH_TOKEN", "ACCESS_KEY"]
+            )
+            func_name = random.choice(["get_data", "fetch_resource", "load_user"])
+            self.code = f"import os\n\ndef {func_name}():\n    {var_name} = '{key_val}'\n    return fetch({var_name})"
+            self.target_secret = key_val
+
+        elif self.task == "medium":
+            table = random.choice(["users", "accounts", "employees", "orders"])
+            field = random.choice(["name", "email", "username", "id"])
+            self.code = f"import sqlite3\n\ndef get_{table}({field}):\n    query = f\"SELECT * FROM {table} WHERE {field}='{{{field}}}'\"\n    cursor.execute(query)"
+
+        elif self.task == "hard":
+            cmd = random.choice(["ping -c 4", "ls -l", "curl", "nmap"])
+            self.code = f"import os\nimport subprocess\n\ndef run_util(user_arg):\n    os.system(f'{cmd} {{user_arg}}')"
+
+        else:
+            self.code = "Unknown task."
+            self.target_secret = ""
+
+        return self.state()
+
+    def state(self) -> Observation:
+        return Observation(
+            step=self.step_count,
+            vulnerability_type=self.task,
+            current_code=self.code,
+            linter_output=self.linter,
+        )
+
+    def step(self, action: Action) -> tuple[Observation, Reward, bool, Info]:
+        self.step_count += 1
+        reward_val = 0.0
+
+        if action.action_type == "run_scan":
+            self.linter = "SECURITY SCAN: Vulnerability detected. Fix hardcoded secrets, SQLi, or Command Injection."
+            reward_val = 0.1  # Incremental progress signal
+
+        elif action.action_type == "submit_patch":
+            patched = action.patched_code if action.patched_code else ""
+
+            # Hybrid AST/String Grading for robustness against formatting
+            if self.task == "easy":
+                if self.target_secret and self.target_secret not in patched:
+                    reward_val += 0.5
+                if uses_os_getenv(patched):
+                    reward_val += 0.5
+
+            elif self.task == "medium":
+                if 'f"SELECT' not in patched and "f'SELECT" not in patched:
+                    reward_val += 0.4
+                if uses_parameterized_query(patched):
+                    reward_val += 0.6
+                elif "?" in patched or "%s" in patched:  # Fallback text format check
+                    reward_val += 0.4
+
+            elif self.task == "hard":
+                if "os.system" not in patched:
+                    reward_val += 0.3
+                if uses_safe_subprocess(patched):
+                    reward_val += 0.7
+                elif (
+                    "subprocess" in patched and "[" in patched and "]" in patched
+                ):  # Fallback text format check
+                    reward_val += 0.4
+
+            self.done = True
+
+        # Hard limit to prevent infinite loops (Penalize logic per OpenEnv spec requirement)
+        if self.step_count >= 5 and not self.done:
+            self.done = True
+            reward_val -= 0.2
+
+        # Ensure reward is strictly between 0.0 and 1.0 per OpenEnv spec requirement
+        reward_val = min(max(reward_val, 0.0), 1.0)
+
+        return self.state(), Reward(value=reward_val), self.done, Info()
+
+    def close(self) -> None:
+        """No-op cleanup method required by the OpenEnv spec."""
+        pass

inference.py

@@ -0,0 +1,97 @@
+import json
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+from openai import OpenAI
+
+from environment import Action, VulnPatchEnv
+
+# Strict Environment Variables required by the hackathon rubric
+API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")
+MODEL_NAME = os.getenv("MODEL_NAME", "meta-llama/Llama-3.3-70B-Instruct")
+HF_TOKEN = os.getenv("HF_TOKEN") or os.getenv("API_KEY")
+
+if not HF_TOKEN:
+    print("WARNING: HF_TOKEN is missing. API calls will fail.", flush=True)
+
+client = OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN or "dummy-key")
+
+
+def run_episode(task_name: str):
+    env = VulnPatchEnv(task=task_name)
+    obs = env.reset()
+
+    # REQUIRED [START] line
+    print(f"[START] task={task_name} env=vuln-patch-env model={MODEL_NAME}", flush=True)
+
+    done = False
+    rewards = []
+
+    try:
+        while not done:
+            prompt = (
+                f"Task: Fix the vulnerability in the code.\n"
+                f"Observation: {obs.model_dump_json()}\n"
+                "Output valid JSON ONLY. Required keys:\n"
+                "- 'action_type': Must be 'run_scan' or 'submit_patch'.\n"
+                "- 'patched_code': The patched python code (string). Only required if submitting."
+            )
+
+            try:
+                response = client.chat.completions.create(
+                    model=MODEL_NAME,
+                    messages=[
+                        {
+                            "role": "system",
+                            "content": "You are a cyber security agent who is an expert in python and security. You are given a task to fix the vulnerability in the code and can find bugs in the code also. Always output valid JSON.",
+                        },
+                        {"role": "user", "content": prompt},
+                    ],
+                    response_format={"type": "json_object"},
+                    temperature=0.0,  # Deterministic LLM response
+                    timeout=60,  # Prevent indefinite hang on slow API
+                )
+                raw_content = response.choices[0].message.content
+                action_data = json.loads(raw_content)
+                action = Action(**action_data)
+                error_msg = "null"
+            except Exception as e:
+                action = Action(action_type="error", patched_code="")
+                error_msg = str(e).replace("\n", " ")
+
+            # Step the environment
+            obs, reward_obj, done, info = env.step(action)
+            reward = reward_obj.value
+            rewards.append(reward)
+
+            # Use environment's info.error if set, else fall back to LLM error, else null
+            env_error = info.error if info.error else None
+            step_error = env_error or (error_msg if error_msg != "null" else None)
+            step_error_str = step_error if step_error else "null"
+
+            # REQUIRED [STEP] line (no newlines, 2 decimal places, lowercase bools)
+            action_safe_str = f"{action.action_type}()"
+            done_str = "true" if done else "false"
+            print(
+                f"[STEP] step={env.step_count} action={action_safe_str} reward={reward:.2f} done={done_str} error={step_error_str}",
+                flush=True,
+            )
+
+    finally:
+        env.close()
+        # REQUIRED [END] line — always emitted even on exception, score to 2 decimal places
+        score = rewards[-1] if rewards else 0.0
+        score = min(max(score, 0.0), 1.0)  # Clamp score to 0.0 - 1.0
+        success_str = "true" if score >= 0.8 else "false"
+        rewards_str = ",".join([f"{r:.2f}" for r in rewards])
+        print(
+            f"[END] success={success_str} steps={env.step_count} score={score:.2f} rewards={rewards_str}",
+            flush=True,
+        )
+
+
+if __name__ == "__main__":
+    for t in ["easy", "medium", "hard"]:
+        run_episode(t)

openenv.yaml

@@ -0,0 +1,8 @@
+name: "vuln-patch-env"
+version: "1.0.0"
+entrypoint: "environment:VulnPatchEnv"
+description: "A real-world code security environment where agents detect and patch CVEs (Hardcoded Secrets, SQLi, Command Injection) without breaking functionality."
+tasks:
+  - easy
+  - medium
+  - hard

pyproject.toml

@@ -0,0 +1,31 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.backends.legacy:build"
+
+[project]
+name = "vuln-patch-env"
+version = "1.0.0"
+description = "A real-world code security environment where AI agents detect and patch vulnerabilities"
+requires-python = ">=3.10"
+dependencies = [
+    "openai",
+    "pydantic",
+    "fastapi",
+    "uvicorn",
+    "openenv-core",
+    "python-dotenv",
+]
+
+[project.scripts]
+server = "server.app:main"
+
+[project.urls]
+Homepage = "https://huggingface.co/spaces"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["server"]
+package-dir = { "server" = "server" }
+
+[tool.openenv]
+environment = "environment:VulnPatchEnv"

requirements.txt

@@ -0,0 +1,6 @@
+openai==2.30.0
+pydantic==2.12.5
+fastapi==0.135.3
+uvicorn==0.44.0
+openenv-core==0.2.3
+python-dotenv==1.2.2

server.py

@@ -0,0 +1,10 @@
+"""
+server.py — Root-level entry point for backwards compatibility.
+The canonical server is defined in server/app.py.
+"""
+from server.app import app, main
+
+__all__ = ["app"]
+
+if __name__ == "__main__":
+    main()

server/__init__.py

@@ -0,0 +1,4 @@
+# server package
+from server.app import app
+
+__all__ = ["app"]

server/app.py

@@ -0,0 +1,124 @@
+"""
+server/app.py — OpenEnv-compatible FastAPI application.
+This is the canonical app entry point expected by `openenv validate`.
+The root server.py re-exports this for backwards compatibility.
+"""
+from fastapi import FastAPI, Request
+from environment import VulnPatchEnv
+
+app = FastAPI(
+    title="vuln-patch-env",
+    description="OpenEnv environment for code vulnerability detection and patching.",
+    version="1.0.0",
+)
+
+# One shared environment instance per server process (stateless reset on each call)
+_env = VulnPatchEnv()
+
+
+@app.get("/")
+async def health_check():
+    return {"status": "running", "message": "vuln-patch-env OpenEnv Server is live"}
+
+
+@app.get("/health")
+async def health():
+    """Required by openenv validate — must return {"status": "healthy"}."""
+    return {"status": "healthy"}
+
+
+@app.get("/metadata")
+async def metadata():
+    """Required by openenv validate — must return name and description."""
+    return {
+        "name": "vuln-patch-env",
+        "description": (
+            "A real-world code security environment where AI agents detect "
+            "and patch vulnerabilities (hardcoded secrets, SQL injection, "
+            "command injection) in Python code."
+        ),
+        "version": "1.0.0",
+        "tasks": ["easy", "medium", "hard"],
+    }
+
+
+@app.get("/schema")
+async def schema():
+    """Required by openenv validate — must return action, observation and state schemas."""
+    from environment import Action, Observation
+    return {
+        "action": Action.model_json_schema(),
+        "observation": Observation.model_json_schema(),
+        "state": Observation.model_json_schema(),  # state has same shape as observation
+    }
+
+
+@app.post("/mcp")
+async def mcp_endpoint(request: Request):
+    """
+    Minimal Model Context Protocol (MCP) endpoint.
+    Required by openenv validate — must return a JSON-RPC 2.0 envelope.
+    """
+    try:
+        body = await request.json()
+    except Exception:
+        body = {}
+
+    return {
+        "jsonrpc": "2.0",
+        "id": body.get("id", 1),
+        "result": {
+            "name": "vuln-patch-env",
+            "description": "OpenEnv environment for code vulnerability patching.",
+            "tools": ["reset", "step", "state"],
+        },
+    }
+
+
+@app.post("/reset")
+async def reset_endpoint(request: Request):
+    """Reset the environment and return the initial observation."""
+    try:
+        body = await request.json()
+        task = body.get("task", "easy")
+    except Exception:
+        task = "easy"
+
+    _env.task = task
+    obs = _env.reset()
+    return {"status": "ok", "observation": obs.model_dump()}
+
+
+@app.post("/step")
+async def step_endpoint(request: Request):
+    """Take one step in the environment."""
+    from environment import Action
+    try:
+        body = await request.json()
+        action = Action(**body)
+    except Exception as e:
+        return {"error": str(e)}, 400
+
+    obs, reward, done, info = _env.step(action)
+    return {
+        "observation": obs.model_dump(),
+        "reward": reward.value,
+        "done": done,
+        "info": info.model_dump(),
+    }
+
+
+@app.get("/state")
+async def state_endpoint():
+    """Return the current environment state."""
+    return {"observation": _env.state().model_dump()}
+
+
+def main():
+    """Entry point for the server script (used by pyproject.toml [project.scripts])."""
+    import uvicorn
+    uvicorn.run("server.app:app", host="0.0.0.0", port=7860, reload=False)
+
+
+if __name__ == "__main__":
+    main()