XDR vs SIEM vs SOAR: What Beginners Should Learn in 2026

#!/usr/bin/env python3
"""
Security Technology Assessment Framework
Helps choose between SIEM, SOAR, and XDR
"""

from typing import Dict, List
from dataclasses import dataclass
from enum import Enum

class TechnologyType(Enum):
    SIEM = "siem"
    SOAR = "soar"
    XDR = "xdr"
    HYBRID = "hybrid"

@dataclass
class Requirements:
    log_volume: int  # logs per day
    automation_needed: bool
    unified_detection: bool
    budget: str  # low, medium, high
    team_size: int
    complexity_tolerance: str  # low, medium, high

class TechnologyAssessment:
    """Assess which technology fits requirements."""
    
    def assess(self, requirements: Requirements) -> TechnologyType:
        """Assess requirements and recommend technology."""
        score_siem = 0
        score_soar = 0
        score_xdr = 0
        
        # Log volume scoring
        if requirements.log_volume > 1000000:
            score_siem += 3
            score_xdr += 2
        else:
            score_xdr += 3
        
        # Automation scoring
        if requirements.automation_needed:
            score_soar += 3
            score_xdr += 2
        
        # Unified detection scoring
        if requirements.unified_detection:
            score_xdr += 3
            score_siem += 1
        
        # Budget scoring
        if requirements.budget == "high":
            score_xdr += 3
            score_siem += 2
            score_soar += 2
        elif requirements.budget == "medium":
            score_siem += 3
            score_soar += 2
        
        # Determine recommendation
        scores = {
            TechnologyType.SIEM: score_siem,
            TechnologyType.SOAR: score_soar,
            TechnologyType.XDR: score_xdr
        }
        
        max_score = max(scores.values())
        if max_score == score_siem == score_xdr:
            return TechnologyType.HYBRID
        
        return max(scores, key=scores.get)

# Usage
requirements = Requirements(
    log_volume=500000,
    automation_needed=True,
    unified_detection=True,
    budget="medium",
    team_size=5,
    complexity_tolerance="medium"
)

assessment = TechnologyAssessment()
recommendation = assessment.assess(requirements)
print(f"Recommended: {recommendation.value}")

#!/usr/bin/env python3
"""
SIEM Implementation Example
Production-ready SIEM log aggregation and correlation
"""

from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
import logging
import json

logger = logging.getLogger(__name__)

@dataclass
class SecurityEvent:
    """Security event from log source."""
    event_id: str
    timestamp: datetime
    source: str
    event_type: str
    severity: str
    data: Dict
    raw_log: str

class SIEMEngine:
    """SIEM log aggregation and correlation engine."""
    
    def __init__(self):
        self.events: List[SecurityEvent] = []
        self.correlations: Dict[str, List[str]] = defaultdict(list)
        self.alerts: List[Dict] = []
    
    def ingest_log(self, log_data: Dict) -> bool:
        """Ingest log from source."""
        try:
            event = SecurityEvent(
                event_id=f"EVT-{len(self.events)+1}",
                timestamp=datetime.fromisoformat(log_data.get('timestamp', datetime.now().isoformat())),
                source=log_data.get('source', 'unknown'),
                event_type=log_data.get('event_type', 'unknown'),
                severity=log_data.get('severity', 'low'),
                data=log_data.get('data', {}),
                raw_log=json.dumps(log_data)
            )
            
            self.events.append(event)
            self._correlate_event(event)
            
            logger.info(f"Event ingested: {event.event_id}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to ingest log: {e}")
            return False
    
    def _correlate_event(self, event: SecurityEvent):
        """Correlate event with existing events."""
        # Simple correlation by source IP
        source_ip = event.data.get('source_ip')
        if source_ip:
            self.correlations[source_ip].append(event.event_id)
            
            # Check for correlation patterns
            if len(self.correlations[source_ip]) >= 5:
                self._generate_alert(source_ip, self.correlations[source_ip])
    
    def _generate_alert(self, source_ip: str, event_ids: List[str]):
        """Generate security alert from correlation."""
        alert = {
            'alert_id': f"ALERT-{len(self.alerts)+1}",
            'timestamp': datetime.now().isoformat(),
            'source_ip': source_ip,
            'event_count': len(event_ids),
            'severity': 'high',
            'description': f"Multiple security events from {source_ip}",
            'event_ids': event_ids
        }
        
        self.alerts.append(alert)
        logger.warning(f"Alert generated: {alert['alert_id']}")
    
    def query_events(self, query: Dict) -> List[SecurityEvent]:
        """Query events based on criteria."""
        results = []
        
        for event in self.events:
            # Simple query matching
            if query.get('source') and event.source != query['source']:
                continue
            if query.get('event_type') and event.event_type != query['event_type']:
                continue
            if query.get('severity') and event.severity != query['severity']:
                continue
            
            results.append(event)
        
        return results
    
    def get_statistics(self) -> Dict:
        """Get SIEM statistics."""
        return {
            'total_events': len(self.events),
            'total_alerts': len(self.alerts),
            'events_by_source': {
                source: len([e for e in self.events if e.source == source])
                for source in set(e.source for e in self.events)
            },
            'events_by_severity': {
                severity: len([e for e in self.events if e.severity == severity])
                for severity in set(e.severity for e in self.events)
            }
        }

