Web Server Potential Command Injection Request

from logs-nginx.access-*, logs-apache.access-*, logs-apache_tomcat.access-*, logs-iis.access-*, logs-traefik.access-*
| where
    // Limit to 200 response code to reduce noise
    http.response.status_code == 200
| eval Esql.url_original_to_lower = to_lower(url.original)
| eval Esql.contains_interpreter = case(Esql.url_original_to_lower like "*python* -c*" or Esql.url_original_to_lower like "*perl* -e*" or Esql.url_original_to_lower like "*ruby* -e*" or Esql.url_original_to_lower like "*ruby* -rsocket*" or Esql.url_original_to_lower like "*lua* -e*" or Esql.url_original_to_lower like "*php* -r*" or Esql.url_original_to_lower like "*node* -e*", 1, 0)
| eval Esql.contains_shell = case(Esql.url_original_to_lower like "*/bin/bash*" or Esql.url_original_to_lower like "*bash*-c*" or Esql.url_original_to_lower like "*/bin/sh*" or Esql.url_original_to_lower rlike "*sh.{1,2}-c*", 1, 0)
| eval Esql.contains_nc = case(Esql.url_original_to_lower like "*netcat*" or Esql.url_original_to_lower like "*ncat*" or Esql.url_original_to_lower rlike """.*nc.{1,2}[0-9]{1,3}(\.[0-9]{1,3}){3}.{1,2}[0-9]{1,5}.*""" or Esql.url_original_to_lower like "*nc.openbsd*" or Esql.url_original_to_lower like "*nc.traditional*" or Esql.url_original_to_lower like "*socat*", 1, 0)
| eval Esql.contains_devtcp = case(Esql.url_original_to_lower like "*/dev/tcp/*" or Esql.url_original_to_lower like "*/dev/udp/*", 1, 0)
| eval Esql.contains_helpers = case((Esql.url_original_to_lower like "*/bin/*" or Esql.url_original_to_lower like "*/usr/bin/*") and (Esql.url_original_to_lower like "*mkfifo*" or Esql.url_original_to_lower like "*nohup*" or Esql.url_original_to_lower like "*setsid*" or Esql.url_original_to_lower like "*busybox*"), 1, 0)
| eval Esql.contains_sus_cli = case(Esql.url_original_to_lower like "*import*pty*spawn*" or Esql.url_original_to_lower like "*import*subprocess*call*" or Esql.url_original_to_lower like "*tcpsocket.new*" or Esql.url_original_to_lower like "*tcpsocket.open*" or Esql.url_original_to_lower like "*io.popen*" or Esql.url_original_to_lower like "*os.execute*" or Esql.url_original_to_lower like "*fsockopen*", 1, 0)
| eval Esql.contains_privileges = case(Esql.url_original_to_lower like "*chmod*+x", 1, 0)
| eval Esql.contains_downloader = case(Esql.url_original_to_lower like "*curl *" or Esql.url_original_to_lower like "*wget *" , 1, 0)
| eval Esql.contains_file_read_keywords = case(Esql.url_original_to_lower like "*/etc/shadow*" or Esql.url_original_to_lower like "*/etc/passwd*" or Esql.url_original_to_lower like "*/root/.ssh/*" or Esql.url_original_to_lower like "*/home/*/.ssh/*" or Esql.url_original_to_lower like "*~/.ssh/*" or Esql.url_original_to_lower like "*/proc/self/environ*", 1, 0)
| eval Esql.contains_base64_cmd = case(Esql.url_original_to_lower like "*base64*-d*" or Esql.url_original_to_lower like "*echo*
| *base64*", 1, 0)
| eval Esql.contains_suspicious_path = case(Esql.url_original_to_lower like "*/tmp/*" or Esql.url_original_to_lower like "*/var/tmp/*" or Esql.url_original_to_lower like "*/dev/shm/*" or Esql.url_original_to_lower like "*/root/*" or Esql.url_original_to_lower like "*/home/*/*" or Esql.url_original_to_lower like "*/var/www/*" or Esql.url_original_to_lower like "*/etc/cron.*/*", 1, 0)
| eval Esql.any_payload_keyword = case(
    Esql.contains_interpreter == 1 or Esql.contains_shell == 1 or Esql.contains_nc == 1 or Esql.contains_devtcp == 1 or
    Esql.contains_helpers == 1 or Esql.contains_sus_cli == 1 or Esql.contains_privileges == 1 or Esql.contains_downloader == 1 or
    Esql.contains_file_read_keywords == 1 or Esql.contains_base64_cmd == 1 or Esql.contains_suspicious_path == 1, 1, 0)
| keep
    @timestamp,
    Esql.url_original_to_lower,
    Esql.any_payload_keyword,
    Esql.contains_interpreter,
    Esql.contains_shell,
    Esql.contains_nc,
    Esql.contains_devtcp,
    Esql.contains_helpers,
    Esql.contains_sus_cli,
    Esql.contains_privileges,
    Esql.contains_downloader,
    Esql.contains_file_read_keywords,
    Esql.contains_base64_cmd,
    Esql.contains_suspicious_path,
    source.ip,
    destination.ip,
    agent.id,
    http.request.method,
    http.response.status_code,
    user_agent.original,
    agent.name,
    data_stream.dataset,
    data_stream.namespace
| stats
    Esql.event_count = count(),
    Esql.url_path_count_distinct = count_distinct(Esql.url_original_to_lower),

    // General fields

    Esql.agent_name_values = values(agent.name),
    Esql.agent_id_values = values(agent.id),
    Esql.url_path_values = values(Esql.url_original_to_lower),
    Esql.http.response.status_code_values = values(http.response.status_code),
    Esql.user_agent_original_values = values(user_agent.original),
    Esql.data_stream_dataset_values = values(data_stream.dataset),
    Esql.data_stream_namespace_values = values(data_stream.namespace),

    // Rule Specific fields
    Esql.any_payload_keyword_max = max(Esql.any_payload_keyword),
    Esql.contains_interpreter_values = values(Esql.contains_interpreter),
    Esql.contains_shell_values = values(Esql.contains_shell),
    Esql.contains_nc_values = values(Esql.contains_nc),
    Esql.contains_devtcp_values = values(Esql.contains_devtcp),
    Esql.contains_helpers_values = values(Esql.contains_helpers),
    Esql.contains_sus_cli_values = values(Esql.contains_sus_cli),
    Esql.contains_privileges_values = values(Esql.contains_privileges),
    Esql.contains_downloader_values = values(Esql.contains_downloader),
    Esql.contains_file_read_keywords_values = values(Esql.contains_file_read_keywords),
    Esql.contains_base64_cmd_values = values(Esql.contains_base64_cmd),
    Esql.contains_suspicious_path_values = values(Esql.contains_suspicious_path)

    by source.ip, agent.id
| where
    // Filter for potential command injection attempts with low event counts to reduce false positives
    Esql.any_payload_keyword_max == 1 and Esql.event_count < 5