deal all auth packet

capture/network.go

@@ -31,17 +31,17 @@ func init() {
 
 // networkCard is network device
 type networkCard struct {
-	name string
+	name       string
 	listenPort int
-	receiver chan model.QueryPiece
+	receiver   chan model.QueryPiece
 }
 
 func NewNetworkCard() (nc *networkCard) {
 	// init device
 	return &networkCard{
-		name: DeviceName,
+		name:       DeviceName,
 		listenPort: snifferPort,
-		receiver: make(chan model.QueryPiece, 100),
+		receiver:   make(chan model.QueryPiece, 100),
 	}
 }
 
@@ -97,7 +97,6 @@ func (nc *networkCard) Listen() (receiver chan model.QueryPiece) {
 	return nc.receiver
 }
 
-
 // Listen get a connection.
 func (nc *networkCard) listenNormal() {
 	go func() {
@@ -112,7 +111,7 @@ func (nc *networkCard) listenNormal() {
 			// capture packets according to a certain probability
 			capturePacketRate := communicator.GetTCPCapturePacketRate()
 			if capturePacketRate <= 0 {
-				time.Sleep(time.Second*1)
+				time.Sleep(time.Second * 1)
 				aliveCounter += 1
 				if aliveCounter >= checkCount {
 					aliveCounter = 0
@@ -124,7 +123,7 @@ func (nc *networkCard) listenNormal() {
 			data, ci, err = handler.ZeroCopyReadPacketData()
 			if err != nil {
 				log.Error(err.Error())
-				time.Sleep(time.Second*3)
+				time.Sleep(time.Second * 3)
 				continue
 			}
 
@@ -135,7 +134,10 @@ func (nc *networkCard) listenNormal() {
 
 			// send FIN tcp packet to avoid not complete session cannot be released
 			tcpPkt := packet.TransportLayer().(*layers.TCP)
-			if tcpPkt.FIN {
+			payLoad := tcpPkt.Payload
+			// deal FIN packet
+			// deal auth packet
+			if tcpPkt.FIN || (len(payLoad) >= 5 && sd.IsAuthPacket(payLoad[4])) {
 				nc.parseTCPPackage(packet)
 				continue
 			}
@@ -274,4 +276,3 @@ func readToServerPackage(
 
 	return
 }
-

session-dealer/controller.go

@@ -24,3 +24,12 @@ func CheckParams()  {
 		mysql.CheckParams()
 	}
 }
+
+func IsAuthPacket(val byte) bool {
+	switch serviceType {
+	case ServiceTypeMysql:
+		return mysql.IsAuthPacket(val)
+	default:
+		return mysql.IsAuthPacket(val)
+	}
+}

session-dealer/mysql/session.go

@@ -144,7 +144,7 @@ func (ms *MysqlSession) readFromClient(seqID int64, bytes []byte) {
 
 	if ms.expectReceiveSize == -1 {
 		// ignore invalid head package
-		if len(bytes) <= 4{
+		if len(bytes) <= 4 {
 			return
 		}
 
@@ -208,6 +208,10 @@ func (ms *MysqlSession) readFromClient(seqID int64, bytes []byte) {
 	// ms.expectReceiveSize = ms.expectReceiveSize - int(contentSize)
 }
 
+func IsAuthPacket(val byte) bool {
+	return val > 32
+}
+
 func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
 	defer ms.clear()
 
@@ -227,7 +231,7 @@ func (ms *MysqlSession) GenerateQueryPiece() (qp model.QueryPiece) {
 
 	var mqp *model.PooledMysqlQueryPiece
 	var querySQLInBytes []byte
-	if ms.cachedStmtBytes[0] > 32 {
+	if IsAuthPacket(ms.cachedStmtBytes[0]) {
 		userName, dbName, err := parseAuthInfo(ms.cachedStmtBytes)
 		if err != nil {
 			log.Errorf("parse auth info failed <-- %s", err.Error())