Advanced Threat Intelligence Integration with Wazuh: MISP and VirusTotal#

1
#!/usr/bin/env python3
2
"""
3
MISP-Wazuh Integration Script
4
Automated threat intelligence correlation and enrichment
5
"""
6

7
import sys
8
import os
9
import json
10
import time
11
import hashlib
12
import logging
13
import requests
14
import argparse
15
from datetime import datetime, timedelta
16
from pathlib import Path
17
from typing import Dict, List, Optional, Any
18
import urllib3
19

20
# Disable SSL warnings for self-signed certificates
21
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
22

23
class MISPWazuhIntegrator:
24
    def __init__(self, config_file: str = '/var/ossec/integrations/misp_config.json'):
25
        self.config = self.load_config(config_file)
26
        self.setup_logging()
27

28
        # MISP Configuration
29
        self.misp_url = self.config['misp']['url']
30
        self.misp_key = self.config['misp']['api_key']
31
        self.misp_verifycert = self.config['misp'].get('verify_cert', False)
32

33
        # Wazuh Configuration
34
        self.wazuh_url = self.config['wazuh']['url']
35
        self.wazuh_user = self.config['wazuh']['user']
36
        self.wazuh_password = self.config['wazuh']['password']
37
        self.wazuh_token = None
38

39
        # Integration settings
40
        self.correlation_threshold = self.config.get('correlation_threshold', 70)
41
        self.max_age_days = self.config.get('max_age_days', 30)
42
        self.batch_size = self.config.get('batch_size', 100)
43

44
        # IOC types mapping
45
        self.ioc_types = {
46
            'ip-src': ['srcip', 'src_ip', 'source_ip'],
47
            'ip-dst': ['dstip', 'dst_ip', 'destination_ip'],
48
            'domain': ['domain', 'hostname', 'fqdn'],
49
            'hostname': ['hostname', 'domain', 'fqdn'],
50
            'url': ['url', 'uri'],
51
            'filename': ['filename', 'file', 'name'],
52
            'md5': ['md5', 'hash', 'file_hash'],
53
            'sha1': ['sha1', 'hash', 'file_hash'],
54
            'sha256': ['sha256', 'hash', 'file_hash'],
55
            'email-src': ['email', 'from_email', 'sender'],
56
            'email-dst': ['email', 'to_email', 'recipient'],
57
            'user-agent': ['user_agent', 'useragent'],
58
            'mutex': ['mutex'],
59
            'registry-key': ['registry_key', 'reg_key']
60
        }
61

62
        # Cache for MISP attributes
63
        self.misp_cache = {}
64
        self.cache_ttl = 3600  # 1 hour
65

66
    def load_config(self, config_file: str) -> Dict:
67
        """Load configuration from JSON file"""
68
        try:
69
            with open(config_file, 'r') as f:
70
                return json.load(f)
71
        except FileNotFoundError:
72
            # Create default configuration
73
            default_config = {
74
                "misp": {
75
                    "url": "https://misp.yourdomain.com",
76
                    "api_key": "YOUR_MISP_API_KEY",
77
                    "verify_cert": False
78
                },
79
                "wazuh": {
80
                    "url": "https://wazuh-manager:55000",
81
                    "user": "wazuh",
82
                    "password": "wazuh"
83
                },
84
                "correlation_threshold": 70,
85
                "max_age_days": 30,
86
                "batch_size": 100,
87
                "log_level": "INFO"
88
            }
89

90
            # Create directory if it doesn't exist
91
            os.makedirs(os.path.dirname(config_file), exist_ok=True)
92

93
            with open(config_file, 'w') as f:
94
                json.dump(default_config, f, indent=2)
95

96
            print(f"Created default configuration at {config_file}")
97
            print("Please update the configuration with your MISP and Wazuh details")
98
            sys.exit(1)
99

100
    def setup_logging(self):
101
        """Setup logging configuration"""
102
        log_level = getattr(logging, self.config.get('log_level', 'INFO'))
103
        log_file = self.config.get('log_file', '/var/ossec/logs/misp_integration.log')
104

105
        # Create log directory if it doesn't exist
106
        os.makedirs(os.path.dirname(log_file), exist_ok=True)
