SOC 2026: How Modern Security Operations Centers Work (Co...

#!/usr/bin/env python3
"""
SOC Alert Processing Workflow
Production-ready SOC workflow with comprehensive error handling, threat intelligence, and automation
"""

from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime, timedelta
import logging
import json
import uuid
from collections import defaultdict
import threading
from queue import PriorityQueue

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class AlertSeverity(Enum):
    """Alert severity levels."""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class AlertStatus(Enum):
    """Alert processing status."""
    NEW = "new"
    ENRICHED = "enriched"
    ANALYZED = "analyzed"
    INVESTIGATING = "investigating"
    RESOLVED = "resolved"
    FALSE_POSITIVE = "false_positive"
    ESCALATED = "escalated"

@dataclass
class ThreatIndicator:
    """Threat intelligence indicator."""
    indicator_type: str  # IP, domain, hash, etc.
    value: str
    threat_score: float
    source: str
    first_seen: datetime
    last_seen: datetime
    tags: List[str] = field(default_factory=list)

@dataclass
class SOCAlert:
    """SOC alert with comprehensive metadata."""
    alert_id: str
    title: str
    severity: AlertSeverity
    source: str
    timestamp: datetime
    description: str
    indicators: List[str]
    status: AlertStatus = AlertStatus.NEW
    threat_indicators: List[ThreatIndicator] = field(default_factory=list)
    enrichment_data: Dict = field(default_factory=dict)
    analysis_result: Optional[Dict] = None
    response_action: Optional[str] = None
    assigned_analyst: Optional[str] = None
    resolution_notes: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    
    def to_dict(self) -> Dict:
        """Convert alert to dictionary for serialization."""
        data = asdict(self)
        data['severity'] = self.severity.value
        data['status'] = self.status.value
        data['timestamp'] = self.timestamp.isoformat()
        data['created_at'] = self.created_at.isoformat()
        data['updated_at'] = self.updated_at.isoformat()
        data['threat_indicators'] = [
            {
                **asdict(ti),
                'first_seen': ti.first_seen.isoformat(),
                'last_seen': ti.last_seen.isoformat()
            }
            for ti in self.threat_indicators
        ]
        return data

class ThreatIntelligenceProvider:
    """Threat intelligence provider for alert enrichment."""
    
    def __init__(self):
        # In production, this would connect to real TI feeds
        self.ioc_cache: Dict[str, ThreatIndicator] = {}
        self._load_sample_indicators()
    
    def _load_sample_indicators(self):
        """Load sample threat indicators for demonstration."""
        sample_ips = [
            ("192.168.1.100", 0.9, "Known malicious IP"),
            ("10.0.0.50", 0.7, "Suspicious activity"),
        ]
        
        for ip, score, source in sample_ips:
            indicator = ThreatIndicator(
                indicator_type="IP",
                value=ip,
                threat_score=score,
                source=source,
                first_seen=datetime.now() - timedelta(days=30),
                last_seen=datetime.now(),
                tags=["malicious", "botnet"]
            )
            self.ioc_cache[ip] = indicator
    
    def lookup_indicator(self, indicator_value: str, indicator_type: str = "IP") -> Optional[ThreatIndicator]:
        """Look up threat indicator in intelligence feeds.
        
        Args:
            indicator_value: Value to look up (IP, domain, hash, etc.)
            indicator_type: Type of indicator (IP, domain, hash, etc.)
            
        Returns:
            ThreatIndicator if found, None otherwise
        """
        cache_key = f"{indicator_type}:{indicator_value}"
        if cache_key in self.ioc_cache:
            logger.info(f"Found threat indicator: {indicator_value}")
            return self.ioc_cache[cache_key]
        
        # In production, query external TI feeds here
        logger.debug(f"No threat intelligence found for {indicator_value}")
        return None
    
    def enrich_indicators(self, indicators: List[str]) -> List[ThreatIndicator]:
        """Enrich list of indicators with threat intelligence.
        
        Args:
            indicators: List of indicator strings
            
        Returns:
            List of ThreatIndicator objects
        """
        enriched = []
        for indicator in indicators:
            # Try to extract type and value
            if ":" in indicator:
                parts = indicator.split(":", 1)
                indicator_type = parts[0].strip()
                value = parts[1].strip()
            else:
                # Default to IP if no type specified
                indicator_type = "IP"
                value = indicator.strip()
            
            ti = self.lookup_indicator(value, indicator_type)
            if ti:
                enriched.append(ti)
        
        return enriched

class SOCWorkflow:
    """Production-ready SOC alert processing workflow."""
    
    def __init__(self, ti_provider: Optional[ThreatIntelligenceProvider] = None):
        """Initialize SOC workflow.
        
        Args:
            ti_provider: Threat intelligence provider instance
        """
        self.alerts: Dict[str, SOCAlert] = {}
        self.alert_queue: PriorityQueue = PriorityQueue()
        self.processed_count = 0
        self.false_positive_count = 0
        self.resolved_count = 0
        self.ti_provider = ti_provider or ThreatIntelligenceProvider()
        self.lock = threading.Lock()
        self.metrics = {
            'total_alerts': 0,
            'processed': 0,
            'false_positives': 0,
            'resolved': 0,
            'escalated': 0,
            'avg_processing_time': 0.0
        }
    
    def receive_alert(self, alert: SOCAlert) -> bool:
        """Receive and queue alert for processing.
        
        Args:
            alert: SOCAlert object to process
            
        Returns:
            True if alert was received successfully, False otherwise
        """
        try:
            with self.lock:
                if alert.alert_id in self.alerts:
                    logger.warning(f"Alert {alert.alert_id} already exists, updating")
                
                self.alerts[alert.alert_id] = alert
                priority = self.prioritize_alert(alert)
                self.alert_queue.put((priority, alert.alert_id))
                self.metrics['total_alerts'] += 1
                
            logger.info(f"Alert received: {alert.alert_id} - {alert.title} (Priority: {priority})")
            return True
            
        except Exception as e:
            logger.error(f"Failed to receive alert {alert.alert_id}: {e}", exc_info=True)
            return False
    
    def prioritize_alert(self, alert: SOCAlert) -> int:
        """Calculate alert priority based on severity and context.
        
        Args:
            alert: SOCAlert to prioritize
            
        Returns:
            Priority score (lower = higher priority)
        """
        base_priority = {
            AlertSeverity.CRITICAL: 1,
            AlertSeverity.HIGH: 2,
            AlertSeverity.MEDIUM: 3,
            AlertSeverity.LOW: 4
        }.get(alert.severity, 5)
        
        # Adjust priority based on threat indicators
        if alert.threat_indicators:
            max_threat_score = max(ti.threat_score for ti in alert.threat_indicators)
            if max_threat_score > 0.8:
                base_priority = max(1, base_priority - 1)
        
        return base_priority
    
    def enrich_alert(self, alert: SOCAlert) -> SOCAlert:
        """Enrich alert with threat intelligence and context.
        
        Args:
            alert: Alert to enrich
            
        Returns:
            Enriched alert
        """
        try:
            logger.info(f"Enriching alert {alert.alert_id}")
            
            # Enrich with threat intelligence
            threat_indicators = self.ti_provider.enrich_indicators(alert.indicators)
            alert.threat_indicators = threat_indicators
            
            # Add enrichment metadata
            alert.enrichment_data = {
                'enriched_at': datetime.now().isoformat(),
                'indicator_count': len(threat_indicators),
                'max_threat_score': max([ti.threat_score for ti in threat_indicators], default=0.0),
                'sources': list(set([ti.source for ti in threat_indicators]))
            }
            
            alert.status = AlertStatus.ENRICHED
            alert.updated_at = datetime.now()
            
            logger.info(f"Alert {alert.alert_id} enriched with {len(threat_indicators)} threat indicators")
            return alert
            
        except Exception as e:
            logger.error(f"Failed to enrich alert {alert.alert_id}: {e}", exc_info=True)
            return alert
    
    def analyze_alert(self, alert: SOCAlert) -> Dict:
        """Analyze alert for threat indicators and determine risk.
        
        Args:
            alert: Alert to analyze
            
        Returns:
            Analysis result dictionary
        """
        try:
            logger.info(f"Analyzing alert {alert.alert_id}")
            
            # Calculate threat score
            base_score = {
                AlertSeverity.CRITICAL: 1.0,
                AlertSeverity.HIGH: 0.75,
                AlertSeverity.MEDIUM: 0.5,
                AlertSeverity.LOW: 0.25
            }.get(alert.severity, 0.0)
            
            # Adjust score based on threat indicators
            if alert.threat_indicators:
                max_ti_score = max([ti.threat_score for ti in alert.threat_indicators])
                threat_score = (base_score + max_ti_score) / 2
            else:
                threat_score = base_score
            
            # Determine recommended action
            if threat_score >= 0.9:
                recommended_action = "immediate_response"
            elif threat_score >= 0.7:
                recommended_action = "priority_investigation"
            elif threat_score >= 0.5:
                recommended_action = "standard_investigation"
            else:
                recommended_action = "review"
            
            analysis = {
                'threat_score': threat_score,
                'base_severity': alert.severity.value,
                'indicator_count': len(alert.threat_indicators),
                'recommended_action': recommended_action,
                'confidence': min(1.0, len(alert.threat_indicators) * 0.2 + 0.5),
                'analysis_timestamp': datetime.now().isoformat()
            }
            
            alert.analysis_result = analysis
            alert.status = AlertStatus.ANALYZED
            alert.updated_at = datetime.now()
            
            logger.info(f"Alert {alert.alert_id} analyzed: threat_score={threat_score:.2f}, action={recommended_action}")
            return analysis
            
        except Exception as e:
            logger.error(f"Failed to analyze alert {alert.alert_id}: {e}", exc_info=True)
            return {'error': str(e)}
    
    def determine_response(self, alert: SOCAlert, analysis: Dict) -> str:
        """Determine automated response action based on analysis.
        
        Args:
            alert: Alert object
            analysis: Analysis result dictionary
            
        Returns:
            Response action string
        """
        try:
            recommended_action = analysis.get('recommended_action', 'review')
            threat_score = analysis.get('threat_score', 0.0)
            
            # Automated response logic
            if threat_score >= 0.9 and alert.severity == AlertSeverity.CRITICAL:
                response = "automated_containment"
            elif threat_score >= 0.8:
                response = "automated_isolation"
            elif threat_score >= 0.7:
                response = "automated_blocking"
            else:
                response = recommended_action
            
            alert.response_action = response
            alert.updated_at = datetime.now()
            
            logger.info(f"Response determined for alert {alert.alert_id}: {response}")
            return response
            
        except Exception as e:
            logger.error(f"Failed to determine response for alert {alert.alert_id}: {e}")
            return "manual_review"
    
    def process_alert(self, alert: SOCAlert) -> Dict:
        """Process alert through complete workflow.
        
        Args:
            alert: Alert to process
            
        Returns:
            Processing result dictionary
        """
        start_time = datetime.now()
        
        try:
            logger.info(f"Processing alert {alert.alert_id}: {alert.title}")
            
            # Step 1: Enrich with threat intelligence
            enriched_alert = self.enrich_alert(alert)
            
            # Step 2: Analyze threat
            analysis = self.analyze_alert(enriched_alert)
            
            if 'error' in analysis:
                return {
                    'alert_id': alert.alert_id,
                    'status': 'failed',
                    'error': analysis['error']
                }
            
            # Step 3: Determine response
            response = self.determine_response(enriched_alert, analysis)
            
            # Update metrics
            processing_time = (datetime.now() - start_time).total_seconds()
            with self.lock:
            self.processed_count += 1
                self.metrics['processed'] += 1
                self.metrics['avg_processing_time'] = (
                    (self.metrics['avg_processing_time'] * (self.processed_count - 1) + processing_time) 
                    / self.processed_count
                )
            
            result = {
                'alert_id': alert.alert_id,
                'status': 'processed',
                'threat_score': analysis.get('threat_score', 0.0),
                'recommended_action': analysis.get('recommended_action'),
                'response_action': response,
                'processing_time_seconds': processing_time,
                'threat_indicators': len(enriched_alert.threat_indicators),
                'enrichment_data': enriched_alert.enrichment_data
            }
            
            logger.info(f"Alert {alert.alert_id} processed successfully in {processing_time:.2f}s")
            return result
            
        except Exception as e:
            logger.error(f"Alert processing failed for {alert.alert_id}: {e}", exc_info=True)
            return {
                'alert_id': alert.alert_id,
                'status': 'failed',
                'error': str(e)
            }
    
    def mark_false_positive(self, alert_id: str, notes: str = "") -> bool:
        """Mark alert as false positive.
        
        Args:
            alert_id: Alert ID to mark
            notes: Optional notes about why it's a false positive
            
        Returns:
            True if successful, False otherwise
        """
        try:
            if alert_id not in self.alerts:
                logger.warning(f"Alert {alert_id} not found")
                return False
            
            alert = self.alerts[alert_id]
            alert.status = AlertStatus.FALSE_POSITIVE
            alert.resolution_notes = notes
            alert.updated_at = datetime.now()
            
            with self.lock:
                self.false_positive_count += 1
                self.metrics['false_positives'] += 1
            
            logger.info(f"Alert {alert_id} marked as false positive")
            return True
            
        except Exception as e:
            logger.error(f"Failed to mark alert {alert_id} as false positive: {e}")
            return False
    
    def resolve_alert(self, alert_id: str, notes: str = "") -> bool:
        """Resolve alert.
        
        Args:
            alert_id: Alert ID to resolve
            notes: Resolution notes
            
        Returns:
            True if successful, False otherwise
        """
        try:
            if alert_id not in self.alerts:
                logger.warning(f"Alert {alert_id} not found")
                return False
            
            alert = self.alerts[alert_id]
            alert.status = AlertStatus.RESOLVED
            alert.resolution_notes = notes
            alert.updated_at = datetime.now()
            
            with self.lock:
                self.resolved_count += 1
                self.metrics['resolved'] += 1
            
            logger.info(f"Alert {alert_id} resolved")
            return True
            
        except Exception as e:
            logger.error(f"Failed to resolve alert {alert_id}: {e}")
            return False
    
    def get_metrics(self) -> Dict:
        """Get SOC workflow metrics.
        
        Returns:
            Dictionary of metrics
        """
        with self.lock:
        return {
                **self.metrics,
                'active_alerts': len([a for a in self.alerts.values() if a.status != AlertStatus.RESOLVED and a.status != AlertStatus.FALSE_POSITIVE]),
                'queue_size': self.alert_queue.qsize()
            }
    
    def export_alert(self, alert_id: str) -> Optional[Dict]:
        """Export alert data.
        
        Args:
            alert_id: Alert ID to export
            
        Returns:
            Alert dictionary or None if not found
        """
        if alert_id in self.alerts:
            return self.alerts[alert_id].to_dict()
        return None
    
    def cleanup(self):
        """Clean up workflow resources."""
        logger.info("Cleaning up SOC workflow resources")
        # In production, close connections, save state, etc.
        pass


# Example usage and testing
if __name__ == "__main__":
    # Initialize workflow
workflow = SOCWorkflow()
    
    # Create sample alerts
    alert1 = SOCAlert(
    alert_id="ALERT-001",
    title="Suspicious Login Activity",
    severity=AlertSeverity.HIGH,
    source="SIEM",
    timestamp=datetime.now(),
        description="Multiple failed login attempts from suspicious IP",
        indicators=["IP: 192.168.1.100", "User: admin", "Domain: example.com"]
    )
    
    alert2 = SOCAlert(
        alert_id="ALERT-002",
        title="Malware Detection",
        severity=AlertSeverity.CRITICAL,
        source="EDR",
        timestamp=datetime.now(),
        description="Malicious file detected on endpoint",
        indicators=["Hash: abc123def456", "IP: 10.0.0.50"]
    )
    
    # Process alerts
    workflow.receive_alert(alert1)
    workflow.receive_alert(alert2)
    
    result1 = workflow.process_alert(alert1)
    result2 = workflow.process_alert(alert2)
    
    print(f"\nAlert 1 Processing Result:")
    print(json.dumps(result1, indent=2))
    
    print(f"\nAlert 2 Processing Result:")
    print(json.dumps(result2, indent=2))
    
    # Get metrics
    metrics = workflow.get_metrics()
    print(f"\nSOC Metrics:")
    print(json.dumps(metrics, indent=2))
    
    # Cleanup
    workflow.cleanup()

#!/usr/bin/env python3
"""
SOC Threat Intelligence Integration
Production-ready TI integration with multiple feed support
"""

from typing import List, Dict, Optional, Set
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
import json
import requests
from abc import ABC, abstractmethod
import hashlib

logger = logging.getLogger(__name__)

@dataclass
class ThreatFeed:
    """Threat intelligence feed configuration."""
    name: str
    url: str
    api_key: Optional[str] = None
    feed_type: str = "json"  # json, csv, stix, etc.
    update_interval: int = 3600  # seconds
    enabled: bool = True

class ThreatIntelligenceFeed(ABC):
    """Abstract base class for threat intelligence feeds."""
    
    @abstractmethod
    def fetch_indicators(self) -> List[Dict]:
        """Fetch indicators from feed."""
        pass
    
    @abstractmethod
    def parse_indicator(self, indicator_data: Dict) -> Optional[Dict]:
        """Parse indicator from feed data."""
        pass

class AbuseIPDBFeed(ThreatIntelligenceFeed):
    """AbuseIPDB threat intelligence feed."""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.abuseipdb.com/api/v2"
    
    def fetch_indicators(self) -> List[Dict]:
        """Fetch indicators from AbuseIPDB."""
        try:
            headers = {
                'Key': self.api_key,
                'Accept': 'application/json'
            }
            
            # Fetch blacklisted IPs
            response = requests.get(
                f"{self.base_url}/blacklist",
                headers=headers,
                params={'limit': 10000},
                timeout=30
            )
            response.raise_for_status()
            
            data = response.json()
            return data.get('data', [])
            
        except Exception as e:
            logger.error(f"Failed to fetch AbuseIPDB indicators: {e}")
            return []
    
    def parse_indicator(self, indicator_data: Dict) -> Optional[Dict]:
        """Parse AbuseIPDB indicator."""
        try:
            return {
                'indicator': indicator_data.get('ipAddress'),
                'indicator_type': 'IP',
                'threat_score': min(1.0, indicator_data.get('abuseConfidencePercentage', 0) / 100),
                'source': 'AbuseIPDB',
                'first_seen': indicator_data.get('lastReportedAt'),
                'tags': ['abuse', 'malicious']
            }
        except Exception as e:
            logger.error(f"Failed to parse AbuseIPDB indicator: {e}")
            return None

class ThreatIntelligenceManager:
    """Manages multiple threat intelligence feeds."""
    
    def __init__(self):
        self.feeds: List[ThreatIntelligenceFeed] = []
        self.indicator_cache: Dict[str, Dict] = {}
        self.last_update: Optional[datetime] = None
        self.update_interval = timedelta(hours=1)
    
    def add_feed(self, feed: ThreatIntelligenceFeed):
        """Add threat intelligence feed."""
        self.feeds.append(feed)
        logger.info(f"Added threat intelligence feed: {feed.__class__.__name__}")
    
    def update_indicators(self) -> int:
        """Update indicators from all feeds.
        
        Returns:
            Number of indicators updated
        """
        total_indicators = 0
        
        for feed in self.feeds:
            try:
                logger.info(f"Fetching indicators from {feed.__class__.__name__}")
                indicators = feed.fetch_indicators()
                
                for indicator_data in indicators:
                    parsed = feed.parse_indicator(indicator_data)
                    if parsed:
                        cache_key = f"{parsed['indicator_type']}:{parsed['indicator']}"
                        self.indicator_cache[cache_key] = parsed
                        total_indicators += 1
                
            except Exception as e:
                logger.error(f"Failed to update feed {feed.__class__.__name__}: {e}")
        
        self.last_update = datetime.now()
        logger.info(f"Updated {total_indicators} indicators from {len(self.feeds)} feeds")
        return total_indicators
    
    def lookup_indicator(self, indicator: str, indicator_type: str = "IP") -> Optional[Dict]:
        """Look up indicator in cache.
        
        Args:
            indicator: Indicator value
            indicator_type: Type of indicator
            
        Returns:
            Indicator data if found, None otherwise
        """
        cache_key = f"{indicator_type}:{indicator}"
        return self.indicator_cache.get(cache_key)
    
    def should_update(self) -> bool:
        """Check if indicators should be updated."""
        if self.last_update is None:
            return True
        return datetime.now() - self.last_update > self.update_interval
    
    def get_statistics(self) -> Dict:
        """Get TI manager statistics."""
        return {
            'total_indicators': len(self.indicator_cache),
            'feeds_count': len(self.feeds),
            'last_update': self.last_update.isoformat() if self.last_update else None,
            'indicator_types': {
                'IP': len([k for k in self.indicator_cache.keys() if k.startswith('IP:')]),
                'Domain': len([k for k in self.indicator_cache.keys() if k.startswith('Domain:')]),
                'Hash': len([k for k in self.indicator_cache.keys() if k.startswith('Hash:')])
            }
        }


# Example usage
if __name__ == "__main__":
    ti_manager = ThreatIntelligenceManager()
    
    # Add feeds (in production, use actual API keys)
    # abuse_feed = AbuseIPDBFeed(api_key="your_api_key_here")
    # ti_manager.add_feed(abuse_feed)
    
    # Update indicators
    if ti_manager.should_update():
        ti_manager.update_indicators()
    
    # Lookup indicator
    result = ti_manager.lookup_indicator("192.168.1.100", "IP")
    if result:
        print(f"Found threat indicator: {json.dumps(result, indent=2)}")
    
    # Get statistics
    stats = ti_manager.get_statistics()
    print(f"\nTI Statistics: {json.dumps(stats, indent=2)}")

#!/usr/bin/env python3
"""
SOC Metrics and Reporting System
Production-ready metrics collection and reporting
"""

from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from enum import Enum
import logging
import json
from collections import defaultdict

logger = logging.getLogger(__name__)

class MetricType(Enum):
    """Metric types."""
    COUNTER = "counter"
    GAUGE = "gauge"
    HISTOGRAM = "histogram"
    TIMER = "timer"

@dataclass
class SOCMetric:
    """SOC metric data point."""
    name: str
    value: float
    metric_type: MetricType
    timestamp: datetime
    tags: Dict[str, str] = None
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            'name': self.name,
            'value': self.value,
            'type': self.metric_type.value,
            'timestamp': self.timestamp.isoformat(),
            'tags': self.tags or {}
        }

