feat: Python OSS-Fuzz build and fuzz scripts

fuzz/build_oss_fuzz.py

@@ -12,99 +12,91 @@
     --sanitizer address \
     --workers 8
 """
-
 import os
 import sys
 import subprocess
 import argparse
-from datetime import datetime
 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"
-LOG_DIR = OSS_FUZZ_DIR / "build_logs"
-
-def setup_logging(project_name: str) -> Path:
-    """Create a timestamped log file for a single project"""
-    LOG_DIR.mkdir(parents=True, exist_ok=True)
-    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
-    return LOG_DIR / f"build_{project_name}_{timestamp}.log"
-
-def log_and_print(message: str, log_file: Path, to_stdout: bool = True):
-    """Write message to log and print to console"""
-    if to_stdout:
-        print(f"[PID:{os.getpid()}] {message}")
-    with open(log_file, "a", encoding="utf-8") as f:
-        f.write(f"{datetime.now().isoformat()} {message}\n")
-
 def run_command(
-    cmd: str, 
-    log_msg: str, 
-    log_file: Path,
+    cmd: str,
+    log_msg: str,
     allowed_exit_codes: Optional[List[int]] = None
 ) -> bool:
-    """Execute a shell command and stream output to log in real-time"""
+    """Execute a shell command and stream output to console"""
     allowed_exit_codes = allowed_exit_codes or []
-    log_and_print(f"▶️ {log_msg}...", log_file, to_stdout=False)
-    log_and_print(f"   $ {cmd}", log_file, to_stdout=False)
+
+    logging.info(f"▶️ {log_msg}")
+    logging.debug(f"$ {cmd}")
 
     try:
         process = subprocess.Popen(
-            f"yes | {cmd}",  # Auto-confirm all prompts
+            f"yes | {cmd}",
             shell=True,
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
             text=True,
             encoding="utf-8",
             errors="replace"
         )
-        with open(log_file, "a", encoding="utf-8") as f:
-            if process.stdout is not None:
-                for line in iter(process.stdout.readline, ""):
-                    f.write(line)
-            else:
-                log_and_print("⚠️ Warning: process.stdout is None", log_file)
+    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]:
-            log_and_print(f"✅ Command completed successfully", log_file, to_stdout=False)
+            logging.info("✅ Command completed successfully")
             return True
-        log_and_print(f"❌ Command failed (exit code: {exit_code})", log_file)
+        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:
-        log_and_print(f"💥 Execution exception: {e}", log_file)
+        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"""
-    log_file = setup_logging(project_name)
     os.chdir(OSS_FUZZ_DIR)
-
-    log_and_print("="*60, log_file)
-    log_and_print(f"🔨 Starting build for project: {project_name}", log_file)
-    log_and_print(f"📝 Log path: {log_file}", log_file)
-    log_and_print("="*60, log_file)
 
-    # 1. Build Docker image
+    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",
-        log_file
+        "Step 1/2: Building Docker image"
     ):
         return (False, project_name)
 
-    # 2. Compile Fuzzers
     if not run_command(
         f"python3 infra/helper.py build_fuzzers --sanitizer {sanitizer} {project_name}",
-        f"Step 2/2: Compiling Fuzzers (sanitizer={sanitizer})",
-        log_file
+        f"Step 2/2: Compiling Fuzzers (sanitizer={sanitizer})"
     ):
         return (False, project_name)
 
-    log_and_print(f"✅ Project {project_name} build completed", log_file)
+    logging.info(f"✅ Project {project_name} build completed")
     return (True, project_name)
 
 def main():
@@ -114,23 +106,33 @@ def main():
     parser.add_argument("--workers", type=int, default=cpu_count())
     args = parser.parse_args()
 
-    # Read project list
+    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") as f:
+        with open(args.project_list, "r", encoding="utf-8") as f:
             projects = [line.strip() for line in f if line.strip()]
-    except Exception as e:
-        print(f"❌ Failed to read project list: {e}")
+    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)
 
-    # Parallel build
     with Pool(args.workers) as pool:
         results = pool.starmap(build_project, [(p, args.sanitizer) for p in projects])
 
-    # Output results
     failed = [p for success, p in results if not success]
-    print(f"\n📊 Build completed: Success {len(projects)-len(failed)}/{len(projects)}")
+    logging.info(f"\n📊 Build completed: Success {len(projects) - len(failed)}/{len(projects)}")
     if failed:
-        print("❌ Failed projects: " + ", ".join(failed))
+        logging.warning("❌ Failed projects: " + ", ".join(failed))
 
 if __name__ == "__main__":
-    main()
+    main()
+