流媒体到网络安全Lakehouse Windows事件日志

流媒体到网络安全Lakehouse windows事件

企业客户经常会问,什么是最简单的和简单的方法来发送实时Windows终端登录数据砖,执行ETL和运行检测搜索安全事件的数据。这是有意义的。Windows登录大型环境必须被监控,但可以很嘈杂,消耗大量资源在传统SIEM产品。摄取系统事件日志到三角洲表和执行流分析有很多成本和性能优势。

这个博客主要关注组织如何从端点收集Windows事件日志,直接进入一个网络安全lakehouse。具体来说,我们将展示如何创建一个管道为微软sysmon处理事件,并将数据转换为公共信息模型(CIM)格式,可用于下游分析。

“我们怎么能摄取和亨特窗端点在规模,同时也维护我们目前的安全架构?”Curious Databricks Customer

建议的体系结构

实际上,Windows端点日志必须通过货代运送到一个中央存储库进行分析。有许多特定于供应商的可执行文件,所以我们专注于最普遍适用的体系结构,使用winlogbeats和卡夫卡集群。的弹性winlogbeats货代已经免费和开源许可,bob下载地址Apache卡夫卡也是一个开源的分布式事件流媒体平台。bob体育客户端下载你可以找到一个示例配置文件都在笔记本或创建您自己的特定配置Windows事件使用winlogbeats手册。如果你想使用卡夫卡服务器出于测试目的,我创建了一个github库使它容易。您可能需要调整这个架构如果你使用其他软件。

数据集

我们还安装了微软系统监控(sysmon)由于其有效性为目标集合用于安全用例。我们将演示如何解析原始JSON日志从sysmon /操作日志和公共信息模型应用到最相关的事件。一旦运行,笔记本将会产生银级三角洲湖表下面的事件。

在笔记本使用winlogbeats配置文件,端点也将发送WinEventLog:安全、WinEventLog:系统,WinEventLog:应用程序中,Windows Powershell WinEventLog: WMI日志文件,也可以使用感兴趣的读者。

通过卡夫卡摄取的数据集

你可能情有可原,获取数据的卡夫卡和湖成三角洲是一个复杂的业务!然而,它不能简单。与Apache火花™,卡夫卡连接器是准备好了,可以直接进入三角洲湖流数据使用火花流。

kafka_df = (spark.readStream。格式(“卡夫卡”).option (“kafka.bootstrap.servers”,“xx.xx.xx.xx: 9094”).option (“订阅”,“winlogbeat”).option (“startingOffsets”,“最新”).option (“failOnDataLoss”,“假”).load ())

告诉火花。readStream使用apache火花卡夫卡连接器,位于
kafka.bootstrap.servers
ip地址。订阅的主题windows事件到达,你是去比赛!

对于可读性,我们将只显示最流行的部分代码,然而,完整的笔记本电脑可以下载使用底部的链接的文章,包括一个链接到一个免费的社区版砖,如果必要的。

winlogbeatDF winlogbeatSchema = read_kafka_topic (bootstrapServers = bootstrapServerAddr端口=“9094”主题=“winlogbeat”)如果类型(winlogbeatDF) = = DataFrame:winlogbeatDF = add_ingest_meta (winlogbeatDF)winlogbeatDF = parser_kafka_winlogbeat (winlogbeatDF阶段=“生”)显示器(winlogbeatDF)其他的:打印(winlogbeatDF winlogbeatSchema)

我们使用上面的代码读原始卡夫卡流使用read_kafka_topic函数,并应用一些顶级拔牙、青铜级别表主要用于分区。

这是一个很好的开始。我们的端点是流媒体实时日志卡夫卡集群和砖dataframe。然而,看来我们之前有更多的工作要做,dataframe准备分析!

一探究竟,event_data字段是嵌套在一个结构体,json和看起来像一个复杂的问题。

在我们开始工作之前转换列,我们写数据帧到青铜级别表,分区_event_date,和_sourcetype。将使我们能够有效地选择这些分区列只读日志源过滤事件时我们需要运用我们的CIM转换。

分区= [“_event_date”,“_sourcetype”]write_table (df = winlogbeatDF,表名=“winlogbeat_kafka_bronze”、表=“青铜”=分区,分区,streamType = streamMode)bronzeWinlogbeatDF = read_table (“winlogbeat_kafka_bronze”).cache ()bronzeWinlogbeatDF = parser_kafka_winlogbeat (bronzeWinlogbeatDF)sysmonProcessDF = bronzeWinlogbeatDF。过滤器((bronzeWinlogbeatDF。_sourcetype = =“Microsoft-Windows-Sysmon /操作”)&((坳(“winlog: event_id”)= =' 1 ')|(坳(“winlog: event_id”)= =“5”)|(坳(“winlog: event_id”)= =“18”)))

上面的数据帧是阅读的青铜的结果表,压扁的列和过滤只处理相关事件(过程开始,过程结束和管道连接)。

夷为平地的列结构和过滤数据帧组成的过程相关的事件,最后阶段是应用标准化字段名称的数据字典。为此,我们使用OSSEM项目命名格式,并应用一个函数,输入dataframe,和一个转换列表,并返回最终dataframe正常化。

transform_cols=({“新”:[“event_message”、“EXPR =情况(event_id = 1),那么过程开始的时候(event_id = 5),那么“过程终止”(event_id = 18)然后管连接结束”)},{“新”:[“event_message_result”、“文字=成功”)},{"重命名":["事件:行动”,“event_status”)},{"重命名":[" winlog: event_data:命令行”、“process_command_line”)},{"重命名":[" winlog: event_data:公司”、“file_company”)},{"重命名":[" winlog: event_data: CurrentDirectory”、“process_file_directory”)},{"重命名":[“dvc_hostname”、“dvc_hostname”]},{"重命名":[“winlog: event_id”、“event_id”)},{"重命名":[“winlog:任务”、“event_category_type”)},。。{"重命名":[“winlog: event_data:散列”,“file_hashes”)},{"重命名":[" winlog: event_data: PipeName ", " pipe_name "]}]sysmonProcess=cim_dataframe (sysmonProcessDF transform_cols)partition_cols=[" _event_date "]write_table (df=sysmonProcess,分区=partition_cols,表=“过程”,表=“银”,streamType=streamMode)
              显示器(sysmonProcess)

由此产生的数据帧被规范化CIM兼容并写入一个银表,由_event_date分区。白银级别表被认为适合运行检测规则。果不其然!

还好下一步增加银表的性能,将是z值它基于列最有可能用于过滤。列process_name和event_id将优秀的候选人。类似的应用布隆过滤器基于user_name列将加快阅读活动时基于实体的搜索。下面的一个例子。

如果optimizeInline:create_bloom_filter(表=“过程”、列= bloom_cols)optimize_table(表=“过程”、列= z_order_cols)

结论

我们已经看到如何创建一个可伸缩的流管道从企业端点包含复杂的结构,直接进入lakehouse。这提供了两个主要的好处。首先,目标的机会,但往往嘈杂的数据,可以使用检测规则,分析了下游或AI威胁检测。其次保持颗粒的能力水平的历史使用三角洲端点数据表成本有效的存储、长期保留,如果需要。

寻找未来的博客,在那里我们将会深入研究分析使用这些数据集。下载完整的笔记本和一个预配置卡夫卡服务器开始流数据窗端点到今天lakehouse !如果你没有一个砖的客户,随时旋转了社区版从这里。