107

108
        logging.basicConfig(
109
            level=log_level,
110
            format='%(asctime)s - %(levelname)s - %(message)s',
111
            handlers=[
112
                logging.FileHandler(log_file),
113
                logging.StreamHandler()
114
            ]
115
        )
116
        self.logger = logging.getLogger(__name__)
117

118
    def get_wazuh_token(self) -> Optional[str]:
119
        """Authenticate with Wazuh API and get token"""
120
        try:
121
            auth_url = f"{self.wazuh_url}/security/user/authenticate"
122

123
            response = requests.post(
124
                auth_url,
125
                json={'user': self.wazuh_user, 'password': self.wazuh_password},
126
                headers={'Content-Type': 'application/json'},
127
                verify=False,
128
                timeout=10
129
            )
130

131
            if response.status_code == 200:
132
                token = response.json()['data']['token']
133
                self.wazuh_token = token
134
                self.logger.info("Successfully authenticated with Wazuh")
135
                return token
136
            else:
137
                self.logger.error(f"Wazuh authentication failed: {response.status_code}")
138
                return None
139

140
        except Exception as e:
141
            self.logger.error(f"Error authenticating with Wazuh: {e}")
142
            return None
143

144
    def query_misp_attributes(self, ioc_value: str, ioc_type: str) -> List[Dict]:
145
        """Query MISP for attributes matching the IOC"""
146
        # Check cache first
147
        cache_key = f"{ioc_type}:{ioc_value}"
148
        if cache_key in self.misp_cache:
149
            cache_entry = self.misp_cache[cache_key]
150
            if time.time() - cache_entry['timestamp'] < self.cache_ttl:
151
                self.logger.debug(f"Cache hit for {cache_key}")
152
                return cache_entry['data']
153

154
        try:
155
            search_url = f"{self.misp_url}/attributes/restSearch"
156
            headers = {
157
                'Authorization': self.misp_key,
158
                'Accept': 'application/json',
159
                'Content-Type': 'application/json'
160
            }
161

162
            # Calculate date range for recent attributes
163
            date_from = (datetime.now() - timedelta(days=self.max_age_days)).strftime('%Y-%m-%d')
164

165
            search_params = {
166
                'returnFormat': 'json',
167
                'value': ioc_value,
168
                'type': ioc_type,
169
                'dateFrom': date_from,
170
                'includeEventTags': True,
171
                'includeGalaxy': True,
172
                'limit': 1000
173
            }
174

175
            response = requests.post(
176
                search_url,
177
                headers=headers,
178
                json=search_params,
179
                verify=self.misp_verifycert,
180
                timeout=30
181
            )
182

183
            if response.status_code == 200:
184
                result = response.json()
185
                attributes = result.get('response', {}).get('Attribute', [])
186

187
                # Cache the result
188
                self.misp_cache[cache_key] = {
189
                    'data': attributes,
190
                    'timestamp': time.time()
191
                }
192

193
                self.logger.debug(f"Found {len(attributes)} attributes for {ioc_value}")
194
                return attributes
195
            else:
196
                self.logger.warning(f"MISP query failed for {ioc_value}: {response.status_code}")
197
                return []
198

199
        except Exception as e:
200
            self.logger.error(f"Error querying MISP for {ioc_value}: {e}")
201
            return []
202

203
    def enrich_with_misp_context(self, attributes: List[Dict]) -> Dict:
204
        """Enrich attributes with additional context from MISP"""
205
        if not attributes:
206
            return {}
207

208
        enrichment = {
209
            'misp_events': [],
210
            'threat_level': 'Unknown',
211
            'categories': set(),
212
            'tags': set(),
213
            'galaxy_clusters': set(),
214
            'confidence': 0,
215
            'first_seen': None,
216
            'last_seen': None,
217
            'threat_actor': None,
218
            'malware_family': None,
219
            'campaign': None
220
        }
221

222
        for attr in attributes:
223
            # Event information
224
            event_info = {
225
                'event_id': attr.get('event_id'),
226
                'event_info': attr.get('Event', {}).get('info', ''),
227
                'event_date': attr.get('Event', {}).get('date', ''),
228
                'threat_level': attr.get('Event', {}).get('threat_level_id', 4)
229
            }