# Example usage
if __name__ == "__main__":
    siem = SIEMEngine()
    
    # Ingest sample logs
    siem.ingest_log({
        'timestamp': datetime.now().isoformat(),
        'source': 'firewall',
        'event_type': 'blocked_connection',
        'severity': 'medium',
        'data': {'source_ip': '192.168.1.100', 'destination': '10.0.0.1'}
    })
    
    # Query events
    results = siem.query_events({'severity': 'medium'})
    print(f"Found {len(results)} medium severity events")
    
    # Get statistics
    stats = siem.get_statistics()
    print(f"SIEM Statistics: {json.dumps(stats, indent=2)}")

#!/usr/bin/env python3
"""
SOAR Implementation Example
Production-ready security orchestration and automation
"""

from typing import List, Dict, Callable, Optional
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class PlaybookStatus(Enum):
    """Playbook execution status."""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class PlaybookStep:
    """Playbook execution step."""
    step_id: str
    name: str
    action: str
    parameters: Dict
    condition: Optional[str] = None

@dataclass
class Playbook:
    """SOAR playbook definition."""
    playbook_id: str
    name: str
    description: str
    steps: List[PlaybookStep]
    enabled: bool = True

class SOAREngine:
    """SOAR orchestration and automation engine."""
    
    def __init__(self):
        self.playbooks: Dict[str, Playbook] = {}
        self.executions: Dict[str, Dict] = {}
        self.actions: Dict[str, Callable] = {}
        self._register_default_actions()
    
    def _register_default_actions(self):
        """Register default automation actions."""
        self.actions['isolate_endpoint'] = self._isolate_endpoint
        self.actions['block_ip'] = self._block_ip
        self.actions['send_notification'] = self._send_notification
        self.actions['create_ticket'] = self._create_ticket
    
    def register_playbook(self, playbook: Playbook) -> bool:
        """Register playbook."""
        try:
            self.playbooks[playbook.playbook_id] = playbook
            logger.info(f"Playbook registered: {playbook.playbook_id}")
            return True
        except Exception as e:
            logger.error(f"Failed to register playbook: {e}")
            return False
    
    def execute_playbook(self, playbook_id: str, context: Dict) -> Dict:
        """Execute playbook."""
        if playbook_id not in self.playbooks:
            return {'error': 'Playbook not found'}
        
        playbook = self.playbooks[playbook_id]
        if not playbook.enabled:
            return {'error': 'Playbook disabled'}
        
        execution_id = f"EXEC-{len(self.executions)+1}"
        self.executions[execution_id] = {
            'playbook_id': playbook_id,
            'status': PlaybookStatus.RUNNING.value,
            'started_at': datetime.now().isoformat(),
            'steps_completed': []
        }
        
        try:
            for step in playbook.steps:
                # Check condition if present
                if step.condition and not self._evaluate_condition(step.condition, context):
                    continue
                
                # Execute action
                if step.action in self.actions:
                    result = self.actions[step.action](step.parameters, context)
                    self.executions[execution_id]['steps_completed'].append({
                        'step_id': step.step_id,
                        'result': result
                    })
                else:
                    logger.warning(f"Unknown action: {step.action}")
            
            self.executions[execution_id]['status'] = PlaybookStatus.COMPLETED.value
            self.executions[execution_id]['completed_at'] = datetime.now().isoformat()
            
            logger.info(f"Playbook {playbook_id} executed successfully")
            return self.executions[execution_id]
            
        except Exception as e:
            self.executions[execution_id]['status'] = PlaybookStatus.FAILED.value
            self.executions[execution_id]['error'] = str(e)
            logger.error(f"Playbook execution failed: {e}")
            return self.executions[execution_id]
    
    def _evaluate_condition(self, condition: str, context: Dict) -> bool:
        """Evaluate playbook condition."""
        # Simplified condition evaluation
        # In production, would use proper expression evaluator
        return True
    
    def _isolate_endpoint(self, parameters: Dict, context: Dict) -> Dict:
        """Isolate endpoint action."""
        endpoint_id = parameters.get('endpoint_id')
        logger.info(f"Isolating endpoint: {endpoint_id}")
        return {'status': 'success', 'endpoint_id': endpoint_id}
    
    def _block_ip(self, parameters: Dict, context: Dict) -> Dict:
        """Block IP action."""
        ip_address = parameters.get('ip_address')
        logger.info(f"Blocking IP: {ip_address}")
        return {'status': 'success', 'ip_address': ip_address}
    
    def _send_notification(self, parameters: Dict, context: Dict) -> Dict:
        """Send notification action."""
        message = parameters.get('message', 'Security alert')
        logger.info(f"Sending notification: {message}")
        return {'status': 'success'}
    
    def _create_ticket(self, parameters: Dict, context: Dict) -> Dict:
        """Create ticket action."""
        title = parameters.get('title', 'Security incident')
        logger.info(f"Creating ticket: {title}")
        return {'status': 'success', 'ticket_id': 'TICKET-001'}

