✄𐘗
text code block:
✄𐘗from logs-gcp.audit-* metadata _id, _index, _version
| where data_stream.dataset == "gcp.audit"
    and service.name == "k8s.io"
    and event.outcome == "success"
    and client.user.email is not null
    and source.ip is not null
    and not to_string(source.ip) in ("127.0.0.1", "::1")
    and not client.user.email in (
      "system:addon-manager",
      "system:apiserver",
      "system:kube-controller-manager"
    )
    and (
      (
        event.action in (
          "io.k8s.authorization.rbac.v1.roles.create",
          "io.k8s.authorization.rbac.v1.roles.update",
          "io.k8s.authorization.rbac.v1.roles.patch",
          "io.k8s.authorization.rbac.v1.clusterroles.create",
          "io.k8s.authorization.rbac.v1.clusterroles.update",
          "io.k8s.authorization.rbac.v1.clusterroles.patch"
        )
        and not (
          client.user.email == "system:serviceaccount:kube-system:clusterrole-aggregation-controller"
          and event.action == "io.k8s.authorization.rbac.v1.clusterroles.patch"
        )
        and (
          KQL("""gcp.audit.request.rules.verbs:("*" or "escalate" or "bind" or "impersonate")""")
          or KQL("""gcp.audit.request.rules.verbs:("*" or "create" or "patch" or "update") and gcp.audit.request.rules.resources:("*" or "clusterroles" or "clusterrolebindings" or "roles" or "rolebindings" or "pods/exec" or "serviceaccounts/token" or "nodes/proxy" or "daemonsets")""")
          or KQL("""gcp.audit.request.rules.verbs:("*" or "get" or "list") and gcp.audit.request.rules.resources:("*" or "secrets")""")
        )
      )
      or event.action in (
        "io.k8s.apps.v1.daemonsets.create",
        "io.k8s.apps.v1.daemonsets.patch",
        "io.k8s.apps.v1.deployments.create",
        "io.k8s.apps.v1.deployments.patch",
        "io.k8s.batch.v1.cronjobs.create",
        "io.k8s.batch.v1.cronjobs.patch"
      )
    )
| eval Esql.is_sensitive_rbac = case(event.action like "io.k8s.authorization.rbac.*", 1, 0),
    Esql.is_workload_modification = case(event.action like "io.k8s.apps.*" or event.action like "io.k8s.batch.*", 1, 0),
    Esql.sensitive_rbac_timestamp = case(event.action like "io.k8s.authorization.rbac.*", @timestamp, null),
    Esql.workload_modification_timestamp = case(event.action like "io.k8s.apps.*" or event.action like "io.k8s.batch.*", @timestamp, null)
| stats
    Esql.sensitive_rbac_count = sum(Esql.is_sensitive_rbac),
    Esql.workload_modification_count = sum(Esql.is_workload_modification),
    Esql.latest_sensitive_rbac_timestamp = max(Esql.sensitive_rbac_timestamp),
    Esql.earliest_workload_modification_timestamp = min(Esql.workload_modification_timestamp),
    Esql.event_action_values = values(event.action),
    Esql.gcp_audit_resource_name_values = values(gcp.audit.resource_name),
    Esql.user_agent_original_values = values(user_agent.original),
    Esql.source_ip_values = values(source.ip)
  by client.user.email
| where Esql.sensitive_rbac_count > 0
    and Esql.workload_modification_count > 0
    and Esql.latest_sensitive_rbac_timestamp <= Esql.earliest_workload_modification_timestamp
| eval Esql.rbac_to_workload_minutes = date_diff(
    "minute",
    Esql.latest_sensitive_rbac_timestamp,
    Esql.earliest_workload_modification_timestamp
  )
| where Esql.rbac_to_workload_minutes <= 5
| keep
    client.user.email,
    Esql.*