这是 pySigma 的 OpenSearch 后端。它提供 sigma.backends.opensearch 包，包含两个后端类：
OpensearchLuceneBackend - 将 Sigma 规则转换为 Lucene 查询语法OpenSearchPPLBackend - 将 Sigma 规则转换为 PPL（管道处理语言）查询Lucene 后端支持以下输出格式：
此后端目前由以下人员维护：
由于基于 Lucene 的查询与 Elasticsearch Lucene 查询非常相似，因此该后端的大部分代码来自 pySigma-backend-elasticsearch。
OpenSearch 特定的更改和输出格式在此后端中完成（例如监控规则）。
PPL（管道处理语言）后端是从零开始实现的，以支持 OpenSearch 的原生查询语言。PPL 提供：
PPL 后端完全支持 Sigma 关联规则，能够检测复杂的多事件场景：
sigma convert \
-t opensearch \
-p ecs_windows \
-f monitor_rule \
/data/sigma/rules/windows/process_creation/proc_creation_win_whoami_priv.yml
sigma convert \
-t opensearch-ppl \
-p ecs_windows \
/data/sigma/rules/windows/process_creation/proc_creation_win_whoami_priv.yml
from sigma.backends.opensearch import OpensearchLuceneBackend
from sigma.pipelines.sysmon import sysmon_pipeline
from sigma.pipelines.elasticsearch.windows import ecs_windows
from sigma.collection import SigmaCollection
from sigma.processing.resolver import ProcessingPipelineResolver
# Create our pipeline resolver
piperesolver = ProcessingPipelineResolver()
# Add wanted pipelines
piperesolver.add_pipeline_class(ecs_windows())
piperesolver.add_pipeline_class(sysmon_pipeline())
# Create a single sorted and prioritzed pipeline
resolved_pipeline = piperesolver.resolve(piperesolver.pipelines)
# Instantiate backend, using our resolved pipeline
# and some backend parameter
backend = OpensearchLuceneBackend(resolved_pipeline, index_names=['logs-*-*', 'beats-*'], monitor_interval=10, monitor_interval_unit="MINUTES")
rules = SigmaCollection.from_yaml("""
title: Run Whoami Showing Privileges
id: 97a80ec7-0e2f-4d05-9ef4-65760e634f6b
status: experimental
description: Detects a whoami.exe executed with the /priv command line flag instructing the tool to show all current user privieleges. This is often used after a privilege escalation attempt.
references:
- https://docs.microsoft.com/en-us/windows-server/administration/windows-commands/whoami
author: Florian Roth
date: 2021/05/05
modified: 2022/05/13
tags:
- attack.privilege_escalation
- attack.discovery
- attack.t1033
logsource:
category: process_creation
product: windows
detection:
selection_img:
- Image|endswith: '\whoami.exe'
- OriginalFileName: 'whoami.exe'
selection_cli:
CommandLine|contains: '/priv'
condition: all of selection*
falsepositives:
- Administrative activity (rare lookups on current privileges)
level: high
""")
# Print converted rule in Lucene syntax
print("Lucene Result: \n" + "\n".join(backend.convert(rules)))
# Print converted rule ready for dsl syntax
print("DSL Result: \n" + json.dumps(backend.convert(rules, output_format="dsl_lucene")[0], indent=2))
# Generate a JSON structure to be imported as monitor rule
print("Monitor Rule Result: \n" + backend.convert(rules, output_format="monitor_rule"))
Lucene 结果：
winlog.channel:Microsoft\-Windows\-Sysmon\/Operational AND (event.code:1 AND ((process.executable:*\\whoami.exe OR process.pe.original_file_name:whoami.exe) AND process.command_line:*\/priv*))
DSL 结果：
{
"query": {
"bool": {
"must": [
{
"query_string": {
"query": "winlog.channel:Microsoft\\-Windows\\-Sysmon\\/Operational AND (event.code:1 AND (winlog.channel:Microsoft\\-Windows\\-Sysmon\\/Operational AND (event.code:1 AND ((process.executable:*\\\\whoami.exe OR process.pe.original_file_name:whoami.exe) AND process.command_line:*\\/priv*))))",
"analyze_wildcard": true
}
}
]
}
}
}
监控规则结果：
{
"type": "monitor",
"name": "SIGMA - Run Whoami Showing Privileges",
"description": "Detects a whoami.exe executed with the /priv command line flag instructing the tool to show all current user privieleges. This is often used after a privilege escalation attempt.",
"enabled": true,
"schedule": {
"period": {
"interval": 10,
"unit": "MINUTES"
}
},
"inputs": [
{
"search": {
"indices": [
"logs-*-*",
"beats-*"
],
"query": {
"size": 1,
"query": {
"bool": {
"must": [
{
"query_string": {
"query": "winlog.channel:Microsoft\\-Windows\\-Sysmon\\/Operational AND (event.code:1 AND (winlog.channel:Microsoft\\-Windows\\-Sysmon\\/Operational AND (event.code:1 AND (winlog.channel:Microsoft\\-Windows\\-Sysmon\\/Operational AND (event.code:1 AND ((process.executable:*\\\\whoami.exe OR process.pe.original_file_name:whoami.exe) AND process.command_line:*\\/priv*))))))",
"analyze_wildcard": true
}
}
]
}
}
}
}
}
],
"tags": [
"attack-privilege_escalation",
"attack-discovery",
"attack-t1033"
],
"triggers": [
{
"name": "generated-trigger",
"severity": 2,
"condition": {
"script": {
"source": "ctx.results[0].hits.total.value > 0",
"lang": "painless"
}
},
"actions": []
}
],
"sigma_meta_data": {
"rule_id": "97a80ec7-0e2f-4d05-9ef4-65760e634f6b",
"threat": []
},
"references": [
"https://docs.microsoft.com/en-us/windows-server/administration/windows-commands/whoami"
]
}
from sigma.backends.opensearch.opensearch_ppl import OpenSearchPPLBackend
from sigma.collection import SigmaCollection
# Instantiate PPL backend
backend = OpenSearchPPLBackend()
# Use the same rule as above
rules = SigmaCollection.from_yaml("""
title: Run Whoami Showing Privileges
id: 97a80ec7-0e2f-4d05-9ef4-65760e634f6b
status: experimental
description: Detects a whoami.exe executed with the /priv command line flag instructing the tool to show all current user privieleges. This is often used after a privilege escalation attempt.
references:
- https://docs.microsoft.com/en-us/windows-server/administration/windows-commands/whoami
author: Florian Roth
date: 2021/05/05
modified: 2022/05/13
tags:
- attack.privilege_escalation
- attack.discovery
- attack.t1033
logsource:
category: process_creation
product: windows
detection:
selection_img:
- Image|endswith: '\whoami.exe'
- OriginalFileName: 'whoami.exe'
selection_cli:
CommandLine|contains: '/priv'
condition: all of selection*
falsepositives:
- Administrative activity (rare lookups on current privileges)
level: high
""")
# Print converted rule in PPL syntax
print("PPL Result: \n" + "\n".join(backend.convert(rules)))
PPL 结果：
source=windows-process_creation-* | where (LIKE(Image, "%\whoami.exe") OR OriginalFileName="whoami.exe") AND LIKE(CommandLine, "%/priv%")
from sigma.backends.opensearch.opensearch_ppl import OpenSearchPPLBackend
from sigma.collection import SigmaCollection
backend = OpenSearchPPLBackend()
# Brute force detection using event_count correlation
rules = SigmaCollection.from_yaml("""
title: Windows Failed Logon Event
name: failed_logon
logsource:
product: windows
service: security
detection:
selection:
EventID: 4625
filter:
SubjectUserName|endswith: $
condition: selection and not filter
---
title: Brute Force Attack Detection
correlation:
type: event_count
rules:
- failed_logon
group-by:
- TargetUserName
- TargetDomainName
timespan: 5m
condition:
gte: 10
""")
print("Correlation PPL Result: \n" + "\n".join(backend.convert(rules)))
关联 PPL 结果：
| search source=windows-security-* | where EventID=4625 AND NOT LIKE(SubjectUserName, "%$") | stats count() as event_count by TargetUserName, TargetDomainName | where event_count >= 10
您可以通过 Sigma 规则中的自定义属性或后端初始化选项来配置 PPL 后端的行为。
PPL 后端支持以下自定义属性，可在 Sigma 规则的 custom 部分中指定：
custom:
opensearch_ppl_index: "custom-logs-*" # Override default index pattern
opensearch_ppl_min_time: "-30d" # Set query time window start
opensearch_ppl_max_time: "now" # Set query time window end
此示例展示了自定义属性如何与关联规则配合使用，其中单个检测规则可以拥有自己的时间窗口，或从关联规则继承：
title: Detection Rule 1 - With Own Time Filter
id: 10000400-0000-0000-0000-000000000004
logsource:
product: windows
category: process_creation
detection:
selection:
CommandLine|contains: 'malware'
condition: selection
custom:
opensearch_ppl_min_time: "-7d" # This rule uses 7 days
opensearch_ppl_max_time: "now"
---
title: Detection Rule 2 - No Time Filter
id: 10000401-0000-0000-0000-000000000004
logsource:
product: windows
category: network_connection
detection:
selection:
DestinationPort: 443
condition: selection
# No custom attributes - will inherit from correlation
---
title: Correlation - Mixed Time Filters
id: 10000402-0000-0000-0000-000000000004
correlation:
type: temporal
rules:
- 10000400-0000-0000-0000-000000000004
- 10000401-0000-0000-0000-000000000004
group-by:
- Computer
timespan: 5m
custom:
opensearch_ppl_min_time: "-30d" # Rule 2 inherits this (30 days)
opensearch_ppl_max_time: "now"
结果：
您还可以在实例化后端时设置默认值：
backend = OpenSearchPPLBackend(
custom_logsource="default-logs-*", # Default index pattern for all rules
min_time="-24h", # Default time window start
max_time="now" # Default time window end
)
各个规则中的自定义属性将覆盖这些后端级别的默认值。