build_oss_fuzz.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
build_oss_fuzz.py

Parallel build of OSS-Fuzz projects (Docker images and Fuzzer compilation).
Uses multiprocessing.Pool to distribute projects across multiple CPU cores for concurrent processing.

Usage: python3 build_oss_fuzz.py [project_list_file] [--sanitizer type] [--workers N]
Example: python3 fuzz/build_oss_fuzz.py data/valid_projects.txt \
    --sanitizer address \
    --workers 8
"""
import os
import sys
import subprocess
import argparse
from pathlib import Path
from typing import List, Optional, Tuple
from multiprocessing import Pool, cpu_count
import logging

# --- Global configuration ---
HOME_DIR = Path.home()
OSS_FUZZ_DIR = HOME_DIR / "FuzzAug" / "fuzz" / "oss-fuzz"
def run_command(
    cmd: str,
    log_msg: str,
    allowed_exit_codes: Optional[List[int]] = None
) -> bool:
    """Execute a shell command and stream output to console"""
    allowed_exit_codes = allowed_exit_codes or []

    logging.info(f"▶️ {log_msg}")
    logging.debug(f"$ {cmd}")

    try:
        process = subprocess.Popen(
            f"yes | {cmd}",
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            encoding="utf-8",
            errors="replace"
        )
    except FileNotFoundError:
        logging.error(f"Command not found: {cmd}")
        return False
    except OSError as e:
        logging.error(f"OS error while executing command: {e}")
        return False
    except ValueError as e:
        logging.error(f"Invalid arguments to Popen: {e}")
        return False

    try:
        if process.stdout:
            for line in iter(process.stdout.readline, ""):
                sys.stdout.write(line)
                sys.stdout.flush()
        process.wait()
        exit_code = process.returncode
        if exit_code in [0, *allowed_exit_codes]:
            logging.info("✅ Command completed successfully")
            return True
        logging.error(f"❌ Command failed (exit code: {exit_code})")
        return False
    except KeyboardInterrupt:
        logging.warning("⛔️ Command interrupted by user")
        process.terminate()
        return False
    except Exception as e:
        logging.exception(f"Unexpected error during process execution: {e}")
        return False


def build_project(project_name: str, sanitizer: str) -> Tuple[bool, str]:
    """Build workflow for a single project"""
    os.chdir(OSS_FUZZ_DIR)

    logging.info("=" * 60)
    logging.info(f"🔨 Starting build for project: {project_name}")
    logging.info("=" * 60)

    if not run_command(
        f"python3 infra/helper.py build_image {project_name}",
        "Step 1/2: Building Docker image"
    ):
        return (False, project_name)

    if not run_command(
        f"python3 infra/helper.py build_fuzzers --sanitizer {sanitizer} {project_name}",
        f"Step 2/2: Compiling Fuzzers (sanitizer={sanitizer})"
    ):
        return (False, project_name)

    logging.info(f"✅ Project {project_name} build completed")
    return (True, project_name)

def main():
    parser = argparse.ArgumentParser(description="OSS-Fuzz Parallel Build Tool")
    parser.add_argument("project_list", help="Project list file path")
    parser.add_argument("--sanitizer", default="address", choices=["address", "memory", "undefined"])
    parser.add_argument("--workers", type=int, default=cpu_count())
    args = parser.parse_args()

    logging.basicConfig(
        level=logging.INFO,
        format='[%(levelname)s] [PID:%(process)d] %(message)s'
    )

    if not os.path.isfile(args.project_list):
        logging.error(f"Project list file not found: {args.project_list}")
        sys.exit(1)

    try:
        with open(args.project_list, "r", encoding="utf-8") as f:
            projects = [line.strip() for line in f if line.strip()]
    except OSError as e:
        logging.error(f"OS error while reading project list: {e}")
        sys.exit(1)
    except UnicodeDecodeError as e:
        logging.error(f"Encoding error while reading file: {e}")
        sys.exit(1)

    with Pool(args.workers) as pool:
        results = pool.starmap(build_project, [(p, args.sanitizer) for p in projects])

    failed = [p for success, p in results if not success]
    logging.info(f"\n📊 Build completed: Success {len(projects) - len(failed)}/{len(projects)}")
    if failed:
        logging.warning("❌ Failed projects: " + ", ".join(failed))

if __name__ == "__main__":
    main()