使用外联接关联数据
本文档介绍了外联接(左联接和右联接)。联接操作用于根据共同的字段值关联和合并来自多个来源的数据。通过将相关安全性事件和实体合并到单个综合视图中,您可以提供有效的威胁检测和调查。
与标准(内)联接不同,外联接会从联接的一侧检索所有记录,即使另一侧没有匹配的条目也是如此。另一侧不匹配的字段通常会填充 null。这样可以防止您丢失不匹配的数据。
外联接的工作原理
YARA-L 2.0 中的外联接概念与标准 SQL 外联接相同:
左外联接会保留联接左侧的所有记录。
右外联接会保留联接右侧的所有记录。
所有查询(无论是否包含 match 条件)都支持外联接语法(左联接和右联接)。外联接的基本行为在搜索、信息中心和检测规则中是相同的,但语法会因上下文而异:
搜索和信息中心:在
events部分中使用显式left join或right join关键字。检测规则:在
events部分中使用等号 (=) 定义联接关系,并在condition部分中指定无界限条件(例如,#data_table >= 0表示左外联接,#data_table = 0表示反联接排除)。
了解左外联接
无论您是在搜索中使用 left join 关键字,还是在检测规则中使用无界限条件 (#data_table >= 0),左外联接(或左联接)都会保留联接左侧(保留)数据源中的所有记录。
如果左侧的记录在右侧事件或数据表中没有匹配项,则右侧的字段将返回为
null。占位符含义:
match部分中使用的任何占位符变量都必须引用左侧事件中的字段,以确保在整个结果集中准确汇总数据。
事件到事件左联接示例
以下示例演示了左外联接,用于将用户登录事件与在同一主机上发生的后续网络连接事件相关联。
左联接可确保结果集中保留所有 USER_LOGIN 事件。如果找到匹配的 NETWORK_CONNECTION 事件 ($e2),则会合并其数据。如果未找到匹配项,则 $e2 的字段为 null。
$e1.metadata.event_type = "USER_LOGIN"
$e2.metadata.event_type = "NETWORK_CONNECTION"
left join $e1.principal.hostname = $e2.principal.hostname
定义左侧事件
以下查询示例定义了联接的左侧 ($e1),即最终结果中保留的事件集:
$e1.metadata.event_type = "USER_LOGIN"
下表表示查询结果,用于标识初始用户登录事件集:
| 事件类型 | 主账号主机名 | IP 地址 |
|---|---|---|
USER_LOGIN
|
workstation-01
|
192.168.1.101
|
USER_LOGIN
|
laptop-hr-02
|
192.168.1.102
|
USER_LOGIN
|
server-db-03
|
10.0.0.50
|
USER_LOGIN
|
kiosk-4
|
192.168.1.104
|
定义合适的事件
以下查询示例定义了联接的右侧 ($e2),即与左侧事件匹配的事件集:
$e2.metadata.event_type = "NETWORK_CONNECTION"
下表列出了可用于匹配的网络连接事件集:
| 事件类型 | 主账号主机名 | IP 地址 |
|---|---|---|
NETWORK_CONNECTION
|
workstation-01
|
192.168.1.101
|
NETWORK_CONNECTION
|
laptop-hr-02
|
192.168.1.101
|
NETWORK_CONNECTION
|
kiosk-4
|
203.0.113.3
|
参加活动
包含匹配部分
以下示例演示了在 principal.hostname 字段上使用左外联接的匹配查询:
$e1.metadata.event_type = "USER_LOGIN"
$e2.metadata.event_type = "NETWORK_CONNECTION"
left join $e1.principal.hostname = $e2.principal.hostname
$host = $e1.principal.hostname
match:
$host over 5m
左外连接可确保每个
USER_LOGIN事件 ($e1) 都包含在最终结果集中。占位符 $host 被赋予
$e1.principal.hostname中的值。 左外联接可确保事件$e1存在,从而确保始终填充 $host 变量以进行聚合。该规则会按主机汇总 5 分钟时间窗口内的结果。
联接结果
生成的数据会显示这两个事件的组合。左侧表 ($e1) 中的所有记录都会保留,如果未找到匹配的主机名(例如,对于 server-db-03),则右侧表 ($e2) 中的字段会设置为 null。
事件类型 ($e1)
|
主账号主机名 ($host)
|
IP 地址 ($e1)
|
事件类型 ($e2)
|
IP 地址 ($e2)
|
匹配状态 |
|---|---|---|---|---|---|
USER_LOGIN
|
workstation-01
|
192.168.1.101
|
NETWORK_CONNECTION
|
192.168.1.101
|
找到匹配项 |
USER_LOGIN
|
laptop-hr-02
|
192.168.1.102
|
NETWORK_CONNECTION
|
192.168.1.101
|
找到匹配项 |
USER_LOGIN
|
server-db-03
|
10.0.0.50
|
null
|
null
|
无匹配 |
USER_LOGIN
|
kiosk-4
|
192.168.1.104
|
NETWORK_CONNECTION
|
203.0.113.3
|
找到匹配项 |
左联接查询示例
本部分提供了左联接查询示例。
具有匹配条件的联接
事件实体
$e1.metadata.event_type = "NETWORK_CONNECTION" $g1.graph.metadata.entity_type = "ASSET" left join $e1.principal.asset.hostname = $g1.graph.entity.asset.hostname $host = $e1.principal.asset.hostname match: $host over 5mEvent-datatable
$host = $e1.principal.hostname left join $e1.principal.hostname = %all_dt_column_types.hostname match: $host by 5m
不含匹配条件的联接
事件-事件
$e1.metadata.event_type = "USER_LOGIN" $e1.principal.ip = "114.241.96.87" $e2.metadata.event_type = "NETWORK_CONNECTION" left join $e1.principal.hostname = $e2.principal.hostname事件实体
$e1.metadata.event_type = "NETWORK_CONNECTION" $g1.graph.metadata.entity_type = "ASSET" left join $e1.principal.asset.hostname = $g1.graph.entity.asset.hostname $host = $e1.principal.asset.hostnameEvent-datatable
$host = $e1.principal.hostname left join $e1.principal.hostname = %all_dt_column_types.hostname
检测规则中的左联接
在检测规则中,您不会在 events 部分中使用 left join 关键字。而是使用等号 (=) 将 UDM 事件与 events 部分中的数据表或次要事件相关联,并在 condition 部分中控制联接行为:
| 联接类型 | events 部分 |
condition 部分 |
不匹配事件的行为 |
|---|---|---|---|
| 内联接 | $key = %data_table.key |
$e and #data_table > 0 |
不匹配的事件会从检测流中丢弃。 |
| 左外连接(丰富) | $key = %data_table.key |
$e and #data_table >= 0 |
$e 中的所有事件都会保留;不匹配的数据表字段为 null。 |
| 反联接(排除) | $key = %data_table.key |
$e and #data_table = 0 |
仅当事件键在数据表中不存在时,才会触发提醒。 |
规则中的事件到数据表左联接示例
以下示例使用名为 corporate_asset_directory 的数据表,通过资产关键性和地理元数据丰富事件,而不会丢弃来自非代管式主机的事件。
定义数据表
corporate_asset_directory 数据表包含以下行,并使用 hostname 作为键列:
hostname(密钥) |
country_code |
country_name |
asset_tier |
owner_dept |
risk_score |
|---|---|---|---|---|---|
ES01-W10 |
ES |
Spain |
TIER_3_WORKSTATION |
Finance |
20 |
CH02-SRV |
CH |
Switzerland |
TIER_1_CRITICAL |
Engineering |
80 |
US03-LAP |
US |
United States |
TIER_2_PRODUCTION |
Executive |
50 |
比较内联接和左外联接条件
内联接 (
#corporate_asset_directory > 0):如果要求数据表数量大于0,则会执行内联接。如果事件源自corporate_asset_directory中未列出的非代管式主机(例如unmanaged-byod-09),则该事件会被滤除,并且不会生成检测结果:rule inner_join_drops_unmatched_events { meta: author = "Google SecOps" description = "Inner join example: drops events whose hostname is not in the data table." events: $e.principal.hostname != "" $hostname = $e.principal.hostname // Direct inner join against data table $hostname = %corporate_asset_directory.hostname match: $hostname over 1h outcome: $country = array_distinct(%corporate_asset_directory.country_code) $asset_tier = array_distinct(%corporate_asset_directory.asset_tier) $risk_score = max(%corporate_asset_directory.risk_score) condition: $e and #corporate_asset_directory > 0 }左外联接 (
#corporate_asset_directory >= 0):使用无界限条件#corporate_asset_directory >= 0会执行左外联接。即使主机名与corporate_asset_directory中的行不匹配,$e中的所有事件也会保留。如果事件与数据表行不匹配,则对
%corporate_asset_directory.column的引用会求值为null。在outcome部分中,对字符串列使用strings.coalesce,对数值列使用条件if表达式,以分配默认后备值:rule left_outer_join_enrichment { meta: author = "Google SecOps" description = "Left outer join using #dt >= 0: preserves all events and populates enrichment fields." events: $e.principal.hostname != "" $hostname = $e.principal.hostname // Outer join equality binding $hostname = %corporate_asset_directory.hostname match: $hostname over 1h outcome: $country = array_distinct(strings.coalesce(%corporate_asset_directory.country_code, "UNKNOWN")) $asset_tier = array_distinct(strings.coalesce(%corporate_asset_directory.asset_tier, "UNMANAGED")) $risk_score = max(if(%corporate_asset_directory.risk_score > 0, %corporate_asset_directory.risk_score, 0)) condition: $e and #corporate_asset_directory >= 0 }
规则联接结果
下表比较了使用 #corporate_asset_directory >= 0 时匹配和不匹配的主机的结果值:
主体主机名 ($hostname) |
数据表匹配状态 | $country |
$asset_tier |
$risk_score |
|---|---|---|---|---|
CH02-SRV |
找到匹配项 | ["CH"] |
["TIER_1_CRITICAL"] |
80 |
ES01-W10 |
找到匹配项 | ["ES"] |
["TIER_3_WORKSTATION"] |
20 |
unmanaged-byod-09 |
不匹配(null已替换为默认值) |
["UNKNOWN"] |
["UNMANAGED"] |
0 |
规则中的左反联接(排除)示例
您可以使用具有零计数条件 (#data_table = 0) 的左外联接来执行左反联接。当事件与排除数据表匹配时,此模式会抑制检测,并且仅在不存在匹配项时触发提醒。
单列排除 (
#authorized_scanners_exclusion = 0):以下规则将$e.principal.hostname与authorized_scanners_exclusion数据表联接,并且仅当主机不存在于表中时才会触发:rule anti_join_scanner_exclusion { meta: author = "Google SecOps" description = "Anti-join exclusion using #dt = 0: triggers only if the host is not in the authorized scanners table." events: $e.principal.hostname != "" $hostname = $e.principal.hostname // Bind join key to data table $hostname = %authorized_scanners_exclusion.hostname match: $hostname over 1h outcome: $total_sent_bytes = sum($e.network.sent_bytes) condition: $e and #authorized_scanners_exclusion = 0 }或者,您也可以在
outcome部分 ($scanner_match_count = count(%authorized_scanners_exclusion.hostname)) 中统计匹配的数据表行数,并在condition部分中评估$e and $scanner_match_count = 0:rule anti_join_outcome_count { meta: author = "Google SecOps" description = "Anti-join via outcome variable count check: triggers when scanner match count is zero." events: $e.principal.hostname != "" $hostname = $e.principal.hostname $hostname = %authorized_scanners_exclusion.hostname match: $hostname over 1h outcome: $total_sent_bytes = sum($e.network.sent_bytes) $scanner_match_count = count(%authorized_scanners_exclusion.hostname) condition: $e and $scanner_match_count = 0 }多列行元组排除 (
#multi_column_admin_exceptions = 0):如需仅在多个字段的特定组合与异常数据表中的同一行匹配时排除事件,请将每个事件字段绑定到events部分中的对应列,并在condition部分中指定#data_table = 0:rule multi_column_tuple_exclusion { meta: author = "Google SecOps" description = "Multi-column row-tuple anti-join: excludes activity matching composite (origin_ip, target_ip, username, action) rows." events: // Bind event fields to match variables $e.principal.ip = $src_ip $e.target.ip = $dst_ip $e.principal.user.userid = $user $e.metadata.product_event_type = $action // Multi-column equality against the same data table row $src_ip = %multi_column_admin_exceptions.origin_ip $dst_ip = %multi_column_admin_exceptions.target_ip $user = %multi_column_admin_exceptions.username $action = %multi_column_admin_exceptions.action match: $src_ip, $dst_ip, $user, $action over 5m outcome: $event_count = count($e.metadata.id) condition: $e and #multi_column_admin_exceptions = 0 }
右外联接
右外联接(或右联接)会保留 right join 关键字右侧数据源中的所有记录。
如果右侧事件中的记录在左侧事件中没有匹配项,则左侧事件中的字段将返回为
null。占位符含义:
match部分中使用的任何占位符变量都必须引用右侧事件中的字段,以确保在整个结果集中准确汇总数据。
事件到事件右联接示例
以下示例演示了如何使用 right outer join 将用户登录事件与同一主机上发生的后续网络连接事件相关联。right join 可确保结果集中保留所有 NETWORK_CONNECTION 事件。如果找到匹配的 USER_LOGIN 事件,则会联接其数据。
如果未找到匹配项,则$e1的字段为 null。
$e1.metadata.event_type = "USER_LOGIN"
$e2.metadata.event_type = "NETWORK_CONNECTION"
right join $e1.principal.hostname = $e2.principal.hostname
定义左侧事件
以下查询定义了联接的左侧 ($e1),即最终结果中的可选事件集:
$e1.metadata.event_type = "USER_LOGIN"
下表表示查询结果,其中标识了初始用户登录事件集:
| 事件类型 | 主账号主机名 | IP 地址 |
|---|---|---|
USER_LOGIN
|
workstation-01
|
192.168.1.101
|
USER_LOGIN
|
laptop-hr-02
|
192.168.1.102
|
USER_LOGIN
|
server-db-03
|
10.0.0.50
|
定义合适的事件
以下查询定义了联接的右侧 ($e2),即最终结果中保留的事件集。
$e2.metadata.event_type = "NETWORK_CONNECTION"
下表列出了可用于匹配的网络连接事件集。
| 事件类型 | 主账号主机名 | IP 地址 |
|---|---|---|
NETWORK_CONNECTION
|
workstation-01
|
192.168.1.101
|
NETWORK_CONNECTION
|
laptop-hr-02
|
192.168.1.101
|
NETWORK_CONNECTION
|
vm-unauth-05
|
203.0.113.3
|
参加活动
以下示例展示了对 principal.hostname 字段进行右外联接的匹配查询:
$e1.metadata.event_type = "USER_LOGIN"
$e2.metadata.event_type = "NETWORK_CONNECTION"
right join $e1.principal.hostname = $e2.principal.hostname
$host = $e1.principal.hostname
match:
$host over 5m
右外连接可确保最终结果集中包含每个
NETWORK_CONNECTION事件 ($e2)。占位符
$host被赋予了$e2.principal.hostname中的值。 右外联接可确保事件$e2存在,从而确保始终填充$host变量以进行汇总。该规则会按主机汇总 5 分钟时间窗口内的结果。
联接结果
生成的数据集会显示这两个事件的组合。右侧表 ($e2) 中的所有记录都会保留,如果未找到匹配的主机名(例如 vm-unauth-05),则左侧表 ($e1) 中的字段会设置为 null。
事件类型 ($e1)
|
主账号主机名 ($e1)
|
IP 地址 ($e1)
|
事件类型 ($e2)
|
主体主机名 ($host) | IP 地址 ($e2)
|
匹配状态 |
|---|---|---|---|---|---|---|
USER_LOGIN
|
workstation-01
|
192.168.1.101
|
NETWORK_CONNECTION
|
workstation-01
|
192.168.1.101
|
找到匹配项 |
USER_LOGIN
|
laptop-hr-02
|
192.168.1.102
|
NETWORK_CONNECTION
|
laptop-hr-02
|
192.168.1.101
|
找到匹配项 |
null
|
null
|
null
|
NETWORK_CONNECTION
|
vm-unauth-05
|
203.0.113.4
|
无匹配 |
右联接查询示例
本部分提供了右联接查询示例。
具有匹配条件的联接
事件-事件
$e1.metadata.event_type = "USER_LOGIN" $e2.metadata.event_type = "NETWORK_CONNECTION" right join $e1.principal.hostname = $e2.principal.hostname $host = $e2.principal.hostname match: $host over 5m实体-事件
$e1.metadata.event_type = "NETWORK_CONNECTION" $g1.graph.metadata.entity_type = "ASSET" right join $g1.graph.entity.asset.hostname = $e1.principal.asset.hostname $host = $e1.principal.asset.hostname match: $host over 5mDatatable-event
$host = $e1.principal.hostname right join %all_dt_column_types.hostname = $e1.principal.hostname match: $host by 5m
不含匹配条件的联接
事件-事件
$e1.metadata.event_type = "USER_LOGIN" $e1.principal.ip = "114.241.96.87" $e2.metadata.event_type = "NETWORK_CONNECTION" right join $e1.principal.hostname = $e2.principal.hostname实体-事件
$e1.metadata.event_type = "NETWORK_CONNECTION" $g1.graph.metadata.entity_type = "ASSET" right join $g1.graph.entity.asset.hostname = $e1.principal.asset.hostname $host = $e1.principal.asset.hostnameDatatable-event
$host = $e1.principal.hostname right join %all_dt_column_types.hostname = $e1.principal.hostname
限制
创建外联接时,请考虑以下限制:
不支持全外联接(左联接和右联接)。
无匹配联接的查询时间范围上限为 14 天。
您无法直接联接两个情境来源(例如,将实体直接联接到数据表)。
主统一数据模型 (UDM) 事件必须是外联接的保留端。如果主要事件位于“可为 null”的一侧,则查询无效。
事件-实体联接必须是左联接。这会正确保留事件 (
$e1)。实体-事件联接必须是右联接。这会正确保留事件 (
$e1)。
以下示例无效,因为 UDM 事件 (
$e1) 位于左侧,但右联接保留了右侧 ($g1),这违反了必须保留 UDM 事件的规则:// Invalid query $e1.metadata.event_type = "NETWORK_CONNECTION" $g1.graph.metadata.entity_type = "ASSET" right join $e1.principal.asset.hostname = $g1.graph.entity.asset.hostname检测规则不支持
events部分中的显式left join和right join关键字,仅 YARA-L 搜索和信息中心支持这些关键字。如需在检测规则中执行左外联接或反联接,请在condition部分中使用无界限条件,例如#data_table >= 0或#data_table = 0。如需了解详情,请参阅检测规则中的左联接。
最佳做法
为防止外联接查询中出现性能缓慢和查询超时的情况,请使用具体且范围较窄的过滤条件。
例如,以下宽泛查询:
$e1.metadata.event_type = "USER_LOGIN"
$e2.metadata.event_type = "NETWORK_CONNECTION"
right join $e1.principal.hostname = $e2.principal.hostname
可通过添加以下特定条件进行优化:
$e1.metadata.event_type = "USER_LOGIN"
$e1.principal.ip = "121.121.121.121"
$e1.principal.user.userid = "alex"
$e2.metadata.event_type = "NETWORK_CONNECTION"
$e2.src.hostname = "altostrat.com"
$e1.principal.hostname = $e2.principal.hostname