Wazuh to ClickHouse Integration: Optimizing Storage and Analytics at Scale#

1
#!/bin/bash
2
# Kafka Cluster Setup for Wazuh Integration
3

4
# Install Kafka
5
wget https://downloads.apache.org/kafka/2.8.2/kafka_2.13-2.8.2.tgz
6
tar -xzf kafka_2.13-2.8.2.tgz
7
sudo mv kafka_2.13-2.8.2 /opt/kafka
8

9
# Configure Kafka server
10
cat > /opt/kafka/config/server.properties << 'EOF'
11
broker.id=1
12
listeners=PLAINTEXT://0.0.0.0:9092
13
advertised.listeners=PLAINTEXT://kafka-server:9092
14
num.network.threads=8
15
num.io.threads=16
16
socket.send.buffer.bytes=102400
17
socket.receive.buffer.bytes=102400
18
socket.request.max.bytes=104857600
19

20
log.dirs=/opt/kafka/kafka-logs
21
num.partitions=6
22
num.recovery.threads.per.data.dir=2
23
offsets.topic.replication.factor=3
24
transaction.state.log.replication.factor=3
25
transaction.state.log.min.isr=2
26

27
log.retention.hours=168
28
log.retention.bytes=1073741824
29
log.segment.bytes=1073741824
30
log.retention.check.interval.ms=300000
31

32
zookeeper.connect=localhost:2181
33
zookeeper.connection.timeout.ms=18000
34

35
group.initial.rebalance.delay.ms=0
36

37
# Performance optimizations for Wazuh logs
38
replica.fetch.max.bytes=10485760
39
message.max.bytes=10485760
40
compression.type=lz4
41
EOF
42

43
# Start Kafka
44
sudo systemctl start kafka
45
sudo systemctl enable kafka
46

47
# Create Wazuh topics
48
/opt/kafka/bin/kafka-topics.sh --create --topic wazuh-alerts --bootstrap-server localhost:9092 --partitions 6 --replication-factor 3
49
/opt/kafka/bin/kafka-topics.sh --create --topic wazuh-archives --bootstrap-server localhost:9092 --partitions 6 --replication-factor 3

1
-- Main alerts table with optimal compression and indexing
2
CREATE TABLE wazuh_logs.alerts_local ON CLUSTER wazuh_cluster
3
(
4
    -- Temporal columns (essential for partitioning)
5
    timestamp DateTime64(3, 'UTC') CODEC(DoubleDelta, ZSTD(3)),
6
    date Date MATERIALIZED toDate(timestamp),
7

8
    -- Agent identification
9
    agent_id LowCardinality(String) CODEC(ZSTD(3)),
10
    agent_name LowCardinality(String) CODEC(ZSTD(3)),
11
    agent_ip IPv4 CODEC(ZSTD(3)),
12

13
    -- Rule information
14
    rule_id UInt32 CODEC(ZSTD(3)),
15
    rule_level UInt8 CODEC(ZSTD(3)),
16
    rule_description String CODEC(ZSTD(3)),
17
    rule_group Array(LowCardinality(String)) CODEC(ZSTD(3)),
18
    rule_mitre Array(LowCardinality(String)) CODEC(ZSTD(3)),
19

20
    -- Location and source
21
    location String CODEC(ZSTD(3)),
22
    decoder_name LowCardinality(String) CODEC(ZSTD(3)),
23

24
    -- Event data
25
    full_log String CODEC(ZSTD(3)),
26
    predecoder_program_name LowCardinality(String) CODEC(ZSTD(3)),
27
    predecoder_hostname LowCardinality(String) CODEC(ZSTD(3)),
28

29
    -- JSON data (flattened for performance)
30
    data_srcip IPv4 CODEC(ZSTD(3)),
31
    data_srcport UInt16 CODEC(ZSTD(3)),
32
    data_dstip IPv4 CODEC(ZSTD(3)),
33
    data_dstport UInt16 CODEC(ZSTD(3)),
34
    data_protocol LowCardinality(String) CODEC(ZSTD(3)),
35
    data_action LowCardinality(String) CODEC(ZSTD(3)),
36
    data_status LowCardinality(String) CODEC(ZSTD(3)),
37

38
    -- Windows-specific fields
39
    data_win_eventdata Nested(
40
        user String,
41
        domain String,
42
        logonType UInt8,
43
        processName String,
44
        commandLine String,
45
        parentProcessName String,
46
        targetUserName String,
47
        workstationName String,
48
        ipAddress IPv4
49
    ) CODEC(ZSTD(3)),
50

51
    -- File integrity monitoring
52
    syscheck_path String CODEC(ZSTD(3)),
53
    syscheck_event LowCardinality(String) CODEC(ZSTD(3)),
54
    syscheck_sha256_after String CODEC(ZSTD(3)),
55
    syscheck_sha256_before String CODEC(ZSTD(3)),
56

57
    -- Vulnerability data
58
    vulnerability Nested(
59
        cve String,
60
        severity String,
61
        cvss2_score Float32,
62
        cvss3_score Float32,
63
        title String
64
    ) CODEC(ZSTD(3)),
65

66
    -- GeoIP data
67
    geoip_src_country_name LowCardinality(String) CODEC(ZSTD(3)),
68
    geoip_src_country_code LowCardinality(String) CODEC(ZSTD(3)),
69
    geoip_src_city_name String CODEC(ZSTD(3)),
70

71
    -- Original JSON for complex queries
72
    json_raw String CODEC(ZSTD(3))
73
)
74
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/wazuh_logs/alerts', '{replica}')
75
PARTITION BY toYYYYMM(timestamp)
76
ORDER BY (timestamp, agent_id, rule_level, rule_id)
77
TTL timestamp + INTERVAL 2 YEAR DELETE
78
SETTINGS
79
    index_granularity = 8192,