class SOCMetricsCollector:
    """Collects and aggregates SOC metrics."""
    
    def __init__(self):
        self.metrics: List[SOCMetric] = []
        self.counters: Dict[str, float] = defaultdict(float)
        self.gauges: Dict[str, float] = {}
        self.timers: Dict[str, List[float]] = defaultdict(list)
    
    def increment_counter(self, name: str, value: float = 1.0, tags: Dict[str, str] = None):
        """Increment counter metric."""
        self.counters[name] += value
        self.metrics.append(SOCMetric(
            name=name,
            value=value,
            metric_type=MetricType.COUNTER,
            timestamp=datetime.now(),
            tags=tags
        ))
    
    def set_gauge(self, name: str, value: float, tags: Dict[str, str] = None):
        """Set gauge metric."""
        self.gauges[name] = value
        self.metrics.append(SOCMetric(
            name=name,
            value=value,
            metric_type=MetricType.GAUGE,
            timestamp=datetime.now(),
            tags=tags
        ))
    
    def record_timer(self, name: str, duration: float, tags: Dict[str, str] = None):
        """Record timer metric."""
        self.timers[name].append(duration)
        self.metrics.append(SOCMetric(
            name=name,
            value=duration,
            metric_type=MetricType.TIMER,
            timestamp=datetime.now(),
            tags=tags
        ))
    
    def get_counter(self, name: str) -> float:
        """Get counter value."""
        return self.counters.get(name, 0.0)
    
    def get_gauge(self, name: str) -> Optional[float]:
        """Get gauge value."""
        return self.gauges.get(name)
    
    def get_timer_stats(self, name: str) -> Dict:
        """Get timer statistics."""
        values = self.timers.get(name, [])
        if not values:
            return {}
        
        return {
            'count': len(values),
            'min': min(values),
            'max': max(values),
            'mean': sum(values) / len(values),
            'p95': sorted(values)[int(len(values) * 0.95)] if values else 0,
            'p99': sorted(values)[int(len(values) * 0.99)] if values else 0
        }
    
    def generate_report(self, time_range: timedelta = timedelta(hours=24)) -> Dict:
        """Generate comprehensive SOC metrics report.
        
        Args:
            time_range: Time range for report
            
        Returns:
            Report dictionary
        """
        cutoff_time = datetime.now() - time_range
        recent_metrics = [m for m in self.metrics if m.timestamp >= cutoff_time]
        
        # Calculate key metrics
        total_alerts = self.get_counter('alerts.received')
        processed_alerts = self.get_counter('alerts.processed')
        false_positives = self.get_counter('alerts.false_positive')
        resolved_alerts = self.get_counter('alerts.resolved')
        
        # Calculate rates
        false_positive_rate = (false_positives / processed_alerts * 100) if processed_alerts > 0 else 0
        resolution_rate = (resolved_alerts / processed_alerts * 100) if processed_alerts > 0 else 0
        
        # Processing time stats
        processing_time_stats = self.get_timer_stats('alert.processing_time')
        
        report = {
            'report_period': {
                'start': cutoff_time.isoformat(),
                'end': datetime.now().isoformat()
            },
            'alert_metrics': {
                'total_received': total_alerts,
                'total_processed': processed_alerts,
                'false_positives': false_positives,
                'resolved': resolved_alerts,
                'false_positive_rate': round(false_positive_rate, 2),
                'resolution_rate': round(resolution_rate, 2)
            },
            'performance_metrics': {
                'avg_processing_time': processing_time_stats.get('mean', 0),
                'p95_processing_time': processing_time_stats.get('p95', 0),
                'p99_processing_time': processing_time_stats.get('p99', 0)
            },
            'current_state': {
                'active_alerts': self.get_gauge('alerts.active') or 0,
                'queue_size': self.get_gauge('alerts.queue_size') or 0
            }
        }
        
        return report
    
    def export_metrics(self, format: str = "json") -> str:
        """Export metrics in specified format.
        
        Args:
            format: Export format (json, csv)
            
        Returns:
            Exported metrics string
        """
        if format == "json":
            return json.dumps({
                'counters': dict(self.counters),
                'gauges': dict(self.gauges),
                'timers': {k: self.get_timer_stats(k) for k in self.timers.keys()}
            }, indent=2)
        else:
            # CSV format
            lines = ["name,value,type,timestamp"]
            for metric in self.metrics:
                lines.append(f"{metric.name},{metric.value},{metric.metric_type.value},{metric.timestamp.isoformat()}")
            return "\n".join(lines)


