incident-response

# Incident Response ## What I do I enable effective security incident handling including detection, analysis, containment, eradication, recovery, and post-incident activities. I provide frameworks for incident classification, response procedures, and documentation. ## When to use me - Responding to active security incidents - Building incident response playbooks - Establishing an incident response team - Conducting incident post-mortems - Setting up detection and alerting - Creating communication templates - Training incident responders - Improving detection capabilities ## Core Concepts - **Incident Lifecycle**: Detection, Analysis, Containment, Eradication, Recovery, Lessons Learned - **NIST Framework**: Identify, Protect, Detect, Respond, Recover - **SANS 6-Step**: Preparation, Identification, Containment, Eradication, Recovery, Lessons Learned - **IOCs**: Indicators of Compromise (IPs, domains, hashes, patterns) - **TTPs**: Tactics, Techniques, and Procedures (MITRE ATT&CK) - **Chain of Custody**: Evidence handling and documentation - **Escalation**: When and how to escalate incidents - **Communication**: Stakeholder updates and notifications - **Forensics**: Preserving and analyzing evidence - **Playbooks**: Pre-defined response procedures ## Code Examples ### Incident Manager ```python from enum import Enum from typing import Dict, List, Optional from dataclasses import dataclass, field from datetime import datetime import json class IncidentSeverity(Enum): LOW = 1 MEDIUM = 2 HIGH = 3 CRITICAL = 4 class IncidentStatus(Enum): NEW = "new" INVESTIGATING = "investigating" CONTAINED = "contained" ERADICATED = "eradicated" RECOVERED = "recovered" CLOSED = "closed" class IncidentType(Enum): MALWARE = "malware" PHISHING = "phishing" RANSOMWARE = "ransomware" DATA_BREACH = "data_breach" DDOS = "ddos" UNAUTHORIZED_ACCESS = "unauthorized_access" INSIDER_THREAT = "insider_threat" APT = "apt" OTHER = "other" @dataclass class IOC: type: str value: str source: str first_seen: datetime last_seen: datetime confidence: float tags: List[str] = field(default_factory=list) @dataclass class Incident: incident_id: str title: str description: str severity: IncidentSeverity status: IncidentStatus type: IncidentType created_at: datetime updated_at: datetime assigned_to: str affected_assets: List[str] = field(default_factory=list) iocs: List[IOC] = field(default_factory=list) timeline: List[Dict] = field(default_factory=list) containment_actions: List[str] = field(default_factory=list) eradication_actions: List[str] = field(default_factory=list) notes: List[str] = field(default_factory=list) affected_systems: List[str] = field(default_factory=list) data_breach: bool = False customers_affected: int = 0 class IncidentManager: def __init__(self): self.incidents: Dict[str, Incident] = {} self.ioc_database: Dict[str, List[str]] = {} def create_incident(self, title: str, description: str, severity: IncidentSeverity, incident_type: IncidentType, assigned_to: str = "") -> Incident: incident_id = f"INC-{datetime.now().strftime('%Y%m%d')}-{len(self.incidents) + 1:04d}" incident = Incident( incident_id=incident_id, title=title, description=description, severity=severity, status=IncidentStatus.NEW, type=incident_type, created_at=datetime.now(), updated_at=datetime.now(), assigned_to=assigned_to ) self.incidents[incident_id] = incident self._add_timeline_entry(incident, "Incident created") return incident def update_status(self, incident_id: str, status: IncidentStatus): if incident_id in self.incidents: self.incidents[incident_id].status = status self.incidents[incident_id].updated_at = datetime.now() self._add_timeline_entry( self.incidents[incident_id], f"Status changed to {status.value}" ) def add_ioc(self, incident_id: str, ioc: IOC): if incident_id in self.incidents: self.incidents[incident_id].iocs.append(ioc) self._add_timeline_entry( self.incidents[incident_id], f"IOC added: {ioc.type}={ioc.value}" ) if ioc.type not in self.ioc_database: self.ioc_database[ioc.type] = [] if ioc.value not in self.ioc_database[ioc.type]: self.ioc_database[ioc.type].append(ioc.value) def add_containment_action(self, incident_id: str, action: str): if incident_id in self.incidents: self.incidents[incident_id].containment_actions.append(action) self._add_timeline_entry( self.incidents[incident_id], f"Containment: {action}" ) def add_eradication_action(self, incident_id: str, action: str): if incident_id in self.incidents: self.incidents[incident_id].eradication_actions.append(action) self._add_timeline_entry( self.incidents[incident_id], f"Eradication: {action}" ) def add_note(self, incident_id: str, note: str, author: str = "analyst"): if incident_id in self.incidents: self.incidents[incident_id].notes.append(f"[{author}] {note}") self._add_timeline_entry( self.incidents[incident_id], f"Note added by {author}" ) def _add_timeline_entry(self, incident: Incident, action: str): incident.timeline.append({ "timestamp": datetime.now().isoformat(), "action": action, "actor": "system" }) def get_active_incidents(self) -> List[Incident]: active_statuses = [IncidentStatus.NEW, IncidentStatus.INVESTIGATING, IncidentStatus.CONTAINED, IncidentStatus.ERADICATED] return [i for i in self.incidents.values() if i.status in active_statuses] def get_incident_summary(self) -> Dict: return { "total": len(self.incidents), "by_status": {s.value: 0 for s in IncidentStatus}, "by_severity": {s.value: 0 for s in IncidentSeverity}, "by_type": {t.value: 0 for t in IncidentType} } ``` ### Playbook Executor ```python from typing import Dict, List, Callable, Optional from dataclasses import dataclass from enum import Enum from datetime import datetime class PlaybookPhase(Enum): PREPARATION = "preparation" DETECTION = "detection" ANALYSIS = "analysis" CONTAINMENT = "containment" ERADICATION = "eradication" RECOVERY = "recovery" POST_INCIDENT = "post_incident" @dataclass class PlaybookStep: step_id: str name: str description: str phase: PlaybookPhase automated: bool manual_instructions: str verification_steps: List[str] rollback_steps: List[str] estimated_time_minutes: int @dataclass class Playbook: playbook_id: str name: str description: str applicable_types: List[IncidentType] severity_range: tuple steps: List[PlaybookStep] created_date: datetime version: str class PlaybookExecutor: def __init__(self, incident_manager: IncidentManager): self.incident_manager = incident_manager self.playbooks: Dict[str, Playbook] = {} def create_ransomware_playbook(self) -> Playbook: steps = [ PlaybookStep( step_id="RANS-001", name="Isolate Affected Systems", description="Immediately isolate infected systems from network", phase=PlaybookPhase.CONTAINMENT, automated=True, manual_instructions="Pull network cables, disable WiFi adapters", verification_steps=["Verify no network connectivity", "Check physical isolation"], rollback_steps=["Document network config before changes", "Take screenshots"], estimated_time_minutes=5 ), PlaybookStep( step_id="RANS-002", name="Preserve Evidence", description="Capture memory and disk images for forensics", phase=PlaybookPhase.ANALYSIS, automated=True, manual_instructions="Use FTK Imager for disk, winpmem for memory", verification_steps=["Verify image hashes match", "Check storage integrity"], rollback_steps=[], estimated_time_minutes=30 ), PlaybookStep( step_id="RANS-003", name="Identify Ransomware Variant", description="Determine the specific ransomware family", phase=PlaybookPhase.ANALYSIS, automated=True, manual_instructions="Check file extensions, ransom note content, encryption pattern", verification_steps=["Cross-reference with known variants", "Check ID Ransomware service"], rollback_steps=[], estimated_time_minutes=15 ), PlaybookStep( step_id="RANS-004", name="Assess Scope", description="Determine extent of encryption across environment", phase=PlaybookPhase.ANALYSIS, automated=True, manual_instructions="Check shared drives, backup systems, cloud storage", verification_steps=["Document all affected paths", "Identify encryption timestamp"], rollback_steps=[], estimated_time_minutes=30 ), PlaybookStep( step_id="RANS-005", name="Check Backups", description="Verify backup integrity and availability", phase=PlaybookPhase.RECOVERY, automated=True, manual_instructions="Check offline backups, air-gapped copies, cloud backups", verification_steps=["Test restore process", "Verify backup timestamps"], rollback_steps=[], estimated_time_minutes=60 ), ] return Playbook( playbook_id="PB-RANSOMWARE-001", name="Ransomware Response", description="Playbook for responding to ransomware incidents", applicable_types=[IncidentType.RANSOMWARE], severity_range=(IncidentSeverity.HIGH, IncidentSeverity.CRITICAL), steps=steps, created_date=datetime.now(), version="1.0" ) def execute_playbook(self, playbook: Playbook, incident_id: str) -> Dict: incident = self.incident_manager.incidents.get(incident_id) if not incident: raise ValueError(f"Incident {incident_id} not found") execution_report = { "playbook_id": playbook.playbook_id, "incident_id": incident_id, "started_at": datetime.now().isoformat(), "steps_executed": [], "completed": [], "failed": [], "skipped": [] } for step in playbook.steps: if incident.severity not in playbook.severity_range: execution_report["skipped"].append({ "step_id": step.step_id, "reason": "Severity out of playbook scope" }) continue try: self.incident_manager.add_note( incident_id, f"Executing step: {step.name} ({step.step_id})" ) execution_report["steps_executed"].append({ "step_id": step.step_id, "name": step.name, "started_at": datetime.now().isoformat(), "status": "in_progress" }) self.incident_manager.add_note( incident_id, f"Step {step.step_id} completed: {step.description}" ) execution_report["completed"].append({ "step_id": step.step_id, "completed_at": datetime.now().isoformat() }) except Exception as e: execution_report["failed"].append({ "step_id": step.step_id, "error": str(e) }) execution_report["completed_at"] = datetime.now().isoformat() return execution_report ``` ### Alert Correlation Engine ```python from typing import Dict, List, Set from dataclasses import dataclass, field from datetime import datetime, timedelta from collections import defaultdict @dataclass class SecurityAlert: alert_id: str rule_name: str source_ip: str destination_ip: str timestamp: datetime severity: str description: str raw_data: Dict processed: bool = False correlated: bool = False class AlertCorrelator: def __init__(self, correlation_window_minutes: int = 15): self.correlation_window = timedelta(minutes=correlation_window_minutes) self.alerts: List[SecurityAlert] = [] self.ip_activity: Dict[str, List[SecurityAlert]] = defaultdict(list) self.alert_patterns: Dict[str, List[SecurityAlert]] = defaultdict(list) def ingest_alert(self, alert: SecurityAlert): self.alerts.append(alert) self.ip_activity[alert.source_ip].append(alert) self.ip_activity[alert.destination_ip].append(alert) self._update_patterns(alert) def _update_patterns(self, alert: SecurityAlert): pattern_key = f"{alert.rule_name}:{alert.severity}" self.alert_patterns[pattern_key].append(alert) def find_related_alerts(self, alert: SecurityAlert) -> List[SecurityAlert]: related = [] window_start = datetime.now() - self.correlation_window for ip in [alert.source_ip, alert.destination_ip]: for related_alert in self.ip_activity[ip]: if related_alert.alert_id != alert.alert_id: if related_alert.timestamp >= window_start: related.append(related_alert)