80
    storage_policy = 'tiered',
81
    merge_with_ttl_timeout = 86400;
82

83
-- Distributed table for cluster queries
84
CREATE TABLE wazuh_logs.alerts ON CLUSTER wazuh_cluster AS wazuh_logs.alerts_local
85
ENGINE = Distributed(wazuh_cluster, wazuh_logs, alerts_local, rand());
86

87
-- Materialized view for real-time aggregations
88
CREATE MATERIALIZED VIEW wazuh_logs.alerts_hourly_stats
89
ENGINE = SummingMergeTree()
90
PARTITION BY toYYYYMM(hour)
91
ORDER BY (hour, agent_id, rule_level)
92
AS SELECT
93
    toStartOfHour(timestamp) AS hour,
94
    agent_id,
95
    agent_name,
96
    rule_level,
97
    count() AS alert_count,
98
    uniq(rule_id) AS unique_rules,
99
    uniq(data_srcip) AS unique_src_ips,
100
    avg(rule_level) AS avg_severity
101
FROM wazuh_logs.alerts_local
102
GROUP BY hour, agent_id, agent_name, rule_level;
103

104
-- High-frequency events table (different compression strategy)
105
CREATE TABLE wazuh_logs.archives_local ON CLUSTER wazuh_cluster
106
(
107
    timestamp DateTime64(3, 'UTC') CODEC(DoubleDelta, LZ4),
108
    agent_id LowCardinality(String) CODEC(LZ4),
109
    agent_name LowCardinality(String) CODEC(LZ4),
110
    location String CODEC(LZ4),
111
    full_log String CODEC(ZSTD(1)),
112
    json_raw String CODEC(ZSTD(1))
113
)
114
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/wazuh_logs/archives', '{replica}')
115
PARTITION BY toYYYYMM(timestamp)
116
ORDER BY (timestamp, agent_id)
117
TTL timestamp + INTERVAL 6 MONTH DELETE
118
SETTINGS
119
    index_granularity = 8192,
120
    storage_policy = 'tiered';
121

