T1071.005 Publish/Subscribe Protocols Detection — KQL & SPL Queries for Sentinel and Splunk

Microsoft Sentinel / Defender

kusto

let TimeWindow = 24h;
let PubSubPorts = dynamic([1883, 8883, 5222, 5223, 5269, 5672, 5671, 61613, 61614]);
// Detect connections to pub/sub protocol ports
DeviceNetworkEvents
| where Timestamp > ago(TimeWindow)
| where RemotePort in (PubSubPorts)
| where ActionType == "ConnectionSuccess"
| extend Protocol = case(
    RemotePort in (1883, 8883), "MQTT",
    RemotePort in (5222, 5223, 5269), "XMPP",
    RemotePort in (5672, 5671), "AMQP",
    RemotePort in (61613, 61614), "STOMP",
    "Unknown")
| extend IsExternal = RemoteIPType == "Public"
| summarize
    ConnectionCount = count(),
    BytesSent = sum(SentBytes),
    BytesReceived = sum(ReceivedBytes),
    FirstSeen = min(Timestamp),
    LastSeen = max(Timestamp)
    by DeviceName, InitiatingProcessFileName, RemoteIP, RemotePort, Protocol, IsExternal, AccountName
| where ConnectionCount > 1
| extend Suspicion = case(
    IsExternal and Protocol in ("XMPP", "STOMP"), "high",
    IsExternal and Protocol == "MQTT" and InitiatingProcessFileName !in~ ("mosquitto_pub", "mosquitto_sub", "mosquitto"), "high",
    IsExternal and Protocol == "AMQP", "medium",
    not(IsExternal) and Protocol in ("XMPP", "STOMP"), "medium",
    "low")
| where Suspicion in ("high", "medium")
| project LastSeen, DeviceName, AccountName, InitiatingProcessFileName, RemoteIP, Protocol, RemotePort, IsExternal, ConnectionCount, BytesSent, BytesReceived, Suspicion
| sort by Suspicion asc, ConnectionCount desc

high severity medium confidence

Data Sources

Network Traffic: Network Connection Creation Network Traffic: Network Traffic Flow Microsoft Defender for Endpoint

Required Tables

DeviceNetworkEvents

False Positives

IoT platforms and home automation systems that legitimately use MQTT for device communication (Home Assistant, AWS IoT Core)
Chat and messaging applications using XMPP (Pidgin, Conversations, Cisco Jabber)
Message queue infrastructure (RabbitMQ, ActiveMQ, Apache Kafka) used for application integration
Development and testing environments with MQTT brokers for IoT prototyping

Splunk

spl

index=network OR index=firewall sourcetype IN ("stream:tcp", "pan:traffic", "cisco:asa")
  (dest_port IN (1883, 8883, 5222, 5223, 5269, 5672, 5671, 61613, 61614))
  action=allowed
| eval Protocol=case(
    dest_port IN (1883, 8883), "MQTT",
    dest_port IN (5222, 5223, 5269), "XMPP",
    dest_port IN (5672, 5671), "AMQP",
    dest_port IN (61613, 61614), "STOMP",
    1=1, "Unknown")
| eval IsExternal=if(NOT (dest_ip="10.*" OR dest_ip="172.16.*" OR dest_ip="192.168.*" OR dest_ip="127.*"), "yes", "no")
| stats count as ConnectionCount, sum(bytes_out) as BytesSent, sum(bytes_in) as BytesReceived, earliest(_time) as FirstSeen, latest(_time) as LastSeen, values(src_user) as Users by src_ip, dest_ip, dest_port, Protocol, IsExternal, process_name
| where ConnectionCount > 1
| eval Suspicion=case(
    IsExternal="yes" AND Protocol IN ("XMPP", "STOMP"), "high",
    IsExternal="yes" AND Protocol="MQTT" AND NOT match(process_name, "(?i)mosquitto"), "high",
    IsExternal="yes" AND Protocol="AMQP", "medium",
    IsExternal="no" AND Protocol IN ("XMPP", "STOMP"), "medium",
    1=1, "low")
| where Suspicion IN ("high", "medium")
| table FirstSeen, LastSeen, src_ip, dest_ip, Protocol, dest_port, IsExternal, process_name, Users, ConnectionCount, BytesSent, BytesReceived, Suspicion
| sort Suspicion, - ConnectionCount

