# building-detection-rule-with-splunk-spl

Build effective detection rules using Splunk Search Processing Language (SPL) correlation searches to identify security threats in SOC environments.

- **Kind:** skill
- **Source:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **Page:** https://forefy.com/skills/e3e66067-1e93-4456-ab11-735de63dd443
- **API (JSON + files):** https://forefy.com/api/asr/e3e66067-1e93-4456-ab11-735de63dd443

---

## LICENSE

```

```

## SKILL.md

---
name: building-detection-rule-with-splunk-spl
description: Build effective detection rules using Splunk Search Processing Language
  (SPL) correlation searches to identify security threats in SOC environments.
domain: cybersecurity
subdomain: soc-operations
tags:
- splunk
- spl
- detection-engineering
- correlation-search
- siem
- soc
- threat-detection
- enterprise-security
version: '1.0'
author: mahipal
license: Apache-2.0
d3fend_techniques:
- Executable Denylisting
- Execution Isolation
- File Metadata Consistency Validation
- Content Format Conversion
- File Content Analysis
nist_csf:
- DE.CM-01
- DE.AE-02
- RS.MA-01
- DE.AE-06
mitre_attack:
- T1059.001
- T1003.001
- T1021.002
- T1110.003
- T1053.005
- T1048
---

# Building Detection Rules with 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.


## 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

## 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

## Core SPL Detection Rule Patterns

### 1. Threshold-Based Detection

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"
```

### 2. Sequence-Based Detection (Failed Login Followed by Success)

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"
```

### 3. Anomaly Detection with Baseline Comparison

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"
```

### 4. Lateral Movement Detection

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"
```

### 5. Data Exfiltration Detection

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"
```

### 6. PowerShell Suspicious Execution Detection

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
```

## Building Correlation Searches in Splunk ES

### Step-by-Step Process

1. **Define the Use Case**: Map to MITRE ATT&CK technique and define what behavior to detect
2. **Identify Data Sources**: Determine which indexes and sourcetypes contain relevant events
3. **Write the Base Search**: Build SPL that extracts relevant events
4. **Add Aggregation**: Use `stats`, `eventstats`, or `streamstats` to summarize
5. **Apply Thresholds**: Set conditions with `where` clause that distinguish normal from anomalous
6. **Enrich Context**: Add lookups for asset information, identity data, and threat intelligence
7. **Configure Notable Event**: Set severity, urgency, and description fields
8. **Schedule and Test**: Run against historical data and validate detection accuracy

### 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"
```

### 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"
)
```

## Performance Optimization

### Use Data Models with 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 *
```

### 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)")
```

### 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"
```

## 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"
]
```

### 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)
```

## MITRE ATT&CK Mapping

| 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 |

## References

- [Splunk ES Correlation Searches Best Practices](https://detect.fyi/splunk-es-correlation-searches-rules-best-cool-practices-06ef94884170)
- [Writing Practical Splunk Detection Rules](https://medium.com/@vitbukac/practical-splunk-detection-rules-how-to-part-1-crawl-a24bc39a4b9d)
- [Configure Correlation Searches - Splunk Documentation](https://help.splunk.com/en/splunk-enterprise-security-8/splunk-app-for-pci-compliance/installation-and-configuration-manual/6.1/configure-correlation-searches/configure-correlation-searches)
- [SOC Prime - Correlation Events in Splunk](https://socprime.com/blog/creating-correlation-events-in-splunk-using-alerts/)

## assets

```

```

## assets/template.md

# Splunk SPL Detection Rule Template

## Rule Metadata

| Field | Value |
|---|---|
| Rule Name | |
| Rule ID | |
| Description | |
| Author | |
| Date Created | |
| Last Modified | |
| Severity | |
| MITRE ATT&CK | |
| Data Sources | |
| Status | Draft / Testing / Production / Retired |

## SPL Query

```spl
| tstats summariesonly=true count
    from datamodel=<DataModel>
    where <conditions>
    by <fields>, _time span=<interval>