230
            enrichment['misp_events'].append(event_info)
231

232
            # Categories
233
            if 'category' in attr:
234
                enrichment['categories'].add(attr['category'])
235

236
            # Tags
237
            event_tags = attr.get('Event', {}).get('Tag', [])
238
            for tag in event_tags:
239
                enrichment['tags'].add(tag.get('name', ''))
240

241
            # Galaxy clusters
242
            event_galaxies = attr.get('Event', {}).get('Galaxy', [])
243
            for galaxy in event_galaxies:
244
                for cluster in galaxy.get('GalaxyCluster', []):
245
                    cluster_value = cluster.get('value', '')
246
                    enrichment['galaxy_clusters'].add(cluster_value)
247

248
                    # Extract threat actor information
249
                    if 'threat-actor' in galaxy.get('type', '').lower():
250
                        enrichment['threat_actor'] = cluster_value
251

252
                    # Extract malware family information
253
                    if 'malware' in galaxy.get('type', '').lower():
254
                        enrichment['malware_family'] = cluster_value
255

256
                    # Extract campaign information
257
                    if 'campaign' in galaxy.get('type', '').lower():
258
                        enrichment['campaign'] = cluster_value
259

260
            # Timestamps
261
            timestamp = attr.get('timestamp')
262
            if timestamp:
263
                if not enrichment['first_seen'] or timestamp < enrichment['first_seen']:
264
                    enrichment['first_seen'] = timestamp
265
                if not enrichment['last_seen'] or timestamp > enrichment['last_seen']:
266
                    enrichment['last_seen'] = timestamp
267

268
        # Convert sets to lists for JSON serialization
269
        enrichment['categories'] = list(enrichment['categories'])
270
        enrichment['tags'] = list(enrichment['tags'])
271
        enrichment['galaxy_clusters'] = list(enrichment['galaxy_clusters'])
272

273
        # Calculate threat level
274
        if enrichment['misp_events']:
275
            threat_levels = [event['threat_level'] for event in enrichment['misp_events']]
276
            enrichment['threat_level'] = min(threat_levels)  # Lower number = higher threat
277

278
        # Calculate confidence based on number of events and recency
279
        num_events = len(enrichment['misp_events'])
280
        if num_events > 0:
281
            base_confidence = min(num_events * 20, 80)  # Max 80 from count
282

283
            # Boost confidence for recent sightings
284
            if enrichment['last_seen']:
285
                days_old = (time.time() - int(enrichment['last_seen'])) / 86400
286
                if days_old < 7:
287
                    base_confidence += 20
288
                elif days_old < 30:
289
                    base_confidence += 10
290

291
            enrichment['confidence'] = min(base_confidence, 100)
292

293
        return enrichment
294

295
    def extract_iocs_from_alert(self, alert: Dict) -> List[Dict]:
296
        """Extract IOCs from Wazuh alert"""
297
        iocs = []
298

299
        # Get alert data
300
        data = alert.get('data', {})
301
        rule = alert.get('rule', {})
302

303
        # Extract IOCs based on rule groups and data fields
304
        for misp_type, wazuh_fields in self.ioc_types.items():
305
            for field in wazuh_fields:
306
                value = self.get_nested_field(data, field)
307
                if value and self.is_valid_ioc(value, misp_type):
308
                    iocs.append({
309
                        'value': value,
310
                        'type': misp_type,
311
                        'field': field,
312
                        'confidence': self.calculate_initial_confidence(field, rule)
313
                    })
314

315
        # Remove duplicates
316
        seen = set()
317
        unique_iocs = []
318
        for ioc in iocs:
319
            key = f"{ioc['type']}:{ioc['value']}"
320
            if key not in seen:
321
                seen.add(key)
322
                unique_iocs.append(ioc)
323

324
        return unique_iocs
325

326
    def get_nested_field(self, data: Dict, field: str) -> Optional[str]:
327
        """Get nested field value from data dictionary"""
328
        if '.' in field:
329
            keys = field.split('.')
330
            current = data
331
            for key in keys:
332
                if isinstance(current, dict) and key in current:
333
                    current = current[key]
334
                else:
335
                    return None
336
            return str(current) if current is not None else None
337
        else:
338
            return data.get(field)
339

340
    def is_valid_ioc(self, value: str, ioc_type: str) -> bool:
341
        """Validate if value is a valid IOC of the specified type"""
342
        import re
343

344
        if not value or len(value) < 3:
345
            return False
346

347
        # Remove quotes and whitespace
348
        value = value.strip('"\'').strip()
349

350
        validators = {
351
            'ip-src': lambda x: self.is_valid_ip(x),
352
            'ip-dst': lambda x: self.is_valid_ip(x),
353
            'domain': lambda x: self.is_valid_domain(x),
354
            'hostname': lambda x: self.is_valid_domain(x),
355
            'url': lambda x: self.is_valid_url(x),
356
            'filename': lambda x: len(x) > 3 and '/' not in x,
357
            'md5': lambda x: re.match(r'^[a-f0-9]{32}$', x.lower()),
358
            'sha1': lambda x: re.match(r'^[a-f0-9]{40}$', x.lower()),
359
            'sha256': lambda x: re.match(r'^[a-f0-9]{64}$', x.lower()),
360
            'email-src': lambda x: '@' in x and '.' in x,
361
            'email-dst': lambda x: '@' in x and '.' in x,
362
            'user-agent': lambda x: len(x) > 10,
363
            'mutex': lambda x: len(x) > 5,
364
            'registry-key': lambda x: 'HKEY' in x.upper() or '\\' in x
365
        }
366

367
        validator = validators.get(ioc_type)
368
        return validator(value) if validator else True
369

370
    def is_valid_ip(self, ip: str) -> bool:
371
        """Validate IP address"""
372
        import ipaddress
373
        try:
374
            addr = ipaddress.ip_address(ip)
375
            # Exclude private and reserved addresses
376
            return not (addr.is_private or addr.is_loopback or addr.is_multicast or addr.is_reserved)
377
        except:
378
            return False
379

380
    def is_valid_domain(self, domain: str) -> bool:
381
        """Validate domain name"""
382
        import re
383
        pattern = r'^([a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
384
        return re.match(pattern, domain) is not None
385

386
    def is_valid_url(self, url: str) -> bool:
387
        """Validate URL"""
388
        import re
389
        pattern = r'^https?://[^\s/$.?#].[^\s]*$'
390
        return re.match(pattern, url) is not None
391

392
    def calculate_initial_confidence(self, field: str, rule: Dict) -> int:
393
        """Calculate initial confidence for IOC based on context"""
394
        base_confidence = 50
395

396
        # Adjust based on rule level
397
        rule_level = rule.get('level', 1)
398
        if rule_level >= 10:
399
            base_confidence += 30
400
        elif rule_level >= 7:
401
            base_confidence += 20
402
        elif rule_level >= 5:
403
            base_confidence += 10
404

405
        # Adjust based on field type
406
        high_confidence_fields = ['srcip', 'dstip', 'md5', 'sha1', 'sha256']
407
        if field in high_confidence_fields:
408
            base_confidence += 10
409

410
        return min(base_confidence, 80)  # Max 80 for initial confidence
411

412
    def process_alert(self, alert: Dict) -> Dict:
413
        """Process a single Wazuh alert for threat intelligence"""
414
        result = {
415
            'alert_id': alert.get('id'),
416
            'timestamp': alert.get('timestamp'),
417
            'rule_id': alert.get('rule', {}).get('id'),
418
            'enrichments': [],
419
            'total_matches': 0,
420
            'highest_confidence': 0,
421
            'threat_level': 'Unknown'
422
        }
423

424
        # Extract IOCs from alert
425
        iocs = self.extract_iocs_from_alert(alert)
426

427
        if not iocs:
428
            self.logger.debug(f"No IOCs extracted from alert {result['alert_id']}")
429
            return result
430

431
        self.logger.info(f"Extracted {len(iocs)} IOCs from alert {result['alert_id']}")
432

433
        # Query MISP for each IOC
434
        for ioc in iocs:
435
            self.logger.debug(f"Querying MISP for {ioc['type']}: {ioc['value']}")
436

