深度学习OSSIM关联分析（附源码注解）

从海量安全事件中挖掘有用的威胁信息与情报是当今讨论的热门话题，同时这也是一个难点？怎么实现呢？这里用到一种技术叫做关联分析，他也是SIEM（Security  Information Event Management安全信息和事件管理）系统中最常见的事件检测手段，这并不是什么新鲜事物，20年前就已经有人提出来了。通常基于时序来对相同数据源或来自不同数据源的安全事件，使用关联规则来进行综合的关联分析，下面介绍关联分析的具体功能。

    通常来说，一次恶意Gong JI会在多个安全设备或应用程序(如网络防火墙，交换机，Web 应用日志，SQL 日志，审核日志等)中留下痕迹． 然而，所有这些信息都是孤立隔绝的，被保存在不同的设备日志中，如果利用了关联分析技术就可以快速定位故障。

关联分析为什么有如此神通广大呢？其实后台有很多复杂的关联规则最为基础的，这些检测规则可以识别 A t t a c k 事件，实际上这些规则是人工在大量实践中总结出来生成对Gong JI事件的一种语言描述，并对其进行了精确分类，分类越细描述越准确，识别率就越高。

Tips: 下文中“Gong Ji”“**”代表铭感词汇，你懂得。

一、关联分析核心思想

关联分析技术的核心思想是通过对某一类事件进行训练建立行为基线，基线范围外的事件视为异常事件来进行分类． 该算法较适合于本文检测场景，通常 SIEM 系统中绝大多数的日志为正常事件，通过对正常事件训练建模，来检测异常或Gong JI事件，这就是理论上说的One-Class SVM（单类支持向量机）。OSSIM系统中需要更多维度特征向量，才能准确判断Gong JI源，避免检测精度低或过拟合情况． 算法输入是经过 OSSIM 归一化后具备相同数据结构的安全事件，输出则为带有标记的安全事件，标记分为两类:正常(Negative)与Gong JI(Positive)． 而OSSIM关联分析引擎的输出就是 One-Class SVM分类算法输出的带标记的正常与Gong JI事件。

通过深入分析 SIEM 系统的关联规则，其中最常见的配置字段如:event_id，timestamp，plugin_id，plugin_sid，src_ip，src_port，des_ip，des_port，protocol 等。 为了使关联分析规则不依赖于传感器配置，本文从 OSSIM 的规则库中抽取了几个重要的字段 plugin_id_name，plu-gin_id_description，sid_name，并将所有字段拆分成关键词标签，然后针对每一条检测规则生成其关键词标签统计模型，利用规则统计模型来代替原始的规则来检测Gong JI。

关联分析的核心通过一个或一组关联指令来完成，下面用SSH**破解的例子加以说明，如图1所示，SSH**破解（Brute Force）大约自UNIX诞生之后，就衍生出来的一种Gong JI行为，据统计发现大约有50%以上的用户名是root。一个低可靠性（low reliability）的SSH 服务器，在经过100登录尝试之后成功登录，而高可靠性（high reliability）的SSH服务器，则经过10000次登录才成功，那么关联引擎可通过在一定时间内，登录的次数，以及这些时间的不同源地址和相同目标IP地址，以及这些IP地址来自于那些国家，它们信誉度如何等信息来综合判定Gong JI以及作出响应。

图1

下面大家将亲生经历关联引擎的关联指令的组成结构。

二、 关联指令配置界面

  OSSIM中关联指令的界面通过Configuration->Threat Intelligence->Directives打开，经过前面几节的内容的学习，大家或多或少的已经接触过关联指令的界面，下面我们进行一下归纳，如下图2所示。

图2  OSSIM  关联指令界面

关键功能解释：

• New Directive:单击此选项以从头开始创建一个新的指令。

• Test Directives:单击此按钮，将检测当前新建/修改的指令是否正确。

• Restart Server:单击此按钮，将重启ossim-server进程，凡是修改了任何一条关联指令都需重新启动Server，按这个按钮比你重启整个USM服务器更明智。