| rename "<DataModel>.*" as *
| stats <aggregation> by <grouping_fields>
| where <threshold_condition>
| lookup asset_lookup ip as src OUTPUT asset_name, asset_priority
| lookup identity_lookup identity as user OUTPUT department, manager
| eval severity=case(<critical_condition>, "critical", <high_condition>, "high", true(), "medium")
| eval description="<dynamic description string>"
| eval mitre_technique="<T-number>"
```

## Detection Logic

### What This Rule Detects
<!-- Describe the adversary behavior being detected -->

### Data Sources Required
<!-- List all required data sources and their sourcetypes -->

| Source | Sourcetype | Index | Required Fields |
|---|---|---|---|
| | | | |

### Threshold Justification
<!-- Explain why the threshold values were chosen -->

### Enrichment Details
<!-- List lookups and threat intel sources used -->

## Testing Plan

### True Positive Test
<!-- Describe how to simulate the attack to validate detection -->

```
Step 1:
Step 2:
Step 3:
Expected Result:
```

### False Positive Analysis
<!-- Document known benign scenarios that trigger this rule -->

| Scenario | Source | Mitigation |
|---|---|---|
| | | |

## Tuning History

| Date | Change | Reason | Impact |
|---|---|---|---|
| | | | |

## Correlation Search Configuration

```
Schedule: */15 * * * *
Time Window: earliest=-20m latest=now
Suppress: 1h by src_ip
Notable Event Security Domain: threat
Adaptive Response: <actions>
```

## Analyst Guidance

### Triage Steps
1.
2.
3.

### Escalation Criteria
-

### Related Rules
-

## references

```

```

## references/api-reference.md

# API Reference: Splunk SPL Detection Rules

## Splunk REST API - Saved Searches
```
POST /servicesNS/{owner}/{app}/saved/searches
Authorization: Bearer TOKEN
```
| Field | Description |
|-------|-------------|
| `name` | Saved search name |
| `search` | SPL query string |
| `is_scheduled` | 1 for scheduled |
| `cron_schedule` | Cron expression (e.g., `*/5 * * * *`) |
| `dispatch.earliest_time` | Start of search window |
| `alert.severity` | 1-5 (info to critical) |
| `alert_type` | `number of events` |
| `alert_threshold` | Trigger threshold |

## Key SPL Commands
| Command | Description |
|---------|-------------|
| `stats count by field` | Aggregate events |
| `where count > N` | Filter results |
| `table field1, field2` | Select fields |
| `eval` | Compute new fields |
| `lookup` | Enrich from lookup table |
| `tstats` | Accelerated data model search |
| `join` | Join two datasets |

## Windows Event IDs for Detection
| EventCode | Source | Description |
|-----------|--------|-------------|
| 4624 | Security | Successful logon |
| 4625 | Security | Failed logon |
| 4648 | Security | Explicit credential logon |
| 4698 | Security | Scheduled task created |
| 4104 | PowerShell | Script block logging |
| 1 | Sysmon | Process creation |
| 3 | Sysmon | Network connection |
| 10 | Sysmon | Process access |

## Alert Severity Levels
| Level | Value | Description |
|-------|-------|-------------|
| Info | 1 | Informational |
| Low | 2 | Low risk |
| Medium | 3 | Medium risk |
| High | 4 | High risk |
| Critical | 5 | Critical risk |

## references/standards.md

# Standards and References - Splunk SPL Detection Rules

## Industry Standards

### MITRE ATT&CK Framework
- Primary mapping standard for detection rule categorization
- Version 18.1 (December 2025) is the latest release
- Use ATT&CK Navigator for visual coverage mapping

### Splunk Common Information Model (CIM)
- Standard field naming convention for normalized data
- Data models: Authentication, Network_Traffic, Endpoint, Web, Email
- Enables cross-sourcetype correlation searches

### NIST SP 800-92 - Guide to Computer Security Log Management
- Log management planning and policy guidance
- Defines log collection, analysis, and retention best practices

### NIST SP 800-61 Rev 2 - Computer Security Incident Handling Guide
- Incident detection and analysis procedures
- Defines severity classification for generated alerts

## Splunk Enterprise Security Resources

### Correlation Search Framework
- Supports scheduled searches with adaptive response actions
- Risk-based alerting (RBA) aggregates risk events by entity
- Notable events are the primary output for SOC analyst review

### Data Model Acceleration
- tstats provides fast summary-based searching
- Accelerated data models required for production correlation searches
- CIM compliance ensures cross-source detection capability

### Key Splunk SPL Commands for Detection

| Command | Purpose |
|---|---|
| `stats` | Aggregate events by fields |
| `tstats` | Fast search over accelerated data models |
| `eventstats` | Add aggregated stats inline to events |
| `streamstats` | Running statistics over ordered events |
| `transaction` | Group related events into transactions |
| `lookup` | Enrich events with external data |
| `where` | Filter results with boolean expressions |
| `eval` | Create calculated fields |

## Detection Engineering Maturity Model

### Level 1 - Basic Threshold Rules
- Simple count-based thresholds
- Single data source correlation

### Level 2 - Multi-Source Correlation
- Cross-source event correlation
- Asset and identity enrichment

### Level 3 - Behavioral Analytics
- Baseline deviation detection
- User and entity behavior profiling

### Level 4 - Risk-Based Alerting
- Cumulative risk scoring per entity
- Context-aware severity assignment

### Level 5 - Automated Response
- Adaptive response action integration
- SOAR playbook triggering from notable events

## references/workflows.md

# Workflows - Building Detection Rules with Splunk SPL

## Detection Rule Development Workflow

```
1. Identify Threat Scenario
   |
   v
