""" Inference Script for SolidityGuard OpenEnv ========================================== MANDATORY Environment Variables: API_BASE_URL The API endpoint for the LLM. MODEL_NAME The model identifier to use for inference. HF_TOKEN Your Hugging Face / API key. SOLIDITYGUARD_TASK The task to run (set by validator for each task) STDOUT FORMAT: [START] task= env= model= [STEP] step= action= reward=<0.00> done= error= [END] success= steps= score= rewards= """ from __future__ import annotations import json import os import sys from typing import Any, Dict, List, Optional from openai import OpenAI from environment import SolidityGuardEnv # Environment variables API_KEY = os.getenv("HF_TOKEN") or os.getenv("API_KEY") API_BASE_URL = os.getenv("API_BASE_URL") or "https://router.huggingface.co/v1" MODEL_NAME = os.getenv("MODEL_NAME") or "Qwen/Qwen2.5-72B-Instruct" # Task configuration - validator may set this for single-task runs TASK_NAME = os.getenv("SOLIDITYGUARD_TASK") or os.getenv("TASK_NAME") BENCHMARK = os.getenv("SOLIDITYGUARD_BENCHMARK", "solidityguard") DEFAULT_TASKS = [ "task_1_best_practices", "task_2_gas_optimization", "task_3_security", ] MAX_STEPS = 1 # Each task has 1 step in our environment SUCCESS_SCORE_THRESHOLD = 0.1 START_TAG = "START" STEP_TAG = "STEP" END_TAG = "END" def log_start(task: str, env: str, model: str) -> None: print(f"[{START_TAG}] task={task} env={env} model={model}", flush=True) def log_step( step: int, action: str, reward: float, done: bool, error: Optional[str] ) -> None: error_val = error if error else "null" done_val = str(done).lower() print( f"[{STEP_TAG}] step={step} action={action} reward={reward:.2f} done={done_val} error={error_val}", flush=True, ) def log_end(success: bool, steps: int, score: float, rewards: List[float]) -> None: rewards_str = ",".join(f"{r:.2f}" for r in rewards) print( f"[{END_TAG}] success={str(success).lower()} steps={steps} score={score:.2f} rewards={rewards_str}", flush=True, ) def _safe_score(value: float) -> float: """Clamp score to valid range (0, 1) exclusive.""" if value <= 0.0: return 0.01 if value >= 1.0: return 0.99 return round(value, 2) def _find_line_number(source_code: str, needle: str, default: int) -> int: for idx, line in enumerate(source_code.splitlines(), start=1): if needle in line: return idx return default def _build_prompt(source_code: str, task_id: str) -> str: task_info = { "task_1_best_practices": "Find syntax and best-practice issues: missing SPDX license, old compiler version (<0.8.x), missing NatSpec comments, deprecated constructor syntax.", "task_2_gas_optimization": "Find gas optimization opportunities: unbounded loops, redundant storage reads, missing custom errors (use custom errors instead of require strings).", "task_3_security": "Find security vulnerabilities: reentrancy bugs, missing access control, tx.origin usage for authorization, integer overflow/underflow.", } return ( "Review this Solidity contract and return ONLY a JSON array of findings. " "Each finding must include: issue_type, line_number, description, severity (Critical/Medium/Low/Info). " f"Focus on: {task_info.get(task_id, task_id)}\n\n" f"Contract:\n{source_code}" ) def _fallback_actions(source_code: str, task_id: str) -> List[Dict[str, Any]]: """Deterministic fallback when LLM call fails or returns empty.""" lowered = source_code.lower() if task_id == "task_1_best_practices": return [ { "issue_type": "missing_spdx", "line_number": 1, "description": "Missing SPDX license identifier", "severity": "Low", }, { "issue_type": "old_compiler_version", "line_number": _find_line_number(source_code, "pragma solidity", 2), "description": "Compiler version below 0.8.x", "severity": "Low", }, ] if task_id == "task_2_gas_optimization": if "for" in lowered and ".length" in lowered: return [ { "issue_type": "unbounded_loop", "line_number": _find_line_number(source_code, "for", 10), "description": "Loop uses dynamic array length without bounds", "severity": "Medium", } ] return [ { "issue_type": "redundant_storage_read", "line_number": _find_line_number(source_code, "fee", 12), "description": "Repeated storage reads could be cached", "severity": "Medium", } ] if task_id == "task_3_security": if "tx.origin" in lowered: return [ { "issue_type": "tx_origin_auth", "line_number": _find_line_number(source_code, "tx.origin", 11), "description": "Authorization uses tx.origin", "severity": "Critical", } ] if "delegatecall" in lowered: return [ { "issue_type": "unsafe_delegatecall", "line_number": _find_line_number(source_code, "delegatecall", 15), "description": "Delegatecall without proper validation", "severity": "Critical", } ] if "call{" in lowered or ".call(" in lowered: return [ { "issue_type": "reentrancy", "line_number": _find_line_number(source_code, "call{", 13), "description": "State update after external call allows reentrancy", "severity": "Critical", } ] return [ { "issue_type": "missing_access_control", "line_number": 9, "description": "Sensitive function lacks access control", "severity": "Critical", } ] # Default fallback return [ { "issue_type": "general_issue", "line_number": 1, "description": "Potential issue detected", "severity": "Low", } ] def _call_model(client: OpenAI, prompt: str) -> List[Dict[str, Any]]: """Call LLM to analyze contract.""" try: response = client.chat.completions.create( model=MODEL_NAME, messages=[ { "role": "system", "content": "You are a Solidity security reviewer. Return ONLY a valid JSON array of findings, no explanation.", }, {"role": "user", "content": prompt}, ], temperature=0.1, max_tokens=800, stream=False, ) content = (response.choices[0].message.content or "[]").strip() # Handle markdown code blocks if content.startswith("```"): parts = content.split("```") if len(parts) >= 2: content = parts[1].replace("json", "", 1).strip() parsed = json.loads(content) if isinstance(parsed, list): return parsed except Exception: pass return [] def main() -> None: client = OpenAI(base_url=API_BASE_URL, api_key=API_KEY or "missing-key") env = SolidityGuardEnv() task_list = [TASK_NAME] if TASK_NAME else DEFAULT_TASKS global_step = 0 for task_id in task_list: rewards: List[float] = [] steps_taken = 0 score = 0.0 success = False log_start(task=task_id, env=BENCHMARK, model=MODEL_NAME) try: observation = env.reset(task_id=task_id) for _ in range(MAX_STEPS): global_step += 1 steps_taken += 1 error: Optional[str] = None action_text = "[]" try: prompt = _build_prompt(observation["source_code"], task_id) actions = _call_model(client, prompt) if not actions: actions = _fallback_actions(observation["source_code"], task_id) result = env.step(actions) reward = _safe_score(float(result.get("reward", 0.0))) done = result.get("done", True) action_text = json.dumps( actions, ensure_ascii=True, separators=(",", ":") ) except Exception as exc: error = str(exc) reward = 0.01 done = True rewards.append(reward) log_step( step=global_step, action=action_text, reward=reward, done=done, error=error, ) if done: break score = _safe_score(sum(rewards) / max(len(rewards), 1)) success = score >= SUCCESS_SCORE_THRESHOLD except Exception as exc: if not rewards: rewards = [0.01] steps_taken = 1 score = _safe_score(sum(rewards) / max(len(rewards), 1)) print(f"[DEBUG] Exception during run for {task_id}: {exc}", flush=True) finally: log_end(success=success, steps=steps_taken, score=score, rewards=rewards) if __name__ == "__main__": main()