需要注意的是，虽然重启时间短暂，但重启期间所有关联数据会丢失。

图3 一条完整指令

● Sensor：传感器，用于收集各种事件信息。

● Protocol:协议，包括ANY、TCP、UDP、ICMP。

● Sticky different：

   还记得Linux基础是指中，用于文件及目录属性设定的sticky位吗？从OSSIM 4.3起在关联指令中添加了新属性，Sticky different（不同的粘粘位）域，规则中Sticky different为None。它是用来设定指令规则的属性，

有关Sticky different的取值大家可参考/etc/ossim/server/directives.xsd文件

<xs:simpleType name="stickydifferenttype">

... ...略

</xs:simpleType>

   当新收集的事件到达关联引擎时，它将和已有的事件相关联，如果事件属性（如IP地址和端口号）相同，但一个指令的属性里可以设置不同的粘粘位，比如None、DST_PORT、SRC_IP、PLUGIN_ID等，用来区别收到的不同事件，以便进行相互关联。例如在端口扫描类Gong JI中，如果你设置目标端口为Sticky different，那么只有来自不同目标端口的事件被关联。

三、在OSSIM 的Web UI中查看效果

下面我们查看实际的例子，如图4所示。

图4  STICKY DIFF

打开/etc/ossim/server/alienvault-dos.xml查看关联指令AvAttacks,Possible DDOS

● Username：事件中指定的用户名

● Pass：     事件中指定的口令

● Userdata1~Userdata8:事件中指定的数据域。

图5  查看更对关联规则属性

注意，From、To、Data source 、Event Type 列下方的号表示可修改，Action下的号代表可添加规则 ，但系统默认指令中的规则不允许修改。

四、深入关联规则

