文章总结： 文章介绍了检测工程中规则调优自动化的实践方法，重点讲解了如何利用MicrosoftSentinel的Watchlists功能简化检测规则的调优过程。文章详细阐述了Watchlist的组织方式、管理方法以及如何通过自动化流程实现检测规则的持续优化，减少手动工作量并提高检测效率。作者还提供了具体的代码示例和实施步骤，展示了如何通过构建验证和持续部署流水线实现检测规则的自动化调优。
综合评分： 93
文章分类： 安全运营,安全工具,网络安全,安全建设,数据安全

---

检测工程: 实践检测即代码 – 规则调优自动化 – 第8部分

Kristof Baute

securitainment

2025年11月25日 23:05
甘肃

本文是系列文章 Detection Engineering: Practicing Detection-as-Code 的第 8 部分。

---

检测规则应当适应被监控环境中的变化。随着组织通过迁移、网络重新配置、引入新系统或软件、弃用现有系统等方式修改基础设施,检测规则需要持续优化,以保持有效性并将告警队列维持在可管理的水平。创建新的检测规则需要研究、学习和理解新的战术和技术,以及问题解决和创造性思维来应对层出不穷的新兴威胁。相比之下,调优检测规则是一个重复性的过程,相比新规则创建显得单调乏味且缺乏吸引力。

在第 7 部分中,我们展示了如何利用自动化来持续监控已部署检测规则的性能和触发率。在本部分中,我们将探讨如何引入自动化并利用持续部署流水线来简化调优检测规则这一繁琐任务。我们将提供来自 Microsoft Sentinel 的示例,特别是其 Watchlists 功能,但这里提供的概念同样适用于其他 SIEM 平台,因为大多数平台都具有类似功能。

Watchlists

Watchlists 是可以上传并用于丰富检测、威胁狩猎和调查的自定义数据集。例如,如果您有高权限账户、计算机资产、IOC 等列表,可以将它们作为 CSV 文件上传,并在威胁狩猎规则或分析规则中引用。本文将展示如何将它们用于调优和过滤。

在底层,watchlist 项是添加到 Sentinel 特殊表 “Watchlist” 的 JSON 对象。该表被缓存以提升查询性能。您可以使用 _GetWatchlist()函数检索 watchlist 中的元素。

需要注意一些限制。Watchlists 不适用于大数据量。活动 watchlist 项(所有 watchlists 的总和)的最大数量为 1000 万,文件上传(单个 watchlist)限制为 3.8 MB。举个例子:如果您有一个网络范围的 watchlist,单个项约为 40 字节(CIDR 表示法的 IP 地址和范围名称),这意味着可以存储 10 万个网络范围——这已经是相当大的数量了。对于我们的调优检测需求,这些限制足够了。如果需要更大的文件,可以将文件上传到 Azure Storage,但这超出了本文范围。如有需要,请参考 MS 文档。1

保留期为 28 天。这可能令人困惑——这并不意味着您的 watchlist 项会在 28 天后被删除,而是指已删除的项将在 28 天后被清除。为了保持其他项处于活动状态,有一个 12 天的自动刷新间隔。

Watchlist 管理

可以通过 Sentinel Web 界面创建 watchlists。这个过程很简单:提供名称、别名和搜索键,然后指向包含数据的 CSV 文件。别名是您在查询中用来引用 watchlist 的名称。名称可以不同。我们保持它们一致,但可以想象它们具有不同名称的场景。例如,如果您是 MSSP,需要处理多个 watchlists,为 watchlist 设置特定名称可能更方便,例如 Servers_CustomerA、Servers_CustomersB 等,但在规则中使用通用引用,即 “Servers”。

Creating Servers watchlist in Sentinel interface

在 Sentinel 中创建 Servers watchlist

SearchKey 是您打算在联接或查找中使用的列名。它专为此用途设计,以提升查询性能。定义 watchlist 时要考虑这一点。大多数情况下,它将是您想在查询中使用的值。

也可以通过 Web 界面编辑 watchlists,这是一个简单的过程。

由于本文关注 Detection-as-Code,我们以编程方式管理 watchlists。有两个 API:一个用于创建和编辑 watchlists2,另一个用于创建和编辑 watchlist 项3。

