building-detection-rule-with-splunk-spl
Compare original and translation side by side
🇺🇸
Original
English🇨🇳
Translation
ChineseBuilding Detection Rules with Splunk SPL
使用Splunk SPL构建检测规则
Overview
概述
Splunk Search Processing Language (SPL) is the primary query language used in Splunk Enterprise Security for building correlation searches that detect suspicious events and patterns. A well-crafted detection rule aggregates, correlates, and enriches security events to generate actionable notable events for SOC analysts. Enterprise SIEMs on average cover only 21% of MITRE ATT&CK techniques, making skilled SPL rule writing essential for closing detection gaps.
Splunk搜索处理语言(SPL)是Splunk Enterprise Security中用于构建关联搜索的主要查询语言,可检测可疑事件和模式。精心编写的检测规则会聚合、关联并丰富安全事件,为SOC分析师生成可执行的重要事件。企业级SIEM平均仅覆盖21%的MITRE ATT&CK技术,因此熟练编写SPL规则对于填补检测缺口至关重要。
When to Use
适用场景
- When deploying or configuring building detection rule with splunk spl capabilities in your environment
- When establishing security controls aligned to compliance requirements
- When building or improving security architecture for this domain
- When conducting security assessments that require this implementation
- 在环境中部署或配置基于Splunk SPL的检测规则时
- 建立符合合规要求的安全控制措施时
- 构建或改进该领域的安全架构时
- 进行需要此实现的安全评估时
Prerequisites
前提条件
- Splunk Enterprise Security (ES) deployed and configured
- Access to Splunk Search & Reporting app with appropriate roles
- Understanding of Common Information Model (CIM) data models
- Familiarity with MITRE ATT&CK framework techniques
- Knowledge of the organization's log sources and data flows
- 已部署并配置Splunk Enterprise Security(ES)
- 拥有Splunk搜索与报告应用的适当角色访问权限
- 了解通用信息模型(CIM)数据模型
- 熟悉MITRE ATT&CK框架技术
- 了解组织的日志源和数据流
Core SPL Detection Rule Patterns
核心SPL检测规则模式
1. Threshold-Based Detection
1. 基于阈值的检测
Detects events exceeding a defined count within a time window.
spl
index=wineventlog sourcetype=WinEventLog:Security EventCode=4625
| stats count as failed_logins dc(TargetUserName) as unique_users by src_ip
| where failed_logins > 10 AND unique_users > 3
| eval severity="high"
| eval description="Brute force attack detected from ".src_ip." with ".failed_logins." failed logins across ".unique_users." accounts"检测在时间窗口内超过定义数量的事件。
spl
index=wineventlog sourcetype=WinEventLog:Security EventCode=4625
| stats count as failed_logins dc(TargetUserName) as unique_users by src_ip
| where failed_logins > 10 AND unique_users > 3
| eval severity="high"
| eval description="Brute force attack detected from ".src_ip." with ".failed_logins." failed logins across ".unique_users." accounts"2. Sequence-Based Detection (Failed Login Followed by Success)
2. 基于序列的检测(失败登录后成功登录)
Correlates a sequence of events indicating a successful brute force attack.
spl
index=wineventlog sourcetype=WinEventLog:Security (EventCode=4625 OR EventCode=4624)
| eval login_status=case(EventCode=4625, "failure", EventCode=4624, "success")
| stats count(eval(login_status="failure")) as failures count(eval(login_status="success")) as successes latest(_time) as last_event by src_ip, TargetUserName
| where failures > 5 AND successes > 0
| eval description="Account ".TargetUserName." compromised via brute force from ".src_ip
| eval urgency="critical"关联表明暴力攻击成功的事件序列。
spl
index=wineventlog sourcetype=WinEventLog:Security (EventCode=4625 OR EventCode=4624)
| eval login_status=case(EventCode=4625, "failure", EventCode=4624, "success")
| stats count(eval(login_status="failure")) as failures count(eval(login_status="success")) as successes latest(_time) as last_event by src_ip, TargetUserName
| where failures > 5 AND successes > 0
| eval description="Account ".TargetUserName." compromised via brute force from ".src_ip
| eval urgency="critical"3. Anomaly Detection with Baseline Comparison
3. 基于基线对比的异常检测
Compares current activity against a baseline period to detect spikes.
spl
index=proxy sourcetype=squid
| bin _time span=1h
| stats count as current_count by src_ip, _time
| join src_ip type=left [
search index=proxy sourcetype=squid earliest=-7d@d latest=-1d@d
| stats avg(count) as avg_count stdev(count) as stdev_count by src_ip
]
| eval threshold=avg_count + (3 * stdev_count)
| where current_count > threshold
| eval deviation=round((current_count - avg_count) / stdev_count, 2)
| eval description="Anomalous web traffic from ".src_ip." - ".deviation." standard deviations above baseline"将当前活动与基线时段进行对比以检测峰值。
spl
index=proxy sourcetype=squid
| bin _time span=1h
| stats count as current_count by src_ip, _time
| join src_ip type=left [
search index=proxy sourcetype=squid earliest=-7d@d latest=-1d@d
| stats avg(count) as avg_count stdev(count) as stdev_count by src_ip
]
| eval threshold=avg_count + (3 * stdev_count)
| where current_count > threshold
| eval deviation=round((current_count - avg_count) / stdev_count, 2)
| eval description="Anomalous web traffic from ".src_ip." - ".deviation." standard deviations above baseline"4. Lateral Movement Detection
4. 横向移动检测
Identifies potential lateral movement using Windows logon events.
spl
index=wineventlog sourcetype=WinEventLog:Security EventCode=4624 Logon_Type=3
| where NOT match(TargetUserName, ".*\$$")
| stats dc(dest) as unique_hosts values(dest) as hosts by src_ip, TargetUserName
| where unique_hosts > 5
| eval severity=case(unique_hosts > 20, "critical", unique_hosts > 10, "high", true(), "medium")
| eval description=TargetUserName." accessed ".unique_hosts." unique hosts from ".src_ip." via network logon"使用Windows登录事件识别潜在的横向移动。
spl
index=wineventlog sourcetype=WinEventLog:Security EventCode=4624 Logon_Type=3
| where NOT match(TargetUserName, ".*\$$")
| stats dc(dest) as unique_hosts values(dest) as hosts by src_ip, TargetUserName
| where unique_hosts > 5
| eval severity=case(unique_hosts > 20, "critical", unique_hosts > 10, "high", true(), "medium")
| eval description=TargetUserName." accessed ".unique_hosts." unique hosts from ".src_ip." via network logon"5. Data Exfiltration Detection
5. 数据泄露检测
Monitors for large outbound data transfers.
spl
index=firewall sourcetype=pan:traffic action=allowed direction=outbound
| stats sum(bytes_out) as total_bytes_out dc(dest_ip) as unique_destinations by src_ip, user
| eval total_mb=round(total_bytes_out/1048576, 2)
| where total_mb > 500 OR unique_destinations > 50
| lookup asset_lookup ip as src_ip OUTPUT asset_category, asset_owner
| eval severity=case(total_mb > 2000, "critical", total_mb > 1000, "high", true(), "medium")
| eval description=user." transferred ".total_mb."MB to ".unique_destinations." unique destinations"监控大规模 outbound 数据传输。
spl
index=firewall sourcetype=pan:traffic action=allowed direction=outbound
| stats sum(bytes_out) as total_bytes_out dc(dest_ip) as unique_destinations by src_ip, user
| eval total_mb=round(total_bytes_out/1048576, 2)
| where total_mb > 500 OR unique_destinations > 50
| lookup asset_lookup ip as src_ip OUTPUT asset_category, asset_owner
| eval severity=case(total_mb > 2000, "critical", total_mb > 1000, "high", true(), "medium")
| eval description=user." transferred ".total_mb."MB to ".unique_destinations." unique destinations"6. PowerShell Suspicious Execution Detection
6. PowerShell可疑执行检测
Detects encoded or obfuscated PowerShell commands.
spl
index=wineventlog sourcetype=WinEventLog:Security EventCode=4104
| where match(ScriptBlockText, "(?i)(encodedcommand|invoke-expression|iex|downloadstring|frombase64string|net\.webclient|invoke-webrequest|bitstransfer|invoke-mimikatz|invoke-shellcode)")
| eval decoded_length=len(ScriptBlockText)
| stats count values(ScriptBlockText) as commands by Computer, UserName
| where count > 0
| eval severity="high"
| eval mitre_technique="T1059.001"
| eval description="Suspicious PowerShell execution on ".Computer." by ".UserName检测编码或混淆的PowerShell命令。
spl
index=wineventlog sourcetype=WinEventLog:Security EventCode=4104
| where match(ScriptBlockText, "(?i)(encodedcommand|invoke-expression|iex|downloadstring|frombase64string|net\.webclient|invoke-webrequest|bitstransfer|invoke-mimikatz|invoke-shellcode)")
| eval decoded_length=len(ScriptBlockText)
| stats count values(ScriptBlockText) as commands by Computer, UserName
| where count > 0
| eval severity="high"
| eval mitre_technique="T1059.001"
| eval description="Suspicious PowerShell execution on ".Computer." by ".UserNameBuilding Correlation Searches in Splunk ES
在Splunk ES中构建关联搜索
Step-by-Step Process
分步流程
- Define the Use Case: Map to MITRE ATT&CK technique and define what behavior to detect
- Identify Data Sources: Determine which indexes and sourcetypes contain relevant events
- Write the Base Search: Build SPL that extracts relevant events
- Add Aggregation: Use ,
stats, oreventstatsto summarizestreamstats - Apply Thresholds: Set conditions with clause that distinguish normal from anomalous
where - Enrich Context: Add lookups for asset information, identity data, and threat intelligence
- Configure Notable Event: Set severity, urgency, and description fields
- Schedule and Test: Run against historical data and validate detection accuracy
- 定义用例:映射到MITRE ATT&CK技术,明确要检测的行为
- 识别数据源:确定哪些索引和源类型包含相关事件
- 编写基础搜索:构建提取相关事件的SPL
- 添加聚合:使用、
stats或eventstats进行汇总streamstats - 应用阈值:通过子句设置区分正常与异常的条件
where - 丰富上下文:添加资产信息、身份数据和威胁情报的查找
- 配置重要事件:设置严重性、紧急性和描述字段
- 调度与测试:针对历史数据运行并验证检测准确性
Correlation Search Configuration Template
关联搜索配置模板
spl
| tstats summariesonly=true count from datamodel=Authentication
where Authentication.action=failure
by Authentication.src, Authentication.user, _time span=5m
| rename "Authentication.*" as *
| stats count as total_failures dc(user) as unique_users values(user) as targeted_users by src
| where total_failures > 20 AND unique_users > 5
| lookup dnslookup clientip as src OUTPUT clienthost as src_dns
| lookup asset_lookup ip as src OUTPUT priority as asset_priority, category as asset_category
| eval urgency=case(asset_priority=="critical", "critical", asset_priority=="high", "high", true(), "medium")
| eval rule_name="Brute Force Against Multiple Accounts"
| eval rule_description="Multiple authentication failures from ".src." targeting ".unique_users." unique accounts"
| eval mitre_attack="T1110.001 - Password Guessing"spl
| tstats summariesonly=true count from datamodel=Authentication
where Authentication.action=failure
by Authentication.src, Authentication.user, _time span=5m
| rename "Authentication.*" as *
| stats count as total_failures dc(user) as unique_users values(user) as targeted_users by src
| where total_failures > 20 AND unique_users > 5
| lookup dnslookup clientip as src OUTPUT clienthost as src_dns
| lookup asset_lookup ip as src OUTPUT priority as asset_priority, category as asset_category
| eval urgency=case(asset_priority=="critical", "critical", asset_priority=="high", "high", true(), "medium")
| eval rule_name="Brute Force Against Multiple Accounts"
| eval rule_description="Multiple authentication failures from ".src." targeting ".unique_users." unique accounts"
| eval mitre_attack="T1110.001 - Password Guessing"Enrichment Best Practices
丰富上下文最佳实践
spl
| lookup identity_lookup identity as user OUTPUT department, manager, risk_score as user_risk
| lookup asset_lookup ip as src_ip OUTPUT asset_name, asset_category, asset_priority, asset_owner
| lookup threatintel_lookup ip as src_ip OUTPUT threat_type, threat_confidence, threat_source
| eval context=case(
isnotnull(threat_type), "Known threat: ".threat_type,
user_risk > 80, "High-risk user: risk score ".user_risk,
asset_priority=="critical", "Critical asset: ".asset_name,
true(), "Standard context"
)spl
| lookup identity_lookup identity as user OUTPUT department, manager, risk_score as user_risk
| lookup asset_lookup ip as src_ip OUTPUT asset_name, asset_category, asset_priority, asset_owner
| lookup threatintel_lookup ip as src_ip OUTPUT threat_type, threat_confidence, threat_source
| eval context=case(
isnotnull(threat_type), "Known threat: ".threat_type,
user_risk > 80, "High-risk user: risk score ".user_risk,
asset_priority=="critical", "Critical asset: ".asset_name,
true(), "Standard context"
)Performance Optimization
性能优化
Use Data Models with tstats
使用数据模型与tstats
spl
| tstats summariesonly=true count from datamodel=Network_Traffic
where All_Traffic.action=allowed
by All_Traffic.src_ip, All_Traffic.dest_ip, All_Traffic.dest_port, _time span=1h
| rename "All_Traffic.*" as *spl
| tstats summariesonly=true count from datamodel=Network_Traffic
where All_Traffic.action=allowed
by All_Traffic.src_ip, All_Traffic.dest_ip, All_Traffic.dest_port, _time span=1h
| rename "All_Traffic.*" as *Limit Time Ranges and Use Indexed Fields
限制时间范围并使用索引字段
spl
index=wineventlog source="WinEventLog:Security" EventCode=4688
earliest=-15m latest=now()
| where NOT match(New_Process_Name, "(?i)(svchost|csrss|lsass|services)")spl
index=wineventlog source="WinEventLog:Security" EventCode=4688
earliest=-15m latest=now()
| where NOT match(New_Process_Name, "(?i)(svchost|csrss|lsass|services)")Use Summary Indexing for Historical Baselines
使用摘要索引存储历史基线
spl
| tstats count from datamodel=Authentication where Authentication.action=failure by Authentication.src, _time span=1h
| collect index=summary source="auth_failure_baseline" marker="report_name=auth_failure_hourly"spl
| tstats count from datamodel=Authentication where Authentication.action=failure by Authentication.src, _time span=1h
| collect index=summary source="auth_failure_baseline" marker="report_name=auth_failure_hourly"Testing and Validation
测试与验证
Test Against Known Attack Patterns
针对已知攻击模式测试
spl
| makeresults count=1
| eval src_ip="10.0.0.50", failed_logins=25, unique_users=8, severity="high"
| eval description="Test brute force detection"
| append [
search index=wineventlog sourcetype=WinEventLog:Security EventCode=4625
earliest=-24h latest=now()
| stats count as failed_logins dc(TargetUserName) as unique_users by src_ip
| where failed_logins > 10 AND unique_users > 3
| eval severity="high"
]spl
| makeresults count=1
| eval src_ip="10.0.0.50", failed_logins=25, unique_users=8, severity="high"
| eval description="Test brute force detection"
| append [
search index=wineventlog sourcetype=WinEventLog:Security EventCode=4625
earliest=-24h latest=now()
| stats count as failed_logins dc(TargetUserName) as unique_users by src_ip
| where failed_logins > 10 AND unique_users > 3
| eval severity="high"
]Calculate Detection Metrics
计算检测指标
spl
index=notable
| search rule_name="Brute Force*"
| stats count as total_alerts count(eval(status_label="Closed - True Positive")) as true_positives count(eval(status_label="Closed - False Positive")) as false_positives by rule_name
| eval precision=round(true_positives / (true_positives + false_positives) * 100, 2)
| eval fpr=round(false_positives / total_alerts * 100, 2)spl
index=notable
| search rule_name="Brute Force*"
| stats count as total_alerts count(eval(status_label="Closed - True Positive")) as true_positives count(eval(status_label="Closed - False Positive")) as false_positives by rule_name
| eval precision=round(true_positives / (true_positives + false_positives) * 100, 2)
| eval fpr=round(false_positives / total_alerts * 100, 2)MITRE ATT&CK Mapping
MITRE ATT&CK映射
| Technique ID | Technique Name | SPL Detection Approach |
|---|---|---|
| T1110.001 | Password Guessing | Threshold on EventCode 4625 by src_ip |
| T1059.001 | PowerShell | Pattern match on EventCode 4104 ScriptBlockText |
| T1021.002 | SMB/Windows Admin Shares | Logon Type 3 with dc(dest) threshold |
| T1048 | Exfiltration Over C2 | bytes_out aggregation over time window |
| T1053.005 | Scheduled Task | EventCode 4698 with suspicious command patterns |
| T1003.001 | LSASS Memory | Process access to lsass.exe via Sysmon EventCode 10 |
| 技术ID | 技术名称 | SPL检测方法 |
|---|---|---|
| T1110.001 | 密码猜测 | 基于src_ip统计EventCode 4625的阈值 |
| T1059.001 | PowerShell | 匹配EventCode 4104的ScriptBlockText模式 |
| T1021.002 | SMB/Windows管理共享 | Logon Type 3结合dc(dest)阈值 |
| T1048 | 通过C2泄露数据 | 时间窗口内bytes_out聚合统计 |
| T1053.005 | 计划任务 | EventCode 4698结合可疑命令模式 |
| T1003.001 | LSASS内存 | 通过Sysmon EventCode 10检测对lsass.exe的进程访问 |