437
            misp_attributes = self.query_misp_attributes(ioc['value'], ioc['type'])
438

439
            if misp_attributes:
440
                enrichment = self.enrich_with_misp_context(misp_attributes)
441
                enrichment.update({
442
                    'ioc_value': ioc['value'],
443
                    'ioc_type': ioc['type'],
444
                    'ioc_field': ioc['field'],
445
                    'misp_matches': len(misp_attributes),
446
                    'initial_confidence': ioc['confidence']
447
                })
448

449
                # Calculate final confidence
450
                final_confidence = min(ioc['confidence'] + enrichment['confidence'], 100)
451
                enrichment['final_confidence'] = final_confidence
452

453
                result['enrichments'].append(enrichment)
454
                result['total_matches'] += len(misp_attributes)
455
                result['highest_confidence'] = max(result['highest_confidence'], final_confidence)
456

457
                # Update threat level
458
                if enrichment['threat_level'] != 'Unknown':
459
                    if result['threat_level'] == 'Unknown' or enrichment['threat_level'] < result['threat_level']:
460
                        result['threat_level'] = enrichment['threat_level']
461

462
                self.logger.info(f"IOC {ioc['value']} matched {len(misp_attributes)} MISP attributes (confidence: {final_confidence})")
463

464
        return result
465

466
    def create_enriched_alert(self, original_alert: Dict, enrichment_result: Dict) -> Dict:
467
        """Create enriched alert with MISP intelligence"""
468
        enriched_alert = original_alert.copy()
469

470
        # Add MISP intelligence data
471
        misp_data = {
472
            'misp_integration': {
473
                'processed_at': datetime.now().isoformat(),
474
                'total_iocs_checked': len(enrichment_result['enrichments']),
475
                'total_misp_matches': enrichment_result['total_matches'],
476
                'highest_confidence': enrichment_result['highest_confidence'],
477
                'threat_level': enrichment_result['threat_level'],
478
                'enrichments': enrichment_result['enrichments']
479
            }
480
        }
481

482
        # Update alert data
483
        if 'data' not in enriched_alert:
484
            enriched_alert['data'] = {}
485

486
        enriched_alert['data'].update(misp_data)
487

488
        # Update rule level if high confidence threats found
489
        if enrichment_result['highest_confidence'] >= self.correlation_threshold:
490
            original_level = enriched_alert.get('rule', {}).get('level', 1)
491
            new_level = min(original_level + 3, 15)  # Increase by 3, max 15
492

493
            if 'rule' not in enriched_alert:
494
                enriched_alert['rule'] = {}
495

496
            enriched_alert['rule']['level'] = new_level
497
            enriched_alert['rule']['description'] = f"{enriched_alert['rule'].get('description', '')} [MISP Intelligence Match]"
498

499
        return enriched_alert
500

501
    def send_enriched_alert(self, enriched_alert: Dict) -> bool:
502
        """Send enriched alert back to Wazuh"""
503
        try:
504
            if not self.wazuh_token:
505
                if not self.get_wazuh_token():
506
                    return False
507

508
            # Send to Wazuh events API
509
            events_url = f"{self.wazuh_url}/events"
510
            headers = {
511
                'Authorization': f'Bearer {self.wazuh_token}',
512
                'Content-Type': 'application/json'
513
            }
514

515
            response = requests.post(
516
                events_url,
517
                headers=headers,
518
                json=enriched_alert,
519
                verify=False,
520
                timeout=10
521
            )
522

523
            if response.status_code in [200, 201]:
524
                self.logger.info(f"Successfully sent enriched alert {enriched_alert.get('id')}")
525
                return True
526
            else:
527
                self.logger.error(f"Failed to send enriched alert: {response.status_code}")
528
                return False
529

530
        except Exception as e:
531
            self.logger.error(f"Error sending enriched alert: {e}")
532
            return False
533

534
    def run_integration(self, alert_data: str = None):
535
        """Main integration function"""
536
        try:
537
            if alert_data:
538
                # Process single alert from stdin
539
                alert = json.loads(alert_data)
540
                self.process_single_alert(alert)
541
            else:
542
                # Continuous mode - monitor Wazuh alerts
543
                self.run_continuous_mode()
544