这里有一个需要注意的地方——可能很容易采用一种方法,即在 watchlist 更改后直接删除并重新上传。这看起来确实很简单:不必关心不同的可能操作(更新、删除或创建),使集成非常直接。但是,Microsoft 建议不要这样做,因为数据摄取有 5 分钟的 SLA。这意味着当您删除 watchlist 并重新创建时,可能会看到两个版本都处于活动状态,但更重要的是:在该时间窗口内也可能根本没有 watchlist 项。因此,在该窗口期间运行的任何规则都将失败:如果用于黑名单将错过恶意内容,如果用于白名单则会产生误报。

一个可能的解决方法是 “hack” _GetWatchlist()函数,使其也检索在过去 5 分钟内被删除的项,但我们没有采用这种方法。相反,为了将存储在仓库中的 watchlists 与 Sentinel 上的 watchlists 同步,我们构建了一个脚本(watchlist_mgmt.py),它利用 Sentinel 的 watchlist API 来同步 watchlist 项,无需删除和重新创建 watchlists 或修改 _GetWatchlist()函数。稍后我们将在 Azure DevOps 流水线中使用该脚本来为调优过程引入自动化,现在先介绍其功能。

下面是 API 消费者代码的屏幕截图:

API consumer code screenshot

脚本 watchlist_mgmt.py 的 “SyncWatchlistItems” 操作处理 “filters” 目录中定义的 watchlists。对于提供的每个 watchlist 名称,它从 JSON 文件中读取 watchlist 的元数据,并从 CSV 文件中读取其内容。然后,脚本使用这些信息在 Sentinel 中创建或更新 watchlist(同一个 API 端点)。之后,它从 Sentinel 检索现有的 watchlist 项,并将它们与仓库 watchlist 进行比较。如果 Sentinel 中的任何项在仓库中不存在,它们将从平台的 watchlist 中删除。这确保了 Sentinel watchlist 与仓库配置保持一致,移除任何不必要的条目。

下面的屏幕截图显示了上述逻辑的实现:

Watchlist synchronization logic implementation

Watchlist 组织

组织 watchlists 有多种方式,需要仔细考虑,否则很快会变成一团混乱。我们决定将过滤项组织成两种 watchlists:

Content Pack Watchlists

这些包含仅由特定 content pack 中的检测使用的变量(content packs 的概念在第 2 部分中解释)。它们不需要大量前期思考。如果创建一个需要特定变量的规则,只需将变量及其值添加到 watchlist 即可。

有一个特定的技术要求——因为需要为特定检测规则查找键值对,所以需要在规则和 watchlist 中都有对该规则的引用。例如,假设我们有一个 content pack “windows_security”,其中有一个用于 DCSync 的检测规则,我们想要允许特定账户执行同步。规则如下所示,包含对自身的引用,用于仅过滤适用于该规则的变量:

let rule_uuid ="28748697-a290-4367-b67c-57f923e21848";
let AllowedAccounts = _GetWatchlist("windows_builtin")
| where column_ifexists("uuid", "") == rule_uuid
| where column_ifexists("Variable", "") =="AllowedAccounts"
| project SearchKey;
SecurityEvent
|&nbsp;where<DETECTION LOGIC>
|&nbsp;where&nbsp;Account !in&nbsp;(AllowedAccounts)

示例 watchlist 包含 2 个不同规则的过滤项,由 UUID 标识:

| uuid | Variable | Value (*) | Tags |
| — | — | — | — |
| 28748697-a290-4367-b67c-57f923e21848 | AllowedAccounts | CONTOSO\aad_sync | Ticket-12345 |
| 28748697-a290-4367-b67c-57f923e21848 | AllowedAccounts | CONTOSO\sp_sync | Ticket-12345 |
| 806fb458-5eaa-45d7-a8a9-78750c637744 | AnotherVariable | SomeValue | Ticket-12346 |

windows_security watchlist() 定义为 SearchKey*

我们创建的其他列:

• Variable: 在过滤中使用的变量名称
• Value: 其实际值,此列定义为 _SearchKey_
• Tags: 可选列,可用于存储工单 ID 或对过滤过程的其他引用

如果运行 _GetWatchlist("windows_security"),整个 watchlist 如下所示:

Windows security watchlist output example

通过在规则中使用额外条件,它将只返回相关的值。结果存储在单列表 “AllowedAccounts” 中,然后可以在排除过滤器中使用此表。请注意,我们使用 “column_ifexists”——这是为了防止在 watchlist 不存在的情况下规则失败。

