modify

Author: joyguoguo

fuzz/build_fuzzers.py

@@ -27,52 +27,53 @@
 from typing import Optional
 from multiprocessing import Pool, cpu_count
 from errors import BuildError, CommandError, PathError, ConfigError
-
-def run_command(
-    cmd: str,
-    oss_fuzz_dir: Path,
-    project: str = "",
-    allowed_exit_codes: Optional[list[int]] = None
-) -> int:
-    """Execute a command and return the exit code"""
-    allowed_exit_codes = allowed_exit_codes or [0]
-    logging.info(f"▶️ Executing command: {cmd}")
+from command_util import run_command_build_fuzz as run_command
+
+# def run_command(
+#     cmd: str,
+#     oss_fuzz_dir: Path,
+#     project: str = "",
+#     allowed_exit_codes: Optional[list[int]] = None
+# ) -> int:
+#     """Execute a command and return the exit code"""
+#     allowed_exit_codes = allowed_exit_codes or [0]
+#     logging.info(f"▶️ Executing command: {cmd}")
 
-    try:
-        process = subprocess.Popen(
-            cmd,
-            shell=True,
-            cwd=str(oss_fuzz_dir),
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            text=True
-        )
+#     try:
+#         process = subprocess.Popen(
+#             cmd,
+#             shell=True,
+#             cwd=str(oss_fuzz_dir),
+#             stdout=subprocess.PIPE,
+#             stderr=subprocess.PIPE,
+#             text=True
+#         )
 
-        stdout, stderr = process.communicate()
-        exit_code = process.returncode
+#         stdout, stderr = process.communicate()
+#         exit_code = process.returncode
 
-        if exit_code in allowed_exit_codes:
-            return exit_code
+#         if exit_code in allowed_exit_codes:
+#             return exit_code
 
-        # Build detailed error message
-        error_msg = f"Command failed (exit code: {exit_code})"
-        if project:
-            error_msg += f" for project: {project}"
+#         # Build detailed error message
+#         error_msg = f"Command failed (exit code: {exit_code})"
+#         if project:
+#             error_msg += f" for project: {project}"
 
-        if stderr.strip():
-            error_msg += f"\nError output:\n{stderr.strip()}"
+#         if stderr.strip():
+#             error_msg += f"\nError output:\n{stderr.strip()}"
 
-        if stdout.strip():
-            error_msg += f"\nOutput:\n{stdout.strip()}"
+#         if stdout.strip():
+#             error_msg += f"\nOutput:\n{stdout.strip()}"
 
-        raise CommandError(error_msg, project=project, exit_code=exit_code)
+#         raise CommandError(error_msg, project=project, exit_code=exit_code)
 
-    except FileNotFoundError as e:
-        raise CommandError(f"Command not found: {cmd.split()[0]}", project=project) from e
-    except OSError as e:
-        raise CommandError(f"System error: {e}", project=project) from e
-    except subprocess.SubprocessError as e:
-        raise CommandError(f"Subprocess error: {e}", project=project) from e
+#     except FileNotFoundError as e:
+#         raise CommandError(f"Command not found: {cmd.split()[0]}", project=project) from e
+#     except OSError as e:
+#         raise CommandError(f"System error: {e}", project=project) from e
+#     except subprocess.SubprocessError as e:
+#         raise CommandError(f"Subprocess error: {e}", project=project) from e
 
 def build_fuzzers(project_name: str, sanitizer: str, oss_fuzz_dir: Path) -> tuple[bool, str]:
     """Fuzzer build workflow"""

fuzz/run_fuzz_all_targets.py

@@ -25,78 +25,75 @@
 from pathlib import Path
 from multiprocessing import Pool, cpu_count
 from returns.maybe import Maybe, Nothing, Some
