1. 项目概述当Pwntools遇见Fuzz提到Pwntools绝大多数安全研究员和CTF选手的第一反应就是“PWN神器”——写EXP、搞ROP链、调交互。这工具在二进制攻防领域几乎是标配。但今天我想聊点不一样的用Pwntools来写一个自动化Fuzz脚本目标是挖掘那些我们日常高频使用的Linux命令行工具里潜藏的漏洞。这个想法源于一次内部安全审计我们团队发现很多看似坚如磐石的coreutils工具比如ls,cat,find或者系统管理工具如tar,awk,sed在面对畸形、超长或精心构造的输入时行为会变得不可预测轻则崩溃重则可能导致权限绕过或信息泄露。手动测试效率太低而一些传统的Fuzz框架如AFL在针对命令行工具的复杂参数组合测试时配置和集成又略显笨重。这时Pwntools这个我们本就熟悉的“老朋友”展现出了惊人的灵活性。它内置的进程控制、管道通信、数据生成和异常状态捕获能力恰好能拼凑出一套轻量、高效且高度可定制的命令行Fuzz解决方案。这篇文章我就来拆解如何利用Pwntools超越其传统的PWN用途构建一个能自动挖掘Linux命令行工具漏洞的Fuzz脚本适合有一定Python和Linux基础希望将安全测试能力扩展到更广泛系统工具的安全从业者、开发者和运维人员。2. 核心思路与架构设计2.1 为什么是Pwntools选择Pwntools而非专门的Fuzzer如AFL、libFuzzer作为核心主要基于几个现实考量。首先场景适配性。命令行工具的Fuzz测试输入向量往往不止是文件内容更包括命令行参数、环境变量、标准输入stdin以及它们之间复杂的组合关系。AFL虽然强大但其主要优化方向是文件输入和代码覆盖率对多输入通道、状态依赖如需要特定前置命令的测试场景配置起来比较曲折。Pwntools的process模块可以无缝地启动子进程、控制其所有I/O通道stdin, stdout, stderr并能方便地设置环境变量和命令行参数这为我们构建复杂的测试用例提供了原子操作能力。其次控制与反馈的精确性。挖掘崩溃Crash或异常行为是Fuzz的核心目标。Pwntools的tube接口recv,sendline,interactive让我们能像与网络服务交互一样与本地进程“对话”精确控制输入时机和内容。更重要的是它能通过process.wait()和信号处理如SIGSEGV,SIGABRT来准确捕获进程的退出状态、返回码以及是否因信号而终止——这是判断一次测试是否触发了潜在漏洞如段错误、断言失败的关键。最后开发效率与可扩展性。Pwntools本身是Python库这意味着我们可以利用Python庞大的生态系统如argparse解析配置、multiprocessing实现并行、logging记录日志快速搭建测试框架。测试逻辑、变异策略、结果去重和报告生成都可以用我们熟悉的Python代码来实现定制化程度极高能够快速适应不同目标工具的测试需求。2.2 Fuzz脚本的整体架构设计一个有效的命令行Fuzz脚本其核心架构可以抽象为以下几个模块目标配置与启动器负责定义要测试的命令行工具路径、基础命令行参数模板、环境变量设置等。例如我们要测试/usr/bin/tar的解压功能基础命令可能是tar -xzf [待测归档文件] -C /tmp/test_dir。测试用例生成器这是Fuzz的“大脑”。它需要根据预设的变异策略生成或修改测试数据。策略可以包括随机字节流最简单的生成随机长度的随机字节。基于字典的变异结合网络热词中提到的“fuzz字典”使用已知的边界值、格式字符串、特殊字符如../,\x00, 超长字符串进行替换或插入。结构感知变异如果目标工具处理特定格式如tar归档头、JSON、XML可以部分解析该格式然后在关键字段如长度字段、类型字段进行有目的的变异。执行与监控引擎使用Pwntools启动目标进程注入生成的测试用例通过命令行参数、文件或stdin并监控其执行状态。关键是要设置超时防止进程挂起比如等待输入密码正如网络片段中提到的sudo测试痛点。结果分析与分类器捕获进程的退出状态、输出stdout/stderr以及可能产生的核心转储core dump。对结果进行分类例如正常退出退出码0、预期内的错误退出如返回码1并打印“invalid argument”、疑似漏洞触发的崩溃如因SIGSEGV信号退出。调度与去重模块管理测试队列实现并行测试以提升效率并对触发的崩溃进行去重避免重复报告相同的漏洞模式。我们的脚本将围绕这几个模块利用Pwntools作为“执行与监控引擎”的核心逐步构建。注意在进行任何Fuzz测试尤其是对系统关键工具进行测试前务必在隔离的虚拟化环境如虚拟机、Docker容器中进行。失控的Fuzz测试可能导致系统不稳定、文件损坏或数据丢失。3. 核心模块实现详解3.1 目标进程的启动与控制这是Pwntools大显身手的地方。我们主要使用pwnlib.tubes.process模块。from pwn import * def run_target(command, input_dataNone, envNone, timeout2): 使用Pwntools启动并控制目标进程。 :param command: 待执行的命令列表如 [/usr/bin/tar, -xzf, test.tar] :param input_data: 通过stdin输入的数据字节串 :param env: 环境变量字典 :param timeout: 进程运行超时时间秒 :return: (returncode, signal, stdout_output, stderr_output) try: # 启动进程 # aslrFalse 在某些调试场景下有用但普通Fuzz可设为True或默认 p process(command, envenv, timeouttimeout) # 如果有stdin输入则发送 if input_data: p.send(input_data) # 发送后关闭stdin告知进程输入结束对于某些工具是必须的 p.shutdown(send) # 等待进程结束并捕获输出 stdout p.recvall(timeouttimeout) stderr p.recvall(timeouttimeout) # 注意对于processstdout和stderr通常合并需要根据目标调整。 # 更精确的做法是使用p.recv()配合条件判断或者使用p.recvrepeat(timeout) # 一个更健壮的版本是 # output b # while p.can_recv(timeout0.1): # try: # output p.recv(timeout0.1) # except EOFError: # break # 获取进程状态 p.wait() returncode p.poll() # pwnlib的process被信号杀死时returncode通常是负数-信号值 # 我们可以将其转换为信号值 if returncode is not None and returncode 0: signal_received -returncode returncode None else: signal_received None p.close() return returncode, signal_received, stdout, stderr except Exception as e: # 处理超时等异常 log.warn(f执行命令 {command} 时发生异常: {e}) # 尝试终止进程 try: p.kill() except: pass return None, None, b, b关键点解析process对象提供了对子进程的完全控制。env参数允许我们设置特定的环境变量这在测试一些依赖环境如LD_PRELOAD的工具时非常有用。send()和shutdown(send)用于向进程的stdin发送数据并关闭写入端模拟用户输入结束。recvall()和wait()的组合用于获取所有输出并等待进程结束。但要注意有些进程可能会分别向stdout和stderr输出上述简单示例可能混合了它们。对于需要严格区分的场景可以考虑使用subprocess.Popen配合pwntools的tube进行更精细的控制或者直接使用pwnlib.tubes.process的stderr管道如果支持。信号处理如果进程因为段错误SIGSEGV信号11或中止SIGABRT信号6而崩溃p.poll()返回的通常是负的信号值。这是我们判断是否触发潜在漏洞的核心依据。3.2 测试用例的生成与变异策略测试用例生成器是Fuzz的创造力来源。我们可以实现一个简单的类来管理多种变异策略。import random import string class FuzzCaseGenerator: def __init__(self, seed_casesNone, dict_pathNone): 初始化生成器。 :param seed_cases: 初始种子用例列表字节串列表 :param dict_path: 外部字典文件路径 self.seed_cases seed_cases or [bnormal_input] self.dictionary [] if dict_path: try: with open(dict_path, rb) as f: # 读取字典每行作为一个条目去除换行符 self.dictionary [line.rstrip(b\n) for line in f] except FileNotFoundError: log.warn(f字典文件 {dict_path} 未找到将使用内置字典。) # 内置一些常见的危险/边界值 self.dictionary.extend([ b../../../../etc/passwd, b%s%n%x%p, # 格式字符串常见payload bA * 1000, # 长字符串 b\x00 * 10, # 空字节 b\xff * 10, # 非ASCII字符 bid, # 命令注入尝试 ]) def mutate(self, data): 对输入数据字节串进行一次随机变异。 if not data: data random.choice(self.seed_cases) mutation_type random.choice([insert, overwrite, delete, repeat, dictionary]) if mutation_type insert: pos random.randint(0, len(data)) new_byte random.randbytes(1) return data[:pos] new_byte data[pos:] elif mutation_type overwrite: if len(data) 0: return data pos random.randint(0, len(data)-1) new_byte random.randbytes(1) return data[:pos] new_byte data[pos1:] elif mutation_type delete: if len(data) 1: return data pos random.randint(0, len(data)-1) return data[:pos] data[pos1:] elif mutation_type repeat: if len(data) 0: return data pos random.randint(0, len(data)-1) repeat_count random.randint(2, 10) return data[:pos] data[pos:pos1] * repeat_count data[pos1:] elif mutation_type dictionary and self.dictionary: dict_entry random.choice(self.dictionary) if random.choice([True, False]): # 替换一段 if len(data) 0: start random.randint(0, len(data)-1) end random.randint(start, len(data)) return data[:start] dict_entry data[end:] else: return dict_entry else: # 插入 pos random.randint(0, len(data)) return data[:pos] dict_entry data[pos:] return data def generate_from_seed(self): 从种子用例中随机选取一个并变异。 seed random.choice(self.seed_cases) # 可以应用多次变异以增加复杂度 for _ in range(random.randint(1, 3)): seed self.mutate(seed) return seed策略选择心得种子质量至关重要。不要只用随机字节开始。最好收集一些目标工具正常处理的有效输入作为初始种子。例如测试tar就用几个正确的小型tar文件作为种子测试grep就用一些简单的文本文件。这能帮助Fuzzer更快地“探索”到程序的深层逻辑路径。字典是加速器。从网络热词可以看到“fuzz字典”的重要性。字典里应该包含目标领域常见的危险模式路径遍历字符串../../../、格式字符串符%n、Shell元字符;、|、、超长字符串、非UTF-8字节序列等。可以结合历史CVE的PoC来丰富字典。变异强度要可控。上述示例的变异是随机的。在实际中可以引入一个“能量”概念对长时间未触发新路径或崩溃的测试用例增加其变异次数和强度进行更激进的测试。3.3 结果捕获与崩溃判定逻辑执行一次测试后我们需要分析结果判断是否发现了有趣的情况。def analyze_result(command, returncode, signal, stdout, stderr, timeout_happenedFalse): 分析单次测试运行结果。 :return: (status, reason) status: normal, expected_error, timeout, crash, hang, interesting # 1. 超时判定 if timeout_happened: # 超时可能是死锁或无限循环值得关注但需排除正常的长时运行 return timeout, Process exceeded timeout limit # 2. 信号导致的崩溃判定 (最可能发现漏洞) if signal is not None: # 常见的崩溃信号 crash_signals { 4: SIGILL, # 非法指令 6: SIGABRT, # 中止信号通常是assert失败或malloc错误 8: SIGFPE, # 浮点异常 11: SIGSEGV, # 段错误 (最常见的内存错误) 7: SIGBUS, # 总线错误 } if signal in crash_signals: return crash, fProcess terminated by signal {signal} ({crash_signals.get(signal, UNKNOWN)}) else: # 其他信号也可能有意义标记为interesting return interesting, fProcess terminated by signal {signal} # 3. 非零返回码 - 可能是预期错误也可能隐藏问题 if returncode ! 0: # 这里需要一些启发式规则或已知的错误模式列表 # 例如某些工具对无效参数返回1内存错误可能返回特定的错误码如139但通常被信号捕获 # 我们可以检查stderr中是否有已知的安全错误信息如“stack smashing detected” error_output stderr.decode(utf-8, errorsignore) stdout.decode(utf-8, errorsignore) security_indicators [ stack smashing, heap corruption, buffer overflow, double free, use after free, segmentation fault, # 有时会打印出来而非发信号 assertion failed, ] for indicator in security_indicators: if indicator in error_output.lower(): return crash, fNon-zero exit ({returncode}) with security indicator: {indicator} # 否则归类为预期错误但可记录以供后续分析 return expected_error, fExit with code {returncode} # 4. 正常退出 return normal, Exited normally with code 0判定逻辑的精细化信号 vs 返回码在Linux中进程如果被信号杀死其退出状态码是128 信号值在Shell中用$?查看。但Pwntools的process.poll()直接给了我们信号值负数或返回码。理解这个区别对正确判定至关重要。“Interesting”结果除了明确的崩溃Crash一些异常行为也值得关注比如程序输出大量乱码到stderr、陷入死循环超时、或产生了非预期的文件操作。这些“有趣”的结果可能指向逻辑漏洞、资源耗尽或条件竞争等问题。去重Deduplication仅仅捕获崩溃是不够的同一个漏洞可能被成千上万个不同的输入触发。简单的去重可以基于崩溃信号和栈回溯哈希。更高级的去重可以使用代码覆盖率信息但这需要插桩AFL做得更好。在我们的脚本中一个初级的去重方法是为每个唯一的(信号, 崩溃地址/函数名)组合只保存第一个触发它的测试用例。获取崩溃地址通常需要调试器如gdb附加这超出了基础脚本的范围但可以作为进阶功能。4. 完整脚本组装与实战演练4.1 构建主循环与并行化我们将上述模块组合起来并加入简单的并行处理以利用多核CPU。import sys import os import time import hashlib from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path def worker(task_id, command_template, generator, output_dir, timeout): 单个测试任务的工作函数。 # 1. 生成测试用例 test_input generator.generate_from_seed() # 为了演示我们假设测试用例通过一个临时文件传递 # 例如测试 cat [file] import tempfile with tempfile.NamedTemporaryFile(modewb, deleteFalse, suffix.fuzz) as tmp: tmp.write(test_input) tmp_path tmp.name # 构造实际命令将占位符替换为临时文件路径 # 假设command_template是一个列表其中包含{INPUT_FILE}占位符 cmd [arg if arg ! {INPUT_FILE} else tmp_path for arg in command_template] # 2. 执行测试 returncode, signal, stdout, stderr run_target(cmd, timeouttimeout) # 3. 分析结果 status, reason analyze_result(cmd, returncode, signal, stdout, stderr) # 4. 清理临时文件 try: os.unlink(tmp_path) except: pass # 5. 记录有趣的结果 if status in [crash, interesting, timeout]: # 生成唯一标识符例如基于信号和输出片段的哈希 crash_hash hashlib.md5(f{signal}:{reason}:{stderr[:200]}.encode()).hexdigest()[:8] crash_dir Path(output_dir) / f{status}_{crash_hash} crash_dir.mkdir(parentsTrue, exist_okTrue) # 保存测试用例、命令和输出 (crash_dir / input.bin).write_bytes(test_input) (crash_dir / command.txt).write_text( .join(cmd)) (crash_dir / stderr.log).write_bytes(stderr) (crash_dir / stdout.log).write_bytes(stdout) (crash_dir / result.txt).write_text(fStatus: {status}\nReason: {reason}\nSignal: {signal}\nReturncode: {returncode}) log.info(f[Worker {task_id}] Found {status}: {reason} (Hash: {crash_hash})) return True, crash_hash else: return False, None def main(): # 配置参数 target_command [/usr/bin/cat, {INPUT_FILE}] # 示例测试cat命令 seed_cases [bHello World\n, bA*100, b\x00\x01\x02, b%s%n] dict_path ./fuzz_dictionary.txt output_dir ./fuzz_results num_workers 4 timeout_per_test 2 # 秒 total_iterations 10000 # 总测试次数 # 初始化 Path(output_dir).mkdir(exist_okTrue) generator FuzzCaseGenerator(seed_casesseed_cases, dict_pathdict_path) seen_crashes set() log.info(fStarting fuzzing for: { .join(target_command)}) log.info(fWorkers: {num_workers}, Timeout: {timeout_per_test}s) # 使用线程池并行执行 with ThreadPoolExecutor(max_workersnum_workers) as executor: future_to_id {} for i in range(total_iterations): future executor.submit(worker, i % num_workers, target_command, generator, output_dir, timeout_per_test) future_to_id[future] i completed 0 for future in as_completed(future_to_id): completed 1 try: found_crash, crash_hash future.result() if found_crash and crash_hash not in seen_crashes: seen_crashes.add(crash_hash) log.success(fUnique crash found! Hash: {crash_hash}) except Exception as e: log.error(fTask failed: {e}) # 简单进度显示 if completed % 100 0: log.info(fProgress: {completed}/{total_iterations}, Unique crashes: {len(seen_crashes)}) log.success(fFuzzing completed. Total tests: {total_iterations}, Unique issues found: {len(seen_crashes)}) if __name__ __main__: context.log_level WARN # 减少Pwntools的调试输出 main()4.2 实战案例挖掘一个简单工具的漏洞假设我们怀疑一个古老的文本处理工具old_processor在处理特定换行符时存在缓冲区溢出。我们编写一个针对性的Fuzz脚本。# 针对 old_processor 的定向Fuzz target_cmd [/usr/local/bin/old_processor, --parse, {INPUT_FILE}] # 种子用例包含正常换行符的文本 seeds [ bline1\nline2\nline3\n, b\r\nline1\r\nline2\r\n, # Windows换行 bline1\n, # 单行 ] # 字典重点变异换行符周围和行长度 dictionary [ b\n, b\r\n, b\r, b\n * 100, # 多个连续换行 bA * 500 b\n, # 超长行后接换行 b\x00\n, # 空字节后换行 ] generator FuzzCaseGenerator(seed_casesseeds, dict_pathNone) generator.dictionary.extend(dictionary) # 调整变异策略增加在换行符附近插入/删除的权重 # 这需要对上面的mutate方法进行扩展此处略运行这个脚本几万次后我们可能会在./fuzz_results/crash_abcd1234目录下发现一个导致SIGSEGV的输入文件。使用gdb加载崩溃的输入进行调试就可能定位到具体的漏洞点。4.3 进阶技巧与调试器联动为了提高效率我们可以让Fuzz脚本在检测到崩溃时自动启动调试器如gdb来捕获回溯轨迹这能极大帮助后续的漏洞分析和去重。def capture_backtrace_on_crash(program, input_file, core_fileNone): 在崩溃时使用gdb自动捕获回溯。 import subprocess gdb_cmds [ run, bt full, # 完整回溯 info registers, x/10i $pc, # 查看崩溃点附近指令 quit ] cmd_input \n.join(gdb_cmds) gdb_command [gdb, --batch, --quiet, program] if core_file: gdb_command.extend([-c, core_file]) else: # 假设崩溃是通过stdin或文件输入触发的这里需要根据实际情况调整 # 一种方法是在gdb中通过run input_file来重定向输入但batch模式处理较复杂 # 更简单的方式是让脚本在崩溃时保存core dump然后分析core文件 pass try: result subprocess.run(gdb_command, inputcmd_input.encode(), capture_outputTrue, timeout10) return result.stdout.decode(utf-8, errorsignore) except subprocess.TimeoutExpired: return [GDB analysis timed out]在worker函数中当判定为crash后可以调用此函数并将输出保存到crash_dir下的backtrace.txt中。注意这需要系统启用core dumpulimit -c unlimited并且程序编译时包含调试信息-g。5. 常见问题、优化方向与避坑指南5.1 常见问题与解决方案目标进程挂起Hang导致Fuzz停滞现象某个测试用例导致目标进程无限循环或等待输入脚本卡在recv()或wait()上。解决务必为每次执行设置合理的timeout参数。在run_target函数中我们使用了Pwntools的timeout参数。一旦超时立即调用p.kill()强制终止进程。将此类超时案例归类为hang或timeout它们可能指向逻辑炸弹或资源死锁漏洞。大量重复的崩溃现象短时间内发现成百上千个崩溃但很多是同一根本原因触发的。解决实现有效的去重。初级方案是基于崩溃信号和栈顶层函数名的哈希通过上述gdb脚本获取。更优的方案是集成简单的覆盖率反馈如使用ptrace或SanitizerCoverage插桩但这会显著增加复杂度。一个折中方案是对崩溃输入进行最小化test case reduction只保存能触发崩溃的最简输入这也有助于去重。Fuzz效率低下现象每秒执行的测试用例数exec/s很低。解决并行化如示例所示使用ThreadPoolExecutor或ProcessPoolExecutor。注意如果目标程序有全局状态如写入同一文件并行可能导致干扰此时需要为每个worker提供独立的工作目录。减少进程启动开销对非常简单的工具进程启动开销可能占大头。考虑使用persistent mode如果目标程序支持让一个进程实例持续运行通过管道或socket反复发送测试数据。这需要目标程序能在处理完一个输入后不退出等待下一个输入。Pwntools的process可以配合sendline和recvuntil来实现类似交互。优化变异策略纯粹的随机变异效率不高。引入覆盖率引导如AFL是最佳方案但实现复杂。可以退而求其次使用基于语法的变异如果知道输入格式或基于历史结果的反馈对产生新路径的输入进行更多变异。误报False Positive现象脚本报告了崩溃但手动验证发现是预期行为如程序自己调用了abort()来提示错误。解决完善analyze_result函数中的启发式规则。例如检查stderr中是否包含“Error: invalid format”等预期错误信息。建立一个“已知安全错误模式”的白名单过滤掉这些情况。同时对每个疑似崩溃必须进行人工复核。5.2 脚本优化与扩展方向结构化输入生成对于处理JSON、XML、特定二进制协议的命令行工具随机变异几乎无效。需要实现一个结构感知的生成器在保持整体语法正确的前提下对关键值进行变异。可以使用库如hypothesis针对Python或自定义语法模板。状态感知Fuzz有些工具的操作依赖于之前命令的状态例如先git init再git add。这就需要维护一个会话状态生成一系列相关的命令序列进行测试。Pwntools的process可以保持连接配合发送多条命令来实现。集成到CI/CD将这个Fuzz脚本作为持续集成的一部分对内部开发或使用的命令行工具进行回归测试。可以设置一个基线任何新的崩溃都被视为回归需要立即调查。结果可视化与报告将发现的崩溃数量、执行速度、代码覆盖率如果实现了等信息实时输出到控制台或生成HTML报告。使用logging模块分级记录日志方便调试和监控。5.3 安全与伦理避坑指南隔离环境再次强调必须在虚拟机或Docker容器中运行Fuzz测试。一个失控的Fuzzer可能填满磁盘、耗光内存或损坏关键系统文件。目标合法性只对你拥有合法测试权限的软件进行Fuzz。对开源软件积极参与负责任的漏洞披露。切勿对未经授权的系统或软件进行测试。资源限制在脚本中设置全局的资源限制如最大运行时间、最大内存使用、最大生成文件大小等防止Fuzzer本身成为拒绝服务攻击的来源。关注副作用有些崩溃可能不会立即导致进程终止但会引发内存泄漏、文件描述符泄漏或状态不一致。监控系统资源如ps,lsof也是一个好习惯。用Pwntools写Fuzz脚本本质上是将我们熟悉的二进制利用工具链的思路应用到了更广泛的软件质量与安全测试领域。它可能没有专业Fuzzer那么“聪明”但其灵活性、可控性和与现有Python技能栈的无缝衔接使得它成为进行快速原型验证、针对性测试或集成到自定义工作流中的一把利器。下次当你对某个命令行工具的行为心存疑虑时不妨花上半小时用这个思路写个小脚本去“问候”一下它或许会有意想不到的发现。