Azure Storage SAS Token Access from External IP
Description
Detects SAS token usage from external public IPs by parsing the signature parameter in storage URIs. Storm-0501 uses stolen SAS tokens from external C2 infrastructure for data exfiltration. Replicates Defender for Cloud alert 'Storage.Blob_AccountSas.InternalSasUsedExternally'.
Query · python
import ipaddress
from urllib.parse import parse_qs, urlparse
def rule(event):
"""
Detects SAS token usage from external IP addresses.
Replicates Defender for Cloud: Storage.Blob_AccountSas.InternalSasUsedExternally
"""
# Must be storage operation
if event.get("category") not in ["StorageRead", "StorageWrite", "StorageDelete"]:
return False
# Must be successful
status_code = event.get("statusCode")
if status_code not in [200, 201, 202, 204]:
return False
# Check if SAS token was used (look for 'sig=' in URI)
uri = event.get("uri", "")
if not uri or "sig=" not in uri:
return False
# Check if IP is external (not private/RFC1918)
caller_ip = extract_caller_ip(event)
if not caller_ip or is_private_ip(caller_ip):
return False
return True
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 is_private_ip(ip_address):
"""Check if IP is in private ranges (RFC1918) or localhost"""
if not ip_address:
return True # Treat empty/missing IPs as private to filter them out
try:
ip_obj = ipaddress.ip_address(ip_address)
return ip_obj.is_private or ip_obj.is_loopback
except ValueError:
# Unparseable IPs are treated as external (suspicious) to avoid missing potential threats
return False
def is_permissive_sas(uri):
"""
Check if SAS token has write, delete, or add permissions.
SAS permissions in 'sp' parameter: r=read, a=add, c=create, w=write, d=delete, l=list
"""
if not uri:
return False # Unknown URIs default to non-permissive (read-only assumption)
parsed = urlparse(uri)
params = parse_qs(parsed.query)
permissions = params.get("sp", [""])[0]
# Check for dangerous permissions
return any(perm in permissions for perm in ["w", "d", "a"])
def title(event):
caller_ip = extract_caller_ip(event) or "<UNKNOWN_IP>"
storage_account = event.deep_get("properties", "accountName", default="<UNKNOWN_ACCOUNT>")
operation = event.get("operationName", "<UNKNOWN_OPERATION>")
return (
f"Azure Storage SAS token used from external IP [{caller_ip}] "
f"to access [{storage_account}] with operation [{operation}]"
)
def severity(event):
"""Higher severity for write/delete operations"""
operation = event.get("operationName", "").lower()
# Check if this is a write/delete operation or has permissive SAS
uri = event.get("uri", "")
if is_permissive_sas(uri):
return "HIGH"
# Delete operations are always high severity
if "delete" in operation:
return "HIGH"
# Write operations are medium severity
if any(op in operation for op in ["put", "write", "create", "set"]):
return "MEDIUM"
# Read-only operations from external IPs are low severity
return "LOW"
def alert_context(event):
# Start with standard Azure activity context
context = {
"caller_ip": extract_caller_ip(event) or "<UNKNOWN>",
"storage_account": event.deep_get("properties", "accountName", default="<UNKNOWN>"),
"operation": event.get("operationName", "<UNKNOWN_OPERATION>"),
"object_key": event.deep_get("properties", "objectKey", default="<UNKNOWN>"),
"user_agent": event.deep_get("properties", "userAgentHeader", default="<UNKNOWN>"),
"uri": event.get("uri", "<UNKNOWN_URI>"),
"status_code": event.get("statusCode"),
"category": event.get("category"),
}
# Extract SAS-specific parameters from URI
uri = event.get("uri", "")
if uri:
parsed = urlparse(uri)
params = parse_qs(parsed.query)
if "sp" in params:
context["sas_permissions"] = params["sp"][0]
if "se" in params:
context["sas_expiry"] = params["se"][0]
return context
def dedup(event):
"""Group alerts by storage account and external IP"""
caller_ip = extract_caller_ip(event) or "unknown"
storage_account = event.deep_get("properties", "accountName", default="unknown")
return f"{storage_account}:{caller_ip}"
Analyst notes
- Query Azure Monitor Activity logs for SAS token generation operations (Microsoft.Storage/storageAccounts/listAccountSas/action) by the same identity in the 24 hours before this access to identify when the token was created and by whom
- Find all storage operations (GetBlob, ListBlobs, DeleteBlob) from the callerIpAddress in the 6 hours before and after the alert to assess if this is isolated access or part of bulk exfiltration
- Check if the callerIpAddress has accessed this storage account in the past 90 days to establish if this external IP is expected
- Review Azure Audit logs for authentication events from the callerIpAddress in the 48 hours before the alert to identify potential account compromise or credential theft
- Search for other alerts with the same callerIpAddress across all storage accounts in the past 7 days to identify if this is targeted or widespread