1. 基本操作

   在OSSIM利用预先定义的规则（/etc/ossim/server/*.xml）来进行关联分析。那么我们如何操作呢，首先，我们通过Web界面，新建一个Correlation Directives，新建两个规则ssh和test，然后我们看看详细文件内容，路径在/etc/ossim/server目录，名为user.xml文件。其他默认规则由/etc/ossim/server/directives.xml定义，如图6所示。

图6 自定义指令

当系统调用Directive下的策略是，首先根据categories.xml配置文件读取相应的XML配置文件，这些文件功能如下:

策略文件存储位置：/etc/ossim/server/及说明

Ø  alienvault-attacks.xml         AlienVault Gong JI类策略

Ø  alienvault-bruteforce.xml      AlienVault**破解类Gong JI

Ø  alienvault-dos.xml             Alienvault拒绝服务类

Ø  alienvault-malware.xml         Alienvault 恶意软件（包括检测各种蠕虫的规则）

Ø  alienvault-misc.xml            Alienvault各种失误类(Miscellaneous)

Ø  alienvault-network.xml         Alienvault网络类 （开源版无此项）

Ø  alienvault-policy.xml          Alienvault策略

Ø  alienvault-scan.xml            Alienvault扫描

Ø  user.xml                       Alienvault用户自定义

如果OSSIM关联引擎读不到这些策略配置文件，在Analysis→Alarm就不会生成报警。接下来我们详细看看一个xml的例子。

2.理解规则树

关联引擎通过规则来实现对安全事件的关联分析，读者需要能够看懂OSSIM的规则，其中采用了特有的树形规则，在规则树中的每一个节点对应一条关联规则(rules)。在匹配时关联引擎将某段时间内收到的统一格式的报警，从根节点开始往叶子节点逐次匹配，系统根据匹配的结果可以进行事件聚合和再次提升报警级别。下面看个实例：

图7 自定义指令内容

从上图可以看出，这种基于事件序列的关联方法中的每条指令相当于一颗由规则组成的树，所以可以把它叫做基于树形指令（Directive）的关联方法，其基本思想是根据相关事件序列创建一系列的规则来表示Gong JI场景，在OSSIM系统中采用了XML来定义关联序列，关联分析引擎启动时就将所有关联导入每个关联规则序列由下面标签组成将上面的实际xml提炼一下，得到如下模板：

<directive id="" name="" priority="">

<rule type="" name="" reliability="" occurence="” from="" to="" port_from="" port_to="" plugin_id=""plugin_sid="">

<rules>

   <rules>

   ......

</rules

    ......

</rule>

</directive>

每个序列开头包括两个标签“directive_id”和“directive name”，而每个序列，又包括很多规则，规则之中又可以嵌套规则，即递归方法。

其中Rule 代表规则，规则后面代表一个可能发生的Gong JI场景，Event 代表在当前已发生的Gong JI场景下，一个Gong JI场景由若干条规则组成，这些规则可能是“与”和“或”的关系。这样表示的好处是，我们清楚的知道，如果一个Gong JI场景发生，那么只需要满足哪些规则即可。

通过计算每个Rule 里的Reliability 来确认这个Gong JI发生的可信度是多少。其中Scene 的id用directive_id来表示；前面讲过Reliability 可以是0~10之间的数，直接给可靠性赋值，也可以使一个修改量，表示当这个规则满足时将前一步Gong JI场景的可靠性做修改，将结果存入New_Reliability中；规则部分的其它属性代表了将和它做匹配的Event 的属性值；而和数据库的具体交互是通过Action 来表示。

下面我们看个实际的例子，下面这段指令主要是用来检测公网服务器是否可用，其中包含了两个规则。

<directive id="500000" name="Public Web Server unavailable" priority="1">

     <rule type="detector" name="server unavailable" reliability="1"

    occurrence="1" from="192.168.11.100" to="ANY" port_from="ANY" port_to="ANY"

    plugin_id="1525" plugin_sid="1,7,9">

    <rules>

    <rule type="monitor" name="Ping Baidu Server  in order to check internet connection"

      reliability="10" from="www.baidu.com" to="www.baidu.com" plugin_id="2009" plugin_sid="1" condition="gt" value="0" interval="20" time_out="120" />

    </rules>

</directive>

关键参数解释：

● Detector: 探测器规则自动收集从代理发来的记录，包括Snort、Apache、Arpwatch等。

● Monitor：监控器负责查询Ntop服务发来的数据和会话。

● Name：事件数据库中的规则名称。

● Reliability:可靠性

● Plugin_id:插件ID,查看更多插件ID请参考《开源安全运维平台OSSIM最佳实践》第一章内容。

● Plugin_sid:插件子ID号，分配给每个插件事件的子ID，比如Snort这个插件ID号为1001,而1501就是它的子ID号，代表Apache事件的响应代码，当然可不止这一件。

● Condition:条件参数和下面6个逻辑有关系，必须符合的逻辑条件匹配规则如下:

Ø  eq – 等于（Equal）

Ø  ne – 不等于（Not equal）

Ø  lt – 小于（Lesser than）

Ø  gt – 大于（Greater than）

Ø  le – 小于等于（Lesser or equal）

Ø  ge – 大于等于（Greater or equal）

●Interval：间隔，这个值类似于time_out,用于监控类规则。

3.Alienvault-scan规则描述

再举一个OSSIM自带的例子，我们查看/etc/ossim/server/alienvault-scan.xml这个扫描Gong JI场景的策略文件，描述的结构就像个逻辑树，基本格式如下：

Gree:level 1

   Yellow : level2

    Orange:level3

    Red: level 4

下面给出部分代码内容，如图8所示

图8 Gong JI扫描指令示例

● Occurrence，表示发生次数，默认为1，Gong JI场景不同这个值也不一样。这个值越大，越要引起管理员的警觉。这里表示的发生次数，也就是计算具有相同的 "from、to、port_from、port_to、plugin_id、plugin_sid" 发生次数，用以进行到关联模式中的下一规则。

这个数值在基于规则的关联分析中非常有用，例如当目标IP地址发生的异常行为事件数量的频率达到了一定数值时（某人针对该系统发动Gong JI），OSSIM会发出预警提示，管理员就可以有针对性开展深入调查。当然对源IP也适用。

From表示来源，源地址有以下几种形式：

①ANY，任意IP地址都可以匹配。

②小数点和数字的IPv4形式。

③以逗号隔开的IPv4地址，不带掩码。

④可以使用任意数目的IP地址，中间用逗号隔开。

⑤网络名称，可以使用网络中事先定义好的网络名称。

⑥相对值，这种情况比较复杂，可以引用上条规则中的IP地址，例如：

l  1:SRC_IP 表示引用前一条规则的源地址

l  2:DST_IP 表示引用前第二条的目的地址作为源地址

⑦否定形式，可以使用地址的否定形式如 :"!192.168.150.200，HOME_NET"。

如果 HOME_NET = 192.168.150.0/24 将匹配一个C类子网排除192.168.150.200。

l  Time_out，表示超时，其等待一定时间以匹配规则，时间超出则匹配失败。

l  Port_from/Port_to ，表示来自哪里/目的端口，Port_from可以是ANY，Port_to，可以是一个端口号或一个逗号分隔的端口序列，1:DST_PORT，也可以否定端口例如，port=“!22,25”。

注意：dst_ip表示目的ip地址，而dst_port则表示目的端口号，ipaddr本地ip地址。

l  Protocol（协议），可以使用以下字符串：TCP、UDP、ANY。

l  Plugin_id（插件ID），参考plugin中的plugin_id。

l  Plugin_sid（插件SID），每个事件都是分配一个子ID。

图9 规则树的嵌套

   在OSSIM系统运行前，必须为已知的Gong JI场景建立对应的树形规则集，在启动时，系统将预先定义好的指令读入内存，在接收到一个事件 （event）后，先将该event与之前已经匹配，而还没有匹配完的指令中的规则进行匹配，然后在于其他规则匹配。那么在一个复杂的Gong JI场景中一条Event就会与多条规则匹配，如果此事件的风险值超过预先设定的阈值，则将其alarm属性设为真，并在Alarm界面中显示。

例如：plugin_id="1001" plugin_sid="2008609,2008641"。

在OSSIM 4.6系统中，有385个数据源，这里ID=1001，代表Snort检测插件，产品类型属于IDS，主要适用于Snort规则，其它插件ID还记得吗？如果忘了请返回本书第1章，查看“插件&功能”对应表。

实际上在OSSIM系统中，Snort插件ID范围是1001～1145。在OSSIM 4.8 版中位于Configuration→Threat Intelligence→Data Source可以查看到所有数据源。如图10所示。

图10 数据源分类

在Ossim的关联引擎运行之前，必须为所有已知的Gong JI场景建立对应的树形指令（在开源系统中提供了84条，在OSSIM 4.15的商业版本提供了1800余条，OSSIM USM 5.x提供2100+条），这也是它的核心价值的体现，在OSSIM启动时，系统会将所有预先定义好的指令读入内存，在接收到一个directive_event后，先将该事件与之前已经匹配，而还没有匹配完的指令中的规则匹配，然后再与其它指令的规则匹配。注意一个事件可以与很多指令中的规则匹配。

五、 完善规则

OSSIM系统中的关联规则生成主要通过安全专家人工编写，这种方式效率低下，且人工成本难以承受． 对于新出现的Gong JI和漏洞无法及时作出响应，检测规则的编写往往出现滞后的情况， 大家也许会考虑能否利用AI技术和数据融合技术进行智能化的数据挖掘呢，这些刚高级的黑科技在实际应用中往往生成的关联规则检测精度较低，检测结果不理想，实际应用效果有限。利用人工神经网络来自动生成关联规则，是关联分析研究领域今后发展的方向。

对于新手而言从一张白纸开始写规则有些难了，但是对照系统给出的默认规则，照葫芦画瓢还是可以实现的，不管这个模型是否完善，先用起来，然后逐步修改。在合适的时间，对自己进行***，然后监控SIEM的反应，观察它是否产生正确的警报，而警报是否提供足够的信息来协助相应这弄清楚发生了什么威胁，如果没有那就需要修改规则，然后你需要不断的优化阈值，你是否发现SIEM报警过于频繁，或者非常稀少，你需要调节阈值来解决，顺便说一句，面对动态的网络Gong JI，这个调节的过程没有终点如果读者在学习关联规则的过程中遇到各种问题也可以参考我的新书《开源安全运维平台OSSIM疑难解析：提高篇》。

 附件：

下面分享的是OSSIM关联分析的一部分源代码。

/*
** * 该指令是否与根节点指令匹配，这里只检查根节点,并不检查指令的子节点**
 */gboolean
