Security Skills Atlas
incident response

Triaging Security Incidents with IR Playbooks

Classify and prioritize security incidents using structured IR playbooks to determine severity, assign response teams, and initiate appropriate response procedures.

incident-responseplaybookseverity-classificationsoctriage
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

MITRE ATT&CK

T1070 on the official MITRE ATT&CK siteT1078 on the official MITRE ATT&CK siteT1486 on the official MITRE ATT&CK siteT1490 on the official MITRE ATT&CK site

NIST CSF 2.0

RC.RP-01 on the official NIST CSF 2.0 siteRS.AN-03 on the official NIST CSF 2.0 siteRS.MA-01 on the official NIST CSF 2.0 siteRS.MA-02 on the official NIST CSF 2.0 site

When to Use

  • New security alert received from SIEM, EDR, or other detection sources
  • SOC analyst needs to determine if an alert is a true positive requiring response
  • Incident needs severity classification and team assignment
  • Multiple concurrent incidents require prioritization
  • Automated triage rules need validation or tuning

Prerequisites

  • SIEM platform with alert correlation (Splunk, Elastic, QRadar, Sentinel)
  • Incident response playbook library (by incident type)
  • Severity classification matrix approved by CISO
  • On-call rotation and escalation procedures
  • Ticketing system for incident tracking (ServiceNow, Jira, TheHive)
  • Threat intelligence feeds for IOC enrichment

Workflow

Step 1: Receive and Acknowledge Alert

# Query Splunk for new critical/high severity alerts
index=notable status=new severity IN ("critical","high")
| table _time, rule_name, src, dest, severity, description
| sort -_time
 
# Query TheHive for new cases
curl -s -H "Authorization: Bearer $THEHIVE_API_KEY" \
  "https://thehive.local/api/v1/query?name=list-alerts" \
  -H "Content-Type: application/json" \
  -d '{"query":[{"_name":"listAlert"},{"_name":"filter","_field":"status","_value":"New"}]}'
 
# Acknowledge alert in SIEM to prevent duplicate triage
curl -X POST "https://splunk.local:8089/services/notable_update" \
  -H "Authorization: Bearer $SPLUNK_TOKEN" \
  -d "ruleUIDs=$RULE_UID&status=1&comment=Triage+initiated+by+analyst"

Step 2: Enrich Alert Data

# Enrich source IP with VirusTotal
curl -s "https://www.virustotal.com/api/v3/ip_addresses/$SRC_IP" \
  -H "x-apikey: $VT_API_KEY" | jq '.data.attributes.last_analysis_stats'
 
# Check IP reputation with AbuseIPDB
curl -s "https://api.abuseipdb.com/api/v2/check?ipAddress=$SRC_IP&maxAgeInDays=90" \
  -H "Key: $ABUSEIPDB_KEY" -H "Accept: application/json" | jq '.data'
 
# Enrich file hash with threat intelligence
curl -s "https://www.virustotal.com/api/v3/files/$FILE_HASH" \
  -H "x-apikey: $VT_API_KEY" | jq '.data.attributes.last_analysis_stats'
 
# Query internal asset database for affected systems
curl -s "https://cmdb.local/api/assets?ip=$DEST_IP" \
  -H "Authorization: Bearer $CMDB_TOKEN" | jq '.asset_criticality, .owner, .environment'

Step 3: Classify Incident Type

# Map alert to incident category using playbook lookup
# Categories: Malware, Phishing, Unauthorized Access, Data Exfiltration,
# DoS/DDoS, Insider Threat, Ransomware, Account Compromise, Web Attack
 
# Check if alert matches known playbook trigger conditions
grep -i "$ALERT_SIGNATURE" /opt/ir/playbooks/trigger_conditions.yaml
 
# Determine incident type from MITRE ATT&CK technique
curl -s "https://attack.mitre.org/api/techniques/$TECHNIQUE_ID" | jq '.name, .tactic'

Step 4: Assign Severity Level

# Severity matrix factors:
# 1. Asset criticality (Critical/High/Medium/Low)
# 2. Data sensitivity (PII/PHI/PCI/Confidential/Public)
# 3. Number of affected systems
# 4. Active vs historical threat
# 5. Confirmed vs suspected compromise
 
# Automated severity calculation
python3 -c "
severity_score = 0
# Asset criticality: Critical=4, High=3, Medium=2, Low=1
severity_score += 4  # Critical server
# Data sensitivity: PII/PHI=4, PCI=3, Confidential=2, Public=1
severity_score += 3  # PCI data
# Scope: Enterprise=4, Department=3, Single system=2, Single user=1
severity_score += 2  # Single system
# Threat status: Active=4, Recent=3, Historical=2, Potential=1
severity_score += 4  # Active threat
 
if severity_score >= 12: print('CRITICAL - P1')
elif severity_score >= 9: print('HIGH - P2')
elif severity_score >= 6: print('MEDIUM - P3')
else: print('LOW - P4')
print(f'Score: {severity_score}/16')
"

