SIEM Event Collection and Correlation: A Comprehensive Technical Guide#

1
<!-- C:\Program Files (x86)\ossec-agent\ossec.conf -->
2
<ossec_config>
3
  <client>
4
    <server>
5
      <address>192.168.1.100</address>
6
      <port>1514</port>
7
      <protocol>tcp</protocol>
8
    </server>
9
    <config-profile>windows, plesk</config-profile>
10
  </client>
11

12
  <!-- Windows Event Log Collection -->
13
  <localfile>
14
    <location>Application</location>
15
    <log_format>eventchannel</log_format>
16
  </localfile>
17

18
  <localfile>
19
    <location>Security</location>
20
    <log_format>eventchannel</log_format>
21
  </localfile>
22

23
  <localfile>
24
    <location>System</location>
25
    <log_format>eventchannel</log_format>
26
  </localfile>
27
</ossec_config>

1
<!-- Plesk-specific log collection -->
2
<localfile>
3
  <location>C:\Program Files (x86)\Plesk\admin\logs\*.log</location>
4
  <log_format>syslog</log_format>
5
</localfile>
6

7
<!-- MySQL/MariaDB logs -->
8
<localfile>
9
  <location>C:\Program Files\MySQL\MySQL Server 8.0\data\*.err</location>
10
  <log_format>mysql_log</log_format>
11
</localfile>
12

13
<!-- ModSecurity logs for web application security -->
14
<localfile>
15
  <location>C:\Program Files (x86)\Plesk\ModSecurity\logs\audit.log</location>
16
  <log_format>apache</log_format>
17
</localfile>
18

19
<!-- FTP logs -->
20
<localfile>
21
  <location>C:\inetpub\logs\LogFiles\FTPSVC\*.log</location>
22
  <log_format>iis</log_format>
23
</localfile>

1
Endpoints → Regional Collectors → Central SIEM → Archive Storage

1
Log Sources → Kafka/Kinesis → Real-time Processing → SIEM Platform

1
On-Premises → Log Forwarders → Cloud SIEM → Data Lake

1
<!-- Step 1: Define decoders for authentication logs -->
2
<decoder name="auth_decoder">
3
  <program_name>sshd</program_name>
4
</decoder>
5

6
<decoder name="auth_extract">
7
  <parent>auth_decoder</parent>
8
  <regex>(\w+) for (\w+) from (\d+\.\d+\.\d+\.\d+)</regex>
9
  <order>status, user, srcip</order>
10
</decoder>
11

12
<!-- Step 2: Create base rules for login events -->
13
<group name="authentication,">
14
  <!-- Successful login -->
15
  <rule id="100001" level="3">
16
    <decoded_as>auth_extract</decoded_as>
17
    <field name="status">Accepted</field>
18
    <description>Successful authentication</description>
19
  </rule>
20

21
  <!-- Failed login -->
22
  <rule id="100002" level="5">
23
    <decoded_as>auth_extract</decoded_as>
24
    <field name="status">Failed</field>
25
    <description>Failed authentication attempt</description>
26
  </rule>
27
</group>
28

29
<!-- Step 3: Correlation rule for multiple failures -->
30
<group name="authentication_attack,">
31
  <rule id="100010" level="10" frequency="5" timeframe="120">
32
    <if_matched_sid>100002</if_matched_sid>
33
    <same_srcip />
34
    <description>Possible brute force attack - Multiple failed logins from same IP</description>
35
    <mitre>
36
      <id>T1110</id>
37
    </mitre>
38
  </rule>
39

40
  <!-- Step 4: Detect successful login after failures -->
41
  <rule id="100011" level="12" timeframe="300">
42
    <if_matched_sid>100010</if_matched_sid>
43
    <field name="status">Accepted</field>
44
    <same_srcip />
45
    <same_user />
46
    <description>CRITICAL: Successful login after brute force attempts - Possible compromise</description>
47
    <mitre>
48
      <id>T1078</id>
49
    </mitre>
50
  </rule>
51
</group>

1
<group name="data_exfiltration,">
2
  <!-- Large file transfer detection -->
3
  <rule id="200001" level="7">
4
    <decoded_as>firewall</decoded_as>
5
    <field name="bytes_sent">^[0-9]{7,}</field> <!-- 10MB+ -->
6
    <description>Large outbound data transfer detected</description>
7
  </rule>
8

9
  <!-- Correlation: Multiple large transfers -->
10
  <rule id="200010" level="12" frequency="3" timeframe="3600">
11
    <if_matched_sid>200001</if_matched_sid>
12
    <same_srcip />
13
    <description>Possible data exfiltration - Multiple large transfers</description>
14
    <mitre>
15
      <id>T1041</id>
16
    </mitre>
17
  </rule>
18

19
  <!-- Correlation: Large transfer to suspicious destination -->
20
  <rule id="200011" level="14">
21
    <if_matched_sid>200001</if_matched_sid>
22
    <list field="dstip" lookup="match_key">etc/lists/suspicious_ips</list>
23
    <description>CRITICAL: Large transfer to known malicious IP</description>
24
  </rule>
25
</group>

1
<rule id="300001" level="10">
2
  <if_matched_sid>100001</if_matched_sid>
3
  <time>21:00-06:00</time>
4
  <weekday>saturday,sunday</weekday>
5
  <description>Suspicious: Login during non-business hours</description>