如果觉得检测规则中的代码有点繁琐,特别是需要在每个规则中使用时,可以将其存储为函数 “GetWatchlistValue”:

GetWatchlistValue function implementation

然后规则将如下所示,确实简洁多了:

let rule_uuid ="28748697-a290-4367-b67c-57f923e21848";
let AllowedAccounts = GetWachlistValue("windows_builtin", rule_uuid, "AllowedAccounts");
SecurityEvent
|&nbsp;where<DETECTION LOGIC>
|&nbsp;where&nbsp;Account !in&nbsp;(AllowedAccounts)

Global Watchlists

Global watchlists 包含可以被多个 content packs 使用的数据:通常是包含资产的列表,例如公司域、防火墙、扫描器、服务器等。如果知道特定 IP 是扫描器,就可以在防火墙 content pack、IDS content pack、WAF content pack、windows content pack 等中引用它,而无需多次维护。

因为 global watchlist 项被多个 content packs 中的多个规则引用,所以需要正确定义它们并进行必要的模式验证。例如,如果有一个包含服务器的 watchlist,可能想添加服务器角色,如 DC、DNS 或 Fileserver,或任何您喜欢的,但请确保使用一致的类型分类,否则可能会破坏规则中的过滤器。

其次,在规则中,需要知道正在处理什么类型的对象。例如,用户账户可以有多种格式,具体取决于正在处理的日志:有时会看到 sAMAccountName (jdoe),有时是 UPN ([email protected]),有时带域 (CONTOSOjdoe) 或不带域:只是像 john.doe 这样的用户名。在组织 watchlist 和创建查询过滤器时要考虑这一点。

第三,需要知道正在使用什么数据类型。用于过滤服务器角色的 Kusto 操作符将不同于用于网络名称的操作符,因为前者是数组(服务器可以有多个角色),而后者通常是标签。至于命名约定,我们遵循 Sentinel 中使用的实体4。我们本来就在那个生态系统中,已经在规则中使用它们。最后,与 Content Pack watchlists 一样,我们还使用标签,分析师或检测工程师可以在添加 watchlist 项时使用这些标签存储服务工单引用。

一些 watchlists 的示例:

• Servers
• Schema: HostName, IPAddress, Role, Tags
• SearchKey: IPAddress
• Role: DC, DNS, …
• Firewalls
• Schema: HostName, IPAddress, Tags
• SearchKey: IPAddress
• Scanners
• Schema: HostName, IPAddress, Tags
• SearchKey: IPAddress
• Networks
• Schema: Subnet, Name, Tags
• SearchKey: Subnet
• Name: Visitor, Byod, Clients, Servers, Management, …
• DomainsInternal
• Schema: Domain, Tags
• SearchKey: Domain
• DomainsTrusted
• Schema: Domain, Tags
• SearchKey: Domain
• UserAccounts
• Schema: Name, UPNSuffix, Sid, AadUserId, Role, Tags
• SearchKey: Name
• Role: Breakglass, Canary, Administrator, …

敏锐的读者会注意到,如果将这些 watchlists 与资产数据库或用户目录结合使用,这种方法会变得更加强大。我们确实建议构建自动化来填充(部分)watchlists,但这超出了本文范围。

使用 Global watchlists 的一些示例:下面是一个非常简单的横向移动检测。其思想是通常不会有到客户端 SMB 端口的流量。如果有一个包含域控制器、文件服务器以及我们预期看到 SMB 流量的其他服务器的 IP 地址的 watchlist,那么可以像这样使用 watchlist:

let ServersAllowList = _GetWatchlist('Servers')
| project SearchKey;
DeviceNetworkEvents
| where RemotePort ==445
| where RemoteIP !in (ServersAllowList)

如果想将过滤限制为仅特定服务器角色,只需添加一个过滤器:

let ServersAllowList = _GetWatchlist('Servers')
| where column_ifexists("Role","") has_any ("DC", "Fileserver")
| project SearchKey;
DeviceNetworkEvents
| where RemotePort ==445
| where RemoteIP !in (ServersAllowList)

如果想使用范围而不是单个 IP 地址,查询会有所不同。对于此规则,可以利用 Networks watchlist 并过滤 Servers 范围,而不是使用 Servers watchlist:

let ServersAllowList = toscalar(_GetWatchlist('Networks')
| where column_ifexists("Name","") in~ ("Servers")
| project SearchKey
| summarize make_set(SearchKey)
);
DeviceNetworkEvents
| where RemotePort ==445
| where not(ipv4_is_in_any_range(RemoteIP, ServersAllowList))

