Install this skill

npx skills add mukul975/Anthropic-Cybersecurity-Skills

Framework mappings

Overview

Alert triage in Elastic Security is the systematic process of reviewing, classifying, and prioritizing security alerts to determine which represent genuine threats. Elastic's AI-driven Attack Discovery feature can triage hundreds of alerts down to discrete attack chains, but skilled analyst triage remains essential. A structured triage workflow typically takes 5-10 minutes per alert cluster using Elastic's built-in tools.

When to Use

When conducting security assessments that involve performing alert triage with elastic siem
When following incident response procedures for related security events
When performing scheduled security testing or auditing activities
When validating security controls through hands-on testing

Prerequisites

Elastic Security deployed (version 8.x or later)
Elastic Agent or Beats configured for endpoint and network data collection
Detection rules enabled and generating alerts
Elastic Common Schema (ECS) compliance across data sources
Analyst access to Kibana Security app with appropriate privileges

Alert Triage Workflow

Step 1: Initial Alert Assessment (2 minutes)

When viewing an alert in Elastic Security, review the alert details panel:

Alert Details Panel:
- Rule Name and Description
- Severity and Risk Score
- MITRE ATT&CK Mapping
- Host and User Context
- Process Tree (for endpoint alerts)
- Timeline of related events

Key Fields to Examine First

Field	Purpose	ECS Field
Rule severity	Initial priority assessment	`kibana.alert.severity`
Risk score	Quantified threat level	`kibana.alert.risk_score`
Host name	Affected system	`host.name`
User name	Affected identity	`user.name`
Process name	Executing process	`process.name`
Source IP	Origin of activity	`source.ip`
Destination IP	Target of activity	`destination.ip`
MITRE tactic	Attack stage	`threat.tactic.name`

Step 2: Context Gathering (3 minutes)

Query Related Events with ES|QL

FROM logs-endpoint.events.*
| WHERE host.name == "affected-host" AND @timestamp > NOW() - 1 HOUR
| STATS count = COUNT(*) BY event.category, event.action
| SORT count DESC

Find All Activity from Suspicious User

FROM logs-*
| WHERE user.name == "suspicious-user" AND @timestamp > NOW() - 24 HOURS
| STATS count = COUNT(*), unique_hosts = COUNT_DISTINCT(host.name) BY event.category
| SORT count DESC

Check for Related Alerts from Same Source

FROM .alerts-security.alerts-default
| WHERE source.ip == "10.0.0.50" AND @timestamp > NOW() - 24 HOURS
| STATS alert_count = COUNT(*) BY kibana.alert.rule.name, kibana.alert.severity
| SORT alert_count DESC

Investigate Lateral Movement from Same IP

FROM logs-system.auth-*
| WHERE source.ip == "10.0.0.50" AND event.outcome == "success"
| STATS login_count = COUNT(*), hosts = COUNT_DISTINCT(host.name) BY user.name
| WHERE hosts > 3

Step 3: Threat Intelligence Enrichment (2 minutes)

Check indicators against threat intelligence:

FROM logs-ti_*
| WHERE threat.indicator.ip == "203.0.113.50"
| KEEP threat.indicator.type, threat.indicator.provider, threat.indicator.confidence, threat.feed.name

Check File Hash Against Known Threats

FROM logs-endpoint.events.file-*
| WHERE file.hash.sha256 == "abc123..."
| STATS occurrences = COUNT(*) BY host.name, file.path, user.name

Step 4: Classification Decision (2 minutes)

Classification	Criteria	Action
True Positive	Confirmed malicious activity	Escalate to incident, begin containment
Benign True Positive	Expected behavior matching rule	Document in alert notes, acknowledge
False Positive	Rule triggered on benign activity	Mark as false positive, create tuning task
Needs Investigation	Insufficient data for determination	Assign for deeper investigation

Step 5: Documentation and Escalation (1 minute)

For each triaged alert, document:

Classification decision with rationale
Evidence artifacts examined
Related alerts or investigations
Recommended next steps

Detection Rules for Triage

Pre-Built Detection Rules

Elastic Security includes 1000+ pre-built detection rules organized by:

MITRE ATT&CK Tactic: Initial Access, Execution, Persistence, etc.
Platform: Windows, Linux, macOS, Cloud
Data Source: Endpoint, Network, Cloud, Identity

Custom Alert Correlation Rule

{
  "name": "Multiple Failed Logins Followed by Success",
  "type": "threshold",
  "query": "event.category:authentication AND event.outcome:failure",
  "threshold": {
    "field": ["source.ip", "user.name"],
    "value": 5,
    "cardinality": [
      {
        "field": "user.name",
        "value": 3
      }
    ]
  },
  "severity": "high",
  "risk_score": 73,
  "threat": [
    {
      "framework": "MITRE ATT&CK",
      "tactic": {
        "id": "TA0006",
        "name": "Credential Access"
      },
      "technique": [
        {
          "id": "T1110",
          "name": "Brute Force"
        }
      ]
    }
  ]
}

