pocsuite3安全工具源码分析

pocsuite3 是由 知道创宇 404实验室 开发维护的开源长途漏洞测试和概念验证开发框架。为了更好理解其运行逻辑，本文将从源码角度分析该项目的初始化，多线程函数，poc模板等等源码。
项目布局

[img=720,858.4615384615385]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711834.png[/img]

api：对要导入的包重命名，方便后续导入调用data：存储用户必要使用的文档数据lib：项目核心代码modules：存储用户自定义的模块plugins：存储用户自定义的插件pocs：存储poc文件shellcodes：存储生成php，java，python等脚本语言的利用代码，以及反弹shell的利用代码cli.py：项目的入口console.py：命令行界面
进入项目入口：/pocsuite3/cli.py

[img=720,350.72]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711836.png[/img]

check_environment() #查抄当前工作目次是否符合当前体系set_paths(). #设置后续必要用到的数据，目次信息banner() #打印命令行页面的横幅
init_options(cmd_line_parser().dict) # 命令行参数处置惩罚跟进cmd_line_parser()查看：

[img=720,466.6350710900474]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711837.png[/img]

此处注意一个参数-c

1. target.add_argument("-c", dest="configFile", help="Load options
2. from a configuration INI file")

复制代码

可以先在pocsuite.ini配置好参数，通过pocsuite -c pocsuite.ini 运行

[img=720,499.87261146496814]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711838.png[/img]

双重跟进init_options()，找到命令行存储参数：

[img=720,550.7311586051743]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711839.png[/img]

可见接纳了类似字典的形式存储，避免了重复数据且另有其它四个参数也接纳了该形式存储，五个参数贯穿整个项目
【----帮助网安学习，以下所有学习资料免费领！加vx：yj520400，备注 “博客园” 获取！】
　① 网安学习发展路径思维导图
　② 60+网安经典常用工具包
　③ 100+SRC漏洞分析报告
　④ 150+网安攻防实战技能电子书
　⑤ 最权势巨子CISSP 认证考试指南+题库
　⑥ 超1800页CTF实战本领手册
　⑦ 最新网安大厂面试题合集（含答案）
　⑧ APP客户端安全检测指南（安卓+IOS）
conf：存储基本配置信息kb：存储了目标地址、加载的PoC、运行模式、输出结果、加载的PoC文件地址、多线程信息等cmd_line_options：是存储命令行输入的参数值merged_options：存储输入值与默认值合并后的结果paths：存储数据、插件、poc等目次地址
参数获取处置惩罚完后，进入项目初始化，init()函数，一下对部分函数进行注解分析：

1. def init():
2. ​
3. """
4. ​
5. Set attributes into both configuration and knowledge base singletons
6. ​
7. based upon command line and configuration file options.
8. ​
9. """
10. ​
11. set_verbosity() #日志输出级别设置
12. ​
13. _adjust_logging_formatter() #调整日志格式器
14. ​
15. _cleanup_options() #将各个配置项格式化，并校验合法性
16. ​
17. _basic_option_validation() #校验seebug,zoomeye等api,token的合法性
18. ​
19. _create_directory() #检测文件路径是否存在，不存在则创建
20. ​
21. _init_kb_comparison()
22. ​
23. update()
24. ​
25. _set_multiple_targets() #读取目标
26. ​
27. _set_user_pocs_path()
28. ​
29. _set_pocs_modules() #动态加载poc
30. ​
31. _set_plugins() #动态加载插件
32. ​
33. _init_targets_plugins()
34. ​
35. _init_pocs_plugins()
36. ​
37. _set_task_queue() #初始化多线程设置
38. ​
39. _init_results_plugins() #初始化输出插件

复制代码

AttribDict类解析

前文也提到过以下五个全局变量，它们均通过创建AttribDict类的实例进行使用，现在我们跟进类详细分析：

AttribDict()类：

[img=720,277.39376770538246]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711841.png[/img]

自定义类，继续自python内建的OrderedDict类，扩展访问方式，简化了对字典键的访问。重要存在三个方法：getattr(),setattr(),delattr()这三个方法在if判定逻辑均相同：1:以双下划线 __ 开头（例如，Python 的内置属性，如 dict）。2:以 _OrderedDict__ 开头（由于 OrderedDict在内部实现中使用的名称）。3:名字存在于 exclude_keys 集合中（排除的键）。假如任一条件成立，说明这个属性不应该通过 obj.attr访问，所以跳过使用自定义的 getattr处置惩罚，直接调用父类对应的方法访问。例：getattr()就调用父类的getattribute()访问
假如属性名不满足，则通过字典的方式，添加大概删除AttribDict中

[img=720,339.2481203007519]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711842.png[/img]

地址处置惩罚代码分析

[img=720,345.97402597402595]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711843.png[/img]

先查看存储初始数据，存在则进行下一步。通过set()创建集合方便去重，再遍历conf.url数据，通过parde_target()进行对url进行分析处置惩罚，并且在不为空的情况下调用集合的add()方法添加，完成后再将，用于暂时存储的target集合里面的数据，放到kb这种全局变量内。parde_target()函数

[img=720,481.2483745123537]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711844.png[/img]

接受参数后先if判定，假如是域名，url，ip:端口形式则直接赋值给target跟进其中一个判定函数：

[img=720,100.92050209205021]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711845.png[/img]

跟进：

[img=720,75.33888228299644]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711846.png[/img]

可见是通过正则进行判定。接着再判定假如为http://ipv6形式，则启动ipv6配置，并进行赋值target，依旧是正则判定。
再判定假如为ipv4则调用python内置ip_address解析赋值，该方法自动区分ipv4大概ipv6并末了返回对应的对象。再通过else判定，对纯ipv6地址，大概ipv6网络进行解析赋值。
动态poc加载

[img=720,378.90410958904107]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711847.png[/img]

Step1：从pocs目次加载先通过os.listdir读取对应目次，返回一个含有poc的py文件的列表。再通过filter()函数过滤init.之类文件，不外此时filter()函数返回的是一个迭代器，所以又通过list()函数将数据处置惩罚成列表再赋值。（lambda x: x not in ['init.py','init.pyc']：这个匿名函数会查抄每个文件名 x 是否不等于'init.py' 或 'init.pyc'。）
再从含有类似thinkphp_poc.py的文件名中，通过x变量循环读取，并通过splitex()函数将其分为"thinkphp_poc",".py"格式的键值队元组。再次通过dict()字典函数，将x元组的第一个元素作为字典的键，第二个元素作为字典的值。
假如poc是目次，则使用 os.walk() 递归遍历该目次下的所有文件，过滤出 .py或 .yaml 文件，并将其完整路径添加到 _pocs 列表中。
Step2：遍历加载 PoC 文件内容并查抄，并对加载失败的poc进行日志记录。
Step3：末了从 Seebug 网站加载 PoC。
poc模版跟据目次找到现存poc：pocsuite3/pocs，thinkphp_rce为例

所有模版均是继续自父类POCBase，跟进：

[img=720,260.1869158878505]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711849.png[/img]

父类在初始化时便设置了一系列可能用到的属性，例如自定义headers，目标url，端口等等。这里关注execute()函数

[img=720,209.22440537745604]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711850.png[/img]

self.url处接纳if判定：假如为http协议则接纳parse_target_url()解析，else接纳build_url()解析：mode值默以为verify。随后调用_execute()根据mode值执行。

[img=720,373.55648535564853]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711851.png[/img]

shell()，attack()，_verify()均需自定义重写。回到例thinkphp_rce例子：_verify()函数如下：

[img=720,306.3498920086393]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711852.png[/img]

调用了_check()函数进行检验：

通过request.post()发送设置好payload的哀求，根据返回包关键字判定是否乐成。（flag自定义）返回的结果在_verify()函数又会调用parse_output()转化为json格式输出。
动态核心load_file_to_module()继续分析_set_pocs_modules()

[img=720,220.16]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711854.png[/img]

将读取文件切割为文件名和后缀名，根据后缀名重构路径file_pth，if判定file_path构建乐成则进入红框代码处。

[img=720,260.48]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711855.png[/img]

通过get_filename()从file_path路径提取文件名，由于wuth.ext=False，则不提取文件名后缀，提取后拼接在pocs_后并赋值给module，例如：pocs_thinkphp_rce。随后三行代码涉及到python中动态模块加载知识：

1. spec = importlib.util.spec_from_file_location(module_name, file_path,
2. loader=PocLoader(module_name, file_path))

复制代码

#创建模块规格，接纳自定义加载器类加载模块，loader:加载器对象，负责如何从文件加载模块
mod = importlib.util.module_from_spec(spec)#根据规格创建模块对象
spec.loader.exec_module(mod) #执行模块代码，确保为完整可用的模块
动态模块注解：
模块是包含 Python 代码的文件，可以通过 import语句加载并使用。通常，当你使用 import 语句导入一个模块时，Python会根据模块的名称查找相应的文件（如 .py 文件），并将其加载到内存中。
然而，在一些特殊的情况下，好比动态加载模块或运行时创建模块，我们必要用到importlib 模块。importlib提供了一些工具，可以帮助我们在运行时加载模块，而不是在编写代码时静态地导入。
例如：importlib.util.spec_from_file_location
spec（模块加载规格）描述了如何加载一个模块。它定义了如何找到模块代码，如何加载它，以及加载时必要的一些元数据。类似于说明书，它告诉Python 模块在哪里、叫什么名字、以及如何加载它。
接着看看是如何调用loader加载器的exec_module()函数进行加载的：

filename接受poc绝对路径，poc_code接受poc文件内容。随后调用check_requires()查抄代码运行中必要的包，通过import函数导入。compile()为python内置函数，将源代码字符串poc_code编译为字节码，'exec'这是一个编译模式，表示代码将作为一段可执行的代码被执行。常见的编译模式有'eval'（用于单个表达式）和 'exec'（用于整个代码块）之后再调用exec()函数执行字节码对象obj当中的代码，并绑定到module.dict上，这样就可以通过module.函数()直接调用poc_code当中的函数。
多线程与输出加载

跟进：_set_task_queue()

[img=720,180.33898305084745]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711857.png[/img]

if判定，poc模版与目标ip均不为空情况下，遍历出poc_module与target。并将它们组成元组，加入kb.task_queue中，确保数据在线程安全传输。
start()函数

[img=720,362.89156626506025]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711858.png[/img]

调用runtime_check()查抄poc是否加载乐成：

[img=720,171.78947368421052]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711859.png[/img]

再调用python尺度库中的queue.Queue类的qsize()方法，获取先前kb.task_queue队列的任务数目。run_threads()函数随后进入start()函数核心：run_threads(conf.threads, task_run)：该函数传入线程数conf.threads()，与多线程执行函数task_run()。

[img=720,88.55172413793103]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711860.png[/img]

这个函数的目的是启动多个线程并执行给定的函数thread_function。num_threads: 必要启动的线程数目。thread_function: 要在线程中运行的目标函数。args: 通报给 thread_function 的参数，默以为空元组。forward_exception: 控制是否在捕捉异常后继续传播异常，默认值为 True。start_msg: 控制是否输出启动线程的消息，默认值为 True。
先threads = []创建空列表，用来存储后续的线程实例
随后进行线程数查抄，假如大于1，则是多线程，并在线程数超过max时发出告警提示，线程不大于1，则直接执行函数

[img=720,297.413073713491]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711861.png[/img]

查抄完为多线程则进行下一步：循环创建线程，并启动

[img=720,241.92]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711862.png[/img]

根据num_threads数目循环创建，并调用setDaemon(TRUE)将所有线程设置为保卫线程。（保卫线程：后台运行，随主线程终止而终止)