这里我们使用 ipv4_is_in_any_range 函数,它需要一个动态数组;因此使用 summarize make_set() 和 toscalar()。顺便说一下,也可以将该代码块用于单个 IP 地址,因为不必使用 CIDR 表示法(/24),只需 IP 地址即可。

过滤流程

为了在 DaC 流水线中最好地利用 watchlists,我们将遵循以下工作流程。检测工程师将新值添加到 watchlist 并创建拉取请求,这会触发上下文化流水线。上下文化流水线识别有多少告警和事件与我们添加到 watchlist 的条目相关联,并用这些结果丰富 PR。这使我们能够估计调优对生成的告警的影响。然后,另一位检测工程师审查并批准或拒绝更改。一旦批准,对 watchlist 的更新就会启动一个 CD 流水线,将更改推送到 Sentinel。

Detection-as-Code allowlist workflow diagram

使用构建验证上下文化拉取请求

根据上述工作流程,我们将使用构建验证和 KQL 来增强拉取请求的上下文。这将为检测工程师提供有关所请求白名单的详细信息。通过这样做,工程师可以在决定是否批准或拒绝拉取请求之前估计白名单中新条目产生的噪音。

我们的第一步是识别 PR 引入的更改。由于构建验证将在主分支上进行,我们可以通过发出以下命令来获取更改。此 Git 命令用于识别在 main分支和 PR 分支之间已更改的文件名。通过指定 --name-only,它仅列出文件名而不显示内容更改。--pretty=""选项确保不显示额外的提交日志信息,仅关注文件名。

git diff --name-only --pretty="" origin/main..HEAD

Git diff output showing modified files

然后,对于每个文件,我们将执行以下命令以显示 main分支和 PR 分支之间文件 filters/scanners.csv的差异。通过使用 --no-pager,输出直接显示在终端中而不分页。--unified=0选项指定 diff 输出应该有零行上下文,仅显示已更改的行。

git --no-pager --unified=0 origin/main..HEAD -- filters/scanners.csv

Git diff output for scanners.csv file

然后,对于每个文件的更改,我们使用正则表达式解析输出,并识别添加到仓库中每个 watchlist 的值。

如果在 watchlist 中添加并删除一个条目,如上面的 9.9.9.9,这只是意味着我们正在更改其在列表中的顺序。

下一步是根据上面识别的信息来上下文化拉取请求,即我们在 watchlist 中添加的值。为了计算估计值,我们将对 Sentinel 事件执行搜索,并尝试识别其中有多少在其实体4中包含白名单值。在 Sentinel 中,实体表示与事件或告警相关的元素。实体用于提供有关正在分析的安全事件的额外上下文和详细信息。常见的实体类型包括 IP 地址、用户账户、主机名、URL、文件等。

为了确定添加到 watchlist 的任何白名单属性是否在我们仓库中管理的规则生成的事件中显示为实体,我们将使用以下 KQL 查询。该查询检查 30 天期间的 Azure Sentinel 事件,将它们与其关联的告警 ID 相关联,因为每个事件可能包含多个告警。然后,我们过滤包含 property_values 变量中定义的任何值的实体的告警,并通过计算每个实体值在事件和告警中的出现次数来汇总结果。

let property_values = dynamic([]);
let lookback_time = 30d;
SecurityIncident
| where TimeGenerated > ago(lookback_time)
| where ProviderName =="Azure Sentinel"
| project IncidentNumber, IncidentName, Title, RelatedAnalyticRuleIds, AlertIds
| mv-expand AlertId = AlertIds
| extend AlertId = tostring(AlertId)
| join kind = inner (
SecurityAlert
| where TimeGenerated > ago(lookback_time)
| where ProductName =="Azure Sentinel"and ProductComponentName =="Scheduled Alerts"
| project SystemAlertId, AlertName, Entities, AlertType
| extend AnalyticRuleId = split(AlertType, "_")[-1]
| mv-expand Entities = todynamic(Entities)
| mv-expand Entities = todynamic(Entities)
| mv-expand kind=array key = bag_keys(Entities)
| extend
    PropertyName = tostring(key),
    PropertyValue = tostring(Entities[tostring(key)])
| where not(PropertyName startswith "$") and PropertyName !=""
| where PropertyName !="Type"
| where PropertyValue in (property_values)
) on $left.AlertId== $right.SystemAlertId
| summarize
    IncidentCount = count_distinct(IncidentNumber),
    AlertCount = count_distinct(SystemAlertId),
    AnalyticRuleIds = make_set(AnalyticRuleId),
    IncidentNumbers=make_set(IncidentNumber),
    SystemAlertIds=make_set(SystemAlertId),
    AlertNames = make_set(AlertName)
    by PropertyName, PropertyValue, Title
