2019-10-26 11:22:23 Author: www.4hou.com(查看原文) 阅读量:121 收藏
导语：在本文中，我将向你展示如何直接使用 Elasticsearch 数据库中的安全事件日志，并将它们保存到一个 DataFrame 中，并通过 Apache Spark Python API 和 SparkSQL 模块执行一些查询。
利用 Apache Spark 查询 Elasticsearch
在前一篇文章中，我介绍了使用 DataFrames 以表格格式表示和分析安全事件日志的概念，并向你展示了如何在名为 Pandas 的 python 库的帮助下实现这一点。
在本文中，我将向你展示如何直接使用 Elasticsearch 数据库中的安全事件日志，并将它们保存到一个 DataFrame 中，并通过 Apache Spark Python API 和 SparkSQL 模块执行一些查询。
要求
· 本文假设你已经阅读了前一篇文章，部署了 HELK 服务器，并且理解了通过 Python DataFrames 进行数据处理的基础知识。
让我们回顾一下这篇文章中将要使用的一些 Apache Spark 概念。
什么是 Apache Spark？
Apache Spark 是一个统一的计算引擎和一组用于计算机集群上并行数据处理的库。
统一分析？
它通过相同的计算引擎和一组一致的 API 支持范围广泛的数据分析任务。 例如，Spark 可以将以下 API 合并为一次数据扫描:
· 简单的数据加载
· SQL 查询
· 机器学习
· 流式计算
计算引擎? ？
· Spark 处理从存储系统加载数据并在系统上执行计算(不是永久存储)
· 移动数据需要大量的资源。 因此，Spark 专注于对数据执行计算，而不管数据位于何处。
库集合? ？
Spark 支持以下库:
· SQL 和结构化数据(Spark SQL)
· 机器学习(MLlib)
· 流程处理(Spark流和新的结构化流)
· 图形分析(GraphX)
· 外部库可在 spark-packages.org 找到(例如: Graphframes 图形框架)
Spark 基础架构
数据分析的挑战之一是数据处理，它通常与可用的计算资源有关。 Spark 引入了并行数据处理的概念，这种并行数据处理具有协调和调度跨集群 worker 执行的可用资源和任务的附加能力。
驱动过程
· 也称为 Spark Session，运行在你的 main () 函数中
· 管理有关 Spark 应用程序的信息。
· 处理对用户代码或输入的响应。
· 分析、分配和安排跨 executor 的工作。
Executors
· 执行驱动程序过程分配的任务
· 执行并向驱动器报告任务计算的状态
集群管理器
· 跟踪可用资源
如何运行 Spark 代码？
· 可以通过 Spark Language API 运行代码，比如 Scala、 Python 等等
· 通过 Language API 表示的概念被翻译成 Spark 代码并在机器集群上运行
Spark API
尽管你可以通过 Spark Language API 与分布式数据交互并运行代码，但是有两个主要的 Spark API 使这一切成为可能:
· 低级的非结构化 API
· 高级结构化 API
结构化 API: Spark Dataframe
· DataFrame 是最常见的结构化 API，它简单地将数据组织成指定的列和行，就像关系数据库中的表一样。
· 你可以将 DataFrame 看作是具有指定列的电子表格。
· Python DataFrame 位于特定位置的一台计算机上，而 Spark DataFrame 可以以分布式方式存在于多台计算机上
单机分析ー Python 数据模型
· 如果你有一个比你的本地硬盘大的 DataFrame，你不可能用一台机器来处理这个 DataFrame。
Dataframe 大于本地端点
· Spark DataFrame 概念允许你通过在多个服务器上分块执行并行分布式分析任务。
什么是 Apache SparkSQL？
· 它是一个 Spark 模块，利用 Sparks 函数式编程 API 来支持 SQL 关系结构化数据处理
· 它提供了一个编程抽象，称为DataFrames。当你在另一种编程语言中运行 SparkSQL 时，结果将作为 Spark DataFrames 返回。 在本文中，我们将使用 Pyspark 的基础知识通过 Spark SQL 模块与 DataFrames 进行交互
什么是 PySpark？
· PySpark 是用于 Spark 的 Python API
· DataFrame API 可用于 Scala，Java,Python 和R 语言中。
· 我们可以创建一个 Jupyter 内核来利用 PySpark API 并通过 notebook 与 Spark 集群进行交互。 HELK 已经提供了这样一个 notebook。
现在你已经了解了 Apache Spark、 Spark DataFrames 和 Spark 语言 API (如 PySpark)的基础知识，接下来我们可以开始读取一些数据并执行一些查询。
我需要一个数据源
正如前面提到的，Spark 专注于对数据的执行计算，而不管数据位于何处。 因此，确定数据驻留在哪里以及 Spark 如何访问数据非常重要。
目前，我将通过网络收集的安全事件日志存储在 Elasticsearch (ES)数据库中，作为 HELK 项目的一部分。 因此，我需要找到一种方法来通过 SparkSQL 查询我的 ES 数据库。
进入 ES-Hadoop 的世界
幸运的是，ES 和 Spark 集群之间的集成是通过 Elastic ES-Hadoop 库完成的。
· 一个开源的、独立的、自包含的小型库，允许 Hadoop 作业与 Elasticsearch 进行交互。
· 它允许数据双向流动，因此应用程序可以透明地利用 Elasticsearch 引擎的能力。
准备 Elasticsearch
让我们用 Elasticsearch 中的真实数据来准备我们的环境。
检查 HELK (ELK 容器)
· 打开你的 HELK 的 IP 地址并输入用户helk 和密码hunting。
· 点击stack monitoring 设置，如下图所示。
· 检查集群是否正在运行，是否有所有的 elastic 容器报告。
下载 Kafkacat
· 如果你使用的是基于 debian 的系统，请确保安装了最新的 Kafkacat deb 包
· 我推荐使用 Ubuntu 18.04。你可以检查Kafkacat deb 包的版本 然后和 Kafkacat GitHub repo 中最新的一个进行比较。
· 你也可以按照 Quick Build 快速构建 的指示从源代码进行安装。
· 安装完成后，通过 -L 元数据参数发起一个快速连接并指向 HELK Kafka topic —— winlogbeat. 该主题应该是HELK 构建的时候自动创建的。
kafkacat -b 192.168.64.138:9092 -t winlogbeat -L Metadata for winlogbeat (from broker 1: 192.168.64.138:9092/1): 1 brokers: broker 1 at 192.168.64.138:9092 1 topics: topic "winlogbeat" with 1 partitions: partition 0, leader 1, replicas: 1, isrs: 1
使用魔多的数据集
我们不再使用用于安全检测的整个实验室机器或者是直接分析几台运行中的 windows 电脑，而是使用我们在前一篇文章中使用的相同的 Mordor 数据集的empire_invoke_wmi，并通过 Kafkacat 将其发送到 HELK 堆栈。
kafkacat -b 192.168.64.138:9092 -t winlogbeat -P -l empire_invoke_wmi_2019-05-18214442.json
刷新 Kibana 发现视图
· 魔多数据集记录的是2019年5月18日的18个小时。
· 确保你选择了正确的时间窗口来验证数据是否到达 Elasticsearch 数据库
通过 Apache Spark 读取 Elasticsearch
我们已经准备好开始使用 ES-Hadoop 库来允许 Spark 通过其结构化的 DataFrame API 和 SQL 模块从 Elasticsearch 读取、分析和表示数据。
创建一个新的 Notebook
· 确保你选择了PySpark 内核。
导入 SparkSession
· 我们从导入来自 PySpark SQL 模块的SparkSession 类作为开始 。
· SparkSession 是 DataFrame 和 SQL 功能的主要入口点。 可以使用 SparkSession 创建一个 DataFrame，将 DataFrame 注册为表，对表执行 SQL，缓存表，甚至读取 parquet 文件。
创建一个 SparkSession
· 为了创建 SparkSession，我们需要使用builder 类。
· 我们给我们的 Spark应用程序起一个名字，并且设置 Spark Master 到我们的helk-spark-master 容器。 所有这些都已经被 HELK 处理好了。 这意味着我们将使用 HELK Spark 集群来执行由SparkSession 安排的任何任务。
spark = SparkSession.builder \
.appName("HELK Reader") \
.master("spark://helk-spark-master:7077") \
.enableHiveSupport() \
.getOrCreate()
验证 Spark 变量
· 一旦SparkSession 构建完成。 我们可以运行用于验证的spark变量
初始化 Elasticsearch 数据读取
· 为了从 Elasticsearch 开始读取数据，我们需要使用DataFrameReader 类 和 read 方法将数据作为一个 DataFrame 读入。
· 我们的基本命令只能初始化 DataFrame reader。
es_reader = (spark.read
.format("org.elasticsearch.spark.sql")
.option("inferSchema", "true")
.option("es.read.field.as.array.include", "tags")
.option("es.nodes","helk-elasticsearch:9200")
.option("es.net.http.auth.user","elastic")
)
从 Elasticsearch 加载数据: Sysmon 索引
· 然后我们就可以通过 DataFrame reader 使用load 方法来加载数据并返回一个 DataFrame。
· 我们可以指定一个特定的索引，在这里我选择使用 Sysmon 索引。
sysmon_df = es_reader.load("logs-endpoint-winevent-sysmon-*/")
筛选 Sysmon DataFrame
· 我们可以使用Filter 方法筛选出只包含ProcessCreate 事件的数据并且使用Select 方法从 Sysmon DataFrame 中返回特定的列。
processcreate_df = sysmon_df.filter(sysmon_df.action == "processcreate")
显示 Sysmon ProcessCreate DataFrame
到此为止，我希望你喜欢这一系列的博文。 在我们知道之前，我们一直在为使用Jupyter Notebook和通过 DataFrames 利用结构化数据处理能力来表示和分析数据打下基础。 还有很多事情可以通过 Apache Spark API 来完成。 如果你有自己的想法，你可以自己编写代码！
在下一篇文章中，我将向你展示如何利用 Apache SparkSQL 提供的关系数据处理来关联分析一些有趣的安全事件，这些事件在单独分析时可能不那么有趣或可疑。
通过 Apache SparkSQL 实现 SQL JOIN
在上文中，我们介绍了 Apache Spark 的基础知识，以及一些用于从 Elasticsearch 数据库中读取和加载数据的基本的 PySpark SQL 类。
在本文中，我将向你展示如何通过 Apache SparkSQL 模块和 Spark Python API 执行关系查询，以便关联分析一些安全事件，提供一些额外的且有趣的上下文。
要求
这篇文章假设你已经阅读过上一篇文章，并且运行了一个将empire_invoke_wmi 魔多数据集存储在 Elasticsearch上的 HELK 服务器(如果你没有这样做建议先阅读上一篇文章)
对数据的理解
· 在我们开始讨论特定安全事件之间的关系查询和 join 之前，我相信首先理解和记录你收集的数据以及它们提供的一直到字段级别的信息是非常重要的。 这不需要一次就完成，但是这是一个帮助安全分析师从数据的角度真正理解他们的差距和覆盖范围的主动权。
· 我认为记录事件日志是应用于开发数据分析和准备威胁狩猎活动的最佳方法的第一步。
· 此步骤还能够让你开始识别出以前可能不知道的数据源之间的数据结构和关系。 这是数据建模的基础。
什么是数据模型？
· 数据模型确定数据集中存在的数据对象的结构以及彼此之间确定的关系。
· 从安全事件的角度来看，数据对象可以定义为“ 用户”、“ 进程”、“ 文件”、“ IP 地址”甚至“ 注册表键”等实体。
· 此外，它们还可以具有”用户名”、”用户域”、”进程路径”或”进程名”等属性，并且根据数据源的结构，它们可以通过”进程 id”或”用户登录 id”等属性具有唯一标识符，这些属性有助于识别和关联映射到同一数据对象的几个事件。
Sysmon 数据模型
· 这是我对OSSEM项目进行最初研究的一部分工作就是记录 Windows Sysmon 事件日志的数据结构，以及它所提供的 approx.25事件之间的已有关系。
· Sysmon 提供了关于几个数据对象的信息，比如“进程”、“ IP”、“ 文件”、“注册表键” ，甚至“命令管道”。
· 此外，大多数数据对象都有一个名为 ProcessGUID 的公共属性，该属性定义了几个 Sysmon 事件之间的直接关系。 根据我的队友 Matt Graeber和 Lee Christensen 在他们最近发布的白皮书“颠覆 Sysmon”中所说的，ProcessGUID 是从机器 GUID、进程启动时间和进程令牌 ID 中获得的唯一值，可以用来关联其他相关事件。
· 在根据 Sysmon 数据对象的 ProcessGUID 属性记录这些数据对象之间的关系之后，就可以得出下面的内容:
Windows 安全审计怎么办？
· 对于 Windows 安全审计事件，由于可用的事件数量、审计类别和子类别的原因，提出数据模型或上图中的内容并不容易。 这个问题有点复杂。
· 但是，与 Sysmon 数据模型类似，大多数 Windows 安全事件都有一个公共属性，称为logon_id ，这个属性允许我们关联在同一个登录会话下发生的事件。
什么是登录会话？
· 根据微软官方的文档,（logon session） 登录会话 是一个计算会话，在用户身份验证成功时开始，在用户注销系统时结束。
· 当用户成功通过身份验证时，身份验证包创建了一个登录会话并将信息返回给Local Security Authority (也就是 LSA)，用于为新用户创建(LSA) （token ）令牌 。 这个令牌包括一个locally unique identifier 本地唯一标识符(LUID) 对于登录会话，调用logon Id 登录 Id.
· 从安全事件的角度来看，logon_id 是在成功的身份验证事件发生时，才会第一次出现。
· 根据微软和OSSEM提供的文档，来自审计登录子类别的事件与登录会话的创建有关，并且发生在被访问的计算机上。对于交互式登录，将在登录到的计算机上生成事件。对于网络登录(例如访问共享) ，将在承载所访问资源的计算机上生成事件。
事件4624: 帐户成功登录
· 这个特定的事件来自审计登录子类别，这个事件是在创建登录会话时(在目标计算机上)生成的。
· 在对4624事件记录特定字段并对有效值进行验证后, 我发现logon_type 字段非常有趣。 这是因为，事件4624是唯一可以告诉我们在登录过程中请求的特定登录类型的事件。
这有什么关系？
· 我相信这是非常有趣的事情，因为我可以使用事件4624的 logon_id 的值来关联在同一个登录会话上发生的其他 Windows 安全事件日志。 此外，由于事件4624中有可用的 logon_type信息，所以我可以根据用户成功验证时请求的特定登录类型对结果进行分类。
我以前就听说过
· 我记得在2017年，我的同事李 · 克里斯滕森 曾提到将登录类型9(NewCredentials)事件与在同一登录会话下发生的其他事件关联分析是非常不错的做法。 然而，在当时，我们并不知道有一个简单的方法可以规模化实现事件分析。
· 一年后，也就是2018年6月左右，我记得我和另一位同事安迪 · 罗宾斯 讨论关于Bloodhound的一些检测用例。 我记得我告诉过他一个潜在的检测方法，那就是在目标机器上，通过将安全事件4624(登录类型3)和logon_id相同的其他事件相关联，就可以检测到横向移动。经过几分钟的讨论之后，他将类似的内容用箭头组合起来，通过logon_id连接事件的想法开始变得更有意义了:
MATCH p = (n:Computer)-[:CitedIn|SameLogonID*1..]->(m:SysmonEvent) RETURN p
· 那年下半年，我发表了一个系列博文，名为基于 KSQL 和 HELK 的实时处理系统 ，我很高兴能够知道 SQL JOIN 在管道级别（随着数据流过）是可能的。 在那篇文章中，我使用Confluent KSQL 关联分析了 SysmonProcessCreate 及 NetworkConnect 事件。很快，我将分享如何在 KSQL 的帮助下关联分析4624安全事件与其他共享相同logon_id 的事件。
JOIN什么？
· 我们可以通过logon_id 字段的值将事件4624与其他事件关联起来，并向所有与逻辑匹配的事件添加logon_type上下文。
根据所请求的登录类型，可以有如下用例:
· 登录类型3: 潜在的横向移动
· 登录类型9: 潜在的模拟登陆
读取，记录，模型，重复
· 我要强调的一点是，仅通过阅读，记录和建模安全事件日志即可识别出潜在的基本用例。 这就是为什么将其作为覆盖所有潜在用例的检测策略的一部分是如此的重要。
· 例如，我不知道你是否已经知道这个事情，那就是Sysmon 中ID为 1的事件（ProcessCreate）是Sysmon中唯一提供logon_id信息的事件。 因此，我可以轻松地将logon_type信息添加到Sysmon提供的ProcessCreate信息中。
Apache SparkSQL 如何为我们提供帮助 ？
· 我们可以轻松地使用 Apache SparkSQL 的关系数据处理功能，并在 Elasticsearch 的数据上执行 SQL JOIN 语句。
· 在继续之前，让我们回顾一下每个 JOIN 类型背后的基本概念。
Inner Join:它返回的是在两个数据源中具有匹配值的数据记录。
Left Outer Join：它从左侧的数据源中返回数据记录，并从右侧的数据源中返回匹配的数据记录。
Full Outer Join：当左侧的数据源或右侧的数据中有匹配项时，将返回所有的数据记录。
Apache SparkSQL JOIN 步骤
· 从 HELK Elasticsearch Windows 安全索引中读取
· 为匹配事件 id 为 4624（成功的身份验证）的事件创建 DataFrame
· 通过 logon_type过滤数据 (如：网络(3) or 新的凭证(9))
· 为匹配事件 id 为1(ProcessCreate)的事件创建 DataFrame
· 通过logon_id对两个 DataFrames 执行 INNER JOIN
检测潜在的横向运动
创建一个新的Notebook
· 确保你选择了PySpark 内核
创建 Spark 会话
· 为了创建 SparkSession，我们使用builder 类
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("HELK JOIN") \
.master("spark://helk-spark-master:7077") \
.enableHiveSupport() \
.getOrCreate()
初始化读取器并读取数据
· 使用 DataFrameReader 类和read 方法从 Elasticsearch 中将数据读取为 DataFrame
· 使用load 方法和 DataFrame 读取器来加载数据并返回一个 DataFrame。我们将首先读取安全索引。
es_reader = (spark.read
.format("org.elasticsearch.spark.sql")
.option("inferSchema", "true")
.option("es.read.field.as.array.include", "tags")
.option("es.nodes","helk-elasticsearch:9200")
.option("es.net.http.auth.user","elastic")
)
security_df = es_reader.load("logs-endpoint-winevent-security-*/")
注册用于安全分析的 DataFrame
· 使用createOrReplaceTempView 方法创建用于安全分析的 DataFrame 的本地临时视图。
· 将此临时表的生命周期绑定到用于创建此 DataFrame 的 SparkSession。
security_df = es_reader.load("logs-endpoint-winevent-security-*/")
创建安全事件4624登录类型为3的DataFrame
· 在这一部分，我喜欢 Apache SparkSQL 分析数据的灵活性。
· 通过event_id 为4624 和logon_type 为3 (潜在的横向移动用例)来筛选数据。请记住，我们正在使用的是 Mordor 数据集empire_invoke_wmi 。因此，我期待一些有趣的结果。
· 筛选数据：只显示具有某个 IP 地址值的事件。
· 筛选特定时间窗口上的数据。 请记住，我们正在使用的是2019年5月18日前后记录的魔多数据集。
security_4624_3 = spark.sql( ''' SELECT event_id, host_name, src_ip_addr, user_logon_id, user_name, logon_type, `@timestamp` FROM security_events WHERE event_id = 4624 AND logon_type = 3 AND src_ip_addr is not null AND `@timestamp` BETWEEN "2019-05-18 00:00:00.000" AND "2019-05-19 00:00:00.000" ''' )
注册新的用于安全分析的 DataFrame
security_4624_3.createOrReplaceTempView("security_4624_3")
准备 Sysmon DataFrame
· 读取 Sysmon 索引并将 DataFrame 注册为临时视图
· 筛选 DataFrame 中 event_id 为1 的数据
· 筛选特定时间窗口上的数据。 请记住，我们正在使用的是2001年5月18日前后记录的魔多数据集
sysmon_df = es_reader.load("logs-endpoint-winevent-sysmon-*/")
sysmon_df.createOrReplaceTempView("sysmon_events")
sysmon_processcreate = spark.sql(
'''
SELECT event_id,
host_name,
process_parent_name,
process_parent_guid,
process_parent_command_line,
process_name,
process_guid,
process_command_line,
user_logon_id,
`@timestamp`
FROM sysmon_events
WHERE event_id = 1
AND `@timestamp` BETWEEN "2019-05-18 00:00:00.000" AND "2019-05-19 00:00:00.000"
'''
)
注册新的 Sysmon DataFrame
sysmon_processcreate.createOrReplaceTempView("sysmon_1")
执行 SQL INNER JOIN
· 创建了新的临时视图之后，我们就可以在logon_id 上执行sql join 操作。
security_sysmon_join = spark.sql( ''' SELECT s.`@timestamp`, s.host_name, s.src_ip_addr, s.logon_type, s.user_logon_id, s.user_name, p.process_parent_name, p.process_parent_guid, p.process_parent_command_line, p.process_name, p.process_guid, p.process_command_line FROM security_4624_3 s INNER JOIN sysmon_1 p ON s.user_logon_id = p.user_logon_id ''' )
显示结果
· 正如你在下面看到的，我们的 JOIN 逻辑能够通过 WMI 行为匹配潜在的横向移动
security_sysmon_join.select( "@timestamp","src_ip_addr","host_name","user_name","process_parent_name","process_parent_command_line","process_name","process_command_line" ).show()
通过 WMI 行为进行横向移动? ？
· 一个众所周知的横向移动技术是通过 WMI 对象类 Win32_Process及其 Create 方法来实现的。 这是因为 Create 方法允许用户在本地或远程创建进程。
· 需要注意的一点是当 Create 方法在远程系统上使用时，该方法是在一个名为“Wmiprvse.exe”的宿主进程中运行的。
· 在Create方法的 CommandLine 参数中定义的进程会由WmiprvSE.exe 生成新进程。 因此，远程创建的新进程将以Wmiprvse.exe作为父进程。
· WmiprvSE.exe 一个是 DCOM 服务器，它由 带有 C:\WINDOWS\system32\svchost.exe -k DcomLaunch -p 参数的 DCOM 服务宿主进程 ——svchost.exe 生成。
WMI 和登录类型为3的登录会话
WmiprvSE.exe 是由 DCOM 服务宿主程序在不同的登录会话中生成的。 但是，无论 WmiprvSE.exe 执行什么操作，都会发生在通过网络身份验证的用户创建的新登录会话上。
根据魔多的数据集的文档的说明:
· 对手使用了帝国模块 invoke_wmi 在电脑 IT001.shire.com(172.18.39.105) 到HR001.shire.com (172.18.39.106) 上生成一个新的代理。
· 对手使用了 SHIRE\pgustavo 的凭证。
· 对手与新代理进行互动并执行了whoami。
如果我们再看一下 SparkSQL JOIN 的结果，我们会发现:
· 执行的 stager 来自172.18.39.106 。
· 帐户pgustavo 被用来进行网络身份验证。
· 由 WmiprvSE 执行了 PowerShell 。
· Powershell 是在新的网络登录会话上下文中创建的。
· PowerShell 执行了 Whoami。
· Whoami 是在新的网络登录会话上下文中创建的。
希望你能喜欢这篇文章。 我尝试使用我们在本系列文章前面几部分中学到的所有基本概念，并将它们应用于记录和对安全事件进行建模。
在下一篇文章中，我将向你展示如何在公司内部和社区中使用 Jupyter Notebook 记录和分享你的威胁狩猎 notebook。
参考资料
http://people.csail.mit.edu/matei/papers/2015/sigmod_spark_sql.pdf
https://docs.microsoft.com/en-us/windows/security/threat-protection/auditing/event-4624
https://docs.microsoft.com/en-us/windows/desktop/secauthn/lsa-logon-sessions
https://spark.apache.org/docs/latest/index.html
https://spark.apache.org/docs/latest/API/python/pyspark.sql.html
https://www.elastic.co/products/hadoop
https://databricks.com/glossary/what-is-spark-sql
本文翻译自：https://posts.specterops.io/threat-hunting-with-jupyter-notebooks-part-3-querying-elasticsearch-via-apache-spark-670054cd9d47如若转载，请注明原文地址： https://www.4hou.com/system/21107.html
如有侵权请联系:admin#unsafe.sh