# Example usage
if __name__ == "__main__":
    collector = SOCMetricsCollector()
    
    # Record some metrics
    collector.increment_counter('alerts.received')
    collector.increment_counter('alerts.processed')
    collector.set_gauge('alerts.active', 5)
    collector.record_timer('alert.processing_time', 1.5)
    collector.record_timer('alert.processing_time', 2.1)
    collector.record_timer('alert.processing_time', 0.8)
    
    # Generate report
    report = collector.generate_report()
    print("SOC Metrics Report:")
    print(json.dumps(report, indent=2))
    
    # Export metrics
    print("\nExported Metrics:")
    print(collector.export_metrics())

#!/usr/bin/env python3
"""
Unit tests for SOC Workflow
Comprehensive test coverage with pytest
"""

import pytest
from datetime import datetime
from soc_workflow import (
    SOCWorkflow, SOCAlert, AlertSeverity, AlertStatus,
    ThreatIntelligenceProvider, ThreatIndicator
)

class TestSOCWorkflow:
    """Unit tests for SOC workflow."""
    
    @pytest.fixture
    def workflow(self):
        """Create SOC workflow instance."""
        return SOCWorkflow()
    
    @pytest.fixture
    def sample_alert(self):
        """Create sample alert."""
        return SOCAlert(
            alert_id="TEST-001",
            title="Test Alert",
            severity=AlertSeverity.MEDIUM,
            source="Test",
            timestamp=datetime.now(),
            description="Test description",
            indicators=["IP: 192.168.1.1"]
        )
    
    def test_receive_alert(self, workflow, sample_alert):
        """Test alert reception."""
        result = workflow.receive_alert(sample_alert)
        assert result is True
        assert sample_alert.alert_id in workflow.alerts
    
    def test_prioritize_alert(self, workflow, sample_alert):
        """Test alert prioritization."""
        priority = workflow.prioritize_alert(sample_alert)
        assert priority == 3  # MEDIUM severity
        
        sample_alert.severity = AlertSeverity.CRITICAL
        priority = workflow.prioritize_alert(sample_alert)
        assert priority == 1  # CRITICAL severity
    
    def test_enrich_alert(self, workflow, sample_alert):
        """Test alert enrichment."""
        enriched = workflow.enrich_alert(sample_alert)
        assert enriched.status == AlertStatus.ENRICHED
        assert 'enrichment_data' in enriched.enrichment_data
    
    def test_analyze_alert(self, workflow, sample_alert):
        """Test alert analysis."""
        enriched = workflow.enrich_alert(sample_alert)
        analysis = workflow.analyze_alert(enriched)
        
        assert 'threat_score' in analysis
        assert 'recommended_action' in analysis
        assert 0.0 <= analysis['threat_score'] <= 1.0
    
    def test_process_alert(self, workflow, sample_alert):
        """Test complete alert processing."""
        workflow.receive_alert(sample_alert)
        result = workflow.process_alert(sample_alert)
        
        assert result['status'] == 'processed'
        assert 'threat_score' in result
        assert 'response_action' in result
    
    def test_mark_false_positive(self, workflow, sample_alert):
        """Test marking alert as false positive."""
        workflow.receive_alert(sample_alert)
        result = workflow.mark_false_positive(sample_alert.alert_id, "Test notes")
        
        assert result is True
        assert workflow.alerts[sample_alert.alert_id].status == AlertStatus.FALSE_POSITIVE
    
    def test_resolve_alert(self, workflow, sample_alert):
        """Test alert resolution."""
        workflow.receive_alert(sample_alert)
        result = workflow.resolve_alert(sample_alert.alert_id, "Resolved")
        
        assert result is True
        assert workflow.alerts[sample_alert.alert_id].status == AlertStatus.RESOLVED
    
    def test_get_metrics(self, workflow, sample_alert):
        """Test metrics retrieval."""
        workflow.receive_alert(sample_alert)
        workflow.process_alert(sample_alert)
        
        metrics = workflow.get_metrics()
        assert 'total_alerts' in metrics
        assert 'processed' in metrics
        assert metrics['processed'] >= 1


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