Step 5: Select and Initiate Playbook

# Load appropriate playbook based on incident type
cat /opt/ir/playbooks/ransomware_playbook.yaml
cat /opt/ir/playbooks/phishing_playbook.yaml
cat /opt/ir/playbooks/unauthorized_access_playbook.yaml
 
# Create incident ticket in TheHive
curl -X POST "https://thehive.local/api/v1/case" \
  -H "Authorization: Bearer $THEHIVE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "title": "IR-2024-XXX: [Incident Type] - [Brief Description]",
    "description": "Triage summary and initial findings",
    "severity": 3,
    "tlp": 2,
    "pap": 2,
    "tags": ["ransomware", "triage-complete"],
    "customFields": {
      "playbook": {"string": "ransomware_v2"},
      "affected_systems": {"integer": 5}
    }
  }'

Step 6: Assign Response Team

# Check on-call schedule
curl -s "https://pagerduty.com/api/v2/oncalls?schedule_ids[]=$SCHEDULE_ID" \
  -H "Authorization: Token token=$PD_TOKEN" | jq '.oncalls[].user.summary'
 
# Page incident responders based on severity
# P1/Critical: Page IR lead + senior analysts + CISO
# P2/High: Page IR lead + available analysts
# P3/Medium: Assign to next available analyst
# P4/Low: Queue for business hours processing
 
curl -X POST "https://events.pagerduty.com/v2/enqueue" \
  -H "Content-Type: application/json" \
  -d '{
    "routing_key": "'$PD_ROUTING_KEY'",
    "event_action": "trigger",
    "payload": {
      "summary": "P1 Security Incident: Ransomware detected on PROD-DB-01",
      "severity": "critical",
      "source": "SIEM-Splunk",
      "custom_details": {"incident_id": "IR-2024-042", "playbook": "ransomware_v2"}
    }
  }'

Step 7: Document Triage Decision and Hand Off

# Update incident ticket with triage summary
curl -X PATCH "https://thehive.local/api/v1/case/$CASE_ID" \
  -H "Authorization: Bearer $THEHIVE_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "status": "InProgress",
    "customFields": {
      "triage_analyst": {"string": "analyst_name"},
      "triage_time": {"date": '$(date +%s000)'},
      "severity_justification": {"string": "Critical asset + active threat + PCI data"}
    }
  }'

Key Concepts

Concept Description
True Positive Alert correctly identifying a real security incident
False Positive Alert incorrectly flagging benign activity as malicious
Severity Classification Ranking incident priority based on impact and urgency
Playbook Selection Choosing the appropriate response procedure based on incident type
IOC Enrichment Adding context to indicators from threat intelligence sources
Escalation Threshold Criteria triggering escalation to higher severity or management
Triage SLA Time target for initial assessment (typically 15-30 min for critical)

Tools & Systems

Tool Purpose
Splunk/Elastic/QRadar SIEM alert correlation and querying
TheHive/SIRP Incident case management and playbook tracking
VirusTotal/AbuseIPDB IOC reputation and enrichment
PagerDuty/OpsGenie On-call management and alerting
MITRE ATT&CK Technique classification and mapping
Cortex XSOAR SOAR platform for automated triage workflows

Common Scenarios

  1. Brute Force Alert: Multiple failed logins from single IP. Enrich IP reputation, check geo-location, verify if account was compromised, assign P3 if unsuccessful.
  2. Malware Detection on Endpoint: AV/EDR quarantined malware. Verify quarantine success, check for lateral movement, assign P2 if persistence detected.
  3. Suspicious Outbound Traffic: Large data transfer to unknown external IP. Check if known cloud service, verify data classification, assign P1 if exfiltration confirmed.
  4. Phishing Email Reported: User reports suspicious email. Extract IOCs, check if others received it, assign P2 if credentials were entered.
  5. Privilege Escalation: User gained admin rights unexpectedly. Verify if authorized change, check for exploitation, assign P1 if unauthorized.

Output Format

  • Triage decision document with severity justification
  • Incident ticket with assigned playbook and team
  • IOC enrichment summary attached to case
  • Escalation notification to appropriate stakeholders
  • Initial timeline of events from alert data
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 3

api-reference.md1.9 KB

API Reference: Triaging Security Incidents with IR Playbooks

Incident Classification Types

Type Keywords Default Severity Playbook
Malware trojan, ransomware, c2, beacon High malware-infection-playbook
Phishing credential harvest, BEC, spear-phishing Medium phishing-response-playbook
Data Exfiltration DLP, dns tunnel, large upload Critical data-exfiltration-playbook
Unauthorized Access brute force, lateral movement High unauthorized-access-playbook
Denial of Service DDoS, SYN flood, volumetric High ddos-response-playbook
Insider Threat policy violation, terminated user High insider-threat-playbook
Web Attack SQLi, XSS, web shell, RCE High web-attack-playbook

Severity Matrix

Context Factor Severity Override
Crown jewel system affected Critical
Active exploitation confirmed Critical
Multiple systems (>5) affected High
Single system affected Medium
Reconnaissance only Low
Minor policy violation Informational

