系统架构 - 灰盒扫描工具

背景

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

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

运行参数与pip安装相同

扫描插件

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

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

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

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-

# 导入scanner插件基类
class ScanPlugin(scan_plugin_base.ScanPluginBase):

    plugin_info = {
        "name": "name",
        "show_name": "plugin",
        "description": "des"
    }

    # 用于生成测试向量
    def mutant(self, rasp_result_ins):
        pass

    # 用于检测测试结果
    def check(self, request_data_list):
        pass

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

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

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

def mutant(self, rasp_result_ins):
    # 首先判断是否需要扫描
    if not rasp_result_ins.has_hook_type("sql"):
        return

    payload_list = [("1'openrasp", "1'openrasp"),
                    ("1\"openrasp", "1\"openrasp"),
                    ("`a openrasp", "`a openrasp")]

    # 获取所有待测试参数
    request_data_ins = self.new_request_data(rasp_result_ins)
    test_params = self.mutant_helper.get_params_list(request_data_ins, ["get", "post", "json", "headers", "cookies"])

    # 生成测试向量的逻辑,返回必须是一个iterable
    for param in test_params:
        # 只测试包含sql类型hook的请求
        if not request_data_ins.is_param_concat_in_hook("sql", param["value"]):
            continue
            # 每个测试点(参数)生成一个payload_seq,防止重复报警,含有相同payload_seq的测试请求仅保留产生的第一个报警
            payload_seq = self.gen_payload_seq()
            for payload in payload_list:
                # 基于RaspResult类的实例生成测试请求RequestData类的实例
                request_data_ins = self.new_request_data(rasp_result_ins, payload_seq, payload[1])
                request_data_ins.set_param(param["type"], param["name"], payload[0])
                request_data_list = [request_data_ins]
                # 每次迭代返回的应该是一个由RequestData类的实例组成的list, 该list中的每个RequestData实例都会被作为测试请求依次发送
                yield request_data_list

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

def check(self, request_data_list):
    # 当前插件每个请求序列仅包含1个请求,取[0]
    request_data_ins = request_data_list[0]
    # 获取检测特征和请求结果
    feature = request_data_ins.get_payload_info()["feature"]
    rasp_result_ins = request_data_ins.get_rasp_result()
    # 检测是否触发sql注入,直接使用checker检测
    if self.checker.check_concat_in_hook(rasp_result_ins, "sql" , feature):

        # 存在漏洞,返回漏洞描述字符串
        return "sql语句逻辑可被用户输入控制"
    else:
        # 不存在漏洞,返回None
        return None

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

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

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

import asyncio

try:
    # do something...
    except asyncio.CancelledError as e:
        raise e
    except Exception as e:
        # do something...
去重插件

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

  • 编写一个去重插件

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

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

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

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

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