| project-reorder PropertyName, PropertyValue, IncidentCount, Title, AlertCount, AlertNames, AnalyticRuleIds, IncidentNumbers, SystemAlertIds

假设我们想要从扫描器的 watchlist 中将 IP 10.16.5.90 加入白名单。在这种情况下,我们会将 IP 添加到 property values 动态列表变量中并执行查询。

let property_values = dynamic(['10.16.5.90']);

下面显示了查询的示例输出。我们在 3 个规则的 Address 实体字段中识别了白名单值 “10.16.5.90”。该实体值存在于 21 个事件和 309 个告警中。

KQL query output showing entity matches in incidents and alerts

我们将所有内容整合到这个脚本中,以便识别 PR 在 watchlist 中引入的更改,然后使用 Jinja6模板制作 KQL 查询。KQL 查询保存在流水线变量中,供流水线中的后续步骤使用。

import re
import subprocess
from jinja2 import Environment, FileSystemLoader

defrun_command(command: list) -> str:
"""Executes a shell command and returns the output."""
try:
# print(f"[R] Running command: {' '.join(command)}")
        output = subprocess.check_output(command, text=True, encoding="utf-8", errors="replace").strip()
# print(f"[O] Command output:\n{'\n'.join(['\t'+line for line in output.splitlines()])}")
return output
except subprocess.CalledProcessErroras e:
print(f"##vso[task.logissue type=error]Error executing command: {' '.join(command)}")
print(f"##vso[task.logissue type=error]Error message: {str(e)}")
return""
exceptUnicodeDecodeErroras e:
print(f"##vso[task.logissue type=error]Unicode decode error: {e}")
return""

defget_pr_modified_files() -> list:
"""Get the pr modified files"""
return run_command(["git", "diff", "--name-only", "--pretty=""", "origin/main..HEAD"]).splitlines()

defget_pr_modified_file_diff_lines(file: str) -> str:
"""Get the pr modified file diff"""
return run_command(["git", "--no-pager", "diff", "--unified=0", "origin/main..HEAD", "--", file]).splitlines()

defget_watchlist_added_values(watchlist_diff: str) -> list:
    added_regex   =r"^\+(?!\+\+)\s*(.*)$"
    removed_regex =r"^\-(?!\-\-)\s*(.*)$"
    added_values = []
    removed_values = []
for line in watchlist_diff:
        match_added = re.search(added_regex, line)
        values = match_added.group(1).split(",") if match_added else []
        values = [v.strip() for v in values]
        added_values += values

        match_removed = re.search(removed_regex, line)
        values = match_removed.group(1).split(",") if match_removed else []
        values = [v.strip() for v in values]
        removed_values += values

# Adding and removing an entry in the same file means just moving it around in the watchlist
    added_values_final = []
    removed_values_final = []
    added_copy = added_values[:]
    removed_copy = removed_values[:]

for val in added_values:
if val in removed_copy:
# cancel out one occurrence from removed
            removed_copy.remove(val)
else:
            added_values_final.append(val)

for val in removed_values:
if val in added_copy:
# cancel out one occurrence from added
            added_copy.remove(val)
else:
            removed_values_final.append(val)

print (f"Added values: {','.join(added_values_final)}")
print (f"Removed values: {','.join(removed_values_final)}")

return added_values_final

defidentify_watchlist_changes():
    all_values = []
    pr_modified_files = get_pr_modified_files()
print(f"Modified Files:\n{', '.join(pr_modified_files)}")

for pr_modified_file in pr_modified_files:
if pr_modified_file.startswith("filters") and pr_modified_file.endswith(".csv"):
print(f"Checking file: {pr_modified_file}")
            pr_modified_file_diff = get_pr_modified_file_diff_lines(pr_modified_file)
            all_values += get_watchlist_added_values(pr_modified_file_diff)

    env = Environment(loader=FileSystemLoader("pipelines/scripts/templates"))
    template = env.get_template("contextualization_query.jinja")

    kql_query = template.render(entity_values=all_values)
