创建多阶段查询

支持的平台：

Google SecOps SIEM

本文档介绍了如何使用 YARA-L 中的多阶段查询，将一个查询阶段的输出直接馈送到后续阶段的输入中。与单个整体查询相比，此过程可让您更好地控制数据转换。

将多阶段查询与现有功能集成

多阶段查询可与 Google Security Operations 中的以下现有功能结合使用：

• 复合检测规则：多阶段查询通过弥合自动化检测和主动调查之间的差距，对复合检测 规则进行补充。虽然复合规则擅长识别较长时间窗口内的复杂多事件关联，但多阶段查询可让分析师从这些检测结果转向实时迭代搜索，以立即验证和确定正在展开的威胁的范围。

• 时间范围和多事件规则： 您可以使用多阶段查询，通过比较数据中的不同时间窗口来检测异常情况。例如，您可以使用初始查询在较长时间内建立基准，然后使用后续查询根据该基准评估近期活动。您还可以使用 多事件规则 创建类似类型的比较。

信息中心 和搜索 都支持 YARA-L 中的多阶段查询。

联接有助于关联来自多个来源的数据，从而为调查提供更多背景信息。通过关联相关事件、实体和其他数据，您可以调查复杂的攻击场景。如需了解详情，请参阅在搜索中使用联接。

主要注意事项

配置多阶段查询时，请注意以下事项：

• 限制阶段：除了根阶段之外，多阶段查询还必须包含一到四个命名 阶段。
• 顺序语法：请务必先定义命名阶段语法，然后再定义 根阶段语法。

我们建议您在实现多阶段查询时查看以下已知问题和建议的解决方法：

• 所有多阶段查询的行为都类似于 统计信息搜索查询 （输出由汇总统计信息组成，而不是未汇总的事件或数据表行）。

• 由于数据集的大小，与 UDM 和实体事件联接的性能可能会较低。我们强烈建议尽可能过滤联接的 UDM 和实体事件端（例如，按事件类型进行过滤）。

如需有关推荐实践的一般指南，请参阅 YARA-L 2.0 最佳实践；如需了解有关联接的特定信息，请参阅 最佳实践。

主要术语

在联接的上下文中， 窗口化阶段 是指包含窗口的 match 部分的阶段。相比之下，表阶段不会输出窗口。

创建多阶段 YARA-L 查询

如需创建多阶段 YARA-L 查询，请完成以下步骤。

阶段结构和语法

前往调查 > 搜索 。定义查询阶段时，请遵循以下结构：

语法：使用以下语法为每个阶段命名，并将其与 其他阶段分开：

stage <stage name> { }

• 大括号：将所有阶段语法放在大括号 {} 内。

• 顺序：先定义所有命名阶段的语法，然后再定义根 阶段。

• 引用：每个阶段都可以引用查询中较早定义的阶段。

• 根阶段：查询必须有一个根阶段，该阶段在所有 命名阶段之后处理。

以下示例阶段 daily_stats 会收集每日网络统计信息：

stage daily_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $source = principal.hostname
  $target = target.ip
  $source != ""
  $target != ""
  match:
    $source, $target by day
  outcome:
    $exchanged_bytes = sum(network.sent_bytes + network.received_bytes)
}

访问阶段输出

后续阶段可以使用阶段字段访问命名阶段的输出。阶段字段与阶段的 match 和 outcome 变量相对应，并且可以与统一数据模型 (UDM) 字段类似的方式使用。

使用以下语法访问阶段字段：

$<stage name>.<variable name>

可选：访问窗口时间戳

如果命名阶段使用跃点窗口、滑动窗口或翻滚窗口，请使用以下预留字段访问每个输出行的窗口开始时间和窗口结束时间：

• $<stage name>.window_start

• $<stage name>.window_end

window_start 和 window_end 是整数字段，以自 Unix 纪元以来的秒数表示。不同阶段的窗口大小可能不同。

多阶段查询示例

本部分中的示例有助于说明如何创建完整的多阶段 YARA-L 查询。

示例：搜索异常活跃的网络连接（小时）