high severity medium confidence

Data Sources

Network Traffic: Network Connection Creation Network Traffic: Network Traffic Flow Firewall Logs

Required Sourcetypes

stream:tcp pan:traffic cisco:asa

False Positives

IoT platforms and home automation systems that legitimately use MQTT for device communication (Home Assistant, AWS IoT Core)
Chat and messaging applications using XMPP (Pidgin, Conversations, Cisco Jabber)
Message queue infrastructure (RabbitMQ, ActiveMQ, Apache Kafka) used for application integration
Development and testing environments with MQTT brokers for IoT prototyping

Elastic Security (EQL)

eql

network where event.action == "connection_attempted" and
  destination.port in (1883, 8883, 5222, 5223, 5269, 5672, 5671, 61613, 61614) and
  not cidrmatch(destination.ip, "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16", "127.0.0.0/8") and
  not (
    destination.port in (1883, 8883) and
    process.name in~ ("mosquitto", "mosquitto_pub", "mosquitto_sub", "mosquitto.exe")
  )

high severity medium confidence

Data Sources

Elastic Endpoint Security agent (network events) Elastic Agent network monitoring integration Packetbeat with flow monitoring enabled

Required Tables

logs-endpoint.events.network-* logs-network_traffic.flow-*

False Positives

IoT gateway software or home-automation platforms (Home Assistant, Node-RED, openHAB) legitimately using MQTT on 1883/8883 for device telemetry — scope-exclude those hosts or add their processes to the allowlist
Enterprise and open-source XMPP chat clients (Pidgin, Psi+, Spark, Gajim) connecting to corporate or cloud-hosted Jabber/XMPP servers on port 5222
Microservice middleware (RabbitMQ client libraries, Celery workers, Pika, aio-pika) making AMQP connections on 5672/5671 to legitimate cloud message brokers such as CloudAMQP or Azure Service Bus

IBM QRadar (AQL)

sql

SELECT
  DATEFORMAT(MIN(starttime)/1000, 'yyyy-MM-dd HH:mm:ss') AS FirstSeen,
  DATEFORMAT(MAX(starttime)/1000, 'yyyy-MM-dd HH:mm:ss') AS LastSeen,
  sourceip AS SrcIP,
  destinationip AS DstIP,
  destinationport AS DstPort,
  CASE
    WHEN destinationport IN (1883, 8883)             THEN 'MQTT'
    WHEN destinationport IN (5222, 5223, 5269)       THEN 'XMPP'
    WHEN destinationport IN (5672, 5671)             THEN 'AMQP'
    WHEN destinationport IN (61613, 61614)           THEN 'STOMP'
    ELSE 'Unknown'
  END AS Protocol,
  username,
  COUNT(*) AS ConnectionCount,
  SUM(sourcebytes)      AS BytesSent,
  SUM(destinationbytes) AS BytesReceived,
  CASE
    WHEN destinationport IN (5222, 5223, 5269, 61613, 61614) THEN 'high'
    WHEN destinationport IN (1883, 8883)                     THEN 'high'
    WHEN destinationport IN (5672, 5671)                     THEN 'medium'
    ELSE 'low'
  END AS Suspicion
FROM flows
WHERE starttime > NOW() - 86400000
  AND destinationport IN (1883, 8883, 5222, 5223, 5269, 5672, 5671, 61613, 61614)
  AND NOT (
        destinationip ILIKE '10.%'
     OR destinationip ILIKE '172.16.%' OR destinationip ILIKE '172.17.%'
     OR destinationip ILIKE '172.18.%' OR destinationip ILIKE '172.19.%'
     OR destinationip ILIKE '172.20.%' OR destinationip ILIKE '172.21.%'
     OR destinationip ILIKE '172.22.%' OR destinationip ILIKE '172.23.%'
     OR destinationip ILIKE '172.24.%' OR destinationip ILIKE '172.25.%'
     OR destinationip ILIKE '172.26.%' OR destinationip ILIKE '172.27.%'
     OR destinationip ILIKE '172.28.%' OR destinationip ILIKE '172.29.%'
     OR destinationip ILIKE '172.30.%' OR destinationip ILIKE '172.31.%'
     OR destinationip ILIKE '192.168.%'
     OR destinationip ILIKE '127.%'
  )