Escalation Paths

Severity Response Time Escalation
Critical 15 minutes IR Team + CISO + Legal
High 1 hour SOC Tier 2 + IR Team
Medium 4 hours SOC Tier 2
Low 24 hours SOC Tier 1
Informational Next business day SOC Tier 1

Python Libraries

Library Version Purpose
json stdlib Alert parsing and report generation
enum stdlib Severity level enumeration
pathlib stdlib Output directory management
datetime stdlib Triage timestamps

References

standards.md2.3 KB

Standards and Framework References - Incident Triage

NIST SP 800-61 Rev. 3 - Incident Triage Alignment

  • Detect (DE): Alert analysis and triage
    • DE.AE-02: Potentially adverse events are analyzed to better understand associated activities
    • DE.AE-03: Information is correlated from multiple sources
    • DE.AE-04: The estimated impact and scope of adverse events is understood
  • Respond (RS): Incident classification and escalation
    • RS.AN-03: Analysis performed to establish awareness of incident scope
    • RS.CO-02: Incidents reported consistent with established criteria

SANS PICERL - Identification Phase

  • Phase 2 focuses on detecting and validating security events
  • Triage determines if an event qualifies as an incident
  • Key activities: alert validation, initial scoping, severity assignment
  • Triage SLAs: P1 <15 min, P2 <30 min, P3 <1 hour, P4 <4 hours

NIST Severity Classification (SP 800-61 Rev. 2, Table 3-2)

Category Definition Examples
CAT 1 - Unauthorized Access Individual gains access without permission Compromised credentials, privilege escalation
CAT 2 - Denial of Service Disruption of service availability DDoS, resource exhaustion
CAT 3 - Malicious Code Infection by malware Virus, worm, trojan, ransomware
CAT 4 - Improper Usage Violation of acceptable use policy Unauthorized software, policy breach
CAT 5 - Scans/Probes Reconnaissance activity Port scans, vulnerability scans
CAT 6 - Investigation Unconfirmed suspicious activity Anomalous behavior under review

MITRE ATT&CK - Triage Technique Mapping

  • Map observed techniques to ATT&CK framework during triage
  • Technique identification helps select appropriate playbook
  • Tactic identification reveals attacker's current phase
  • Reference: https://attack.mitre.org/

FIRST CSIRT Services Framework

US-CERT Federal Incident Reporting Guidelines

workflows.md4.6 KB

Incident Triage with IR Playbooks - Detailed Workflow

Triage Decision Tree

Alert Received
    |
    v
Is alert from trusted/tuned detection rule?
    |-- No --> Check rule logic, verify data source --> Potential false positive
    |-- Yes --> Continue
    |
    v
Does alert match known false positive pattern?
    |-- Yes --> Document, close as false positive, tune rule
    |-- No --> Continue
    |
    v
Can indicator be enriched with external threat intel?
    |-- Yes --> Enrich with VT, AbuseIPDB, OTX --> Add context
    |-- No --> Continue with available data
    |
    v
What is the incident type?
    |-- Malware --> Malware playbook
    |-- Phishing --> Phishing playbook
    |-- Unauthorized Access --> Access compromise playbook
    |-- Data Exfiltration --> Data breach playbook
    |-- Ransomware --> Ransomware playbook
    |-- DoS/DDoS --> Availability playbook
    |-- Insider Threat --> Insider playbook
    |
    v
Assign severity based on:
    - Asset criticality x Threat level x Data sensitivity
    |
    v
Route to appropriate team with playbook

Severity Assignment Matrix

Impact Score (1-4)

Score Asset Criticality Examples
4 Critical Domain controllers, production databases, financial systems
3 High Email servers, web applications, file servers
2 Medium Development systems, internal tools
1 Low Test systems, non-production workstations

Urgency Score (1-4)

Score Threat Status Indicators
4 Active exploitation Ongoing attack, real-time data loss
3 Confirmed compromise Evidence of breach, but not active
2 Attempted attack Blocked attack, no evidence of success
1 Reconnaissance Scanning, probing, no exploitation attempt

Final Severity = Impact x Urgency

Score Range Severity Response Time Escalation
12-16 P1 Critical Immediate (15 min) CISO + IR Lead + Senior Analysts
8-11 P2 High 30 minutes IR Lead + Available Analysts
4-7 P3 Medium 2 hours Next available analyst
1-3 P4 Low 24 hours (business hours) Queued for analyst review

Playbook Selection Guide

By Alert Source

Alert Source Likely Playbook Key Triage Actions
EDR - Malware detection Malware IR Check quarantine status, verify family
Email gateway - Phishing Phishing IR Extract IOCs, check delivery scope
SIEM - Authentication anomaly Account Compromise Verify account, check lateral movement
IDS/IPS - Exploit attempt Vulnerability Exploitation Verify patch status, check success
DLP - Data transfer Data Exfiltration Classify data, verify authorization
Cloud - Impossible travel Cloud Account Compromise Verify user, check API calls