此多阶段 YARA-L 示例用于识别网络活动高于正常水平的 IP 地址对，目标是保持高活动状态超过三小时的 IP 地址对。该查询包含两个必需的组件：命名阶段 hourly_stats 和 root 阶段。

hourly_stats 阶段会搜索网络活动级别较高的 principal.ip 和 target.ip 对。

此阶段会为以下字段返回单个小时值：

• 来源 IP 的统计信息（字符串）：$hourly_stats.src_ip

• 目标 IP 的统计信息（字符串）：$hourly_stats.dst_ip

• 事件计数的统计信息（整数）：$hourly_stats.count

• 接收字节数的标准差（浮点数）：$hourly_stats.std_recd_bytes

• 接收字节数的平均值（浮点数）：$hourly_stats.avg_recd_bytes

• 小时存储桶开始时间（自 Unix 纪元以来的秒数，整数）： $hourly_stats.window_start

• 小时存储桶结束时间（自 Unix 纪元以来的秒数，整数）： $hourly_stats.window_end

根阶段会处理 hourly_stats 阶段的输出。它会计算活动量超过 $hourly_stats 指定阈值的 principal.ip 和 target.ip 对的统计信息。然后，它会过滤掉高活动状态超过三小时的 IP 地址对。

stage hourly_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $src_ip = principal.ip
  $dst_ip = target.ip
  $src_ip != ""
  $dst_ip != ""

  match:
    $src_ip, $dst_ip by hour

  outcome:
    $count = count(metadata.id)
    $avg_recd_bytes = avg(network.received_bytes)
    $std_recd_bytes = stddev(network.received_bytes)

  condition:
    $avg_recd_bytes > 100 and $std_recd_bytes > 50
}

$src_ip = $hourly_stats.src_ip
$dst_ip = $hourly_stats.dst_ip
$time_bucket_count = strings.concat(timestamp.get_timestamp($hourly_stats.window_start), "|", $hourly_stats.count)

match:
 $src_ip, $dst_ip

outcome:
 $list = array_distinct($time_bucket_count)
 $count = count_distinct($hourly_stats.window_start)

condition:
 $count > 3

如果您按如下方式更改根阶段中的匹配条件，则可以为多阶段查询引入按天进行的窗口化聚合。

match:
 $src_ip, $dst_ip by day

示例：搜索异常活跃的网络连接（使用 Z 分数）

此多阶段查询使用 Z 分数计算（衡量与平均值的标准差数）将每日平均网络活动与今天的活动进行比较。此查询实际上是在搜索内部资产和外部系统之间异常高的网络活动。

前提条件：查询时间窗口必须大于或等于 2 天 并且包含当前日期，以便计算出的 Z 分数有效。

此多阶段查询包含 daily_stats 阶段和 root 阶段，这两个阶段协同工作以计算网络活动的 Z 分数：

• daily_stats 阶段执行初始每日聚合。它会计算每个 IP 地址对（source 和 target）每天交换的总字节数，并返回以下阶段字段（与输出行中的列相对应）：

  • $daily_stats.source：单数，字符串
  • $daily_stats.target：单数，字符串
  • $daily_stats.exchanged_bytes：单数，整数
  • $daily_stats.window_start：单数，整数
  • $daily_stats.window_end：单数，整数

• 根阶段会聚合每个 IP 地址对的 daily_stats 阶段输出。它会计算整个搜索范围内每日交换字节数的平均值和标准差，以及今天交换的字节数。它会使用这三个计算出的值来确定 Z 分数。

• 输出会列出今天所有 IP 地址对的 Z 分数，并按降序排序。

// Calculate the total bytes exchanged per day by source and target

stage daily_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $source = principal.hostname
  $target = target.ip
  $source != ""
  $target != ""

  match:
    $source, $target by day

  outcome:
    $exchanged_bytes = sum(network.sent_bytes + network.received_bytes)
}

// Calculate the average per day over the time window and compare with the bytes exchanged today

$source = $daily_stats.source
$target = $daily_stats.target
$date = timestamp.get_date($daily_stats.window_start)

match:
  $source, $target