AI-Assisted Triage

Elastic AI Assistant Integration

Open alert in Elastic Security
Click AI Assistant panel
Use quick prompts:
- "Summarize this alert" - Get initial assessment
- "Generate ES|QL query to find related activity" - Expand investigation
- "What are the recommended response actions?" - Get playbook guidance
- "Is this likely a false positive?" - Get AI confidence assessment

Attack Discovery

Elastic's Attack Discovery automatically:

Groups related alerts into attack chains
Maps alerts to MITRE ATT&CK kill chain stages
Filters false positives using ML models
Prioritizes based on business impact
Provides narrative summary of the attack

Triage Prioritization Matrix

Risk Score	Severity	Asset Criticality	Response SLA
90-100	Critical	High	15 minutes
70-89	High	High	30 minutes
70-89	High	Medium	1 hour
50-69	Medium	Any	4 hours
21-49	Low	Any	8 hours
1-20	Informational	Any	24 hours

Triage Metrics and KPIs

Metric	Target	Measurement
Mean Time to Triage (MTTT)	< 10 minutes	Time from alert creation to classification
False Positive Rate	< 30%	False positives / total alerts
Escalation Rate	10-20%	Escalated alerts / total alerts
Alert Coverage	> 80%	Triaged alerts / generated alerts per shift
Reclassification Rate	< 5%	Changed classifications / total classified

References

Source materials

References and resources

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

References 3

api-reference.md2.8 KB

Alert Triage with Elastic SIEM - API Reference

elasticsearch-py Client

Connection

from elasticsearch import Elasticsearch
es = Elasticsearch(
    hosts=["https://elastic:9200"],
    api_key="base64-api-key",
    verify_certs=True
)

SIEM Signals Index

Elastic Security stores alerts in .siem-signals-<space>-* indices.

Querying Alerts

Search Open Alerts

es.search(
    index=".siem-signals-*",
    query={"bool": {"must": [
        {"range": {"@timestamp": {"gte": "now-24h"}}},
        {"term": {"signal.status": "open"}}
    ]}},
    sort=[{"@timestamp": {"order": "desc"}}],
    size=500
)

Alert Fields

Field	Path	Description
Rule name	`signal.rule.name`	Detection rule that triggered
Rule ID	`signal.rule.id`	Unique rule identifier
Severity	`signal.rule.severity`	critical, high, medium, low
Risk score	`signal.rule.risk_score`	0-100 numeric score
Status	`signal.status`	open, acknowledged, closed
Source IP	`source.ip`	Alert source address
Destination IP	`destination.ip`	Alert destination address
User	`user.name`	Associated username
Host	`host.name`	Affected hostname
Process	`process.name`	Triggering process

Aggregations

es.search(
    index=".siem-signals-*",
    query={"bool": {"must": [...]}},
    aggs={
        "by_severity": {"terms": {"field": "signal.rule.severity", "size": 10}},
        "by_rule": {"terms": {"field": "signal.rule.name.keyword", "size": 20}},
        "by_host": {"terms": {"field": "host.name.keyword", "size": 20}}
    },
    size=0
)

Alert Status Management

Update Alert Status

es.update(
    index=".siem-signals-default-000001",
    id="alert_doc_id",
    body={"doc": {"signal": {"status": "closed"}}}
)

Triage Prioritization

Severity Priority

Critical (risk score 90-100)
High (risk score 70-89)
Medium (risk score 40-69)
Low (risk score 0-39)

Alert Clustering

Alerts from the same host within a time window are grouped as potential incidents. Three or more alerts from the same host suggest a multi-stage attack.

Elastic Security API

List Detection Rules

GET /api/detection_engine/rules/_find?per_page=100

Get Rule Execution Status

GET /api/detection_engine/rules/_find_statuses

Output Schema

{
  "report": "elastic_siem_alert_triage",
  "total_open_alerts": 45,
  "severity_summary": {"critical": 3, "high": 12, "medium": 20, "low": 10},
  "alert_clusters": [{"host": "web01", "alert_count": 5, "max_severity": "high"}],
  "aggregations": {"by_severity": [{"key": "high", "count": 12}]}
}

CLI Usage

python agent.py --host https://elastic:9200 --api-key "key" --hours 24 --output report.json

standards.md1.7 KB

Standards and References - Alert Triage with Elastic SIEM

Elastic Common Schema (ECS)

ECS is a standardized field naming convention for Elasticsearch data. All Elastic Security detections and triage workflows rely on ECS compliance.

Key ECS Field Categories for Triage