# Example usage
if __name__ == "__main__":
    soar = SOAREngine()
    
    # Create playbook
    playbook = Playbook(
        playbook_id="PB-001",
        name="Malware Detection Response",
        description="Automated response to malware detection",
        steps=[
            PlaybookStep(
                step_id="STEP-1",
                name="Isolate Endpoint",
                action="isolate_endpoint",
                parameters={'endpoint_id': 'ENDPOINT-001'}
            ),
            PlaybookStep(
                step_id="STEP-2",
                name="Block IP",
                action="block_ip",
                parameters={'ip_address': '192.168.1.100'}
            ),
            PlaybookStep(
                step_id="STEP-3",
                name="Send Notification",
                action="send_notification",
                parameters={'message': 'Malware detected and contained'}
            )
        ]
    )
    
    soar.register_playbook(playbook)
    
    # Execute playbook
    result = soar.execute_playbook("PB-001", {})
    print(f"Playbook execution: {result['status']}")

#!/usr/bin/env python3
"""
XDR Implementation Example
Production-ready extended detection and response
"""

from typing import List, Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import logging

logger = logging.getLogger(__name__)

class DataSource(Enum):
    """XDR data sources."""
    ENDPOINT = "endpoint"
    NETWORK = "network"
    CLOUD = "cloud"
    EMAIL = "email"

@dataclass
class XDREvent:
    """XDR unified event."""
    event_id: str
    timestamp: datetime
    data_source: DataSource
    event_type: str
    severity: str
    entities: Dict  # hosts, users, IPs, etc.
    raw_data: Dict

class XDREngine:
    """XDR unified detection and response engine."""
    
    def __init__(self):
        self.events: List[XDREvent] = []
        self.detections: List[Dict] = []
        self.investigations: List[Dict] = []
    
    def ingest_event(self, source: DataSource, event_data: Dict) -> bool:
        """Ingest event from data source."""
        try:
            event = XDREvent(
                event_id=f"XDR-{len(self.events)+1}",
                timestamp=datetime.fromisoformat(event_data.get('timestamp', datetime.now().isoformat())),
                data_source=source,
                event_type=event_data.get('event_type', 'unknown'),
                severity=event_data.get('severity', 'low'),
                entities=event_data.get('entities', {}),
                raw_data=event_data
            )
            
            self.events.append(event)
            self._analyze_event(event)
            
            logger.info(f"XDR event ingested: {event.event_id}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to ingest event: {e}")
            return False
    
    def _analyze_event(self, event: XDREvent):
        """Analyze event for threats."""
        # Cross-correlate with other events
        related_events = self._find_related_events(event)
        
        if len(related_events) >= 3:
            detection = {
                'detection_id': f"DET-{len(self.detections)+1}",
                'timestamp': datetime.now().isoformat(),
                'severity': 'high',
                'description': 'Multi-source threat detected',
                'events': [e.event_id for e in related_events] + [event.event_id],
                'entities': event.entities
            }
            
            self.detections.append(detection)
            self._auto_investigate(detection)
            logger.warning(f"Threat detected: {detection['detection_id']}")
    
    def _find_related_events(self, event: XDREvent) -> List[XDREvent]:
        """Find related events across data sources."""
        related = []
        
        # Find events with same entities
        for e in self.events:
            if e.event_id == event.event_id:
                continue
            
            # Check for entity overlap
            if self._entities_overlap(event.entities, e.entities):
                related.append(e)
        
        return related
    
    def _entities_overlap(self, entities1: Dict, entities2: Dict) -> bool:
        """Check if entities overlap."""
        for key in ['host', 'user', 'ip']:
            if key in entities1 and key in entities2:
                if entities1[key] == entities2[key]:
                    return True
        return False
    
    def _auto_investigate(self, detection: Dict):
        """Automated investigation."""
        investigation = {
            'investigation_id': f"INV-{len(self.investigations)+1}",
            'detection_id': detection['detection_id'],
            'status': 'in_progress',
            'started_at': datetime.now().isoformat(),
            'findings': []
        }
        
        # Automated investigation steps
        investigation['findings'].append({
            'finding': 'Cross-source correlation detected',
            'confidence': 'high'
        })
        
        self.investigations.append(investigation)
        logger.info(f"Investigation started: {investigation['investigation_id']}")
    
    def get_unified_view(self, entity_id: str, entity_type: str = 'host') -> Dict:
        """Get unified view of entity across all data sources."""
        entity_events = [
            e for e in self.events
            if entity_type in e.entities and e.entities[entity_type] == entity_id
        ]
        
        return {
            'entity_id': entity_id,
            'entity_type': entity_type,
            'total_events': len(entity_events),
            'events_by_source': {
                source.value: len([e for e in entity_events if e.data_source == source])
                for source in DataSource
            },
            'events': [e.event_id for e in entity_events]
        }