By MITRE ATT&CK Tactic

Tactic Playbook Priority
Initial Access (TA0001) Perimeter Breach P1-P2
Execution (TA0002) Malware/Code Execution P1-P2
Persistence (TA0003) Backdoor/Implant P2
Privilege Escalation (TA0004) Privilege Escalation P1
Defense Evasion (TA0005) Security Tool Bypass P2
Credential Access (TA0006) Credential Theft P1-P2
Discovery (TA0007) Reconnaissance P3
Lateral Movement (TA0008) Lateral Movement P1
Collection (TA0009) Data Staging P2
Exfiltration (TA0010) Data Breach P1
Impact (TA0040) Ransomware/Destruction P1

IOC Enrichment Workflow

Step 1: Automated Enrichment

  1. Submit IPs to VirusTotal, AbuseIPDB, Shodan
  2. Submit file hashes to VirusTotal, MalwareBazaar, Hybrid Analysis
  3. Submit domains to URLScan.io, VirusTotal, PassiveTotal
  4. Check against internal IOC database and watchlists

Step 2: Context Addition

  1. Look up asset in CMDB for criticality and owner
  2. Check user in HR system for role and access level
  3. Verify network zone and data classification
  4. Cross-reference with recent threat intelligence reports

Step 3: Correlation

  1. Search SIEM for related alerts in past 72 hours
  2. Check if same IOCs appeared in other incidents
  3. Correlate with ongoing threat campaigns
  4. Verify if alert is part of a larger attack chain

Triage Documentation Requirements

  1. Alert details (source, time, raw data)
  2. Enrichment results (reputation scores, intelligence hits)
  3. Classification decision (incident type, severity, justification)
  4. Selected playbook and version
  5. Assigned team/analyst
  6. Initial timeline of observed events
  7. Known affected assets and accounts

Scripts 2

agent.py8.5 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Agent for triaging security incidents with IR playbooks.