122
-- Specialized table for security analytics
123
CREATE TABLE wazuh_logs.security_events ON CLUSTER wazuh_cluster
124
(
125
    timestamp DateTime64(3, 'UTC'),
126
    agent_id LowCardinality(String),
127
    rule_id UInt32,
128
    rule_level UInt8,
129
    mitre_technique Array(LowCardinality(String)),
130
    src_ip IPv4,
131
    dst_ip IPv4,
132
    user_name String,
133
    process_name String,
134
    command_line String,
135
    file_path String,
136
    hash_sha256 String,
137

138
    -- Pre-computed threat indicators
139
    is_privilege_escalation UInt8,
140
    is_lateral_movement UInt8,
141
    is_persistence UInt8,
142
    is_defense_evasion UInt8,
143
    threat_score Float32
144
)
145
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/wazuh_logs/security_events', '{replica}')
146
PARTITION BY toYYYYMM(timestamp)
147
ORDER BY (timestamp, threat_score, rule_level)
148
SETTINGS index_granularity = 8192;

1
-- Create secondary indexes for common queries
2
ALTER TABLE wazuh_logs.alerts_local ON CLUSTER wazuh_cluster
3
ADD INDEX idx_rule_id rule_id TYPE minmax GRANULARITY 4;
4

5
ALTER TABLE wazuh_logs.alerts_local ON CLUSTER wazuh_cluster
6
ADD INDEX idx_src_ip data_srcip TYPE minmax GRANULARITY 4;
7

8
ALTER TABLE wazuh_logs.alerts_local ON CLUSTER wazuh_cluster
9
ADD INDEX idx_agent_name agent_name TYPE set(100) GRANULARITY 8;
10

11
-- Bloom filter for high cardinality strings
12
ALTER TABLE wazuh_logs.alerts_local ON CLUSTER wazuh_cluster
13
ADD INDEX idx_full_log_bloom full_log TYPE bloom_filter(0.01) GRANULARITY 4;

1
#!/usr/bin/env python3
2
"""
3
Wazuh to ClickHouse Integration Service
4
High-performance log processor with Kafka integration
5
"""
6

7
import json
8
import logging
9
import threading
10
import time
11
from datetime import datetime
12
from typing import Dict, List, Optional, Any
13
from dataclasses import dataclass
14
from queue import Queue
15
import hashlib
16
import ipaddress
17

18
from kafka import KafkaConsumer
19
from clickhouse_driver import Client
20
import yaml
21
import signal
22
import sys
23
from concurrent.futures import ThreadPoolExecutor
24

25
# Configuration
26
@dataclass
27
class Config:
28
    kafka_bootstrap_servers: List[str]
29
    kafka_topics: List[str]
30
    clickhouse_host: str
31
    clickhouse_port: int
32
    clickhouse_database: str
33
    clickhouse_user: str
34
    clickhouse_password: str
35
    batch_size: int
36
    flush_interval: int
37
    worker_threads: int
38
    consumer_group: str
39
    max_retries: int
40

41
class WazuhClickHouseProcessor:
42
    def __init__(self, config: Config):
43
        self.config = config
44
        self.running = True
45
        self.batch_queue = Queue(maxsize=10000)
46
        self.stats = {
47
            'processed': 0,
48
            'errors': 0,
49
            'last_process_time': 0
50
        }
51

52
        # Initialize connections
53
        self.clickhouse_client = self._init_clickhouse()
54
        self.kafka_consumer = self._init_kafka()
55

56
        # Thread pool for parallel processing
57
        self.executor = ThreadPoolExecutor(max_workers=config.worker_threads)
58

59
        # Setup logging
60
        logging.basicConfig(
61
            level=logging.INFO,
62
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
63
        )
64
        self.logger = logging.getLogger(__name__)
65

66
        # Setup signal handlers
67
        signal.signal(signal.SIGTERM, self._signal_handler)
68
        signal.signal(signal.SIGINT, self._signal_handler)
69

70
    def _init_clickhouse(self) -> Client:
71
        """Initialize ClickHouse connection with optimization"""
72

73
        client = Client(
74
            host=self.config.clickhouse_host,
75
            port=self.config.clickhouse_port,
76
            database=self.config.clickhouse_database,
77
            user=self.config.clickhouse_user,
78
            password=self.config.clickhouse_password,
79
            send_receive_timeout=300,
80
            sync_request_timeout=300,
81
            compress_block_size=65536,
82
            compression=True
83
        )
84

85
        # Verify connection and create tables if needed
86
        self._ensure_tables_exist(client)
87

88
        return client
89

90
    def _init_kafka(self) -> KafkaConsumer:
91
        """Initialize Kafka consumer with optimal settings"""
92

93
        return KafkaConsumer(
94
            *self.config.kafka_topics,
95
            bootstrap_servers=self.config.kafka_bootstrap_servers,
96
            group_id=self.config.consumer_group,
97
            auto_offset_reset='latest',
98
            enable_auto_commit=True,
99
            auto_commit_interval_ms=5000,
100
            session_timeout_ms=30000,
101
            max_poll_records=1000,
102
            max_poll_interval_ms=300000,
103
            fetch_min_bytes=1024,
104
            fetch_max_wait_ms=100,
105
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
106
            consumer_timeout_ms=1000
107
        )
108

109
    def _ensure_tables_exist(self, client: Client):
110
        """Ensure required tables exist with proper structure"""
111

112
        # Check if alerts table exists
113
        tables = client.execute("SHOW TABLES FROM wazuh_logs")
114
        table_names = [table[0] for table in tables]
115

116
        if 'alerts_local' not in table_names:
117
            self.logger.info("Creating alerts table...")
118
            # Table creation SQL would be executed here
119
            # (Using the schema defined earlier)
120

121
        if 'archives_local' not in table_names:
122
            self.logger.info("Creating archives table...")
123
            # Archive table creation
124

125
    def _signal_handler(self, signum, frame):
126
        """Handle shutdown signals gracefully"""
127
        self.logger.info(f"Received signal {signum}, shutting down...")
128
        self.running = False
129
        self.kafka_consumer.close()
130
        self.executor.shutdown(wait=True)
131
        sys.exit(0)
132

133
    def _extract_ip_address(self, ip_str: str) -> Optional[str]:
134
        """Extract and validate IP address"""
135
        if not ip_str:
136
            return None
137

138
        try:
139
            # Handle IPv4
140
            ipaddress.ip_address(ip_str)
141
            return ip_str
142
        except ValueError:
143
            # Try to extract IP from string
144
            import re
145
            ip_pattern = r'\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b'
146
            match = re.search(ip_pattern, ip_str)
147
            return match.group() if match else None
148

149
    def _transform_wazuh_alert(self, raw_alert: Dict) -> Dict:
150
        """Transform Wazuh alert to ClickHouse-optimized format"""
151

152
        try:
153
            # Extract timestamp
154
            timestamp = raw_alert.get('timestamp', datetime.utcnow().isoformat())
155
            if isinstance(timestamp, str):
156
                # Parse ISO format timestamp
157
                timestamp = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
158

159
            # Extract agent information
160
            agent = raw_alert.get('agent', {})
161
            agent_id = agent.get('id', 'unknown')
162
            agent_name = agent.get('name', 'unknown')
163
            agent_ip = self._extract_ip_address(agent.get('ip', ''))
164

165
            # Extract rule information
166
            rule = raw_alert.get('rule', {})
167
            rule_id = rule.get('id', 0)
168
            rule_level = rule.get('level', 0)
169
            rule_description = rule.get('description', '')
170
            rule_groups = rule.get('groups', [])
171
            rule_mitre = rule.get('mitre', [])
172

173
            # Extract data section
174
            data = raw_alert.get('data', {})
175

176
            # Windows event data
177
            win_data = data.get('win', {}).get('eventdata', {})
178

179
            # System check data
180
            syscheck = data.get('syscheck', {})
181

182
            # Vulnerability data
183
            vulnerability = data.get('vulnerability', {})
184

185
            # Build optimized record
186
            record = {
187
                'timestamp': timestamp,
188
                'agent_id': agent_id,
189
                'agent_name': agent_name,
190
                'agent_ip': agent_ip,
191
                'rule_id': rule_id,
192
                'rule_level': rule_level,
193
                'rule_description': rule_description,
194
                'rule_group': rule_groups,
195
                'rule_mitre': rule_mitre,
196
                'location': raw_alert.get('location', ''),
197
                'decoder_name': raw_alert.get('decoder', {}).get('name', ''),
198
                'full_log': raw_alert.get('full_log', ''),
199
                'predecoder_program_name': raw_alert.get('predecoder', {}).get('program_name', ''),
200
                'predecoder_hostname': raw_alert.get('predecoder', {}).get('hostname', ''),
201

202
                # Network data
203
                'data_srcip': self._extract_ip_address(data.get('srcip', '')),
204
                'data_srcport': data.get('srcport', 0),
205
                'data_dstip': self._extract_ip_address(data.get('dstip', '')),
206
                'data_dstport': data.get('dstport', 0),
207
                'data_protocol': data.get('protocol', ''),
208
                'data_action': data.get('action', ''),
209
                'data_status': data.get('status', ''),
210

211
                # Windows nested data
212
                'data_win_eventdata.user': [win_data.get('user', '')],
213
                'data_win_eventdata.domain': [win_data.get('domain', '')],
214
                'data_win_eventdata.logonType': [win_data.get('logonType', 0)],
215
                'data_win_eventdata.processName': [win_data.get('processName', '')],
216
                'data_win_eventdata.commandLine': [win_data.get('commandLine', '')],
217
                'data_win_eventdata.parentProcessName': [win_data.get('parentProcessName', '')],
218
                'data_win_eventdata.targetUserName': [win_data.get('targetUserName', '')],
219
                'data_win_eventdata.workstationName': [win_data.get('workstationName', '')],
220
                'data_win_eventdata.ipAddress': [self._extract_ip_address(win_data.get('ipAddress', ''))],
221

222
                # File integrity monitoring
223
                'syscheck_path': syscheck.get('path', ''),
224
                'syscheck_event': syscheck.get('event', ''),
225
                'syscheck_sha256_after': syscheck.get('sha256_after', ''),
226
                'syscheck_sha256_before': syscheck.get('sha256_before', ''),
227

228
                # Vulnerability nested data
229
                'vulnerability.cve': [vulnerability.get('cve', '')],
230
                'vulnerability.severity': [vulnerability.get('severity', '')],
231
                'vulnerability.cvss2_score': [vulnerability.get('cvss2_score', 0.0)],
232
                'vulnerability.cvss3_score': [vulnerability.get('cvss3_score', 0.0)],
233
                'vulnerability.title': [vulnerability.get('title', '')],
234

235
                # GeoIP data (if available)
236
                'geoip_src_country_name': data.get('geoip', {}).get('country_name', ''),
237
                'geoip_src_country_code': data.get('geoip', {}).get('country_code', ''),
238
                'geoip_src_city_name': data.get('geoip', {}).get('city_name', ''),
239

240
                # Raw JSON for complex queries
241
                'json_raw': json.dumps(raw_alert)
242
            }
243

244
            # Clean null values and convert types
245
            cleaned_record = self._clean_record(record)
246

247
            return cleaned_record
248

249
        except Exception as e:
250
            self.logger.error(f"Error transforming alert: {e}")
251
            self.stats['errors'] += 1
252
            return None
253

254
    def _clean_record(self, record: Dict) -> Dict:
255
        """Clean and validate record data"""
256

257
        cleaned = {}
258

259
        for key, value in record.items():
260
            if value is None or value == '' or value == []:
261
                # Set appropriate defaults
262
                if 'ip' in key.lower():
263
                    cleaned[key] = '0.0.0.0'
264
                elif 'port' in key.lower() or key.endswith('_id') or 'level' in key:
265
                    cleaned[key] = 0
266
                elif isinstance(value, list):
267
                    cleaned[key] = ['']
268
                else:
269
                    cleaned[key] = ''
270
            else:
271
                cleaned[key] = value
272

273
        return cleaned
274

275
    def _batch_insert_alerts(self, records: List[Dict]):