print(f"KQL Query: \n{kql_query}")
print(f"##vso[task.setvariable variable=kql_query]{kql_query.replace("\n", " ")}")

defmain():
    identify_watchlist_changes()

if__name__=="__main__":
    main()

jinja 模板如下:

let entity_values = dynamic({{ entity_values }});
let lookback_time = 30d;
SecurityIncident
| where TimeGenerated > ago(lookback_time)
| where ProviderName =="Azure Sentinel"
| project IncidentNumber, IncidentName, Title, RelatedAnalyticRuleIds, AlertIds
| mv-expand AlertId = AlertIds
| extend AlertId = tostring(AlertId)
| join kind = inner (
    SecurityAlert
    | where TimeGenerated > ago(lookback_time)
    | where ProductName =="Azure Sentinel"and ProductComponentName =="Scheduled Alerts"
    | project SystemAlertId, AlertName, Entities, AlertType
    | extend AnalyticRuleId = split(AlertType, "_")[-1]
    | mv-expand Entities = todynamic(Entities)
    | mv-expand Entities = todynamic(Entities)
    | mv-expand kind=array key = bag_keys(Entities)
    | extend
        EntityName = tostring(key),
        EntityValue = tostring(Entities[tostring(key)])
    | where not(EntityName startswith "$") and EntityName !=""
    | where EntityName !="Type"
    | where EntityValue in (entity_values)
    )
on $left.AlertId== $right.SystemAlertId
| summarize
    IncidentCount = count_distinct(IncidentNumber),
    AlertCount = count_distinct(SystemAlertId),
    AnalyticRuleIds = make_set(AnalyticRuleId),
    IncidentNumbers=make_set(IncidentNumber),
    SystemAlertIds=make_set(SystemAlertId),
    AlertNames = make_set(AlertName)
    by EntityName, EntityValue, Title
| project-reorder
    EntityName,
    EntityValue,
    IncidentCount,
    Title,
    AlertCount,
    AlertNames,
    AnalyticRuleIds,
    IncidentNumbers,
    SystemAlertIds

在将 KQL 查询保存为流水线变量后,我们可以使用我们在第 7 部分中创建的检测监控脚本来查询环境,并识别在其实体中包含我们添加到 watchlist 的值的事件。这将让我们了解此过滤将对环境产生的影响。

name: Contextualize Watchlist Change

trigger: none

jobs:
- job: ContextualizeWatchlistChange
displayName: "Contextualize Watchlist Change"
steps:
    - checkout: self
fetchDepth: 0
path: 's/$(Build.Repository.Name)'
    - script: |
python $(Pipeline.Workspace)/s/$(Build.Repository.Name)/pipelines/scripts/identify_watchlist_changes.py
displayName: 'Run Identify Watchlist Changes'
    - script: |
pip install -r $(Pipeline.Workspace)/s/$(Build.Repository.Name)/pipelines/scripts/requirements.txt
displayName: 'Python Dependencies Installation'
    - bash: |
python $(Pipeline.Workspace)/s/$(Build.Repository.Name)/pipelines/scripts/detection_monitoring.py --tenant 'QA' --platform 'sentinel' --detection-compare-field 'AnalyticRuleIds'
env:
QUERY: $(kql_query)
displayName: "Run Detection Monitoring Script"

我们将流水线添加到主分支的构建验证中。但这次,我们将策略要求设置为可选,因为不希望运行的潜在失败阻止我们合并拉取请求的能力。

Build validation policy configuration

为了测试实现,我们将把 IP 10.16.5.90 添加到扫描器 watchlist。SOC 分析师将创建一个拉取请求,将 IP 添加到 filters/scanners.csv 文件。

Pull request adding IP to scanners watchlist

构建验证将运行流水线,识别拉取请求引入的 watchlist 更改,并从 jinja 模板创建 KQL 查询。

Pipeline identifying watchlist changes and generating KQL query

然后查询将在目标平台上运行,并识别添加的值在实体中的事件和告警。

Query results showing incidents and alerts with whitelisted entity

使用 Watchlists 自动调优

在创建拉取请求并获得批准后,成功合并到主分支后将触发另一个流水线,该流水线将识别哪些 watchlists 已更新并将它们同步到目标平台。但首先,我们将介绍实现所述逻辑所需的一些 git 命令。我们将使用类似于第 6 部分中由仓库更新触发的自动部署的方法。

