[5.6] Backport packetbeat fixes

CHANGELOG.asciidoc

@@ -36,6 +36,15 @@ https://github.com/elastic/beats/compare/v5.6.8...5.6[Check the HEAD diff]
 
 *Packetbeat*
 
+- Fix http status phrase parsing not allow spaces. {pull}5312[5312]
+- 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]
+- Fix out of bounds access to slice in MongoDB parser. {pull}6256[6256]
+- Fix sniffer hanging on exit under Linux. {pull}6535[6535]
+- Fix bounds check error in http parser causing a panic. {pull}6750[6750]
+
 *Winlogbeat*
 
 ==== Added
@@ -50,6 +59,9 @@ https://github.com/elastic/beats/compare/v5.6.8...5.6[Check the HEAD diff]
 
 *Packetbeat*
 
+- HTTP parses successfully on empty status phrase. {issue}6176[6176]
+- HTTP parser supports broken status line. {pull}6631[6631]
+
 *Winlogbeat*
 
 ==== Deprecated

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"})

packetbeat/protos/http/http_parser.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"time"
+	"unicode"
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/streambuf"
@@ -74,8 +75,9 @@ var (
 
 	constCRLF = []byte("\r\n")
 
-	constClose     = []byte("close")
-	constKeepAlive = []byte("keep-alive")
+	constClose       = []byte("close")
+	constKeepAlive   = []byte("keep-alive")
+	constHTTPVersion = []byte("HTTP/")
 
 	nameContentLength    = []byte("content-length")
 	nameContentType      = []byte("content-type")
@@ -139,13 +141,13 @@ func (*parser) parseHTTPLine(s *stream, m *message) (cont, ok, complete bool) {
 	var version []byte
 	var err error
 	fline := s.data[s.parseOffset:i]
-	if len(fline) < 8 {
+	if len(fline) < 9 {
 		if isDebug {
 			debugf("First line too small")
 		}
 		return false, false, false
 	}
-	if bytes.Equal(fline[0:5], []byte("HTTP/")) {
+	if bytes.Equal(fline[0:5], constHTTPVersion) {
 		//RESPONSE
 		m.isRequest = false
 		version = fline[5:8]
@@ -160,20 +162,23 @@ func (*parser) parseHTTPLine(s *stream, m *message) (cont, ok, complete bool) {
 		}
 	} else {
 		// REQUEST
-		slices := bytes.Fields(fline)
-		if len(slices) != 3 {
+		afterMethodIdx := bytes.IndexFunc(fline, unicode.IsSpace)
+		afterRequestURIIdx := bytes.LastIndexFunc(fline, unicode.IsSpace)
+
+		// Make sure we have the VERB + URI + HTTP_VERSION
+		if afterMethodIdx == -1 || afterRequestURIIdx == -1 || afterMethodIdx == afterRequestURIIdx {
 			if isDebug {
 				debugf("Couldn't understand HTTP request: %s", fline)
 			}
 			return false, false, false
 		}
 
-		m.method = common.NetString(slices[0])
-		m.requestURI = common.NetString(slices[1])
+		m.method = common.NetString(fline[:afterMethodIdx])
+		m.requestURI = common.NetString(fline[afterMethodIdx+1 : afterRequestURIIdx])
 
-		if bytes.Equal(slices[2][:5], []byte("HTTP/")) {
+		if bytes.Equal(fline[afterRequestURIIdx+1:afterRequestURIIdx+len(constHTTPVersion)+1], constHTTPVersion) {
 			m.isRequest = true
-			version = slices[2][5:]
+			version = fline[afterRequestURIIdx+len(constHTTPVersion)+1:]
 		} else {
 			if isDebug {
 				debugf("Couldn't understand HTTP version: %s", fline)
@@ -207,19 +212,18 @@ func parseResponseStatus(s []byte) (uint16, []byte, error) {
 		debugf("parseResponseStatus: %s", s)
 	}
 
+	var phrase []byte
 	p := bytes.IndexByte(s, ' ')
 	if p == -1 {
-		return 0, nil, errors.New("Not able to identify status code")
+		p = len(s)
+	} else {
+		phrase = s[p+1:]
 	}
-
-	code, _ := parseInt(s[0:p])
-
-	p = bytes.LastIndexByte(s, ' ')
-	if p == -1 {
-		return uint16(code), nil, errors.New("Not able to identify status code")
+	statusCode, err := parseInt(s[0:p])
+	if err != nil {
+		return 0, nil, fmt.Errorf("Unable to parse status code from [%s]", s)
 	}
-	phrase := s[p+1:]
-	return uint16(code), phrase, nil
+	return uint16(statusCode), phrase, nil
 }
 
 func parseVersion(s []byte) (uint8, uint8, error) {
@@ -354,8 +358,8 @@ func (parser *parser) parseHeader(m *message, data []byte) (bool, bool, int) {
 				if val, ok := m.headers[string(headerName)]; ok {
 					composed := make([]byte, len(val)+len(headerVal)+2)
 					off := copy(composed, val)
-					off = copy(composed[off:], []byte(", "))
-					copy(composed[off:], headerVal)
+					copy(composed[off:], []byte(", "))
+					copy(composed[off+2:], headerVal)
 
 					m.headers[string(headerName)] = composed
 				} else {

packetbeat/protos/http/http_test.go

@@ -4,6 +4,7 @@ package http
 
 import (
 	"bytes"
+	"fmt"
 	"net"
 	"regexp"
 	"strings"
@@ -380,6 +381,24 @@ func TestHttpParser_Response_HTTP_10_without_content_length(t *testing.T) {
 	assert.Equal(t, 4, message.contentLength)
 }
 
+func TestHttpParser_Response_without_phrase(t *testing.T) {
+	for idx, testCase := range []struct {
+		ok, complete bool
+		code         int
+		request      string
+	}{
+		{true, true, 200, "HTTP/1.1 200 \r\nContent-Length: 0\r\n\r\n"},
+		{true, true, 301, "HTTP/1.1 301\r\nContent-Length: 0\r\n\r\n"},
+	} {
+		msg := fmt.Sprintf("failed test case[%d]: \"%s\"", idx, testCase.request)
+		r, ok, complete := testParse(nil, testCase.request)
+		assert.Equal(t, testCase.ok, ok, msg)
+		assert.Equal(t, testCase.complete, complete, msg)
+		assert.Equal(t, testCase.code, int(r.statusCode), msg)
+		assert.Equal(t, "", string(r.statusPhrase), msg)
+	}
+}
+
 func TestHttpParser_splitResponse_midBody(t *testing.T) {
 	data1 := "HTTP/1.1 200 OK\r\n" +
 		"Date: Tue, 14 Aug 2012 22:31:45 GMT\r\n" +
@@ -525,6 +544,54 @@ func TestHttpParser_301_response(t *testing.T) {
 	assert.Equal(t, 290, msg.contentLength)
 }
 
+func TestHttpParser_PhraseContainsSpaces(t *testing.T) {
+	if testing.Verbose() {
+		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"http"})
+	}
+	response_404 := "HTTP/1.1 404 Not Found\r\n" +
+		"Server: Apache-Coyote/1.1\r\n" +
+		"Content-Type: text/html;charset=utf-8\r\n" +
+		"Content-Length: 18\r\n" +
+		"Date: Mon, 31 Jul 2017 11:31:53 GMT\r\n" +
+		"\r\n" +
+		"Http Response Body"
+
+	r, ok, complete := testParse(nil, response_404)
+	assert.True(t, ok)
+	assert.True(t, complete)
+	assert.Equal(t, 18, r.contentLength)
+	assert.Equal(t, "Not Found", string(r.statusPhrase))
+	assert.Equal(t, 404, int(r.statusCode))
+
+	response_500 := "HTTP/1.1 500 Internal Server Error\r\n" +
+		"Server: Apache-Coyote/1.1\r\n" +
+		"Content-Type: text/html;charset=utf-8\r\n" +
+		"Content-Length: 2\r\n" +
+		"Date: Mon, 30 Jul 2017 00:00:00 GMT\r\n" +
+		"\r\n" +
+		"xx"
+	r, ok, complete = testParse(nil, response_500)
+	assert.True(t, ok)
+	assert.True(t, complete)
+	assert.Equal(t, 2, r.contentLength)
+	assert.Equal(t, "Internal Server Error", string(r.statusPhrase))
+	assert.Equal(t, 500, int(r.statusCode))
+
+	broken := "HTTP/1.1 500 \r\n" +
+		"Server: Apache-Coyote/1.1\r\n" +
+		"Content-Type: text/html;charset=utf-8\r\n" +
+		"Content-Length: 2\r\n" +
+		"Date: Mon, 30 Jul 2017 00:00:00 GMT\r\n" +
+		"\r\n" +
+		"xx"
+	r, ok, complete = testParse(nil, broken)
+	assert.True(t, ok)
+	assert.True(t, complete)
+	assert.Equal(t, 2, r.contentLength)
+	assert.Equal(t, "", string(r.statusPhrase))
+	assert.Equal(t, 500, int(r.statusCode))
+}
+
 func TestEatBodyChunked(t *testing.T) {
 	if testing.Verbose() {
 		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"http", "httpdetailed"})
@@ -654,6 +721,45 @@ func TestEatBodyChunkedWaitCRLF(t *testing.T) {
 	}
 }
 
+func TestHttpParser_requestURIWithSpace(t *testing.T) {
+	if testing.Verbose() {
+		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"http", "httpdetailed"})
+	}
+
+	http := httpModForTests()
+	http.hideKeywords = []string{"password", "pass"}
+	http.parserConfig.sendHeaders = true
+	http.parserConfig.sendAllHeaders = true
+
+	// Non URL-encoded string, RFC says it should be encoded
+	data1 := "GET http://localhost:8080/test?password=two secret HTTP/1.1\r\n" +
+		"Host: www.google.com\r\n" +
+		"Connection: keep-alive\r\n" +
+		"User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_4) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/21.0.1180.75 Safari/537.1\r\n" +
+		"Accept: */*\r\n" +
+		"X-Chrome-Variations: CLa1yQEIj7bJAQiftskBCKS2yQEIp7bJAQiptskBCLSDygE=\r\n" +
+		"Referer: http://www.google.com/\r\n" +
+		"Accept-Encoding: gzip,deflate,sdch\r\n" +
+		"Accept-Language: en-US,en;q=0.8\r\n" +
+		"Content-Type: application/x-www-form-urlencoded\r\n" +
+		"Content-Length: 23\r\n" +
+		"Accept-Charset: ISO-8859-1,utf-8;q=0.7,*;q=0.3\r\n" +
+		"Cookie: PREF=ID=6b67d166417efec4:U=69097d4080ae0e15:FF=0:TM=1340891937:LM=1340891938:S=8t97UBiUwKbESvVX; NID=61=sf10OV-t02wu5PXrc09AhGagFrhSAB2C_98ZaI53-uH4jGiVG_yz9WmE3vjEBcmJyWUogB1ZF5puyDIIiB-UIdLd4OEgPR3x1LHNyuGmEDaNbQ_XaxWQqqQ59mX1qgLQ\r\n" +
+		"\r\n" +
+		"username=ME&pass=twosecret"
+	tp := newTestParser(http, data1)
+
+	msg, ok, complete := tp.parse()
+	assert.True(t, ok)
+	assert.True(t, complete)
+	rawMsg := tp.stream.data[tp.stream.message.start:tp.stream.message.end]
+	path, params, err := http.extractParameters(msg, rawMsg)
+	assert.Nil(t, err)
+	assert.Equal(t, "/test", path)
+	assert.Equal(t, string(msg.requestURI), "http://localhost:8080/test?password=two secret")
+	assert.False(t, strings.Contains(params, "two secret"))
+}
+
 func TestHttpParser_censorPasswordURL(t *testing.T) {
 	if testing.Verbose() {
 		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"http", "httpdetailed"})
@@ -1048,6 +1154,28 @@ func Test_gap_in_body_http1dot0(t *testing.T) {
 
 }
 
+func TestHttpParser_composedHeaders(t *testing.T) {
+	data := "HTTP/1.1 200 OK\r\n" +
+		"Content-Length: 0\r\n" +
+		"Date: Tue, 14 Aug 2012 22:31:45 GMT\r\n" +
+		"Set-Cookie: aCookie=yummy\r\n" +
+		"Set-Cookie: anotherCookie=why%20not\r\n" +
+		"\r\n"
+	http := httpModForTests()
+	http.parserConfig.sendHeaders = true
+	http.parserConfig.sendAllHeaders = true
+	message, ok, complete := testParse(http, data)
+
+	assert.True(t, ok)
+	assert.True(t, complete)
+	assert.False(t, message.isRequest)
+	assert.Equal(t, 200, int(message.statusCode))
+	assert.Equal(t, "OK", string(message.statusPhrase))
+	header, ok := message.headers["set-cookie"]
+	assert.True(t, ok)
+	assert.Equal(t, "aCookie=yummy, anotherCookie=why%20not", string(header))
+}
+
 func testCreateTCPTuple() *common.TCPTuple {
 	t := &common.TCPTuple{
 		IPLength: 4,

packetbeat/protos/mongodb/mongodb_parser.go

@@ -341,6 +341,9 @@ func (d *decoder) readDocument() (bson.M, error) {
 	start := d.i
 	documentLength, err := d.readInt32()
 	d.i = start + documentLength
+	if len(d.in) < d.i {
+		return nil, errors.New("document out of bounds")
+	}
 
 	documentMap := bson.M{}
 

packetbeat/protos/mongodb/mongodb_test.go

@@ -7,11 +7,12 @@ import (
 	"net"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/packetbeat/protos"
 	"github.com/elastic/beats/packetbeat/publish"
-	"github.com/stretchr/testify/assert"
 )
 
 // Helper function returning a Mongodb module that can be used
@@ -333,3 +334,31 @@ func TestMaxDocSize(t *testing.T) {
 
 	assert.Equal(t, "\"1234 ...\n\"123\"\n\"12\"", res["response"])
 }
+
+// Test for a (recovered) panic parsing document length in request/response messages
+func TestDocumentLengthBoundsChecked(t *testing.T) {
+	if testing.Verbose() {
+		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"mongodb", "mongodbdetailed"})
+	}
+
+	mongodb := mongodbModForTests()
+
+	// request and response from tests/pcaps/mongo_one_row.pcap
+	reqData, err := hex.DecodeString(
+		// Request message with out of bounds document
+		"320000000a000000ffffffffd4070000" +
+			"00000000746573742e72667374617572" +
+			"616e7473000000000001000000" +
+			// Document length (including itself)
+			"06000000" +
+			// Document (1 byte instead of 2)
+			"00")
+	assert.Nil(t, err)
+
+	tcptuple := testTCPTuple()
+	req := protos.Packet{Payload: reqData}
+	private := protos.ProtocolData(new(mongodbConnectionData))
+
+	private = mongodb.Parse(&req, tcptuple, 0, private)
+	assert.NotNil(t, private, "mongodb parser recovered from a panic")
+}