6
</rule>

1
<rule id="400001" level="12" timeframe="60">
2
  <if_matched_group>web_attack</if_matched_group>
3
  <if_matched_group>ids_alert</if_matched_group>
4
  <same_srcip />
5
  <description>Coordinated attack detected across multiple systems</description>
6
</rule>

1
<rule id="500001" level="9" frequency="10" timeframe="600">
2
  <if_matched_sid>5402</if_matched_sid> <!-- User added to group -->
3
  <same_user />
4
  <description>Unusual activity: Multiple privilege changes for single user</description>
5
</rule>

1
#!/bin/bash
2
# Brute force simulation script
3

4
TARGET_HOST="192.168.1.100"
5
TARGET_USER="testuser"
6

7
# Generate failed login attempts
8
for i in {1..10}; do
9
    ssh -o BatchMode=yes $TARGET_USER@$TARGET_HOST 2>/dev/null
10
    sleep 2
11
done
12

13
# Successful login after failures
14
sshpass -p 'correct_password' ssh $TARGET_USER@$TARGET_HOST "echo 'Login successful'"

1
import socket
2
import time
3
from datetime import datetime
4

5
def inject_test_logs(siem_host, siem_port):
6
    """Inject test logs to validate correlation rules"""
7

8
    test_scenarios = [
9
        # Brute force pattern
10
        {"event": "Failed login", "count": 5, "interval": 1},
11
        {"event": "Successful login", "count": 1, "interval": 0},
12

13
        # Data exfiltration pattern
14
        {"event": "Large transfer", "count": 3, "interval": 300},
15
    ]
16

17
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
18

19
    for scenario in test_scenarios:
20
        for i in range(scenario["count"]):
21
            timestamp = datetime.now().strftime("%b %d %H:%M:%S")
22
            log_entry = f"{timestamp} test-host {scenario['event']} from 10.0.0.{i}"
23

24
            sock.sendto(log_entry.encode(), (siem_host, siem_port))
25
            time.sleep(scenario["interval"])
26

27
    sock.close()
28

29
# Execute test
30
inject_test_logs("192.168.1.100", 514)

1
<integration>
2
  <name>slack</name>
3
  <hook_url>https://hooks.slack.com/services/YOUR/WEBHOOK/URL</hook_url>
4
  <level>12</level>
5
  <alert_format>json</alert_format>
6
  <options>
7
    {
8
      "pretext": "🚨 Critical Security Alert",
9
      "color": "danger",
10
      "fields": [
11
        {"title": "Rule", "value": "_rule.description_"},
12
        {"title": "Level", "value": "_rule.level_"},
13
        {"title": "Source IP", "value": "_srcip_"},
14
        {"title": "User", "value": "_dstuser_"}
15
      ]
16
    }
17
  </options>
18
</integration>

1
#!/usr/bin/env python3
2
# Automated response to brute force attacks
3

4
import json
5
import subprocess
6
from datetime import datetime, timedelta
7

8
def block_attacker(alert_data):
9
    """Automatically block IPs after brute force detection"""
10

11
    alert = json.loads(alert_data)
12

13
    if alert['rule']['id'] == '100010':  # Brute force detection
14
        source_ip = alert['data']['srcip']
15

16
        # Add to firewall blocklist
17
        subprocess.run([
18
            'iptables', '-A', 'INPUT',
19
            '-s', source_ip,
20
            '-j', 'DROP'
21
        ])
22

23
        # Set expiration (24 hours)
24
        expiry = datetime.now() + timedelta(hours=24)
25

26
        # Log the action
27
        with open('/var/log/auto_blocks.log', 'a') as f:
28
            f.write(f"{datetime.now()}: Blocked {source_ip} until {expiry}\n")
29

30
        return True
31

32
    return False

1
<group name="gdpr_compliance,">
2
  <!-- Personal data access monitoring -->
3
  <rule id="700001" level="8">
4
    <decoded_as>database_query</decoded_as>
5
    <match>SELECT .* FROM users</match>
6
    <description>Personal data accessed - GDPR Article 15</description>
7
  </rule>
8

9
  <!-- Data deletion requests -->
10
  <rule id="700002" level="6">
11
    <decoded_as>database_query</decoded_as>
12
    <match>DELETE FROM users WHERE</match>
13
    <description>Right to erasure executed - GDPR Article 17</description>
14
  </rule>
15

16
  <!-- Consent tracking -->
17
  <rule id="700003" level="5">
18
    <program_name>webapp</program_name>
19
    <match>consent_updated</match>
20
    <description>User consent modified - GDPR Article 7</description>
21
  </rule>
22
</group>

1
<group name="pci_dss,">
2
  <!-- Requirement 8.1.6: Lock after failed attempts -->
3
  <rule id="800001" level="10" frequency="6" timeframe="900">
4
    <if_matched_sid>5503</if_matched_sid>
5
    <same_user />
6
    <description>PCI DSS 8.1.6: Account lockout threshold reached</description>
7
    <pci_dss>8.1.6</pci_dss>
8
  </rule>
9

10
  <!-- Requirement 10.2.2: Admin actions -->
11
  <rule id="800002" level="8">
12
    <if_matched_group>sudo</if_matched_group>
13
    <description>PCI DSS 10.2.2: Administrative action logged</description>
14
    <pci_dss>10.2.2</pci_dss>
15
  </rule>
16
</group>