系统架构 - 灰盒扫描工具

背景

IAST（交互式扫描）技术是一种实时动态交互的漏洞检测技术，通过在服务端部署agent程序，收集、监控Web应用程序运行时函数执行、数据传输，并与扫描器端进行实时交互，高效、准确的识别安全缺陷及漏洞。目前OpenRASP项目已实现相当于IAST agent端的OpenRASP agent，在此基础上引入一个扫描端，即可实现一个完整的IAST扫描工具。

系统架构

灰盒扫描工具采用Python3实现，数据库采用MySQL，通讯采用HTTP+JSON的方式。整体架构图如下：iast-main工具分为扫描器端和agent端，Agent端用于收集web应用的运行信息，扫描器端用于处理插件产生的请求信息，并完成整个IAST扫描逻辑

agent端

由一个单独的OpenRASP插件构成，用于提取http请求中hook点产生的信息，并通过HTTP协议以JSON形式发送至扫描器端。

扫描器端

该部分是一个独立运行的扫描工具，借助从OpenRASP插件部分获取的信息完成扫描任务。

  • 功能设计和模块划分

    扫描模块包括三个模块：预处理模块（Preprocessor）、扫描模块（Scanner）、监控模块（Monitor）

预处理模块即图中HTTPServer部分，用于接收agent插件的http请求，处理、存储、分发http请求信息

扫描模块用于运行扫描插件，执行漏洞扫描逻辑

监控模块用于定期获取其他模块的运行时信息，调整参数，提供控制台的HTTP服务等

  • 运行流程示例

    一个典型扫描流程，以SQL注入的扫描过程为例：

    1、运行扫描器端，初始化所有模块

    2、测试人员发送了一条HTTP请求到web server，产生请求及其对应的HOOK信息被OpenRASP插件获取，发送至http_server

    3、http_server发现请求不是扫描器发出的，对其进行去重后写入数据库

    4、扫描模块从数据库获取一条HOOK信息，下发到所有扫描插件

    5、sql注入扫描插件分析HOOK信息发现用户输入参数拼接进了sql查询，运行对应扫描逻辑

    6、扫描插件生成扫描请求，把原始请求进入query的输入参数替换为单引号

    7、扫描插件在请求头添加用于识别扫描请求的scan_request_id，发送扫描请求给web server

    10、web server处理请求并返回结果，扫描插件获得http response，同时OpenRASP插件获取到请求hook信息，发送至http_server

    11、http_server发现请求是扫描器发出的，将其写入rasp_result_queue队列

    12、扫描模块读取rasp_result_queue队列，将rasp_result传给对应的扫描插件

    13、扫描插件检查收到的hook信息，发现query逻辑被改变，认为存在SQL注入，将漏洞信息写入数据库

插件开发

获取代码

在安装了python3.7的环境中可以直接运行IAST

  1. git clone https://github.com/baidu-security/openrasp-iast.git
  2. cd openrasp-iast
  3. python3 openrasp_iast/main.py start -f

运行参数与pip安装相同

扫描插件

所有扫描插件均位于plugin/scanner目录下，所有扩展名为.py的文件都会在启动扫描器时被加载，ScanPluginBase.py为插件基础类，提供编写插件所需的一系列接口，具体参考接口文档

这里以编写一个简单的sql注入fuzz插件为例，介绍插件的编写方法：

首先在plugin/scanner目录下新建一个new_plugin.py文件，粘贴以下内容：

  1. #!/usr/bin/env python3
  2. # -*- coding: UTF-8 -*-
  4. # 导入scanner插件基类
  5. class ScanPlugin(scan_plugin_base.ScanPluginBase):
  7.     plugin_info = {
  8.         "name": "name",
  9.         "show_name": "plugin",
  10.         "description": "des"
  11.     }
  13.     # 用于生成测试向量
  14.     def mutant(self, rasp_result_ins):
  15.         pass
  17.     # 用于检测测试结果
  18.     def check(self, request_data_list):
  19.         pass

在plugin_info中填写插件基本信息：

  1. plugin_info = {
  2.     "name": "sql_basic", # 插件文件名去除扩展名.py后的部分
  3.     "show_name": "SQL注入检测插件", # 插件在后台显示的名称
  4.     "description": "基础sql注入漏洞检测插件" # 插件功能描述
  5. }