sim_directive_match_by_event (SimDirective  *directive,
                                                      SimEvent      *event)
{
  SimRule *rule;
  gboolean match;

  g_return_val_if_fail (directive, FALSE);
  g_return_val_if_fail (SIM_IS_DIRECTIVE (directive), FALSE);
  g_return_val_if_fail (!directive->_priv->matched, FALSE);
  g_return_val_if_fail (directive->_priv->rule_root, FALSE);
  g_return_val_if_fail (directive->_priv->rule_root->data, FALSE);
  g_return_val_if_fail (SIM_IS_RULE (directive->_priv->rule_root->data), FALSE);
  g_return_val_if_fail (event, FALSE);
  g_return_val_if_fail (SIM_IS_EVENT (event), FALSE);

  rule = SIM_RULE (directive->_priv->rule_root->data);

  match = sim_rule_match_by_event (rule, event); 

  return match;
}/*
** *这将检查事件是否可以与backlog中的某些数据匹配。backlog实际上是一个包含事件数据的指令。每个backlog条目都是一个树，其中包含来自一个指令的所有规则（它相当于是一个指令克隆）。其中每个规则（simrule）还包含与规则匹配的事件的数据。**
 * 
 */gboolean
sim_directive_backlog_match_by_event (SimDirective  *directive,
                                                                      SimEvent    *event)
{
  GNode      *node = NULL;

  g_return_val_if_fail (directive, FALSE);
  g_return_val_if_fail (SIM_IS_DIRECTIVE (directive), FALSE);
  g_return_val_if_fail (!directive->_priv->matched, FALSE);
  g_return_val_if_fail (directive->_priv->rule_curr, FALSE);
  g_return_val_if_fail (directive->_priv->rule_curr->data, FALSE);
  g_return_val_if_fail (SIM_IS_RULE (directive->_priv->rule_curr->data), FALSE);
  g_return_val_if_fail (event, FALSE);
  g_return_val_if_fail (SIM_IS_EVENT (event), FALSE);

  node = directive->_priv->rule_curr->children;  while (node)      //**我们必须对照backlog中的所有规则节点检查事件,除了根节点，因为它签入了sim_directive_match_by_event是从sim_organizer_correlation调用的.**
  {
    SimRule *rule = (SimRule *) node->data;    if (sim_rule_match_by_event (rule, event))
        {
            g_log (G_LOG_DOMAIN, G_LOG_LEVEL_DEBUG, "sim_directive_backlog_match_by_event; sim_rule_match_by_event: True");
          time_t time_last = time (NULL);
            directive->_priv->rule_curr = node;     // 每次事件匹配时，该指令都下一级到匹配的节点。下次将根据此级别检查事件。

                                                                                        //FIXME: 父节点中可能存在内存泄漏.
          directive->_priv->time_last = time_last;
          directive->_priv->time_out = sim_directive_get_rule_curr_time_out_max (directive);

            sim_rule_set_event_data (rule, event);      //这里我们将事件中的各个字段分配到规则中,所以每次我们进入规则时，我们可以看到匹配的事件.

          sim_rule_set_time_last (rule, time_last);          if (!G_NODE_IS_LEAF (node))
        {
          GNode *children = node->children;          while (children)
                {
                  SimRule *rule_child = (SimRule *) children->data;

                  sim_rule_set_time_last (rule_child, time_last);

                  sim_directive_set_rule_vars (directive, children);
                  children = children->next;
                    g_log (G_LOG_DOMAIN, G_LOG_LEVEL_DEBUG, "sim_directive_backlog_match_by_event: There are childrens in %d directive", directive->_priv->id);
                }
            }          else
          {
              directive->_priv->matched = TRUE;
                g_log (G_LOG_DOMAIN, G_LOG_LEVEL_DEBUG, "sim_directive_backlog_match_by_event: The directive %d has matched", directive->_priv->id);
          }          return TRUE;
        }        else
        {
            g_log (G_LOG_DOMAIN, G_LOG_LEVEL_DEBUG, "sim_directive_backlog_match_by_event: sim_rule_match_by_event: False");
        }

      node = node->next;
    }  return FALSE;
}/*
 * 检查指令中的所有节点规则,以查看.......
 */gboolean