Category	Fields	Usage
Base	`@timestamp`, `message`, `tags`	Event timing and classification
Agent	`agent.name`, `agent.type`	Data source identification
Host	`host.name`, `host.ip`, `host.os`	Affected system context
User	`user.name`, `user.domain`	Identity attribution
Process	`process.name`, `process.pid`, `process.command_line`	Execution context
Network	`source.ip`, `destination.ip`, `destination.port`	Network activity
File	`file.name`, `file.hash.sha256`, `file.path`	File-related events
Threat	`threat.tactic.name`, `threat.technique.id`	MITRE ATT&CK mapping

MITRE ATT&CK Integration

Elastic Security maps detection rules and alerts to MITRE ATT&CK tactics and techniques, providing a common taxonomy for triage prioritization.

NIST SP 800-61 Rev 2

Triage aligns with NIST incident handling phases:

Detection and Analysis (triage is the core of this phase)
Prioritization based on functional impact, information impact, and recoverability

SOC Maturity Model

Triage Capability Levels

Level	Capability
Level 1	Manual review of individual alerts
Level 2	Grouped alert triage with correlation
Level 3	AI-assisted triage with automated enrichment
Level 4	Automated classification with human oversight
Level 5	Fully autonomous triage with exception-based review

workflows.md1.6 KB

Workflows - Alert Triage with Elastic SIEM

5-Step Rapid Triage Framework

1. Alert Reception (30 seconds)
   - Review alert title, severity, risk score
   - Check MITRE ATT&CK mapping
   |
   v
2. Context Assessment (2 minutes)
   - Examine affected host and user
   - Check asset criticality
   - Review process tree for endpoint alerts
   |
   v
3. Intelligence Enrichment (2 minutes)
   - Check threat intelligence feeds
   - Query for related alerts (same source/user)
   - Search for known IOCs
   |
   v
4. Classification (1 minute)
   - True Positive / False Positive / Needs Investigation
   - Assign confidence level
   |
   v
5. Action (2 minutes)
   - Document findings in alert notes
   - Escalate or close with rationale
   - Create tuning task if false positive

Alert Grouping Strategy

Smart Grouping Criteria

Time window: Group alerts within 15-minute windows
Entity: Group by affected host or user
Kill chain stage: Group by MITRE ATT&CK tactic
Source: Group by originating IP or detection rule

Group Triage Process

Sort alert groups by highest severity member
Triage group as single unit when correlated
Escalate entire group if attack chain detected
Close group if false positive pattern identified

Shift-Based Triage Queue Management

Queue Priority	Alert Criteria	Analyst Tier
Immediate	Critical severity, critical assets	Tier 2+
High	High severity or multiple related alerts	Tier 1/2
Standard	Medium severity, standard assets	Tier 1
Low	Low/info severity, non-critical	Tier 1 (batch review)

Scripts 2

agent.py6.6 KB

Display-only source. This catalog never executes bundled scripts.

#!/usr/bin/env python3
"""Alert Triage with Elastic SIEM agent - queries Elasticsearch SIEM signals
index to retrieve, prioritize, and triage security alerts using
elasticsearch-py client."""

import argparse
import json
import sys
from collections import Counter
from datetime import datetime
from pathlib import Path

try:
    from elasticsearch import Elasticsearch
except ImportError:
    print("Install elasticsearch: pip install elasticsearch", file=sys.stderr)
    sys.exit(1)


SIEM_SIGNALS_INDEX = ".siem-signals-*"

SEVERITY_PRIORITY = {"critical": 1, "high": 2, "medium": 3, "low": 4}


def create_client(host: str, api_key: str = None, username: str = None,
                  password: str = None, verify_certs: bool = True) -> Elasticsearch:
    """Create Elasticsearch client."""
    kwargs = {"hosts": [host], "verify_certs": verify_certs}
    if api_key:
        kwargs["api_key"] = api_key
    elif username and password:
        kwargs["basic_auth"] = (username, password)
    return Elasticsearch(**kwargs)


def get_open_alerts(es: Elasticsearch, hours_back: int = 24,
                    severity: list[str] = None, size: int = 500) -> list[dict]:
    """Retrieve open SIEM alerts from the signals index."""
    must_clauses = [
        {"range": {"@timestamp": {"gte": f"now-{hours_back}h", "lte": "now"}}},
        {"term": {"signal.status": "open"}},
    ]
    if severity:
        must_clauses.append({"terms": {"signal.rule.severity": severity}})

    query = {"bool": {"must": must_clauses}}
    response = es.search(index=SIEM_SIGNALS_INDEX, query=query, size=size,
                         sort=[{"@timestamp": {"order": "desc"}}])
    alerts = []
    for hit in response["hits"]["hits"]:
        src = hit["_source"]
        signal = src.get("signal", {})
        rule = signal.get("rule", {})
        alerts.append({
            "alert_id": hit["_id"],
            "timestamp": src.get("@timestamp", ""),
            "rule_name": rule.get("name", ""),
            "rule_id": rule.get("id", ""),
            "severity": rule.get("severity", "unknown"),
            "risk_score": rule.get("risk_score", 0),
            "status": signal.get("status", ""),
            "source_ip": src.get("source", {}).get("ip", ""),
            "destination_ip": src.get("destination", {}).get("ip", ""),
            "user": src.get("user", {}).get("name", ""),
            "host": src.get("host", {}).get("name", ""),
            "process": src.get("process", {}).get("name", ""),
        })
    return alerts


