Mastering SIEM Correlation Rules: From Fundamentals to Advanced Threat Detection#

1
<group name="correlation_rules,">
2
  <rule id="100001" level="12" frequency="5" timeframe="120">
3
    <decoded_as>json</decoded_as>
4
    <if_sid>100000</if_sid>
5
    <match>pattern_to_match</match>
6
    <same_srcip />
7
    <description>Rule description</description>
8
    <mitre>
9
      <id>T1110</id>
10
    </mitre>
11
  </rule>
12
</group>

1
<!-- Complex rule using multiple functions -->
2
<rule id="700001" level="12">
3
  <if_group>web_attack</if_group>
4
  <match>contains("UNION SELECT") OR contains("1=1")</match>
5
  <frequency>count() > 5</frequency>
6
  <timeframe>60</timeframe>
7
  <same_srcip />
8
  <description>Persistent SQL injection attempts</description>
9
</rule>

1
<group name="lateral_movement_detection,">
2
  <!-- Rule 1: Detect multiple IP logins -->
3
  <rule id="800001" level="10">
4
    <decoded_as>json</decoded_as>
5
    <field name="event.action">authentication_success</field>
6
    <unique_diff>source.ip</unique_diff>
7
    <frequency>5</frequency>
8
    <timeframe>600</timeframe>
9
    <description>User logged in from 5+ different IPs in 10 minutes</description>
10
    <mitre>
11
      <id>T1021</id>
12
    </mitre>
13
  </rule>
14

15
  <!-- Rule 2: Unusual process execution paths -->
16
  <rule id="800002" level="9">
17
    <field name="process.executable">^C:\\Windows\\Temp\\</field>
18
    <field name="process.parent">^C:\\Windows\\System32\\</field>
19
    <description>Suspicious process spawn from temp directory</description>
20
    <mitre>
21
      <id>T1055</id>
22
    </mitre>
23
  </rule>
24

25
  <!-- Rule 3: Correlation - Movement after compromise -->
26
  <rule id="800003" level="14">
27
    <if_matched_sid>800001,800002</if_matched_sid>
28
    <same_user />
29
    <timeframe>1800</timeframe>
30
    <description>CRITICAL: Lateral movement pattern detected</description>
31
  </rule>
32
</group>

1
<group name="privilege_escalation,">
2
  <!-- Sudden privilege gain -->
3
  <rule id="900001" level="12">
4
    <field name="event.action">group_membership_add</field>
5
    <field name="group.name">Domain Admins|Enterprise Admins</field>
6
    <description>User added to high-privilege group</description>
7
    <mitre>
8
      <id>T1078.002</id>
9
    </mitre>
10
  </rule>
11

12
  <!-- Suspicious privilege usage -->
13
  <rule id="900002" level="10">
14
    <if_matched_sid>900001</if_matched_sid>
15
    <field name="event.action">sensitive_file_access</field>
16
    <timeframe>300</timeframe>
17
    <same_user />
18
    <description>Immediate sensitive access after privilege escalation</description>
19
  </rule>
20
</group>

1
<group name="data_exfiltration,">
2
  <!-- Large data transfer -->
3
  <rule id="1000001" level="8">
4
    <field name="network.bytes_sent">^[0-9]{8,}</field> <!-- 10MB+ -->
5
    <field name="destination.port">443|80|22</field>
6
    <description>Large outbound data transfer detected</description>
7
  </rule>
8

9
  <!-- Unusual destination -->
10
  <rule id="1000002" level="10">
11
    <if_matched_sid>1000001</if_matched_sid>
12
    <list field="destination.ip" lookup="not_match_key">etc/lists/trusted_ips</list>
13
    <description>Data transfer to untrusted destination</description>
14
  </rule>
15

16
  <!-- Exfiltration pattern -->
17
  <rule id="1000003" level="13">
18
    <if_matched_sid>1000002</if_matched_sid>
19
    <frequency>3</frequency>
20
    <timeframe>3600</timeframe>
21
    <same_srcip />
