Chapter 5: 模糊测试执行器 (Fuzzing Executor)

欢迎来到 AutoSafeCoder 教程的第五章！在上一章第 4 章：模糊测试输入生成器 (Fuzzing Input Generator) 中，我们认识了那位充满创意的“捣蛋工匠”，它为我们准备了大量奇奇怪怪、专门用来“找茬”的测试输入数据。

现在我们有了测试代码的“弹药”（那些模糊测试输入），但光有弹药还不够，我们还需要一个“靶场”和一个“射击手”来把这些弹药打向我们的目标（也就是程序员智能体编写的代码），并且仔细观察代码在中弹后会有什么反应——会不会“爆炸”（崩溃）、“卡壳”（出错）或者表现出其它异常行为？

这就是我们本章要介绍的主角——模糊测试执行器 (Fuzzing Executor) 的职责所在。

什么是模糊测试执行器？

模糊测试执行器 (Fuzzing Executor) 是 AutoSafeCoder 团队中的“碰撞测试工程师”。它的核心任务是：

接收代码和输入：它从多智能体协作框架那里拿到由程序员智能体生成的代码，以及由模糊测试输入生成器提供的大量、多样化的输入数据。
在受控环境中执行：它在一个相对隔离和安全的环境中，用这些输入数据去实际运行代码。这一点与静态安全分析器不同，后者只检查代码文本而不运行。
监控和报告结果：在运行过程中，它会密切监控代码的行为。如果代码成功运行并返回结果，它会记录下来。更重要的是，如果代码因为某个“奇怪”的输入而崩溃、抛出异常、运行超时或者表现出任何意外行为，它会捕捉到这些情况，并生成报告。

正如概念描述中提到的，模糊测试执行器就像一个汽车碰撞测试机构。这个机构拿到制造好的汽车（代码），把它放在模拟的极端碰撞场景（使用模糊测试输入运行）中进行测试，以评估其在真实世界压力下的稳定性和安全性。输入生成器负责设计各种碰撞场景（生成输入），而执行器则负责实际进行碰撞测试并记录结果。

模糊测试执行器如何工作？（高层视角）

想象一下我们的“碰撞测试工程师”（执行器）拿到了一段由程序员智能体写的代码（比如一个处理用户配置文件的函数）和一堆由模糊测试输入生成器准备好的“奇葩”配置文件（测试输入）。

准备测试环境：工程师会准备一个安全的测试场地（一个隔离的运行环境），确保测试过程不会影响到外部系统。
逐个进行测试：工程师拿起第一个“奇葩”配置文件（输入），加载到测试代码中，然后启动代码运行。
观察并记录：
- 如果代码顺利处理了配置文件并正常退出，工程师记录“通过”。
- 如果代码在处理过程中突然崩溃（比如程序意外终止），工程师记录“崩溃”，并记下是哪个配置文件导致的。
- 如果代码运行了很久都没有反应（超过了预设时间），工程师记录“超时”。
- 如果代码抛出了一个未被处理的错误（异常），工程师记录“错误”及错误信息。
重复测试：工程师继续拿起下一个配置文件，重复步骤 3，直到所有准备好的输入都测试完毕。
汇总报告：最后，工程师整理所有记录，形成一份详细的测试报告，交给多智能体协作框架。这份报告会清楚地说明哪些输入导致了问题，以及问题的具体表现。

这份报告对于改进代码至关重要。框架会根据报告中指出的问题，让程序员智能体去修复代码中的缺陷。

如何使用模糊测试执行器？

在 AutoSafeCoder 项目中，模糊测试执行器的功能主要由 executor_agent_safe.py 文件中的 execute_fuzz 函数提供。虽然它不像其他组件那样被封装成一个单独的“Agent”类，但它扮演了执行器的角色。多智能体协作框架（main.py 中的 MultiAgentSystem 类）在模糊测试的循环中调用这个函数。

让我们看看 main.py 中框架是如何调用它的：

1. 在模糊测试循环中调用 execute_fuzz