Classifies alerts by incident type, assigns severity using a
structured matrix, selects the appropriate IR playbook, and
generates triage decisions with escalation recommendations.
"""

import json
import sys
from pathlib import Path
from datetime import datetime
from enum import Enum


class Severity(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "informational"


INCIDENT_TYPES = {
    "malware": {
        "keywords": ["malware", "trojan", "ransomware", "virus", "worm", "dropper", "c2", "beacon"],
        "default_severity": Severity.HIGH,
        "playbook": "malware-infection-playbook",
        "escalation": "SOC Tier 2 + IR Team",
    },
    "phishing": {
        "keywords": ["phishing", "credential harvest", "suspicious email", "spear-phishing", "bec"],
        "default_severity": Severity.MEDIUM,
        "playbook": "phishing-response-playbook",
        "escalation": "SOC Tier 2",
    },
    "data_exfiltration": {
        "keywords": ["exfiltration", "data leak", "dlp", "large upload", "dns tunnel", "unusual transfer"],
        "default_severity": Severity.CRITICAL,
        "playbook": "data-exfiltration-playbook",
        "escalation": "IR Team + CISO",
    },
    "unauthorized_access": {
        "keywords": ["brute force", "credential stuffing", "privilege escalation", "lateral movement",
                      "pass-the-hash", "kerberoasting", "golden ticket"],
        "default_severity": Severity.HIGH,
        "playbook": "unauthorized-access-playbook",
        "escalation": "SOC Tier 2 + AD Team",
    },
    "denial_of_service": {
        "keywords": ["ddos", "dos", "syn flood", "amplification", "volumetric", "resource exhaustion"],
        "default_severity": Severity.HIGH,
        "playbook": "ddos-response-playbook",
        "escalation": "NOC + SOC Tier 2",
    },
    "insider_threat": {
        "keywords": ["insider", "policy violation", "unauthorized copy", "after hours", "terminated user"],
        "default_severity": Severity.HIGH,
        "playbook": "insider-threat-playbook",
        "escalation": "IR Team + HR + Legal",
    },
    "web_attack": {
        "keywords": ["sqli", "xss", "rce", "web shell", "injection", "traversal", "deserialization"],
        "default_severity": Severity.HIGH,
        "playbook": "web-attack-playbook",
        "escalation": "SOC Tier 2 + AppSec",
    },
}

SEVERITY_MATRIX = {
    "crown_jewel_affected": Severity.CRITICAL,
    "active_exploitation": Severity.CRITICAL,
    "multiple_systems": Severity.HIGH,
    "single_system": Severity.MEDIUM,
    "reconnaissance_only": Severity.LOW,
    "policy_violation_minor": Severity.INFO,
}


class IncidentTriageAgent:
    """Triages security incidents using structured IR playbooks."""

    def __init__(self, output_dir="./incident_triage"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.triage_results = []

    def classify_incident(self, alert_text):
        """Classify alert into incident type based on keyword matching."""
        alert_lower = alert_text.lower()
        scores = {}
        for inc_type, config in INCIDENT_TYPES.items():
            score = sum(1 for kw in config["keywords"] if kw in alert_lower)
            if score > 0:
                scores[inc_type] = score
        if not scores:
            return {"type": "unknown", "confidence": 0}
        best = max(scores, key=scores.get)
        return {"type": best, "confidence": scores[best], "all_matches": scores}

    def assess_severity(self, classification, context=None):
        """Determine severity using classification and contextual factors."""
        ctx = context or {}
        inc_type = classification.get("type", "unknown")
        config = INCIDENT_TYPES.get(inc_type, {})
        base_severity = config.get("default_severity", Severity.MEDIUM)

        if ctx.get("crown_jewel_affected"):
            return Severity.CRITICAL
        if ctx.get("active_exploitation"):
            return Severity.CRITICAL
        if ctx.get("systems_affected", 1) > 5:
            return Severity.HIGH if base_severity != Severity.CRITICAL else Severity.CRITICAL
        return base_severity

    def select_playbook(self, classification):
        """Select appropriate IR playbook for the incident type."""
        inc_type = classification.get("type", "unknown")
        config = INCIDENT_TYPES.get(inc_type)
        if not config:
            return {"playbook": "generic-incident-playbook", "escalation": "SOC Tier 1"}
        return {"playbook": config["playbook"], "escalation": config["escalation"]}

    def build_triage_decision(self, alert_text, context=None):
        """Complete triage: classify, assess, assign playbook."""
        classification = self.classify_incident(alert_text)
        severity = self.assess_severity(classification, context)
        playbook = self.select_playbook(classification)

        decision = {
            "timestamp": datetime.utcnow().isoformat(),
            "alert_summary": alert_text[:200],
            "classification": classification,
            "severity": severity.value,
            "playbook": playbook["playbook"],
            "escalation_to": playbook["escalation"],
            "immediate_actions": self._get_immediate_actions(classification["type"], severity),
            "containment_needed": severity in (Severity.CRITICAL, Severity.HIGH),
        }
        self.triage_results.append(decision)
        return decision

    def _get_immediate_actions(self, inc_type, severity):
        actions = {
            "malware": ["Isolate affected host from network", "Collect memory dump",
                        "Block C2 indicators at firewall", "Preserve disk image"],
            "phishing": ["Block sender domain at email gateway", "Search for other recipients",
                         "Reset credentials if clicked", "Report to anti-phishing service"],
            "data_exfiltration": ["Block destination IPs/domains", "Disable compromised account",
                                  "Preserve DLP logs", "Notify legal/compliance"],
            "unauthorized_access": ["Disable compromised account", "Reset credentials",
                                    "Review authentication logs", "Check for persistence"],
            "denial_of_service": ["Enable DDoS mitigation", "Contact ISP/CDN",
                                  "Capture traffic sample", "Identify attack vector"],
            "insider_threat": ["Preserve evidence chain of custody", "Restrict account access",
                               "Monitor user activity", "Coordinate with HR"],
            "web_attack": ["Enable WAF blocking mode", "Capture attack payloads",
                           "Check for web shells", "Review application logs"],
        }
        return actions.get(inc_type, ["Acknowledge and investigate", "Escalate to Tier 2"])

    def prioritize_queue(self, alerts):
        """Prioritize multiple alerts by severity and type."""
        severity_order = {Severity.CRITICAL: 0, Severity.HIGH: 1, Severity.MEDIUM: 2,
                          Severity.LOW: 3, Severity.INFO: 4}
        decisions = [self.build_triage_decision(a["text"], a.get("context")) for a in alerts]
        decisions.sort(key=lambda d: severity_order.get(Severity(d["severity"]), 5))
        return decisions

    def generate_report(self, alerts=None):
        if alerts:
            self.prioritize_queue(alerts)
        report = {
            "report_date": datetime.utcnow().isoformat(),
            "total_triaged": len(self.triage_results),
            "by_severity": {},
            "triage_decisions": self.triage_results,
        }
        for d in self.triage_results:
            sev = d["severity"]
            report["by_severity"][sev] = report["by_severity"].get(sev, 0) + 1

        out = self.output_dir / "incident_triage_report.json"
        with open(out, "w") as f:
            json.dump(report, f, indent=2)
        print(json.dumps(report, indent=2))
        return report


def main():
    if len(sys.argv) < 2:
        print("Usage: agent.py '<alert_text>' [--crown-jewel] [--active-exploit]")
        sys.exit(1)
    alert = sys.argv[1]
    context = {}
    if "--crown-jewel" in sys.argv:
        context["crown_jewel_affected"] = True
    if "--active-exploit" in sys.argv:
        context["active_exploitation"] = True
    agent = IncidentTriageAgent()
    agent.build_triage_decision(alert, context)
    agent.generate_report()


if __name__ == "__main__":
    main()
process.py15.2 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""
Security Incident Triage Automation Script

Automates incident triage workflow:
- Enriches IOCs with threat intelligence APIs
- Calculates severity based on asset criticality and threat level
- Selects appropriate IR playbook
- Creates incident tickets
- Generates triage report

Requirements:
    pip install requests pyyaml
"""