545
        except Exception as e:
546
            self.logger.error(f"Integration error: {e}")
547
            sys.exit(1)
548

549
    def process_single_alert(self, alert: Dict):
550
        """Process a single alert"""
551
        self.logger.info(f"Processing alert {alert.get('id')} from rule {alert.get('rule', {}).get('id')}")
552

553
        # Process alert for threat intelligence
554
        enrichment_result = self.process_alert(alert)
555

556
        # Only enrich if we found matches above threshold
557
        if enrichment_result['highest_confidence'] >= self.correlation_threshold:
558
            self.logger.info(f"Alert {alert.get('id')} exceeded correlation threshold ({enrichment_result['highest_confidence']})")
559

560
            # Create enriched alert
561
            enriched_alert = self.create_enriched_alert(alert, enrichment_result)
562

563
            # Send enriched alert
564
            if self.send_enriched_alert(enriched_alert):
565
                self.logger.info(f"Successfully enriched alert {alert.get('id')}")
566
            else:
567
                self.logger.error(f"Failed to send enriched alert {alert.get('id')}")
568
        else:
569
            self.logger.debug(f"Alert {alert.get('id')} below correlation threshold ({enrichment_result['highest_confidence']})")
570

571
    def run_continuous_mode(self):
572
        """Run in continuous mode monitoring Wazuh alerts"""
573
        self.logger.info("Starting MISP-Wazuh integration in continuous mode")
574

575
        # This would typically connect to Wazuh's real-time alert stream
576
        # For now, we'll simulate by reading from a file or API
577
        self.logger.info("Continuous mode not implemented - use single alert mode")
578

579
def main():
580
    parser = argparse.ArgumentParser(description='MISP-Wazuh Threat Intelligence Integration')
581
    parser.add_argument('--config', default='/var/ossec/integrations/misp_config.json',
582
                       help='Configuration file path')
583
    parser.add_argument('--alert', help='Process single alert (JSON string)')
584
    parser.add_argument('--continuous', action='store_true',
585
                       help='Run in continuous mode')
586

587
    args = parser.parse_args()
588

589
    # Initialize integrator
590
    integrator = MISPWazuhIntegrator(args.config)
591

592
    if args.alert:
593
        # Process single alert
594
        integrator.process_single_alert(json.loads(args.alert))
595
    elif args.continuous:
596
        # Run continuous mode
597
        integrator.run_continuous_mode()
598
    else:
599
        # Read from stdin
600
        alert_data = sys.stdin.read()
601
        if alert_data.strip():
602
            integrator.run_integration(alert_data)
603
        else:
604
            parser.print_help()
605

606
if __name__ == "__main__":
607
    main()

1
<!-- Wazuh Manager Configuration for MISP Integration -->
2
<ossec_config>
3

4
  <!-- Custom MISP Integration -->
5
  <integration>
6
    <name>custom-misp</name>
7
    <hook_url>/var/ossec/integrations/misp_integration.py</hook_url>
8
    <level>7</level>
9
    <group>misp_enabled</group>
10
    <alert_format>json</alert_format>
11
    <max_log>1MB</max_log>
12
  </integration>
13

14
  <!-- High-priority events for immediate MISP correlation -->
15
  <integration>
16
    <name>custom-misp-priority</name>
17
    <hook_url>/var/ossec/integrations/misp_integration.py --priority</hook_url>
18
    <level>10</level>
19
    <alert_format>json</alert_format>
20
    <max_log>1MB</max_log>
21
  </integration>
22

23
  <!-- MISP integration for specific rule groups -->
24
  <integration>
25
    <name>custom-misp-malware</name>
26
    <hook_url>/var/ossec/integrations/misp_integration.py</hook_url>
27
    <group>malware,trojan,virus</group>
28
    <alert_format>json</alert_format>
29
    <max_log>1MB</max_log>
30
  </integration>
31

32
  <integration>
33
    <name>custom-misp-network</name>
34
    <hook_url>/var/ossec/integrations/misp_integration.py</hook_url>
35
    <group>network,intrusion_detection</group>
36
    <alert_format>json</alert_format>
37
    <max_log>1MB</max_log>
38
  </integration>
39

40
</ossec_config>