Azure NSG Flow Log 引发的自嗨 -- 日志查询分析

        上一篇中通过 Function 服务来实现了 NSG Flow Log 的 ETL 和流化处理，然后将处理好的 NSG Flow Log 推入到 Event Hub 中。NSG Flow Log 的分析引擎这里选用 Azure Data Explorer，是因为 Data Explorer 有强大的查询能力，对时序数据有良好的支持，内置机器学习能力，以及和 Azure 上其它产品原生的对接能力。通过 Event Hub 将 NSG Flow Log 推入 Data Explorer 来保证消息的准实时性，如果对于实时性容忍度高也可以选择通过 Function 直接将 NSG Flow Log 写入到 Data Explorer。再来看一下架构图，本文将介绍如何从 Event Hub 中将消息注入到 Data Explorer，以及在 Data Explorer 中的一些基础查询示例。

         Event Hub 作为 Azure 的 1st Party 产品，原生已经内置 Data Explorer 的连接器，所以在 Data Explorer 中配置 Event Hub 连接器即可。在配置连接器之前需要在 Data Explorer 中创建 Database 和 Table，创建 Table 定义 Schema，通过消息 Mapping 规则将 Event Hub 中注入的 Json 格式内的消息属性字段映射到 Table 中定义的属性字段里。

        Table 的 Schema 依照之前在 Function 中 NSG Flow Log 日志消息的格式进行定义，下面是一个从 Function 中发出的 NSG Flow Log 的示例：

{
    'category': 'NetworkSecurityGroupFlowEvent', 
    'macAddress': '000D3A0F2885', 
    'operationName': 'NetworkSecurityGroupFlowEvents', 
    'resourceId': '/SUBSCRIPTIONS/C04B3C63-8DFE-4F98-BE18-E71FF67A1F4E/RESOURCEGROUPS/ADXNSG/PROVIDERS/MICROSOFT.NETWORK/NETWORKSECURITYGROUPS/NSGLOGDEMO-NSG', 
    'systemId': '23246691-2d22-4668-879e-32b396c63e75', 
    'time': '2019-06-11T02:40:39.6296239Z', 
    'rule': 'UserRule_SSH', 
    'mac': '000D3A0F2885', 
    'flowtuples': '1560220793,218.92.0.210,172.16.2.4,28119,22,T,I,A,B,,,,'
}

        基于上述消息格式定义 Table Schema，以下示例创建名称为 nsglog 的 Table，详细的 Data Explorer 创建 Table 步骤可参阅：https://docs.microsoft.com/en-in/azure/data-explorer/ingest-data-event-hub

.create table nsglog (Category: string, MacAddress: string, OperationName: string, ResourceId: string, SystemId: string, Time: datetime, Rule: string, Mac: string, Flowtuples: string)

        创建 Mapping 规则将 Event Hub 中注入的 Json 消息属性映射到 Table 属性中，通过 $. 访问 Event Hub 消息中 Json 属性并定义映射规则。

.create table nsglog ingestion json mapping 'NsglogMapping' '[{"column":"Category","path":"$.category","datatype":"string"},{"column":"MacAddress","path":"$.macAddress","datatype":"string"},{"column":"OperationName","path":"$.operationName","datatype":"string"},{"column":"ResourceId","path":"$.resourceId","datatype":"string"},{"column":"SystemId","path":"$.systemId","datatype":"string"},{"column":"Time","path":"$.time","datatype":"datetime"},{"column":"Rule","path":"$.rule","datatype":"string"},{"column":"Mac","path":"$.mac","datatype":"string"},{"column":"Flowtuples","path":"$.flowtuples","datatype":"string"}]'

        NSG Flow Log 注入到 Data Explorer 后，便可通过 Data Explorer 的分析引擎进行相关查询，在定义 Table 属性映射规则时注意到 flowtuples 字段里面的属性直接没有进行任何处理直接映射的一个 Flowtuples 属性中，所以在做查询分析编写查询语句时候需要加入很多字符函数来对其中的属性值做拆解，这里我们先来定义一个通用处理语句将 Flowtuples 内的属性值拆解成具体的属性值（如时间戳，源 IP，目的 IP，源端口，目的端口，协议类型，流方向，规则决策，流状态，收发包及字节统计）。

