Fix mysql parser handling of connection phase (#8173)

The mysql protocol parser assumes that transactions can only be started
by the client. This is true once the connection has been negotiated
("command phase"), but not during initial handshake ("connection phase").

This causes parsing problems when a connection is monitored from the
start, as sometimes the connection phase leaves the parser confused on
which side is client.

This patch modifies how client-side is detected, which can only be done
by looking at the destination port.

CHANGELOG.asciidoc

@@ -80,6 +80,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 - Fixed a seccomp related error where the `fcntl64` syscall was not permitted
   on 32-bit Linux and the sniffer failed to start. {issue}7839[7839]
 - Added missing `cmdline` and `client_cmdline` fields to index template. {pull}8258[8258]
+- Fixed the mysql missing transactions if monitoring a connection from the start. {pull}8173[8173]
 
 *Winlogbeat*
 

packetbeat/protos/mysql/mysql.go

@@ -208,10 +208,18 @@ func (stream *mysqlStream) prepareForNewMessage() {
 	stream.data = stream.data[stream.parseOffset:]
 	stream.parseState = mysqlStateStart
 	stream.parseOffset = 0
-	stream.isClient = false
 	stream.message = nil
 }
 
+func (mysql *mysqlPlugin) isServerPort(port uint16) bool {
+	for _, sPort := range mysql.ports {
+		if uint16(sPort) == port {
+			return true
+		}
+	}
+	return false
+}
+
 func mysqlMessageParser(s *mysqlStream) (bool, bool) {
 	logp.Debug("mysqldetailed", "MySQL parser called. parseState = %s", s.parseState)
 
@@ -229,27 +237,20 @@ func mysqlMessageParser(s *mysqlStream) (bool, bool) {
 			m.seq = hdr[3]
 			m.typ = hdr[4]
 
-			logp.Debug("mysqldetailed", "MySQL Header: Packet length %d, Seq %d, Type=%d", m.packetLength, m.seq, m.typ)
+			logp.Debug("mysqldetailed", "MySQL Header: Packet length %d, Seq %d, Type=%d isClient=%v", m.packetLength, m.seq, m.typ, s.isClient)
 
-			if m.seq == 0 {
+			if s.isClient {
 				// starts Command Phase
-
-				if m.typ == mysqlCmdQuery {
+				if m.seq == 0 && m.typ == mysqlCmdQuery {
 					// parse request
 					m.isRequest = true
 					m.start = s.parseOffset
 					s.parseState = mysqlStateEatMessage
-
 				} else {
 					// ignore command
 					m.ignoreMessage = true
 					s.parseState = mysqlStateEatMessage
 				}
-
-				if !s.isClient {
-					s.isClient = true
-				}
-
 			} else if !s.isClient {
 				// parse response
 				m.isRequest = false
@@ -275,11 +276,6 @@ func mysqlMessageParser(s *mysqlStream) (bool, bool) {
 					m.ignoreMessage = true
 					s.parseState = mysqlStateEatMessage
 				}
-
-			} else {
-				// something else, not expected
-				logp.Debug("mysql", "Unexpected MySQL message of type %d received.", m.typ)
-				return false, false
 			}
 
 		case mysqlStateEatMessage:
@@ -500,9 +496,14 @@ func (mysql *mysqlPlugin) Parse(pkt *protos.Packet, tcptuple *common.TCPTuple,
 	}
 
 	if priv.data[dir] == nil {
+		dstPort := tcptuple.DstPort
+		if dir == tcp.TCPDirectionReverse {
+			dstPort = tcptuple.SrcPort
+		}
 		priv.data[dir] = &mysqlStream{
-			data:    pkt.Payload,
-			message: &mysqlMessage{ts: pkt.Ts},
+			data:     pkt.Payload,
+			message:  &mysqlMessage{ts: pkt.Ts},
+			isClient: mysql.isServerPort(dstPort),
 		}
 	} else {
 		// concatenate bytes
@@ -521,7 +522,7 @@ func (mysql *mysqlPlugin) Parse(pkt *protos.Packet, tcptuple *common.TCPTuple,
 		}
 
 		ok, complete := mysqlMessageParser(priv.data[dir])
-		//logp.Debug("mysqldetailed", "mysqlMessageParser returned ok=%b complete=%b", ok, complete)
+		logp.Debug("mysqldetailed", "mysqlMessageParser returned ok=%v complete=%v", ok, complete)
 		if !ok {
 			// drop this tcp stream. Will retry parsing with the next
 			// segment in it

packetbeat/protos/mysql/mysql_test.go

@@ -23,6 +23,7 @@ import (
 	"encoding/hex"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -31,10 +32,11 @@ import (
 	"github.com/elastic/beats/libbeat/logp"
 
 	"github.com/elastic/beats/packetbeat/protos"
-
-	"time"
+	"github.com/elastic/beats/packetbeat/protos/tcp"
 )
 
+const serverPort = 3306
+
 type eventStore struct {
 	events []beat.Event
 }
@@ -55,6 +57,7 @@ func mysqlModForTests(store *eventStore) *mysqlPlugin {
 
 	var mysql mysqlPlugin
 	config := defaultConfig
+	config.Ports = []int{serverPort}
 	mysql.init(callback, &config)
 	return &mysql
 }
@@ -82,7 +85,7 @@ func TestMySQLParser_simpleRequest(t *testing.T) {
 		t.Errorf("Failed to decode hex string")
 	}
 
-	stream := &mysqlStream{data: message, message: new(mysqlMessage)}
+	stream := &mysqlStream{data: message, message: new(mysqlMessage), isClient: true}
 
 	ok, complete := mysqlMessageParser(stream)
 
@@ -364,7 +367,7 @@ func TestParseMySQL_simpleUpdateResponse(t *testing.T) {
 		countHandleMysql++
 	}
 
-	mysql.Parse(&pkt, &tuple, 1, private)
+	mysql.Parse(&pkt, &tuple, tcp.TCPDirectionOriginal, private)
 
 	if countHandleMysql != 1 {
 		t.Errorf("handleMysql not called")
@@ -405,7 +408,7 @@ func TestParseMySQL_threeResponses(t *testing.T) {
 		countHandleMysql++
 	}
 
-	mysql.Parse(&pkt, &tuple, 1, private)
+	mysql.Parse(&pkt, &tuple, tcp.TCPDirectionOriginal, private)
 
 	if countHandleMysql != 3 {
 		t.Errorf("handleMysql not called three times")
@@ -447,7 +450,7 @@ func TestParseMySQL_splitResponse(t *testing.T) {
 		countHandleMysql++
 	}
 
-	private = mysql.Parse(&pkt, &tuple, 1, private).(mysqlPrivateData)
+	private = mysql.Parse(&pkt, &tuple, tcp.TCPDirectionOriginal, private).(mysqlPrivateData)
 	if countHandleMysql != 0 {
 		t.Errorf("handleMysql called on first run")
 	}
@@ -482,7 +485,7 @@ func testTCPTuple() *common.TCPTuple {
 		IPLength: 4,
 		BaseTuple: common.BaseTuple{
 			SrcIP: net.IPv4(192, 168, 0, 1), DstIP: net.IPv4(192, 168, 0, 2),
-			SrcPort: 6512, DstPort: 3306,
+			SrcPort: 6512, DstPort: serverPort,
 		},
 	}
 	t.ComputeHashables()
@@ -540,17 +543,17 @@ func Test_gap_in_response(t *testing.T) {
 
 	private := protos.ProtocolData(new(mysqlPrivateData))
 
-	private = mysql.Parse(&req, tcptuple, 0, private)
-	private = mysql.Parse(&resp, tcptuple, 1, private)
+	private = mysql.Parse(&req, tcptuple, tcp.TCPDirectionOriginal, private)
+	private = mysql.Parse(&resp, tcptuple, tcp.TCPDirectionReverse, private)
 
 	logp.Debug("mysql", "Now sending gap..")
 
-	_, drop := mysql.GapInStream(tcptuple, 1, 10, private)
+	_, drop := mysql.GapInStream(tcptuple, tcp.TCPDirectionReverse, 10, private)
 	assert.Equal(t, true, drop)
 
 	trans := expectTransaction(t, store)
 	assert.NotNil(t, trans)
-	assert.Equal(t, trans["notes"], []string{"Packet loss while capturing the response"})
+	assert.Equal(t, []string{"Packet loss while capturing the response"}, trans["notes"])
 }
 
 // Test that loss of data during the request doesn't result in a
@@ -567,7 +570,7 @@ func Test_gap_in_eat_message(t *testing.T) {
 			"66726f6d20746573")
 	assert.Nil(t, err)
 
-	stream := &mysqlStream{data: reqData, message: new(mysqlMessage)}
+	stream := &mysqlStream{data: reqData, message: new(mysqlMessage), isClient: true}
 	ok, complete := mysqlMessageParser(stream)
 	assert.Equal(t, true, ok)
 	assert.Equal(t, false, complete)

packetbeat/tests/system/test_0067_mysql_connection_phase.py

@@ -0,0 +1,23 @@
+from packetbeat import BaseTest
+
+"""
+Tests that the negotiation phase at the beginning of a mysql connection
+doesn't leave the parser in a broken state.
+"""
+
+
+class Test(BaseTest):
+
+    def test_connection_phase(self):
+        """
+        This tests that requests are still captured from a mysql stream that
+        starts with the "connection phase" negotiation.
+        """
+        self.render_config_template(
+            mysql_ports=[3306],
+        )
+        self.run_packetbeat(pcap="mysql_connection.pcap")
+
+        objs = self.read_output()
+        assert len(objs) == 1
+        assert objs[0]['query'] == 'SELECT DATABASE()'