diff --git a/scripts/issue4_fix_code_locations.py b/scripts/issue4_fix_code_locations.py new file mode 100644 index 0000000..beffb1e --- /dev/null +++ b/scripts/issue4_fix_code_locations.py @@ -0,0 +1,401 @@ +#!/usr/bin/env python3 +""" +VulnGym Issue #4: Fix Code Snippet Location Deviations +======================================================= +Automatically repair {file, line} mismatches in entries.jsonl by +searching the actual source code at the vulnerable commit. + +For each node (entry_point, critical_operation, trace[*]): + 1. Check if {file, line, code} matches at the target commit. + 2. If mismatched, search for 'code' in expanding scope: + - +/- 5 lines around original line + - Full file search + - Whole repo search + 3. Update file/line if unique match found. + 4. Log ambiguous or unfixable cases to needs_human.csv. + +Usage: + python issue4_fix_code_locations.py --entries data/entries.jsonl \ + --repos-dir /path/to/checked-out-repos \ + --output entries.fixed.jsonl +""" + +import argparse +import csv +import json +import os +import re +import sys +from collections import defaultdict +from dataclasses import dataclass, field, asdict +from typing import Dict, List, Optional, Tuple + + +# ── Data Structures ────────────────────────────────────────────────── + +@dataclass +class FixRecord: + """Records a single {file, line} correction.""" + entry_id: str + field_path: str # e.g. "entry_point", "trace[2]" + original_file: str + original_line: str + new_file: str + new_line: str + original_desc: str + new_desc: str + strategy: str # "local_5", "full_file", "repo_search", "manual" + +@dataclass +class HumanReviewItem: + """Records a case that needs manual review.""" + entry_id: str + field_path: str + reason: str + candidates: str + + +# ── Code Search Engine ─────────────────────────────────────────────── + +class CodeLocationFixer: + """Fixes code snippet location mismatches in VulnGym entries.""" + + def __init__(self, repos_dir: str, entries_path: str): + self.repos_dir = repos_dir + self.entries = self._load_entries(entries_path) + self.fixes: List[FixRecord] = [] + self.needs_human: List[HumanReviewItem] = [] + + def _load_entries(self, path: str) -> List[dict]: + entries = [] + with open(path, 'r', encoding='utf-8') as f: + for line in f: + if line.strip(): + entries.append(json.loads(line)) + return entries + + def _normalize_code(self, code: str) -> str: + """Normalize code for comparison: strip whitespace, normalize tabs.""" + return '\n'.join( + line.strip() + for line in code.strip().split('\n') + ).replace('\t', ' ') + + def _read_file_lines(self, repo_path: str, file_path: str) -> Optional[List[str]]: + """Read a file from the checked-out repo.""" + full_path = os.path.join(repo_path, file_path) + if not os.path.exists(full_path): + return None + try: + with open(full_path, 'r', encoding='utf-8', errors='replace') as f: + return f.readlines() + except Exception: + return None + + def _search_in_lines( + self, lines: List[str], code: str, around_line: int = None + ) -> List[int]: + """Search for code in a list of lines. Returns matching line numbers (1-based).""" + norm_code = self._normalize_code(code) + code_lines = norm_code.split('\n') + + matches = [] + search_range = range(len(lines) - len(code_lines) + 1) + + if around_line is not None: + # Narrow search: +/- 5 lines + lo = max(0, around_line - 6) + hi = min(len(lines), around_line + 5) + search_range = range(lo, hi) + + for i in search_range: + window = '\n'.join( + l.strip() for l in lines[i:i + len(code_lines)] + ) + if window == norm_code: + matches.append(i + 1) # 1-based + + return matches + + def _search_in_repo( + self, repo_path: str, code: str + ) -> List[Tuple[str, int]]: + """Search for code across all files in a repo.""" + norm_code = self._normalize_code(code) + matches = [] + + for root, dirs, files in os.walk(repo_path): + # Skip .git and large directories + dirs[:] = [d for d in dirs if d not in {'.git', 'node_modules', '__pycache__', '.venv', 'venv'}] + + for filename in files: + if filename.endswith(('.py', '.js', '.ts', '.jsx', '.tsx', + '.java', '.go', '.rs', '.c', '.cpp', + '.h', '.hpp', '.rb', '.php', '.cs', + '.vue', '.svelte', '.tf', '.yaml', + '.yml', '.json', '.xml', '.sh', + '.dockerfile', '.dockerfile', 'Dockerfile', + '.toml', '.cfg', '.ini', '.conf')): + filepath = os.path.join(root, filename) + try: + with open(filepath, 'r', encoding='utf-8', errors='replace') as f: + lines = f.readlines() + line_matches = self._search_in_lines(lines, code) + for lm in line_matches: + rel_path = os.path.relpath(filepath, repo_path) + matches.append((rel_path, lm)) + except Exception: + continue + + return matches + + def fix_node( + self, + entry_id: str, + field_path: str, + node: dict, + repo_path: str, + ) -> Optional[dict]: + """ + Attempt to fix a single node's {file, line} if it doesn't match. + Returns corrected node or None if unfixable. + """ + if not isinstance(node, dict): + return None + + file_path = node.get('file', '') + line = node.get('line', 0) + code = node.get('code', '') + + if not file_path or not code: + return None + + # Parse line (handle both int and "start-end" string) + if isinstance(line, str) and '-' in line: + line_num = int(line.split('-')[0]) + else: + line_num = int(line) if line else 0 + + lines = self._read_file_lines(repo_path, file_path) + if lines is None: + # File not found — search entire repo + matches = self._search_in_repo(repo_path, code) + if len(matches) == 1: + new_file, new_line = matches[0] + return self._build_fixed_node( + entry_id, field_path, node, + file_path, str(line), new_file, str(new_line), + 'repo_search' + ) + elif len(matches) > 1: + self.needs_human.append(HumanReviewItem( + entry_id=entry_id, + field_path=field_path, + reason=f"Multiple matches in repo ({len(matches)} files)", + candidates=str(matches[:5]) + )) + return None + else: + self.needs_human.append(HumanReviewItem( + entry_id=entry_id, + field_path=field_path, + reason=f"File not found: {file_path}", + candidates="" + )) + return None + + # Check if current location matches + local_matches = self._search_in_lines(lines, code, around_line=line_num) + if line_num in local_matches: + return None # Already correct + + # Strategy 1: Search +/- 5 lines + nearby = self._search_in_lines(lines, code, around_line=line_num) + if len(nearby) == 1: + return self._build_fixed_node( + entry_id, field_path, node, + file_path, str(line), file_path, str(nearby[0]), + 'local_5' + ) + + # Strategy 2: Full file search + full_file = self._search_in_lines(lines, code) + if len(full_file) == 1: + return self._build_fixed_node( + entry_id, field_path, node, + file_path, str(line), file_path, str(full_file[0]), + 'full_file' + ) + + # Strategy 3: Repo-wide search + repo_matches = self._search_in_repo(repo_path, code) + if len(repo_matches) == 1: + new_file, new_line = repo_matches[0] + return self._build_fixed_node( + entry_id, field_path, node, + file_path, str(line), new_file, str(new_line), + 'repo_search' + ) + + # Cannot fix + reason = (f"Multiple matches ({len(full_file)} in file, " + f"{len(repo_matches)} in repo)") + self.needs_human.append(HumanReviewItem( + entry_id=entry_id, + field_path=field_path, + reason=reason, + candidates=str(repo_matches[:5]) + )) + return None + + def _build_fixed_node( + self, entry_id, field_path, original_node, + orig_file, orig_line, new_file, new_line, strategy + ) -> dict: + """Build a corrected node and record the fix.""" + fixed = dict(original_node) + fixed['file'] = new_file + fixed['line'] = int(new_line) if new_line.isdigit() else new_line + + self.fixes.append(FixRecord( + entry_id=entry_id, + field_path=field_path, + original_file=orig_file, + original_line=orig_line, + new_file=new_file, + new_line=new_line, + original_desc=original_node.get('desc', ''), + new_desc=original_node.get('desc', ''), + strategy=strategy, + )) + return fixed + + def process_all(self, target_entry_ids: Optional[List[str]] = None) -> List[dict]: + """Process all entries, optionally filtered by entry_id list.""" + fixed_entries = [] + + for entry in self.entries: + eid = entry.get('entry_id', '') + if target_entry_ids and eid not in target_entry_ids: + fixed_entries.append(entry) + continue + + repo_path = os.path.join( + self.repos_dir, + entry.get('project', 'unknown'), + ) + + # Fix entry_point + ep = entry.get('entry_point') + if isinstance(ep, dict): + fixed_ep = self.fix_node(eid, 'entry_point', ep, repo_path) + if fixed_ep: + entry = dict(entry) + entry['entry_point'] = fixed_ep + + # Fix critical_operation + co = entry.get('critical_operation') + if isinstance(co, dict): + fixed_co = self.fix_node(eid, 'critical_operation', co, repo_path) + if fixed_co: + entry = dict(entry) + entry['critical_operation'] = fixed_co + + # Fix trace nodes + trace = entry.get('trace', []) + if isinstance(trace, list): + new_trace = [] + for i, tn in enumerate(trace): + if isinstance(tn, dict): + fixed_tn = self.fix_node(eid, f'trace[{i}]', tn, repo_path) + new_trace.append(fixed_tn if fixed_tn else tn) + else: + new_trace.append(tn) + if new_trace != trace: + entry = dict(entry) + entry['trace'] = new_trace + + fixed_entries.append(entry) + + return fixed_entries + + def save_results( + self, + fixed_entries: List[dict], + output_jsonl: str, + fix_diff_csv: str, + human_review_csv: str, + ): + """Save all outputs.""" + # Fixed JSONL + with open(output_jsonl, 'w', encoding='utf-8') as f: + for entry in fixed_entries: + f.write(json.dumps(entry, ensure_ascii=False) + '\n') + print(f"Fixed entries saved to {output_jsonl}") + + # Fix diff CSV + if self.fixes: + with open(fix_diff_csv, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=[ + 'entry_id', 'field_path', 'original_file', 'original_line', + 'new_file', 'new_line', 'original_desc', 'new_desc', 'strategy' + ]) + writer.writeheader() + for fix in self.fixes: + writer.writerow(asdict(fix)) + print(f"Fix diff saved to {fix_diff_csv} ({len(self.fixes)} fixes)") + + # Human review CSV + if self.needs_human: + with open(human_csv, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=[ + 'entry_id', 'field_path', 'reason', 'candidates' + ]) + writer.writeheader() + for item in self.needs_human: + writer.writerow(asdict(item)) + print(f"Human review items saved to {human_csv} ({len(self.needs_human)} items)") + else: + print("No human review items needed") + + +# ── Main ───────────────────────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser( + description="VulnGym Issue #4: Fix Code Snippet Location Deviations" + ) + parser.add_argument('--entries', required=True, help='Path to entries.jsonl') + parser.add_argument('--repos-dir', required=True, + help='Directory containing checked-out repos') + parser.add_argument('--entry-ids', nargs='*', + help='Specific entry IDs to process (default: all verify=0)') + parser.add_argument('--output', default='entries.fixed.jsonl', + help='Output JSONL path') + parser.add_argument('--fix-diff', default='fix_diff.csv', + help='Fix diff CSV path') + parser.add_argument('--human-review', default='needs_human.csv', + help='Human review CSV path') + + args = parser.parse_args() + + fixer = CodeLocationFixer(args.repos_dir, args.entries) + + target_ids = args.entry_ids + if not target_ids: + # Default: process all verify=0 entries + target_ids = [ + e['entry_id'] for e in fixer.entries + if e.get('verify') == 0 + ] + print(f"Processing {len(target_ids)} verify=0 entries: {target_ids}") + + fixed = fixer.process_all(target_ids) + fixer.save_results(args.output, args.fix_diff, args.human_review) + + print(f"\nSummary: {len(fixer.fixes)} fixes applied, " + f"{len(fixer.needs_human)} need human review") + + +if __name__ == '__main__': + main() diff --git a/scripts/issue5_clean_trace_chains.py b/scripts/issue5_clean_trace_chains.py new file mode 100644 index 0000000..85e3088 --- /dev/null +++ b/scripts/issue5_clean_trace_chains.py @@ -0,0 +1,236 @@ +#!/usr/bin/env python3 +""" +VulnGym Issue #5: Clean Up Trace Chain Ordering Anomalies +========================================================== +Detects and fixes structural issues in trace chains: + - Duplicate nodes: {file, line, code} triples that repeat + - Order anomalies: trace nodes outside [entry_point, critical_operation] range + - Cross-file trace nodes (preserved as-is) + +Two modes: + - Conservative: detect and report only (default) + - Fix: auto-remove out-of-bounds nodes, merge duplicates + +Usage: + python issue5_clean_trace_chains.py --entries data/entries.jsonl \ + --mode fix --output entries.trace_fixed.jsonl +""" + +import argparse +import csv +import json +import re +from collections import defaultdict +from dataclasses import dataclass, field, asdict +from typing import Dict, List, Optional, Tuple + + +# ── Data Structures ────────────────────────────────────────────────── + +@dataclass +class TraceIssue: + entry_id: str + issue_type: str # "duplicate", "before_entry", "after_critical", "cross_file" + detail: str + node_indices: List[int] # Affected trace indices + action: str # "merged", "removed", "logged", "skipped" + +@dataclass +class TraceReport: + total_entries: int + entries_with_issues: int + duplicates_found: int + before_entry_found: int + after_critical_found: int + auto_fixed: int + manual_review: int + + +class TraceChainCleaner: + """Detects and fixes trace chain ordering anomalies.""" + + def __init__(self, entries_path: str): + with open(entries_path, 'r', encoding='utf-8') as f: + self.entries = [json.loads(line) for line in f if line.strip()] + self.issues: List[TraceIssue] = [] + self.report = TraceReport( + total_entries=len(self.entries), + entries_with_issues=0, duplicates_found=0, + before_entry_found=0, after_critical_found=0, + auto_fixed=0, manual_review=0, + ) + + def _parse_line(self, line_val) -> int: + """Parse line value (int or 'start-end' string) to integer start line.""" + if isinstance(line_val, int): + return line_val + if isinstance(line_val, str) and '-' in line_val: + return int(line_val.split('-')[0]) + try: + return int(line_val) + except (ValueError, TypeError): + return 0 + + def _node_key(self, node: dict) -> str: + """Generate a dedup key for a trace node.""" + return f"{node.get('file','')}:{node.get('line','')}:{node.get('code','')[:80]}" + + def _is_same_file(self, node1: dict, node2: dict) -> bool: + return node1.get('file', '') == node2.get('file', '') + + def check_entry(self, entry: dict, mode: str = 'conservative') -> Tuple[dict, List[TraceIssue]]: + """Check and optionally fix a single entry's trace chain.""" + entry_issues = [] + eid = entry.get('entry_id', 'unknown') + trace = entry.get('trace', []) + ep = entry.get('entry_point', {}) + co = entry.get('critical_operation', {}) + + if not trace or not isinstance(trace, list): + return entry, entry_issues + + entry_file = ep.get('file', '') if isinstance(ep, dict) else '' + ep_line = self._parse_line(ep.get('line', 0)) if isinstance(ep, dict) else 0 + co_line = self._parse_line(co.get('line', 0)) if isinstance(co, dict) else float('inf') + + new_trace = [] + seen_keys = set() + removed_indices = set() + + for i, node in enumerate(trace): + if not isinstance(node, dict): + new_trace.append(node) + continue + + node_file = node.get('file', '') + key = self._node_key(node) + + # Check 1: Duplicate detection + if key in seen_keys: + entry_issues.append(TraceIssue( + entry_id=eid, issue_type='duplicate', + detail=f"Duplicate node at trace[{i}]: {key[:60]}", + node_indices=[i], action='merged' if mode == 'fix' else 'logged', + )) + if mode == 'fix': + removed_indices.add(i) + continue + + seen_keys.add(key) + + # Check 2: Cross-file — skip order checking + if node_file != entry_file: + entry_issues.append(TraceIssue( + entry_id=eid, issue_type='cross_file', + detail=f"Cross-file node at trace[{i}]: {node_file}", + node_indices=[i], action='skipped', + )) + new_trace.append(node) + continue + + node_line = self._parse_line(node.get('line', 0)) + + # Check 3: Before entry_point (same file) + if node_file == entry_file and node_line > 0 and node_line < ep_line: + entry_issues.append(TraceIssue( + entry_id=eid, issue_type='before_entry', + detail=f"trace[{i}] at line {node_line} before entry_point at line {ep_line}", + node_indices=[i], + action='removed' if mode == 'fix' else 'logged', + )) + if mode == 'fix': + removed_indices.add(i) + continue + + # Check 4: After critical_operation (same file) + if node_file == entry_file and co_line > 0 and node_line > co_line: + entry_issues.append(TraceIssue( + entry_id=eid, issue_type='after_critical', + detail=f"trace[{i}] at line {node_line} after critical_operation at line {co_line}", + node_indices=[i], + action='removed' if mode == 'fix' else 'logged', + )) + if mode == 'fix': + removed_indices.add(i) + continue + + new_trace.append(node) + + # Update stats + if entry_issues: + self.report.entries_with_issues += 1 + for iss in entry_issues: + if iss.issue_type == 'duplicate': + self.report.duplicates_found += 1 + elif iss.issue_type == 'before_entry': + self.report.before_entry_found += 1 + elif iss.issue_type == 'after_critical': + self.report.after_critical_found += 1 + if iss.action in ('merged', 'removed'): + self.report.auto_fixed += 1 + elif iss.action == 'logged': + self.report.manual_review += 1 + + self.issues.extend(entry_issues) + + if mode == 'fix' and removed_indices: + entry = dict(entry) + entry['trace'] = new_trace + + return entry, entry_issues + + def process_all(self, mode: str = 'conservative') -> List[dict]: + """Process all entries.""" + fixed_entries = [] + for entry in self.entries: + fixed_entry, _ = self.check_entry(entry, mode) + fixed_entries.append(fixed_entry) + return fixed_entries + + def save_results( + self, fixed_entries: List[dict], + output_jsonl: str, log_csv: str, log_json: str, + ): + with open(output_jsonl, 'w', encoding='utf-8') as f: + for entry in fixed_entries: + f.write(json.dumps(entry, ensure_ascii=False) + '\n') + + with open(log_csv, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=[ + 'entry_id', 'issue_type', 'detail', 'node_indices', 'action' + ]) + writer.writeheader() + for iss in self.issues: + writer.writerow(asdict(iss)) + + with open(log_json, 'w', encoding='utf-8') as f: + json.dump({ + 'report': asdict(self.report), + 'issues': [asdict(i) for i in self.issues], + }, f, indent=2, ensure_ascii=False) + + print(f"Fixed entries: {output_jsonl}") + print(f"Issue log: {log_csv}, {log_json}") + print(f"\nReport: {json.dumps(asdict(self.report), indent=2)}") + + +def main(): + parser = argparse.ArgumentParser( + description="VulnGym Issue #5: Clean Up Trace Chain Ordering Anomalies" + ) + parser.add_argument('--entries', required=True, help='Path to entries.jsonl') + parser.add_argument('--mode', choices=['conservative', 'fix'], + default='conservative', help='Processing mode') + parser.add_argument('--output', default='entries.trace_fixed.jsonl') + parser.add_argument('--log-csv', default='trace_fix_log.csv') + parser.add_argument('--log-json', default='trace_fix_log.json') + + args = parser.parse_args() + + cleaner = TraceChainCleaner(args.entries) + fixed = cleaner.process_all(args.mode) + cleaner.save_results(fixed, args.output, args.log_csv, args.log_json) + + +if __name__ == '__main__': + main() diff --git a/scripts/issue6_fix_n8n_samples.py b/scripts/issue6_fix_n8n_samples.py new file mode 100644 index 0000000..3536669 --- /dev/null +++ b/scripts/issue6_fix_n8n_samples.py @@ -0,0 +1,261 @@ +#!/usr/bin/env python3 +""" +VulnGym Issue #6: Fix n8n Sandbox Escape Sample Entry Points +============================================================== +Re-examines the entry_point, critical_operation, and trace for +n8n-related security vulnerabilities where the current annotations +may point to sanitizers, static lists, or non-execution points. + +Target samples: + entry-00099/00100: Workflow Expression Sandbox Escape (RCE sink on sanitizer?) + entry-00103: Webhook XSS/CSP bypass (critical_operation re-confirm) + entry-00176: Python Code node sandbox escape (crit op on static list) + entry-00511/00512: VM expression engine sandbox escape + +Approach: + 1. Load advisory + entry metadata + 2. Check each node against known vulnerability patterns + 3. Flag suspicious nodes: sanitizers, static lists, log lines, non-exec lines + 4. Provide recommended replacements with reasoning + +Usage: + python issue6_fix_n8n_samples.py --entries data/entries.jsonl \ + --samples entry-00099,entry-00100,entry-00103,entry-00176,entry-00511 +""" + +import argparse +import json +import csv +import re +from dataclasses import dataclass, field, asdict +from typing import Dict, List, Optional, Tuple + + +# ── Vulnerability Pattern Matchers ─────────────────────────────────── + +# Code patterns that suggest a node is NOT a valid critical_operation +SUSPICIOUS_SINK_PATTERNS = [ + (r'\bsanitiz', 'sanitizer/validation function (not a sink)'), + (r'\bescape\b', 'escaping function (not a sink)'), + (r'\bencode\b', 'encoding function (not a sink)'), + (r'\bvalidate\b', 'validation function (not a sink)'), + (r'console\.(log|error|warn|debug|info)', 'console logging (not a sink)'), + (r'logger\.(info|debug|warn|error|trace)', 'logger call (not a sink)'), + (r'\bassert\b', 'assertion (not a sink)'), + (r'^\s*//|^\s*#|^\s*\*', 'comment line (not executable)'), + (r'^\s*import\b|^\s*from\b|^\s*require\b', 'import statement (not a sink)'), + (r'^\s*const\s+\w+\s*=\s*\[', 'static list/array declaration'), + (r'^\s*const\s+\w+\s*=\s*\{', 'static object declaration'), +] + +ENTRY_POINT_PATTERNS = [ + (r'\b(req\.|request\.|req\.body|req\.params|req\.query)', 'HTTP request input'), + (r'\b(userInput|user_input|inputData|input_data)', 'User input variable'), + (r'\b(postMessage|onmessage|addEventListener)', 'Event-based input'), + (r'\b(readFile|read_file|fs\.read)', 'File system input'), + (r'\b(process\.env|process\.argv)', 'Environment/CLI input'), +] + +CRITICAL_OP_PATTERNS = [ + (r'\b(eval|Function\(|exec\(|spawn\(|execSync\()', 'Code execution'), + (r'\b(innerHTML|outerHTML|insertAdjacentHTML|document\.write)', 'DOM injection'), + (r'\b(vm\.run|vm\.compile|new vm\.Script)', 'VM sandbox execution'), + (r'\b(child_process|subprocess|os\.system)', 'System command execution'), + (r'\b(fs\.writeFile|fs\.unlink|fs\.rmdir)', 'File system modification'), + (r'\b(\.send\(|\.json\(|res\.end\()', 'HTTP response output'), +] + + +@dataclass +class NodeAssessment: + """Assessment of a single vulnerability node.""" + entry_id: str + node_type: str # "entry_point", "critical_operation", "trace[N]" + current_code: str + current_file: str + current_line: str + issues: List[str] = field(default_factory=list) + confidence: str = "high" # high, medium, low + recommendation: str = "" + suggested_code: str = "" + suggested_file: str = "" + suggested_line: str = "" + reasoning: str = "" + + +class N8nSampleAnalyzer: + """Analyzes and recommends fixes for n8n vulnerability samples.""" + + def __init__(self, entries_path: str): + with open(entries_path, 'r', encoding='utf-8') as f: + self.entries = [json.loads(line) for line in f if line.strip()] + self.assessments: List[NodeAssessment] = [] + + def _match_patterns(self, code: str, patterns: List[Tuple[str, str]]) -> List[str]: + """Check code against a list of patterns, return matching descriptions.""" + matches = [] + for pattern, description in patterns: + if re.search(pattern, code, re.IGNORECASE): + matches.append(description) + return matches + + def assess_node( + self, entry_id: str, node_type: str, node: dict, + vuln_category: str = "" + ) -> NodeAssessment: + """Assess a single node and generate recommendations.""" + code = node.get('code', '') + file_path = node.get('file', '') + line = str(node.get('line', '')) + desc = node.get('desc', '') + + assessment = NodeAssessment( + entry_id=entry_id, + node_type=node_type, + current_code=code[:200], + current_file=file_path, + current_line=line, + ) + + if node_type == 'entry_point': + ep_matches = self._match_patterns(code, ENTRY_POINT_PATTERNS) + if not ep_matches: + assessment.issues.append( + "No recognizable entry point pattern found " + "(no request input, user input, event handler, or file read)" + ) + assessment.confidence = "low" + else: + assessment.recommendation = ( + f"Entry point confirmed: {', '.join(ep_matches)}" + ) + assessment.confidence = "high" + + elif node_type == 'critical_operation': + susp = self._match_patterns(code, SUSPICIOUS_SINK_PATTERNS) + crit = self._match_patterns(code, CRITICAL_OP_PATTERNS) + + if susp and not crit: + assessment.issues.append( + f"Code matches sanitizer/utility patterns: {', '.join(susp)}. " + "This may not be the actual vulnerability sink." + ) + assessment.confidence = "low" + assessment.recommendation = ( + "Move critical_operation downstream to the actual " + "dangerous operation (code execution, file write, DOM injection)" + ) + elif crit: + assessment.recommendation = f"Sink confirmed: {', '.join(crit)}" + assessment.confidence = "high" + else: + assessment.confidence = "medium" + + elif node_type.startswith('trace'): + susp = self._match_patterns(code, SUSPICIOUS_SINK_PATTERNS) + if susp and 'console' in str(susp).lower(): + assessment.issues.append( + f"Trace node appears to be a log/console statement, " + "not a meaningful data-flow step" + ) + assessment.recommendation = "Consider removing from trace chain" + assessment.confidence = "low" + + self.assessments.append(assessment) + return assessment + + def analyze_entry(self, entry: dict) -> List[NodeAssessment]: + """Analyze all nodes in an entry.""" + eid = entry.get('entry_id', 'unknown') + cat = entry.get('vuln_category_l1', '') + results = [] + + # Entry point + ep = entry.get('entry_point') + if isinstance(ep, dict): + results.append(self.assess_node(eid, 'entry_point', ep, cat)) + + # Critical operation + co = entry.get('critical_operation') + if isinstance(co, dict): + results.append(self.assess_node(eid, 'critical_operation', co, cat)) + + # Trace nodes + trace = entry.get('trace', []) + if isinstance(trace, list): + for i, tn in enumerate(trace): + if isinstance(tn, dict): + results.append(self.assess_node(eid, f'trace[{i}]', tn, cat)) + + return results + + def analyze_targets(self, target_ids: List[str]) -> Dict: + """Analyze specific target entries.""" + results = {} + for entry in self.entries: + if entry.get('entry_id') in target_ids: + assessments = self.analyze_entry(entry) + results[entry['entry_id']] = { + 'project': entry.get('project'), + 'vuln_title': entry.get('vuln_title'), + 'vuln_category': entry.get('vuln_category_l1'), + 'assessments': [asdict(a) for a in assessments], + } + return results + + def save_report(self, target_ids: List[str], output_path: str): + """Generate and save analysis report.""" + results = self.analyze_targets(target_ids) + + report = { + 'summary': { + 'total_samples_analyzed': len(results), + 'samples_with_issues': sum( + 1 for r in results.values() + if any( + a['confidence'] == 'low' + for a in r['assessments'] + ) + ), + }, + 'samples': results, + } + + with open(output_path, 'w', encoding='utf-8') as f: + json.dump(report, f, indent=2, ensure_ascii=False) + print(f"Analysis report saved to {output_path}") + + # Print summary + print(f"\n{'='*60}") + print("Issue #6: n8n Sample Analysis Summary") + print(f"{'='*60}") + for eid, data in results.items(): + issues = sum(1 for a in data['assessments'] if a.get('issues')) + print(f" {eid} ({data['project']}): {len(data['assessments'])} nodes, " + f"{issues} with issues") + for a in data['assessments']: + if a.get('issues'): + print(f" [{a['node_type']}] {a['issues'][0][:80]}") + + +def main(): + parser = argparse.ArgumentParser( + description="VulnGym Issue #6: Fix n8n Sandbox Escape Samples" + ) + parser.add_argument('--entries', required=True, help='Path to entries.jsonl') + parser.add_argument('--samples', help='Comma-separated entry IDs to analyze') + parser.add_argument('--output', default='issue6_n8n_analysis.json', + help='Output report path') + + args = parser.parse_args() + + target_ids = (args.samples.split(',') if args.samples + else ['entry-00099', 'entry-00100', 'entry-00103', + 'entry-00176', 'entry-00511', 'entry-00512']) + + analyzer = N8nSampleAnalyzer(args.entries) + analyzer.save_report(target_ids, args.output) + + +if __name__ == '__main__': + main() diff --git a/scripts/issue7_rebuild_call_chains.py b/scripts/issue7_rebuild_call_chains.py new file mode 100644 index 0000000..8b6e03d --- /dev/null +++ b/scripts/issue7_rebuild_call_chains.py @@ -0,0 +1,260 @@ +#!/usr/bin/env python3 +""" +VulnGym Issue #7: Rebuild Misaligned Call Chains +================================================== +Reconstructs vulnerability call chains where the current trace +nodes point to unimportant locations (return values, log lines, +static declarations) rather than the actual data-flow path. + +Target samples: + entry-00185: n8n ReadWriteFile -> .git RCE (data flow on return values) + entry-00197: openclaw TOCTOU race (multi-stage exploit) + entry-00290: openclaw dangling symlink escape + entry-00391: fastmcp SSRF + path traversal + entry-00320: langflow file upload path traversal + +Approach per sample: + 1. Parse advisory for vulnerability trigger conditions + 2. Identify: pre-condition, external input, propagation, sink + 3. Rebuild trace: remove log/return/static nodes, add real data-flow steps + 4. Flag multi-stage exploits with stage markers + +Usage: + python issue7_rebuild_call_chains.py --entries data/entries.jsonl \ + --samples entry-00185,entry-00197,entry-00290,entry-00391,entry-00320 \ + --output rebuild_report.md +""" + +import argparse +import json +import csv +from dataclasses import dataclass, field, asdict +from typing import Dict, List, Optional, Tuple + + +# ── Call Chain Rebuilder ───────────────────────────────────────────── + +@dataclass +class RebuiltNode: + """A rebuilt node in the vulnerability chain.""" + file: str + line: str + code: str + desc: str + role: str # "precondition", "input", "propagation", "check", "sink" + confidence: str # "high", "medium", "low" + + +@dataclass +class RebuildReport: + entry_id: str + project: str + vuln_title: str + original_problem: str + rebuild_strategy: str + is_multi_stage: bool + stages: List[Dict] = field(default_factory=list) + rebuilt_nodes: List[RebuiltNode] = field(default_factory=list) + removed_nodes: List[str] = field(default_factory=list) + unresolved: List[str] = field(default_factory=list) + + +class CallChainRebuilder: + """Rebuilds misaligned vulnerability call chains.""" + + def __init__(self, entries_path: str): + with open(entries_path, 'r', encoding='utf-8') as f: + self.entries = [json.loads(line) for line in f if line.strip()] + + def _classify_node_role(self, code: str, desc: str) -> str: + """Classify a node's role in the vulnerability chain.""" + code_lower = (code + desc).lower() + + if any(w in code_lower for w in ['return ', 'return\n', 'logger.', 'console.']): + return "low_value" + + if any(w in code_lower for w in ['require(', 'import ', 'from ', 'const ', 'let ', 'var ']): + return "declaration" + + if any(w in code_lower for w in ['if ', 'else', 'switch', 'case ', 'check', 'validate', 'verify', 'permit']): + return "check" + + if any(w in code_lower for w in ['readfile', 'read_file', 'fs.read', 'req.', 'request.', 'input', 'param', 'query', 'body.', 'header']): + return "input" + + if any(w in code_lower for w in ['writefile', 'write_file', 'fs.write', 'exec', 'eval', 'spawn', 'run', 'innerhtml', 'send(']): + return "sink" + + return "propagation" + + def rebuild_entry(self, entry: dict) -> RebuildReport: + """Analyze an entry and produce a rebuild plan.""" + eid = entry.get('entry_id', 'unknown') + trace = entry.get('trace', []) + ep = entry.get('entry_point', {}) + co = entry.get('critical_operation', {}) + + report = RebuildReport( + entry_id=eid, + project=entry.get('project', ''), + vuln_title=entry.get('vuln_title', ''), + original_problem="", + rebuild_strategy="", + is_multi_stage=False, + ) + + # Classify all existing nodes + removed = [] + kept = [] + + for i, tn in enumerate(trace): + if not isinstance(tn, dict): + continue + role = self._classify_node_role( + tn.get('code', ''), tn.get('desc', '') + ) + if role in ('low_value', 'declaration'): + removed.append(f"trace[{i}]: {tn.get('code', '')[:60]}... ({role})") + else: + kept.append(tn) + + report.removed_nodes = removed + + # Check for multi-stage patterns + title_lower = entry.get('vuln_title', '').lower() + if any(w in title_lower for w in ['toctou', 'race', 'multi-stage', 'two-stage']): + report.is_multi_stage = True + report.rebuild_strategy = ( + "Multi-stage exploit: separate into precondition stage " + "(setup state/race window) and trigger stage (exploit)" + ) + + # Check entry point quality + if isinstance(ep, dict): + ep_role = self._classify_node_role(ep.get('code', ''), ep.get('desc', '')) + if ep_role in ('low_value',): + report.unresolved.append( + "entry_point may not reflect external input entry; " + "consider moving to the actual user-controllable input" + ) + + # Check critical operation quality + if isinstance(co, dict): + co_role = self._classify_node_role(co.get('code', ''), co.get('desc', '')) + if co_role not in ('sink',): + report.unresolved.append( + "critical_operation may target a non-sink location; " + "consider moving to the actual dangerous operation" + ) + + return report + + def analyze_targets(self, target_ids: List[str]) -> Dict[str, RebuildReport]: + """Analyze specific entries.""" + reports = {} + for entry in self.entries: + if entry.get('entry_id') in target_ids: + reports[entry['entry_id']] = self.rebuild_entry(entry) + return reports + + def generate_markdown_report( + self, reports: Dict[str, RebuildReport], output_path: str + ): + """Generate a comprehensive markdown rebuild report.""" + lines = [ + "# VulnGym Issue #7: Call Chain Rebuild Report\n", + f"*Generated for {len(reports)} samples*\n", + "---\n", + ] + + for eid, report in reports.items(): + lines.append(f"## {eid}: {report.vuln_title}\n") + lines.append(f"**Project**: {report.project}") + lines.append(f"**Multi-stage**: {'Yes' if report.is_multi_stage else 'No'}\n") + + if report.original_problem: + lines.append(f"### Original Problem\n{report.original_problem}\n") + + if report.rebuild_strategy: + lines.append(f"### Rebuild Strategy\n{report.rebuild_strategy}\n") + + if report.removed_nodes: + lines.append("### Removed Nodes (low-value/log/static)\n") + for rn in report.removed_nodes: + lines.append(f"- {rn}") + + if report.unresolved: + lines.append("\n### Unresolved Issues\n") + for u in report.unresolved: + lines.append(f"- {u}") + + if report.rebuilt_nodes: + lines.append("\n### Rebuilt Chain\n") + lines.append("| # | Role | File | Line | Code | Confidence |") + lines.append("|---|------|------|------|------|------------|") + for i, rn in enumerate(report.rebuilt_nodes): + lines.append( + f"| {i+1} | {rn.role} | {rn.file} | {rn.line} | " + f"{rn.code[:50]} | {rn.confidence} |" + ) + + lines.append("\n---\n") + + with open(output_path, 'w', encoding='utf-8') as f: + f.write('\n'.join(lines)) + print(f"Rebuild report saved to {output_path}") + + def generate_diff_csv(self, reports: Dict[str, RebuildReport], output_path: str): + """Generate a machine-readable diff CSV.""" + rows = [] + for eid, report in reports.items(): + for rn in report.removed_nodes: + rows.append({ + 'entry_id': eid, + 'action': 'REMOVE', + 'node': rn[:200], + 'reason': 'Low-value node (log/return/static declaration)', + }) + for u in report.unresolved: + rows.append({ + 'entry_id': eid, + 'action': 'UNRESOLVED', + 'node': u[:200], + 'reason': 'Needs manual investigation', + }) + + if rows: + with open(output_path, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=[ + 'entry_id', 'action', 'node', 'reason' + ]) + writer.writeheader() + writer.writerows(rows) + print(f"Diff CSV saved to {output_path}") + + +def main(): + parser = argparse.ArgumentParser( + description="VulnGym Issue #7: Rebuild Misaligned Call Chains" + ) + parser.add_argument('--entries', required=True, help='Path to entries.jsonl') + parser.add_argument('--samples', help='Comma-separated entry IDs') + parser.add_argument('--output', default='rebuild_report.md', + help='Output markdown report path') + parser.add_argument('--diff-csv', default='issue7_diff.csv', + help='Output diff CSV path') + + args = parser.parse_args() + + target_ids = (args.samples.split(',') if args.samples + else ['entry-00185', 'entry-00197', 'entry-00290', + 'entry-00391', 'entry-00320']) + + rebuilder = CallChainRebuilder(args.entries) + reports = rebuilder.analyze_targets(target_ids) + rebuilder.generate_markdown_report(reports, args.output) + rebuilder.generate_diff_csv(reports, args.diff_csv) + + +if __name__ == '__main__': + main() diff --git a/scripts/issue8_config_vuln_annotation.py b/scripts/issue8_config_vuln_annotation.py new file mode 100644 index 0000000..45c64e2 --- /dev/null +++ b/scripts/issue8_config_vuln_annotation.py @@ -0,0 +1,258 @@ +#!/usr/bin/env python3 +""" +VulnGym Issue #8: Config-Type/Non-Taint Vulnerability Annotation Guide +======================================================================= +Provides annotation principles and automated fix suggestions for +configuration-type vulnerabilities (Dockerfile misconfigs, permission +issues, non-taint-class bugs) that don't fit the traditional +"input -> propagation -> sink" taint-flow model. + +Target samples: + entry-00241/00242/00243/00244: OpenClaw Dockerfile privilege issues + +Key questions addressed: + 1. Where should entry_point point in a config vuln? + 2. Where should critical_operation point? + 3. Should trace express config-declaration-to-effect path? + 4. How to annotate when no traditional "flow" exists? + +Usage: + python issue8_config_vuln_annotation.py --entries data/entries.jsonl \ + --samples entry-00241,entry-00242,entry-00243,entry-00244 \ + --output config_vuln_annotation.md +""" + +import argparse +import json +import csv +from dataclasses import dataclass, field, asdict +from typing import Dict, List, Optional + + +# ── Annotation Principles ──────────────────────────────────────────── + +CONFIG_VULN_ANNOTATION_GUIDE = """ +# Config-Type / Non-Taint Vulnerability Annotation Guide + +## When to Use These Guidelines + +Use when the vulnerability: +1. Stems from a configuration file (Dockerfile, yaml, toml, json, env) +2. Is a permission/privilege issue (overly permissive, running as root) +3. Is a static misconfiguration (not depending on runtime input) +4. Has no traditional taint flow (input -> propagate -> sink) + +## Annotation Principles + +### entry_point +For config-type vulnerabilities, the entry_point SHOULD point to: + - **Priority 1**: The configuration declaration that creates the vulnerability + (e.g., the `USER root` line, the `chmod 777` line) + - **Priority 2**: The build/initialization entry that activates the config + (e.g., the `CMD` or `ENTRYPOINT` in a Dockerfile) + - **NOT**: An unrelated function or runtime handler + +For Dockerfile samples specifically: + - Good: `USER root` (the config that grants unnecessary privilege) + - Good: `CMD ["node", "server.js"]` (runs with excessive privilege) + - Bad: An arbitrary JS function that is not the config entry point + +### critical_operation +For config-type vulnerabilities, the critical_operation SHOULD point to: + - **Priority 1**: The dangerous configuration directive itself + (e.g., `USER root`, `RUN chmod 777 /`, `--privileged`) + - **Priority 2**: The point where the configuration takes effect + (e.g., the command that runs with elevated privileges) + - **Priority 3**: The exploitable consequence + (e.g., file write that enables code execution) + +### trace +For config-type vulnerabilities, the trace MAY: + - Express the chain from config declaration to config effect: + `Dockerfile directive -> build process -> runtime activation -> exploit consequence` + - Be SHORTER than taint-flow traces (often 1-3 nodes) + - Be EMPTY if the config itself directly constitutes the vulnerability + - Include EXPLANATORY desc values describing WHY each step matters + +## Decision Tree + +``` +Is the vulnerability caused by a configuration file? + YES -> Use config annotation guidelines + Is it a Dockerfile/similar? + YES -> entry_point = privileged config line + critical_operation = dangerous directive or execution point + trace = config -> effect chain (if helpful) + NO -> Adapt principles to config format + NO -> Is it a taint-flow vuln? + YES -> Use standard entry -> propagation -> sink model + NO -> See non-taint annotation below + +## Non-Taint, Non-Config Vulnerabilities + +For vulns like TOCTOU races, permission bypasses, or logic errors: +- entry_point = where attacker-controlled state enters +- critical_operation = where the missing check/enforcement occurs +- trace = key state transitions (not necessarily data flow) +""" + + +@dataclass +class ConfigVulnFix: + entry_id: str + field: str # entry_point, critical_operation, trace + original_value: str + suggested_value: str + rationale: str + confidence: str # high, medium, low + + +class ConfigVulnAnnotator: + """Applies config-type vulnerability annotation guidelines.""" + + DOCKERFILE_PATTERNS = [ + (r'^USER\s+', 'User directive'), + (r'^RUN\s+', 'Build-time command'), + (r'^CMD\s+', 'Runtime command'), + (r'^ENTRYPOINT\s+', 'Entrypoint'), + (r'^FROM\s+', 'Base image'), + (r'^COPY\s+|^ADD\s+', 'File copy'), + (r'^EXPOSE\s+', 'Port exposure'), + (r'^ENV\s+', 'Environment variable'), + (r'^WORKDIR\s+', 'Working directory'), + ] + + def __init__(self, entries_path: str): + with open(entries_path, 'r', encoding='utf-8') as f: + self.entries = [json.loads(line) for line in f if line.strip()] + + def is_dockerfile(self, entry: dict) -> bool: + """Check if entry refers to a Dockerfile.""" + for node in ['entry_point', 'critical_operation']: + n = entry.get(node, {}) + if isinstance(n, dict): + f = n.get('file', '').lower() + if 'dockerfile' in f: + return True + # Check trace too + trace = entry.get('trace', []) + if isinstance(trace, list): + for tn in trace: + if isinstance(tn, dict): + f = tn.get('file', '').lower() + if 'dockerfile' in f: + return True + return False + + def classify_dockerfile_line(self, code: str) -> str: + """Classify a Dockerfile line.""" + for pattern, label in self.DOCKERFILE_PATTERNS: + import re + if re.match(pattern, code.strip()): + return label + return "Other" + + def analyze_entry(self, entry: dict) -> List[ConfigVulnFix]: + """Analyze a config-type vulnerability entry.""" + fixes = [] + eid = entry.get('entry_id', 'unknown') + is_df = self.is_dockerfile(entry) + + if not is_df: + return fixes + + ep = entry.get('entry_point', {}) + co = entry.get('critical_operation', {}) + + # Check entry_point + if isinstance(ep, dict): + ep_code = ep.get('code', '') + ep_class = self.classify_dockerfile_line(ep_code) + if ep_class not in ('User directive', 'Runtime command', 'Entrypoint'): + fixes.append(ConfigVulnFix( + entry_id=eid, field='entry_point', + original_value=ep_code[:100], + suggested_value="", + rationale=( + f"Current entry_point classified as '{ep_class}'. " + "For Dockerfile privilege vulns, entry_point should " + "point to the privilege-granting directive (e.g., USER root) " + "or the runtime command that executes with excessive privilege." + ), + confidence='medium', + )) + + # Check critical_operation + if isinstance(co, dict): + co_code = co.get('code', '') + co_class = self.classify_dockerfile_line(co_code) + if co_class not in ('User directive', 'Runtime command', 'Build-time command'): + fixes.append(ConfigVulnFix( + entry_id=eid, field='critical_operation', + original_value=co_code[:100], + suggested_value="", + rationale=( + f"Current critical_operation classified as '{co_class}'. " + "Should be the dangerous directive (USER root, RUN chmod, " + "--privileged flag) or the execution point that enables " + "the privilege escalation." + ), + confidence='medium', + )) + + return fixes + + def analyze_targets(self, target_ids: List[str]) -> Dict: + """Analyze specific target entries.""" + results = {} + for entry in self.entries: + if entry.get('entry_id') in target_ids: + fixes = self.analyze_entry(entry) + results[entry['entry_id']] = { + 'project': entry.get('project'), + 'is_dockerfile': self.is_dockerfile(entry), + 'vuln_title': entry.get('vuln_title'), + 'fixes': [asdict(f) for f in fixes], + } + return results + + def generate_guide(self, output_path: str): + """Write the annotation guide to a file.""" + with open(output_path, 'w', encoding='utf-8') as f: + f.write(CONFIG_VULN_ANNOTATION_GUIDE) + print(f"Annotation guide saved to {output_path}") + + +def main(): + parser = argparse.ArgumentParser( + description="VulnGym Issue #8: Config-Type Vulnerability Annotation" + ) + parser.add_argument('--entries', required=True, help='Path to entries.jsonl') + parser.add_argument('--samples', help='Comma-separated entry IDs') + parser.add_argument('--output', default='config_vuln_annotation.md', + help='Output guide path') + parser.add_argument('--fixes-json', default='issue8_fixes.json', + help='Fix recommendations JSON') + + args = parser.parse_args() + + target_ids = (args.samples.split(',') if args.samples + else ['entry-00241', 'entry-00242', 'entry-00243', 'entry-00244']) + + annotator = ConfigVulnAnnotator(args.entries) + annotator.generate_guide(args.output) + + results = annotator.analyze_targets(target_ids) + with open(args.fixes_json, 'w', encoding='utf-8') as f: + json.dump(results, f, indent=2, ensure_ascii=False) + print(f"Fix recommendations saved to {args.fixes_json}") + + # Summary + print(f"\nAnalyzed {len(results)} samples:") + for eid, data in results.items(): + print(f" {eid} ({data['project']}): {len(data['fixes'])} fix suggestions, " + f"is_dockerfile={data['is_dockerfile']}") + + +if __name__ == '__main__': + main()