def get_alert_aggregations(es: Elasticsearch, hours_back: int = 24) -> dict:
    """Aggregate alerts by rule, severity, and host."""
    query = {
        "bool": {
            "must": [
                {"range": {"@timestamp": {"gte": f"now-{hours_back}h"}}},
                {"term": {"signal.status": "open"}},
            ]
        }
    }
    aggs = {
        "by_severity": {"terms": {"field": "signal.rule.severity", "size": 10}},
        "by_rule": {"terms": {"field": "signal.rule.name.keyword", "size": 20}},
        "by_host": {"terms": {"field": "host.name.keyword", "size": 20}},
        "by_user": {"terms": {"field": "user.name.keyword", "size": 20}},
    }
    response = es.search(index=SIEM_SIGNALS_INDEX, query=query, aggs=aggs, size=0)
    result = {}
    for agg_name, agg_data in response.get("aggregations", {}).items():
        result[agg_name] = [
            {"key": bucket["key"], "count": bucket["doc_count"]}
            for bucket in agg_data.get("buckets", [])
        ]
    return result


def prioritize_alerts(alerts: list[dict]) -> list[dict]:
    """Sort and prioritize alerts by severity and risk score."""
    return sorted(alerts, key=lambda a: (
        SEVERITY_PRIORITY.get(a.get("severity", "low"), 5),
        -a.get("risk_score", 0)
    ))


def identify_alert_clusters(alerts: list[dict]) -> list[dict]:
    """Group related alerts that may represent a single incident."""
    clusters = []
    by_host = {}
    for alert in alerts:
        host = alert.get("host", "unknown")
        if host not in by_host:
            by_host[host] = []
        by_host[host].append(alert)

    for host, host_alerts in by_host.items():
        if len(host_alerts) >= 3:
            rules = list(set(a["rule_name"] for a in host_alerts))
            max_severity = min(host_alerts, key=lambda a: SEVERITY_PRIORITY.get(a.get("severity", "low"), 5))
            clusters.append({
                "host": host,
                "alert_count": len(host_alerts),
                "unique_rules": len(rules),
                "rules": rules[:10],
                "max_severity": max_severity["severity"],
                "recommendation": "Investigate as potential incident - multiple alerts on same host",
            })
    return clusters


def generate_report(host: str, api_key: str = None, username: str = None,
                    password: str = None, hours_back: int = 24,
                    severity: list[str] = None) -> dict:
    """Run alert triage and build consolidated report."""
    es = create_client(host, api_key, username, password)
    alerts = get_open_alerts(es, hours_back, severity)
    prioritized = prioritize_alerts(alerts)
    aggregations = get_alert_aggregations(es, hours_back)
    clusters = identify_alert_clusters(alerts)

    severity_counts = Counter(a["severity"] for a in alerts)
    return {
        "report": "elastic_siem_alert_triage",
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "time_window_hours": hours_back,
        "total_open_alerts": len(alerts),
        "severity_summary": dict(severity_counts),
        "alert_clusters": clusters,
        "aggregations": aggregations,
        "prioritized_alerts": prioritized[:50],
    }


def main():
    parser = argparse.ArgumentParser(description="Elastic SIEM Alert Triage Agent")
    parser.add_argument("--host", required=True, help="Elasticsearch URL")
    parser.add_argument("--api-key", help="Elasticsearch API key")
    parser.add_argument("--username", help="Elasticsearch username")
    parser.add_argument("--password", help="Elasticsearch password")
    parser.add_argument("--hours", type=int, default=24, help="Hours to look back (default: 24)")
    parser.add_argument("--severity", nargs="+", help="Filter by severity levels")
    parser.add_argument("--output", help="Output JSON file path")
    args = parser.parse_args()

    report = generate_report(args.host, args.api_key, args.username,
                             args.password, args.hours, args.severity)
    output = json.dumps(report, indent=2)
    if args.output:
        Path(args.output).write_text(output, encoding="utf-8")
        print(f"Report written to {args.output}")
    else:
        print(output)


if __name__ == "__main__":
    main()

process.py11.2 KB