使用 “版本 ”下拉列表切换服务。 了解有关导航的详细信息。
适用于: ✅ Microsoft Sentinel
Kusto 查询语言(KQL)是一种功能强大的工具,用于查询和分析Microsoft Sentinel 中的数据。 作为安全分析师,掌握 KQL 可以显著增强检测威胁和有效应对事件的能力。 本文提供了使用 KQL 执行常见任务的综合指南,可帮助你高效作和分析数据。
在本教程中,我们将介绍 KQL 的基础知识,包括了解查询结构、获取、限制、排序和筛选数据、汇总数据和联接表。 此外,我们还探讨了高级概念,例如使用 evaluate 运算符和 let 语句创建更复杂的可维护查询。
先决条件
阅读本文之前,请确保已熟悉 Kusto 查询语言(KQL)的基础知识。 如果您是 KQL 的新手,请参阅:
了解查询结构基础知识
开始学习 Kusto 查询语言的一个良好起点是了解整体查询结构。 查看 Kusto 查询时,你注意到的第一件事是使用管道符号 (|)。 Kusto 查询的结构首先从数据源获取数据,然后通过“管道”传递数据,每个步骤提供某种级别的处理,然后将数据传递到下一步。 在流水线末尾你将获得最终结果。 实际上,这是我们的工作流程:
Get Data | Filter | Summarize | Sort | Select
通过将数据沿管道传递的概念,可以形成一种直观的结构,因为这使得在每一步很容易在心中构建数据的图景。
为了说明这一点,让我们看看以下查询,该查询将查看 Microsoft Entra 登录日志。 当你通读每一行时,可以看到指示数据正在发生什么的关键字。 我们已将管道中的相关阶段作为注释包含在每行中。
注释
可以通过用双斜杠//()在查询中的任何行前面添加注释。
SigninLogs // Get data
| evaluate bag_unpack(LocationDetails) // Ignore this line for now; we'll come back to it at the end.
| where RiskLevelDuringSignIn == 'none' // Filter
and TimeGenerated >= ago(7d) // Filter
| summarize Count = count() by city // Summarize
| sort by Count desc // Sort
| take 5 // Select
由于每个步骤的输出都作为下一步的输入,因此步骤的顺序可以确定查询的结果并影响其性能。 您需要根据希望从查询中获得的结果来排序步骤。
一个很好的经验法则是尽早筛选数据,这样你只会将相关数据传递给数据处理管道。 这极大地提高了性能,并确保不会意外地在汇总步骤中包含不相关的数据。 有关详细信息,请参阅 Kusto 查询语言查询的最佳做法。
希望你现在对 Kusto 查询语言中的查询的整体结构表示赞赏。 现在,让我们看看实际查询运算符本身,这些运算符用于创建查询。
获取、限制、排序和筛选数据
Kusto 查询语言的核心词汇(可用于完成大部分任务的基础)是用于筛选、排序和选择数据的运算符集合。 剩余的任务要求你扩展语言知识以满足更高级的需求。 让我们对 前面示例中 使用的一些命令进行一些扩展,并查看 take, sort以及 where。
对于这些运算符中的每一个,我们将检查其在之前的SigninLogs示例中的用法,并学习实用技巧或最佳实践。
获取数据
任何基本查询的第一行会指定您要操作的表。 对于 Microsoft Sentinel,这可能是工作区中的日志类型的名称,例如 SigninLogs、 SecurityAlert 或 CommonSecurityLog。 例如:
SigninLogs
在 Kusto 查询语言中,日志名称区分大小写,因此SigninLogssigninLogs以不同的方式解释。 为自定义日志选择名称时,请小心,以便可以轻松识别它们,并且与另一个日志不太类似。
限制数据:采取 / 限制
take 运算符(和相同的限制运算符)用于仅返回给定数量的行来限制结果。 后跟一个整数,指定要返回的行数。 通常,它在确定排序顺序后在查询末尾使用,在这种情况下,它将返回排序顺序顶部的给定行数。
在查询中较早使用take可以在不返回大型数据集的情况下进行查询测试。 但是,如果在任何take作之前放置sort作,take则返回随机选择的行,并且每次运行查询时可能都返回一组不同的行。 下面是使用“take”的示例:
SigninLogs
| take 5
小窍门
在处理一个可能对查询结构不太了解的全新查询时,在开头放置 take 语句可以人工限制数据集,以提高处理和实验的速度会很有用。 对完整查询感到满意后,可以删除初始 take 步骤。
排序数据: 排序 / 顺序
排序运算符(和相同的顺序运算符)用于按指定列对数据进行排序。 在以下示例中,我们按 TimeGenerated 对结果进行排序,并将顺序方向设置为使用 desc 参数降序,首先放置最高值;对于升序,我们将使用 asc。
注释
排序的默认方向为降序,因此从技术上来说,你只需指定是否要按升序排序。 但是,在任何情况下指定排序方向会使查询更具可读性。
SigninLogs
| sort by TimeGenerated desc
| take 5
如前所述,我们将 sort 运算符放在 take 运算符之前。 我们需要先进行排序,以确保获得相应的五条记录。
顶部
top运算符允许我们将sort和take操作合并为单个运算符:
SigninLogs
| top 5 by TimeGenerated desc
如果两条或更多条记录在用于排序的列中具有相同的值,你可以添加更多列以帮助排序。 在逗号分隔的列表中添加额外的排序列,该列位于第一个排序列之后,但在排序顺序关键字之前。 例如:
SigninLogs
| sort by TimeGenerated, Identity desc
| take 5
现在,如果 TimeGenerated 在多个记录之间相同,则它会尝试按 Identity 列中的值进行排序。
注释
何时使用 sort , take以及何时使用 top
如果只对一个字段进行排序,请使用
top,因为它提供的性能优于组合sort和take。如果需要对多个字段进行排序(如上一个示例中
top),则无法执行此作,因此必须使用sort和take。
筛选数据: 其中
其中运算符可以说是最重要的运算符,因为它是确保仅处理与方案相关的数据子集的关键。 应尽量尽早在查询中筛选数据,因为这样做会减少后续步骤中需要处理的数据量,从而提高查询性能:它还可确保仅对所需数据执行计算。 请参阅此示例:
SigninLogs
| where TimeGenerated >= ago(7d)
| sort by TimeGenerated, Identity desc
| take 5
该 where 运算符指定变量、比较(标量)运算符和值。 在本例中,我们用于 >= 表示 TimeGenerated 列中的值需要大于(即晚于)或等于七天前。
Kusto 查询语言中有两种类型的比较运算符:字符串和数字。 字符串运算符支持区分大小写、子字符串位置、前缀、后缀及更多功能。
该 == 运算符既是数字运算符,也是字符串运算符,这意味着它可用于数字和文本。 例如,以下两个语句都是有效的 where 语句:
| where ResultType == 0| where Category == 'SignInLogs'
最佳做法: 在大多数情况下,你可能希望按多个列筛选数据,或以多种方式筛选同一列。 在这些情况下,应记住两种最佳做法。
可以使用where关键字将多个语句合并为单个步骤。 例如:
SigninLogs
| where Resource == ResourceGroup
and TimeGenerated >= ago(7d)
当使用where关键字将多个筛选器连接到单个语句时,通过将仅引用单一列的筛选器放在前面,可以获得更好的性能。 因此,编写上一个查询的更好方法是:
SigninLogs
| where TimeGenerated >= ago(7d)
and Resource == ResourceGroup
在此示例中,第一个筛选器提到单个列(TimeGenerated),而第二个筛选器引用了两列(Resource 和 ResourceGroup)。
汇总数据
Summarize 是 Kusto 查询语言中最重要的表格运算符之一,但对于一般查询语言的新手来说,它也是学习起来较为复杂的运算符之一。 该任务summarize用于接收一个数据表,然后输出一个由一列或多列聚合生成的新表。
汇总语句的结构
一个summarize语句的基本结构如下所示:
| summarize <aggregation> by <column>
例如,下面将返回 Perf 表中每个 CounterName 值的记录计数:
Perf
| summarize count() by CounterName
由于summarize的输出是一张新表,因此summarize语句中未显式指定的任何列都不会传递到管道中。 若要说明此概念,请考虑以下示例:
Perf
| project ObjectName, CounterValue, CounterName
| summarize count() by CounterName
| sort by ObjectName asc
第二行中,我们指定仅关注 ObjectName、CounterValue 和 CounterName 列。 然后总结为按 CounterName 获取记录计数,最后尝试根据 ObjectName 列按升序对数据进行排序。 遗憾的是,此查询失败并显示错误(指示 ObjectName 未知),因为在汇总时,我们只在新表中包含 Count 和 CounterName 列。 为了避免此错误,我们可以将 ObjectName 添加到步骤末尾 summarize ,如下所示:
Perf
| project ObjectName, CounterValue , CounterName
| summarize count() by CounterName, ObjectName
| sort by ObjectName asc
在你的脑海中读取 summarize 行的方式可以是:“按 CounterName 汇总记录数,并按 ObjectName 进行分组。” 你可以继续在 summarize 语句的末尾添加用逗号分隔的列。
在前面的示例基础上,如果我们想同时聚合多个列,可以通过向运算符 summarize 添加聚合(用逗号分隔)来实现此目的。 在下面的示例中,我们不仅获取所有记录的计数,还获取所有记录的 CounterValue 列中的值的总和(与查询中的任何筛选器匹配):
Perf
| project ObjectName, CounterValue , CounterName
| summarize count(), sum(CounterValue) by CounterName, ObjectName
| sort by ObjectName asc
重命名聚合列
这似乎是谈论这些聚合列的列名的好时机。
在本部分的开头,我们说 summarize 运算符接受一个数据表并生成一个新表,并且只有在 summarize 语句中指定的列会继续沿管道传递。 因此,如果要运行上述示例,则聚合的结果列将会是 count_ 和 sum_CounterValue。
Kusto 引擎会自动创建列名称,无需我们明确指定,但通常你可能更希望新列具有更友好的名称。 您可以在summarize语句中通过指定一个新名称(后跟=和聚合)轻松重命名列,如下所示:
Perf
| project ObjectName, CounterValue , CounterName
| summarize Count = count(), CounterSum = sum(CounterValue) by CounterName, ObjectName
| sort by ObjectName asc
现在,汇总的列名称为 Count 和 CounterSum。
操作符有比此处能涵盖的更多的内容,但你应该投入时间来了解它,因为它是你计划对 Microsoft Sentinel 数据执行的任何数据分析的关键组成部分。
聚合参考
这是许多聚合函数,但一些最常用的函数是 sum(), count()以及 avg()。 有关详细信息,请参阅 聚合函数类型一目了然。
选择:添加和删除列
当你处理更多查询时,你可能会发现,获得的信息比对你的主题所需的信息更多(即表中列数过多)。 或者,可能需要比你拥有的更多信息(也就是说,需要添加包含其他列分析结果的新列)。 让我们看一下列操作的一些关键运算符。
Project和投射
Project大致相当于多种语言的选择语句。 它允许您选择保留哪些列。 返回的列顺序与在 project 语句中列出的列的顺序匹配,如以下示例所示:
Perf
| project ObjectName, CounterValue, CounterName
设想一下,当你处理列数多的数据集时,你可能有许多需要保留的列,而逐一命名所有这些列将需要大量的输入。 对于这些情况,你有 project-away,这样就可以指定要删除的列,而不是要保留哪些列,如下所示:
Perf
| project-away MG, _ResourceId, Type
小窍门
在查询中,在开头和末尾使用project会很有用。 在查询开始时使用 project 可以通过去除无需传递到流水线的大数据块来提高性能。 在末尾再使用它,可以删除在前面步骤中创建但在最终输出中不需要的任何列。
Extend
Extend 用于创建新的计算列。 如果要对现有列执行计算,并查看每一行的输出,这非常有用。 让我们看看一个简单的示例,在其中计算一个名为 Kbytes 的新列,我们可以通过将 MB 值(在现有 数量 列中)乘以 1,024 来计算该列。
Usage
| where QuantityUnit == 'MBytes'
| extend KBytes = Quantity * 1024
| project DataType, MBytes=Quantity, KBytes
在 project 语句的最后一行中,我们将 Quantity 列重命名为 Mbytes,以便我们可以轻松地判断哪个度量单位与每列相关。
值得注意的是,extend 也适用于已经计算过的列。 例如,我们可以添加一个名为 Bytes 的列,该列是从 Kbytes 计算得出的:
Usage
| where QuantityUnit == 'MBytes'
| extend KBytes = Quantity * 1024
| extend Bytes = KBytes * 1024
| project DataType, MBytes=Quantity, KBytes, Bytes
联接表
Microsoft Sentinel 中的大部分工作都可以使用单个日志类型来执行,但有时你想要将数据关联在一起或针对另一组数据执行查找。 与大多数查询语言一样,Kusto 查询语言提供了一些用于执行各种类型的联接的运算符。 在本部分中,我们将介绍最常用的运算符和 unionjoin。
Union
联合 只是将两个或多个表组合并返回所有行。 例如:
OfficeActivity
| union SecurityEvent
这将从 OfficeActivity 和 SecurityEvent 表返回所有行。
Union 提供了一些参数,可用于调整联合的行为方式。 最有用的两个是 带源 和 类型:
OfficeActivity
| union withsource = SourceTable kind = inner SecurityEvent
withsource 参数允许您指定一个新列的名称,其在给定行中的值是该行来源的表的名称。 在本示例中,我们命名了 SourceTable 列,根据行,该值为 OfficeActivity 或 SecurityEvent。
我们指定的另一个参数是 类型,它具有两个选项: 内部 或 外部。 在本示例中,我们指定 了内部,这意味着在联合期间保留的唯一列是存在于这两个表中的列。 或者,如果我们指定了 外部连接(即默认值),则会返回这两个表中的所有列。
加入
联接 的运作方式与 union 类似,但不是通过联接表生成新表,而是联接 行 以创建新表。 与大多数数据库语言一样,可以执行多种类型的联接。
join 的一般语法为:
T1
| join kind = <join type>
(
T2
) on $left.<T1Column> == $right.<T2Column>
在join运算符之后,指定我们要执行的联接类型,后面跟一个左括号。 括号中是您指定要联接的表,以及您希望在该表上添加的任何其他查询语句的位置。 右括号后,我们使用 on 关键字,其后是我们的左列关键字 ($left.<columnName>) 和右列关键字 ($right.<columnName>),中间使用 == 运算符分隔。 下面是 内部联接的示例:
OfficeActivity
| where TimeGenerated >= ago(1d)
and LogonUserSid != ''
| join kind = inner (
SecurityEvent
| where TimeGenerated >= ago(1d)
and SubjectUserSid != ''
) on $left.LogonUserSid == $right.SubjectUserSid
注释
如果两个表对要对其执行联接的列具有相同的名称,则无需使用 $left 和 $right;可以改为指定列名称。 然而,使用 $left 和 $right更为明确,通常被视为一种良好做法。
小窍门
最佳做法是将最小的表放在左侧。 在某些情况下,遵循此规则可以带来巨大的性能优势,具体取决于要执行的联接类型以及表的大小。
有关详细信息,请参阅 联接运算符。
Evaluate
你可能会记得,回到 第一个示例中,我们在其中一行中看到了 评估 运算符。 运算符 evaluate 的使用频率比我们之前接触过的运算符要低。 但是,了解 evaluate 运算符的工作方式非常值得投入时间。 再次,这是第一个查询,可以在第二行看到 evaluate。
SigninLogs
| evaluate bag_unpack(LocationDetails)
| where RiskLevelDuringSignIn == 'none'
and TimeGenerated >= ago(7d)
| summarize Count = count() by city
| sort by Count desc
| take 5
使用此运算符可以调用可用的插件(内置函数)。 其中许多插件侧重于数据科学,例如 autocluster、 diffpattern 和 sequence_detect,使你能够执行高级分析和发现统计异常和离群值。
示例中使用的插件称为 bag_unpack,因此只需获取一部分动态数据并将其转换为列即可。 请记住, 动态数据 是类似于 JSON 的数据类型,如以下示例所示:
{
"countryOrRegion":"US",
"geoCoordinates": {
"longitude":-122.12094116210936,
"latitude":47.68050003051758
},
"state":"Washington",
"city":"Redmond"
}
在这种情况下,我们希望按城市汇总数据,但 City 作为 LocationDetails 列中的属性包含。 若要在查询中使用 city 属性,必须先使用 bag_unpack将其转换为列。
回到我们最初的管道步骤,我们看到了:
Get Data | Filter | Summarize | Sort | Select
现在,我们已经考虑了 evaluate 运算符,可以看到它表示管道中的新阶段,现在如下所示:
Get Data | Parse | Filter | Summarize | Sort | Select
还有其他许多运算符和函数示例可用于将数据源分析为更具可读性和可作的格式。 可以在 Kusto 查询语言学习资源和工作簿中了解它们以及 Kusto 查询语言的其余部分。
Let 语句
现在,我们已经介绍了许多主要的运算符和数据类型,让我们总结一下 let 语句,这是让你的查询更易于读取、编辑和维护的好方法。
允许 创建和设置变量,或将名称分配给表达式。 此表达式可以是单个值,但它也可以是整个查询。 下面是一个简单示例:
let aWeekAgo = ago(7d);
SigninLogs
| where TimeGenerated >= aWeekAgo
在这里,我们指定了 aWeekAgo 的名称,并将其设置为等于 时间跨度 函数的输出,该函数返回 日期/时间 值。 然后,我们使用分号终止 let 语句。 现在,我们有一个名为 aWeekAgo 的新变量,可在查询中的任何位置使用。
如前所述,可以使用 let 语句获取整个查询,并为结果命名。 由于查询结果(作为表格表达式)可用作查询的输入,因此可以将此命名结果视为表,以便运行另一个查询。 下面是对上一个示例的轻微修改:
let aWeekAgo = ago(7d);
let getSignins = SigninLogs
| where TimeGenerated >= aWeekAgo;
getSignins
在本例中,我们创建了第二个 let 语句,在该语句中,我们将整个查询包装成名为 getSignins 的新变量。 就像以前一样,我们用分号终止第二个 let 语句。 然后,我们在运行查询的最后一行上调用变量。 请注意,我们能够在第二个 let 语句中使用 aWeekAgo。 这是因为我们在上一行中指定了它;如果我们要交换 let 语句,以便 getSignins 先来,我们将收到错误。
现在,我们可以使用 getSignins 作为另一个查询的基础(在同一窗口中):
let aWeekAgo = ago(7d);
let getSignins = SigninLogs
| where TimeGenerated >= aWeekAgo;
getSignins
| where level >= 3
| project IPAddress, UserDisplayName, Level
Let 语句可让你更强大和灵活地帮助组织查询。 让我们 定义标量值和表格值,以及创建用户定义的函数。 当你构建可能执行多个联接的更复杂查询时,它们非常有用。
后续步骤
充分利用 Microsoft Sentinel 自带的 Kusto 查询语言工作簿——Microsoft Sentinel 的高级 KQL 工作簿。 它为你提供了许多在日常安全作中可能遇到的情况的分步帮助和示例,并指出了大量现成的现成分析规则、工作簿、搜寻规则以及更多使用 Kusto 查询的元素示例。 从Microsoft Sentinel 中的 “工作簿 ”页启动此工作簿。
有关详细信息,请参见: