AWS SSM Distributed Command

Description

Detect an attacker utilizing AWS Systems Manager (SSM) to execute commands through SendCommand on multiple EC2 instances.

Query · python

import datetime as dt
import json

from panther_aws_helpers import aws_cloudtrail_success, aws_rule_context
from panther_core import PantherEvent
from panther_detection_helpers.caching import get_string_set, put_string_set

# Determine how separate instances need be commanded in order to trigger an alert
INSTANCE_THRESHOLD = 2

all_instance_ids = set()


def rule(event: PantherEvent) -> bool:
    # Exclude events of the wrong type
    if event.get("eventName") != "SendCommand":
        return False

    # Determine if this actor accessed any other params in this account
    key = get_cache_key(event)
    cached_ids = get_cached_instance_ids(key)
    target_instance_ids = set(event.deep_get("requestParameters", "instanceIds", default=[]))

    # Determine if the cache needs updating with new entries
    global all_instance_ids  # pylint: disable=global-statement
    all_instance_ids = cached_ids | target_instance_ids
    if all_instance_ids - cached_ids:
        # Only set the TTL if this is the first time we're adding to the cache
        #   Otherwise we'll be perpetually extending the lifespan of the cached data every time we
        #   add more.
        put_string_set(key, all_instance_ids, epoch_seconds=(3600 if not cached_ids else None))

    # Check combined number of params
    return len(all_instance_ids) > INSTANCE_THRESHOLD


def title(event: PantherEvent) -> str:
    actor = event.udm("actor_user")
    account_name = event.get("recipientAccountId")
    return f"Commands distributed to many EC2 instances by [{actor}] in [{account_name}]"


def severity(event: PantherEvent) -> str:
    # Demote to LOW if attempt was denied
    if not aws_cloudtrail_success(event):
        return "LOW"
    return "DEFAULT"


def alert_context(event: PantherEvent) -> dict:
    global all_instance_ids
    context = aws_rule_context(event)
    context.update({"instanceIds": list(all_instance_ids)})
    return context


def get_cache_key(event) -> str:
    """Use the field values in the event to generate a cache key unique to this actor and
    account ID."""
    offset = (
        dt.datetime.fromisoformat(event.get("p_event_time", "1970-01-01T00:00:00")).timestamp()
        // 3600
        * 3600
    )
    actor = event.udm("actor_user")
    account = event.get("recipientAccountId")
    rule_id = "AWS.SSM.DistributedCommand"
    return f"{rule_id}-{account}-{actor}-{offset}"


def get_cached_instance_ids(key: str) -> set[str]:
    """Get any previously cached parameter names. Included automatic converstion from string in
    the case of a unit test mock."""
    cached_ids = get_string_set(key, force_ttl_check=True)
    if isinstance(cached_ids, str):
        # This is a unit test
        cached_ids = set(json.loads(cached_ids))
    return cached_ids

Analyst notes

Detetmine who issued the command, the command content and arguments, and which EC2 instances were affected. Determine the risk of an attacker creating a persistent point of access within one of the instances. Review behaviour logs for the EC2 instances (and their associated IAM roles).

Detection tier

Atomic IOCs
Network & Host Artifacts
Tools
TTPs
Pending classification

Metadata

Source: panther-labs/panther-analysis
License: Apache 2.0
Severity: Informational
Status: Stable
Query type: python
Created: 2025-03-18
Modified: 2025-12-01
Authored by: panther-labs
Source ID
Upstream: rules/aws_cloudtrail_rules/aws_ssm_distributed_command.yml
Type: rule
Version: 3
References: https://stratus-red-team.cloud/attack-techniques/AWS/aws.execution.ssm-send-command/
Data sources: AWS.CloudTrail

ATT&CK

TA0002 T1203