import argparse
import json
import logging
import os
import sys
from datetime import datetime, timezone
from typing import Optional

try:
    import requests
except ImportError:
    print("Install requests: pip install requests")
    sys.exit(1)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("incident_triage")

# Incident type to playbook mapping
PLAYBOOK_MAP = {
    "malware": {"playbook": "malware_ir_v2", "team": "malware_analysis"},
    "ransomware": {"playbook": "ransomware_ir_v3", "team": "ransomware_response"},
    "phishing": {"playbook": "phishing_ir_v2", "team": "email_security"},
    "unauthorized_access": {"playbook": "access_compromise_v2", "team": "identity_response"},
    "data_exfiltration": {"playbook": "data_breach_v2", "team": "data_protection"},
    "ddos": {"playbook": "ddos_response_v1", "team": "network_operations"},
    "insider_threat": {"playbook": "insider_threat_v2", "team": "insider_risk"},
    "account_compromise": {"playbook": "account_compromise_v2", "team": "identity_response"},
    "web_attack": {"playbook": "web_attack_v2", "team": "application_security"},
    "privilege_escalation": {"playbook": "privesc_ir_v1", "team": "identity_response"},
    "lateral_movement": {"playbook": "lateral_movement_v1", "team": "network_defense"},
    "supply_chain": {"playbook": "supply_chain_v1", "team": "third_party_risk"},
}

# Severity calculation weights
SEVERITY_WEIGHTS = {
    "asset_criticality": {"critical": 4, "high": 3, "medium": 2, "low": 1},
    "data_sensitivity": {"pii_phi": 4, "pci": 3, "confidential": 2, "public": 1},
    "threat_status": {"active": 4, "confirmed": 3, "attempted": 2, "recon": 1},
    "scope": {"enterprise": 4, "department": 3, "single_system": 2, "single_user": 1},
}


class IOCEnricher:
    """Enrich IOCs with external threat intelligence sources."""

    def __init__(self, vt_api_key: str = "", abuseipdb_key: str = ""):
        self.vt_api_key = vt_api_key or os.getenv("VT_API_KEY", "")
        self.abuseipdb_key = abuseipdb_key or os.getenv("ABUSEIPDB_KEY", "")

    def enrich_ip(self, ip_address: str) -> dict:
        result = {"ip": ip_address, "sources": {}}

        # VirusTotal
        if self.vt_api_key:
            try:
                resp = requests.get(
                    f"https://www.virustotal.com/api/v3/ip_addresses/{ip_address}",
                    headers={"x-apikey": self.vt_api_key},
                    timeout=10,
                )
                if resp.status_code == 200:
                    data = resp.json().get("data", {}).get("attributes", {})
                    stats = data.get("last_analysis_stats", {})
                    result["sources"]["virustotal"] = {
                        "malicious": stats.get("malicious", 0),
                        "suspicious": stats.get("suspicious", 0),
                        "harmless": stats.get("harmless", 0),
                        "undetected": stats.get("undetected", 0),
                        "reputation": data.get("reputation", 0),
                        "country": data.get("country", "unknown"),
                        "as_owner": data.get("as_owner", "unknown"),
                    }
                    logger.info(f"VT enrichment for {ip_address}: {stats.get('malicious', 0)} malicious")
            except Exception as e:
                logger.warning(f"VT enrichment failed for {ip_address}: {e}")

        # AbuseIPDB
        if self.abuseipdb_key:
            try:
                resp = requests.get(
                    f"https://api.abuseipdb.com/api/v2/check",
                    params={"ipAddress": ip_address, "maxAgeInDays": 90},
                    headers={"Key": self.abuseipdb_key, "Accept": "application/json"},
                    timeout=10,
                )
                if resp.status_code == 200:
                    data = resp.json().get("data", {})
                    result["sources"]["abuseipdb"] = {
                        "abuse_confidence": data.get("abuseConfidenceScore", 0),
                        "total_reports": data.get("totalReports", 0),
                        "country_code": data.get("countryCode", ""),
                        "isp": data.get("isp", ""),
                        "is_tor": data.get("isTor", False),
                    }
                    logger.info(f"AbuseIPDB for {ip_address}: confidence={data.get('abuseConfidenceScore', 0)}%")
            except Exception as e:
                logger.warning(f"AbuseIPDB enrichment failed for {ip_address}: {e}")

        # Calculate overall threat score
        vt_malicious = result.get("sources", {}).get("virustotal", {}).get("malicious", 0)
        abuse_score = result.get("sources", {}).get("abuseipdb", {}).get("abuse_confidence", 0)
        result["threat_score"] = min(100, (vt_malicious * 5) + abuse_score)
        result["threat_level"] = (
            "critical" if result["threat_score"] >= 80
            else "high" if result["threat_score"] >= 50
            else "medium" if result["threat_score"] >= 20
            else "low"
        )
        return result

    def enrich_hash(self, file_hash: str) -> dict:
        result = {"hash": file_hash, "sources": {}}
        if self.vt_api_key:
            try:
                resp = requests.get(
                    f"https://www.virustotal.com/api/v3/files/{file_hash}",
                    headers={"x-apikey": self.vt_api_key},
                    timeout=10,
                )
                if resp.status_code == 200:
                    data = resp.json().get("data", {}).get("attributes", {})
                    stats = data.get("last_analysis_stats", {})
                    result["sources"]["virustotal"] = {
                        "malicious": stats.get("malicious", 0),
                        "suspicious": stats.get("suspicious", 0),
                        "detection_names": list(
                            name for eng, det in data.get("last_analysis_results", {}).items()
                            if det.get("category") == "malicious"
                            for name in [det.get("result", "")]
                        )[:10],
                        "file_type": data.get("type_description", ""),
                        "file_name": data.get("meaningful_name", ""),
                    }
            except Exception as e:
                logger.warning(f"VT hash enrichment failed: {e}")
        return result

    def enrich_domain(self, domain: str) -> dict:
        result = {"domain": domain, "sources": {}}
        if self.vt_api_key:
            try:
                resp = requests.get(
                    f"https://www.virustotal.com/api/v3/domains/{domain}",
                    headers={"x-apikey": self.vt_api_key},
                    timeout=10,
                )
                if resp.status_code == 200:
                    data = resp.json().get("data", {}).get("attributes", {})
                    stats = data.get("last_analysis_stats", {})
                    result["sources"]["virustotal"] = {
                        "malicious": stats.get("malicious", 0),
                        "suspicious": stats.get("suspicious", 0),
                        "reputation": data.get("reputation", 0),
                        "creation_date": data.get("creation_date", ""),
                        "registrar": data.get("registrar", ""),
                    }
            except Exception as e:
                logger.warning(f"VT domain enrichment failed: {e}")
        return result


