Implementing Ticketing System For Incidents

incident_taxonomy:
  categories:
    - malware_infection
    - phishing_campaign
    - unauthorized_access
    - data_exfiltration
    - denial_of_service
    - ransomware
    - insider_threat
    - vulnerability_exploitation
    - account_compromise
    - policy_violation

  severity_levels:
    critical:
      definition: "Active data breach, ransomware, or business-critical system compromise"
      response_sla: 15 minutes
      resolution_sla: 4 hours
      escalation: immediate to Tier 3 + CISO notification
      examples: ["Active ransomware", "Domain admin compromise", "Customer data breach"]

    high:
      definition: "Confirmed compromise of business systems or multiple user accounts"
      response_sla: 30 minutes
      resolution_sla: 8 hours
      escalation: Tier 2 immediate, Tier 3 if unresolved in 2 hours
      examples: ["Malware with C2", "Lateral movement detected", "Phishing with credential theft"]

    medium:
      definition: "Confirmed security event requiring investigation and remediation"
      response_sla: 2 hours
      resolution_sla: 24 hours
      escalation: Tier 2 within 4 hours
      examples: ["Single phishing click", "Unauthorized software", "Policy violation"]

    low:
      definition: "Minor security event with limited impact"
      response_sla: 8 hours
      resolution_sla: 72 hours
      escalation: Tier 1 standard queue
      examples: ["Scan attempt", "Failed brute force (no compromise)", "Info disclosure"]

import requests
import json
from datetime import datetime

class IncidentTicketManager:
    def __init__(self, snow_url, snow_user, snow_password):
        self.snow_url = snow_url
        self.auth = (snow_user, snow_password)
        self.headers = {
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

    def create_incident(self, alert_data):
        """Create ServiceNow incident from SIEM alert"""
        severity_map = {
            "critical": "1",
            "high": "2",
            "medium": "3",
            "low": "4"
        }

        payload = {
            "short_description": f"[SEC] {alert_data['rule_name']} — {alert_data['src']}",
            "description": self._build_description(alert_data),
            "category": "Security",
            "subcategory": alert_data.get("category", "Investigation"),
            "urgency": severity_map.get(alert_data["severity"], "3"),
            "impact": severity_map.get(alert_data["severity"], "3"),
            "assignment_group": self._get_assignment_group(alert_data["severity"]),
            "caller_id": "soc_automation",
            "u_siem_event_id": alert_data.get("notable_id", ""),
            "u_mitre_technique": alert_data.get("mitre_technique", ""),
            "u_affected_hosts": ", ".join(alert_data.get("affected_hosts", [])),
            "u_iocs": json.dumps(alert_data.get("iocs", {}))
        }

        response = requests.post(
            f"{self.snow_url}/api/now/table/incident",
            auth=self.auth,
            headers=self.headers,
            json=payload
        )
        result = response.json()["result"]
        return {
            "ticket_number": result["number"],
            "sys_id": result["sys_id"],
            "state": result["state"]
        }

    def _build_description(self, alert_data):
        return f"""
SECURITY INCIDENT — Auto-generated from SIEM
================================================
Alert Rule:       {alert_data['rule_name']}
SIEM Event ID:    {alert_data.get('notable_id', 'N/A')}
Detection Time:   {alert_data['detection_time']}
Severity:         {alert_data['severity'].upper()}
MITRE ATT&CK:    {alert_data.get('mitre_technique', 'N/A')}

Source:           {alert_data.get('src', 'N/A')}
Destination:      {alert_data.get('dest', 'N/A')}
User:             {alert_data.get('user', 'N/A')}

Initial Context:
{alert_data.get('description', 'See SIEM for details.')}

IOCs:
{json.dumps(alert_data.get('iocs', {}), indent=2)}
"""

    def _get_assignment_group(self, severity):
        if severity in ("critical", "high"):
            return "SOC Tier 2"
        return "SOC Tier 1"

    def update_incident(self, ticket_number, updates):
        """Update an existing incident"""
        # First get sys_id from ticket number
        response = requests.get(
            f"{self.snow_url}/api/now/table/incident",
            auth=self.auth,
            headers=self.headers,
            params={"sysparm_query": f"number={ticket_number}", "sysparm_limit": 1}
        )
        sys_id = response.json()["result"][0]["sys_id"]

        # Update
        response = requests.patch(
            f"{self.snow_url}/api/now/table/incident/{sys_id}",
            auth=self.auth,
            headers=self.headers,
            json=updates
        )
        return response.json()["result"]

    def add_work_note(self, ticket_number, note):
        """Add investigation note to incident"""
        self.update_incident(ticket_number, {"work_notes": note})

    def escalate_incident(self, ticket_number, reason):
        """Escalate to next tier"""
        self.update_incident(ticket_number, {
            "assignment_group": "SOC Tier 3",
            "urgency": "1",
            "work_notes": f"ESCALATED: {reason}"
        })

    def resolve_incident(self, ticket_number, resolution):
        """Resolve and close incident"""
        self.update_incident(ticket_number, {
            "state": "6",  # Resolved
            "close_code": "Resolved",
            "close_notes": resolution,
            "u_incident_disposition": resolution.split(":")[0] if ":" in resolution else "Resolved"
        })

import requests

class TheHiveCaseManager:
    def __init__(self, thehive_url, api_key):
        self.url = thehive_url
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }

    def create_case(self, alert_data):
        """Create case in TheHive from SIEM alert"""
        case = {
            "title": f"[{alert_data['severity'].upper()}] {alert_data['rule_name']}",
            "description": self._build_markdown_description(alert_data),
            "severity": {"critical": 4, "high": 3, "medium": 2, "low": 1}.get(
                alert_data["severity"], 2
            ),
            "tlp": 2,  # TLP:AMBER
            "pap": 2,  # PAP:AMBER
            "tags": [
                alert_data.get("mitre_technique", ""),
                alert_data.get("category", ""),
                f"source:{alert_data.get('src', 'unknown')}"
            ],
            "tasks": self._generate_tasks(alert_data["severity"]),
            "customFields": {
                "siem-event-id": {"string": alert_data.get("notable_id", "")},
                "mitre-technique": {"string": alert_data.get("mitre_technique", "")},
                "detection-source": {"string": "Splunk ES"}
            }
        }

        response = requests.post(
            f"{self.url}/api/case",
            headers=self.headers,
            json=case
        )
        return response.json()

    def _generate_tasks(self, severity):
        """Generate investigation tasks based on severity"""
        tasks = [
            {"title": "Initial Triage", "group": "Phase 1", "description": "Review SIEM alert and validate findings"},
            {"title": "IOC Enrichment", "group": "Phase 1", "description": "Enrich all IOCs with VT, AbuseIPDB"},
            {"title": "Scope Assessment", "group": "Phase 2", "description": "Determine affected systems and users"},
        ]
        if severity in ("critical", "high"):
            tasks.extend([
                {"title": "Containment Actions", "group": "Phase 2", "description": "Isolate affected systems"},
                {"title": "Evidence Collection", "group": "Phase 3", "description": "Preserve forensic artifacts"},
                {"title": "Eradication", "group": "Phase 3", "description": "Remove threat from environment"},
                {"title": "Recovery", "group": "Phase 4", "description": "Restore systems to normal operations"},
                {"title": "Post-Incident Review", "group": "Phase 4", "description": "Document lessons learned"},
            ])
        else:
            tasks.append(
                {"title": "Resolution and Documentation", "group": "Phase 2", "description": "Document findings and close"}
            )
        return tasks

    def add_observable(self, case_id, ioc_type, ioc_value, description=""):
        """Add IOC observable to case"""
        observable = {
            "dataType": ioc_type,
            "data": ioc_value,
            "message": description,
            "tlp": 2,
            "ioc": True,
            "tags": ["auto-extracted"]
        }
        response = requests.post(
            f"{self.url}/api/case/{case_id}/artifact",
            headers=self.headers,
            json=observable
        )
        return response.json()

