这是 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()