let nsglogextend = nsglog 
| extend timestamp =  datetime(1970-01-01) + tolong(split(Flowtuples, ",")[0]) * 1sec  // extract flow timestamp
| extend Srcip = split(Flowtuples, ",")[1]  // extract flow source ip
| extend Destip = split(Flowtuples, ",")[2]  // extract flow destination ip
| extend Srcport = toint(split(Flowtuples, ",")[3])  //  extract flow source port
| extend Destport = toint(split(Flowtuples, ",")[4])  // extract flow destination port
| extend Protocol = split(Flowtuples, ",")[5]  // extract flow protocol type
| extend Direction = split(Flowtuples, ",")[6]  // extract flow direction
| extend Decision = split(Flowtuples, ",")[7]  // extract flow rule decision Allow or Deny
| extend Flowstate= split(Flowtuples, ",")[8]  // extract flow state
| extend S2dPacket= iif(split(Flowtuples, ",")[9] == "", long(0), tolong(split(Flowtuples, ",")[9]))  //  extract flow S2D packet statstics 
| extend S2dByte = iif(split(Flowtuples, ",")[10] == "", long(0), tolong(split(Flowtuples, ",")[10]))  // extract flow S2D byte statstics 
| extend D2sPacket= iif(split(Flowtuples, ",")[11] == "", long(0), tolong(split(Flowtuples, ",")[11]))  // extract flow D2S packet statstics 
| extend D2sByte = iif(split(Flowtuples, ",")[12] == "", long(0), tolong(split(Flowtuples, ",")[12])) ;  // extract flow S2D byte statstics

        万事具备，日志在我手，怎么用起来？这里举一个栗子，带大家一起自嗨一下。网络安全部门经常遇到的问题是开放的端口有恶意的扫描以及访问，如何在日常运维中及时定位发现并且进行处理，这里我们按照 flow pattern based 异常监测作为理论基础，正常的业务流量会按照一个基础特征运行，有异常流量产生的时候会产生与基础特征相悖的特征。在测试环境中有一台 Linux 虚拟机开启了 SSH 端口，这里通过 ADX 来发现是否有异常 IP 在尝试连接该虚拟机。

let min_t = toscalar(nsglogextend | summarize min(timestamp))  // extract min timestamp in log
let max_t = toscalar(nsglogextend | summarize max(timestamp))  // extract max timestamp in log
nsglogextend
| filter Flowstate == "B"  // 1
| make-series num=count() default=0 on timestamp in range(min_t, max_t, 1h) by Rule  // 2
| render timechart  // 3

        上述查询语句说明： 1. 在所有日志中搜索所有 Flow 建立日志，2. 将日志按照小时为时间窗口单位，统计每小时内 Flow 的数量，并按照 NSG Rule 规则分类，3. 将上述统计按照时序图标进行显示，

        从上述输出中可以看到 2019-06-10 当日有共计 1861 个 SSH 连接请求，明显高于其它时间。下面继续挖掘到底谁在发起这些 SSH 连接，将时间窗口聚焦到 2019-06-10 单日，按照分钟为时间窗口，查看到底异常连接发生在什么时段。

let min_t = toscalar(nsglogextend | summarize min(timestamp));
let max_t = toscalar(nsglogextend | summarize max(timestamp));
nsglogextend
| filter Flowstate == "B"
| make-series num=count() default=0 on timestamp in range(datetime(2019-06-10 14:00:00), datetime(2019-06-10 15:00:00), 1m) by Rule
| render timechart 

        可以看到从 14：28 分开始连接请求开始高于平常值，到底是哪些机器发起了这些连接，地址是什么，是不是某个 IP 发起了很多请求？

let min_t = toscalar(nsglogextend | summarize min(timestamp));
let max_t = toscalar(nsglogextend | summarize max(timestamp));
nsglogextend
| filter Flowstate == "B"
| where timestamp between (datetime(2019-06-10 14:00:00)..datetime(2019-06-10 15:00:00))
| summarize count(tostring(Srcport)) by tostring(Srcip)

        上述查询语句对 2019-06-10 14：00 到 2019-06-10 15：00 这个时间窗口内的所有发起 SSH 请求的源地址进行归类并统计数量。

        可以看到 51.75.90.236 这个 IP 非常异常，在短短一小时内共计发起了 1017 次连接。通过上述这个栗子可以看到 Azure Data Explorer 具有非常强大的分析引擎，可以对 NSG Flow Log 进行分析，这里引入的 Flow Pattern Based 异常检查是一个非常基础的分析方式，这里以 SSH 恶意登陆为栗子抛砖引玉，大家可以借助分析已经来建立比如：Flow 端口数量特征（分析恶意端口扫描事件），Flow IP 数量特征（分析 DDos 攻击事件），流量与数据包比例特征（分析数据盗取事件）等。除此之外，现在业界 ML 技术应用与 Abnormal Detection 也有很多的实践，Azure Data Explorer 内置 ML 能力，可以通过 ML 来对 Flow Pattern 进行建模，从而对 Flow 行为已经异常判别。

        至此我们四期的自嗨之旅要告一段落了，希望大家都有所收获，我们下期再见！