SOC Automation in 2026: Build Your First Playbook (Beginn...

#!/usr/bin/env python3
"""
SOC Automation Playbook Framework
Production-ready playbook implementation
"""

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

class PlaybookStatus(Enum):
    DRAFT = "draft"
    ACTIVE = "active"
    PAUSED = "paused"
    DEPRECATED = "deprecated"

@dataclass
class PlaybookStep:
    step_id: str
    name: str
    action: str
    condition: Optional[str] = None
    on_failure: str = "continue"

@dataclass
class Playbook:
    playbook_id: str
    name: str
    description: str
    trigger: str
    steps: List[PlaybookStep]
    status: PlaybookStatus

class PlaybookEngine:
    """SOC automation playbook engine."""
    
    def __init__(self):
        self.playbooks: Dict[str, Playbook] = {}
        self.execution_history: List[Dict] = []
    
    def create_playbook(self, playbook: Playbook) -> bool:
        """Create new playbook."""
        try:
            self.playbooks[playbook.playbook_id] = playbook
            return True
        except Exception as e:
            print(f"Failed to create playbook: {e}")
            return False
    
    def execute_playbook(self, playbook_id: str, context: Dict) -> Dict:
        """Execute playbook with context."""
        playbook = self.playbooks.get(playbook_id)
        if not playbook:
            return {'error': 'Playbook not found'}
        
        if playbook.status != PlaybookStatus.ACTIVE:
            return {'error': 'Playbook not active'}
        
        results = []
        for step in playbook.steps:
            # Check condition
            if step.condition and not self.evaluate_condition(step.condition, context):
                continue
            
            # Execute step
            try:
                result = self.execute_step(step, context)
                results.append({
                    'step_id': step.step_id,
                    'status': 'success',
                    'result': result
                })
            except Exception as e:
                results.append({
                    'step_id': step.step_id,
                    'status': 'failed',
                    'error': str(e)
                })
                
                if step.on_failure == "stop":
                    break
        
        execution_record = {
            'playbook_id': playbook_id,
            'timestamp': datetime.now(),
            'context': context,
            'results': results
        }
        self.execution_history.append(execution_record)
        
        return execution_record
    
    def execute_step(self, step: PlaybookStep, context: Dict) -> Dict:
        """Execute individual playbook step."""
        # Simplified - would execute actual actions
        return {'action': step.action, 'completed': True}
    
    def evaluate_condition(self, condition: str, context: Dict) -> bool:
        """Evaluate playbook condition."""
        # Simplified condition evaluation
        return True

# Usage
engine = PlaybookEngine()

playbook = Playbook(
    playbook_id="PB-001",
    name="Alert Triage Playbook",
    description="Automated alert triage and enrichment",
    trigger="alert_received",
    steps=[
        PlaybookStep("step1", "Enrich Alert", "enrich_with_threat_intel"),
        PlaybookStep("step2", "Check Severity", "assess_severity", condition="severity == 'high'"),
        PlaybookStep("step3", "Create Ticket", "create_incident_ticket")
    ],
    status=PlaybookStatus.ACTIVE
)

engine.create_playbook(playbook)
result = engine.execute_playbook("PB-001", {'alert_id': 'ALERT-001'})
print(f"Playbook executed: {result}")

#!/usr/bin/env python3
"""
Advanced SOC Automation Playbook Framework
Production-ready playbook engine with comprehensive actions
"""

from typing import List, Dict, Optional, Callable, Any
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime, timedelta
import logging
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class PlaybookStatus(Enum):
    """Playbook status."""
    DRAFT = "draft"
    ACTIVE = "active"
    PAUSED = "paused"
    DEPRECATED = "deprecated"

class StepStatus(Enum):
    """Step execution status."""
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"

@dataclass
class PlaybookStep:
    """Playbook step definition."""
    step_id: str
    name: str
    action: str
    parameters: Dict[str, Any] = field(default_factory=dict)
    condition: Optional[str] = None
    on_failure: str = "continue"  # continue, stop, retry
    retry_count: int = 0
    timeout: Optional[int] = None
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)

