Feedback for ransomware correlation using Elastic

Hello everyone,

I'm currently working on an Alert/workflow that aims to proactively trigger when a ransomware pattern is detected. Our SOC uses Elastic Security as both the SIEM and EDR.

To build it, I started from real ransomware incidents we've handled in our SOC, along with the MITRE ATT&CK Top 10 ransomware techniques:

Rather than manually reviewing and correlating multiple alerts to determine whether a ransomware attack is actually in progress, I wanted to estimate a confidence score based on the observed MITRE ATT&CK techniques.

So far, this is the best approach I've come up with, but I'm not sure it's the right one. One concern is that some ransomware attacks can unfold over several days or even weeks, making it difficult to determine the most appropriate correlation window and query strategy.

I'd really appreciate your feedback. Has anyone implemented something similar ? How do you approach ransomware detection in your environment ? If you see any weaknesses or blind spots in this correlation approach, I'd really appreciate your feedback.

Here is the ES|QL query I'm currently using:

FROM
 .alerts-security.alerts-* 
METADATA
 _id
| 
WHERE
 `@timestamp` > NOW() - 1 HOURS
| 
WHERE
 `kibana.alert.workflow_status` != "closed"
  AND `host.id` IS NOT NULL


| 
EVAL
 technique_str = CONCAT(
    COALESCE(MV_CONCAT(`kibana.alert.rule.threat.technique.id`, ","), ""),
    ",",
    COALESCE(MV_CONCAT(`kibana.alert.rule.threat.technique.subtechnique.id`, ","), "")
)


| 
EVAL
 score_T1059 = CASE(technique_str LIKE "*T1059*", 5, 0)
| 
EVAL
 score_T1078 = CASE(technique_str LIKE "*T1078*", 25, 0)
| 
EVAL
 score_T1021_001 = CASE(technique_str LIKE "*T1021.001*", 40, 0)
| 
EVAL
 score_T1047 = CASE(technique_str LIKE "*T1047*", 15, 0)
| 
EVAL
 score_T1490 = CASE(technique_str LIKE "*T1490*", 50, 0)
| 
EVAL
 score_T1105 = CASE(technique_str LIKE "*T1105*", 30, 0)
| 
EVAL
 score_T1083 = CASE(technique_str LIKE "*T1083*", 5, 0)
| 
EVAL
 score_T1486 = CASE(technique_str LIKE "*T1486*", 100, 0)
| 
EVAL
 score_T1190 = CASE(technique_str LIKE "*T1190*", 50, 0)
| 
EVAL
 score_T1489 = CASE(technique_str LIKE "*T1489*", 15, 0)


| 
STATS
    score_T1059 = MAX(score_T1059),
    score_T1078 = MAX(score_T1078),
    score_T1021_001 = MAX(score_T1021_001),
    score_T1047 = MAX(score_T1047),
    score_T1490 = MAX(score_T1490),
    score_T1105 = MAX(score_T1105),
    score_T1083 = MAX(score_T1083),
    score_T1486 = MAX(score_T1486),
    score_T1190 = MAX(score_T1190),
    score_T1489 = MAX(score_T1489),
    techniques = VALUES(technique_str),
    rules = VALUES(`kibana.alert.rule.name`),
    source_alert_ids = VALUES(_id),
    users = VALUES(`user.name`),
    alert_count = COUNT()
  
BY
 `host.id`, `host.name`


| 
EVAL
 ransomware_score =
    score_T1059 +
    score_T1078 +
    score_T1021_001 +
    score_T1047 +
    score_T1490 +
    score_T1105 +
    score_T1083 +
    score_T1486 +
    score_T1190 +
    score_T1489


| 
WHERE
 ransomware_score >= 60
| 
SORT
 ransomware_score DESC

And here is an example of the output produced for a ransomware test:

{

"score_T1059": 5,

"score_T1078": 0,

"score_T1021_001": 0,

"score_T1047": 0,

"score_T1490": 50,

"score_T1105": 30,

"score_T1083": 0,

"score_T1486": 0,

"score_T1190": 0,

"score_T1489": 0,

"techniques": [

"T1112,T1562,T1562.001",

"T1112,T1562,T1562.001,T1562.006",

"T1204,T1204.002",

"T1490",

"T1070,T1070.001",

"T1027,T1059,T1105,T1140,T1027.010,T1059.001"

],

"rules": [

"Malicious Behavior Detection Alert: Suspicious PowerShell Base64 Decoding",

"Malicious Behavior Detection Alert: Potential Privilege Escalation via Token Impersonation",

"Malicious Behavior Detection Alert: Privilege Escalation via EXTENDED STARTUPINFO",

"Malware Detection Alert",

"Microsoft Windows Defender Tampering",

"Windows Defender Disabled via Registry Modification",

"Volume Shadow Copy Deleted or Resized via VssAdmin",

"Windows Event Logs Cleared",

"Suspicious Windows Powershell Arguments"

],

"source_alert_ids": [

"alert-id-redacted-001",

"alert-id-redacted-002",

"alert-id-redacted-003",

"..."

],

"users": [

"Administrator",

"SYSTEM"

],

"alert_count": 90,

"host.id": "endpoint-id-redacted",

"host.name": "example-server-01",

"ransomware_score": 85

}