框架在 run 方法中，拿到初始输入后，会进入一个循环。在每次循环中，它调用 execute_fuzz 来运行代码，然后使用模糊测试输入生成器（InputMutatorAgent）生成下一轮的输入。

# main.py (run 方法片段 - 模糊测试循环简化)
from executor_agent_safe import execute_fuzz # 导入执行函数
# ... 其他导入 ...

class MultiAgentSystem:
    # ... (init, 静态分析, 生成初始输入等) ...
    def run(self, iterations=120):
        # ... (前面的代码) ...
        current_inputs = self.test_inputs # 使用初始或上一轮变异的输入
        failed_inputs_fuzz = [] # 记录导致失败的输入
        fuzzing_test_status = None # 记录模糊测试最终状态

        for iteration in range(iterations): # 进行 N 轮模糊测试
            print(f"\n模糊测试迭代 {iteration + 1}")

            # === 关键步骤: 调用 execute_fuzz 执行代码 ===
            try:
                # 传入当前代码、当前输入和超时时间 (例如 3 秒)
                result_str, passed, inputs_used, func_name = execute_fuzz(
                    self.code,
                    current_inputs,
                    3
                )
            except Exception as e:
                # 处理 execute_fuzz 本身可能出现的异常
                print(f"执行模糊测试时出错: {e}")
                fuzzing_test_status = "error running function"
                break # 中断模糊测试循环

            print(f"执行结果: {result_str}") # 打印执行器返回的字符串信息

            # === 检查执行结果 ===
            if not passed: # 如果 passed 为 False，表示代码执行失败
                # (省略了处理特定错误类型的代码, 如模块缺失等)
                print("代码执行失败或超时！")
                failed_inputs_fuzz.append({'inputs': inputs_used, 'result': result_str})
                # 如果失败次数过多，提前终止
                if len(failed_inputs_fuzz) > 10:
                    break
            else:
                print("代码执行成功。")

            # === 变异输入，准备下一轮 ===
            # (调用 InputMutatorAgent.mutate_inputs() 的代码，见上一章)
            # current_inputs = mutator_agent.mutate_inputs()
            # ...

        # === 模糊测试循环结束，处理结果 ===
        # (如果 failed_inputs_fuzz 不为空，则反馈给 ProgrammerAgent 尝试修复)
        # ... (省略) ...

代码解释:

导入: 首先，我们从 executor_agent_safe.py 文件导入 execute_fuzz 函数。
循环: 代码在一个 for 循环中运行，模拟多轮模糊测试。
调用 execute_fuzz: 在循环内部，最关键的一行是调用 execute_fuzz。我们传入三个主要参数：
- self.code: 当前需要测试的 Python 代码字符串。
- current_inputs: 模糊测试输入生成器准备好的、用于本次测试的输入数据（通常是一个字典）。
- 3: 超时时间（单位：秒）。如果代码运行超过 3 秒还没结束，执行器会强制终止它并报告超时。
获取结果: execute_fuzz 函数会返回几个值：
- result_str: 一个描述执行结果的字符串（比如 “passed”, “timed out”, “failed: <错误信息>”）。
- passed: 一个布尔值，True 表示代码成功执行，False 表示失败（包括错误、超时等）。
- inputs_used: 实际传递给被测函数的输入（通常与 current_inputs 相同）。
- func_name: 从代码中提取出的函数名。
检查结果: 框架检查 passed 变量。如果为 False，说明本次测试发现了问题。框架会记录下导致失败的输入 (inputs_used) 和结果 (result_str) 到 failed_inputs_fuzz 列表中。
后续处理: 循环结束后，如果 failed_inputs_fuzz 列表不为空，框架就会将这些失败信息反馈给程序员智能体，让它尝试修复代码。

示例输入:

self.code:

def divide(a, b):
    # 这个函数没有处理除以零的情况
    return a / b

current_inputs: {'a': 10, 'b': 0} (由模糊测试输入生成器生成)
timeout: 3

可能的输出 (execute_fuzz 返回值):