outcome:
  $today_bytes = sum(if($date = timestamp.get_date(timestamp.current_seconds()), cast.as_int($daily_stats.exchanged_bytes), 0))
  $average_bytes = window.avg($daily_stats.exchanged_bytes)
  $stddev_bytes = window.stddev($daily_stats.exchanged_bytes)
  $zscore = ($today_bytes - $average_bytes) / $stddev_bytes

order:
  $zscore desc

从阶段导出未聚合的变量

在典型的多阶段查询中，数据通常通过聚合过程在阶段之间传递，该过程可以将多个事件“折叠”为单个摘要。但是，在某些情况下，您需要保留每个事件的具体详细信息（例如唯一进程 ID 或特定命令行），而不会丢失该粒度。为了支持这一点，您可以直接导出变量，而无需使用分组函数。

命名阶段可以包含未聚合的 outcome 部分。这意味着在该 outcome 部分中定义的变量会直接从阶段输出，让后续阶段可以将其作为阶段字段进行访问，而无需进行分组聚合。

此详细信息很有用，因为它具有以下优点：

• 保留数据保真度：您可以将事件的确切属性（例如特定文件路径）传递到下一阶段，而无需使用人工“占位符”聚合（如 max() 或 array_distinct()）。
• 降低查询复杂性：它简化了 YARA-L 逻辑，无需使用 match 部分或分组语句，只需将值从阶段 1 传输到阶段 2。
• 优化性能：通过绕过聚合引擎，系统可以更高效地在阶段之间流式传输数据，从而缩短复杂、大容量搜索的执行时间。

示例：导出未聚合的变量

此示例演示了如何导出未聚合的变量。请注意以下逻辑：

• top_5_bytes_sent 阶段会搜索网络活动最高的五个事件。

• top_5_bytes_sent 阶段会输出以下阶段字段（与输出行中的列相对应）：

  • $top_5_bytes_sent.bytes_sent：单数，整数
  • $top_5_bytes_sent.timestamp_seconds：单数，整数

• root 阶段会计算网络活动最高的五个事件的最新和最早时间戳。

stage top_5_bytes_sent {
  metadata.event_type = "NETWORK_CONNECTION"
  network.sent_bytes > 0

  outcome:
    $bytes_sent = network.sent_bytes
    $timestamp_seconds = metadata.event_timestamp.seconds

  order:
    $bytes_sent desc

  limit:
    5
}

outcome:
  $latest_timestamp = timestamp.get_timestamp(max($top_5_bytes_sent.timestamp_seconds))
  $earliest_timestamp = timestamp.get_timestamp(min($top_5_bytes_sent.timestamp_seconds))

在多阶段查询中实现窗口化

在多阶段检测中，窗口化可让您为阶段内的事件关联定义特定的时间边界。通过将数据划分为离散的时间存储分区（例如 5 分钟的滑动窗口或 1 小时的翻滚窗口），您可以识别暴力破解攻击或信标行为等模式，这些模式只有在作为时间限定序列进行分析时才会显示出来。

多阶段查询在命名阶段中支持所有类型的窗口化（跳跃窗口、滑动窗口和翻滚窗口）。这有助于您在阶段之间传递时间情境化数据，例如使用阶段 1 识别高频事件窗口，并使用阶段 2 将该特定窗口与后续管理操作相关联。

如果命名阶段包含窗口，则可以使用以下预留字段访问每个输出行的窗口开始时间和窗口结束时间：

• $stage_window_start：标记窗口开始的 Unix 时间戳。
• $stage_window_end：标记窗口结束的 Unix 时间戳。

如需详细了解窗口化，请参阅 YARA-L 2.0 窗口化逻辑。

常见使用场景

• 顺序检测：将检测到的登录失败次数峰值的具体时间范围传递到第二个阶段，该阶段会查找此后不久的成功登录。
• 时长分析：通过将初始利用阶段的 $stage_window_start 与后续阶段中事件的时间戳进行比较，计算“入侵时间”。
• 历史基准化：使用窗口将当前事件计数与上一个窗口的输出变量进行比较。

示例：跳跃窗口

