Integrating Wazuh with Fluentd for Unified Logging and Big Data Analytics#

1
flowchart LR
2
    subgraph "Wazuh Server"
3
        A1[Alerts JSON] --> S1[Socket<br/>fluent.sock]
4
        S1 --> FF[Fluentd<br/>Forwarder]
5
    end
6

7
    subgraph "Fluentd Server"
8
        FF -->|TCP/24224| FS[Fluentd<br/>Receiver]
9
        FS --> P1[Parser]
10
        P1 --> R1[Router]
11
    end
12

13
    subgraph "Data Destinations"
14
        R1 --> H1[Hadoop HDFS]
15
        R1 --> E1[Elasticsearch]
16
        R1 --> S3[AWS S3]
17
        R1 --> K1[Kafka]
18
    end
19

20
    style FF fill:#51cf66
21
    style FS fill:#4dabf7
22
    style H1 fill:#ffd43b

1
<match wazuh.**>
2
  @type copy
3

4
  # Hadoop for long-term storage
5
  <store>
6
    @type webhdfs
7
    host hadoop-server
8
    port 9870
9
    path "/security/wazuh/%Y/%m/%d/#{Socket.gethostname}.json"
10
    <buffer time>
11
      timekey 1h
12
      timekey_wait 10m
13
    </buffer>
14
  </store>
15

16
  # Elasticsearch for real-time search
17
  <store>
18
    @type elasticsearch
19
    host elasticsearch-server
20
    port 9200
21
    index_name wazuh-alerts-%Y.%m
22
    type_name _doc
23
  </store>
24

25
  # S3 for backup
26
  <store>
27
    @type s3
28
    aws_key_id YOUR_AWS_KEY
29
    aws_sec_key YOUR_AWS_SECRET
30
    s3_bucket wazuh-backup
31
    s3_region us-east-1
32
    path logs/wazuh/%Y/%m/%d/
33
    <buffer time>
34
      timekey 3600
35
      timekey_wait 10m
36
    </buffer>
37
  </store>
38
</match>

1
# Filter high-severity alerts
2
<filter wazuh.**>
3
  @type grep
4
  <regexp>
5
    key rule.level
6
    pattern /^(9|10|11|12|13|14|15)$/
7
  </regexp>
8
</filter>
9

10
# Add metadata
11
<filter wazuh.**>
12
  @type record_transformer
13
  <record>
14
    fluentd_hostname "#{Socket.gethostname}"
15
    fluentd_timestamp ${time}
16
    environment "production"
17
  </record>
18
</filter>
19

20
# Parse and enrich
21
<filter wazuh.**>
22
  @type parser
23
  key_name full_log
24
  <parse>
25
    @type regexp
26
    expression /^(?<time>[^ ]+) (?<host>[^ ]+) (?<process>[^:]+): (?<message>.*)$/
27
  </parse>
28
</filter>

1
# Optimize buffer settings
2
<match wazuh.**>
3
  @type webhdfs
4
  host hadoop-server
5
  port 9870
6
  path "/wazuh/optimized/%Y%m%d/alerts.json"
7

8
  <buffer time,tag>
9
    @type file
10
    path /var/log/fluent/buffer/wazuh
11
    timekey 300  # 5 minutes
12
    timekey_wait 60
13
    chunk_limit_size 256m
14
    total_limit_size 2g
15
    overflow_action drop_oldest_chunk
16
    compress gzip
17
  </buffer>
18

19
  <format>
20
    @type json
21
  </format>
22

23
  # Retry settings
24
  retry_forever false
25
  retry_max_times 3
26
  retry_wait 10s
27
  retry_exponential_backoff_base 2
28
</match>

1
# PySpark analysis example
2
from pyspark.sql import SparkSession
3

4
spark = SparkSession.builder \
5
    .appName("WazuhSecurityAnalytics") \
6
    .getOrCreate()
7

8
# Read Wazuh alerts from HDFS
9
alerts_df = spark.read.json("hdfs://localhost:9000/Wazuh/*/alerts.json")
10

11
# Analyze top threats
12
threats = alerts_df.groupBy("rule.description") \
13
    .count() \
14
    .orderBy("count", ascending=False) \
15
    .limit(10)
16

17
threats.show()