Your approach looks solid. I'd suggest using multiple correlation windows with score decay instead of a fixed 1-hour window, since many ransomware attacks unfold over days. Also consider adjusting scores based on the sequence of techniques and adding behavioral indicators to reduce false positives.

Thanks a lot for the advice!

Your input helped me a lot to improve my detection rule: decay factors, strict sequencing between kill chain phases, and calibration against 15 days of real data.

That's exactly what I was missing.

I will also add an IOC hunt to my workflow to help reduce false positives.

Here's the result (AI-assisted implementation). hope it can help:

FROM .alerts-security.alerts-* METADATA _id
| WHERE `@timestamp` > NOW() - 15 DAYS
  AND `host.id` IS NOT NULL
  AND `kibana.alert.building_block_type` IS NULL


| EVAL technique_str = CONCAT(
    COALESCE(
        MV_CONCAT(`kibana.alert.rule.threat.technique.id`, ","),
        ""
    ),
    ",",
    COALESCE(
        MV_CONCAT(`kibana.alert.rule.threat.technique.subtechnique.id`, ","),
        ""
    )
)


| WHERE
    technique_str LIKE "*T1059*" OR
    technique_str LIKE "*T1078*" OR
    technique_str LIKE "*T1021.001*" OR
    technique_str LIKE "*T1047*" OR
    technique_str LIKE "*T1490*" OR
    technique_str LIKE "*T1105*" OR
    technique_str LIKE "*T1083*" OR
    technique_str LIKE "*T1486*" OR
    technique_str LIKE "*T1190*" OR
    technique_str LIKE "*T1489*"


| EVAL decay_factor = CASE(
    `@timestamp` > NOW() - 1 HOUR, 1.0,
    `@timestamp` > NOW() - 1 DAY, 0.80,
    `@timestamp` > NOW() - 3 DAYS, 0.60,
    `@timestamp` > NOW() - 7 DAYS, 0.40,
    `@timestamp` > NOW() - 15 DAYS, 0.20,
    0.0
)


| EVAL score_T1059 = CASE(
    technique_str LIKE "*T1059*", 5.0 * decay_factor,
    0.0
)
| EVAL score_T1078 = CASE(
    technique_str LIKE "*T1078*", 25.0 * decay_factor,
    0.0
)
| EVAL score_T1021_001 = CASE(
    technique_str LIKE "*T1021.001*", 40.0 * decay_factor,
    0.0
)
| EVAL score_T1047 = CASE(
    technique_str LIKE "*T1047*", 15.0 * decay_factor,
    0.0
)
| EVAL score_T1490 = CASE(
    technique_str LIKE "*T1490*", 50.0 * decay_factor,
    0.0
)
| EVAL score_T1105 = CASE(
    technique_str LIKE "*T1105*", 30.0 * decay_factor,
    0.0
)
| EVAL score_T1083 = CASE(
    technique_str LIKE "*T1083*", 5.0 * decay_factor,
    0.0
)
| EVAL score_T1486 = CASE(
    technique_str LIKE "*T1486*", 100.0 * decay_factor,
    0.0
)
| EVAL score_T1190 = CASE(
    technique_str LIKE "*T1190*", 50.0 * decay_factor,
    0.0
)
| EVAL score_T1489 = CASE(
    technique_str LIKE "*T1489*", 15.0 * decay_factor,
    0.0
)


| EVAL ts_initial_access = CASE(
    technique_str LIKE "*T1190*"
      OR technique_str LIKE "*T1078*",
    `@timestamp`,
    NULL
)

| EVAL ts_execution = CASE(
    technique_str LIKE "*T1059*"
      OR technique_str LIKE "*T1047*"
      OR technique_str LIKE "*T1021.001*",
    `@timestamp`,
    NULL
)

| EVAL ts_preparation = CASE(
    technique_str LIKE "*T1105*"
      OR technique_str LIKE "*T1083*",
    `@timestamp`,
    NULL
)