以下示例说明了如何在多阶段查询中使用跳跃窗口：

• hourly_stats 阶段会搜索在同一小时内具有高网络活动的 IP 地址对。

• hourly_stats 会输出以下阶段字段（与输出行中的列相对应）：

  • $hourly_stats.src_ip：单数，字符串
  • $hourly_stats.dst_ip：单数，字符串
  • $hourly_stats.count：单数，整数
  • $hourly_stats.std_recd_bytes：单数，浮点数
  • $hourly_stats.avg_recd_bytes：单数，浮点数
  • $hourly_stats.window_start：单数，整数
  • $hourly_stats.window_end：单数，整数

• 根阶段会过滤掉高活动状态超过 3 小时的 IP 地址对。由于在 hourly_stats 阶段使用了跳跃窗口，因此这些小时可能会重叠。

stage hourly_stats {
  metadata.event_type = "NETWORK_CONNECTION"
  $src_ip = principal.ip
  $dst_ip = target.ip
  $src_ip != ""
  $dst_ip != ""

  match:
    $src_ip, $dst_ip over 1h

  outcome:
    $count = count(metadata.id)
    $avg_recd_bytes = avg(network.received_bytes)
    $std_recd_bytes = stddev(network.received_bytes)

  condition:
    $avg_recd_bytes > 100 and $std_recd_bytes > 50
}

$src_ip = $hourly_stats.src_ip
$dst_ip = $hourly_stats.dst_ip
$time_bucket_count = strings.concat(timestamp.get_timestamp($hourly_stats.window_start), "|", $hourly_stats.count)

match:
 $src_ip, $dst_ip

outcome:
 $list = array_distinct($time_bucket_count)
 $count = count_distinct($hourly_stats.window_start)

condition:
 $count > 3

多阶段查询中的内联接

内联接可让您关联不同阶段或来源类型的数据，从而创建复杂的分析工作流，例如将实时事件与预先计算的统计基准进行比较。通过联接阶段，您可以使用有状态数据（例如中位数或查找表）丰富原始遥测数据，以识别单个阶段事件过滤器会遗漏的离群值或多向量威胁。

多阶段查询的阶段内和阶段间都支持内联接。内联接功能支持以下类型：

• UDM 和 UDM：关联两组不同的安全事件（例如，将登录事件与后续文件访问进行匹配）。
• UDM 和 ECG：将事件数据与实体情境图信息合并，以丰富身份或资产。
• UDM 和数据表：将实时事件与静态或上传的参考列表（例如，高价值资产列表或部门专用 IP 地址范围）联接。

以下示例展示了如何在 UDM 事件和计算出的表阶段之间配置无匹配联接（在 outcome 或 events 部分而不是 match 部分中执行的联接）。借助此模式，您可以执行统计异常检测，如以下平均绝对偏差 (MAD) 计算所示：

• median 阶段：计算每个来源主机和目标 IP 地址对的中位数发送字节数。
  • $median.host：单数，字符串
  • $median.target：单数，字符串
  • $median.median：单数，浮点数
• absolute_deviations 阶段：将每个 UDM 事件与中位数阶段中的相应行联接。这样，您就可以计算每个单独事件相对于其对等组的发送字节数的绝对偏差。
  • $absolute_deviations.host：单数，字符串
  • $absolute_deviations.target：单数，字符串
  • $absolute_deviations.absolute_deviation：单数，浮点数
• 根阶段：计算所有 UDM 事件中这些绝对偏差的平均值，以建立异常阈值。

示例：配置无匹配联接

stage median {
  metadata.event_type = "NETWORK_CONNECTION"
  $host = principal.hostname
  $target = target.ip

  match:
    $host, $target

  outcome:
    $median = window.median(network.sent_bytes, true)
}

stage absolute_deviations {
  metadata.event_type = "NETWORK_CONNECTION"
  $join_host = principal.hostname
  $join_host = $median.host
  $join_target = target.ip[0]
  $join_target = $median.target

  outcome:
    $host = $join_host
    $target = $join_target
    $absolute_deviation = math.abs(network.sent_bytes - $median.median)
}