2. Map to MITRE ATT&CK Technique
   |
   v
3. Identify Required Data Sources
   |
   v
4. Validate Data Availability in Splunk
   |
   v
5. Write Base SPL Query
   |
   v
6. Add Aggregation and Filtering
   |
   v
7. Add Enrichment (Lookups, Threat Intel)
   |
   v
8. Test Against Historical Data
   |
   v
9. Calculate False Positive Rate
   |
   v
10. Deploy as Correlation Search
    |
    v
11. Monitor Detection Metrics
    |
    v
12. Tune and Iterate
```

## Rule Testing Workflow

### Phase 1: Development
- Write SPL query in Search & Reporting
- Test with `earliest=-7d latest=now()`
- Verify expected events are captured

### Phase 2: Validation
- Run Atomic Red Team tests to generate known-bad events
- Confirm detection triggers on simulated attacks
- Check no duplicate or redundant notable events generated

### Phase 3: Tuning
- Identify false positives from 7-day burn-in period
- Add exclusions for known benign activity
- Adjust thresholds based on environment baseline

### Phase 4: Production
- Schedule as correlation search in ES
- Configure adaptive response actions
- Set notable event severity and urgency mapping

## Correlation Search Scheduling Guide

| Rule Severity | Schedule Interval | Time Window |
|---|---|---|
| Critical | Every 5 minutes | 10 minutes |
| High | Every 15 minutes | 20 minutes |
| Medium | Every 30 minutes | 35 minutes |
| Low | Every 60 minutes | 65 minutes |
| Informational | Every 4 hours | 4.5 hours |

Note: Time window should slightly exceed schedule interval to prevent event gaps.

## Alert Output Workflow

```
Correlation Search Fires
    |
    v
Notable Event Created in ES
    |
    v
SOC Analyst Reviews in Incident Review Dashboard
    |
    v
Analyst Triages: True Positive / False Positive / Needs Investigation
    |
    v
True Positive --> Create Investigation --> Escalate if needed
False Positive --> Document exclusion --> Update correlation search
```

## scripts

```

```

## scripts/agent.py

```python
#!/usr/bin/env python3
"""Splunk SPL Detection Rule Builder Agent - Generates and validates Splunk detection rules."""

import json
import logging
import os
import argparse
from datetime import datetime

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

DETECTION_TEMPLATES = {
    "brute_force": {
        "spl": 'index=main sourcetype=WinEventLog:Security EventCode=4625 | stats count by src_ip, user | where count > {threshold}',
        "description": "Detect brute force login attempts",
        "mitre": "T1110",
        "severity": "high",
    },
    "lateral_movement_rdp": {
        "spl": 'index=main sourcetype=WinEventLog:Security EventCode=4624 Logon_Type=10 | stats count by src_ip, dest, user | where count > 1',
        "description": "Detect RDP lateral movement",
        "mitre": "T1021.001",
        "severity": "high",
    },
    "powershell_encoded": {
        "spl": 'index=main sourcetype=WinEventLog:Security EventCode=4104 ScriptBlockText="*-EncodedCommand*" OR ScriptBlockText="*FromBase64String*" | table _time, ComputerName, ScriptBlockText',
        "description": "Detect encoded PowerShell execution",
        "mitre": "T1059.001",
        "severity": "critical",
    },
    "suspicious_process": {
        "spl": 'index=main sourcetype=sysmon EventCode=1 (ParentImage="*\\\\cmd.exe" OR ParentImage="*\\\\powershell.exe") (Image="*\\\\whoami.exe" OR Image="*\\\\net.exe" OR Image="*\\\\nltest.exe") | table _time, Computer, User, ParentImage, Image, CommandLine',
        "description": "Detect suspicious child process spawning",
        "mitre": "T1059.003",
        "severity": "high",
    },
    "scheduled_task_creation": {
        "spl": 'index=main sourcetype=WinEventLog:Security EventCode=4698 | table _time, SubjectUserName, TaskName, TaskContent',
        "description": "Detect new scheduled task creation",
        "mitre": "T1053.005",
        "severity": "medium",
    },
    "credential_dumping": {
        "spl": 'index=main sourcetype=sysmon EventCode=10 TargetImage="*\\\\lsass.exe" GrantedAccess IN ("0x1010", "0x1038", "0x1fffff") | table _time, SourceImage, TargetImage, GrantedAccess',
        "description": "Detect LSASS memory access (credential dumping)",
        "mitre": "T1003.001",
        "severity": "critical",
    },
    "data_exfiltration": {
        "spl": 'index=main sourcetype=proxy | stats sum(bytes_out) as total_bytes by src_ip, dest | where total_bytes > {threshold_bytes} | eval MB=round(total_bytes/1024/1024,2)',
        "description": "Detect large data transfers (exfiltration)",
        "mitre": "T1048",
        "severity": "high",
    },
}


def generate_spl_rule(template_name, params=None):
    """Generate an SPL detection rule from template."""
    if template_name not in DETECTION_TEMPLATES:
        return {"error": f"Unknown template: {template_name}"}
    template = DETECTION_TEMPLATES[template_name]
    spl = template["spl"]
    if params:
        for key, value in params.items():
            spl = spl.replace(f"{{{key}}}", str(value))
    return {
        "name": template_name,
        "spl": spl,
        "description": template["description"],
        "mitre_technique": template["mitre"],
        "severity": template["severity"],
    }


def deploy_saved_search(splunk_url, token, rule_name, spl, severity="high"):
    """Deploy a saved search to Splunk via REST API."""
    headers = {"Authorization": f"Bearer {token}"}
    data = {
        "name": f"Detection - {rule_name}",
        "search": spl,
        "is_scheduled": 1,
        "cron_schedule": "*/5 * * * *",
        "dispatch.earliest_time": "-5m",
        "dispatch.latest_time": "now",
        "alert.severity": {"info": 1, "low": 2, "medium": 3, "high": 4, "critical": 5}.get(severity, 4),
        "alert_type": "number of events",
        "alert_comparator": "greater than",
        "alert_threshold": "0",
        "actions": "email",
    }
    try:
        resp = requests.post(f"{splunk_url}/servicesNS/admin/search/saved/searches", headers=headers, data=data, verify=not os.environ.get("SKIP_TLS_VERIFY", "").lower() == "true", timeout=30)  # Set SKIP_TLS_VERIFY=true for self-signed certs in lab environments
        return {"status": resp.status_code, "deployed": resp.status_code in (200, 201)}
    except requests.RequestException as e:
        return {"error": str(e)}


def generate_report(rules, deployment_results=None):
    """Generate detection rule report."""
    report = {
        "timestamp": datetime.utcnow().isoformat(),
        "rules_generated": len(rules),
        "rules": rules,
        "deployment_results": deployment_results or [],
    }
    print(f"SPL REPORT: {len(rules)} detection rules generated")
    return report


def main():
    parser = argparse.ArgumentParser(description="Splunk SPL Detection Rule Builder Agent")
    parser.add_argument("--templates", nargs="*", choices=list(DETECTION_TEMPLATES.keys()), default=list(DETECTION_TEMPLATES.keys()))
    parser.add_argument("--threshold", type=int, default=10)
    parser.add_argument("--threshold-bytes", type=int, default=104857600)
    parser.add_argument("--splunk-url", help="Splunk URL for deployment")
    parser.add_argument("--splunk-token", help="Splunk auth token")
    parser.add_argument("--output", default="splunk_rules.json")
    args = parser.parse_args()

    rules = []
    for template in args.templates:
        rule = generate_spl_rule(template, {"threshold": args.threshold, "threshold_bytes": args.threshold_bytes})
        rules.append(rule)

    deployments = []
    if args.splunk_url and args.splunk_token:
        for rule in rules:
            result = deploy_saved_search(args.splunk_url, args.splunk_token, rule["name"], rule["spl"], rule["severity"])
            deployments.append({"rule": rule["name"], "result": result})

    report = generate_report(rules, deployments)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2)
    logger.info("Report saved to %s", args.output)


if __name__ == "__main__":
    main()
```

## scripts/process.py

```python
#!/usr/bin/env python3
"""
Splunk SPL Detection Rule Builder and Validator

Generates, validates, and manages Splunk SPL detection rules
for SOC correlation searches. Supports MITRE ATT&CK mapping
and rule quality scoring.
"""

import json
import re
import hashlib
from datetime import datetime
from typing import Optional


MITRE_TECHNIQUES = {
    "T1110.001": {"name": "Password Guessing", "tactic": "Credential Access"},
    "T1110.003": {"name": "Password Spraying", "tactic": "Credential Access"},
    "T1059.001": {"name": "PowerShell", "tactic": "Execution"},
    "T1059.003": {"name": "Windows Command Shell", "tactic": "Execution"},
    "T1021.002": {"name": "SMB/Windows Admin Shares", "tactic": "Lateral Movement"},
    "T1021.001": {"name": "Remote Desktop Protocol", "tactic": "Lateral Movement"},
    "T1048": {"name": "Exfiltration Over C2 Channel", "tactic": "Exfiltration"},
    "T1048.003": {"name": "Exfiltration Over Unencrypted Protocol", "tactic": "Exfiltration"},
    "T1053.005": {"name": "Scheduled Task", "tactic": "Persistence"},
    "T1003.001": {"name": "LSASS Memory", "tactic": "Credential Access"},
    "T1078": {"name": "Valid Accounts", "tactic": "Defense Evasion"},
    "T1078.002": {"name": "Domain Accounts", "tactic": "Defense Evasion"},
    "T1547.001": {"name": "Registry Run Keys", "tactic": "Persistence"},
    "T1055": {"name": "Process Injection", "tactic": "Defense Evasion"},
    "T1071.001": {"name": "Web Protocols", "tactic": "Command and Control"},
    "T1036.005": {"name": "Match Legitimate Name", "tactic": "Defense Evasion"},
    "T1027": {"name": "Obfuscated Files or Information", "tactic": "Defense Evasion"},
    "T1218.011": {"name": "Rundll32", "tactic": "Defense Evasion"},
    "T1543.003": {"name": "Windows Service", "tactic": "Persistence"},
    "T1105": {"name": "Ingress Tool Transfer", "tactic": "Command and Control"},
}


class SplunkDetectionRule:
    """Represents a Splunk SPL detection rule with metadata and validation."""

    def __init__(
        self,
        name: str,
        description: str,
        spl_query: str,
        mitre_techniques: list,
        severity: str = "medium",
        schedule_cron: str = "*/15 * * * *",
        time_window: str = "-20m",
        data_sources: Optional[list] = None,
        false_positive_notes: Optional[list] = None,
    ):
        self.name = name
        self.description = description
        self.spl_query = spl_query
        self.mitre_techniques = mitre_techniques
        self.severity = severity
        self.schedule_cron = schedule_cron
        self.time_window = time_window
        self.data_sources = data_sources or []
        self.false_positive_notes = false_positive_notes or []
        self.created = datetime.utcnow().isoformat()
        self.rule_id = self._generate_rule_id()

    def _generate_rule_id(self) -> str:
        hash_input = f"{self.name}:{self.spl_query}"
        return f"SPL-{hashlib.sha256(hash_input.encode()).hexdigest()[:12].upper()}"

    def validate(self) -> dict:
        """Validate the SPL detection rule for common issues."""
        issues = []
        score = 100

        # Check for missing time constraint
        if "earliest=" not in self.spl_query and "span=" not in self.spl_query:
            issues.append("WARNING: No time constraint in query - may scan too much data")
            score -= 10

        # Check for wildcard-heavy searches
        wildcard_count = self.spl_query.count("*")
        if wildcard_count > 5:
            issues.append(f"WARNING: {wildcard_count} wildcards detected - may impact performance")
            score -= 5 * min(wildcard_count - 5, 4)

        # Check for aggregation
        agg_commands = ["stats", "eventstats", "streamstats", "tstats", "chart", "timechart"]
        has_aggregation = any(cmd in self.spl_query.lower() for cmd in agg_commands)
        if not has_aggregation:
            issues.append("WARNING: No aggregation command - rule may generate excessive alerts")
            score -= 15

        # Check for threshold
        if "where" not in self.spl_query.lower():
            issues.append("WARNING: No where clause - rule has no threshold filtering")
            score -= 15

        # Check for enrichment
        if "lookup" not in self.spl_query.lower():
            issues.append("INFO: No lookup enrichment - consider adding asset/identity context")
            score -= 5

        # Check MITRE mapping
        if not self.mitre_techniques:
            issues.append("WARNING: No MITRE ATT&CK technique mapped")
            score -= 10

        for tech_id in self.mitre_techniques:
            if tech_id not in MITRE_TECHNIQUES:
                issues.append(f"WARNING: Unknown MITRE technique ID: {tech_id}")
                score -= 5

        # Check severity is valid
        valid_severities = ["informational", "low", "medium", "high", "critical"]
        if self.severity not in valid_severities:
            issues.append(f"ERROR: Invalid severity '{self.severity}' - must be one of {valid_severities}")
            score -= 20

        # Check for eval description
        if "eval description" not in self.spl_query.lower() and "eval rule_description" not in self.spl_query.lower():
            issues.append("INFO: No description field in output - analysts will lack context")
            score -= 5

        # Check for CIM data model usage
        if "datamodel=" in self.spl_query.lower() or "tstats" in self.spl_query.lower():
            score += 5  # Bonus for using CIM-accelerated searches

        return {
            "rule_id": self.rule_id,
            "rule_name": self.name,
            "valid": score >= 60,
            "quality_score": max(0, min(100, score)),
            "issues": issues,
            "issue_count": len(issues),
        }

    def to_splunk_savedsearch_conf(self) -> str:
        """Generate Splunk savedsearches.conf stanza for the rule."""
        mitre_str = ", ".join(self.mitre_techniques)
        stanza = f"""[{self.name}]
search = {self.spl_query}
description = {self.description}
dispatch.earliest_time = {self.time_window}
dispatch.latest_time = now
cron_schedule = {self.schedule_cron}
is_scheduled = 1
enableSched = 1
alert.severity = {self._severity_to_int()}
alert.suppress = 1
alert.suppress.period = 1h
alert.suppress.fields = src_ip
action.notable = 1
action.notable.param.rule_title = {self.name}
action.notable.param.rule_description = {self.description}
action.notable.param.severity = {self.severity}
action.notable.param.security_domain = threat
action.notable.param.drilldown_name = View triggering events
action.notable.param.drilldown_search = {self.spl_query}
action.notable.param.mitre_attack = {mitre_str}
"""
        return stanza

    def _severity_to_int(self) -> int:
        mapping = {"informational": 1, "low": 2, "medium": 3, "high": 4, "critical": 5}
        return mapping.get(self.severity, 3)

    def to_json(self) -> str:
        return json.dumps(
            {
                "rule_id": self.rule_id,
                "name": self.name,
                "description": self.description,
                "spl_query": self.spl_query,
                "mitre_techniques": self.mitre_techniques,
                "severity": self.severity,
                "schedule_cron": self.schedule_cron,
                "time_window": self.time_window,
                "data_sources": self.data_sources,
                "false_positive_notes": self.false_positive_notes,
                "created": self.created,
            },
            indent=2,
        )