276
        """Insert batch of alerts into ClickHouse"""
277

278
        if not records:
279
            return
280

281
        try:
282
            start_time = time.time()
283

284
            self.clickhouse_client.execute(
285
                'INSERT INTO wazuh_logs.alerts_local VALUES',
286
                records,
287
                types_check=True
288
            )
289

290
            processing_time = time.time() - start_time
291
            self.stats['processed'] += len(records)
292
            self.stats['last_process_time'] = processing_time
293

294
            self.logger.info(f"Inserted {len(records)} records in {processing_time:.2f}s")
295

296
        except Exception as e:
297
            self.logger.error(f"Error inserting batch: {e}")
298
            self.stats['errors'] += len(records)
299

300
            # Retry logic
301
            for retry in range(self.config.max_retries):
302
                try:
303
                    time.sleep(2 ** retry)  # Exponential backoff
304
                    self.clickhouse_client.execute(
305
                        'INSERT INTO wazuh_logs.alerts_local VALUES',
306
                        records
307
                    )
308
                    self.logger.info(f"Retry {retry + 1} successful")
309
                    break
310
                except Exception as retry_error:
311
                    self.logger.error(f"Retry {retry + 1} failed: {retry_error}")
312

313
    def _process_batch(self, batch: List[Dict]):
314
        """Process a batch of messages"""
315

316
        transformed_records = []
317

318
        for message in batch:
319
            try:
320
                # Transform message
321
                record = self._transform_wazuh_alert(message)
322
                if record:
323
                    transformed_records.append(record)
324

325
            except Exception as e:
326
                self.logger.error(f"Error processing message: {e}")
327
                self.stats['errors'] += 1
328

329
        # Insert batch
330
        if transformed_records:
331
            self._batch_insert_alerts(transformed_records)
332

333
    def _consumer_worker(self):
334
        """Kafka consumer worker thread"""
335

336
        batch = []
337
        last_flush = time.time()
338

339
        for message in self.kafka_consumer:
340
            if not self.running:
341
                break
342

343
            try:
344
                # Add to batch
345
                batch.append(message.value)
346

347
                # Check if batch is ready
348
                current_time = time.time()
349
                batch_ready = (
350
                    len(batch) >= self.config.batch_size or
351
                    (current_time - last_flush) >= self.config.flush_interval
352
                )
353

354
                if batch_ready:
355
                    # Submit batch for processing
356
                    self.executor.submit(self._process_batch, batch.copy())
357
                    batch.clear()
358
                    last_flush = current_time
359

360
            except Exception as e:
361
                self.logger.error(f"Consumer error: {e}")
362
                self.stats['errors'] += 1
363

364
        # Process remaining batch
365
        if batch:
366
            self.executor.submit(self._process_batch, batch)
367

368
    def _stats_reporter(self):
369
        """Report processing statistics"""
370

371
        while self.running:
372
            time.sleep(60)  # Report every minute
373

374
            self.logger.info(
375
                f"Stats - Processed: {self.stats['processed']}, "
376
                f"Errors: {self.stats['errors']}, "
377
                f"Last batch time: {self.stats['last_process_time']:.2f}s"
378
            )
379

380
    def run(self):
381
        """Main processing loop"""
382

383
        self.logger.info("Starting Wazuh to ClickHouse processor...")
384

385
        # Start stats reporter
386
        stats_thread = threading.Thread(target=self._stats_reporter, daemon=True)
387
        stats_thread.start()
388

389
        # Start consumer
390
        try:
391
            self._consumer_worker()
392
        except KeyboardInterrupt:
393
            self.logger.info("Received interrupt signal")
394
        finally:
395
            self.running = False
396
            self.kafka_consumer.close()
397
            self.executor.shutdown(wait=True)
398
            self.logger.info("Processor shutdown complete")
399

400
# Configuration loading
401
def load_config(config_file: str) -> Config:
402
    """Load configuration from YAML file"""
403

404
    with open(config_file, 'r') as f:
405
        config_data = yaml.safe_load(f)
406