@dataclass
class Playbook:
    """Playbook definition."""
    playbook_id: str
    name: str
    description: str
    trigger: str
    steps: List[PlaybookStep]
    status: PlaybookStatus = PlaybookStatus.DRAFT
    version: str = "1.0"
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            **asdict(self),
            'status': self.status.value,
            'created_at': self.created_at.isoformat(),
            'updated_at': self.updated_at.isoformat()
        }

@dataclass
class StepExecution:
    """Step execution result."""
    step_id: str
    status: StepStatus
    result: Optional[Dict] = None
    error: Optional[str] = None
    execution_time: float = 0.0
    timestamp: datetime = field(default_factory=datetime.now)
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            **asdict(self),
            'status': self.status.value,
            'timestamp': self.timestamp.isoformat()
        }

@dataclass
class PlaybookExecution:
    """Playbook execution record."""
    execution_id: str
    playbook_id: str
    status: str
    steps: List[StepExecution]
    context: Dict
    started_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None
    duration: Optional[float] = None
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return {
            **asdict(self),
            'started_at': self.started_at.isoformat(),
            'completed_at': self.completed_at.isoformat() if self.completed_at else None
        }

class PlaybookEngine:
    """Production-ready SOC automation playbook engine."""
    
    def __init__(self):
        self.playbooks: Dict[str, Playbook] = {}
        self.executions: Dict[str, PlaybookExecution] = {}
        self.actions: Dict[str, Callable] = {}
        self._register_default_actions()
    
    def _register_default_actions(self):
        """Register default automation actions."""
        self.actions['enrich_alert'] = self._enrich_alert
        self.actions['create_ticket'] = self._create_ticket
        self.actions['isolate_endpoint'] = self._isolate_endpoint
        self.actions['block_ip'] = self._block_ip
        self.actions['send_notification'] = self._send_notification
        self.actions['query_threat_intel'] = self._query_threat_intel
        self.actions['escalate'] = self._escalate
    
    def register_action(self, action_name: str, action_func: Callable):
        """Register custom action.
        
        Args:
            action_name: Name of the action
            action_func: Function to execute
        """
        self.actions[action_name] = action_func
        logger.info(f"Registered action: {action_name}")
    
    def create_playbook(self, playbook: Playbook) -> bool:
        """Create new playbook.
        
        Args:
            playbook: Playbook to create
            
        Returns:
            True if successful
        """
        try:
            if playbook.playbook_id in self.playbooks:
                logger.warning(f"Playbook {playbook.playbook_id} already exists, updating")
            
            playbook.updated_at = datetime.now()
            self.playbooks[playbook.playbook_id] = playbook
            logger.info(f"Playbook created: {playbook.playbook_id}")
            return True
            
        except Exception as e:
            logger.error(f"Failed to create playbook: {e}", exc_info=True)
            return False
    
    def execute_playbook(self, playbook_id: str, context: Dict) -> PlaybookExecution:
        """Execute playbook with context.
        
        Args:
            playbook_id: Playbook ID to execute
            context: Execution context
            
        Returns:
            PlaybookExecution record
        """
        playbook = self.playbooks.get(playbook_id)
        if not playbook:
            raise ValueError(f"Playbook {playbook_id} not found")
        
        if playbook.status != PlaybookStatus.ACTIVE:
            raise ValueError(f"Playbook {playbook_id} is not active")
        
        execution_id = f"EXEC-{len(self.executions)+1}-{datetime.now().strftime('%Y%m%d%H%M%S')}"
        started_at = datetime.now()
        
        step_executions = []
        
        try:
            for step in playbook.steps:
                step_exec = self._execute_step(step, context)
                step_executions.append(step_exec)
                
                if step_exec.status == StepStatus.FAILED:
                    if step.on_failure == "stop":
                        break
                    elif step.on_failure == "retry" and step.retry_count > 0:
                        # Retry logic would go here
                        pass
            
            completed_at = datetime.now()
            duration = (completed_at - started_at).total_seconds()
            
            execution = PlaybookExecution(
                execution_id=execution_id,
                playbook_id=playbook_id,
                status="completed",
                steps=step_executions,
                context=context,
                started_at=started_at,
                completed_at=completed_at,
                duration=duration
            )
            
            self.executions[execution_id] = execution
            logger.info(f"Playbook {playbook_id} executed successfully: {execution_id}")
            return execution
            
        except Exception as e:
            logger.error(f"Playbook execution failed: {e}", exc_info=True)
            completed_at = datetime.now()
            duration = (completed_at - started_at).total_seconds()
            
            execution = PlaybookExecution(
                execution_id=execution_id,
                playbook_id=playbook_id,
                status="failed",
                steps=step_executions,
                context=context,
                started_at=started_at,
                completed_at=completed_at,
                duration=duration
            )
            self.executions[execution_id] = execution
            raise
    
    def _execute_step(self, step: PlaybookStep, context: Dict) -> StepExecution:
        """Execute individual playbook step.
        
        Args:
            step: Step to execute
            context: Execution context
            
        Returns:
            StepExecution result
        """
        start_time = datetime.now()
        
        # Check condition
        if step.condition and not self._evaluate_condition(step.condition, context):
            return StepExecution(
                step_id=step.step_id,
                status=StepStatus.SKIPPED,
                timestamp=start_time
            )
        
        # Execute action
        if step.action not in self.actions:
            return StepExecution(
                step_id=step.step_id,
                status=StepStatus.FAILED,
                error=f"Action {step.action} not found",
                timestamp=start_time
            )
        
        try:
            action_func = self.actions[step.action]
            result = action_func(step.parameters, context)
            
            execution_time = (datetime.now() - start_time).total_seconds()
            
            return StepExecution(
                step_id=step.step_id,
                status=StepStatus.SUCCESS,
                result=result,
                execution_time=execution_time,
                timestamp=start_time
            )
            
        except Exception as e:
            execution_time = (datetime.now() - start_time).total_seconds()
            logger.error(f"Step {step.step_id} failed: {e}", exc_info=True)
            
            return StepExecution(
                step_id=step.step_id,
                status=StepStatus.FAILED,
                error=str(e),
                execution_time=execution_time,
                timestamp=start_time
            )
    
    def _evaluate_condition(self, condition: str, context: Dict) -> bool:
        """Evaluate playbook condition.
        
        Args:
            condition: Condition expression
            context: Execution context
            
        Returns:
            True if condition is met
        """
        # Simplified condition evaluation
        # In production, would use proper expression evaluator
        try:
            # Simple variable substitution
            for key, value in context.items():
                condition = condition.replace(f"{{{{{key}}}}}", str(value))
            
            # Basic evaluation (simplified - use safe eval in production)
            return eval(condition)
        except:
            return False
    
    # Default action implementations
    def _enrich_alert(self, parameters: Dict, context: Dict) -> Dict:
        """Enrich alert with threat intelligence."""
        alert_id = parameters.get('alert_id') or context.get('alert_id')
        logger.info(f"Enriching alert: {alert_id}")
        return {'enriched': True, 'alert_id': alert_id}
    
    def _create_ticket(self, parameters: Dict, context: Dict) -> Dict:
        """Create incident ticket."""
        title = parameters.get('title', 'Security Incident')
        logger.info(f"Creating ticket: {title}")
        return {'ticket_id': 'TICKET-001', 'status': 'created'}
    
    def _isolate_endpoint(self, parameters: Dict, context: Dict) -> Dict:
        """Isolate endpoint."""
        endpoint_id = parameters.get('endpoint_id')
        logger.info(f"Isolating endpoint: {endpoint_id}")
        return {'endpoint_id': endpoint_id, 'isolated': True}
    
    def _block_ip(self, parameters: Dict, context: Dict) -> Dict:
        """Block IP address."""
        ip_address = parameters.get('ip_address')
        logger.info(f"Blocking IP: {ip_address}")
        return {'ip_address': ip_address, 'blocked': True}
    
    def _send_notification(self, parameters: Dict, context: Dict) -> Dict:
        """Send notification."""
        message = parameters.get('message', 'Security alert')
        logger.info(f"Sending notification: {message}")
        return {'sent': True}
    
    def _query_threat_intel(self, parameters: Dict, context: Dict) -> Dict:
        """Query threat intelligence."""
        indicator = parameters.get('indicator')
        logger.info(f"Querying threat intel: {indicator}")
        return {'indicator': indicator, 'malicious': False}
    
    def _escalate(self, parameters: Dict, context: Dict) -> Dict:
        """Escalate to analyst."""
        reason = parameters.get('reason', 'Manual review required')
        logger.info(f"Escalating: {reason}")
        return {'escalated': True, 'reason': reason}
    
    def get_execution_history(self, playbook_id: Optional[str] = None, limit: int = 100) -> List[PlaybookExecution]:
        """Get execution history.
        
        Args:
            playbook_id: Optional playbook ID to filter
            limit: Maximum number of results
            
        Returns:
            List of executions
        """
        executions = list(self.executions.values())
        
        if playbook_id:
            executions = [e for e in executions if e.playbook_id == playbook_id]
        
        executions.sort(key=lambda x: x.started_at, reverse=True)
        return executions[:limit]
    
    def get_statistics(self) -> Dict:
        """Get playbook statistics.
        
        Returns:
            Statistics dictionary
        """
        total_executions = len(self.executions)
        successful_executions = len([e for e in self.executions.values() if e.status == "completed"])
        failed_executions = total_executions - successful_executions
        
        return {
            'total_playbooks': len(self.playbooks),
            'active_playbooks': len([p for p in self.playbooks.values() if p.status == PlaybookStatus.ACTIVE]),
            'total_executions': total_executions,
            'successful_executions': successful_executions,
            'failed_executions': failed_executions,
            'success_rate': successful_executions / total_executions if total_executions > 0 else 0.0
        }
    
    def cleanup(self):
        """Clean up resources."""
        logger.info("Cleaning up playbook engine resources")