class SeverityCalculator:
    """Calculate incident severity based on multiple factors."""

    @staticmethod
    def calculate(asset_criticality: str, data_sensitivity: str,
                  threat_status: str, scope: str) -> dict:
        score = (
            SEVERITY_WEIGHTS["asset_criticality"].get(asset_criticality, 1)
            + SEVERITY_WEIGHTS["data_sensitivity"].get(data_sensitivity, 1)
            + SEVERITY_WEIGHTS["threat_status"].get(threat_status, 1)
            + SEVERITY_WEIGHTS["scope"].get(scope, 1)
        )
        if score >= 13:
            severity, priority, response_time = "Critical", "P1", "15 minutes"
        elif score >= 10:
            severity, priority, response_time = "High", "P2", "30 minutes"
        elif score >= 6:
            severity, priority, response_time = "Medium", "P3", "2 hours"
        else:
            severity, priority, response_time = "Low", "P4", "24 hours"

        return {
            "score": score,
            "max_score": 16,
            "severity": severity,
            "priority": priority,
            "response_time_sla": response_time,
            "factors": {
                "asset_criticality": asset_criticality,
                "data_sensitivity": data_sensitivity,
                "threat_status": threat_status,
                "scope": scope,
            },
        }


class PlaybookSelector:
    """Select appropriate IR playbook based on incident type."""

    @staticmethod
    def select(incident_type: str) -> dict:
        playbook = PLAYBOOK_MAP.get(incident_type.lower())
        if not playbook:
            return {
                "playbook": "generic_ir_v1",
                "team": "general_ir",
                "note": f"No specific playbook for type '{incident_type}', using generic",
            }
        return playbook


class TheHiveClient:
    """Create and manage incidents in TheHive."""

    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.api_key = api_key

    def _headers(self):
        return {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}

    def create_case(self, title: str, description: str, severity: int,
                    tags: list, custom_fields: dict = None) -> dict:
        payload = {
            "title": title,
            "description": description,
            "severity": severity,
            "tlp": 2,
            "pap": 2,
            "tags": tags,
        }
        if custom_fields:
            payload["customFields"] = custom_fields
        try:
            resp = requests.post(
                f"{self.base_url}/api/v1/case",
                headers=self._headers(),
                json=payload,
                timeout=10,
            )
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            logger.error(f"Failed to create TheHive case: {e}")
            return {"error": str(e)}


def generate_triage_report(alert_data: dict, enrichment: dict,
                           severity: dict, playbook: dict, output_path: str):
    """Generate a triage assessment report."""
    report = {
        "triage_report": {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "analyst": os.getenv("USERNAME", os.getenv("USER", "unknown")),
            "alert_data": alert_data,
            "enrichment_results": enrichment,
            "severity_assessment": severity,
            "playbook_assignment": playbook,
            "decision": "escalate" if severity["priority"] in ("P1", "P2") else "investigate",
        }
    }
    with open(output_path, "w") as f:
        json.dump(report, f, indent=2)
    logger.info(f"Triage report saved to: {output_path}")
    return report