22
    <description>Sustained data exfiltration pattern detected</description>
23
    <mitre>
24
      <id>T1041</id>
25
    </mitre>
26
  </rule>
27
</group>

1
<!-- Example: Escalating severity based on context -->
2
<group name="escalating_severity,">
3
  <!-- Level 5: Single failed login -->
4
  <rule id="120001" level="5">
5
    <field name="event.outcome">failure</field>
6
    <description>Failed authentication attempt</description>
7
  </rule>
8

9
  <!-- Level 10: Multiple failures -->
10
  <rule id="120002" level="10" frequency="5">
11
    <if_matched_sid>120001</if_matched_sid>
12
    <timeframe>300</timeframe>
13
    <description>Multiple authentication failures</description>
14
  </rule>
15

16
  <!-- Level 14: Successful after failures -->
17
  <rule id="120003" level="14">
18
    <if_matched_sid>120002</if_matched_sid>
19
    <field name="event.outcome">success</field>
20
    <timeframe>600</timeframe>
21
    <description>Successful login after brute force attempts</description>
22
  </rule>
23
</group>

1
Kernel Events → eBPF Programs → BPF Maps → Wazuh Agent → Wazuh Manager → Correlation Rules

1
#!/usr/bin/env python3
2
# Enhanced eBPF monitoring with Wazuh integration
3

4
from bcc import BPF
5
import json
6
import socket
7
import time
8

9
# eBPF program for comprehensive TCP monitoring
10
bpf_program = """
11
#include <uapi/linux/ptrace.h>
12
#include <net/sock.h>
13
#include <bcc/proto.h>
14

15
BPF_HASH(connections, u32, u64);
16
BPF_PERF_OUTPUT(events);
17

18
struct data_t {
19
    u32 pid;
20
    u32 saddr;
21
    u32 daddr;
22
    u16 dport;
23
    char comm[TASK_COMM_LEN];
24
    u64 timestamp;
25
};
26

27
int trace_connect(struct pt_regs *ctx, struct sock *sk) {
28
    struct data_t data = {};
29
    u32 pid = bpf_get_current_pid_tgid() >> 32;
30

31
    data.pid = pid;
32
    data.saddr = sk->__sk_common.skc_rcv_saddr;
33
    data.daddr = sk->__sk_common.skc_daddr;
34
    data.dport = sk->__sk_common.skc_dport;
35
    data.timestamp = bpf_ktime_get_ns();
36

37
    bpf_get_current_comm(&data.comm, sizeof(data.comm));
38
    events.perf_submit(ctx, &data, sizeof(data));
39

40
    return 0;
41
}
42
"""
43

44
def send_to_wazuh(event_data):
45
    """Send formatted event to Wazuh via local socket"""
46
    wazuh_socket = '/var/ossec/queue/sockets/queue'
47
    message = f"1:eBPF:{json.dumps(event_data)}"
48

49
    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
50
        sock.sendto(message.encode(), wazuh_socket)
51

52
# Initialize BPF
53
b = BPF(text=bpf_program)
54
b.attach_kprobe(event="tcp_connect", fn_name="trace_connect")
55

56
# Process events
57
def process_event(cpu, data, size):
58
    event = b["events"].event(data)
59
    event_dict = {
60
        "pid": event.pid,
61
        "process": event.comm.decode('utf-8', 'ignore'),
62
        "src_ip": socket.inet_ntoa(event.saddr.to_bytes(4, 'little')),
63
        "dst_ip": socket.inet_ntoa(event.daddr.to_bytes(4, 'little')),
64
        "dst_port": socket.ntohs(event.dport),
65
        "timestamp": event.timestamp
66
    }
67
    send_to_wazuh(event_dict)
68

69
b["events"].open_perf_buffer(process_event)
70

71
while True:
72
    b.perf_buffer_poll()

1
#!/usr/bin/env python3
2
# Correlation rule testing framework
3

4
import json
5
import subprocess
6
import tempfile
7
from datetime import datetime
8

