Fuzzing

packages/fuzzing/
├── afl_runner.py          # AFL++ campaign orchestration
├── crash_collector.py     # Crash collection & deduplication
└── corpus_manager.py      # Corpus generation & management

from pathlib import Path
from packages.fuzzing import AFLRunner

# Initialize AFL runner
runner = AFLRunner(
    binary_path=Path("/path/to/target_binary"),
    corpus_dir=Path("seeds/"),
    output_dir=Path("out/fuzz_results")
)

# Check instrumentation
is_instrumented = runner.check_binary_instrumentation()

# Start fuzzing (60 seconds)
result = runner.run_single_fuzzer(
    duration_seconds=60,
    memory_limit="500M"
)

print(f"Execs: {result['total_execs']}")
print(f"Crashes: {result['unique_crashes']}")

# Run with multiple workers
result = runner.run_parallel_fuzzers(
    num_workers=4,
    duration_seconds=300,
    memory_limit="500M"
)

print(f"Workers: {len(result['worker_stats'])}")
print(f"Total crashes: {result['total_unique_crashes']}")

from packages.fuzzing import CrashCollector

collector = CrashCollector(
    binary_path=Path("/path/to/target"),
    output_dir=Path("out/crashes")
)

# Collect crashes from AFL output
crashes = collector.collect_crashes(
    afl_output_dir=Path("out/fuzz_results")
)

for crash in crashes:
    print(f"Crash ID: {crash.crash_id}")
    print(f"Signal: {crash.signal}")
    print(f"Hash: {crash.stack_hash}")
    print(f"Input: {crash.input_file}")

class AFLRunner:
    def __init__(
        self,
        binary_path: Path,
        corpus_dir: Optional[Path] = None,
        output_dir: Optional[Path] = None,
        dict_path: Optional[Path] = None,
        input_mode: str = "stdin",
        check_sanitizers: bool = False,
        recompile_guide: bool = False,
        use_showmap: bool = False
    )
    
    def run_single_fuzzer(
        self,
        duration_seconds: int,
        memory_limit: str = "none",
        extra_args: List[str] = None
    ) -> Dict[str, Any]
    
    def run_parallel_fuzzers(
        self,
        num_workers: int,
        duration_seconds: int,
        memory_limit: str = "none"
    ) -> Dict[str, Any]
    
    def check_binary_instrumentation(self) -> bool

class CrashCollector:
    def __init__(
        self,
        binary_path: Path,
        output_dir: Path
    )
    
    def collect_crashes(
        self,
        afl_output_dir: Path
    ) -> List[Crash]
    
    def deduplicate_crashes(
        self,
        crashes: List[Crash]
    ) -> List[Crash]
    
    def triage_crash(
        self,
        crash: Crash
    ) -> Dict[str, Any]

class CorpusManager:
    def __init__(self, corpus_dir: Path)
    
    def generate_seeds(
        self,
        num_seeds: int = 10,
        seed_type: str = "random"
    ) -> List[Path]
    
    def import_corpus(
        self,
        source_dir: Path
    ) -> int
    
    def minimize_corpus(
        self,
        afl_cmin: str = "afl-cmin"
    ) -> int

from pathlib import Path
from packages.fuzzing import AFLRunner, CrashCollector, CorpusManager
from packages.binary_analysis import CrashAnalyser

# 1. Prepare corpus
corpus = CorpusManager(Path("seeds/"))
corpus.generate_seeds(num_seeds=20, seed_type="mixed")

# 2. Run fuzzing campaign
runner = AFLRunner(
    binary_path=Path("target_binary"),
    corpus_dir=Path("seeds/"),
    output_dir=Path("out/fuzz")
)

print("Starting fuzzing campaign...")
result = runner.run_parallel_fuzzers(
    num_workers=4,
    duration_seconds=3600,  # 1 hour
    memory_limit="1G"
)

