Elastic high stable kql

Multiple Cloud Secrets Accessed by Source Address

This rule detects authenticated sessions accessing secret stores across multiple environments from the same source address within a short period of time, including cloud providers (AWS, GCP, Azure) and Kubernetes clusters. Adversaries with access to compromised credentials or session tokens may attempt to retrieve secrets from services such as AWS Secrets Manager, Google Secret Manager, Azure Key Vault, or Kubernetes Secrets in rapid succession to expand their access or exfiltrate sensitive information.

View Source

Detection Logic

FROM logs-azure.platformlogs-*, logs-aws.cloudtrail-*, logs-gcp.audit-*, logs-kubernetes.audit_logs-* METADATA _id, _version, _index
| WHERE 
  ( 
    /* AWS Secrets Manager */ 
    (data_stream.dataset == "aws.cloudtrail" 
AND event.action == "GetSecretValue") 
OR 
    
    // Azure Key Vault (platform logs)
    (data_stream.dataset == "azure.platformlogs" 
AND event.action IN ("SecretGet", "KeyGet")) or 
    
    /* Google Secret Manager */ 
    (data_stream.dataset IN ("googlecloud.audit", "gcp.audit") 
AND 
     event.action IN ("google.cloud.secretmanager.v1.SecretManagerService.AccessSecretVersion", "google.cloud.secretmanager.v1.SecretManagerService.GetSecretRequest")) 
OR 

    /* Kubernetes Secrets */
    (data_stream.dataset == "kubernetes.audit_logs" 
AND kubernetes.audit.objectRef.resource == "secrets" 
AND kubernetes.audit.verb IN ("get", "list"))

   ) 
AND source.ip IS NOT NULL

// Cloud vendor label based on dataset
| EVAL Esql.cloud_vendor = CASE(
    data_stream.dataset == "aws.cloudtrail", "aws",
    data_stream.dataset == "azure.platformlogs", "azure",
    data_stream.dataset IN ("googlecloud.audit","gcp.audit"), "gcp",
    data_stream.dataset == "kubernetes.audit_logs", "k8s",
    "unknown"
  )
// Vendor+tenant label, e.g. aws:123456789012, azure:tenant, gcp:project
| EVAL Esql.tenant_label = CASE(
    Esql.cloud_vendor == "aws", CONCAT("aws:", cloud.account.id),
    Esql.cloud_vendor == "azure", CONCAT("azure:", cloud.account.id),
    Esql.cloud_vendor == "gcp", CONCAT("gcp:", cloud.account.id),
    Esql.cloud_vendor == "k8s", CONCAT("k8s:", orchestrator.cluster.name),
    "unknown"
  )
| WHERE Esql.cloud_vendor != "unknown"
| STATS
    // Core counts
    Esql.events_count = COUNT(*),
    Esql.vendor_count_distinct = COUNT_DISTINCT(Esql.cloud_vendor),
    // Action & data source context
    Esql.event_action_values = VALUES(event.action),
    Esql.data_source_values = VALUES(data_stream.dataset),
    // Cloud vendor + tenant context
    Esql.cloud_vendor_values = VALUES(Esql.cloud_vendor),
    Esql.tenant_label_values = VALUES(Esql.tenant_label),
    // Hyperscaler-specific IDs
    Esql.aws_account_id_values = VALUES(CASE(Esql.cloud_vendor == "aws", cloud.account.id, "unknown")),
    Esql.azure_tenant_id_values = VALUES(CASE(Esql.cloud_vendor == "azure", cloud.account.id, "unknown")),
    Esql.gcp_project_id_values = VALUES(CASE(Esql.cloud_vendor == "gcp", cloud.account.id, "unknown")),
    // Generic cloud metadata
    Esql.cloud_region_values = VALUES(cloud.region),
    Esql.k8s_namespace_values = VALUES(kubernetes.audit.objectRef.namespace),
    // Namespace values
    Esql.data_stream_namespace_values = VALUES(data_stream.namespace), 
    Esql_priv.user_name_values = VALUES(user.name)
  BY source.ip
// Require multi-vendor cred-access from same source IP
| WHERE Esql.vendor_count_distinct >= 2
| SORT Esql.events_count DESC
| KEEP Esql.*, Esql_priv.*, source.ip

Field Validations

Loading…

Comments (0)

Loading comments...