-
-
-
-
-
-def run_command(
-    cmd: str,
-    log_msg: str,
-    logger: logging.Logger,
-    allowed_exit_codes: Maybe[list[int]] = Nothing,
-    timeout: int = 3600  # Default 1-hour timeout
-) -> bool:
-    """Execute commands with real-time logging and precise error handling"""
-    allowed_codes = allowed_exit_codes.value_or([])
-    logger.info(f"▶️ {log_msg}...")
-    logger.debug(f"   $ {cmd}")
-
-    process = None
-    try:
-        process = subprocess.Popen(
-            cmd,
-            shell=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            text=True,
-            encoding="utf-8",
-            errors="replace"
-        )
-
-        start_time = time.time()
-        while process.poll() is None:
-            if time.time() - start_time > timeout:
-                logger.error(f"⌛ Command timed out after {timeout} seconds")
-                process.terminate()
-                try:
-                    process.wait(timeout=5)
-                except subprocess.TimeoutExpired:
-                    process.kill()
-                return False
-
-            if process.stdout:
-                line = process.stdout.readline()
-                if line:
-                    logger.debug(line.strip())
-            else:
-                time.sleep(0.1)
-
-        exit_code = process.returncode
-        if exit_code not in [0, *allowed_codes]:
-            logger.error(f"❌ Command execution failed, exit code: {exit_code}")
-            return False
-        return True
-
-    except FileNotFoundError:
-        logger.error(f"🔍 Command not found: {cmd.split()[0]}")
-        return False
-    except PermissionError:
-        logger.error(f"🔒 Insufficient permissions to execute command: {cmd}")
-        return False
-    except subprocess.SubprocessError as e:
-        logger.exception(f"💥 Subprocess error: {e}")
-        return False
-    except OSError as e:
-        logger.exception(f"💥 Operating system error during command execution: {e}")
-        return False
-    finally:
-        if process and process.poll() is None:
-            try:
-                process.terminate()
-                process.wait(timeout=5)
-            except Exception:
-                pass
+from command_util import run_command_fuzz_all_targets as run_command
+
+# def run_command(
+#     cmd: str,
+#     log_msg: str,
+#     logger: logging.Logger,
+#     allowed_exit_codes: Maybe[list[int]] = Nothing,
+#     timeout: int = 3600  # Default 1-hour timeout
+# ) -> bool:
+#     """Execute commands with real-time logging and precise error handling"""
+#     allowed_codes = allowed_exit_codes.value_or([])
+#     logger.info(f"▶️ {log_msg}...")
+#     logger.debug(f"   $ {cmd}")
+
+#     process = None
+#     try:
+#         process = subprocess.Popen(
+#             cmd,
+#             shell=True,
+#             stdout=subprocess.PIPE,
+#             stderr=subprocess.STDOUT,
+#             text=True,
+#             encoding="utf-8",
+#             errors="replace"
+#         )
+
+#         start_time = time.time()
+#         while process.poll() is None:
+#             if time.time() - start_time > timeout:
+#                 logger.error(f"⌛ Command timed out after {timeout} seconds")
+#                 process.terminate()
+#                 try:
+#                     process.wait(timeout=5)
+#                 except subprocess.TimeoutExpired:
+#                     process.kill()
+#                 return False
+
+#             if process.stdout:
+#                 line = process.stdout.readline()
+#                 if line:
+#                     logger.debug(line.strip())
+#             else:
+#                 time.sleep(0.1)
+
+#         exit_code = process.returncode
+#         if exit_code not in [0, *allowed_codes]:
+#             logger.error(f"❌ Command execution failed, exit code: {exit_code}")
+#             return False
+#         return True
+
+#     except FileNotFoundError:
+#         logger.error(f"🔍 Command not found: {cmd.split()[0]}")
+#         return False
+#     except PermissionError:
+#         logger.error(f"🔒 Insufficient permissions to execute command: {cmd}")
+#         return False
+#     except subprocess.SubprocessError as e:
+#         logger.exception(f"💥 Subprocess error: {e}")
+#         return False
+#     except OSError as e:
+#         logger.exception(f"💥 Operating system error during command execution: {e}")
+#         return False
+#     finally:
+#         if process and process.poll() is None:
+#             try:
+#                 process.terminate()
+#                 process.wait(timeout=5)
+#             except Exception:
+#                 pass
 
 
 def discover_targets(project_name: str, oss_fuzz_dir: Path, logger: logging.Logger) -> list[str]: