Home 数据分析 CEP引擎Esper在入侵检测系统中的实践

CEP引擎Esper在入侵检测系统中的实践

by zinan

1、名词

CEP(Complex Event Processing):复杂事件处理、复合事件处理

EPL(Event Processing Language):SQL-LIKE的事件处理语言,用于描述CEP任务

2、CEP引擎

离线引擎

待补充。

实时引擎

Esper:

优点:更新活跃,轻量级嵌入式引擎,规则验证方便,本地导入包后,即可运行。官方提供的文档和case study很全面,支持标准SQL语法,唯一编译型CEP引擎,吞吐量很大单机600W/S。

缺点:分布式需要自己实现,EPL编写没有专门的IDE。

Fink CEP:

优点:分布式部署。

缺点:规则验证不方便,需要上线后才可验证,据说有性能问题。

Siddhi:

优点:轻量级,有web接口,语法编写有错误提示。

缺点:据说句法使用上不够灵活,功能支持不如Esper多,官方文档描述不够详细。

3、数据流图

选择采用Storm + Esper模式提供CEP能力,Storm拓扑中的大致数据流图如下:

4、关于Storm中的数据分组

在Storm中,假如将嵌入了Esper引擎的bolt设置为500并发,以此来增大数据处理效率,而在Storm底层,实际是启动了500个Java虚拟机,每个虚拟机跑一个搭载了Esper引擎的bolt,这样就出现了一个问题,不同Esper引擎之间的数据是隔离的,上游的数据在传递给Esper Bolts的时候,会分散到不同的Esper引擎中,例如A日志给了A引擎,B数据给了B引擎,那么就无法将A和B进行关联分析,基于这样的问题,必须合理的规划分组方式,将需要关联分析的数据,全部发送给同一个bolt。

这里的Esper Bolts接收到了5种类型的日志,分别是:
a.终端EDR相关的原始日志

b.Suricata相关的原始日志

c.Sigma告警

d.Suricata告警

e.Esper引擎自己产生的CEP告警(CEP告警重入Esper引擎是为了方便告警级联分析)

a、b作为原始日志采用默认分组方式『field grouping』,例如suricata的tcp flow日志可以选择使用src_ip字段作为grouping,sysmon日志可以选择computer_name字段作为grouping。但这依然不能满足一些检测场景,例如需要聚合dest_ip字段,因为同一个dest_ip可能被分到了不同的Esper引擎,所以聚合的结果一定是不准确的,解决方案后文介绍。

b、c、d、e作为告警日志,采用了三种不同的分组方式:

Field Grouping:提取当前告警中指定的字段值,作为grouping条件。

One Grouping:将所有one grouping的告警都发送到同一个Esper引擎中。

All Grouping:将告警发送给每一个Esper引擎。

由于每个告警都需要对应的规则,那么在规则定义阶段,也可以指定好告警产生时的grouping方式,例如在一个Sigma规则中,可以如下定义:

# 告警产生后,在Esper引擎中的分组方式,不加此字段则会同时进行2种默认分组方式:
# 1、通过computer_name字段进行field grouping
# 2、one grouping方式,把所有告警发送到同一个bolt中,如果log_level为0则默认不进行one grouping
# 自定义分组方式:
# 1、all_grouping,当前告警发送给每一个bolt节点
# 2、one_grouping,当前告警发送给同一个节点
# 3、define_one_grouping,用户自定义one_grouping字段,后接一个随机HASH值,例如:『define_one_grouping:12345』
# 4、{field_name},直接填写告警中的字段名,则会提取这个字段值作为grouping id
# 示例:
# storm_grouping_field:
#    - 'event_data.IpAddress'       # 通过告警日志中的event_data.IpAddress字段来分组
#    - 'one_grouping'               # 通过one_grouping方式分组
#    - 'all_grouping'               # 通过all_grouping方式分组
#    - 'define_one_grouping:12345'  # 通过one_grouping方式分组,但是自定义分组ID为12345,
#                                   # 相同的ID会分组到同一个bolt,不同的ID可能会分组到不同的bolt
storm_grouping_field:
    - 'event_data.IpAddress'
    - 'one_grouping'

这里使用了两种分组方式,『field grouping』和『one grouping』,也就是说告警进入Esper引擎时,不但会以『event_data.IpAddress』字段的值分组,还会以『one grouping』方式分组。

而对于Suricata告警,由于Snort不方便做如此详细的规则定义方式,所以需要额外的维护一个告警映射表,例如当Suricata告警产生后,给它添加一些附加信息,告警描述,威胁等级,引用链接,告警分类,标签等等,当然还有grouping方式,一个精简的示例模板如下:

# 匹配Suricata的规则名,支持模糊匹配算法
signature: 'SCWX * Privilege Escalation Attempt *CVE-2020-1472*'

# 告警描述
description: '源主机尝试探测了目标主机是否存在CVE-2020-1472(ZeroLogon)漏洞。'

# 一级分类
category_1: 漏洞利用

#二级分类
category_2: CVE漏洞

# 相关参考链接
references:
   - https://cert.360.cn/warning/detail?id=ead0801e639052498d4bf396e08ef775
   - https://www.freebuf.com/articles/system/249860.html

# 告警标签
tags:
   - ZeroLogon

# 当前mapping的匹配优先级,如果同一个告警命中了多个mapping,优先使用precedence高的mapping,无此字段默认为0
precedence: 100


# 告警产生后,在Esper引擎中的分组方式,不加此字段则会同时进行2种默认分组方式:
# 1、通过attacker_ip字段进行field grouping
# 2、one grouping方式,把所有告警发送到同一个bolt中,如果log_level为0则默认不进行one grouping
# 自定义分组方式:
# 1、all_grouping,当前告警发送给每一个bolt节点
# 2、one_grouping,当前告警发送给同一个节点
# 3、define_one_grouping,用户自定义one_grouping字段,后接一个随机HASH值,例如:『define_one_grouping:12345』
# 4、{field_name},直接填写告警中的字段名,则会提取这个字段值作为grouping id
# 示例:
# storm_grouping_field:
#    - 'http.hostname'              # 通过告警日志中的http.hostname字段来分组
#    - 'one_grouping'               # 通过one_grouping方式分组
#    - 'all_grouping'               # 通过all_grouping方式分组
#    - 'define_one_grouping:12345'  # 通过one_grouping方式分组,但是自定义分组ID为12345,
#                                   # 相同的ID会分组到同一个bolt,不同的ID可能会分组到不同的bolt
storm_grouping_field:
    - 'attacker_ip'

已知了各种告警的分组方式,就可以选择合适的字段和分组方法,让多个需要关联的日志,发送到同一个Esper中。但假如非要指定原始日志的grouping方式,也可以使用变通的方式,先通过规则将需要关注的原始日志筛选出来,然后设置告警为内联告警,内联告警仅仅用于系统内部级联分析,不会入库和发出:

# 规则产生的告警日志等级,可为,0:系统内部级联告警,不输出,1:一级告警(普通告警),2:二级告警(运营告警),3:三级告警(应急告警)。
log_level: 3

然后再到Esper中写规则和分析这些被筛选出的日志即可。

5、实时Sigma引擎

Sigma规则基于yaml语言编写,学习成本很低,十分适用于基于文本特征和简单逻辑判断的日志检测场景,可以用它去做易于提取到强特征的攻击检测,也可以用它初筛原始日志后,做内部级联检测。

详情可看Wiki介绍:https://github.com/Neo23x0/sigma

但作者发布的Sigma引擎是用Python编写,基于ES,Splunk等平台离线计算检测原始日志,而且仅支持模糊匹配算法,不支持正则,这样存在两个问题:

1)离线计算实时性不高,不能达到微秒级检测,而且可能存在大量重复计算,比较浪费资源,再加上ES适用场景为小数据集,如果在海量数据面前,会存在性能以及稳定性的问题。例如TPS在50w以上,原始Sigma引擎这种离线计算的模式更是难以稳定和快速的完成检测。

2)模糊匹配算法在实际使用过程中,优点是易于理解和编写,缺点是不容易精细化的匹配特征,容易造成误报的情况。