result_str: "failed: division by zero" (或者类似的错误信息)
passed: False
inputs_used: {'a': 10, 'b': 0}
func_name: "divide"

框架会捕捉到这个 passed = False 的结果，并将 {'inputs': {'a': 10, 'b': 0}, 'result': 'failed: division by zero'} 添加到 failed_inputs_fuzz 列表中。

通过这种方式，模糊测试执行器（execute_fuzz 函数）充当了实际运行代码并报告问题的关键角色。

内部实现揭秘

我们已经了解了模糊测试执行器（execute_fuzz 函数）的作用和用法，现在让我们稍微深入一点，看看它是如何在 executor_agent_safe.py 文件内部安全地执行潜在不安全的代码的。

非代码流程图解:

当框架调用 execute_fuzz 函数时，内部大致发生了什么？

sequenceDiagram
    participant MAS as 多智能体协作框架 (main.py)
    participant EF as 执行器函数 (execute_fuzz)
    participant MP as Python多进程模块 (multiprocessing)
    participant ChildP as 子进程 (运行unsafe_execute)
    participant Code as 被测代码 (在子进程中运行)

    MAS->>EF: 调用 execute_fuzz(代码, 输入, 超时时间)
    EF->>MP: 创建Manager用于进程间通信 (结果列表)
    EF->>MP: 准备启动一个新进程，目标是 unsafe_execute 函数
    MP->>ChildP: 启动子进程
    Note right of ChildP: 子进程开始执行 unsafe_execute
    ChildP->>ChildP: 设置安全防护 (reliability_guard)
    ChildP->>ChildP: 构造包含代码和输入的完整脚本
    ChildP->>Code: 使用 exec() 执行构造好的脚本
    alt 代码执行成功
        Code-->>ChildP: 正常结束 (可能打印结果到标准输出)
        ChildP->>MP: 将 "passed" 写入结果列表
    else 代码执行出错/超时
        Code-->>ChildP: 抛出异常 或 超时被捕获
        ChildP->>MP: 将错误信息或 "timed out" 写入结果列表
    end
    ChildP-->>MP: 子进程结束
    MP-->>EF: 父进程检测到子进程结束
    EF->>MP: 从Manager获取结果列表
    EF-->>MAS: 返回解析后的结果 (result_str, passed, inputs, func_name)

图解说明:

框架 (MAS) 调用 execute_fuzz 函数 (EF)，传入代码、输入和超时时间。
execute_fuzz (EF) 使用 Python 的 multiprocessing 模块 (MP) 来创建一个新的、独立的子进程 (ChildP)。这样做是为了隔离执行，防止被测代码如果崩溃或行为不当，影响到主程序（框架）。它还创建了一个共享的 Manager.list() 用于从子进程取回结果。
子进程 (ChildP) 开始执行一个名为 unsafe_execute 的内部函数。
在 unsafe_execute 内部，首先会调用 reliability_guard() 设置一些安全限制（比如禁用某些危险的系统调用），这是安全代码执行环境的一部分。
然后，它会将传入的代码和输入数据动态地组装成一个完整的 Python 脚本字符串。
最关键的一步：子进程使用 exec() 函数来执行这个构造好的脚本字符串。exec() 可以执行存储在字符串中的 Python 代码。同时，会设置一个时间限制 (time_limit)。
被测代码 (Code) 开始在子进程中运行。
- 如果代码正常运行结束，子进程会捕获其输出（如果有），并将表示成功的状态（如 “passed”）写入共享的结果列表中。
- 如果代码运行出错（比如抛出未捕获的异常，如 ZeroDivisionError），或者运行时间超过了设定的阈值，exec() 的调用会被中断或捕获。子进程会将相应的错误信息（如 “failed: division by zero”）或 “timed out” 写入结果列表。
子进程执行完毕后退出。
父进程 (EF) 通过 p.join() 等待子进程结束（或者超时后强制结束 p.kill()）。
父进程从共享的结果列表中读取子进程写入的状态信息。
execute_fuzz 函数解析这个状态信息，转换成 result_str 和 passed 等返回值，最终交还给框架 (MAS)。