1
# Anomaly detection on authentication patterns
2
from pyspark.ml.feature import VectorAssembler
3
from pyspark.ml.clustering import KMeans
4

5
# Feature engineering
6
features = alerts_df.filter(alerts_df.rule.groups.contains("authentication")) \
7
    .select("hour", "agent.id", "data.srcip")
8

9
# ML pipeline
10
assembler = VectorAssembler(inputCols=["hour", "agent_id_encoded"],
11
                           outputCol="features")
12
kmeans = KMeans(k=5, seed=1)
13

14
# Detect anomalous authentication patterns
15
model = kmeans.fit(assembled_data)
16
predictions = model.transform(assembled_data)

1
-- Hive query for compliance reports
2
CREATE EXTERNAL TABLE wazuh_alerts (
3
    timestamp STRING,
4
    rule STRUCT<
5
        level: INT,
6
        description: STRING,
7
        pci_dss: ARRAY<STRING>,
8
        gdpr: ARRAY<STRING>
9
    >,
10
    agent STRUCT<
11
        id: STRING,
12
        name: STRING
13
    >
14
)
15
STORED AS JSON
16
LOCATION '/Wazuh/';
17

18
-- PCI DSS compliance query
19
SELECT
20
    rule.pci_dss[0] as requirement,
21
    COUNT(*) as violation_count,
22
    collect_list(DISTINCT agent.name) as affected_systems
23
FROM wazuh_alerts
24
WHERE size(rule.pci_dss) > 0
25
GROUP BY rule.pci_dss[0]
26
ORDER BY violation_count DESC;

1
# Prevent data loss during outages
2
<match wazuh.**>
3
  @type forward
4
  <server>
5
    name primary
6
    host primary-fluentd
7
    port 24224
8
  </server>
9
  <server>
10
    name secondary
11
    host secondary-fluentd
12
    port 24224
13
    standby
14
  </server>
15

16
  <buffer>
17
    @type file
18
    path /var/log/fluent/buffer/
19
    flush_mode interval
20
    flush_interval 10s
21
    retry_type exponential_backoff
22
    retry_forever true
23
  </buffer>
24
</match>

1
# Enable TLS and authentication
2
<source>
3
  @type forward
4
  port 24224
5
  bind 0.0.0.0
6

7
  <transport tls>
8
    cert_path /etc/fluent/certs/server.crt
9
    private_key_path /etc/fluent/certs/server.key
10
    client_cert_auth true
11
    ca_path /etc/fluent/certs/ca.crt
12
  </transport>
13

14
  <security>
15
    self_hostname fluentd-server
16
    shared_key YOUR_SHARED_KEY
17
  </security>
18
</source>

1
# Monitor Fluentd health
2
<source>
3
  @type monitor_agent
4
  bind 127.0.0.1
5
  port 24220
6
</source>
7

8
# Prometheus metrics
9
<source>
10
  @type prometheus
11
  bind 0.0.0.0
12
  port 24231
13
  metrics_path /metrics
14
</source>
15

16
# Alert on errors
17
<match fluent.**>
18
  @type mail
19
  host smtp.example.com
20
  port 587
21
  from fluentd@example.com
22
  to ops@example.com
23
  subject "Fluentd Error Alert"
24
  <filter>
25
    @type grep
26
    <regexp>
27
      key message
28
      pattern /error|Error|ERROR/
29
    </regexp>
30
  </filter>
31
</match>

1
<match wazuh.**>
2
  @type kafka2
3
  brokers kafka1:9092,kafka2:9092,kafka3:9092
4
  default_topic wazuh-alerts
5

6
  <format>
7
    @type json
8
  </format>
9

10
  <buffer>
11
    @type file
12
    path /var/log/fluent/buffer/kafka
13
    flush_interval 3s
14
  </buffer>
15

16
  # Partition by agent ID
17
  partition_key_key agent.id
18
  max_send_retries 3
19
  required_acks -1
20
</match>

1
<match wazuh.**>
2
  @type splunk_hec
3
  hec_host splunk.example.com
4
  hec_port 8088
5
  hec_token YOUR_HEC_TOKEN
6

7
  source wazuh
8
  sourcetype _json
9

10
  <format>
11
    @type json
12
  </format>
13
</match>