基于以上问题,需要按照原始的Sigma规则逻辑,用Java重写整个引擎,使其可以运行在Storm中,并且增加支持正则的功能。

6、Esper规则模板

一个Esper规则由两部分构成:

1)用于描述检测逻辑的EPL语句。

2)用于处理通过EPL语句筛选出日志的Listener。

由于每个规则通过EPL筛选出来的日志可能不一样,所以Listener也不能有相同的处理流程,而在官方给出的Case Study中,Listener则全部由Java代码编写,虽然满足了灵活处理日志的能力,但在实践中却有诸多无法忍受的缺陷:

1)规则开发时间长,毕竟写代码本身就麻烦,后续还有经过线下规则验证,线上规则验证,不断调试代码,修改检测特征,总体下来耗时很长,规则无法快速添加和更新。

2)规则编写门槛较高,非Java开发人员无法参与规则编写。

3)规则不支持热更新,每一次规则的上线和更新,都需要重启线上Storm拓扑,导致监控中断,甚至在重启过程中可能导致日志丢失。

解决方法是将Esper编写规则的方式模板化,通过配置文件的方式来描述Esper规则,并且通过Java的反射机制来控制Listener的处理流程,可以继续使用YAML来定义整个Esper规则,以下简单描述Esper规则模板的关键构成。
首先定义EPL查询语句:

# EPL语句,多条EPL语句需要『;』符号分隔,并且两个换行符。
epl: '
@name("可疑进程通过1个单向匿名管道执行CMD命令_无周期性心跳_Sysmon_查询同时有两个行为的进程")
select * from pattern [
    every-distinct(
            a.computer_name, a.event_data_sourceip, a.event_data_destinationip, a.event_data_destinationport,
            a.event_data_image, 8 hour
        )
        a=SysmonRawLogs(
            event_id = "3"
            and event_data_direction = "outbound"

            and IPAddress4Utils.is_valid_ipv4(event_data_sourceip)
            and IPAddress4Utils.is_valid_ipv4(event_data_destinationip)

            and not StringUtils.wildcard_match(event_data_image, "*\\google\\*")
            and not StringUtils.wildcard_match(event_data_image, "*\\inetsrv\\w3wp.exe")
            and not StringUtils.wildcard_match(event_data_image, "*\\system32\\dns.exe")
            and not StringUtils.wildcard_match(event_data_image, "*\\system32\\svchost.exe")
        ) ->

    b=ComplexAttackAlerts(
        rule_name = "通过绑定1个单向匿名管道执行CMD命令_Sysmon" and
        a.computer_name = host_computer_name and
        a.event_data_image = host_parent_images and
        a.event_data_processguid = host_parent_guid and
        a.event_data_processid = host_parent_pid
    ) where timer:within(8 hour)
];

'

# 需要添加回调函数的EPL语句索引编号,不加这个字段表示不添加回调函数,例如仅仅创建全局窗口的规则,就不需要加此字段
# 如果为多个语句添加回调函数,则写成数组形式,例如:
# listener_epl_index:
#     - 0
#     - 1
#     - 3
listener_epl_index: 0

然后定义Listener的处理流程,可以先定义将会用到的局部常量和数据类型:

# 设置局部常量
# 表达式为『常量名: '数据类型:常量值'』
# 支持的基本数据类型有:
#   String:字符串类型,不填数据类型则默认为String类型,例如:string_const: 'aaaaa'
#   Integer:整数类型
#   int:等同Integer
#   Long:长整数类型
#   long: 等同Long
#   Boolean:布尔类型
#   boolean:等同Boolean
#   Float:浮点小数类型
#   float:等同Float
#   Double:双精度浮点小数类型
#   double:等同Double
#
# 如果创建数组,则通过以下方式表达
# 字符串数组String[]的表达形式,只需要第一个元素申明基本类型即可:
# string_list:
#     - 'String:aaaa'
#     - 'bbbbb'
#     - 'ccccc'
# 整数数组Integer[]的表达形式:
# integer_list:
#     - 'int:111'
#     - '222'
#     - '333'
#
# 如果需要创建一个null常量,先申明数据类型,再通过『$null』关键字赋值为null,例如:
# string_null: 'String:$null'
# long_null: 'Long:$null'
local_constant:
    total_alert_count: 'int:1'
    end_unix_timestamp: 'Long:0'

    splitRegex: 'String:,'

    aboveDeep: 'int:30'
    belowDeep: 'int:10'
    procDeep: 'int:30'

    event_id:
        - 'String:1'
        - '10'
        - '17'
        - '18'
        - '3'

