Detect Publish/Subscribe Protocols in Google Chronicle

rule t1071_005_pubsub_protocol_external_c2 {
  meta:
    author          = "Detection Engineering"
    description     = "Detects repeated outbound connections to pub/sub protocol ports (MQTT, XMPP, AMQP, STOMP) targeting non-RFC-1918 addresses. High suspicion for XMPP/STOMP and non-broker MQTT; medium for AMQP."
    mitre_attack_tactic   = "Command and Control"
    mitre_attack_technique = "T1071.005"
    severity        = "HIGH"
    priority        = "HIGH"

  events:
    $net.metadata.event_type = "NETWORK_CONNECTION"
    $net.network.direction   = "OUTBOUND"
    $net.target.port in (
      1883, 8883,
      5222, 5223, 5269,
      5672, 5671,
      61613, 61614
    )
    // Exclude RFC-1918 and loopback destinations
    not re.regex(
      $net.target.ip,
      `^(10\.|172\.(1[6-9]|2[0-9]|3[01])\.|192\.168\.|127\.)`
    )
    // Exclude known-legitimate MQTT clients
    not (
      $net.target.port in (1883, 8883) and
      re.regex($net.principal.process.file.full_path, `(?i)(mosquitto_pub|mosquitto_sub|mosquitto)(\.exe)?$`)
    )
    $net.principal.hostname = $host

  match:
    $host over 1h

  outcome:
    $connection_count  = count_distinct($net.target.ip)
    $protocols_seen    = array_distinct(
      if($net.target.port = 1883 or $net.target.port = 8883,  "MQTT",
      if($net.target.port = 5222 or $net.target.port = 5223 or $net.target.port = 5269, "XMPP",
      if($net.target.port = 5672 or $net.target.port = 5671,  "AMQP",
      if($net.target.port = 61613 or $net.target.port = 61614, "STOMP", "Unknown"))))
    )
    $initiating_process = array_distinct($net.principal.process.file.full_path)
    $target_ips         = array_distinct($net.target.ip)

  condition:
    #net > 1
}