[img=720,259.2563600782779]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711863.png[/img]

随后再调用python尺度库函数isAlive()进行循环查抄，直到所有线程完成才跳出循环。(python3发起使用is_Alive()函数)。

[img=720,169.32203389830508]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711864.png[/img]

执行完run_threads()函数后，finally代码再执行task_done()，跟进该函数，内部存在三个函数：

show_task_result()：会取出poc执行结果，然后格式化输出

[img=720,394.24]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711866.png[/img]

result_plugins_start():该函数负责调用file_record.py中的start()函数

[img=720,230.51975051975052]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711867.png[/img]

[img=720,124.71428571428571]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711868.png[/img]

result_compare_handle():显示来自各个搜索引擎的对比数据

[img=720,277.037037037037]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711869.png[/img]

先前已经分析了start(0函数核心在于run_threads(conf.threads,task_run)，我们接着跟进分析多线程执行函数：task_run()
多线程执行函数：
task_run():

[img=720,169.5483870967742]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711870.png[/img]

先确认task_queue不为空，并且thread_continue为真，随后从task_queue获取目标ip与poc模版
（之前通过task_queue.put((target,poc_module))存储进去的）
随后调用python尺度库copy模块中的deepcpy，进行深拷贝操纵，复制poc模版，防止原始poc模块被修改。
poc_name获取poc模块名称方便日志打印。
随后处置惩罚用户自定义参数，查抄是否尝试修改白名单内容，并校验是否存在必选参数未设置。

[img=720,486.2411347517731]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711871.png[/img]

随后进入核心代码块，根据传参调用excute()函数：

[img=720,106.18296529968454]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711872.png[/img]

后续则是根据测试乐成大概失败，对结果进行处置惩罚输出

[img=720,201.32701421800948]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711873.png[/img]

综合文章分析，pocsuite3项目被我分成如下执行流程：

[img=720,353.7931034482759]https://m-1254331109.cos.ap-guangzhou.myqcloud.com/202502171711874.png[/img]

在clip.py中调用main()函数，整个项目则开始执行，进行环境查抄，参数获取后，则进入核心代码：在main()函数中调用init()与start()函数，末了则是我上文刚分析过的数据处置惩罚与输出格式化。
更多网安技能的在线实操练习，请点击这里>>
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。