Kubernetes CronJob Created or Modified


Description

This detection monitors for creation or modification of CronJobs in Kubernetes clusters. Attackers may create or modify scheduled jobs to achieve cluster persistence, execute malicious code on a schedule, or maintain backdoor access to compromised clusters. This detection works across AWS EKS, Azure AKS, and GCP GKE clusters.

Query · python

from panther_kubernetes_helpers import (
    is_failed_request,
    is_system_namespace,
    is_system_principal,
    k8s_alert_context,
)


def rule(event):
    verb = event.udm("verb")
    resource = event.udm("resource")
    subresource = event.udm("subresource")
    response_status = event.udm("responseStatus")
    namespace = event.udm("namespace")
    username = event.udm("username")

    # Check for CronJob create/update/patch operations
    if verb in ("create", "update", "patch") and resource == "cronjobs":
        # Only alert on successful operations
        if is_failed_request(response_status):
            return False

        # Exclude status updates (routine execution tracking, not spec changes)
        if subresource == "status":
            return False

        # Exclude system controllers creating/modifying CronJobs in system namespaces
        if is_system_namespace(namespace) and is_system_principal(username):
            return False

        return True

    return False


def title(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    namespace = event.udm("namespace") or "<UNKNOWN_NAMESPACE>"
    cronjob_name = event.udm("name") or "<UNKNOWN>"
    verb = event.udm("verb")

    action = "created" if verb == "create" else "modified"
    return f"[{username}] {action} CronJob " f"[{namespace}/{cronjob_name}]"


def dedup(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    namespace = event.udm("namespace") or "<UNKNOWN_NAMESPACE>"
    return f"k8s_cronjob_{username}_{namespace}"


def alert_context(event):
    return k8s_alert_context(
        event,
        extra_fields={
            "cronjob_name": event.udm("name"),
            "requestObject": event.udm("requestObject"),
        },
    )

Analyst notes

  1. Query all CronJob creation and modification events by username in the 24 hours before the alert to establish deployment patterns
  2. Analyze the CronJob schedule specification and container image from requestObject to identify suspicious commands or known malicious patterns
  3. Search for other CronJobs created by this user in the past 30 days to determine if this represents established activity or anomalous behavior
Raw source Kubernetes CronJob Created or Modified · Panther Python
Esc
Published by panther-labs/panther-analysis ↗, licensed under Apache 2.0 ↗. Reproduced here unmodified.
AnalysisType: rule
RuleID: "Kubernetes.CronJob.CreatedOrModified"
DisplayName: "Kubernetes CronJob Created or Modified"
Enabled: true
Status: Experimental
Filename: k8s_cronjob_created_or_modified.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Security Control
  - Persistence
  - Scheduled Task
  - Unified Detection
Severity: Info
Description: >
  This detection monitors for creation or modification of CronJobs in Kubernetes clusters.
  Attackers may create or modify scheduled jobs to achieve cluster persistence, execute
  malicious code on a schedule, or maintain backdoor access to compromised clusters.
  This detection works across AWS EKS, Azure AKS, and GCP GKE clusters.
Runbook: |
  1. Query all CronJob creation and modification events by username in the 24 hours before the alert to establish deployment patterns
  2. Analyze the CronJob schedule specification and container image from requestObject to identify suspicious commands or known malicious patterns
  3. Search for other CronJobs created by this user in the past 30 days to determine if this represents established activity or anomalous behavior
Reference: https://kubernetes.io/docs/concepts/workloads/controllers/cron-jobs/
Reports:
  MITRE ATT&CK:
    - TA0003:T1053.003 # Persistence: Scheduled Task/Job - Cron
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - namespace
  - p_source_label
Tests:
  - Name: EKS CronJob Created
    ExpectedResult: true
    Log:
      {
        "kind": "Event",
        "apiVersion": "audit.k8s.io/v1",
        "auditID": "abc-123",
        "verb": "create",
        "user": {"username": "admin@example.com"},
        "sourceIPs": ["1.2.3.4"],
        "userAgent": "kubectl/v1.28.0",
        "objectRef": {
          "resource": "cronjobs",
          "namespace": "default",
          "name": "backup-job",
          "apiVersion": "batch/v1"
        },
        "responseStatus": {"code": 201},
        "requestURI": "/apis/batch/v1/namespaces/default/cronjobs",
        "requestObject": {
          "kind": "CronJob",
          "metadata": {"name": "backup-job", "namespace": "default"},
          "spec": {
            "schedule": "0 2 * * *",
            "jobTemplate": {
              "spec": {
                "template": {
                  "spec": {
                    "containers": [{
                      "name": "backup",
                      "image": "backup:latest"
                    }]
                  }
                }
              }
            }
          }
        },
        "p_log_type": "Amazon.EKS.Audit",
        "p_source_label": "eks-cluster"
      }
  - Name: AKS CronJob Updated
    ExpectedResult: true
    Log:
      {
        "p_log_type": "Azure.MonitorActivity",
        "category": "kube-audit",
        "operationName": "Microsoft.ContainerService/managedClusters/diagnosticLogs/Read",
        "properties": {
          "log": "{\"kind\":\"Event\",\"apiVersion\":\"audit.k8s.io/v1\",\"verb\":\"update\",\"user\":{\"username\":\"admin@example.com\"},\"sourceIPs\":[\"10.0.0.1\"],\"objectRef\":{\"resource\":\"cronjobs\",\"namespace\":\"production\",\"name\":\"nightly-job\"},\"responseStatus\":{\"code\":200},\"requestURI\":\"/apis/batch/v1/namespaces/production/cronjobs/nightly-job\"}"
        },
        "p_source_label": "aks-cluster"
      }
  - Name: GCP GKE CronJob Created
    ExpectedResult: true
    Log:
      {
        "protoPayload": {
          "authenticationInfo": {"principalEmail": "user@company.com"},
          "authorizationInfo": [{
            "granted": true,
            "permission": "io.k8s.batch.v1.cronjobs.create",
            "resource": "batch/v1/namespaces/default/cronjobs/backup-job"
          }],
          "methodName": "io.k8s.batch.v1.cronjobs.create",
          "requestMetadata": {
            "callerIP": "1.2.3.4",
            "callerSuppliedUserAgent": "kubectl/v1.27.0"
          },
          "resourceName": "batch/v1/namespaces/default/cronjobs/backup-job",
          "serviceName": "k8s.io",
          "request": {
            "kind": "CronJob",
            "metadata": {"name": "backup-job"},
            "spec": {"schedule": "0 2 * * *"}
          }
        },
        "resource": {
          "type": "k8s_cluster",
          "labels": {"project_id": "test-project"}
        },
        "p_log_type": "GCP.AuditLog",
        "p_source_label": "gke-cluster"
      }
  - Name: CronJob Creation Failed
    ExpectedResult: false
    Log:
      {
        "kind": "Event",
        "verb": "create",
        "objectRef": {
          "resource": "cronjobs",
          "namespace": "default",
          "name": "test-job"
        },
        "responseStatus": {"code": 403, "status": "Failure"},
        "p_log_type": "Amazon.EKS.Audit"
      }
  - Name: Pod Creation (Not CronJob)
    ExpectedResult: false
    Log:
      {
        "kind": "Event",
        "verb": "create",
        "objectRef": {
          "resource": "pods",
          "namespace": "default"
        },
        "responseStatus": {"code": 201},
        "p_log_type": "Amazon.EKS.Audit"
      }
  - Name: CronJob Get (Not Create/Update)
    ExpectedResult: false
    Log:
      {
        "kind": "Event",
        "verb": "get",
        "objectRef": {
          "resource": "cronjobs",
          "namespace": "default",
          "name": "existing-job"
        },
        "responseStatus": {"code": 200},
        "p_log_type": "Amazon.EKS.Audit"
      }
  - Name: System controller creating CronJob in kube-system (excluded)
    ExpectedResult: false
    Log:
      {
        "kind": "Event",
        "apiVersion": "audit.k8s.io/v1",
        "verb": "create",
        "user": {"username": "system:serviceaccount:kube-system:cronjob-controller"},
        "sourceIPs": ["10.0.0.1"],
        "objectRef": {
          "resource": "cronjobs",
          "namespace": "kube-system",
          "name": "system-cleanup",
          "apiVersion": "batch/v1"
        },
        "responseStatus": {"code": 201},
        "p_log_type": "Amazon.EKS.Audit",
        "p_source_label": "eks-cluster"
      }


# ------ paired body: k8s_cronjob_created_or_modified.py ------

from panther_kubernetes_helpers import (
    is_failed_request,
    is_system_namespace,
    is_system_principal,
    k8s_alert_context,
)


def rule(event):
    verb = event.udm("verb")
    resource = event.udm("resource")
    subresource = event.udm("subresource")
    response_status = event.udm("responseStatus")
    namespace = event.udm("namespace")
    username = event.udm("username")

    # Check for CronJob create/update/patch operations
    if verb in ("create", "update", "patch") and resource == "cronjobs":
        # Only alert on successful operations
        if is_failed_request(response_status):
            return False

        # Exclude status updates (routine execution tracking, not spec changes)
        if subresource == "status":
            return False

        # Exclude system controllers creating/modifying CronJobs in system namespaces
        if is_system_namespace(namespace) and is_system_principal(username):
            return False

        return True

    return False


def title(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    namespace = event.udm("namespace") or "<UNKNOWN_NAMESPACE>"
    cronjob_name = event.udm("name") or "<UNKNOWN>"
    verb = event.udm("verb")

    action = "created" if verb == "create" else "modified"
    return f"[{username}] {action} CronJob " f"[{namespace}/{cronjob_name}]"


def dedup(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    namespace = event.udm("namespace") or "<UNKNOWN_NAMESPACE>"
    return f"k8s_cronjob_{username}_{namespace}"


def alert_context(event):
    return k8s_alert_context(
        event,
        extra_fields={
            "cronjob_name": event.udm("name"),
            "requestObject": event.udm("requestObject"),
        },
    )

Detection rules belong to the projects that publish them and remain under their own licenses. This site indexes and links to them; it claims no rights in them.