9
class RuleTester:
10
    def __init__(self, rule_file):
11
        self.rule_file = rule_file
12
        self.test_cases = []
13

14
    def add_test_case(self, log_entry, expected_rule_id, should_trigger=True):
15
        """Add a test case for rule validation"""
16
        self.test_cases.append({
17
            'log': log_entry,
18
            'expected_rule': expected_rule_id,
19
            'should_trigger': should_trigger
20
        })
21

22
    def run_tests(self):
23
        """Execute all test cases"""
24
        results = []
25

26
        for test in self.test_cases:
27
            # Create temporary file with test log
28
            with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
29
                f.write(test['log'])
30
                temp_file = f.name
31

32
            # Run through wazuh-logtest
33
            result = subprocess.run(
34
                ['/var/ossec/bin/wazuh-logtest'],
35
                stdin=open(temp_file),
36
                capture_output=True,
37
                text=True
38
            )
39

40
            # Parse output
41
            triggered = test['expected_rule'] in result.stdout
42
            success = triggered == test['should_trigger']
43

44
            results.append({
45
                'test': test,
46
                'success': success,
47
                'output': result.stdout
48
            })
49

50
        return results
51

52
    def generate_report(self, results):
53
        """Generate test report"""
54
        total = len(results)
55
        passed = sum(1 for r in results if r['success'])
56

57
        report = f"""
58
        Rule Testing Report
59
        ==================
60
        Date: {datetime.now()}
61
        Rule File: {self.rule_file}
62

63
        Results: {passed}/{total} tests passed
64

65
        Details:
66
        """
67

68
        for i, result in enumerate(results, 1):
69
            status = "✓ PASS" if result['success'] else "✗ FAIL"
70
            report += f"\n{i}. {status}: {result['test']['expected_rule']}"
71
            if not result['success']:
72
                report += f"\n   Expected: {result['test']['should_trigger']}"
73
                report += f"\n   Log: {result['test']['log'][:100]}..."
74

75
        return report
76

77
# Example usage
78
tester = RuleTester('/var/ossec/etc/rules/local_rules.xml')
79

80
# Add test cases for brute force detection
81
tester.add_test_case(
82
    'Failed login for user admin from 192.168.1.100',
83
    expected_rule_id='110001',
84
    should_trigger=False  # Single failure shouldn't trigger
85
)
86

87
# Test multiple failures
88
for i in range(10):
89
    tester.add_test_case(
90
        f'Failed login attempt {i} from 192.168.1.100',
91
        expected_rule_id='110001',
92
        should_trigger=(i == 9)  # Should trigger on 10th attempt
93
    )
94

95
results = tester.run_tests()
96
print(tester.generate_report(results))

1
# Rule effectiveness metrics
2
def calculate_rule_metrics(rule_id, time_period):
3
    """Calculate key metrics for rule performance"""
4

5
    metrics = {
6
        'total_triggers': count_rule_triggers(rule_id, time_period),
7
        'true_positives': count_validated_alerts(rule_id, time_period),
8
        'false_positives': count_false_positives(rule_id, time_period),
9
        'detection_time': average_detection_time(rule_id, time_period),
10
        'coverage': calculate_threat_coverage(rule_id)
11
    }
12

13
    # Calculate derived metrics
14
    if metrics['total_triggers'] > 0:
15
        metrics['precision'] = metrics['true_positives'] / metrics['total_triggers']
16
        metrics['false_positive_rate'] = metrics['false_positives'] / metrics['total_triggers']
17

18
    return metrics
19

20
# Generate rule effectiveness report
21
def generate_effectiveness_report(rules):
22
    """Generate comprehensive effectiveness report"""
23

24
    report = []
25
    for rule_id in rules:
26
        metrics = calculate_rule_metrics(rule_id, '30d')
27

28
        effectiveness_score = (
29
            metrics['precision'] * 0.4 +
30
            (1 - metrics['false_positive_rate']) * 0.3 +
31
            metrics['coverage'] * 0.3
32
        )