#!/usr/bin/env python3
"""
SOC Workflow Cleanup
Production-ready cleanup and resource management
"""

import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class SOCWorkflowCleanup:
    """Handles cleanup operations for SOC workflow."""
    
    def __init__(self, workflow):
        """Initialize cleanup handler.
        
        Args:
            workflow: SOCWorkflow instance to clean up
        """
        self.workflow = workflow
    
    def cleanup_old_alerts(self, days: int = 90):
        """Remove alerts older than specified days.
        
        Args:
            days: Number of days to keep alerts
        """
        cutoff_date = datetime.now() - timedelta(days=days)
        removed_count = 0
        
        alert_ids_to_remove = [
            alert_id for alert_id, alert in self.workflow.alerts.items()
            if alert.created_at < cutoff_date and alert.status in [AlertStatus.RESOLVED, AlertStatus.FALSE_POSITIVE]
        ]
        
        for alert_id in alert_ids_to_remove:
            del self.workflow.alerts[alert_id]
            removed_count += 1
        
        logger.info(f"Cleaned up {removed_count} old alerts")
        return removed_count
    
    def cleanup(self):
        """Perform complete cleanup."""
        logger.info("Starting SOC workflow cleanup")
        
        # Clean up old alerts
        self.cleanup_old_alerts()
        
        # Clean up workflow resources
        self.workflow.cleanup()
        
        logger.info("SOC workflow cleanup complete")