这种使用独立进程和超时控制的方法是执行不可信代码（比如由 AI 生成的代码）时常用的安全实践。

代码层面的深入了解:

让我们看看 executor_agent_safe.py 中 execute_fuzz 函数的核心部分。

# executor_agent_safe.py (片段 - 简化)
import multiprocessing
import signal
import contextlib
import tempfile
import os
import json
# ... 其他导入 ...

# (FResult 枚举定义，省略)
# (remove_json_prefix 函数定义，省略)

def execute_fuzz(completion: str, input_json, timeout: float,
                      completion_id: Optional[int] = None) -> Dict:

    # 内部函数：实际在子进程中运行的代码
    def unsafe_execute():
        # 在临时目录中运行，并设置资源限制
        with create_tempdir():
            # 禁用危险函数，这是安全措施的一部分
            reliability_guard()

            # 准备要执行的代码字符串
            # (get_exec_code 会将用户代码和输入包装成一个完整脚本)
            code_to_run, inputs, func_name = get_exec_code(completion, input_json)
            inputs_list.append(inputs) # 通过 manager list 把输入传回父进程
            func_names.append(func_name) # 把函数名传回父进程

            try:
                exec_globals = {}
                # 在受控的 IO 环境下执行
                with swallow_io():
                    # 设置超时限制
                    with time_limit(timeout):
                        # === 核心：执行代码 ===
                        exec(code_to_run, exec_globals)
                # 如果 exec 成功并没有抛异常，标记为 passed
                result.append("passed")
            except TimeoutException:
                result.append("timed out") # 捕获超时异常
            except BaseException as e:
                # 捕获所有其他异常
                result.append(f"failed: {e}")

    # 使用 multiprocessing Manager 来在进程间共享结果
    manager = multiprocessing.Manager()
    result = manager.list() # 存储执行结果 ("passed", "timed out", "failed: ...")
    inputs_list = manager.list() # 存储实际使用的输入
    func_names = manager.list() # 存储提取的函数名

    # 创建并启动子进程，运行 unsafe_execute 函数
    p = multiprocessing.Process(target=unsafe_execute)
    p.start()
    # 等待子进程结束，设置一个比内部超时稍长的时间
    p.join(timeout=timeout + 1)

    # 如果子进程仍在运行 (join 超时)，则强制终止
    if p.is_alive():
        p.kill()
        # 如果结果列表为空，说明可能是被 kill 了，标记为超时
        if not result:
             result.append("timed out")

    # 如果结果列表仍然为空 (极端情况)，也标记为超时
    if not result:
        result.append("timed out")

    # 从 manager list 中获取结果
    final_result_str = result[0]
    final_passed = (final_result_str == 'passed')
    final_inputs = inputs_list[0] if inputs_list else {}
    final_func_name = func_names[0] if func_names else None

    # 返回结果给调用者 (main.py)
    return final_result_str, final_passed, final_inputs, final_func_name

# (其他辅助函数如 time_limit, swallow_io, create_tempdir, reliability_guard 定义省略)
# (get_exec_code 函数定义，将用户代码和输入包装成可执行脚本，省略)

代码解释:

unsafe_execute 函数: 这是将在子进程中执行的代码逻辑。
- create_tempdir(): 创建一个临时目录并在其中运行，避免污染主环境。
- reliability_guard(): 调用一个辅助函数，禁用 Python 中一些可能被滥用的功能（如 os.fork, shutil.rmtree），增加安全性。这是安全代码执行环境的关键组成部分。
- get_exec_code(): 这是一个辅助函数（代码中未完全展示），它负责将传入的函数代码 (completion) 和输入 (input_json) 包装成一个完整的、可以直接执行的 Python 脚本字符串。例如，它会添加必要的 import json，定义一个 main 函数来调用用户的函数，并处理输入和输出的 JSON 序列化。
- manager.list().append(): 将解析后的输入 inputs 和函数名 func_name 添加到由 multiprocessing.Manager 创建的共享列表中，以便父进程可以获取它们。
- try...except: 使用 try...except 块来包裹核心的 exec() 调用。
- swallow_io(): 一个上下文管理器，用于抑制被测代码的标准输出和标准错误输出，防止它们干扰主程序的输出。
- time_limit(timeout): 另一个上下文管理器，使用 signal 模块设置一个定时器。如果在 timeout 秒内 exec() 没有执行完毕，它会抛出 TimeoutException。
- exec(code_to_run, exec_globals): 这是执行代码的核心。它在当前的全局命名空间 (exec_globals) 中执行 code_to_run 字符串中的 Python 代码。
- 结果记录: 如果 exec 成功，则将 “passed” 添加到共享的 result 列表；如果捕获到 TimeoutException，则添加 “timed out”；如果捕获到其他任何异常 (BaseException)，则将 “failed: <错误信息>” 添加到 result 列表。
主函数逻辑 (execute_fuzz): 这是在父进程（被 main.py 调用）中执行的代码。
- multiprocessing.Manager(): 创建一个 Manager 对象，用于创建可以在不同进程之间安全共享的数据结构（这里是列表 result, inputs_list, func_names）。
- multiprocessing.Process(): 创建一个新的进程对象 p，指定其目标函数为 unsafe_execute。
- p.start(): 启动子进程。
- p.join(timeout=timeout + 1): 等待子进程结束。设置的超时时间略长于子进程内部的超时时间，以确保子进程有时间完成或超时。
- p.is_alive(), p.kill(): 如果 join 超时后子进程仍然在运行，说明可能卡死了，父进程会强制终止 (kill) 它，并记录结果为 “timed out”。
- 结果解析: 从共享列表 result, inputs_list, func_names 中取出子进程写入的数据。
- 返回: 将最终的执行结果字符串、表示是否通过的布尔值、实际使用的输入和函数名返回给 main.py 中的调用者。

这种结合了多进程隔离、资源限制 (reliability_guard) 和超时控制 (time_limit) 的方法，使得 execute_fuzz 能够相对安全地运行来自不可信来源（如 AI 模型）的代码，并有效地捕捉运行时错误和崩溃。

总结

在本章中，我们深入了解了 AutoSafeCoder 的“碰撞测试工程师”——**模糊测试执行器 (Fuzzing Executor)**。我们学习到：

它的核心职责是实际运行由程序员智能体编写的代码，并使用由模糊测试输入生成器提供的多样化输入进行测试。
它像一个碰撞测试机构，通过模拟极端情况（运行模糊测试输入）来评估代码的健壮性和安全性，目的是发现运行时错误、崩溃或超时。
在 AutoSafeCoder 中，这个功能主要由 executor_agent_safe.py 文件中的 execute_fuzz 函数实现。
多智能体协作框架在模糊测试循环中调用 execute_fuzz，并根据其返回的 passed 状态和结果信息来判断测试是否成功，以及是否需要反馈给程序员进行修复。
我们通过流程图和代码分析，了解了 execute_fuzz 的内部实现关键：使用 Python 的 multiprocessing 模块将代码执行隔离在子进程中，并通过超时控制 (time_limit) 和安全防护 (reliability_guard) 来确保测试过程的相对安全和可控。

模糊测试执行器是 AutoSafeCoder 进行动态代码分析、发现隐藏运行时问题的关键环节。它依赖于一个安全可靠的执行环境来完成任务。

下一章预告: 第 6 章：安全代码执行环境 (Secure Code Execution Environment) - 我们将更详细地探讨 AutoSafeCoder 是如何构建和利用这个安全环境来运行潜在不安全代码的。

Generated by AI Codebase Knowledge Builder

ChenSir

AutoSafeCoder-Fuzzing Executor

Chapter 5: 模糊测试执行器 (Fuzzing Executor)

什么是模糊测试执行器？

模糊测试执行器如何工作？（高层视角）

如何使用模糊测试执行器？

内部实现揭秘

总结