Create Snapshot

// Detect suspicious cloud snapshot creation and sharing across Azure (AzureActivity) and AWS (AWSCloudTrail)
let LookbackWindow = 24h;
let BulkThreshold = 5;
// --- Azure: Snapshot creation via Resource Manager ---
let AzureSnapshots = AzureActivity
| where TimeGenerated > ago(LookbackWindow)
| where OperationNameValue =~ "MICROSOFT.COMPUTE/SNAPSHOTS/WRITE"
| where ActivityStatusValue =~ "Succeeded"
| extend SnapshotName = tostring(split(ResourceId, "/")[-1])
| extend SubscriptionId = tostring(split(ResourceId, "/")[2])
| extend ResourceGroupName = tostring(split(ResourceId, "/")[4])
| project TimeGenerated, CloudProvider="Azure",
          Caller, CallerIpAddress,
          ResourceId, SnapshotName, ResourceGroupName, SubscriptionId,
          OperationName=OperationNameValue,
          IsExternalShare=false;
// --- AWS: Snapshot creation and sharing via CloudTrail ---
let AWSSnapshots = AWSCloudTrail
| where TimeGenerated > ago(LookbackWindow)
| where EventName in ("CreateSnapshot", "CopySnapshot", "CreateDBSnapshot",
                       "CreateDBClusterSnapshot", "ModifySnapshotAttribute",
                       "ModifyDBSnapshotAttribute")
| extend SourceVolumeId = tostring(parse_json(RequestParameters).volumeId)
| extend SnapshotId = tostring(parse_json(ResponseElements).snapshotId)
| extend IsExternalShare = (EventName == "ModifySnapshotAttribute" and
         tostring(parse_json(RequestParameters).attributeType) == "createVolumePermission")
| project TimeGenerated, CloudProvider="AWS",
          Caller=UserIdentityArn, CallerIpAddress=SourceIpAddress,
          ResourceId=coalesce(SnapshotId, SourceVolumeId),
          SnapshotName=coalesce(SnapshotId, EventName),
          ResourceGroupName="", SubscriptionId=RecipientAccountId,
          OperationName=EventName, IsExternalShare;
// --- Combine and enrich ---
union AzureSnapshots, AWSSnapshots
| extend HourOfDay = hourofday(TimeGenerated)
| extend IsOffHours = (HourOfDay < 7 or HourOfDay > 19)
| extend IsWeekend = (dayofweek(TimeGenerated) == 0d or dayofweek(TimeGenerated) == 6d)
| summarize SnapshotCount=count(),
            FirstEvent=min(TimeGenerated),
            LastEvent=max(TimeGenerated),
            Operations=make_set(OperationName),
            SnapshotNames=make_set(SnapshotName, 10),
            SourceIPs=make_set(CallerIpAddress),
            ExternalShareCount=countif(IsExternalShare == true),
            OffHoursCount=countif(IsOffHours == true)
            by Caller, CloudProvider, SubscriptionId
| extend SuspicionScore = case(
    ExternalShareCount > 0, 3,
    SnapshotCount > BulkThreshold and OffHoursCount > 0, 3,
    SnapshotCount > BulkThreshold, 2,
    OffHoursCount > 0, 1,
    1)
| where SnapshotCount > 1 or ExternalShareCount > 0
| project FirstEvent, LastEvent, CloudProvider, Caller, SourceIPs,
          SnapshotCount, ExternalShareCount, OffHoursCount,
          Operations, SnapshotNames, SubscriptionId, SuspicionScore
| sort by SuspicionScore desc, SnapshotCount desc

index=cloud (sourcetype="aws:cloudtrail" OR sourcetype="azure:monitor:activity")
(
  (sourcetype="aws:cloudtrail"
   (EventName="CreateSnapshot" OR EventName="CopySnapshot" OR EventName="CreateDBSnapshot"
    OR EventName="CreateDBClusterSnapshot" OR EventName="ModifySnapshotAttribute"
    OR EventName="ModifyDBSnapshotAttribute")
  )
  OR
  (sourcetype="azure:monitor:activity"
   operationName="MICROSOFT.COMPUTE/SNAPSHOTS/WRITE" status="Succeeded"
  )
)
| eval CloudProvider=if(sourcetype="aws:cloudtrail", "AWS", "Azure")
| eval Caller=coalesce(userIdentity.arn, caller)
| eval CallerIP=coalesce(sourceIPAddress, callerIpAddress)
| eval Operation=coalesce(EventName, operationName)
| eval ResourceRef=coalesce(requestParameters.volumeId, resourceId)
| eval IsExternalShare=if(
    (Operation="ModifySnapshotAttribute" AND requestParameters.attributeType="createVolumePermission"),
    1, 0)
| eval HourOfDay=strftime(_time, "%H")
| eval IsOffHours=if(HourOfDay < "07" OR HourOfDay > "19", 1, 0)
| eval IsWeekend=if(strftime(_time, "%w")="0" OR strftime(_time, "%w")="6", 1, 0)
| stats count as SnapshotCount,
        earliest(_time) as FirstEvent,
        latest(_time) as LastEvent,
        values(Operation) as Operations,
        values(ResourceRef) as Resources,
        dc(CallerIP) as UniqueSourceIPs,
        sum(IsExternalShare) as ExternalShareCount,
        sum(IsOffHours) as OffHoursCount
        by Caller, CloudProvider