| EVAL ts_impact_preparation = CASE(
    technique_str LIKE "*T1489*"
      OR technique_str LIKE "*T1490*",
    `@timestamp`,
    NULL
)

| EVAL ts_encryption = CASE(
    technique_str LIKE "*T1486*",
    `@timestamp`,
    NULL
)


| EVAL is_closed = CASE(
    `kibana.alert.workflow_status` == "closed",
    1,
    0
)

| EVAL is_active = CASE(
    `kibana.alert.workflow_status` != "closed",
    1,
    0
)


| EVAL recent_active_contributor = CASE(
    `@timestamp` > NOW() - 1 HOUR
    AND `kibana.alert.workflow_status` != "closed",
    1,
    0
)

| STATS
    score_T1059 = MAX(score_T1059),
    score_T1078 = MAX(score_T1078),
    score_T1021_001 = MAX(score_T1021_001),
    score_T1047 = MAX(score_T1047),
    score_T1490 = MAX(score_T1490),
    score_T1105 = MAX(score_T1105),
    score_T1083 = MAX(score_T1083),
    score_T1486 = MAX(score_T1486),
    score_T1190 = MAX(score_T1190),
    score_T1489 = MAX(score_T1489),

    first_initial_access = MIN(ts_initial_access),
    first_execution = MIN(ts_execution),
    first_preparation = MIN(ts_preparation),
    first_impact_preparation = MIN(ts_impact_preparation),
    first_encryption = MIN(ts_encryption),

    recent_active_contributor = MAX(recent_active_contributor),

    first_seen = MIN(`@timestamp`),
    last_seen = MAX(`@timestamp`),

    techniques = VALUES(technique_str),
    rules = VALUES(`kibana.alert.rule.name`),
    source_alert_ids = VALUES(_id),
    users = VALUES(`user.name`),
    workflow_statuses = VALUES(`kibana.alert.workflow_status`),

    alert_count = COUNT(),
    closed_alert_count = SUM(is_closed),
    active_alert_count = SUM(is_active)

  BY `host.id`, `host.name`

| EVAL base_ransomware_score =
    score_T1059 +
    score_T1078 +
    score_T1021_001 +
    score_T1047 +
    score_T1490 +
    score_T1105 +
    score_T1083 +
    score_T1486 +
    score_T1190 +
    score_T1489


| EVAL sequence_bonus = CASE(
    first_initial_access IS NOT NULL
      AND first_execution IS NOT NULL
      AND first_preparation IS NOT NULL
      AND first_impact_preparation IS NOT NULL
      AND first_encryption IS NOT NULL
      AND first_initial_access < first_execution
      AND first_execution < first_preparation
      AND first_preparation < first_impact_preparation
      AND first_impact_preparation < first_encryption,
    20,

    first_execution IS NOT NULL
      AND first_preparation IS NOT NULL
      AND first_impact_preparation IS NOT NULL
      AND first_execution < first_preparation
      AND first_preparation < first_impact_preparation,
    15,

    first_initial_access IS NOT NULL
      AND first_execution IS NOT NULL
      AND first_preparation IS NOT NULL
      AND first_initial_access < first_execution
      AND first_execution < first_preparation,
    10,

    (
        first_initial_access IS NOT NULL
        AND first_execution IS NOT NULL
        AND first_initial_access < first_execution
    )
    OR (
        first_execution IS NOT NULL
        AND first_preparation IS NOT NULL
        AND first_execution < first_preparation
    )
    OR (
        first_preparation IS NOT NULL
        AND first_impact_preparation IS NOT NULL
        AND first_preparation < first_impact_preparation
    )
    OR (
        first_impact_preparation IS NOT NULL
        AND first_encryption IS NOT NULL
        AND first_impact_preparation < first_encryption
    ),
    5,

    0
)

| EVAL ransomware_score =
    base_ransomware_score + sequence_bonus


| WHERE ransomware_score >= 60

| WHERE recent_active_contributor == 1

| SORT ransomware_score DESC

Really like this idea. Correlating multiple detections into a host-level confidence score is much more useful than looking at individual alerts.

One thought: how do you see this fitting alongside Entity Analytics? Is the score intentionally host-centric, or could it evolve into an incident-level risk score that also incorporates user, identity, asset criticality, and related activity across multiple hosts? That could make it even more valuable for prioritization, not just detection.

I'd also consider weighting detections by their fidelity, not just their ATT&CK technique. For example, an Elastic Defend behavioral ransomware alert should probably contribute much more than a generic PowerShell detection.