GROUP BY
  sourceip, destinationip, destinationport, Protocol, username, Suspicion
HAVING ConnectionCount > 1
   AND Suspicion IN ('high', 'medium')
ORDER BY Suspicion ASC, ConnectionCount DESC

high severity medium confidence

Data Sources

QRadar Network Activity (flows) — requires flow collector with Layer 4 visibility Firewall or network tap log sources forwarding to QRadar QRadar Network Insights (QNI) for deep packet metadata

Required Tables

flows

False Positives

Industrial control systems or OT networks with MQTT-based telemetry pipelines (Ignition, Kepware, AWS IoT Greengrass) regularly making outbound MQTT connections to cloud brokers
SaaS-connected business applications using AMQP-over-TLS (port 5671) for event streaming to Azure Service Bus, CloudAMQP, or Amazon MQ — source IPs will be application servers, not user endpoints
Remote workforce using XMPP-based softphones or video conferencing clients (Cisco Jabber, Openfire-backed tools) that open persistent sessions to external Jabber servers on port 5222

Sumo Logic CSE

sql

(_sourceCategory=network OR _sourceCategory=firewall OR _sourceCategory=pan OR _sourceCategory=cisco/asa)
| where dest_port in (1883, 8883, 5222, 5223, 5269, 5672, 5671, 61613, 61614)
| where action = "allowed" or action = "permit" or isnull(action)
| eval Protocol = if(dest_port = 1883 or dest_port = 8883, "MQTT",
    if(dest_port = 5222 or dest_port = 5223 or dest_port = 5269, "XMPP",
    if(dest_port = 5672 or dest_port = 5671, "AMQP",
    if(dest_port = 61613 or dest_port = 61614, "STOMP", "Unknown"))))
| eval IsExternal = if(
    !starts_with(dest_ip, "10.") and
    !starts_with(dest_ip, "192.168.") and
    !starts_with(dest_ip, "127.") and
    !matches(dest_ip, "172\.(1[6-9]|2[0-9]|3[01])\\..*"),
    "yes", "no")
| where IsExternal = "yes"
| count as ConnectionCount, sum(bytes_out) as BytesSent, sum(bytes_in) as BytesReceived,
    min(_messageTime) as FirstSeen, max(_messageTime) as LastSeen
    by src_ip, dest_ip, dest_port, Protocol, IsExternal, process_name
| where ConnectionCount > 1
| eval Suspicion = if((Protocol = "XMPP" or Protocol = "STOMP"), "high",
    if(Protocol = "MQTT" and !matches(process_name, "(?i)mosquitto.*"), "high",
    if(Protocol = "AMQP", "medium", "low")))
| where Suspicion in ("high", "medium")
| fields LastSeen, FirstSeen, src_ip, dest_ip, Protocol, dest_port, process_name,
    ConnectionCount, BytesSent, BytesReceived, Suspicion
| sort by Suspicion, -ConnectionCount

high severity medium confidence

Data Sources

Palo Alto Networks firewall logs (_sourceCategory=network/paloalto or pan) Cisco ASA/FTD logs (_sourceCategory=cisco/asa) Generic firewall or flow logs with normalized dest_port, src_ip, dest_ip fields Sumo Logic Cloud SIEM (CSE) network schema records

Required Tables

Network/firewall log sources with dest_port, src_ip, dest_ip, bytes_out, bytes_in fields

False Positives

Cloud-connected building management or SCADA systems sending MQTT telemetry (sensor readings, alarms) to external IoT platforms such as AWS IoT Core, Azure IoT Hub, or HiveMQ Cloud
Customer service or sales platforms that use XMPP under the hood (some legacy Salesforce integrations, LiveChat products) will appear as repeated port-5222 connections from application servers
Development and CI/CD pipelines running integration tests against external RabbitMQ or ActiveMQ instances — engineers' workstations or build agents will show AMQP connections to cloud broker endpoints

Google Chronicle / SecOps

yaral

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
}

high severity medium confidence

Data Sources

Google Chronicle UDM — network connection events from endpoint agents (CrowdStrike, Carbon Black, SentinelOne ingested via Chronicle forwarder) Chronicle network sensor or firewall log ingestion with UDM normalization Google Chronicle Security Operations with SIEM log ingestion

Required Tables

