✄𐘗
text code block:
✄𐘗from logs-endpoint.events.process-* metadata _id, _index, _version
| mv_expand event.action 
| where
    host.os.type == "linux" and
    event.type == "start" and
    event.action == "exec" and
    process.name in ("scp", "ftp", "sftp", "vsftpd", "sftp-server", "rsync") and (
        (process.parent.name in ("bash", "dash", "sh", "tcsh", "csh", "zsh", "ksh", "fish")) or
        (
          process.parent.name like ".*" or
          process.parent.name like "*.elf" or
          process.parent.name like "*.sh" or
          process.parent.name like "*.py" or
          process.parent.name like "*.rb" or
          process.parent.name like "*.pl" or
          process.parent.name like "*.lua*" or
          process.parent.name like "*.php*" or
          process.parent.name like "*.js"
        ) or
        (
          process.parent.executable like "/tmp/*" or
          process.parent.executable like "/var/tmp/*" or
          process.parent.executable like "/dev/shm/*" or
          process.parent.executable like "./*" or
          process.parent.executable like "/run/*" or
          process.parent.executable like "/var/run/*" or
          process.parent.executable like "/boot/*" or
          process.parent.executable like "/sys/*" or
          process.parent.executable like "/lost+found/*" or
          process.parent.executable like "/proc/*" or
          process.parent.executable like "/var/mail/*" or
          process.parent.executable like "/var/www/*" or
          process.parent.executable like "/home/*" or
          process.parent.executable like "/root/*"
        )
    )

| keep
    @timestamp,
    _id,
    _index,
    _version,
    host.os.type,
    event.type,
    event.action,
    process.name,
    process.parent.name,
    process.parent.executable,
    process.executable,
    agent.id,
    host.name,
    event.dataset,
    data_stream.namespace

| stats
    Esql.event_count = count(),
    Esql.agent_id_count_distinct = count_distinct(agent.id),
    Esql.host_name_values = values(host.name),
    Esql.agent_id_values = values(agent.id),
    Esql.event_dataset_values = values(event.dataset),
    Esql.data_stream_namespace_values = values(data_stream.namespace)

  by process.executable, process.parent.executable

| where
    Esql.agent_id_count_distinct == 1 and
    Esql.event_count < 5
| sort Esql.event_count asc

// Extract unique values to ECS fields for alerts exclusion
| eval agent.id = mv_min(Esql.agent_id_values),
       host.name = mv_min(Esql.host_name_values)

| keep agent.id, host.name, process.executable, process.parent.executable, Esql.*