407
    return Config(
408
        kafka_bootstrap_servers=config_data['kafka']['bootstrap_servers'],
409
        kafka_topics=config_data['kafka']['topics'],
410
        clickhouse_host=config_data['clickhouse']['host'],
411
        clickhouse_port=config_data['clickhouse']['port'],
412
        clickhouse_database=config_data['clickhouse']['database'],
413
        clickhouse_user=config_data['clickhouse']['user'],
414
        clickhouse_password=config_data['clickhouse']['password'],
415
        batch_size=config_data.get('processing', {}).get('batch_size', 1000),
416
        flush_interval=config_data.get('processing', {}).get('flush_interval', 5),
417
        worker_threads=config_data.get('processing', {}).get('worker_threads', 4),
418
        consumer_group=config_data.get('processing', {}).get('consumer_group', 'wazuh-clickhouse'),
419
        max_retries=config_data.get('processing', {}).get('max_retries', 3)
420
    )
421

422
if __name__ == "__main__":
423
    # Load configuration
424
    config = load_config('/etc/wazuh-clickhouse/config.yaml')
425

426
    # Create and run processor
427
    processor = WazuhClickHouseProcessor(config)
428
    processor.run()

1
-- Top 10 Most Frequent Attacks by MITRE Technique
2
SELECT
3
    arrayJoin(rule_mitre) AS mitre_technique,
4
    count() AS attack_count,
5
    uniq(agent_id) AS affected_agents,
6
    uniq(data_srcip) AS unique_sources,
7
    max(rule_level) AS max_severity,
8
    toStartOfHour(max(timestamp)) AS last_seen
9
FROM wazuh_logs.alerts
10
WHERE timestamp >= now() - INTERVAL 24 HOUR
11
    AND rule_level >= 5
12
    AND mitre_technique != ''
13
GROUP BY mitre_technique
14
ORDER BY attack_count DESC
15
LIMIT 10;
16

17
-- Lateral Movement Detection
18
SELECT
19
    data_win_eventdata.user[1] AS user,
20
    count(DISTINCT agent_name) AS hosts_accessed,
21
    count() AS total_logons,
22
    uniq(data_srcip) AS source_ips,
23
    min(timestamp) AS first_seen,
24
    max(timestamp) AS last_seen,
25
    arrayDistinct(groupArray(agent_name)) AS affected_hosts
26
FROM wazuh_logs.alerts
27
WHERE timestamp >= now() - INTERVAL 4 HOUR
28
    AND rule_id IN (60106, 60108, 60122)  -- Logon events
29
    AND data_win_eventdata.user[1] != ''
30
GROUP BY user
31
HAVING hosts_accessed >= 3
32
    AND total_logons >= 5
33
ORDER BY hosts_accessed DESC, total_logons DESC;
34

35
-- Privilege Escalation Timeline
36
WITH privilege_events AS (
37
    SELECT
38
        timestamp,
39
        agent_name,
40
        rule_id,
41
        rule_description,
42
        data_win_eventdata.user[1] AS user,
43
        data_win_eventdata.processName[1] AS process,
44
        syscheck_path
45
    FROM wazuh_logs.alerts
46
    WHERE timestamp >= now() - INTERVAL 24 HOUR
47
        AND (
48
            has(rule_mitre, 'T1068') OR  -- Exploitation for Privilege Escalation
49
            has(rule_mitre, 'T1078') OR  -- Valid Accounts
50
            has(rule_mitre, 'T1134') OR  -- Access Token Manipulation
51
            rule_id IN (92016, 92026, 200001)  -- Custom privilege escalation rules
52
        )
53
)
54
SELECT
55
    agent_name,
56
    user,
57
    count() AS escalation_attempts,
58
    min(timestamp) AS first_attempt,
59
    max(timestamp) AS last_attempt,
60
    arrayDistinct(groupArray(rule_description)) AS attack_types,
61
    arrayDistinct(groupArray(process)) AS processes,
62
    arrayDistinct(groupArray(syscheck_path)) AS modified_files