33

34
        report.append({
35
            'rule_id': rule_id,
36
            'effectiveness_score': effectiveness_score,
37
            'metrics': metrics,
38
            'recommendation': get_optimization_recommendation(metrics)
39
        })
40

41
    return sorted(report, key=lambda x: x['effectiveness_score'], reverse=True)

1
<!-- Active response configuration -->
2
<ossec_config>
3
  <active-response>
4
    <command>firewall-drop</command>
5
    <location>local</location>
6
    <rules_id>110001,110006,110007</rules_id>
7
    <timeout>3600</timeout>
8
  </active-response>
9

10
  <active-response>
11
    <command>disable-account</command>
12
    <location>local</location>
13
    <rules_id>900001,110008</rules_id>
14
  </active-response>
15

16
  <active-response>
17
    <command>isolate-endpoint</command>
18
    <location>local</location>
19
    <rules_id>110006</rules_id>
20
  </active-response>
21
</ossec_config>

1
#!/usr/bin/env python3
2
# SOAR platform integration for automated response
3

4
import requests
5
import json
6
from typing import Dict, Any
7

8
class SOARIntegration:
9
    def __init__(self, soar_url: str, api_key: str):
10
        self.soar_url = soar_url
11
        self.api_key = api_key
12
        self.headers = {
13
            'Authorization': f'Bearer {api_key}',
14
            'Content-Type': 'application/json'
15
        }
16

17
    def create_incident(self, alert_data: Dict[str, Any]) -> str:
18
        """Create incident in SOAR platform"""
19

20
        incident = {
21
            'title': alert_data['rule']['description'],
22
            'severity': self.map_severity(alert_data['rule']['level']),
23
            'source': 'Wazuh SIEM',
24
            'details': {
25
                'rule_id': alert_data['rule']['id'],
26
                'mitre_tactics': alert_data['rule'].get('mitre', {}).get('id', []),
27
                'source_ip': alert_data.get('data', {}).get('srcip'),
28
                'user': alert_data.get('data', {}).get('user'),
29
                'timestamp': alert_data['timestamp']
30
            },
31
            'automated_actions': self.get_automated_actions(alert_data['rule']['id'])
32
        }
33

34
        response = requests.post(
35
            f'{self.soar_url}/api/incidents',
36
            headers=self.headers,
37
            json=incident
38
        )
39

40
        return response.json()['incident_id']
41

42
    def map_severity(self, wazuh_level: int) -> str:
43
        """Map Wazuh severity to SOAR severity"""
44
        if wazuh_level >= 12:
45
            return 'critical'
46
        elif wazuh_level >= 9:
47
            return 'high'
48
        elif wazuh_level >= 6:
49
            return 'medium'
50
        else:
51
            return 'low'
52

53
    def get_automated_actions(self, rule_id: str) -> list:
54
        """Define automated actions based on rule"""
55

56
        action_mapping = {
57
            '110001': ['block_ip', 'gather_forensics'],
58
            '110006': ['isolate_endpoint', 'snapshot_system', 'notify_soc'],
59
            '900001': ['disable_account', 'force_password_reset'],
60
            '1000003': ['block_data_transfer', 'alert_dlp_team']
61
        }
62

63
        return action_mapping.get(rule_id, ['investigate'])
64

65
# Webhook receiver for Wazuh alerts
66
from flask import Flask, request
67

68
app = Flask(__name__)
69
soar = SOARIntegration('https://soar.company.com', 'API_KEY')
70

71
@app.route('/wazuh/webhook', methods=['POST'])
72
def handle_wazuh_alert():
73
    alert = request.json
74

75
    # Filter for high-severity alerts
76
    if alert['rule']['level'] >= 10:
77
        incident_id = soar.create_incident(alert)
78
        return {'status': 'success', 'incident_id': incident_id}
79

80
    return {'status': 'logged'}
81

82
if __name__ == '__main__':
83
    app.run(host='0.0.0.0', port=8080, ssl_context='adhoc')