我们将使用的第一个命令将获取 main分支上最近提交的哈希。-1选项将输出限制为仅最新提交,确保仅显示一个提交。--pretty=format:%H部分自定义输出以仅显示完整的提交哈希。

git log main -1 --pretty=format:%H

Git log showing latest commit hash

然后我们使用以下命令显示提交消息,使用从前一个命令返回的提交哈希。

git show --pretty=format:%s f88594f5acbb0e0d9bc5652249dd44897ed23b40

Git show displaying commit message

当我们合并拉取请求时,Azure DevOps Repos 自动创建了此提交(“Merged PR “)。然后我们执行以下 Git 命令来获取提交父级。

git show --pretty=format:%P f88594f5acbb0e0d9bc5652249dd44897ed23b40

Git show displaying commit parents

然后,我们通过执行下面的命令来获取父提交之间的提交列表。范围 <3b9a0c…>.. 定义了起始和结束提交,使我们能够看到在 main分支上这两个哈希之间的所有提交。通过指定 --pretty=format:%H选项,输出被自定义为仅显示提交哈希。

git log main --pretty=format:%H 3b9a0c0d926bae8fa6295186e06d7e17033c0ebe..f767e8edadd6783a542de5ed3f53c1c79ed2575b

Git log showing commit list between parent commits

对于每个提交,我们执行 git diff-tree,它检查提交引入的差异。--no-commit-id标志从输出中省略提交 ID,仅关注文件更改。--name-status选项提供更改的摘要,显示每个文件的状态(例如,添加、修改、删除)以及文件名。

git diff-tree --no-commit-id --name-status -r f767e8edadd6783a542de5ed3f53c1c79ed2575b

Git diff-tree output showing file changes

根据您使用的 Git 客户端版本,可能需要验证输出是否如屏幕截图中显示的那样。

对于上述命令中显示的输出,我们使用无快进合并。如果使用另一种合并类型,可能需要修改 git 命令。

下一步是将所有内容整合到一个脚本中。该脚本检索最后一个提交哈希,检查提交消息是否指示合并的拉取请求,并识别合并的开始和结束提交以列出所有涉及的提交。然后,它检查每个提交在 filters 目录中的修改。该目录中修改或添加的文件名存储在 watchlist_names变量中,供流水线在稍后步骤中使用。

import subprocess
import os

base_paths = ["filters/*.csv", "filters/*.json"]

defrun_command(command: list) -> str:
"""Executes a shell command and returns the output."""
try:
#print(f"[R] Running command: {' '.join(command)}")
        output = subprocess.check_output(command, text=True, encoding="utf-8", errors="replace").strip()
#print(f"[O] Command output:\n{'\n'.join(['\t'+line for line in output.splitlines()])}")
return output
except subprocess.CalledProcessErroras e:
print(f"##vso[task.logissue type=error] Error executing command: {' '.join(command)}")
print(f"##vso[task.logissue type=error] Error message: {str(e)}")
return""
exceptUnicodeDecodeErroras e:
print(f"##vso[task.logissue type=error] Unicode decode error: {e}")
return""

defget_last_commit() -> str:
"""Retrieve the most recent commit hash on the main branch."""
return run_command(["git", "log", "main", "-1", "--pretty=format:%H"])

defget_commit_message(commit_hash: str) -> str:
"""Retrieve a commit message"""
return run_command(["git", "show", "--pretty=format:%s", commit_hash])

defget_commit_parents(commit_hash: str) -> list:
"""Retrieve the commit parents"""
return run_command(["git", "show", "--pretty=format:%P", commit_hash]).split(" ")

defget_commit_list(start_commit:str, end_commit:str) -> list:
"""Retrieve a commit list"""
return run_command(["git", "log", "main", "--pretty=format:%H", f"{start_commit}..{end_commit}"]).splitlines()

defget_commit_modified_files(commit_hash: str) -> list:
"""Get a list of modified files in the commit, along with their status"""
return run_command(["git", "diff-tree", "--no-commit-id", "--name-status", "-r", commit_hash, "--"]+ base_paths).splitlines()

defidentify_filters():
    filters = []

    input_commit_hash = get_last_commit()
print(f"Last commit ID: {input_commit_hash}")

    commit_message = get_commit_message(input_commit_hash)
print(f"Commit message: {commit_message}")