sim_directive_backlog_match_by_not (SimDirective  *directive)
{
  GNode      *node = NULL;
  GNode      *children = NULL;

  g_return_val_if_fail (directive, FALSE);
  g_return_val_if_fail (SIM_IS_DIRECTIVE (directive), FALSE);
  g_return_val_if_fail (!directive->_priv->matched, FALSE);
  g_return_val_if_fail (directive->_priv->rule_curr, FALSE);
  g_return_val_if_fail (directive->_priv->rule_curr->data, FALSE);
  g_return_val_if_fail (SIM_IS_RULE (directive->_priv->rule_curr->data), FALSE);

  node = directive->_priv->rule_curr->children;  while (node) 
  {
    SimRule *rule = (SimRule *) node->data;        //如果规则已超时 &&       
    if ((sim_rule_is_time_out (rule)) && (sim_rule_get_not (rule)) && (!sim_rule_is_not_invalid (rule))) 
        {
          time_t time_last = time (NULL);
        directive->_priv->rule_curr = node;
          directive->_priv->time_last = time_last;
          directive->_priv->time_out = sim_directive_get_rule_curr_time_out_max (directive);

        sim_rule_set_not_data (rule);          if (!G_NODE_IS_LEAF (node)) //这不是最后的节点，他还有一些子节点.
        {
          children = node->children;          while (children)
                {
                SimRule *rule_child = (SimRule *) children->data;

                  sim_rule_set_time_last (rule_child, time_last);

                  sim_directive_set_rule_vars (directive, children);
                  children = children->next;
                }
        }        else //last node!
        {
          directive->_priv->matched = TRUE;
        }        return TRUE;
        }
    node = node->next;
  }  return FALSE;
}/*
 * backlog&directives几乎是相同的：backlog是存储指令并填充事件数据的地方。
 *“node”是子节点函数。我们需要从引用其级别的节点向该节点添加src_ip、port等。如果“node”参数是根节点->子节点1->子节点2中的children2，并且我们在children2中有1:plugin-sid，那么我们必须将根节点中的plugin-sid添加到children2中。
 */void
sim_directive_set_rule_vars (SimDirective     *directive,
                                                     GNode            *node)
{
  SimRule    *rule;
  SimRule    *rule_up;
  GNode      *node_up;
  GList      *vars;
  GInetAddr  *ia;
  GInetAddr  *sensor;
  gint        port;
  gint        sid;
  SimProtocolType  protocol;
    gchar               *aux = NULL;

  g_return_if_fail (directive);
  g_return_if_fail (SIM_IS_DIRECTIVE (directive));
  g_return_if_fail (node);
  g_return_if_fail (g_node_depth (node) > 1);

  rule = (SimRule *) node->data;
  vars = sim_rule_get_vars (rule);

 2019 最 新 作 品