Panther rules: k8s

from panther_kubernetes_helpers import is_failed_request, is_system_principal, k8s_alert_context

# Admission controller webhook resource types
WEBHOOK_RESOURCES = {
    "mutatingwebhookconfigurations",
    "validatingwebhookconfigurations",
}


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

    # Only check webhook creation events
    if verb != "create":
        return False

    if resource not in WEBHOOK_RESOURCES:
        return False

    # Skip failed requests
    if is_failed_request(response_status):
        return False

    # Exclude noise from cluster maintenance
    username = event.udm("username")
    if is_system_principal(username):
        return False

    # Alert on any admission controller webhook creation
    return True


def title(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    resource = event.udm("resource") or "webhook"
    name = event.udm("name") or "<UNKNOWN_NAME>"

    webhook_type = "Mutating" if "mutating" in resource.lower() else "Validating"

    return f"[{username}] created {webhook_type} admission controller webhook [{name}] "


def dedup(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    return f"k8s_admission_webhook_{username}"


def severity(event):
    """Increase severity for webhooks that intercept all resources."""
    webhooks = event.udm("webhooks") or []

    for webhook in webhooks:
        rules = webhook.get("rules", [])
        for rule_config in rules:
            resources = rule_config.get("resources", [])
            api_groups = rule_config.get("apiGroups", [])

            # Check for wildcard rules that intercept everything
            if "*" in resources or "*" in api_groups:
                return "HIGH"

    return "MEDIUM"


def alert_context(event):
    webhooks = event.udm("webhooks") or []

    # Extract webhook details
    webhook_details = []
    for webhook in webhooks:
        client_config = webhook.get("clientConfig", {})
        webhook_details.append(
            {
                "name": webhook.get("name"),
                "url": client_config.get("url"),
                "service": client_config.get("service"),
                "failure_policy": webhook.get("failurePolicy"),
                "rules": webhook.get("rules", []),
            }
        )

    return k8s_alert_context(
        event,
        extra_fields={
            "webhook_name": event.udm("name"),
            "webhook_type": event.udm("resource"),
            "webhooks": webhook_details,
        },
    )

AnalysisType: rule
RuleID: "Kubernetes.AdmissionController.Created"
DisplayName: "Kubernetes Admission Controller Webhook Created"
Enabled: true
Filename: k8s_admission_controller_created.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Persistence
  - Credential Access
  - Collection
  - Unified Detection
Severity: Medium
Description: >
  This detection monitors for creation of MutatingWebhookConfiguration or ValidatingWebhookConfiguration
  resources. Admission controller webhooks can intercept all API requests to the Kubernetes API server,
  allowing attackers to inspect, modify, or block any resource creation or modification. This provides
  powerful capabilities for persistence (modifying deployments to inject backdoors), credential theft
  (intercepting secrets), and reconnaissance (enumerating all cluster activity).
Runbook: |
  1. Review the webhook configuration details including the target webhook service URL and failure policy
  2. Identify all API operations performed by the username in the 48 hours before webhook creation to establish intent
  3. Search for other webhook configurations or suspicious API activity from this user across all clusters in the past 7 days
Reports:
  MITRE ATT&CK:
    - TA0003:T1546 # Persistence: Event Triggered Execution
    - TA0006:T1552 # Credential Access: Unsecured Credentials
    - TA0009:T1530 # Collection: Data from Cloud Storage Object
Reference: https://kubernetes.io/docs/reference/access-authn-authz/extensible-admission-controllers/
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - p_source_label

from panther_kubernetes_helpers import is_failed_request, is_system_principal, k8s_alert_context

# System components that legitimately list all secrets cluster-wide
ALLOWED_SECRET_LISTERS = {
    "system:serviceaccount:kube-system:namespace-controller",
    "system:serviceaccount:kube-system:kube-state-metrics",
    "system:apiserver",
}


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

    # Only check secret list operations that are successful
    if verb != "list" or resource != "secrets" or is_failed_request(response_status):
        return False

    # Key indicator: namespace is empty (cluster-wide list, not namespaced)
    # and requestURI doesn't contain /namespaces/ in path
    if namespace or (request_uri and "/namespaces/" in request_uri):
        return False

    # Exclude system principals and specific components
    if is_system_principal(username) or username in ALLOWED_SECRET_LISTERS:
        return False

    return True


def title(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    request_uri = event.udm("requestURI") or ""

    return f"[{username}] dumped all secrets across all namespaces (URI: {request_uri})"


def dedup(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    return f"k8s_secrets_dump_{username}"


def alert_context(event):
    request_uri = event.udm("requestURI") or ""

    # Extract query parameters if present (e.g., limit=500)
    query_params = {}
    if "?" in request_uri:
        query_string = request_uri.split("?", 1)[1]
        for param in query_string.split("&"):
            if "=" in param:
                key, value = param.split("=", 1)
                query_params[key] = value

    return k8s_alert_context(
        event,
        extra_fields={
            "request_uri": request_uri,
            "query_parameters": query_params,
            "operation": "list_all_secrets",
        },
    )

AnalysisType: rule
RuleID: "Kubernetes.Secrets.DumpAllNamespaces"
DisplayName: "Kubernetes All Secrets Dumped Across Namespaces"
Enabled: true
Status: Experimental
Filename: k8s_secrets_dump_all_namespaces.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Credential Access
  - Mass Exfiltration
  - Secrets Management
  - Unified Detection
Severity: Critical
Description: >
  This detection monitors for cluster-wide secret list operations that dump all secrets across all namespaces in a
  single API call. Attackers with list secrets permissions at the cluster level can trivially access every secret in
  the cluster using the LIST /api/v1/secrets API, exposing all credentials, tokens, and sensitive configuration data.
  This is a known attack technique documented by Stratus Red Team and represents mass credential theft.
Runbook: |
  1. Immediately investigate the user or service account performing this operation and determine if they are compromised
  2. Review what secrets exist in the cluster to assess the scope of credential exposure and rotate all sensitive credentials
  3. Search for other suspicious API activity by this user in the past 24 hours and check for data exfiltration or lateral movement attempts
Reports:
  Stratus Red Team:
    - k8s.credential-access.dump-secrets
  MITRE ATT&CK:
    - TA0006:T1552.007 # Credential Access: Unsecured Credentials - Container API
    - TA0009:T1530 # Collection: Data from Cloud Storage Object
Reference: https://stratus-red-team.cloud/attack-techniques/kubernetes/k8s.credential-access.dump-secrets/
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - p_source_label

from panther_kubernetes_helpers import k8s_alert_context


def is_system_access(event):
    # Skip localhost and health check traffic
    # nosec: Fallback to empty list is safe - we check for localhost separately
    src_ip = event.udm("sourceIPs") or []
    if src_ip == ["127.0.0.1"]:
        return True

    user_agent = event.udm("userAgent") or ""

    # AWS EKS: Exclude ELB health checker from internal IPs
    if user_agent == "ELB-HealthChecker/2.0" and src_ip and src_ip[0].startswith("10.0."):
        return True

    system_user_agents = (
        "kube-probe/",
        "GoogleHC/",
    )
    # GCP GKE & Azure AKS: Exclude kube-probe (Kubernetes liveness/readiness probes)
    # GCP GKE: Exclude Google Cloud Load Balancer health checks

    if any(user_agent.startswith(pattern) for pattern in system_user_agents):
        return True
    return False


def is_health_check(event):
    request_uri = event.udm("requestURI") or ""
    health_patterns = (
        "/healthz",
        "/readyz",
        "/livez",
        "/apis/healthz",
        "/apis/readyz",
        "/apis/livez",
    )
    if any(request_uri.startswith(pattern) for pattern in health_patterns):
        return True
    return False


def rule(event):
    if event.udm("username") == "system:anonymous":
        if not is_system_access(event) and not is_health_check(event):
            return True
    return False


def title(event):
    # For failed attempts or /version endpoint, use generic titles
    annotations = event.udm("annotations") or {}
    if annotations.get("authorization.k8s.io/decision") != "allow":
        return "Failed Anonymous Kubernetes API Access Attempt(s) Detected"
    if event.udm("requestURI") == "/version":
        return "Anonymous Kubernetes API Access to /version Endpoint Detected"

    # For successful access to other endpoints, provide detailed information
    source_ips = event.udm("sourceIPs") or []
    source_ip = source_ips[0] if source_ips else "<UNKNOWN_IP>"
    request_uri = event.udm("requestURI") or "<UNKNOWN_URI>"
    return (
        f"Anonymous API access detected on Kubernetes API server "
        f"from [{source_ip}] to [{request_uri}]"
    )


def severity(event):
    annotations = event.udm("annotations") or {}
    if annotations.get("authorization.k8s.io/decision") != "allow":
        return "INFO"
    if event.udm("requestURI") == "/version":
        return "INFO"
    return "DEFAULT"


def dedup(event):

    user_agent = event.udm("userAgent") or "<UNKNOWN_USER_AGENT>"
    return f"anonymous_access_{user_agent}"


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

AnalysisType: rule
Filename: k8s_anonymous_api_access.py
RuleID: "Kubernetes.API.AnonymousAccess"
DisplayName: "Kubernetes Anonymous API Access Detected"
Enabled: true
Status: Experimental
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Severity: Info
Reports:
  MITRE ATT&CK:
    - TA0001:T1190 # Initial Access: Exploit Public-Facing Application
Description: >
  This rule detects anonymous API requests made to Kubernetes API servers across
  AWS EKS, Azure AKS, and GCP GKE clusters. In production environments, anonymous
  access should be disabled to prevent unauthorized access to the API server.
DedupPeriodMinutes: 60
Reference:
  https://raesene.github.io/blog/2023/03/18/lets-talk-about-anonymous-access-to-Kubernetes/
Runbook: |
  1. Query all API requests by the anonymous username (system:anonymous) in the 24 hours before and after the alert to identify scope of unauthorized access
  2. Check if the sourceIPs field contains internal, cloud provider, or external addresses to determine access vector
  3. Review other authentication attempts from the same sourceIPs in the past 7 days to identify related suspicious activity
SummaryAttributes:
  - username
  - p_any_ip_addresses
  - p_source_label
Tags:
  - Kubernetes
  - Security Control
  - API
  - Initial Access
  - Unified Detection

from panther_kubernetes_helpers import is_k8s_log, k8s_alert_context


def rule(event):
    # Check if this is a Kubernetes audit log with Tor exit node enrichment
    if is_k8s_log(event) and event.deep_get("p_enrichment", "tor_exit_nodes"):
        return True

    return False


def title(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    tor_nodes = event.deep_get("p_enrichment", "tor_exit_nodes", default=[])
    tor_ip = tor_nodes[0] if tor_nodes else "<UNKNOWN_IP>"

    return f"Kubernetes API activity from Tor exit node [{tor_ip}] by user [{username}]"


def dedup(event):
    tor_nodes = event.deep_get("p_enrichment", "tor_exit_nodes", default=[])
    tor_ip = tor_nodes[0] if tor_nodes else "<UNKNOWN_IP>"
    return f"k8s_tor_{tor_ip}"


def alert_context(event):
    return k8s_alert_context(
        event,
        extra_fields={"tor_exit_nodes": event.deep_get("p_enrichment", "tor_exit_nodes")},
    )

AnalysisType: rule
RuleID: "Kubernetes.API.IOC.Activity"
DisplayName: "Kubernetes API Activity from Tor Exit Node"
Enabled: true
Filename: k8s_ioc_activity.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Security Control
  - Command and Control
  - Encrypted Channel
  - Unified Detection
Severity: Medium
Description: >
  This detection monitors for Kubernetes API requests originating from known Indicators
  of Compromise, specifically Tor exit nodes. Tor usage may indicate attempts to hide
  the true source of malicious activity or unauthorized access attempts. This detection
  works across AWS EKS, Azure AKS, and GCP GKE clusters.
Runbook: |
  1. Find all API operations performed by username through the Tor exit node IP in the 6 hours before and after the alert
  2. Compare the operations and resources accessed against normal baseline activity for this user in the past 30 days to identify anomalous behavior
  3. Check if the Tor exit node IP has accessed other clusters or sensitive resources in the past 24 hours to assess campaign scope
Reference: https://medium.com/snowflake/from-logs-to-detection-using-snowflake-and-panther-to-detect-k8s-threats-d72f70a504d7
Reports:
  MITRE ATT&CK:
    - TA0011:T1573.002 # Command and Control: Encrypted Channel - Asymmetric Cryptography
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - p_any_ip_addresses
  - p_source_label

from ipaddress import ip_address

from panther_kubernetes_helpers import k8s_alert_context


def rule(event):
    response_status = event.udm("responseStatus") or {}
    source_ips = event.udm("sourceIPs") or []

    # Check for 403 response (HTTP) or 7 (gRPC PERMISSION_DENIED)
    status_code = response_status.get("code")
    if status_code not in (403, 7):
        return False
    if not source_ips:
        return False

    try:
        source_ip = source_ips[0]
        ip_obj = ip_address(source_ip)
        # alert on public IPs
        # Check if is_global attribute exists (Python 3.4+)
        if hasattr(ip_obj, "is_global") and ip_obj.is_global:
            return True
    except (ValueError, IndexError):
        return False

    return False


def title(event):
    source_ips = event.udm("sourceIPs") or []
    source_ip = source_ips[0] if source_ips else "<UNKNOWN_IP>"

    return f"Multiple 403 responses from public IP [{source_ip}]"


def dedup(event):
    source_ips = event.udm("sourceIPs") or []
    source_ip = source_ips[0] if source_ips else "<UNKNOWN_IP>"
    return f"k8s_403_{source_ip}"


def alert_context(event):
    return k8s_alert_context(event)

AnalysisType: rule
Filename: k8s_multiple_403_public_ip.py
RuleID: "Kubernetes.API.Multiple403.PublicIP"
DisplayName: "Kubernetes API Multiple 403 Responses from Single Public IP"
Enabled: true
Status: Experimental
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Security Control
  - Container and Resource Discovery
  - Unified Detection
Reports:
  MITRE ATT&CK:
    - TA0007:T1613 # Container and Resource Discovery
Reference: https://aws.github.io/aws-eks-best-practices/security/docs/detective/
Severity: Info
Description: >
  This detection identifies when a public source IP generates multiple 403 (Forbidden)
  responses from the Kubernetes API server. This pattern may indicate reconnaissance
  attempts, permission enumeration, brute force attacks, or misconfigured access.
  Private IPs are excluded as they typically represent legitimate internal traffic.
Runbook: |
  1. Find all API requests from the sourceIPs address in the 1 hour before and after the alert to identify attempted operations
  2. Identify which API resources or operations the source IP attempted to access to determine if this is systematic probing or random scanning
  3. Check if the same source IP has generated 403 errors against other Kubernetes clusters in the past 24 hours to assess campaign scope
DedupPeriodMinutes: 30
Threshold: 10
SummaryAttributes:
  - username
  - p_any_ip_addresses
  - p_source_label

from panther_kubernetes_helpers import is_failed_request, is_system_principal, k8s_alert_context


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

    # Only check CertificateSigningRequest creation
    if verb != "create" or resource != "certificatesigningrequests":
        return False

    # Skip failed requests
    if is_failed_request(response_status):
        return False

    # Exclude system principals
    if is_system_principal(username):
        return False

    # Exclude node bootstrap processes that legitimately create CSRs during cluster operations
    if username == "kubelet-nodepool-bootstrap":
        return False

    # Check if this is for client authentication
    request_object = event.udm("requestObject") or {}
    spec = request_object.get("spec", {})
    usages = spec.get("usages", [])
    signer_name = spec.get("signerName", "")

    # Look for client auth certificates
    if "client auth" in usages or "kubernetes.io/kube-apiserver-client" in signer_name:
        return True

    return False


def title(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    name = event.udm("name") or "<UNKNOWN_CSR>"

    return f"[{username}] created client certificate signing request [{name}]"


def dedup(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    name = event.udm("name") or "<UNKNOWN_CSR>"
    return f"k8s_client_cert_{username}_{name}"


def alert_context(event):
    request_object = event.udm("requestObject") or {}
    spec = request_object.get("spec", {})

    return k8s_alert_context(
        event,
        extra_fields={
            "csr_name": event.udm("name"),
            "signer_name": spec.get("signerName"),
            "usages": spec.get("usages"),
            "groups": spec.get("groups"),
        },
    )

AnalysisType: rule
RuleID: "Kubernetes.ClientCertificate.Created"
DisplayName: "Kubernetes Client Certificate Credential Created"
Enabled: true
Filename: k8s_client_certificate_created.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Persistence
  - Credential Access
  - Unified Detection
Severity: Info
Description: >
  Detects the creation of client certificate signing requests (CSRs) for Kubernetes API authentication. Attackers with
  appropriate RBAC permissions can create and approve client certificates to establish persistent access to the cluster.
  Client certificates provide long-term authentication that bypasses service account token expiration and can be harder
  to revoke. This technique is documented by Stratus Red Team as a persistence mechanism.
Runbook: |
  1. Immediately investigate the user creating the certificate and verify if they are authorized to create client auth credentials
  2. Check if the CSR was approved and review who approved it within the last 10 minutes by searching audit logs for certificate approval events
  3. Search for API activity using the newly created certificate in the past 24 hours to identify any unauthorized access
Reports:
  Stratus Red Team:
    - k8s.persistence.create-client-certificate
  MITRE ATT&CK:
    - TA0003:T1098 # Persistence: Account Manipulation
    - TA0006:T1552 # Credential Access: Unsecured Credentials
Reference: https://stratus-red-team.cloud/attack-techniques/kubernetes/k8s.persistence.create-client-certificate/
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - name
  - p_source_label

from panther_base_helpers import deep_get
from panther_kubernetes_helpers import is_failed_request, is_system_principal, k8s_alert_context

PRIVILEGED_CLUSTER_ROLES = {
    "cluster-admin",  # Full cluster control
    "system:masters",  # Superuser group
    "admin",  # Namespace admin capabilities
    "system:kube-controller-manager",  # Controller manager privileges
    "system:kube-scheduler",  # Scheduler privileges
}


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

    # Only check ClusterRoleBinding creation events
    if verb != "create" or resource != "clusterrolebindings":
        return False

    # Skip failed requests
    if is_failed_request(response_status):
        return False

    # Exclude system principals to reduce false positives from legitimate operators
    if is_system_principal(username):
        return False

    # Check if binding references a privileged cluster role
    request_object = event.udm("requestObject") or {}
    role_name = deep_get(request_object, "roleRef", "name", default="")
    if role_name in PRIVILEGED_CLUSTER_ROLES:
        return True

    return False


def title(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    name = event.udm("name") or "<UNKNOWN_BINDING>"

    request_object = event.udm("requestObject") or {}
    role_ref = request_object.get("roleRef", {})
    role_name = role_ref.get("name", "<UNKNOWN_ROLE>")

    # Extract subject information
    subjects = request_object.get("subjects", [])
    subject_names = []
    for subject in subjects:
        subject_type = subject.get("kind", "")
        subject_name = subject.get("name", "")
        if subject_type and subject_name:
            subject_names.append(f"{subject_type}:{subject_name}")

    subjects_str = ", ".join(subject_names) if subject_names else "<UNKNOWN_SUBJECTS>"

    return (
        f"[{username}] created ClusterRoleBinding [{name}] granting [{role_name}] "
        f"to [{subjects_str}]"
    )


def dedup(event):
    username = event.udm("username") or "<UNKNOWN_USER>"
    name = event.udm("name") or "<UNKNOWN_BINDING>"
    return f"k8s_clusterrolebinding_{username}_{name}"


def severity(event):
    """Increase severity for cluster-admin and system:masters roles."""
    request_object = event.udm("requestObject") or {}
    role_ref = request_object.get("roleRef", {})
    role_name = role_ref.get("name", "")

    # High for cluster-admin and system:masters (full cluster control)
    if role_name in {"cluster-admin", "system:masters"}:
        return "HIGH"

    # Medium for other privileged system roles
    return "DEFAULT"


def alert_context(event):
    request_object = event.udm("requestObject") or {}
    role_ref = request_object.get("roleRef", {})
    subjects = request_object.get("subjects", [])

    return k8s_alert_context(
        event,
        extra_fields={
            "binding_name": event.udm("name"),
            "role_name": role_ref.get("name"),
            "role_kind": role_ref.get("kind"),
            "subjects": subjects,
        },
    )

AnalysisType: rule
RuleID: "Kubernetes.ClusterRoleBinding.Privileged"
DisplayName: "Kubernetes ClusterRoleBinding to Privileged Role"
Enabled: true
Filename: k8s_clusterrolebinding_privileged.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Privilege Escalation
  - Persistence
  - RBAC
  - Unified Detection
Severity: Medium
Description: >
  This detection monitors for ClusterRoleBindings being created that grant privileged cluster
  roles like cluster-admin or system:masters. Attackers who gain initial cluster access often create
  ClusterRoleBindings to escalate privileges and gain full control over all cluster resources and namespaces.
  While some bindings to privileged roles are legitimate for cluster operators, unexpected bindings should be
  investigated immediately as they may indicate compromise or insider threat.
Runbook: |
  1. Review the subjects being granted the privileged role and immediately delete the ClusterRoleBinding if unauthorized
  2. Identify all API operations by the creating user in the 2 hours before and after the alert and audit all actions by the subjects that were granted privileges
  3. Review all ClusterRoleBindings across all clusters and search for other RBAC changes in the past 7 days
Reports:
  Stratus Red Team:
    - k8s.persistence.create-admin-clusterrole
    - k8s.privilege-escalation.create-admin-clusterrole
  MITRE ATT&CK:
    - TA0004:T1078.004 # Privilege Escalation: Valid Accounts - Cloud Accounts
    - TA0003:T1098 # Persistence: Account Manipulation
Reference: >
  - https://kubernetes.io/docs/reference/access-authn-authz/rbac/
  - https://seifrajhi.github.io/blog/kubernetes-rbac-privilege-escalation-mitigation/#%EF%B8%8F-rolebinding-permissions-in-kubernetes-implications-and-safeguards
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - name
  - p_source_label

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"),
        },
    )

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

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")
    response_status = event.udm("responseStatus")
    namespace = event.udm("namespace")
    username = event.udm("username")

    # Check for DaemonSet create operation
    if verb == "create" and resource == "daemonsets":
        # Only alert on successful operations
        if is_failed_request(response_status):
            return False

        # Exclude system controllers creating DaemonSets 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>"
    daemonset_name = event.udm("name") or "<UNKNOWN>"

    return f"[{username}] created DaemonSet [{namespace}/{daemonset_name}]"


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


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

AnalysisType: rule
RuleID: "Kubernetes.DaemonSet.Created"
DisplayName: "Kubernetes DaemonSet Created"
Enabled: true
Filename: k8s_daemonset_created.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Security Control
  - Persistence
  - Deploy Container
  - Unified Detection
Severity: Info
Description: >
  This detection monitors for creation of DaemonSets in Kubernetes clusters.
  DaemonSets ensure that a copy of a pod runs on all (or selected) nodes in the cluster.
  Attackers may abuse DaemonSets to deploy malicious containers across all nodes for
  cluster-wide persistence, credential harvesting, cryptomining, or lateral movement.
  This detection works across AWS EKS, Azure AKS, and GCP GKE clusters.
Runbook: |
  1. Review all DaemonSet and Deployment creation events by username in the 24 hours before the alert to establish if this is routine activity
  2. Analyze the container image, hostPath volumes, and privileged security context from the requestObject specification to identify risk factors
  3. Search for similar DaemonSet deployments by this user in the past 90 days and compare images, namespaces, and security settings to identify anomalies
Reference: https://kubernetes.io/docs/concepts/workloads/controllers/daemonset/
Reports:
  MITRE ATT&CK:
    - TA0002:T1610 # Execution: Deploy Container
    - TA0003:T1543 # Persistence: Create or Modify System Process
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - namespace
  - p_source_label

from urllib.parse import parse_qs, urlparse

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


def get_exec_command(event):
    """Extract exec command from requestObject (GCP/Azure) or requestURI query params (EKS).

    Returns:
        List of command arguments, or empty list if not found
    """
    # Try requestObject first (GCP/Azure format)
    request_object = event.udm("requestObject") or {}
    command = request_object.get("command")
    if command:
        return command

    # Fall back to parsing requestURI query parameters (EKS format)
    # EKS format: /api/v1/.../exec?command=tar&command=cf&command=-
    request_uri = event.udm("requestURI") or ""
    if "command=" in request_uri:
        try:
            parsed = urlparse(request_uri)
            params = parse_qs(parsed.query)
            return params.get("command", [])
        except Exception:  # pylint: disable=broad-except
            return []

    return []


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

    # Only check exec subresource operations
    if verb not in ("create", "get") or resource != "pods" or subresource != "exec":
        return False

    # Skip failed requests
    if is_failed_request(response_status):
        return False

    # Exclude system principals creating pods in system namespaces (legitimate)
    # but alert on system principals in user namespaces (malicious Deployments)
    # and alert on user-created pods in system namespaces (suspicious)
    if is_system_principal(username) and is_system_namespace(namespace):
        return False

    # Extract command from either requestObject or requestURI
    command = get_exec_command(event)
    if not command:
        return False

    # Check if command contains tar with cf - pattern
    # tar cf - indicates copying FROM pod (stdout output = exfil)
    tar_found = False
    cf_flag_found = False
    stdout_dash_found = False

    for i, arg in enumerate(command):
        arg_str = str(arg).lower()

        if "tar" in arg_str:
            tar_found = True

        # Look for cf flag (create+file) - handles: cf, -cf, czf, -czf, etc.
        if "cf" in arg_str and arg_str != "-c":
            cf_flag_found = True

        # Check if - appears after cf flag was found (stdout redirect)
        if arg_str == "-" and i > 0 and cf_flag_found:
            stdout_dash_found = True

    return tar_found and cf_flag_found and stdout_dash_found


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

    command = get_exec_command(event)

    # Extract the path being copied if possible
    # Command format: ["tar", "cf", "-", "/path/to/file"]
    path = "<UNKNOWN_PATH>"
    if len(command) >= 4:
        for i, arg in enumerate(command):
            if arg == "-" and i + 1 < len(command):
                path = str(command[i + 1])
                break

    return f"[{username}] copied data from pod [{namespace}/{name}] path [{path}] via kubectl cp"


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


def severity(event):
    """Increase severity for copying from sensitive paths."""
    command = get_exec_command(event)
    command_str = " ".join(str(arg) for arg in command).lower()

    # Critical severity for copying credentials or SSH keys
    critical_patterns = [
        "/root",
        ".ssh",
        "id_rsa",
        "id_ecdsa",
        "id_ed25519",
        "credentials",
        "secrets",
        "token",
        ".kube",
        "serviceaccount",
    ]
    if any(pattern in command_str for pattern in critical_patterns):
        return "CRITICAL"

    # High severity for copying from sensitive system directories
    sensitive_paths = ["/etc", "/var/run", "/proc", "config", "password", "shadow"]
    if any(path in command_str for path in sensitive_paths):
        return "HIGH"

    return "MEDIUM"


def alert_context(event):
    command = get_exec_command(event)
    request_object = event.udm("requestObject") or {}

    return k8s_alert_context(
        event,
        extra_fields={
            "pod_name": event.udm("name"),
            "command": command,
            "container": request_object.get("container"),
        },
    )

AnalysisType: rule
RuleID: "Kubernetes.Kubectl.CP.Operation"
DisplayName: "Kubernetes Data Copy via kubectl cp"
Enabled: true
Filename: k8s_kubectl_cp_operation.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Exfiltration
  - Data Theft
  - Credential Access
  - Unified Detection
Severity: Medium
Description: >
  This detection monitors for kubectl cp operations that copy files from pods to local machines,
  which can indicate data exfiltration. When kubectl cp is used to copy files from a pod, it
  executes a tar command with stdout output (tar cf -) inside the container and streams the data
  back through the Kubernetes API server. Attackers who gain cluster access can use this technique
  to steal application secrets, credentials, configuration files, or sensitive data from container
  filesystems without leaving obvious traces inside the pod itself. While kubectl cp has legitimate
  uses for debugging and backup, unexpected usage should be investigated.
Runbook: |
  1. Review all exec and kubectl cp operations by this user in the 24 hours before and after the alert
  2. Identify what files or directories were exfiltrated and assess their sensitivity
  3. Search for other suspicious API activity from this user or service account across all clusters in the past 7 days
Reports:
  MITRE ATT&CK:
    - TA0010:T1530 # Exfiltration: Data from Cloud Storage
    - TA0006:T1552 # Credential Access: Unsecured Credentials
Reference: https://kubernetes.io/docs/reference/generated/kubectl/kubectl-commands#cp
DedupPeriodMinutes: 60
SummaryAttributes:
  - username
  - namespace
  - name
  - p_source_label

from panther_kubernetes_helpers import k8s_alert_context


def rule(event):
    verb = event.udm("verb")
    resource = event.udm("resource")
    subresource = event.udm("subresource")

    # Check for exec action on pods
    if verb in ("create", "get") and resource == "pods" and subresource == "exec":
        return True

    return False


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

    return f"[{username}] executed into pod [{namespace}/{pod_name}]"


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


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

AnalysisType: rule
RuleID: "Kubernetes.Pod.Exec"
DisplayName: "Kubernetes Exec Into Pod"
Enabled: false
Status: Experimental
Filename: k8s_exec_into_pod.py
LogTypes:
  - Amazon.EKS.Audit
  - Azure.MonitorActivity
  - GCP.AuditLog
Tags:
  - Kubernetes
  - Security Control
  - Configuration Required
  - Unified Detection
Severity: Medium
Description: >
  Detects when users exec into pods across Kubernetes clusters. Execing into pods should be
  monitored as it can be used for unauthorized access, privilege escalation, or persistent
  access to workloads. This detection is disabled by default and should be configured with
  inline filters in the Panther UI to exclude legitimate use cases (e.g., specific service
  accounts, namespaces, or authorized users).
Runbook: |
  1. Query all exec operations by the username in the 24 hours before and after the alert to identify scope of pod access
  2. Review the namespace and pod name from the alert context to determine if accessing sensitive workloads or data
  3. Cross-reference the username against authorized administrator or DevOps groups in the past 30 days to verify authorization
Reference: https://kubernetes.io/docs/tasks/debug/debug-application/get-shell-running-container/