print(f"Campaign complete: {result['total_execs']:,} execs")
print(f"Found {result['total_unique_crashes']} unique crashes")

# 3. Collect crashes
collector = CrashCollector(
    binary_path=Path("target_binary"),
    output_dir=Path("out/crashes")
)

crashes = collector.collect_crashes(Path("out/fuzz"))
print(f"Collected {len(crashes)} crashes")

# 4. Analyze crashes
analyser = CrashAnalyser(Path("target_binary"))

for crash in crashes[:5]:  # Top 5 crashes
    context = analyser.analyze_crash(
        input_file=crash.input_file,
        signal=crash.signal
    )
    print(f"\nCrash {crash.crash_id}:")
    print(f"  Type: {context.crash_type}")
    print(f"  Exploitability: {context.exploitability}")

# Compile with AFL instrumentation
export CC=afl-clang-fast
export CXX=afl-clang-fast++

# Build with ASAN for better crash detection
export AFL_USE_ASAN=1

./configure && make

runner = AFLRunner(binary_path=Path("target"))

if runner.check_binary_instrumentation():
    print("✓ Binary is instrumented")
else:
    print("⚠ Binary not instrumented - will use QEMU mode (slower)")

from packages.fuzzing import CorpusManager

corpus = CorpusManager(Path("seeds/"))

# Generate mixed seeds
seeds = corpus.generate_seeds(
    num_seeds=20,
    seed_type="mixed"  # random, structured, or mixed
)

print(f"Generated {len(seeds)} seed files")

# Import seeds from another source
count = corpus.import_corpus(Path("/path/to/external/seeds"))
print(f"Imported {count} seeds")

# Remove redundant seeds (keeps only unique coverage)
remaining = corpus.minimize_corpus()
print(f"Minimized to {remaining} unique seeds")

runner = AFLRunner(
    binary_path=Path("target"),
    corpus_dir=Path("seeds/"),
    output_dir=Path("out/fuzz"),
    dict_path=Path("target.dict"),  # Mutation dictionary
    input_mode="stdin",              # or "file", "network"
    check_sanitizers=True,           # Check for ASAN
    use_showmap=True                 # Verify coverage
)

# Run with memory limit
result = runner.run_single_fuzzer(
    duration_seconds=300,
    memory_limit="500M"  # Prevent OOM
)

# macOS requires higher shared memory limits
sudo afl-system-config

# Or manually:
sudo sysctl kern.sysv.shmmax=8388608
sudo sysctl kern.sysv.shmall=2048

out/fuzz_results/
├── main_node/              # Primary fuzzer
│   ├── crashes/
│   │   ├── id:000000,sig:11    # Crash inputs
│   │   └── id:000001,sig:06
│   ├── queue/             # Corpus queue
│   └── fuzzer_stats       # Statistics
├── secondary_node_01/     # Secondary fuzzers
├── secondary_node_02/
└── secondary_node_03/

out/crashes/               # Collected & analyzed crashes
├── crash_abc123/
│   ├── input              # Crashing input
│   ├── stack_trace.txt    # Stack trace
│   ├── registers.txt      # Register dump
│   └── analysis.json      # LLM analysis
└── crash_def456/

from packages.fuzzing import AFLRunner, CrashCollector
from packages.binary_analysis import CrashAnalyser

# Fuzz -> Collect -> Analyze
runner = AFLRunner(...)
result = runner.run_parallel_fuzzers(...)

collector = CrashCollector(...)
crashes = collector.collect_crashes(...)

analyser = CrashAnalyser(...)
for crash in crashes:
    analysis = analyser.analyze_crash(...)

from packages.llm_analysis import AutonomousSecurityAgentV2

# Analyze crash with AI
agent = AutonomousSecurityAgentV2(...)

for crash in crashes:
    context = analyser.analyze_crash(crash.input_file, crash.signal)
    exploit = agent.generate_exploit_for_crash(context)