Azure Storage Blob Bulk Extraction
Description
Detects high-volume blob extraction (50+ GetBlob operations in 15 minutes) from Azure Storage accounts. Storm-0501 rapidly extracts data using stolen credentials before ransomware deployment. Replicates Defender for Cloud alert 'Unusual amount of data extracted from a storage account'.
Query · python
from panther_azureactivity_helpers import (
azure_activity_alert_context,
azure_resource_logs_success,
extract_resource_name_from_id,
)
def rule(event):
# Must be GetBlob operation (actual data retrieval)
operation = event.get("operationName", "").upper()
if operation != "GETBLOB":
return False
# Must be successful
return azure_resource_logs_success(event)
def extract_caller_ip(event):
"""Extract IP address from callerIpAddress field, removing port if present"""
caller_ip_address = event.get("callerIpAddress", "")
if not caller_ip_address:
return ""
# Split by colon to remove port if present
return caller_ip_address.split(":")[0] if ":" in caller_ip_address else caller_ip_address
def title(event):
caller_ip = extract_caller_ip(event) or "<UNKNOWN_IP>"
resource_id = event.get("resourceId", "")
storage_account = (
extract_resource_name_from_id(resource_id, "storageAccounts", default="<UNKNOWN_ACCOUNT>")
if resource_id
else "<UNKNOWN_ACCOUNT>"
)
return (
f"Unusual volume of blobs extracted from Azure Storage account [{storage_account}] "
f"by [{caller_ip}]"
)
def dedup(event):
"""Group by storage account and caller IP for 15-minute aggregation"""
caller_ip = extract_caller_ip(event) or "unknown"
resource_id = event.get("resourceId", "")
storage_account = (
extract_resource_name_from_id(resource_id, "storageAccounts", default="unknown")
if resource_id
else "unknown"
)
return f"{storage_account}:{caller_ip}"
def severity(event):
"""Higher severity if user agent suggests automated exfiltration"""
user_agent = event.deep_get("properties", "userAgentHeader", default="").lower()
# Scripts, curl, wget suggest malicious automation
suspicious_agents = ["python", "curl", "wget", "powershell", "bash", "script"]
if any(agent in user_agent for agent in suspicious_agents):
return "HIGH"
return "MEDIUM"
def alert_context(event):
context = azure_activity_alert_context(event)
# Add blob-specific context
context["blob_path"] = event.deep_get("properties", "objectKey", default="<UNKNOWN>")
context["user_agent"] = event.deep_get("properties", "userAgentHeader", default="<UNKNOWN>")
return context
Analyst notes
- Query Azure MonitorActivity logs for all GetBlob operations from the callerIpAddress in the 2 hours before and after the alert to calculate total blob count
- Compare the blob count from step 1 to the callerIpAddress's 30-day average for this storage account to determine if this is anomalous volume
- Find all SAS token generation operations (listAccountSas, listServiceSas) in the 6 hours before the first GetBlob to identify if stolen SAS tokens are being used
- Check if the callerIpAddress has accessed this storage account in the past 90 days to determine if this is a new or known accessor
- Search for authentication events from the callerIpAddress in Azure.Audit logs in the 48 hours before the alert to assess if account compromise preceded the extraction