Avoid eager MQTT response buffer access (#2515)

Author: MrAlias

pkg/ebpf/common/mqtt_detect_transform.go

@@ -57,13 +57,16 @@ func packetTypeToMethod(packetType mqttparser.PacketType) string {
 // should be ignored for span creation (e.g., control packets like CONNECT).
 func ProcessPossibleMQTTEvent(event *TCPRequestInfo, pkt *largebuf.LargeBuffer, rpkt *largebuf.LargeBuffer) (*MQTTInfo, bool, error) {
 	pktView := pkt.UnsafeView()
-	rpktView := rpkt.UnsafeView()
 
 	m, ignore, err := ProcessMQTTEvent(pktView)
 	if err != nil {
+		if rpkt == nil {
+			return m, ignore, err
+		}
+
 		// If we are getting the information in the response buffer, the event
 		// must be reversed and that's how we captured it.
-		m, ignore, err = ProcessMQTTEvent(rpktView)
+		m, ignore, err = ProcessMQTTEvent(rpkt.UnsafeView())
 		if err == nil && !ignore {
 			reverseTCPEvent(event)
 		}

pkg/ebpf/common/mqtt_detect_transform_test.go

@@ -269,6 +269,21 @@ func TestProcessPossibleMQTTEvent(t *testing.T) {
 				PacketID:   0,
 			},
 		},
+		{
+			name: "PUBLISH in request buffer without response buffer",
+			request: []byte{
+				0x30,       // PUBLISH, QoS 0
+				0x0c,       // Remaining length: 12
+				0x00, 0x0a, // Topic length: 10
+				't', 'e', 's', 't', '/', 't', 'o', 'p', 'i', 'c',
+			},
+			expected: &MQTTInfo{
+				PacketType: mqttparser.PacketTypePUBLISH,
+				Topic:      "test/topic",
+				QoS:        mqttparser.QoSAtMostOnce,
+				PacketID:   0,
+			},
+		},
 		{
 			name:    "PUBLISH in response buffer (reversed)",
 			request: []byte{},
@@ -291,12 +306,21 @@ func TestProcessPossibleMQTTEvent(t *testing.T) {
 			response: []byte{0x00, 0x00, 0x00},
 			err:      true,
 		},
+		{
+			name:    "Invalid request without response buffer",
+			request: []byte{0x00, 0x00, 0x00},
+			err:     true,
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			event := &TCPRequestInfo{}
-			res, _, err := ProcessPossibleMQTTEvent(event, largebuf.NewLargeBufferFrom(tt.request), largebuf.NewLargeBufferFrom(tt.response))
+			var response *largebuf.LargeBuffer
+			if tt.response != nil {
+				response = largebuf.NewLargeBufferFrom(tt.response)
+			}
+			res, _, err := ProcessPossibleMQTTEvent(event, largebuf.NewLargeBufferFrom(tt.request), response)
 			if tt.err {
 				assert.Error(t, err)
 				return