| eval SuspicionScore=case(
    ExternalShareCount > 0, 3,
    SnapshotCount > 5 AND OffHoursCount > 0, 3,
    SnapshotCount > 5, 2,
    OffHoursCount > 0, 1,
    true(), 1)
| where SnapshotCount > 1 OR ExternalShareCount > 0
| eval FirstEvent=strftime(FirstEvent, "%Y-%m-%d %H:%M:%S")
| eval LastEvent=strftime(LastEvent, "%Y-%m-%d %H:%M:%S")
| table FirstEvent, LastEvent, CloudProvider, Caller, SnapshotCount,
        ExternalShareCount, OffHoursCount, UniqueSourceIPs,
        Operations, Resources, SuspicionScore
| sort - SuspicionScore - SnapshotCount

any where event.dataset == "azure.activitylogs"
  and event.action : ("Microsoft.Compute/snapshots/write", "Microsoft.Compute/virtualMachines/write",
                      "Microsoft.Compute/virtualMachines/delete", "Microsoft.Compute/disks/write")
  and not user.name : ("azure-backup-service", "azure-site-recovery")

SELECT
  DATEFORMAT(devicetime,'yyyy-MM-dd HH:mm:ss') AS EventTime,
  username, "Operation" AS CloudOperation,
  "ResourceType" AS CloudResource,
  "ResourceGroup", "SubscriptionId",
  CASE
    WHEN "Operation" ILIKE '%delete%' OR "Operation" ILIKE '%destroy%' THEN 90
    WHEN "Operation" ILIKE '%snapshot%' AND "ResultType" = 'Success' THEN 70
    WHEN "Operation" ILIKE '%create%instance%' THEN 60
    ELSE 40
  END AS RiskScore,
  CASE
    WHEN "Operation" ILIKE '%delete%' THEN 'Cloud Resource Deletion'
    WHEN "Operation" ILIKE '%snapshot%' THEN 'Snapshot Operation'
    WHEN "Operation" ILIKE '%create%' THEN 'Cloud Resource Creation'
    ELSE 'Cloud Modification'
  END AS AlertType
FROM events
WHERE LOGSOURCETYPENAME(devicetype) ILIKE '%azure%' OR LOGSOURCETYPENAME(devicetype) ILIKE '%cloudtrail%'
  AND ("Operation" ILIKE '%compute%' OR "Operation" ILIKE '%instance%' OR "Operation" ILIKE '%snapshot%' OR "Operation" ILIKE '%virtualMachine%')
  AND username NOT ILIKE '%azure%automation%'
  AND username NOT ILIKE '%backup%service%'
ORDER BY RiskScore DESC
LAST 1 HOURS

_sourceCategory=*azure*activitylog* OR _sourceCategory=*aws*cloudtrail*
| json auto
| where matches(operationName, "(?i)(snapshot|virtualMachine|instance|compute|create|delete|modify)")
| eval RiskLevel = if(matches(operationName, "(?i)(delete|destroy|terminate)"), "High",
    if(matches(operationName, "(?i)(snapshot|create)"), "Medium", "Low"))
| where RiskLevel in ("High","Medium")
| eval AlertType = if(matches(operationName, "(?i)delete"), "Cloud Resource Deletion",
    if(matches(operationName, "(?i)snapshot"), "Snapshot Operation",
    if(matches(operationName, "(?i)create"), "Resource Creation", "Cloud Modification")))
| where caller != "azure-backup" and !matches(caller, "(?i)automation")
| stats count AS OpCount, values(operationName) AS Operations, values(RiskLevel) AS RiskLevels by _sourceHost, caller, resourceGroup
| sort by OpCount desc

rule T1578_001_cloud_infra_modification {
  meta:
    author = "Detection Engineering"
    description = "Detects suspicious cloud infrastructure modifications"
    severity = "high"
    confidence = "medium"
    mitre_attack = "T1578.001"
    reference = "https://attack.mitre.org/techniques/T1578/001/"

  events:
    $e.metadata.event_type = "USER_RESOURCE_ACCESS"
    re.regex($e.metadata.product_event_type, `(?i)(snapshot|virtualMachine|instance|compute|create|delete|modify)`)
    not re.regex($e.principal.user.email_addresses, `(?i)(backup|automation|serviceaccount)`)

  condition:
    $e
}

#event_simpleName = "CloudResourceModification"
| OperationName = /(?i)(snapshot|virtualMachine|instance|compute|create|delete|modify|revert)/
| ActorUserName != /(?i)(backup|automation|serviceaccount|scheduler)/
| groupBy([aid, ActorUserName, OperationName, ResourceType, ResourceName], function=[count(as=OpCount), min(timestamp, as=FirstSeen)])
| case {
    OperationName = /(?i)(delete|destroy|terminate)/ => RiskScore := "High";
    OperationName = /(?i)(snapshot|create)/ => RiskScore := "Medium";
    * => RiskScore := "Low";
  }
| where RiskScore in ("High", "Medium")
| table([ActorUserName, OperationName, ResourceType, ResourceName, OpCount, RiskScore, FirstSeen])
| sort(RiskScore)