# WIA-SEC-020: Security Incident Response - Phases 2, 3, 4 Specification **Standard ID:** WIA-SEC-020 **Version:** 1.0 **Status:** ACTIVE **Last Updated:** 2025-12-25 --- ## Phase 2: Advanced Detection and Analysis ### 2.1 Threat Intelligence Integration #### 2.1.1 STIX/TAXII Integration **STIX (Structured Threat Information Expression):** ```json { "type": "indicator", "spec_version": "2.1", "id": "indicator--8e2e2d2b-17d4-4cbf-938f-98ee46b3cd3f", "created": "2025-12-25T10:00:00.000Z", "modified": "2025-12-25T10:00:00.000Z", "name": "Malicious IP Indicator", "description": "IP address associated with APT29 C2 infrastructure", "indicator_types": ["malicious-activity"], "pattern": "[ipv4-addr:value = '192.168.1.100']", "pattern_type": "stix", "valid_from": "2025-12-25T10:00:00Z", "kill_chain_phases": [ { "kill_chain_name": "mitre-attack", "phase_name": "command-and-control" } ] } ``` **TAXII Server Configuration:** ```python # TAXII 2.1 Client Implementation from taxii2client.v21 import Server, Collection # Connect to threat intelligence server server = Server("https://threatintel.example.com/taxii/", user="csirt", password="***") # Get available collections collections = server.api_root.collections # Poll for new indicators for collection in collections: if collection.title == "APT Indicators": indicators = collection.get_objects() for indicator in indicators: process_threat_indicator(indicator) ``` #### 2.1.2 MITRE ATT&CK Mapping **Technique Mapping:** ```python class MitreAttackMapper: """Map observed behaviors to MITRE ATT&CK techniques""" def __init__(self): self.technique_db = self.load_mitre_database() def map_behavior(self, observed_behavior): """ Map observed behavior to ATT&CK technique """ mappings = { "powershell_execution": { "technique": "T1059.001", "tactic": "Execution", "name": "PowerShell" }, "credential_dumping": { "technique": "T1003", "tactic": "Credential Access", "name": "OS Credential Dumping" }, "lateral_movement_rdp": { "technique": "T1021.001", "tactic": "Lateral Movement", "name": "Remote Desktop Protocol" }, "data_exfiltration_http": { "technique": "T1041", "tactic": "Exfiltration", "name": "Exfiltration Over C2 Channel" } } return mappings.get(observed_behavior, None) def build_attack_chain(self, observed_techniques): """ Reconstruct attack chain from observed techniques """ kill_chain = { "initial_access": [], "execution": [], "persistence": [], "privilege_escalation": [], "defense_evasion": [], "credential_access": [], "discovery": [], "lateral_movement": [], "collection": [], "exfiltration": [], "command_and_control": [] } for technique in observed_techniques: tactic = technique.get("tactic", "").lower().replace(" ", "_") if tactic in kill_chain: kill_chain[tactic].append(technique) return kill_chain ``` #### 2.1.3 Automated Threat Hunting **Hunt Hypothesis Development:** ```python class ThreatHuntingFramework: """Proactive threat hunting using hypothesis-driven approach""" def create_hunt_hypothesis(self, threat_intel): """ Create testable hunt hypothesis from threat intelligence """ hypothesis = { "id": f"HUNT-{generate_id()}", "threat_actor": threat_intel.get("actor"), "hypothesis": "", "ttps": [], "data_sources": [], "search_queries": [], "expected_evidence": [] } # Example: APT29 hunt if threat_intel.get("actor") == "APT29": hypothesis.update({ "hypothesis": "APT29 using compromised credentials for initial access", "ttps": ["T1078", "T1071", "T1059.001"], "data_sources": [ "Windows Event Logs (4624, 4625)", "PowerShell logs", "Network traffic to known C2" ], "search_queries": [ { "query": 'EventID=4624 AND LogonType=3 AND TargetUserName!=*$ AND NOT SourceNetworkAddress IN (known_admin_ips)', "expected_results": "Unusual remote logons from unexpected IPs" }, { "query": 'process_name=powershell.exe AND (command_line=*-encodedcommand* OR command_line=*-enc*)', "expected_results": "Encoded PowerShell commands" } ] }) return hypothesis def execute_hunt(self, hypothesis): """ Execute hunt queries and analyze results """ results = { "hypothesis_id": hypothesis["id"], "findings": [], "iocs_discovered": [], "recommended_actions": [] } for query in hypothesis["search_queries"]: findings = self.siem.search(query["query"]) if len(findings) > 0: results["findings"].append({ "query": query["query"], "hits": len(findings), "sample_events": findings[:10], "severity": self.assess_severity(findings) }) return results ``` ### 2.2 Behavioral Analytics #### 2.2.1 User and Entity Behavior Analytics (UEBA) **Anomaly Detection Models:** ```python import numpy as np from sklearn.ensemble import IsolationForest from sklearn.preprocessing import StandardScaler class UEBASystem: """User and Entity Behavior Analytics""" def __init__(self): self.model = IsolationForest(contamination=0.1, random_state=42) self.scaler = StandardScaler() def build_user_baseline(self, user_id, historical_data): """ Build baseline behavior profile for user """ features = self.extract_features(historical_data) baseline = { "user_id": user_id, "typical_login_times": self.analyze_login_patterns(features), "typical_locations": self.analyze_locations(features), "typical_data_access": self.analyze_data_access(features), "typical_network_activity": self.analyze_network(features), "peer_group": self.identify_peer_group(user_id) } return baseline def detect_anomalies(self, current_behavior, baseline): """ Detect deviations from baseline behavior """ anomalies = [] # Login time anomaly if self.is_unusual_time(current_behavior["login_time"], baseline["typical_login_times"]): anomalies.append({ "type": "UNUSUAL_LOGIN_TIME", "severity": "MEDIUM", "description": f"Login at {current_behavior['login_time']} outside typical hours" }) # Geo-location anomaly if current_behavior["location"] not in baseline["typical_locations"]: distance = self.calculate_distance( current_behavior["location"], baseline["typical_locations"][-1] ) if distance > 500: # km anomalies.append({ "type": "IMPOSSIBLE_TRAVEL", "severity": "HIGH", "description": f"Login from {current_behavior['location']}, {distance}km from last location" }) # Data access anomaly if current_behavior["files_accessed"] > baseline["typical_data_access"]["max"] * 3: anomalies.append({ "type": "EXCESSIVE_DATA_ACCESS", "severity": "HIGH", "description": "Accessing significantly more files than typical" }) return anomalies def calculate_risk_score(self, user_id, anomalies): """ Calculate overall risk score based on detected anomalies """ weights = { "UNUSUAL_LOGIN_TIME": 2, "IMPOSSIBLE_TRAVEL": 8, "EXCESSIVE_DATA_ACCESS": 7, "PRIVILEGE_ESCALATION": 9, "UNUSUAL_COMMAND": 6 } risk_score = sum(weights.get(a["type"], 1) for a in anomalies) if risk_score >= 15: return {"score": risk_score, "level": "CRITICAL"} elif risk_score >= 10: return {"score": risk_score, "level": "HIGH"} elif risk_score >= 5: return {"score": risk_score, "level": "MEDIUM"} else: return {"score": risk_score, "level": "LOW"} ``` #### 2.2.2 Network Traffic Analysis **Deep Packet Inspection:** ```python from scapy.all import * class NetworkBehaviorAnalyzer: """Analyze network traffic for suspicious patterns""" def analyze_dns_tunneling(self, packets): """ Detect DNS tunneling for data exfiltration """ dns_queries = [p for p in packets if p.haslayer(DNS) and p[DNS].qr == 0] suspicious = [] for query in dns_queries: domain = query[DNS].qd.qname.decode() # Check for suspicious patterns if len(domain) > 50: # Unusually long domain suspicious.append({ "type": "LONG_DNS_QUERY", "domain": domain, "length": len(domain), "src_ip": query[IP].src }) # Check for high entropy (encoded data) entropy = self.calculate_entropy(domain) if entropy > 4.5: suspicious.append({ "type": "HIGH_ENTROPY_DNS", "domain": domain, "entropy": entropy, "src_ip": query[IP].src }) return suspicious def detect_beaconing(self, connections): """ Detect C2 beaconing behavior """ # Group connections by destination conn_groups = {} for conn in connections: key = (conn["src_ip"], conn["dst_ip"], conn["dst_port"]) if key not in conn_groups: conn_groups[key] = [] conn_groups[key].append(conn["timestamp"]) beacons = [] for key, timestamps in conn_groups.items(): if len(timestamps) < 10: continue # Calculate intervals between connections intervals = [timestamps[i+1] - timestamps[i] for i in range(len(timestamps)-1)] # Check for regular intervals (beaconing) mean_interval = np.mean(intervals) std_interval = np.std(intervals) if std_interval / mean_interval < 0.1: # Very consistent intervals beacons.append({ "src_ip": key[0], "dst_ip": key[1], "dst_port": key[2], "interval": mean_interval, "confidence": "HIGH", "description": f"Regular beaconing every {mean_interval:.1f}s" }) return beacons ``` --- ## Phase 3: Advanced Containment and Eradication ### 3.1 Automated Response (SOAR) #### 3.1.1 Security Orchestration Platform **Playbook Automation:** ```yaml # SOAR Playbook: Ransomware Response name: "Ransomware Incident Response" trigger: type: "alert" conditions: - alert_type: "ransomware_detected" - severity: ["CRITICAL", "HIGH"] actions: - name: "immediate_isolation" type: "network_isolation" targets: "{{ affected_hosts }}" timeout: 60 on_failure: "notify_admin" - name: "snapshot_memory" type: "forensic_capture" capture_type: "memory" targets: "{{ affected_hosts }}" storage: "/forensics/{{ incident_id }}/" - name: "block_c2_communication" type: "firewall_rule" action: "block" destinations: "{{ ioc_ips }}" scope: "global" - name: "disable_accounts" type: "account_action" action: "disable" accounts: "{{ compromised_accounts }}" notify_users: false - name: "create_incident_ticket" type: "ticketing" system: "ServiceNow" priority: "P1" assignment_group: "CSIRT" description: "Ransomware detected on {{ affected_hosts | length }} hosts" - name: "notify_stakeholders" type: "notification" channels: ["email", "sms", "slack"] recipients: ["csirt-lead", "ciso", "cto"] template: "ransomware_alert" - name: "begin_recovery" type: "workflow" workflow_name: "ransomware_recovery" wait_for_approval: true approver: "csirt-lead" ``` **SOAR Integration Code:** ```python class SOAROrchestrator: """Security Orchestration, Automation and Response""" def __init__(self): self.integrations = { "firewall": FirewallAPI(), "edr": EDRPlatform(), "siem": SIEMConnector(), "ticketing": ServiceNowAPI(), "communication": SlackAPI() } def execute_playbook(self, playbook, incident): """ Execute automated response playbook """ results = { "playbook": playbook["name"], "incident_id": incident["id"], "actions": [], "success": True } for action in playbook["actions"]: try: result = self.execute_action(action, incident) results["actions"].append({ "action": action["name"], "status": "SUCCESS", "result": result }) except Exception as e: results["actions"].append({ "action": action["name"], "status": "FAILED", "error": str(e) }) # Handle failure if action.get("on_failure") == "stop_playbook": results["success"] = False break elif action.get("on_failure") == "notify_admin": self.notify_admin(action, e) return results def execute_action(self, action, incident): """ Execute individual action """ action_type = action["type"] if action_type == "network_isolation": hosts = self.resolve_variable(action["targets"], incident) return self.integrations["edr"].isolate_hosts(hosts) elif action_type == "firewall_rule": ips = self.resolve_variable(action["destinations"], incident) return self.integrations["firewall"].block_ips(ips) elif action_type == "account_action": accounts = self.resolve_variable(action["accounts"], incident) if action["action"] == "disable": return self.disable_accounts(accounts) elif action_type == "notification": return self.send_notifications(action, incident) # Add more action types... ``` ### 3.2 Advanced Forensics #### 3.2.1 Memory Forensics **Volatility Framework Analysis:** ```bash #!/bin/bash # Automated memory analysis MEMORY_IMAGE=$1 PROFILE=$2 OUTPUT_DIR=$3 # Determine OS profile if [ -z "$PROFILE" ]; then echo "[*] Identifying OS profile..." PROFILE=$(volatility -f $MEMORY_IMAGE imageinfo | grep "Suggested Profile" | cut -d ":" -f 2 | cut -d "," -f 1 | tr -d ' ') fi echo "[+] Using profile: $PROFILE" mkdir -p $OUTPUT_DIR # Process listing echo "[*] Extracting process list..." volatility -f $MEMORY_IMAGE --profile=$PROFILE pslist > $OUTPUT_DIR/pslist.txt volatility -f $MEMORY_IMAGE --profile=$PROFILE pstree > $OUTPUT_DIR/pstree.txt # Network connections echo "[*] Extracting network connections..." volatility -f $MEMORY_IMAGE --profile=$PROFILE netscan > $OUTPUT_DIR/netscan.txt # Command line history echo "[*] Extracting command history..." volatility -f $MEMORY_IMAGE --profile=$PROFILE cmdscan > $OUTPUT_DIR/cmdscan.txt volatility -f $MEMORY_IMAGE --profile=$PROFILE consoles > $OUTPUT_DIR/consoles.txt # Loaded DLLs and modules echo "[*] Extracting loaded modules..." volatility -f $MEMORY_IMAGE --profile=$PROFILE dlllist > $OUTPUT_DIR/dlllist.txt # Malware detection echo "[*] Scanning for malware..." volatility -f $MEMORY_IMAGE --profile=$PROFILE malfind > $OUTPUT_DIR/malfind.txt # Extract suspicious processes echo "[*] Dumping suspicious processes..." volatility -f $MEMORY_IMAGE --profile=$PROFILE procdump -D $OUTPUT_DIR/dumps/ # Registry analysis echo "[*] Analyzing registry..." volatility -f $MEMORY_IMAGE --profile=$PROFILE hivelist > $OUTPUT_DIR/hivelist.txt echo "[+] Memory analysis complete. Results in $OUTPUT_DIR" ``` #### 3.2.2 Malware Analysis **Static and Dynamic Analysis:** ```python import pefile import hashlib import yara class MalwareAnalyzer: """Automated malware analysis framework""" def analyze_sample(self, file_path): """ Perform comprehensive malware analysis """ analysis = { "file_path": file_path, "hashes": self.calculate_hashes(file_path), "file_type": self.identify_file_type(file_path), "static_analysis": {}, "dynamic_analysis": {}, "threat_classification": {} } # Static analysis if analysis["file_type"] == "PE": analysis["static_analysis"] = self.analyze_pe(file_path) # YARA scanning analysis["yara_matches"] = self.scan_yara(file_path) # Threat intelligence lookup analysis["threat_intel"] = self.lookup_threat_intel( analysis["hashes"]["sha256"] ) # Dynamic analysis (sandboxed) if self.should_detonate(analysis): analysis["dynamic_analysis"] = self.sandbox_execute(file_path) return analysis def analyze_pe(self, file_path): """ Analyze PE (Windows executable) file """ pe = pefile.PE(file_path) analysis = { "imports": [], "exports": [], "sections": [], "suspicious_indicators": [] } # Analyze imports if hasattr(pe, 'DIRECTORY_ENTRY_IMPORT'): for entry in pe.DIRECTORY_ENTRY_IMPORT: dll_name = entry.dll.decode() for imp in entry.imports: func_name = imp.name.decode() if imp.name else f"Ordinal_{imp.ordinal}" analysis["imports"].append(f"{dll_name}:{func_name}") # Check for suspicious APIs if self.is_suspicious_api(func_name): analysis["suspicious_indicators"].append({ "type": "SUSPICIOUS_API", "value": f"{dll_name}:{func_name}" }) # Analyze sections for section in pe.sections: section_info = { "name": section.Name.decode().strip('\x00'), "virtual_address": hex(section.VirtualAddress), "virtual_size": section.Misc_VirtualSize, "raw_size": section.SizeOfRawData, "entropy": section.get_entropy() } analysis["sections"].append(section_info) # High entropy may indicate packing/encryption if section.get_entropy() > 7.0: analysis["suspicious_indicators"].append({ "type": "HIGH_ENTROPY_SECTION", "section": section_info["name"], "entropy": section.get_entropy() }) return analysis def sandbox_execute(self, file_path): """ Execute malware in isolated sandbox """ sandbox_result = { "network_activity": [], "file_operations": [], "registry_operations": [], "process_activity": [], "behavioral_indicators": [] } # This would integrate with Cuckoo Sandbox or similar # Simplified example: sandbox_result["behavioral_indicators"] = [ "Creates scheduled task for persistence", "Attempts to delete shadow copies", "Encrypts files with .locked extension", "Contacts C2 server at 192.168.1.100:443" ] return sandbox_result ``` --- ## Phase 4: Recovery and Resilience ### 4.1 Business Continuity Integration #### 4.1.1 Disaster Recovery Procedures **Recovery Time Objectives (RTO) and Recovery Point Objectives (RPO):** ```python class BCPIntegration: """Business Continuity Planning integration""" def assess_recovery_priority(self, affected_systems): """ Determine recovery priority based on business criticality """ priorities = { "TIER_1_CRITICAL": { "rto": "1 hour", "rpo": "15 minutes", "systems": ["payment_processing", "customer_database"] }, "TIER_2_HIGH": { "rto": "4 hours", "rpo": "1 hour", "systems": ["email", "crm", "erp"] }, "TIER_3_MEDIUM": { "rto": "24 hours", "rpo": "4 hours", "systems": ["reporting", "analytics"] }, "TIER_4_LOW": { "rto": "72 hours", "rpo": "24 hours", "systems": ["archive", "development"] } } recovery_plan = [] for tier, config in priorities.items(): affected = [s for s in affected_systems if s in config["systems"]] if affected: recovery_plan.append({ "tier": tier, "systems": affected, "rto": config["rto"], "rpo": config["rpo"], "order": len(recovery_plan) + 1 }) return sorted(recovery_plan, key=lambda x: x["order"]) def validate_backup_integrity(self, backup_id): """ Verify backup is clean and restorable """ validation = { "backup_id": backup_id, "integrity_check": False, "malware_scan": False, "restore_test": False, "approved_for_recovery": False } # Check backup integrity checksum_valid = self.verify_checksum(backup_id) validation["integrity_check"] = checksum_valid # Scan for malware if checksum_valid: scan_result = self.malware_scan_backup(backup_id) validation["malware_scan"] = scan_result["clean"] # Test restore to isolated environment if validation["malware_scan"]: restore_test = self.test_restore(backup_id) validation["restore_test"] = restore_test["success"] # Final approval validation["approved_for_recovery"] = all([ validation["integrity_check"], validation["malware_scan"], validation["restore_test"] ]) return validation ``` #### 4.1.2 Phased Recovery Process **Recovery Stages:** ```yaml # Recovery Workflow recovery_phases: phase_1_emergency_response: duration: "0-4 hours" objectives: - Activate incident response team - Isolate affected systems - Preserve evidence - Assess scope of impact phase_2_containment: duration: "4-8 hours" objectives: - Stop spread of incident - Identify all affected systems - Begin evidence collection - Communicate with stakeholders phase_3_eradication: duration: "8-24 hours" objectives: - Remove malware/threats - Close attack vectors - Patch vulnerabilities - Validate clean state phase_4_recovery: duration: "24-72 hours" objectives: - Restore from clean backups - Rebuild compromised systems - Reset credentials - Implement enhanced monitoring phase_5_post_incident: duration: "1-2 weeks" objectives: - Lessons learned review - Update procedures - Implement preventive measures - Security improvements ``` ### 4.2 Continuous Improvement #### 4.2.1 Metrics and KPIs **Incident Response Dashboard:** ```python class IncidentMetricsDashboard: """Track and visualize incident response metrics""" def calculate_metrics(self, incidents, timeframe="30d"): """ Calculate key incident response metrics """ metrics = { "mean_time_to_detect": 0, "mean_time_to_respond": 0, "mean_time_to_contain": 0, "mean_time_to_recover": 0, "total_incidents": len(incidents), "by_severity": {}, "by_category": {}, "false_positive_rate": 0, "recurrence_rate": 0 } # Calculate MTTD, MTTR, MTTC detection_times = [] response_times = [] containment_times = [] recovery_times = [] for incident in incidents: if incident["detection_time"] and incident["occurrence_time"]: detection_times.append( (incident["detection_time"] - incident["occurrence_time"]).total_seconds() / 60 ) if incident["response_time"] and incident["detection_time"]: response_times.append( (incident["response_time"] - incident["detection_time"]).total_seconds() / 60 ) if incident["containment_time"] and incident["response_time"]: containment_times.append( (incident["containment_time"] - incident["response_time"]).total_seconds() / 60 ) metrics["mean_time_to_detect"] = np.mean(detection_times) if detection_times else 0 metrics["mean_time_to_respond"] = np.mean(response_times) if response_times else 0 metrics["mean_time_to_contain"] = np.mean(containment_times) if containment_times else 0 # Categorize incidents for incident in incidents: # By severity severity = incident["severity"] metrics["by_severity"][severity] = metrics["by_severity"].get(severity, 0) + 1 # By category category = incident["category"] metrics["by_category"][category] = metrics["by_category"].get(category, 0) + 1 return metrics ``` --- ## Appendix: Advanced Topics ### A. Threat Intelligence Feeds **Recommended Sources:** - CISA AIS (Automated Indicator Sharing) - FBI InfraGard - MS-ISAC Threat Intelligence - Commercial feeds (CrowdStrike, Recorded Future, etc.) - Open source (AlienVault OTX, abuse.ch) ### B. Forensic Tool References **Essential Tools:** - **Memory Analysis:** Volatility, Rekall, WinDbg - **Disk Forensics:** Autopsy, FTK Imager, EnCase - **Network Forensics:** Wireshark, NetworkMiner, Zeek - **Malware Analysis:** IDA Pro, Ghidra, Cuckoo Sandbox, Any.run - **Log Analysis:** Splunk, ELK Stack, Graylog ### C. Compliance Mapping **Regulatory Frameworks:** - **GDPR Article 33:** 72-hour breach notification - **HIPAA:** 60-day breach notification - **PCI-DSS Requirement 12.10:** Incident response plan - **SOC 2:** Incident management processes - **ISO 27035:** Incident management standard --- **Document Control:** - Version: 1.0 - Status: ACTIVE - Last Review: 2025-12-25 - Next Review: 2026-06-25 - Owner: WIA Security Standards Committee **弘益人間 (홍익인간) - Benefit All Humanity** © 2025 SmileStory Inc. / WIA