if commit_message.startswith("Merged PR"):
print("PR merge commit identified. Identifying changes...")
        commit_parents = get_commit_parents(input_commit_hash)

iflen(commit_parents) ==2:
            start_commit = commit_parents[0]
            end_commit = commit_parents[1]
print(f"Start commit:{start_commit}..End commit:{end_commit}")

            commit_list = get_commit_list(start_commit, end_commit)
print(f"Commit list:\n {', '.join(commit_list)}")
for commit_hash in commit_list:
print(f"Processing commit:{commit_hash}")
                commit_modified_files = get_commit_modified_files(commit_hash)

for commit_modified_file in commit_modified_files:
                    status, filepath = commit_modified_file.split("\t")

if filepath.startswith("filters/") and (status in ["A", "M"]):
print(f"Filter watchlist {filepath} {"created" if status=="A" else "modified"}.")
                        filters.append(os.path.basename(filepath.removesuffix(".json").removesuffix(".csv")))
else:
print(f"##vso[task.logissue type=error]Could not identify parents of {input_commit_hash}")

    filters =list(set(filters))
print(f"Filter watchlists identified for deployment: {', '.join(filters)}")
print(f"##vso[task.setvariable variable=watchlist_names]{', '.join(filters)}")
return

defmain():
    identify_filters()

if__name__=="__main__":
    main()

流水线由主分支中 filters/*目录内的更改触发。流水线检出主分支的最新代码,安装必要的 Python 依赖项,并运行上述脚本以更新 watchlists 中的更改。然后,它根据仓库 watchlists 中识别的更改同步 watchlist 项。

name: Automatic Watchlist Deployment Triggered By Repo Changes

trigger:
branches:
include:
      - main
paths:
include:
      - "filters/*"

jobs:
- job: IdentifyFilterWatchlistChanges
displayName: "Identify Filter Watchlist Changes"
condition: eq(variables['Build.SourceBranchName'], 'main')
steps:
    - checkout: self
fetchDepth: 0
    - script: |
git fetch origin main
git checkout -b main origin/main
displayName: "Fetch Branches Locally"
    - script: |
pip install -r pipelines/scripts/requirements.txt
displayName: 'Python Dependencies Installation'
    - script: |
python pipelines/scripts/identify_filter_changes.py
displayName: 'Run Identify Filter Changes Script'
    - script: |
python pipelines/scripts/watchlist_mgmt.py --tenant '<Tenant Name>' --action 'SyncWatchlistItems' --names '$(watchlist_names)'
displayName:&nbsp;"Watchlist Management Script Run"

作为示例,我们将在仓库中的扫描器过滤文件中进行以下两项更改:添加一个条目(10.16.5.90)并删除另一个(1.1.1.1)。

Scanners filter file changes in repository

在合并上述更改后,identify_filter_changes.py 脚本运行并识别修改的 watchlist。

Script identifying modified watchlists

然后,watchlist_mgmt.py 将 watchlist 同步到 Sentinel。

Watchlist synchronization to Sentinel

总结

总结而言,我们探讨了持续部署流水线如何通过使用 watchlists 来简化检测的调优。这个过程帮助我们减少手动工作量,并更有效地扩展检测库。

参考文献

1. https://learn.microsoft.com/en-us/azure/sentinel/watchlists ↩︎
2. https://learn.microsoft.com/en-us/rest/api/securityinsights/watchlists?view=rest-securityinsights-2025-09-01 ↩︎
3. https://learn.microsoft.com/en-us/rest/api/securityinsights/watchlist-items?view=rest-securityinsights-2025-09-01 ↩︎
4. https://learn.microsoft.com/en-us/azure/sentinel/entities ↩︎
5. https://learn.microsoft.com/en-us/azure/sentinel/entities ↩︎
6. https://jinja.palletsprojects.com/ ↩︎

---

Detection Engineering: Practicing Detection-as-Code – Tuning – Part 8

免责声明：本博客文章仅用于教育和研究目的。提供的所有技术和代码示例旨在帮助防御者理解攻击手法并提高安全态势。请勿使用此信息访问或干扰您不拥有或没有明确测试权限的系统。未经授权的使用可能违反法律和道德准则。作者对因应用所讨论概念而导致的任何误用或损害不承担任何责任。

---

免责声明：

本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。

任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。

本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。

本文转载自：securitainment Kristof Baute《检测工程: 实践检测即代码 – 规则调优自动化 – 第8部分》