$host = $absolute_deviations.host
$target = $absolute_deviations.target

match:
  $host, $target

outcome:
  $mean_absolute_deviation = avg($absolute_deviations.absolute_deviation)

示例：窗口化阶段和表阶段之间的无匹配联接

以下示例说明了如何在多阶段查询中配置窗口化阶段和表阶段之间的无匹配联接。

• hourly_stats 阶段会计算每个来源和目标主机对以及小时存储桶的总发送字节数。
• hourly_stats 阶段会输出以下阶段字段（与输出行中的列相对应）：
  • $hourly_stats.source_host：单数，字符串
  • $hourly_stats.dst_host：单数，字符串
  • $hourly_stats.total_bytes_sent：单数，浮点数
  • $hourly_stats.window_start：单数，整数
  • $hourly_stats.window_end：单数，整数
• agg_stats 阶段会计算每个来源和目标主机对每小时的平均字节数和标准差。

• agg_stats 会输出以下阶段字段（与输出行中的列相对应）：

  • $agg_stats.source_host：单数，字符串
  • $agg_stats.dst_host：单数，字符串
  • $agg_stats.avg_bytes_sent：单数，浮点数
  • $agg_stats.stddev_bytes_sent：单数，浮点数

• 根阶段会将 hourly_stats 中的每一行与 agg_stats 中同一来源和目标主机对的行联接。对于每个来源和目标主机对，它会使用该主机对存储桶的总发送字节数和汇总统计信息来计算 z 分数。

stage hourly_stats {
 $source_host = principal.hostname
 $dst_host = target.hostname
 principal.hostname != ""
 target.hostname != ""
 match:
   $source_host, $dst_host by hour
 outcome:
   $total_bytes_sent = sum(network.sent_bytes)
}

stage agg_stats {
  $source_host = $hourly_stats.source_host
  $dst_host = $hourly_stats.dst_host
  match:
    $source_host, $dst_host
  outcome:
   $avg_bytes_sent = avg($hourly_stats.total_bytes_sent)
   $stddev_bytes_sent = stddev($hourly_stats.total_bytes_sent)
}

$source_host = $agg_stats.source_host
$source_host = $hourly_stats.source_host

$dst_host = $agg_stats.dst_host
$dst_host = $hourly_stats.dst_host

outcome:
  $hour_bucket = timestamp.get_timestamp($hourly_stats.window_start)
  $z_score = ($hourly_stats.total_bytes_sent - $agg_stats.avg_bytes_sent)/$agg_stats.stddev_bytes_sent

多阶段查询中的交叉联接

使用 Google SecOps 搜索或信息中心时，多阶段查询中的交叉联接可让您将单个 UDM 事件数据与在其他 YARA-L 阶段中计算出的汇总统计信息进行比较。

在 YARA-L 中，cross join 关键字适用于仅返回一行的阶段。

当在限制为 1 的阶段和另一个数据集（例如 UDM 事件）之间使用交叉联接时，该阶段的单行输出会附加到另一个数据集的每一行。这样可以丰富包含总体统计信息的事件数据。

示例：查找异常登录活动

以下示例用于识别登录频率高于正常水平的用户。它通过将每个用户的登录计数（使用 user_login_counts 阶段）与所有用户的平均登录计数（使用 total_users 阶段）进行比较来计算此值。登录次数异常多的用户可以在搜索结果中排序。

然后，您可以使用交叉联接关键字将 total_users 阶段的结果与 user_login_counts 阶段的结果相关联。

stage user_login_counts {
    $user = principal.user.userid
    metadata.event_type = "USER_LOGIN"
    security_result.action = "ALLOW"

    match:
        $user

    outcome:
        $login_count = count(metadata.id)
}

stage total_users {
    outcome:
        $count = count($user_login_counts.user)
    limit:
        1
}

cross join $total_users, $user_login_counts

$login_count = $user_login_counts.login_count
$user = $user_login_counts.user
$tot_users = $total_users.count

// all users who logged in the same number of times are grouped together.
match:
    $login_count
outcome:
    $num_users = count($user)
    $frequency_percent = (count($user) / max($tot_users) ) * 100