class DetectionRuleLibrary:
    """Manages a collection of Splunk detection rules."""

    def __init__(self):
        self.rules = []

    def add_rule(self, rule: SplunkDetectionRule):
        self.rules.append(rule)

    def validate_all(self) -> dict:
        results = {"total_rules": len(self.rules), "valid_rules": 0, "invalid_rules": 0, "details": []}
        for rule in self.rules:
            validation = rule.validate()
            results["details"].append(validation)
            if validation["valid"]:
                results["valid_rules"] += 1
            else:
                results["invalid_rules"] += 1
        return results

    def get_mitre_coverage(self) -> dict:
        coverage = {}
        for rule in self.rules:
            for tech_id in rule.mitre_techniques:
                if tech_id not in coverage:
                    coverage[tech_id] = {
                        "technique": MITRE_TECHNIQUES.get(tech_id, {}).get("name", "Unknown"),
                        "tactic": MITRE_TECHNIQUES.get(tech_id, {}).get("tactic", "Unknown"),
                        "rules": [],
                    }
                coverage[tech_id]["rules"].append(rule.name)
        return {
            "techniques_covered": len(coverage),
            "total_known_techniques": len(MITRE_TECHNIQUES),
            "coverage_percentage": round(len(coverage) / len(MITRE_TECHNIQUES) * 100, 1),
            "coverage_map": coverage,
        }

    def export_savedsearches_conf(self) -> str:
        output = "# Auto-generated Splunk savedsearches.conf\n"
        output += f"# Generated: {datetime.utcnow().isoformat()}\n"
        output += f"# Total Rules: {len(self.rules)}\n\n"
        for rule in self.rules:
            output += rule.to_splunk_savedsearch_conf() + "\n"
        return output


def build_sample_detection_library() -> DetectionRuleLibrary:
    """Build a sample detection rule library with common SOC use cases."""
    library = DetectionRuleLibrary()

    library.add_rule(
        SplunkDetectionRule(
            name="Brute Force - Multiple Failed Logins",
            description="Detects brute force attacks with multiple failed login attempts from a single source",
            spl_query=(
                '| 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 > 3 '
                '| lookup asset_lookup ip as src OUTPUT priority as asset_priority '
                '| eval severity=case(unique_users > 10, "critical", unique_users > 5, "high", true(), "medium") '
                '| eval description="Brute force detected from ".src." targeting ".unique_users." accounts"'
            ),
            mitre_techniques=["T1110.001"],
            severity="high",
            schedule_cron="*/5 * * * *",
            time_window="-10m",
            data_sources=["Windows Security Event Log", "Linux Auth Log"],
            false_positive_notes=["Service accounts with expired passwords", "Misconfigured applications"],
        )
    )

    library.add_rule(
        SplunkDetectionRule(
            name="Suspicious PowerShell Execution",
            description="Detects encoded or obfuscated PowerShell commands indicating potential malicious activity",
            spl_query=(
                'index=wineventlog sourcetype=WinEventLog:Security EventCode=4104 '
                '| where match(ScriptBlockText, "(?i)(encodedcommand|invoke-expression|iex|downloadstring|frombase64string|net\\.webclient|invoke-mimikatz)") '
                '| stats count values(ScriptBlockText) as commands by Computer, UserName '
                '| where count > 0 '
                '| lookup identity_lookup identity as UserName OUTPUT department, manager '
                '| eval severity="high" '
                '| eval description="Suspicious PowerShell on ".Computer." by ".UserName'
            ),
            mitre_techniques=["T1059.001", "T1027"],
            severity="high",
            data_sources=["Windows PowerShell Script Block Logging"],
            false_positive_notes=["IT automation scripts using encoded commands", "SCCM deployment scripts"],
        )
    )

    library.add_rule(
        SplunkDetectionRule(
            name="Lateral Movement - Multiple Host Access",
            description="Detects a user or source IP accessing an unusual number of hosts via network logon",
            spl_query=(
                '| tstats summariesonly=true dc(Authentication.dest) as unique_hosts '
                'from datamodel=Authentication where Authentication.action=success Authentication.Logon_Type=3 '
                'by Authentication.src, Authentication.user, _time span=1h '
                '| rename "Authentication.*" as * '
                '| where unique_hosts > 5 '
                '| lookup asset_lookup ip as src OUTPUT asset_name, asset_category '
                '| eval severity=case(unique_hosts > 20, "critical", unique_hosts > 10, "high", true(), "medium") '
                '| eval description=user." accessed ".unique_hosts." hosts from ".src." in 1 hour"'
            ),
            mitre_techniques=["T1021.002", "T1078.002"],
            severity="high",
            data_sources=["Windows Security Event Log"],
            false_positive_notes=["Vulnerability scanners", "IT management tools", "Software deployment systems"],
        )
    )

    return library


if __name__ == "__main__":
    library = build_sample_detection_library()

    print("=" * 70)
    print("SPLUNK SPL DETECTION RULE LIBRARY")
    print("=" * 70)

    # Validate all rules
    validation = library.validate_all()
    print(f"\nTotal Rules: {validation['total_rules']}")
    print(f"Valid Rules: {validation['valid_rules']}")
    print(f"Invalid Rules: {validation['invalid_rules']}")

    for detail in validation["details"]:
        print(f"\n--- {detail['rule_name']} ---")
        print(f"  Rule ID: {detail['rule_id']}")
        print(f"  Quality Score: {detail['quality_score']}/100")
        print(f"  Valid: {detail['valid']}")
        for issue in detail["issues"]:
            print(f"  {issue}")

    # MITRE coverage
    coverage = library.get_mitre_coverage()
    print(f"\nMITRE ATT&CK Coverage: {coverage['techniques_covered']}/{coverage['total_known_techniques']} ({coverage['coverage_percentage']}%)")
    for tech_id, info in coverage["coverage_map"].items():
        print(f"  {tech_id} ({info['technique']}): {', '.join(info['rules'])}")

    # Export savedsearches.conf
    conf = library.export_savedsearches_conf()
    print(f"\n{'=' * 70}")
    print("GENERATED savedsearches.conf")
    print("=" * 70)
    print(conf)
```