SOAR Playbook in Python: Incident Response Automation
A Security Orchestration, Automation and Response (SOAR) playbook is much more than a script: it is a formal, repeatable orchestration of actions in response to a security incident. When an alert triggers a playbook, a coordinated sequence of operations - data enrichment, malware analysis, automatic containment, notifications, and documentation - happens in seconds instead of hours, measurably reducing Mean Time to Respond (MTTR).
Recent industry data confirms the impact: Python-based implementations have achieved 2.5 minute MTTR with 92% accuracy, handling 500 incidents/hour. Cortex XSOAR, TheHive with Cortex, and platforms like Shuffle (open-source) allow implementing playbooks in Python with full access to the security ecosystem.
This article builds complete SOAR playbooks in Python for three common scenarios: phishing, endpoint malware, and brute force. For each, the entire flow is implemented: triage, enrichment, containment, escalation, and automatic documentation.
What You Will Learn
- Architecture of a modular SOAR playbook in Python
- Automatic enrichment: VirusTotal, Shodan, Active Directory
- Automatic containment: endpoint isolation, IP blocking, user disabling
- Integration with TheHive and Cortex for orchestration
- Playbook testing and versioning
- Patterns for human escalation with pre-populated context
Architecture of a Modular SOAR Playbook
A well-designed playbook is composed of independent functional blocks that can be reused across different playbooks. The modular structure facilitates testing, maintenance, and evolution over time.
The fundamental blocks of any SOAR playbook are:
- Trigger: which event starts the playbook (SIEM alert, suspicious email, user report)
- Enrichment: additional context collection (threat intel, asset info, user history)
- Decision: automatic triage logic (false positive? escalation? immediate containment?)
- Action: response actions (isolation, blocking, notification)
- Documentation: ticket update, playbook log, evidence collection
- Closure/Escalation: automatic closure or escalation to human analyst
# SOAR Playbook Base Framework
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
from enum import Enum
import logging
class PlaybookStatus(Enum):
RUNNING = "running"
COMPLETED = "completed"
ESCALATED = "escalated"
FAILED = "failed"
class ActionResult(Enum):
SUCCESS = "success"
FAILURE = "failure"
SKIPPED = "skipped"
ESCALATE = "escalate"
@dataclass
class PlaybookContext:
"""Shared context across all playbook steps."""
alert_id: str
alert_type: str
alert_data: dict
start_time: datetime = field(default_factory=datetime.now)
enrichment_data: dict = field(default_factory=dict)
actions_taken: list[dict] = field(default_factory=list)
evidence: list[dict] = field(default_factory=list)
escalation_reason: Optional[str] = None
status: PlaybookStatus = PlaybookStatus.RUNNING
def add_action(self, action: str, result: ActionResult, details: dict = None) -> None:
self.actions_taken.append({
'timestamp': datetime.now().isoformat(),
'action': action,
'result': result.value,
'details': details or {}
})
def set_escalation(self, reason: str) -> None:
self.escalation_reason = reason
self.status = PlaybookStatus.ESCALATED
class PlaybookStep(ABC):
def __init__(self, name: str):
self.name = name
self.logger = logging.getLogger(f"soar.{name}")
@abstractmethod
def execute(self, context: PlaybookContext) -> ActionResult:
pass
class Playbook:
def __init__(self, name: str, steps: list[PlaybookStep]):
self.name = name
self.steps = steps
self.logger = logging.getLogger(f"soar.playbook.{name}")
def run(self, alert_id: str, alert_type: str, alert_data: dict) -> PlaybookContext:
context = PlaybookContext(alert_id=alert_id, alert_type=alert_type, alert_data=alert_data)
for step in self.steps:
if context.status in [PlaybookStatus.ESCALATED, PlaybookStatus.FAILED]:
break
try:
result = step.execute(context)
context.add_action(step.name, result)
if result == ActionResult.ESCALATE:
context.status = PlaybookStatus.ESCALATED
break
except Exception as e:
self.logger.error(f"Step '{step}' failed: {e}")
context.add_action(step.name, ActionResult.FAILURE, {'error': str(e)})
context.status = PlaybookStatus.FAILED
break
if context.status == PlaybookStatus.RUNNING:
context.status = PlaybookStatus.COMPLETED
return contextPhishing Playbook: From Triage to Containment
Phishing is the most common scenario in SOCs. A mature playbook reduces response time from 45 minutes (manual) to less than 2 minutes (automated), analyzing URLs, attachments, verifying email headers, and isolating users who clicked.
# Complete Phishing Playbook
import httpx, re, base64
from email.parser import Parser
from email.policy import default as default_policy
class ExtractEmailArtifactsStep(PlaybookStep):
def __init__(self):
super().__init__("extract_email_artifacts")
def execute(self, context: PlaybookContext) -> ActionResult:
email_raw = context.alert_data.get('email_raw', '')
if not email_raw:
return ActionResult.SKIPPED
msg = Parser(policy=default_policy).parsestr(email_raw)
headers = {
'from': msg.get('From', ''),
'reply_to': msg.get('Reply-To', ''),
'dkim': 'DKIM-Signature' in msg,
'spf': 'Received-SPF' in email_raw,
}
body = msg.get_body(preferencelist=('plain', 'html'))
body_text = body.get_content() if body else ''
urls = re.findall(r'https?://[^\s<>"]+', body_text)
context.enrichment_data['email_headers'] = headers
context.enrichment_data['urls'] = list(set(urls))
context.add_evidence('extracted_urls', urls, 'email_parser')
return ActionResult.SUCCESS
class TriageDecisionStep(PlaybookStep):
def __init__(self):
super().__init__("triage_decision")
def execute(self, context: PlaybookContext) -> ActionResult:
vt_verdict = context.enrichment_data.get('vt_verdict', 'unknown')
headers = context.enrichment_data.get('email_headers', {})
risk_score = 0
if vt_verdict == 'malicious':
risk_score += 40
if not headers.get('dkim', True):
risk_score += 15
if not headers.get('spf', True):
risk_score += 15
context.enrichment_data['risk_score'] = risk_score
if risk_score >= 50:
context.enrichment_data['triage_result'] = 'high_risk'
return ActionResult.SUCCESS
elif risk_score >= 25:
context.enrichment_data['triage_result'] = 'medium_risk'
context.set_escalation(f"Risk score {risk_score}: requires human review")
return ActionResult.ESCALATE
else:
context.enrichment_data['triage_result'] = 'low_risk'
return ActionResult.SUCCESS
class ContainPhishingStep(PlaybookStep):
def __init__(self, exchange_client, ad_client):
super().__init__("contain_phishing")
self.exchange = exchange_client
self.ad = ad_client
def execute(self, context: PlaybookContext) -> ActionResult:
if context.enrichment_data.get('triage_result') != 'high_risk':
return ActionResult.SKIPPED
sender = context.alert_data.get('sender_email', '')
if sender:
self.exchange.add_to_blocklist(sender)
if context.alert_data.get('user_clicked_link', False):
victim = context.alert_data.get('recipient_email', '')
self.ad.force_password_reset(victim)
self.ad.revoke_sessions(victim)
return ActionResult.SUCCESSMalware Endpoint Playbook: Enrichment and Isolation
The malware endpoint playbook must respond more aggressively: endpoint isolation is critical to prevent lateral movement. But before isolation, it is necessary to collect as much forensic data as possible.
# Malware Endpoint Playbook
class CollectForensicDataStep(PlaybookStep):
def __init__(self, edr_client):
super().__init__("collect_forensic_data")
self.edr = edr_client
def execute(self, context: PlaybookContext) -> ActionResult:
endpoint = context.alert_data.get('endpoint_hostname', '')
forensics = {
'process_tree': self.edr.get_process_tree(endpoint, context.alert_data.get('process_name')),
'network_connections': self.edr.get_process_network(endpoint, context.alert_data.get('pid')),
'file_changes': self.edr.get_recent_file_changes(endpoint, hours_back=2),
'autoruns': self.edr.get_autoruns(endpoint)
}
context.enrichment_data['forensics'] = forensics
context.add_evidence('forensic_collection', forensics, 'edr')
return ActionResult.SUCCESS
class IsolateEndpointStep(PlaybookStep):
def __init__(self, edr_client, cmdb_client):
super().__init__("isolate_endpoint")
self.edr = edr_client
self.cmdb = cmdb_client
def execute(self, context: PlaybookContext) -> ActionResult:
verdict = context.enrichment_data.get('malware_verdict', 'unknown')
if verdict not in ['confirmed_malware', 'suspicious']:
return ActionResult.SKIPPED
endpoint = context.alert_data.get('endpoint_hostname', '')
asset_info = self.cmdb.get_asset_info(endpoint)
if asset_info.get('criticality') == 'critical':
context.set_escalation(
f"Endpoint '{endpoint}' is critical (type: {asset_info.get('type')}). "
f"Isolation requires manual approval."
)
return ActionResult.ESCALATE
isolation_result = self.edr.isolate_endpoint(
endpoint,
reason=f"Malware detection - alert {context.alert_id}",
allow_edr_communication=True
)
context.add_evidence('isolation', {
'endpoint': endpoint,
'result': isolation_result,
'timestamp': datetime.now().isoformat()
}, 'edr')
return ActionResult.SUCCESSAutomatic Documentation and Ticket Creation
A mature playbook automatically generates incident documentation, populating the ticket with all enrichment data, actions taken, and remediation recommendations. This eliminates the "reporting overhead" that often occupies 40% of analyst time.
# Automatic Documentation with Jinja2
import jinja2
INCIDENT_REPORT_TEMPLATE = """
## Incident Report - {{ context.alert_id }}
**Status**: {{ context.status.value }}
**Alert Type**: {{ context.alert_type }}
**Risk Score**: {{ context.enrichment_data.get('risk_score', 'N/A') }}
### Actions Taken
{% for action in context.actions_taken %}
- {{ action.timestamp }} - {{ action.action }}: {{ action.result }}
{% endfor %}
### Evidence Collected
{% for ev in context.evidence %}
- {{ ev.type }} (from {{ ev.source }})
{% endfor %}
{% if context.escalation_reason %}
### Escalation Required
**Reason**: {{ context.escalation_reason }}
{% endif %}
"""
class DocumentIncidentStep(PlaybookStep):
def __init__(self, thehive_client=None):
super().__init__("document_incident")
self.thehive = thehive_client
self.template = jinja2.Template(INCIDENT_REPORT_TEMPLATE)
def execute(self, context: PlaybookContext) -> ActionResult:
report = self.template.render(context=context)
context.enrichment_data['incident_report'] = report
if self.thehive:
case_id = context.alert_data.get('thehive_case_id')
if case_id:
self.thehive.update_case(case_id, description=report)
else:
self.thehive.create_case({
'title': f"[SOAR] {context.alert_type} - {context.alert_id}",
'description': report,
'severity': 2
})
return ActionResult.SUCCESSPlaybook Testing Framework
SOAR playbooks must be tested with the same rigor as application code. A faulty playbook in production can cause over-containment (isolating legitimate hosts) or under-containment (failing to act on real threats).
# Testing Framework for SOAR Playbooks
import pytest
from unittest.mock import MagicMock, patch
class MockExchangeClient:
def __init__(self):
self.blocked_senders = []
def add_to_blocklist(self, sender: str) -> None:
self.blocked_senders.append(sender)
def search_and_delete(self, sender: str, url_contains: str) -> dict:
return {"deleted_count": 3, "status": "success"}
class MockADClient:
def __init__(self):
self.password_resets = []
def force_password_reset(self, user: str) -> None:
self.password_resets.append(user)
def revoke_sessions(self, user: str) -> None:
pass
class TestPhishingPlaybook:
def setup_method(self):
self.exchange = MockExchangeClient()
self.ad = MockADClient()
def test_high_risk_triggers_containment(self):
"""High risk phishing must trigger containment."""
alert = {
"alert_id": "test-001",
"alert_type": "phishing",
"sender_email": "attacker@evil.com",
"recipient_email": "victim@company.com",
"user_clicked_link": True,
"email_raw": "From: attacker@evil.com\n\nClick: http://evil.com/phish"
}
playbook = Playbook("phishing_response", [
ExtractEmailArtifactsStep(),
TriageDecisionStep(),
ContainPhishingStep(self.exchange, self.ad),
])
context = playbook.run("test-001", "phishing", alert)
assert context.status in [PlaybookStatus.COMPLETED, PlaybookStatus.ESCALATED]
def test_low_risk_no_containment(self):
"""Low risk email must NOT trigger containment."""
alert = {
"alert_id": "test-002",
"alert_type": "phishing",
"sender_email": "news@legit.com",
"user_clicked_link": False,
"email_raw": "From: news@legit.com\nSubject: Newsletter\n\nHello!"
}
playbook = Playbook("phishing_response", [
ExtractEmailArtifactsStep(),
TriageDecisionStep(),
ContainPhishingStep(self.exchange, self.ad),
])
context = playbook.run("test-002", "phishing", alert)
assert len(self.exchange.blocked_senders) == 0
assert len(self.ad.password_resets) == 0Best Practice: Dry Run Mode
Before deploying to production, run every playbook in "dry run" mode for 2 weeks. Dry run executes all enrichment and decision steps but skips actual containment actions (isolation, blocks, password resets), only logging what it would have done. This allows calibrating triage logic without risk.
Critical: Always Check Asset Criticality Before Containment
Never automate isolation of assets without checking the CMDB first. Automatically isolating a production database or domain controller to respond to a false positive can cause a self-inflicted outage far more damaging than the original threat. Always require human approval for critical assets.
Conclusions and Key Takeaways
Python SOAR playbooks represent the frontier of modern incident response: they combine the flexibility of a general-purpose language with the richest security integration ecosystem available. The modular framework ensures reusability, testability, and maintainability over time.
Related deep dive: 05 - Threat Intelligence Ingestion: Processore Feed STIX/TAXII.
Related deep dive: 06 - Anomaly Detection Comportamentale: ML su Log Data.
Key Takeaways
- The modular architecture (Step + Playbook + Context) is the foundation of any enterprise SOAR
- Multi-source enrichment (VT, EDR, CMDB, AD) is essential for accurate triage decisions
- Automatic escalation for critical assets prevents collateral damage from over-containment
- Automatic documentation eliminates reporting overhead (40% of analyst time)
- Testing with mock clients ensures quality without impacting production systems
- Dry run mode is essential for calibrating triage logic before go-live
Related Articles
- Alert Triage Automation: Reducing MTTD with Graph Analysis
- Threat Intelligence Ingestion: STIX/TAXII Feed Processor
- Detection Engineering as a Discipline: From Scripts to Pipeline
Frequently asked questions
What is a SOAR playbook and how does it help in incident response?
A SOAR (Security Orchestration, Automation and Response) playbook is a formal, repeatable orchestration of actions in response to a security incident that automates the process of responding to threats.
How do Python-based implementations of SOAR playbooks compare to manual incident response processes?
Recent industry data confirms that Python-based implementations have achieved 2.5 minute Mean Time to Respond (MTTR) with 92% accuracy, compared to hours for manual processes.
What is the importance of testing a SOAR playbook before deploying it in production?
A faulty playbook can cause over-containment or under-containment, and therefore must be tested with the same rigor as application code using a "dry run" mode that simulates containment actions without actually taking them.
Why should human approval always be required for critical assets such as production databases or domain controllers?
Automatically isolating these assets can cause a self-inflicted outage far more damaging than the original threat, and therefore requires manual intervention to prevent unintended consequences.
Commenti
Caricamento commenti...
Sign in to comment