63
FROM privilege_events
64
GROUP BY agent_name, user
65
ORDER BY escalation_attempts DESC, last_attempt DESC;
66

67
-- Malware Family Analysis by Hash Correlation
68
WITH malware_hashes AS (
69
    SELECT
70
        syscheck_sha256_after AS hash,
71
        count() AS occurrence_count,
72
        uniq(agent_id) AS infected_hosts,
73
        arrayDistinct(groupArray(agent_name)) AS host_names,
74
        min(timestamp) AS first_seen,
75
        max(timestamp) AS last_seen
76
    FROM wazuh_logs.alerts
77
    WHERE timestamp >= now() - INTERVAL 7 DAY
78
        AND syscheck_sha256_after != ''
79
        AND rule_level >= 7
80
    GROUP BY hash
81
    HAVING occurrence_count > 1
82
)
83
SELECT
84
    hash,
85
    occurrence_count,
86
    infected_hosts,
87
    host_names,
88
    first_seen,
89
    last_seen,
90
    dateDiff('hour', first_seen, last_seen) AS spread_duration_hours
91
FROM malware_hashes
92
ORDER BY infected_hosts DESC, occurrence_count DESC;
93

94
-- Network Attack Correlation
95
SELECT
96
    data_srcip AS attacker_ip,
97
    count() AS total_attacks,
98
    uniq(agent_id) AS targets_count,
99
    uniq(rule_id) AS attack_types,
100
    avg(rule_level) AS avg_severity,
101
    arrayDistinct(groupArray(rule_description)) AS attack_descriptions,
102
    arrayDistinct(groupArray(agent_name)) AS target_hosts,
103
    min(timestamp) AS campaign_start,
104
    max(timestamp) AS campaign_end,
105
    geoip_src_country_name AS source_country
106
FROM wazuh_logs.alerts
107
WHERE timestamp >= now() - INTERVAL 24 HOUR
108
    AND data_srcip != '0.0.0.0'
109
    AND rule_level >= 5
110
GROUP BY data_srcip, source_country
111
HAVING total_attacks >= 10
112
    OR targets_count >= 3
113
ORDER BY total_attacks DESC, targets_count DESC
114
LIMIT 20;

1
-- Storage Compression Analysis
2
SELECT
3
    database,
4
    table,
5
    formatReadableSize(sum(bytes_on_disk)) AS compressed_size,
6
    formatReadableSize(sum(data_uncompressed_bytes)) AS uncompressed_size,
7
    round(sum(data_uncompressed_bytes) / sum(bytes_on_disk), 2) AS compression_ratio,
8
    sum(rows) AS total_rows,
9
    sum(primary_key_bytes_in_memory) AS index_memory
10
FROM system.parts
11
WHERE database = 'wazuh_logs'
12
    AND active = 1
13
GROUP BY database, table
14
ORDER BY bytes_on_disk DESC;
15

16
-- Query Performance Analysis
17
SELECT
18
    query_duration_ms,
19
    result_rows,
20
    result_bytes,
21
    read_rows,
22
    read_bytes,
23
    memory_usage,
24
    query_start_time,
25
    substring(query, 1, 100) AS query_preview
26
FROM system.query_log
27
WHERE type = 'QueryFinish'
28
    AND event_time >= now() - INTERVAL 1 HOUR
29
    AND query LIKE '%wazuh_logs%'
30
    AND query_duration_ms > 1000
31
ORDER BY query_duration_ms DESC
32
LIMIT 10;
33

34
-- Table Partitioning Analysis
35
SELECT
36
    partition,
37
    count() AS parts_count,
38
    formatReadableSize(sum(bytes_on_disk)) AS size_on_disk,
39
    sum(rows) AS total_rows,
40
    min(min_date) AS oldest_data,
41
    max(max_date) AS newest_data,
42
    sum(primary_key_bytes_in_memory) AS index_memory
43
FROM system.parts
44
WHERE database = 'wazuh_logs'
45
    AND table = 'alerts_local'
46
    AND active = 1
47
GROUP BY partition
48
ORDER BY partition DESC;