GCP K8s New Daemonset Deployed

AnalysisType: rule
RuleID: "GCP.K8s.New.Daemonset.Deployed"
DisplayName: "GCP K8s New Daemonset Deployed"
Description: "Detects Daemonset creation in GCP Kubernetes clusters."
Enabled: false
Status: Deprecated
Filename: gcp_k8s_new_daemonset_deployed.py
LogTypes:
  - GCP.AuditLog
Severity: Medium
Tags:
  - Deprecated
DedupPeriodMinutes: 60
Threshold: 1
Reference: https://medium.com/snowflake/from-logs-to-detection-using-snowflake-and-panther-to-detect-k8s-threats-d72f70a504d7
Runbook: Investigate a reason of creating Daemonset. Create ticket if appropriate.
Reports:
  MITRE ATT&CK:
    - TA0002:T1610 # Deploy Container
Tests:
  - Name: privilege-escalation
    ExpectedResult: true
    Log:
      protoPayload:
        authorizationInfo:
          - granted: true
            permission: io.k8s.apps.v1.daemonsets.create
        methodName: v2.deploymentmanager.deployments.insert
        serviceName: deploymentmanager.googleapis.com
      receiveTimestamp: "2024-01-19 13:47:19.465856238"
      resource:
        labels:
          name: test-vm-deployment
          project_id: panther-threat-research
        type: deployment
      severity: NOTICE
      timestamp: "2024-01-19 13:47:18.279921000"
  - Name: fail
    ExpectedResult: false
    Log:
      protoPayload:
        authorizationInfo:
          - granted: false
            permission: io.k8s.apps.v1.daemonsets.create
        methodName: v2.deploymentmanager.deployments.insert
        serviceName: deploymentmanager.googleapis.com
      receiveTimestamp: "2024-01-19 13:47:19.465856238"
      resource:
        labels:
          name: test-vm-deployment
          project_id: panther-threat-research
        type: deployment
      severity: NOTICE
      timestamp: "2024-01-19 13:47:18.279921000"


# ------ paired body: gcp_k8s_new_daemonset_deployed.py ------

from panther_gcp_helpers import gcp_alert_context


def rule(event):
    authorization_info = event.deep_walk("protoPayload", "authorizationInfo")
    if not authorization_info:
        return False
    for auth in authorization_info:
        if (
            auth.get("permission") == "io.k8s.apps.v1.daemonsets.create"
            and auth.get("granted") is True
        ):
            return True
    return False


def title(event):
    actor = event.deep_get(
        "protoPayload", "authenticationInfo", "principalEmail", default="<ACTOR_NOT_FOUND>"
    )
    operation = event.deep_get("protoPayload", "methodName", default="<OPERATION_NOT_FOUND>")
    project_id = event.deep_get("resource", "labels", "project_id", default="<PROJECT_NOT_FOUND>")

    return f"[GCP]: [{actor}] performed [{operation}] on project [{project_id}]"


def alert_context(event):
    return gcp_alert_context(event)