Add fuzzlog_analyzer helper script

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 <log_file_paths...> [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
+```

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()