def main():
    parser = argparse.ArgumentParser(description="Security Incident Triage Automation")
    parser.add_argument("--alert-source", required=True, help="Source of the alert (e.g., SIEM, EDR)")
    parser.add_argument("--alert-name", required=True, help="Alert rule name or title")
    parser.add_argument("--incident-type", required=True,
                        choices=list(PLAYBOOK_MAP.keys()),
                        help="Classified incident type")
    parser.add_argument("--src-ip", help="Source IP address to enrich")
    parser.add_argument("--dest-ip", help="Destination IP address")
    parser.add_argument("--file-hash", help="File hash (SHA256) to enrich")
    parser.add_argument("--domain", help="Domain to enrich")
    parser.add_argument("--asset-criticality", default="medium",
                        choices=["critical", "high", "medium", "low"])
    parser.add_argument("--data-sensitivity", default="confidential",
                        choices=["pii_phi", "pci", "confidential", "public"])
    parser.add_argument("--threat-status", default="confirmed",
                        choices=["active", "confirmed", "attempted", "recon"])
    parser.add_argument("--scope", default="single_system",
                        choices=["enterprise", "department", "single_system", "single_user"])
    parser.add_argument("--output-dir", default="./triage_output")
    parser.add_argument("--thehive-url", default=os.getenv("THEHIVE_URL", ""))
    parser.add_argument("--thehive-key", default=os.getenv("THEHIVE_API_KEY", ""))

    args = parser.parse_args()
    os.makedirs(args.output_dir, exist_ok=True)

    # Enrich IOCs
    enricher = IOCEnricher()
    enrichment = {}
    if args.src_ip:
        enrichment["src_ip"] = enricher.enrich_ip(args.src_ip)
    if args.file_hash:
        enrichment["file_hash"] = enricher.enrich_hash(args.file_hash)
    if args.domain:
        enrichment["domain"] = enricher.enrich_domain(args.domain)

    # Calculate severity
    severity = SeverityCalculator.calculate(
        args.asset_criticality, args.data_sensitivity,
        args.threat_status, args.scope,
    )
    logger.info(f"Severity: {severity['severity']} ({severity['priority']}) - Score: {severity['score']}/{severity['max_score']}")

    # Select playbook
    playbook = PlaybookSelector.select(args.incident_type)
    logger.info(f"Playbook: {playbook['playbook']} - Team: {playbook['team']}")

    # Create ticket in TheHive if configured
    if args.thehive_url and args.thehive_key:
        thehive = TheHiveClient(args.thehive_url, args.thehive_key)
        severity_map = {"Critical": 4, "High": 3, "Medium": 2, "Low": 1}
        case = thehive.create_case(
            title=f"[{severity['priority']}] {args.alert_name}",
            description=f"Triage: {args.incident_type} incident from {args.alert_source}",
            severity=severity_map.get(severity["severity"], 2),
            tags=[args.incident_type, severity["priority"], "triage-complete"],
            custom_fields={"playbook": {"string": playbook["playbook"]}},
        )
        logger.info(f"TheHive case created: {case}")

    # Generate report
    alert_data = {
        "source": args.alert_source,
        "name": args.alert_name,
        "type": args.incident_type,
        "src_ip": args.src_ip,
        "dest_ip": args.dest_ip,
    }
    report_path = os.path.join(args.output_dir, f"triage_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
    generate_triage_report(alert_data, enrichment, severity, playbook, report_path)

    print(f"\nTriage Complete")
    print(f"Severity: {severity['severity']} ({severity['priority']})")
    print(f"Playbook: {playbook['playbook']}")
    print(f"Response SLA: {severity['response_time_sla']}")
    print(f"Report: {report_path}")


if __name__ == "__main__":
    main()

Assets 1

template.mdtext/markdown · 2.9 KB
Keep exploring
Collecting Volatile Evidence from Compromised HostsCollect volatile forensic evidence from a compromised system following order of volatility, preserving memory, network connections, processes, and system state before they are lost.Conducting Post-Incident Lessons LearnedFacilitate structured post-incident reviews to identify root causes, document what worked and failed, and produce actionable recommendations to improve future incident response.Eradicating Malware from Infected SystemsSystematically remove malware, backdoors, and attacker persistence mechanisms from infected systems while ensuring complete eradication and preventing re-infection.Implementing Velociraptor for IR CollectionDeploy and configure Velociraptor for scalable endpoint forensic artifact collection during incident response using VQL queries, hunts, and pre-built artifact packs across Windows, Linux, and macOS environments.Testing Ransomware Recovery ProceduresTest and validate ransomware recovery procedures including backup restore operations, RTO/RPO target verification, recovery sequencing, and clean restore validation to ensure organizational resilience against destructive ransomware attacks.Triaging Security IncidentsPerforms initial triage of security incidents to determine severity, scope, and required response actions using the NIST SP 800-61r3 and SANS PICERL frameworks. Classifies incidents by type, assigns priority based on business impact, and routes to appropriate response teams. Activates for requests involving incident triage, security alert classification, severity assessment, incident prioritization, or initial incident analysis.