限制

多阶段查询具有以下功能和结构限制：

结构要求

构建查询时，您必须遵循以下结构要求：

• 根阶段：每个查询只允许有一个根阶段。
• 命名阶段：最多支持四个命名阶段。
• 阶段引用：阶段只能引用同一查询中逻辑上在其之前定义的阶段。
• 交叉联接：交叉联接只能引用返回单行的阶段。您必须在引用的阶段中添加限制 (1) 才能满足此要求。这很有用，因为它可以让您将单个全局统计信息（例如最大值或平均值）附加到每个单独的事件行以进行比较。
• 联接：所有阶段最多允许四个非数据表联接。
• 结果要求：每个命名阶段（根阶段除外）都必须包含 match 部分或 outcome 部分，其中 outcome 部分不需要聚合。

窗口和兼容性限制

以下限制适用于您使用窗口的方式以及运行查询的位置：

• 功能支持：多阶段查询可在搜索和信息中心中使用，但规则不支持此功能。
• 窗口类型：避免在单个查询中混用不同的窗口类型。
• 窗口依赖项：使用跳跃窗口或滑动窗口的阶段不能依赖于也使用跳跃窗口或滑动窗口的另一个阶段。
• 翻滚窗口大小：虽然不同阶段的翻滚窗口大小可能不同，但大小差异必须小于 720 倍。

示例：阶段聚合差异（无效）

以下配置无效，因为一个月包含 44,640 分钟（44,640 / 1 > 720 ）：

阶段: monthly_stats { ... match: by month }

根: match: by minute

示例：阶段聚合差异（有效）

如需解决此问题，请确保阶段之间的比率较小。例如，将每小时数据聚合到每日报告中：

阶段: daily_stats { ... match: by day }

根: match: by hour

由于 24（一天中的小时数）小于 720，因此系统可以高效地将阶段数据映射到根阶段。

阶段和查询限制

多阶段查询中的每个单独阶段都有特定的限制。适用于单阶段查询的大多数限制也适用于每个单独阶段：

• 输出要求：每个阶段都必须输出至少一个匹配或结果变量（阶段字段）。

• 查询时间范围：

  • 标准查询：最长为 30 天。
  • 具有无匹配联接的多阶段查询：最长限制为 14 天。

• 窗口大小限制：窗口（跳跃窗口、滑动窗口或翻滚窗口）的最大大小取决于查询是否包含联接：

  • 包含联接：任何类型（跳跃窗口、滑动窗口或翻滚窗口）的最大窗口大小为 2 天。如需了解详情，请参阅搜索联接限制。

  • 不包含联接（单个事件）：

  • 跳跃窗口和滑动窗口：最长为 2 天。

  • 翻滚窗口：最长增加到 30 天。

• 结果变量上限：

  • 默认值为 20
  • 对于选择使用较大限制的客户，为 50
  • 数组限制：数组值结果变量中最多允许 10,000 个元素。

• 每个查询的事件限制：

  • 最多两个 UDM 事件
  • 最多一个 ECG 事件
  • 最多两个数据表

注意： 您无法在单个查询中将数据表、UDM 和 ECG 事件联接在一起。

服务和性能限制

多阶段查询受到与统计信息查询相同的限制：

• 统计信息查询：120 QPH（API 和界面）。
• 搜索视图：每分钟 100 个视图。
• API 支持：Google SecOps 系统和 EventService.UDMSearch API 支持多阶段联接，但 SearchService.UDMSearch API 不支持。系统还允许您运行不包含联接的多阶段查询。

事件和全局限制

您必须遵守以下事件和平台级限制：

事件上限

多阶段查询严格限制可以同时处理的事件数量：

• UDM 事件：最多允许 2 个 UDM 事件。
• 实体情境图 (ECG) 事件：最多允许 1 个 ECG 事件。

全局查询限制

这些平台级限制控制着多阶段查询返回的数据回溯时间和数据量：

• 查询时间范围：标准查询的最长时间范围为 30 天。
• 结果集总数：结果集总数上限为 10,000 个结果。

需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。