Spaces:
Sleeping
Sleeping
| """ | |
| SecureCodeEnv - Pre-Submission Validator | |
| Run this before pushing to HuggingFace Spaces. | |
| All checks must pass before submission. | |
| Usage: | |
| python validate.py | |
| python validate.py --url https://vishaldhakad-securecodeenv.hf.space | |
| """ | |
| import sys | |
| import os | |
| import json | |
| import requests | |
| import argparse | |
| import subprocess | |
| PASS = "✅" | |
| FAIL = "❌" | |
| WARN = "⚠️ " | |
| def check(name: str, ok: bool, detail: str = "") -> bool: | |
| icon = PASS if ok else FAIL | |
| line = f" {icon} {name}" | |
| if detail: | |
| line += f" — {detail}" | |
| print(line) | |
| return ok | |
| def validate_files() -> bool: | |
| print("\n── File Structure ──────────────────────────────────────────") | |
| required = [ | |
| "openenv.yaml", | |
| "Dockerfile", | |
| "inference.py", | |
| "requirements.txt", | |
| "README.md", | |
| "app/main.py", | |
| "app/routes.py", | |
| "app/models.py", | |
| "app/state.py", | |
| "graders/reward_aggregator.py", | |
| "graders/correctness.py", | |
| "graders/attacks.py", | |
| "graders/static_analysis.py", | |
| "graders/performance.py", | |
| "graders/consistency.py", | |
| "graders/documentation.py", | |
| "codegraph/graph.py", | |
| "codegraph/extractor.py", | |
| "codegraph/serializer.py", | |
| "sandbox/executor.py", | |
| "sandbox/payload_gen.py", | |
| "tasks/task_registry.py", | |
| ] | |
| all_ok = True | |
| for path in required: | |
| exists = os.path.exists(path) | |
| if not check(path, exists): | |
| all_ok = False | |
| return all_ok | |
| def validate_imports() -> bool: | |
| print("\n── Python Imports ──────────────────────────────────────────") | |
| checks = [ | |
| ("fastapi", "from fastapi import FastAPI"), | |
| ("pydantic", "from pydantic import BaseModel"), | |
| ("uvicorn", "import uvicorn"), | |
| ("bandit CLI", None), | |
| ] | |
| all_ok = True | |
| for name, stmt in checks: | |
| if stmt: | |
| try: | |
| exec(stmt) | |
| check(name, True) | |
| except ImportError as e: | |
| check(name, False, str(e)) | |
| all_ok = False | |
| else: | |
| # Check CLI tool | |
| result = subprocess.run(["bandit", "--version"], capture_output=True, text=True) | |
| ok = result.returncode == 0 | |
| check(f"bandit CLI", ok, result.stdout.strip()[:40] if ok else "not found — pip install bandit") | |
| if not ok: | |
| all_ok = False | |
| return all_ok | |
| def validate_task_registry() -> bool: | |
| print("\n── Task Registry ───────────────────────────────────────────") | |
| try: | |
| sys.path.insert(0, ".") | |
| from tasks.task_registry import TASK_REGISTRY, TASKS_BY_DIFFICULTY | |
| total = len(TASK_REGISTRY) | |
| check("Task registry loads", True, f"{total} tasks loaded") | |
| for diff in ["easy", "medium", "hard"]: | |
| n = len(TASKS_BY_DIFFICULTY.get(diff, [])) | |
| check(f"{diff} tasks", n >= 3, f"{n} tasks (need ≥ 3)") | |
| # Validate task structure | |
| for tid, task in TASK_REGISTRY.items(): | |
| has_required = all(k in task for k in ["id", "difficulty", "cwe_targets", "problem_statement", "test_cases"]) | |
| check(f"task {tid} structure", has_required) | |
| return True | |
| except Exception as e: | |
| check("Task registry import", False, str(e)[:80]) | |
| return False | |
| def validate_api(base_url: str) -> bool: | |
| print(f"\n── Live API: {base_url} ─────────────────────────────────────") | |
| all_ok = True | |
| # Health check | |
| try: | |
| r = requests.get(f"{base_url}/health", timeout=10) | |
| ok = r.status_code == 200 | |
| check("GET /health → 200", ok, r.json().get("env", "") if ok else f"HTTP {r.status_code}") | |
| if not ok: | |
| all_ok = False | |
| except Exception as e: | |
| check("GET /health", False, str(e)[:60]) | |
| return False | |
| # Reset | |
| for diff in ["easy", "medium", "hard"]: | |
| try: | |
| r = requests.post(f"{base_url}/reset", json={"difficulty": diff}, timeout=15) | |
| ok = r.status_code == 200 | |
| if ok: | |
| data = r.json() | |
| has_fields = all(k in data for k in ["session_id", "task_id", "problem_statement", "cwe_targets"]) | |
| check(f"POST /reset ({diff})", has_fields, data.get("task_id", "")) | |
| if not has_fields: | |
| all_ok = False | |
| # Step with trivial code | |
| sid = data["session_id"] | |
| step_r = requests.post(f"{base_url}/step", json={ | |
| "session_id": sid, | |
| "code": "def solution(): pass", | |
| "filename": "test.py", | |
| }, timeout=60) | |
| step_ok = step_r.status_code == 200 | |
| if step_ok: | |
| sdata = step_r.json() | |
| reward = sdata.get("total_reward", -1) | |
| in_range = 0.0 <= reward <= 1.0 | |
| check(f"POST /step ({diff}) → reward in [0,1]", in_range, f"reward={reward:.3f}") | |
| if not in_range: | |
| all_ok = False | |
| else: | |
| check(f"POST /step ({diff})", False, f"HTTP {step_r.status_code}") | |
| all_ok = False | |
| else: | |
| check(f"POST /reset ({diff})", False, f"HTTP {r.status_code}") | |
| all_ok = False | |
| except Exception as e: | |
| check(f"POST /reset ({diff})", False, str(e)[:60]) | |
| all_ok = False | |
| # State | |
| try: | |
| r2 = requests.post(f"{base_url}/reset", json={"difficulty": "easy"}, timeout=10) | |
| if r2.status_code == 200: | |
| sid = r2.json()["session_id"] | |
| state_r = requests.get(f"{base_url}/state", params={"session_id": sid}, timeout=10) | |
| check("GET /state", state_r.status_code == 200) | |
| except Exception: | |
| pass | |
| return all_ok | |
| def validate_openenv_yaml() -> bool: | |
| print("\n── openenv.yaml ────────────────────────────────────────────") | |
| try: | |
| import yaml | |
| with open("openenv.yaml") as f: | |
| spec = yaml.safe_load(f) | |
| required_keys = ["name", "version", "description", "action_space", "observation_space", "tasks", "reward"] | |
| for k in required_keys: | |
| check(f"has '{k}' field", k in spec) | |
| check("9 tasks defined", len(spec.get("tasks", [])) == 9, f"found {len(spec.get('tasks', []))}") | |
| return True | |
| except ImportError: | |
| print(f" {WARN} yaml not installed — skipping YAML validation (pip install pyyaml)") | |
| return True | |
| except Exception as e: | |
| check("openenv.yaml parses", False, str(e)[:80]) | |
| return False | |
| def main(): | |
| parser = argparse.ArgumentParser(description="SecureCodeEnv pre-submission validator") | |
| parser.add_argument("--url", default="http://localhost:7860", help="Base URL of the running environment") | |
| parser.add_argument("--skip-api", action="store_true", help="Skip live API checks") | |
| args = parser.parse_args() | |
| print("=" * 60) | |
| print(" SecureCodeEnv — Pre-Submission Validator") | |
| print("=" * 60) | |
| results = [ | |
| validate_files(), | |
| validate_imports(), | |
| validate_task_registry(), | |
| validate_openenv_yaml(), | |
| ] | |
| if not args.skip_api: | |
| results.append(validate_api(args.url)) | |
| else: | |
| print(f"\n {WARN} Skipping live API checks (--skip-api)") | |
| print("\n" + "=" * 60) | |
| passed = sum(results) | |
| total = len(results) | |
| if passed == total: | |
| print(f" {PASS} ALL CHECKS PASSED ({passed}/{total}) — ready to submit!") | |
| sys.exit(0) | |
| else: | |
| print(f" {FAIL} {total - passed} check group(s) failed ({passed}/{total} passed)") | |
| print(" Fix failures before submitting.") | |
| sys.exit(1) | |
| if __name__ == "__main__": | |
| main() | |