从proxy获取client ip地址

capture/network.go

@@ -15,7 +15,7 @@ import (
 	"golang.org/x/net/bpf"
 
 	"github.com/google/gopacket/pcapgo"
-	proto "github.com/pires/go-proxyproto"
+	pp "github.com/pires/go-proxyproto"
 	log "github.com/sirupsen/logrus"
 	"github.com/zr-hebo/sniffer-agent/model"
 	sd "github.com/zr-hebo/sniffer-agent/session-dealer"
@@ -147,14 +147,8 @@ func (nc *networkCard) listenNormal() {
 
 			// deal auth packet
 			if sd.IsAuthPacket(tcpPkt.Payload) {
-				reader := bufio.NewReader(bytes.NewReader(data))
-				header, _ := proto.Read(reader)
-				var clientIP *string
-				if header != nil {
-					clientIPContent := header.SourceAddress.String()
-					clientIP = &clientIPContent
-				}
-				nc.parseTCPPackage(packet, clientIP)
+				authHeader, _ := pp.Read(bufio.NewReader(bytes.NewReader(tcpPkt.Payload)))
+				nc.parseTCPPackage(packet, authHeader)
 				continue
 			}
 
@@ -174,7 +168,7 @@ func (nc *networkCard) listenNormal() {
 	return
 }
 
-func (nc *networkCard) parseTCPPackage(packet gopacket.Packet, clientIP *string) {
+func (nc *networkCard) parseTCPPackage(packet gopacket.Packet, authHeader *pp.Header) {
 	var err error
 	defer func() {
 		if err != nil {
@@ -203,9 +197,19 @@ func (nc *networkCard) parseTCPPackage(packet gopacket.Packet, clientIP *string)
 	dstIP := ipInfo.DstIP.String()
 	srcPort := int(tcpPkt.SrcPort)
 	dstPort := int(tcpPkt.DstPort)
+
 	if dstPort == nc.listenPort {
+		// get client ip from proxy auth info
+		var clientIP *string
+		var clientPort int
+		if authHeader != nil && authHeader.SourceAddress.String() != srcIP {
+			clientIPContent := authHeader.SourceAddress.String()
+			clientIP = &clientIPContent
+			clientPort = int(authHeader.SourcePort)
+		}
+
 		// deal mysql server response
-		err = readToServerPackage(clientIP, &srcIP, srcPort, tcpPkt, nc.receiver)
+		err = readToServerPackage(clientIP, clientPort, &srcIP, srcPort, tcpPkt, nc.receiver)
 		if err != nil {
 			return
 		}
@@ -255,7 +259,8 @@ func readFromServerPackage(
 }
 
 func readToServerPackage(
-	clientIP, srcIP *string, srcPort int, tcpPkt *layers.TCP, receiver chan model.QueryPiece) (err error) {
+	clientIP *string, clientPort int, srcIP *string, srcPort int, tcpPkt *layers.TCP,
+	receiver chan model.QueryPiece) (err error) {
 	defer func() {
 		if err != nil {
 			log.Error("read package send from client to mysql server failed <-- %s", err.Error())
@@ -282,7 +287,7 @@ func readToServerPackage(
 	sessionKey := spliceSessionKey(srcIP, srcPort)
 	session := sessionPool[*sessionKey]
 	if session == nil {
-		session = sd.NewSession(sessionKey, clientIP, srcIP, srcPort, localIPAddr, snifferPort, receiver)
+		session = sd.NewSession(sessionKey, clientIP, clientPort, srcIP, srcPort, localIPAddr, snifferPort, receiver)
 		sessionPool[*sessionKey] = session
 	}
 

model/mysql_query_piece.go

@@ -26,7 +26,7 @@ type PooledMysqlQueryPiece struct {
 }
 
 func NewPooledMysqlQueryPiece(
-	sessionID, clientIP, visitUser, visitDB, clientHost, serverIP *string,
+	sessionID, clientIP, visitUser, visitDB, serverIP *string,
 	clientPort, serverPort int, throwPacketRate float64, stmtBeginTime int64) (
 	mqp *PooledMysqlQueryPiece) {
 	mqp = mqpp.Dequeue()
@@ -35,7 +35,6 @@ func NewPooledMysqlQueryPiece(
 	mqp.SessionID = sessionID
 	mqp.ClientHost = clientIP
 	mqp.ClientPort = clientPort
-	mqp.ClientHost = clientHost
 	mqp.ServerIP = serverIP
 	mqp.ServerPort = serverPort
 	mqp.VisitUser = visitUser

session-dealer/controller.go

@@ -5,13 +5,13 @@ import (
 	"github.com/zr-hebo/sniffer-agent/session-dealer/mysql"
 )
 
-func NewSession(sessionKey, clientAlias, clientIP *string, clientPort int, serverIP *string, serverPort int,
+func NewSession(sessionKey, clientIP *string, clientPort int, srcIP *string, srcPort int, serverIP *string, serverPort int,
 	receiver chan model.QueryPiece) (session ConnSession) {
 	switch serviceType {
 	case ServiceTypeMysql:
-		session = mysql.NewMysqlSession(sessionKey, clientAlias, clientIP, clientPort, serverIP, serverPort, receiver)
+		session = mysql.NewMysqlSession(sessionKey, clientIP, clientPort, srcIP, srcPort, serverIP, serverPort, receiver)
 	default:
-		session = mysql.NewMysqlSession(sessionKey, clientAlias, clientIP, clientPort, serverIP, serverPort, receiver)
+		session = mysql.NewMysqlSession(sessionKey, clientIP, clientPort, srcIP, srcPort, serverIP, serverPort, receiver)
 	}
 	return
 }

session-dealer/mysql/session.go

@@ -15,9 +15,10 @@ type MysqlSession struct {
 	connectionID  *string
 	visitUser     *string
 	visitDB       *string
-	clientAlias   *string
-	clientHost    *string
+	clientIP      *string
 	clientPort    int
+	srcIP         *string
+	srcPort       int
 	serverIP      *string
 	serverPort    int
 	stmtBeginTime int64
@@ -47,13 +48,14 @@ type prepareInfo struct {
 }
 
 func NewMysqlSession(
-	sessionKey, clientAlias, clientIP *string, clientPort int, serverIP *string, serverPort int,
+	sessionKey, clientIP *string, clientPort int, srcIP *string, srcPort int, serverIP *string, serverPort int,
 	receiver chan model.QueryPiece) (ms *MysqlSession) {
 	ms = &MysqlSession{
 		connectionID:       sessionKey,
-		clientAlias:        clientAlias,
-		clientHost:         clientIP,
+		clientIP:           clientIP,
 		clientPort:         clientPort,
+		srcIP:              srcIP,
+		srcPort:            srcPort,
 		serverIP:           serverIP,
 		serverPort:         serverPort,
 		stmtBeginTime:      time.Now().UnixNano() / millSecondUnit,
@@ -335,11 +337,13 @@ func filterQueryPieceBySQL(mqp *model.PooledMysqlQueryPiece, querySQL []byte) (*
 }
 
 func (ms *MysqlSession) composeQueryPiece() (mqp *model.PooledMysqlQueryPiece) {
-	clientIP := ms.clientAlias
+	clientIP := ms.clientIP
+	clientPort := ms.clientPort
 	if clientIP == nil || len(*clientIP) < 1 {
-		clientIP = ms.clientHost
+		clientIP = ms.srcIP
+		clientPort = ms.serverPort
 	}
 	return model.NewPooledMysqlQueryPiece(
-		ms.connectionID, ms.clientHost, ms.visitUser, ms.visitDB, ms.clientHost, ms.serverIP,
-		ms.clientPort, ms.serverPort, communicator.GetMysqlCapturePacketRate(), ms.stmtBeginTime)
+		ms.connectionID, clientIP, ms.visitUser, ms.visitDB, ms.serverIP,
+		clientPort, ms.serverPort, communicator.GetMysqlCapturePacketRate(), ms.stmtBeginTime)
 }