实现mutant函数，生成扫描向量

  1. def mutant(self, rasp_result_ins):
  2.     # 首先判断是否需要扫描
  3.     if not rasp_result_ins.has_hook_type("sql"):
  4.         return
  6.     payload_list = [("1'openrasp", "1'openrasp"),
  7.                     ("1\"openrasp", "1\"openrasp"),
  8.                     ("`a openrasp", "`a openrasp")]
  10.     # 获取所有待测试参数
  11.     request_data_ins = self.new_request_data(rasp_result_ins)
  12.     test_params = self.mutant_helper.get_params_list(request_data_ins, ["get", "post", "json", "headers", "cookies"])
  14.     # 生成测试向量的逻辑，返回必须是一个iterable
  15.     for param in test_params:
  16.         # 只测试包含sql类型hook的请求
  17.         if not request_data_ins.is_param_concat_in_hook("sql", param["value"]):
  18.             continue
  19.             # 每个测试点(参数)生成一个payload_seq，防止重复报警，含有相同payload_seq的测试请求仅保留产生的第一个报警
  20.             payload_seq = self.gen_payload_seq()
  21.             for payload in payload_list:
  22.                 # 基于RaspResult类的实例生成测试请求RequestData类的实例
  23.                 request_data_ins = self.new_request_data(rasp_result_ins, payload_seq, payload[1])
  24.                 request_data_ins.set_param(param["type"], param["name"], payload[0])
  25.                 request_data_list = [request_data_ins]
  26.                 # 每次迭代返回的应该是一个由RequestData类的实例组成的list, 该list中的每个RequestData实例都会被作为测试请求依次发送
  27.                 yield request_data_list

每个被发送并获取结果的请求序列都会回调check函数，在check函数中来实现漏洞判定：

  1. def check(self, request_data_list):
  2.     # 当前插件每个请求序列仅包含1个请求，取[0]
  3.     request_data_ins = request_data_list[0]
  4.     # 获取检测特征和请求结果
  5.     feature = request_data_ins.get_payload_info()["feature"]
  6.     rasp_result_ins = request_data_ins.get_rasp_result()
  7.     # 检测是否触发sql注入，直接使用checker检测
  8.     if self.checker.check_concat_in_hook(rasp_result_ins, "sql" , feature):
  10.         # 存在漏洞，返回漏洞描述字符串
  11.         return "sql语句逻辑可被用户输入控制"
  12.     else:
  13.         # 不存在漏洞，返回None
  14.         return None

如果自行实现检测函数，需要在检测到漏洞时，使用set_vuln_hook标记有漏洞的hook点信息:

  1. for hook_item in rasp_result_ins.get_hook_info():
  2.     if has_vuln(hook_item):
  3.         rasp_result_ins.set_vuln_hook(hook_item)

需要注意的是，扫描模块是基于asyncio实现的异步扫描，每个插件会以一个coroutine的形式运行，因此应避免在插件中使用同步的io操作以免影响扫描性能。在进行异常捕获时，如果要捕获全部异常，应单独处理asyncio.CancelledError异常，以免扫描任务无法被终止：

  1. import asyncio
  3. try:
  4.     # do something...
  5.     except asyncio.CancelledError as e:
  6.         raise e
  7.     except Exception as e:
  8.         # do something...
去重插件

所有扫描插件均位于plugin/deduplicate目录下，IAST启动时会加载core/config.py中指定的去重插件，去重插件是为了避免同一个请求被反复扫描，应当根据扫描目标的url结构、参数等特性来编写，DedupPluginBase.py为插件基础类，提供编写插件所需的一系列接口，具体参考接口文档

  • 编写一个去重插件

    在plugin/deduplicate目录下新建一个new_plugin.py文件，写入以下内容：

  1.   #!/usr/bin/env python3
  2.   # -*- coding: UTF-8 -*-
  4.   # 导入去重插件基类
  5.   from plugin.deduplicate import DedupPluginBase
  7.   # 插件类必须命名为dedupPlugin，并继承自DedupPluginBase.DedupPluginBase
  8.   class dedupPlugin(DedupPluginBase.DedupPluginBase):
  9.       # 实现get_hash函数,返回hash字符串
  10.       def get_hash(self, rasp_result_ins):
  11.           return self.get_hash_default(rasp_result_ins)

去重插件只需要实现一个get_hash方法即可，该方法传入当前待去重请求对应的raspResult类的实例，返回一个hash字符串，返回hash相同的请求会被视为重复请求，这里直接调用了默认的get_hash_default方法，其具体实现如下：

  1.   import hashlib
  3.   # 计算urlpath、参数、堆栈等数据的hash，连接后生成结果hash
  4.   def get_hash_default(self, rasp_result_ins):
  5.           path_str = rasp_result_ins.get_path()
  6.           stack_hash = rasp_result_ins.get_all_stack_hash()
  7.           param_keys = "".join(rasp_result_ins.get_parameters().keys())
  8.           query_keys = "".join(rasp_result_ins.get_query_parameters().keys())
  9.           json_struct = rasp_result_ins.get_json_struct()
  10.           contact_str = "".join([path_str, stack_hash, param_keys, json_struct, query_keys]).encode("utf-8")
  11.           return hashlib.md5(contact_str).hexdigest()

注意：如果插件运行中抛出异常，会自动调用get_hash_default函数生成默认的hash