提取EPL返回的行:

# 提取EPL的返回值为局部变量
# 表达式为『变量名: '数据类型:EPL返回值中的字段名'』
# 数据类型一致
local_variable:
    start_unix_timestamp: 'Long:a.unix_timestamp'

    dest_ip: 'String:a.event_data_destinationip'
    dest_port: 'int:a.event_data_destinationport'

    proto: 'String:a.event_data_protocol'

    computer_name: 'String:a.computer_name'
    host_ip: 'String:a.host_ip'
    user: 'String:b.host_users'

    parentImage: 'String:b.host_parent_images'
    parentCommand: 'String:b.host_parent_commandline'
    parentGuid: 'String:b.host_parent_guid'
    parentPid: 'String:b.host_parent_pid'

    image: 'String:b.host_images'
    command: 'String:b.host_commandline'
    guid: 'String:b.host_guid'
    pid: 'String:b.host_pid'

    sysmonLogIdStr: 'String:b.sysmon_log_id'

如果需要调用一些方法来完成即席查询任务,只需提前写好UDF然后调用即可:

# 动态调用Java代码中EsperQueryUtils()类中的函数
# 表达式为『返回值的变量名: '数据类型:函数名(参数1, 参数2, 参数3, ......)'』
# 函数传参可以调用局部常量和局部变量,需要在局部变量/常量名前面加上『$』符号,如果不加『$』符号则表示传入当前字符串。
local_variable_invoke_function:
    sysmonLogIdList: 'String[]:split($sysmonLogIdStr, $splitRegex)'
    commandContextList: 'String[]:query_win_context_command_sysmon($computer_name, $pid, $parentGuid,
                         $parentPid, $parentImage, $start_unix_timestamp, $aboveDeep, $belowDeep, $command)'
    processChainList: 'String[]:query_win_process_chain_sysmon($computer_name, $parentCommand, $parentImage,
                       $parentGuid, $parentPid, $command, $image, $guid, $pid, $procDeep)'

最后,将前面准备好的局部变量,填充到CEP告警模板中,生成一个完整的CEP告警:

# 动态调用Java代码中告警模板ComplexAttackAlert()类中的函数,向模板中添加告警信息字段
# 表达式为『函数名: '局部变量/常量名'』
# 如果对同一个函数进行多次调用传入不同的参数,则需要写成数组形式,例如:
# add_sysmon_log_id:
#     - '$aid'
#     - '$bid'
#     - '$cid'
create_alert_invoke_function:
    add_host_computer_name: '$computer_name'
    add_host_user: '$user'
    add_host_ip: '$host_ip'
    add_total_alert_count: '$total_alert_count'

    add_host_image: '$image'
    add_host_guid: '$guid'
    add_host_pid: '$pid'
    add_host_commandline: '$command'

    add_host_parent_image: '$parentImage'
    add_host_parent_commandline: '$parentCommand'
    add_host_parent_guid: '$parentGuid'
    add_host_parent_pid: '$parentPid'

    add_dest_ip: '$dest_ip'
    add_dest_port: '$dest_port'
    add_attacker_ip: '$dest_ip'
    add_attacker_port: '$dest_port'

    add_sysmon_log_id: '$sysmonLogIdList'
    add_host_command_context: '$commandContextList'
    add_host_process_chain: '$processChainList'

    add_host_event_id: '$event_id'

7、Case Study

通过Sysmon日志检测Cobalt Strike木马

8、结尾

结尾有点仓促,很多东西没有交代,也是长时间不写博客了写到最后脑子疼,最后欢迎一起交流其它的CEP方案。

 

打赏
0 comment

You may also like

Leave a Comment

*

code