diff --git a/scripts/README.md b/scripts/README.md new file mode 100644 index 0000000..145c3ff --- /dev/null +++ b/scripts/README.md @@ -0,0 +1,36 @@ +# Scripts + +This directory contains utility scripts for analyzing and processing SHERPA results. + +## Available Scripts + +### `fuzzlog_analyzer.py` + +Analyzes libFuzzer/OSS-Fuzz logs to extract key information and surface actionable findings. + +**Features:** +- Tracks fuzzer execution stats (runtime, exec/s, corpus growth) +- Detects crashes, timeouts, OOMs, and sanitizer findings +- Identifies artifact locations and interesting coverage discoveries +- Provides both human-readable and JSON output formats + +**Usage:** +```bash +python3 fuzzlog_analyzer.py [options] +``` + +**Options:** +- `--json`: Output results in JSON format instead of human-readable text +- `--max-funcs N`: Maximum NEW_FUNC lines to display per file (default: 10) + +**Examples:** +```bash +# Analyze a single log file +python3 fuzzlog_analyzer.py fuzzer_run.txt + +# Analyze multiple log files with JSON output +python3 fuzzlog_analyzer.py *.txt --json + +# Limit NEW_FUNC output +python3 fuzzlog_analyzer.py fuzzer_run.txt --max-funcs 5 +``` \ No newline at end of file diff --git a/scripts/fuzzlog_analyzer.py b/scripts/fuzzlog_analyzer.py new file mode 100644 index 0000000..71cc0b0 --- /dev/null +++ b/scripts/fuzzlog_analyzer.py @@ -0,0 +1,345 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +fuzzlog_analyzer.py +Analyze libFuzzer / OSS-Fuzz style logs and report: +- Did the fuzzer run? For how long? Execs, exec/s, corpus growth +- Crashes / timeouts / OOMs / sanitizer findings +- Artifact locations (crash-, timeout-, leaks, "Test unit written to", artifact_prefix) +- Interesting items (NEW_FUNC lines, UBSan/ASan summaries, warnings) +- Pointers for next steps + +Works best with libFuzzer logs (standalone or inside OSS-Fuzz base-runner output), +but is defensive against variations. +""" + +import argparse +import json +import re +from pathlib import Path +from typing import List, Dict, Any, Tuple + +# --- Regex library (broad but precise enough) --- + +RE_STARTED = re.compile(r'\bINFO:\s*seed corpus\b|\bstat::number_of_executed_units\b|\bRunning with\s+seed\b', re.IGNORECASE) +RE_EMPTY_CORPUS = re.compile(r'A corpus is not provided, starting from an empty corpus', re.IGNORECASE) + +# Progress & stats (multiple shapes) +RE_DONE = re.compile(r'\bDone\b.*?\bin\s+(\d+)\s*(?:s|second)', re.IGNORECASE) # Done 223k in 600s +RE_EXECS = re.compile(r'(?:stat::number_of_executed_units|executed\s+units)\D+(\d+)', re.IGNORECASE) +RE_EXECS_PER_SEC = re.compile(r'(?:exec/s|executions per second)\D+([0-9]+(?:\.[0-9]+)?)', re.IGNORECASE) +RE_NEW_UNITS = re.compile(r'(?:stat::new_units_added|NEW units added|added\s+(\d+)\s+new units)|\bNEW:\s*(\d+)', re.IGNORECASE) +RE_PEAK_RSS = re.compile(r'(?:stat::peak_rss_mb|peak rss|peak\s+RSS)\D+([0-9]+)', re.IGNORECASE) +RE_SLOWEST = re.compile(r'(?:slowest unit time|slowest unit)\D+([0-9]+(?:\.[0-9]+)?)\s*(?:s|sec|seconds)?', re.IGNORECASE) + +# NEW_FUNC discoveries (libFuzzer’s -print_pcs / coverage discovery logs) +RE_NEW_FUNC = re.compile(r'\bNEW_FUNC\b.*', re.IGNORECASE) + +# Crashes / sanitizers / timeouts / OOM +RE_ERROR_HDR = re.compile(r'==\d+==\s*ERROR:\s*([A-Za-z]+Sanitizer):\s*(.*)', re.IGNORECASE) +RE_SAN_SUMMARY = re.compile(r'\bSUMMARY:\s*([A-Za-z]+Sanitizer):\s*(.*)', re.IGNORECASE) +RE_TIMEOUT = re.compile(r'\bERROR:\s*libFuzzer:\s*timeout|\btimeout after\b', re.IGNORECASE) +RE_OOM = re.compile(r'\bout[-\s]?of[-\s]?memory\b|\bOOM\b', re.IGNORECASE) +RE_DEADLY_SIGNAL = re.compile(r'\bdeadly signal\b|\bSignal\s*\d+\b', re.IGNORECASE) + +# Artifacts and prefixes +RE_ARTIFACT_PREFIX = re.compile(r"artifact_prefix(?:=|:)\s*'?([^\s']+)'?", re.IGNORECASE) +RE_TEST_UNIT_WRITTEN = re.compile(r'\bTest unit written to\b\s*(.*)', re.IGNORECASE) +RE_SAVED_ARTIFACT = re.compile(r'\b(?:crash|timeout|leak|oom)-[0-9a-fA-F]+', re.IGNORECASE) + +# Hints about corpus +RE_CORPUS_DIR = re.compile(r'\bcorpus\b.*?(?:→|->|:)\s*(\S+)|\bseed corpus:\s*(\S+)', re.IGNORECASE) + +# Warnings / oddities worth surfacing +RE_WARN = re.compile(r'\bWARNING:.*|==\d+==\s*WARNING:.*', re.IGNORECASE) + +# Generic k-v stat lines (libFuzzer final stats often look like 'stat::name: value') +RE_STAT = re.compile(r'\bstat::([a-zA-Z0-9_]+)\s*:\s*([^\s].*)') + + +def parse_log(text: str) -> Dict[str, Any]: + lines = text.splitlines() + findings: Dict[str, Any] = { + "ran": False, + "started_from_empty_corpus": False, + "executions": None, + "execs_per_sec": None, + "duration_s": None, + "new_units": None, + "peak_rss_mb": None, + "slowest_unit_s": None, + "new_funcs": [], + "warnings": [], + "errors": [], + "timeouts": [], + "ooms": [], + "signals": [], + "sanitizer_summaries": [], + "artifacts": [], + "artifact_prefix": None, + "corpus_dir_hints": [], + "raw_stats": {}, # any stat::foo: bar captured + "interesting": [], # synthesized highlights + } + + # Basic presence + for ln in lines: + if RE_STARTED.search(ln): + findings["ran"] = True + if RE_EMPTY_CORPUS.search(ln): + findings["started_from_empty_corpus"] = True + + # Stats + m = RE_DONE.search(ln) + if m: + try: + findings["duration_s"] = int(m.group(1)) + except Exception: + pass + + m = RE_EXECS.search(ln) + if m and findings["executions"] is None: + try: + findings["executions"] = int(m.group(1)) + except Exception: + pass + + m = RE_EXECS_PER_SEC.search(ln) + if m and findings["execs_per_sec"] is None: + findings["execs_per_sec"] = float(m.group(1)) + + m = RE_NEW_UNITS.search(ln) + if m and findings["new_units"] is None: + # RE_NEW_UNITS may have two alternative groups + val = next((g for g in m.groups() if g), None) + if val: + try: + findings["new_units"] = int(val) + except Exception: + pass + + m = RE_PEAK_RSS.search(ln) + if m and findings["peak_rss_mb"] is None: + try: + findings["peak_rss_mb"] = int(m.group(1)) + except Exception: + pass + + m = RE_SLOWEST.search(ln) + if m and findings["slowest_unit_s"] is None: + try: + findings["slowest_unit_s"] = float(m.group(1)) + except Exception: + pass + + # NEW_FUNC list + m = RE_NEW_FUNC.search(ln) + if m: + findings["new_funcs"].append(m.group(0).strip()) + + # Errors / sanitizers + m = RE_ERROR_HDR.search(ln) + if m: + findings["errors"].append({"sanitizer": m.group(1), "message": m.group(2).strip()}) + + m = RE_SAN_SUMMARY.search(ln) + if m: + findings["sanitizer_summaries"].append({"sanitizer": m.group(1), "summary": m.group(2).strip()}) + + if RE_TIMEOUT.search(ln): + findings["timeouts"].append(ln.strip()) + if RE_OOM.search(ln): + findings["ooms"].append(ln.strip()) + if RE_DEADLY_SIGNAL.search(ln): + findings["signals"].append(ln.strip()) + + # Artifacts / prefixes + m = RE_ARTIFACT_PREFIX.search(ln) + if m and not findings["artifact_prefix"]: + findings["artifact_prefix"] = m.group(1).strip() + + m = RE_TEST_UNIT_WRITTEN.search(ln) + if m: + findings["artifacts"].append(m.group(1).strip()) + + m = RE_SAVED_ARTIFACT.search(ln) + if m: + findings["artifacts"].append(m.group(0).strip()) + + # Corpus hints + m = RE_CORPUS_DIR.search(ln) + if m: + for g in m.groups(): + if g: + findings["corpus_dir_hints"].append(g.strip()) + + # Warnings + if RE_WARN.search(ln): + findings["warnings"].append(ln.strip()) + + # Generic stats + m = RE_STAT.search(ln) + if m: + key, val = m.group(1), m.group(2).strip() + findings["raw_stats"][key] = val + + # Synthesize “interesting” bullets + if findings["errors"] or findings["sanitizer_summaries"]: + findings["interesting"].append("Sanitizer findings present.") + if findings["timeouts"]: + findings["interesting"].append(f"{len(findings['timeouts'])} timeout indication(s).") + if findings["ooms"]: + findings["interesting"].append(f"{len(findings['ooms'])} out-of-memory indication(s).") + if findings["signals"]: + findings["interesting"].append(f"{len(findings['signals'])} deadly signal indication(s).") + if findings["artifacts"]: + findings["interesting"].append(f"Artifact(s) saved: {len(findings['artifacts'])}.") + if findings["new_funcs"]: + findings["interesting"].append(f"Exploration: {len(findings['new_funcs'])} NEW_FUNC discoveries.") + if findings["new_units"]: + findings["interesting"].append(f"Corpus growth: +{findings['new_units']} new units.") + if findings["artifact_prefix"]: + findings["interesting"].append(f"artifact_prefix set: {findings['artifact_prefix']}") + if findings["started_from_empty_corpus"]: + findings["interesting"].append("Started from EMPTY corpus (consider seeding).") + if findings["warnings"]: + findings["interesting"].append(f"{len(findings['warnings'])} warning(s) encountered.") + + return findings + + +def summarize_human(path: Path, f: Dict[str, Any], max_show_funcs: int = 10) -> str: + lines: List[str] = [] + lines.append(f"\n=== {path} ===") + lines.append(f"Ran: {'yes' if f['ran'] else 'no/unknown'}") + if f["ran"]: + lines.append(f"- Executions: {f['executions'] if f['executions'] is not None else 'n/a'}") + lines.append(f"- Exec/s: {f['execs_per_sec'] if f['execs_per_sec'] is not None else 'n/a'}") + lines.append(f"- Duration (s): {f['duration_s'] if f['duration_s'] is not None else 'n/a'}") + lines.append(f"- New units: {f['new_units'] if f['new_units'] is not None else 'n/a'}") + lines.append(f"- Peak RSS (MB): {f['peak_rss_mb'] if f['peak_rss_mb'] is not None else 'n/a'}") + lines.append(f"- Slowest unit (s): {f['slowest_unit_s'] if f['slowest_unit_s'] is not None else 'n/a'}") + if f["started_from_empty_corpus"]: + lines.append("- Note: started from empty corpus") + + has_findings = any([f["errors"], f["sanitizer_summaries"], f["timeouts"], f["ooms"], f["signals"]]) + lines.append(f"Findings worth investigating: {'YES' if has_findings else 'no'}") + + if f["errors"]: + lines.append("\n-- Sanitizer Errors --") + for e in f["errors"][:5]: + lines.append(f" * {e['sanitizer']}: {e['message']}") + if len(f["errors"]) > 5: + lines.append(f" ... and {len(f['errors'])-5} more") + + if f["sanitizer_summaries"]: + lines.append("\n-- Sanitizer Summaries --") + for s in f["sanitizer_summaries"][:5]: + lines.append(f" * {s['sanitizer']}: {s['summary']}") + if len(f["sanitizer_summaries"]) > 5: + lines.append(f" ... and {len(f['sanitizer_summaries'])-5} more") + + if f["timeouts"]: + lines.append("\n-- Timeouts --") + for t in f["timeouts"][:5]: + lines.append(f" * {t}") + if len(f["timeouts"]) > 5: + lines.append(f" ... and {len(f['timeouts'])-5} more") + + if f["ooms"]: + lines.append("\n-- OOMs --") + for o in f["ooms"][:5]: + lines.append(f" * {o}") + if len(f["ooms"]) > 5: + lines.append(f" ... and {len(f['ooms'])-5} more") + + if f["signals"]: + lines.append("\n-- Signals --") + for s in f["signals"][:5]: + lines.append(f" * {s}") + if len(f["signals"]) > 5: + lines.append(f" ... and {len(f['signals'])-5} more") + + if f["artifact_prefix"] or f["artifacts"]: + lines.append("\n-- Artifacts --") + if f["artifact_prefix"]: + lines.append(f" artifact_prefix: {f['artifact_prefix']}") + for a in f["artifacts"][:5]: + lines.append(f" * {a}") + if len(f["artifacts"]) > 5: + lines.append(f" ... and {len(f['artifacts'])-5} more") + + if f["corpus_dir_hints"]: + lines.append("\n-- Corpus Hints --") + for c in list(dict.fromkeys(f["corpus_dir_hints"]))[:5]: + lines.append(f" * {c}") + + if f["new_funcs"]: + lines.append("\n-- NEW_FUNC (coverage discoveries) --") + for nf in f["new_funcs"][:max_show_funcs]: + lines.append(f" * {nf}") + if len(f["new_funcs"]) > max_show_funcs: + lines.append(f" ... and {len(f['new_funcs'])-max_show_funcs} more") + + if f["warnings"]: + lines.append("\n-- Warnings --") + for w in f["warnings"][:5]: + lines.append(f" * {w}") + if len(f["warnings"]) > 5: + lines.append(f" ... and {len(f['warnings'])-5} more") + + if f["interesting"]: + lines.append("\n-- Highlights --") + for h in f["interesting"]: + lines.append(f" * {h}") + + # Helpful next steps if there are findings or no findings + lines.append("\n-- Next Steps --") + if has_findings: + lines.append(" * Reproduce the crashing/timeout unit locally with the target binary.") + lines.append(" * Inspect saved artifacts (see above) and run under ASan/UBSan with symbols.") + lines.append(" * Use -runs=1 @artifact to confirm and reduce; apply -minimize_crash=1 if needed.") + else: + lines.append(" * Consider longer runs, parallel jobs, value profiling (-use_value_profile=1).") + lines.append(" * Add UBSan/MSan builds and re-run on existing corpus.") + lines.append(" * Seed corpus with realistic samples if started empty.") + + return "\n".join(lines) + + +def analyze_paths(paths: List[Path]) -> Tuple[List[Dict[str, Any]], List[str]]: + results: List[Dict[str, Any]] = [] + render: List[str] = [] + for p in paths: + try: + text = p.read_text(errors="replace") + except Exception as e: + results.append({"path": str(p), "error": str(e)}) + render.append(f"\n=== {p} ===\nERROR: {e}") + continue + + parsed = parse_log(text) + parsed["path"] = str(p) + results.append(parsed) + render.append(summarize_human(p, parsed)) + return results, render + + +def main(): + ap = argparse.ArgumentParser(description="Summarize libFuzzer/OSS-Fuzz logs and surface actionable findings.") + ap.add_argument("paths", nargs="+", help="Log files (supporting globs via shell).") + ap.add_argument("--json", action="store_true", help="Emit JSON instead of human-readable text.") + ap.add_argument("--max-funcs", type=int, default=10, help="Max NEW_FUNC lines to display per file.") + args = ap.parse_args() + + files: List[Path] = [Path(p) for p in args.paths] + results, render = analyze_paths(files) + + if args.json: + print(json.dumps(results, indent=2)) + else: + print("\n".join(render)) + + +if __name__ == "__main__": + main()