AMQP: Fix panic during parse of partial message (elastic#6384)

A message with a client header consisting on a partial frame (not
all data received for this frame) could result in a panic.

CHANGELOG.asciidoc

@@ -40,6 +40,7 @@ https://github.com/elastic/beats/compare/v5.6.8...5.6[Check the HEAD diff]
 - Fix http parse to allow to parse get request with space in the URI. {pull}5495[5495]
 - Fix mysql SQL parser to trim `\r` from Windows Server `SELECT\r\n\t1`. {pull}5572[5572]
 - Fix corruption when parsing repeated headers in an HTTP request or response. {pull}6325[6325]
+- Fix panic when parsing partial AMQP messages. {pull}6384[6384]
 
 *Winlogbeat*
 

packetbeat/protos/amqp/amqp_parser.go

@@ -73,7 +73,10 @@ func isProtocolHeader(data []byte) (isHeader bool, version string) {
 //func to read a frame header and check if it is valid and complete
 func readFrameHeader(data []byte) (ret *amqpFrame, err bool) {
 	var frame amqpFrame
-
+	if len(data) < 8 {
+		logp.Warn("Partial frame header, waiting for more data")
+		return nil, false
+	}
 	frame.size = binary.BigEndian.Uint32(data[3:7])
 	if len(data) < int(frame.size)+8 {
 		logp.Warn("Frame shorter than declared size, waiting for more data")

packetbeat/protos/amqp/amqp_test.go

@@ -83,6 +83,30 @@ func TestAmqp_FrameSize(t *testing.T) {
 	}
 }
 
+// Test that the parser doesn't panic on a partial message that includes
+// a client header
+func TestAmqp_PartialFrameSize(t *testing.T) {
+	if testing.Verbose() {
+		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"amqp", "amqpdetailed"})
+	}
+
+	amqp := amqpModForTests()
+
+	//incomplete frame
+	data, err := hex.DecodeString("414d515000060606010000000000")
+	assert.Nil(t, err)
+
+	stream := &amqpStream{data: data, message: new(amqpMessage)}
+	ok, complete := amqp.amqpMessageParser(stream)
+
+	if !ok {
+		t.Errorf("Parsing should not raise an error")
+	}
+	if complete {
+		t.Errorf("message should not be complete")
+	}
+}
+
 func TestAmqp_WrongShortStringSize(t *testing.T) {
 	if testing.Verbose() {
 		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"amqp", "amqpdetailed"})