# Example usage
if __name__ == "__main__":
    xdr = XDREngine()
    
    # Ingest events from different sources
    xdr.ingest_event(DataSource.ENDPOINT, {
        'timestamp': datetime.now().isoformat(),
        'event_type': 'malware_detected',
        'severity': 'high',
        'entities': {'host': 'HOST-001', 'user': 'USER-001'}
    })
    
    xdr.ingest_event(DataSource.NETWORK, {
        'timestamp': datetime.now().isoformat(),
        'event_type': 'suspicious_connection',
        'severity': 'medium',
        'entities': {'host': 'HOST-001', 'ip': '192.168.1.100'}
    })
    
    # Get unified view
    view = xdr.get_unified_view('HOST-001', 'host')
    print(f"Unified view: {view}")

#!/usr/bin/env python3
"""
Unit tests for SIEM/SOAR/XDR
"""

import pytest
from datetime import datetime
from siem_soar_xdr import (
    SIEMEngine, SecurityEvent,
    SOAREngine, Playbook, PlaybookStep,
    XDREngine, DataSource
)

class TestSIEMEngine:
    """Tests for SIEMEngine."""
    
    @pytest.fixture
    def siem(self):
        return SIEMEngine()
    
    def test_ingest_log(self, siem):
        """Test log ingestion."""
        result = siem.ingest_log({
            'timestamp': datetime.now().isoformat(),
            'source': 'firewall',
            'event_type': 'blocked',
            'severity': 'medium',
            'data': {}
        })
        assert result is True
        assert len(siem.events) > 0

class TestSOAREngine:
    """Tests for SOAREngine."""
    
    @pytest.fixture
    def soar(self):
        return SOAREngine()
    
    def test_register_playbook(self, soar):
        """Test playbook registration."""
        playbook = Playbook(
            playbook_id="PB-001",
            name="Test Playbook",
            description="Test",
            steps=[]
        )
        result = soar.register_playbook(playbook)
        assert result is True

class TestXDREngine:
    """Tests for XDREngine."""
    
    @pytest.fixture
    def xdr(self):
        return XDREngine()
    
    def test_ingest_event(self, xdr):
        """Test event ingestion."""
        result = xdr.ingest_event(DataSource.ENDPOINT, {
            'timestamp': datetime.now().isoformat(),
            'event_type': 'test',
            'severity': 'low',
            'entities': {}
        })
        assert result is True

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

#!/usr/bin/env python3
"""
SIEM/SOAR/XDR Cleanup
Production-ready cleanup and resource management
"""

import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class SecurityPlatformCleanup:
    """Handles cleanup operations."""
    
    def __init__(self, siem=None, soar=None, xdr=None):
        self.siem = siem
        self.soar = soar
        self.xdr = xdr
    
    def cleanup_old_events(self, days: int = 90):
        """Clean up old events."""
        cutoff_date = datetime.now() - timedelta(days=days)
        
        if self.siem:
            initial_count = len(self.siem.events)
            self.siem.events = [
                e for e in self.siem.events
                if e.timestamp >= cutoff_date
            ]
            removed = initial_count - len(self.siem.events)
            logger.info(f"Cleaned up {removed} SIEM events")
        
        if self.xdr:
            initial_count = len(self.xdr.events)
            self.xdr.events = [
                e for e in self.xdr.events
                if e.timestamp >= cutoff_date
            ]
            removed = initial_count - len(self.xdr.events)
            logger.info(f"Cleaned up {removed} XDR events")
    
    def cleanup(self):
        """Perform complete cleanup."""
        logger.info("Starting security platform cleanup")
        self.cleanup_old_events()
        logger.info("Security platform cleanup complete")