波弗兹如何检测崩溃并记录崩溃测试用例

Question

我们使用总线对远程服务进行模糊处理。模糊脚本如下所示。

session = Session(target = Target(connection = SocketConnection(host, port, proto='tcp')))
s_initialize("Test")
s_string("Fuzz", fuzzable = True)
session.connect(s_get("Test"))
session.fuzz()

过了一会儿，我们注意到远程服务崩溃了，但模糊程序只是重复尝试重新启动。fuzzer没有检测到远程服务已经关闭，崩溃的测试用例也没有存储。

[2022-02-02 04:18:42,231]    Test Step: Restarting target
[2022-02-02 04:18:42,231]     Info: Restarting target process using CallbackMonitor
[2022-02-02 04:18:42,231]    Test Step: Cleaning up connections from callbacks
[2022-02-02 04:18:42,231]     Info: Closing target connection...
[2022-02-02 04:18:42,231]     Info: Connection closed.
[2022-02-02 04:18:42,231]     Info: No reset handler available... sleeping for 5 seconds
[2022-02-02 04:18:47,236]     Info: Opening target connection (xxx)...
[2022-02-02 04:18:47,237]     Info: Cannot connect to target; retrying. Note: This likely indicates a failure caused by the previous test case, or a target that is slow to restart.
[2022-02-02 04:18:47,237]    Test Step: Restarting target
[2022-02-02 04:18:47,237]     Info: Restarting target process using CallbackMonitor
[2022-02-02 04:18:47,237]    Test Step: Cleaning up connections from callbacks
[2022-02-02 04:18:47,237]     Info: Closing target connection...
[2022-02-02 04:18:47,237]     Info: Connection closed.
[2022-02-02 04:18:47,237]     Info: No reset handler available... sleeping for 5 seconds
[2022-02-02 04:18:52,243]     Info: Opening target connection (xxx)...
[2022-02-02 04:18:52,244]     Info: Cannot connect to target; retrying. Note: This likely indicates a failure caused by the previous test case, or a target that is slow to restart.

我们如何自定义boofuzz脚本，以便：

1. 我们可以检测到远程服务已关闭(例如，尝试tcp连接)？
2. 我们可以将未截断的崩溃测试用例存储到磁盘？

中。

Answer 1

如果不使用监视器，则可以添加一个post_test_case_callbacks来检查服务器是否处于活动状态。此函数将在每个测试用例之后调用。

post_test_case_callbacks (方法列表)-注册的方法将在每个模糊测试用例之后调用。默认无。

例如：

logger = FuzzLoggerText()
def target_alive(target, fuzz_data_logger, session, sock, *args, **kwargs):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sock.connect((host, port))
        logger.log_pass(description="alive")
    except ConnectionRefusedError:
        logger.log_fail(description="Server down")

session = Session(target=Target(SocketConnection(host, int(port))), post_test_case_callbacks=[target_alive])