--- Active incidents approaching SLA breach
index=servicenow sourcetype="snow:incident" category="Security" state IN ("New", "In Progress")
| eval sla_minutes = case(
    urgency="1", 15,
    urgency="2", 30,
    urgency="3", 120,
    urgency="4", 480
  )
| eval age_minutes = round((now() - strptime(opened_at, "%Y-%m-%d %H:%M:%S")) / 60, 0)
| eval sla_remaining = sla_minutes - age_minutes
| eval sla_status = case(
    sla_remaining < 0, "BREACHED",
    sla_remaining < sla_minutes * 0.25, "AT RISK",
    1=1, "ON TRACK"
  )
| where sla_status IN ("BREACHED", "AT RISK")
| sort sla_remaining
| table number, short_description, urgency, assignment_group, assigned_to,
        age_minutes, sla_minutes, sla_remaining, sla_status

def check_sla_breaches(ticket_manager):
    """Check for SLA breaches and auto-escalate"""
    open_incidents = ticket_manager.get_open_incidents()

    for incident in open_incidents:
        age_minutes = (datetime.utcnow() - incident["opened_at"]).total_seconds() / 60
        sla_minutes = {"1": 15, "2": 30, "3": 120, "4": 480}[incident["urgency"]]

        if age_minutes > sla_minutes and incident["state"] == "New":
            ticket_manager.escalate_incident(
                incident["number"],
                f"SLA BREACH: {int(age_minutes)}min elapsed, {sla_minutes}min SLA. Auto-escalating."
            )

--- Monthly incident metrics
index=servicenow sourcetype="snow:incident" category="Security"
opened_at > "2024-03-01" opened_at < "2024-04-01"
| stats count AS total,
        avg(eval((resolved_at - opened_at) / 3600)) AS avg_resolution_hours,
        sum(eval(if(urgency="1", 1, 0))) AS critical,
        sum(eval(if(urgency="2", 1, 0))) AS high,
        sum(eval(if(urgency="3", 1, 0))) AS medium,
        sum(eval(if(urgency="4", 1, 0))) AS low
| eval avg_resolution = round(avg_resolution_hours, 1)

--- SLA compliance rate
index=servicenow sourcetype="snow:incident" category="Security" state="Resolved"
| eval sla_target = case(urgency="1", 4, urgency="2", 8, urgency="3", 24, urgency="4", 72)
| eval resolution_hours = (resolved_at - opened_at) / 3600
| eval sla_met = if(resolution_hours <= sla_target, 1, 0)
| stats sum(sla_met) AS met, count AS total
| eval compliance_pct = round(met / total * 100, 1)