UDM events with metadata.event_type = NETWORK_CONNECTION principal.process.file.full_path populated (requires endpoint agent, not just firewall)

False Positives

Legitimate IoT fleet management agents installed on endpoints (Balena, Fleet.io, AWS IoT Greengrass Edge) that maintain persistent MQTT sessions to cloud brokers — these will fire repeatedly and should be whitelisted by process path or target IP
Corporate unified communications clients making persistent XMPP connections to Google Chat, Cisco WebEx, or on-premises Openfire servers — high connection-count sessions will match the #net > 1 threshold
Containers or VM agents on developer workstations running local-to-remote AMQP tests against CloudAMQP trial instances or public RabbitMQ test brokers during application development cycles

CrowdStrike LogScale (CQL)

cql

#event_simpleName = "NetworkConnectIP4"
| RemotePort in [1883, 8883, 5222, 5223, 5269, 5672, 5671, 61613, 61614]
// Exclude RFC-1918 and loopback
| !cidr(RemoteAddressIP4, subnet="10.0.0.0/8")
| !cidr(RemoteAddressIP4, subnet="172.16.0.0/12")
| !cidr(RemoteAddressIP4, subnet="192.168.0.0/16")
| !cidr(RemoteAddressIP4, subnet="127.0.0.0/8")
// Classify protocol
| Protocol := case {
    RemotePort in [1883, 8883]             => "MQTT";
    RemotePort in [5222, 5223, 5269]       => "XMPP";
    RemotePort in [5672, 5671]             => "AMQP";
    RemotePort in [61613, 61614]           => "STOMP";
    *                                      => "Unknown"
  }
// Score suspicion — exclude known MQTT broker tools before flagging MQTT
| Suspicion := case {
    Protocol in ["XMPP", "STOMP"]                                                     => "high";
    Protocol = "MQTT" AND ImageFileName != /(?i)(mosquitto_pub|mosquitto_sub|mosquitto)(\.exe)?$/ => "high";
    Protocol = "AMQP"                                                                 => "medium";
    *                                                                                  => "low"
  }
| Suspicion in ["high", "medium"]
| groupBy(
    [ComputerName, UserName, ImageFileName, RemoteAddressIP4, RemotePort, Protocol, Suspicion],
    function=[
      count(as=ConnectionCount),
      min(@timestamp, as=FirstSeen),
      max(@timestamp, as=LastSeen)
    ]
  )
| ConnectionCount > 1
| sort(Suspicion, order=asc)
| sort(ConnectionCount, order=desc)

high severity medium confidence

Data Sources

CrowdStrike Falcon Insight XDR endpoint telemetry (NetworkConnectIP4 events) Falcon Data Replicator (FDR) feed ingested into LogScale CrowdStrike Falcon LogScale (SIEM) — requires network connection telemetry enabled on sensor policy

Required Tables

Falcon telemetry: #event_simpleName = NetworkConnectIP4 Fields required: RemotePort, RemoteAddressIP4, ImageFileName, ComputerName, UserName

False Positives

Falcon-protected OT/IoT gateway hosts running MQTT broker clients (mosquitto, HiveMQ Edge, EMQX) for telemetry forwarding — the mosquitto regex exclusion handles standard installs but custom binary names or paths may still fire
Security tools and monitoring agents (Nagios, Zabbix, custom health-check scripts) that use AMQP or STOMP to push metrics to external observability platforms — these generate high-frequency repeated connections that will exceed the ConnectionCount > 1 threshold immediately
Developers running local pub/sub client tools (MQTT Explorer, AnotherMQTTClient, MQTTX) for debugging or testing against public broker sandboxes (broker.hivemq.com, test.mosquitto.org) — common on developer workstations and will appear as high-suspicion MQTT from non-standard process names

Publish/Subscribe Protocols

Data Sources

Required Tables

False Positives

Data Sources

Required Sourcetypes

False Positives

Data Sources

Required Tables

False Positives

Data Sources

Required Tables

False Positives

Data Sources

Required Tables

False Positives

Data Sources

Required Tables

False Positives

Data Sources

Required Tables

False Positives

Response Playbook

Investigation Guide

Atomic Red Team Tests

Unlock Pro Content

Related Detections

Parent Technique

Related Sub-techniques

Same Tactic: Command and Control

Popular Detections