# Example usage
if __name__ == "__main__":
    engine = PlaybookEngine()
    
    # Create playbook
    playbook = Playbook(
        playbook_id="PB-001",
        name="Malware Detection Response",
        description="Automated response to malware detection",
        trigger="malware_detected",
        steps=[
            PlaybookStep(
                step_id="step1",
                name="Enrich Alert",
                action="enrich_alert",
                parameters={'alert_id': '{{alert_id}}'}
            ),
            PlaybookStep(
                step_id="step2",
                name="Isolate Endpoint",
                action="isolate_endpoint",
                parameters={'endpoint_id': '{{endpoint_id}}'},
                condition="{{severity}} == 'high'"
            ),
            PlaybookStep(
                step_id="step3",
                name="Create Ticket",
                action="create_ticket",
                parameters={'title': 'Malware Detected'}
            ),
            PlaybookStep(
                step_id="step4",
                name="Send Notification",
                action="send_notification",
                parameters={'message': 'Malware incident created'}
            )
        ],
        status=PlaybookStatus.ACTIVE
    )
    
    engine.create_playbook(playbook)
    
    # Execute playbook
    context = {
        'alert_id': 'ALERT-001',
        'endpoint_id': 'ENDPOINT-001',
        'severity': 'high'
    }
    
    execution = engine.execute_playbook("PB-001", context)
    print(f"Playbook executed: {execution.execution_id}")
    print(f"Status: {execution.status}")
    print(f"Steps completed: {len([s for s in execution.steps if s.status == StepStatus.SUCCESS])}")
    
    # Get statistics
    stats = engine.get_statistics()
    print(f"Statistics: {json.dumps(stats, indent=2)}")

#!/usr/bin/env python3
"""
Unit tests for Playbook Engine
"""

import pytest
from datetime import datetime
from playbook_engine import (
    PlaybookEngine, Playbook, PlaybookStep, PlaybookStatus
)

class TestPlaybookEngine:
    """Tests for PlaybookEngine."""
    
    @pytest.fixture
    def engine(self):
        return PlaybookEngine()
    
    @pytest.fixture
    def sample_playbook(self):
        return Playbook(
            playbook_id="TEST-001",
            name="Test Playbook",
            description="Test",
            trigger="test",
            steps=[
                PlaybookStep("step1", "Test Step", "send_notification", {"message": "test"})
            ],
            status=PlaybookStatus.ACTIVE
        )
    
    def test_create_playbook(self, engine, sample_playbook):
        """Test playbook creation."""
        result = engine.create_playbook(sample_playbook)
        assert result is True
        assert "TEST-001" in engine.playbooks
    
    def test_execute_playbook(self, engine, sample_playbook):
        """Test playbook execution."""
        engine.create_playbook(sample_playbook)
        execution = engine.execute_playbook("TEST-001", {})
        
        assert execution.playbook_id == "TEST-001"
        assert execution.status in ["completed", "failed"]
    
    def test_statistics(self, engine):
        """Test statistics."""
        stats = engine.get_statistics()
        assert 'total_playbooks' in stats
        assert 'total_executions' in stats

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

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

import logging
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

class PlaybookEngineCleanup:
    """Handles cleanup operations for playbook engine."""
    
    def __init__(self, engine):
        """Initialize cleanup handler."""
        self.engine = engine
    
    def cleanup_old_executions(self, days: int = 90):
        """Remove executions older than specified days."""
        cutoff_date = datetime.now() - timedelta(days=days)
        initial_count = len(self.engine.executions)
        
        self.engine.executions = {
            exec_id: exec_record
            for exec_id, exec_record in self.engine.executions.items()
            if exec_record.started_at >= cutoff_date
        }
        
        removed = initial_count - len(self.engine.executions)
        logger.info(f"Cleaned up {removed} old executions")
        return removed
    
    def cleanup(self):
        """Perform complete cleanup."""
        logger.info("Starting playbook engine cleanup")
        
        # Clean up old executions
        self.cleanup_old_executions()
        
        # Clean up engine resources
        self.engine.cleanup()
        
        logger.info("Playbook engine cleanup complete")

Real-World Case Study

Challenge: SOC struggling with manual processes:

• 4-hour average response time
• High analyst workload
• Inconsistent procedures
• Human error common
• Limited scalability

Solution: Implemented SOC automation:

• Developed playbooks
• Automated workflows
• Integrated tools
• Standardized procedures
• Added decision logic

Results:

• 75% faster response: Automation reduces time
• 3x efficiency: Handle more incidents
• Zero human error: Automated steps reliable
• 24/7 operations: Automation runs continuously
• Cost savings: Reduced manual effort
• Scalability: Handle growth efficiently

FAQ

Q: How do I start with SOC automation?

A: Start with simple, repetitive tasks. Develop playbooks for common scenarios, automate alert enrichment, and gradually expand automation coverage.

Q: What tools do I need for automation?

A: SOAR platforms (Splunk SOAR, XSOAR, etc.), SIEM integration, security tools APIs, and orchestration capabilities.

Q: Should I automate everything?

A: No. Automate repetitive, well-defined tasks. Keep complex decisions and critical actions for human review. Balance automation with human oversight.

Conclusion

SOC automation improves efficiency and reduces response time. Develop playbooks, automate workflows, and continuously improve automation capabilities.

Action Steps

1. Identify automation opportunities
2. Develop playbooks
3. Automate workflows
4. Integrate tools
5. Test thoroughly
6. Deploy gradually
7. Monitor and improve

Related Topics

• SOC 2026
• XDR vs SIEM vs SOAR

Educational Use Only: This content is for educational purposes. Implement SOC automation to improve security operations.