代码拉取完成,页面将自动刷新
package mysql
import (
"fmt"
"packetbeat/common"
"packetbeat/config"
"packetbeat/logp"
"packetbeat/procs"
"packetbeat/protos"
"packetbeat/protos/tcp"
"strings"
"time"
)
// Packet types
const (
MYSQL_CMD_QUERY = 3
)
const MAX_PAYLOAD_SIZE = 100 * 1024
type MysqlMessage struct {
start int
end int
Ts time.Time
IsRequest bool
PacketLength uint32
Seq uint8
Typ uint8
NumberOfRows int
NumberOfFields int
Size uint64
Fields []string
Rows [][]string
Tables string
IsOK bool
AffectedRows int
InsertId int
IsError bool
ErrorCode int
ErrorInfo string
Query string
IgnoreMessage bool
Direction uint8
IsTruncated bool
TcpTuple common.TcpTuple
CmdlineTuple *common.CmdlineTuple
Raw []byte
}
type MysqlTransaction struct {
Type string
tuple common.TcpTuple
Src common.Endpoint
Dst common.Endpoint
ResponseTime int32
Ts int64
JsTs time.Time
ts time.Time
Query string
Method string
Path string // for mysql, Path refers to the mysql table queried
Size uint64
Mysql common.MapStr
Request_raw string
Response_raw string
timer *time.Timer
}
type MysqlStream struct {
tcptuple *common.TcpTuple
data []byte
parseOffset int
parseState int
isClient bool
message *MysqlMessage
}
const (
TransactionsHashSize = 2 ^ 16
TransactionTimeout = 10 * 1e9
)
const (
MysqlStateStart = iota
MysqlStateEatMessage
MysqlStateEatFields
MysqlStateEatRows
)
type Mysql struct {
transactionsMap map[common.HashableTcpTuple]*MysqlTransaction
maxStoreRows int
maxRowLength int
Send_request bool
Send_response bool
results chan common.MapStr
// function pointer for mocking
handleMysql func(mysql *Mysql, m *MysqlMessage, tcp *common.TcpTuple,
dir uint8, raw_msg []byte)
}
func (mysql *Mysql) InitDefaults() {
mysql.maxRowLength = 1024
mysql.maxStoreRows = 10
mysql.Send_request = false
mysql.Send_response = false
}
func (mysql *Mysql) setFromConfig() error {
if config.ConfigSingleton.Mysql.Max_row_length > 0 {
mysql.maxRowLength = config.ConfigSingleton.Mysql.Max_row_length
}
if config.ConfigSingleton.Mysql.Max_rows > 0 {
mysql.maxStoreRows = config.ConfigSingleton.Mysql.Max_rows
}
if config.ConfigMeta.IsDefined("protocols", "mysql", "send_request") {
mysql.Send_request = config.ConfigSingleton.Protocols["mysql"].Send_request
}
if config.ConfigMeta.IsDefined("protocols", "mysql", "send_response") {
mysql.Send_response = config.ConfigSingleton.Protocols["mysql"].Send_response
}
return nil
}
func (mysql *Mysql) Init(test_mode bool, results chan common.MapStr) error {
mysql.InitDefaults()
if !test_mode {
err := mysql.setFromConfig()
if err != nil {
return err
}
}
mysql.transactionsMap = make(map[common.HashableTcpTuple]*MysqlTransaction, TransactionsHashSize)
mysql.handleMysql = handleMysql
mysql.results = results
return nil
}
func (stream *MysqlStream) PrepareForNewMessage() {
stream.data = stream.data[stream.message.end:]
stream.parseState = MysqlStateStart
stream.parseOffset = 0
stream.isClient = false
stream.message = nil
}
func mysqlMessageParser(s *MysqlStream) (bool, bool) {
logp.Debug("mysqldetailed", "MySQL parser called. parseState = %d", s.parseState)
m := s.message
for s.parseOffset < len(s.data) {
switch s.parseState {
case MysqlStateStart:
m.start = s.parseOffset
if len(s.data[s.parseOffset:]) < 5 {
logp.Warn("MySQL Message too short. Ignore it.")
return false, false
}
hdr := s.data[s.parseOffset : s.parseOffset+5]
m.PacketLength = uint32(hdr[0]) | uint32(hdr[1])<<8 | uint32(hdr[2])<<16
m.Seq = uint8(hdr[3])
m.Typ = uint8(hdr[4])
logp.Debug("mysqldetailed", "MySQL Header: Packet length %d, Seq %d, Type=%d", m.PacketLength, m.Seq, m.Typ)
if m.Seq == 0 {
// starts Command Phase
if m.Typ == MYSQL_CMD_QUERY {
// 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
if uint8(hdr[4]) == 0x00 {
logp.Debug("mysqldetailed", "Received OK response")
m.start = s.parseOffset
s.parseState = MysqlStateEatMessage
m.IsOK = true
} else if uint8(hdr[4]) == 0xff {
logp.Debug("mysqldetailed", "Received ERR response")
m.start = s.parseOffset
s.parseState = MysqlStateEatMessage
m.IsError = true
} else if m.PacketLength == 1 {
logp.Debug("mysqldetailed", "Query response. Number of fields %d", uint8(hdr[4]))
m.NumberOfFields = int(hdr[4])
m.start = s.parseOffset
s.parseOffset += 5
s.parseState = MysqlStateEatFields
} else {
// something else. ignore
m.IgnoreMessage = true
s.parseState = MysqlStateEatMessage
}
} else {
// something else, not expected
logp.Warn("Unexpected MySQL message of type %d received.", m.Typ)
return false, false
}
break
case MysqlStateEatMessage:
if len(s.data[s.parseOffset:]) >= int(m.PacketLength)+4 {
s.parseOffset += 4 //header
s.parseOffset += int(m.PacketLength)
m.end = s.parseOffset
if m.IsRequest {
m.Query = string(s.data[m.start+5 : m.end])
} else if m.IsOK {
m.AffectedRows = int(s.data[m.start+5])
m.InsertId = int(s.data[m.start+6])
} else if m.IsError {
m.ErrorCode = int(uint16(s.data[m.start+6])<<8 | uint16(s.data[m.start+7]))
m.ErrorInfo = string(s.data[m.start+9:m.start+14]) + ": " + string(s.data[m.start+15:])
}
logp.Debug("mysqldetailed", "Message complete. remaining=%d", len(s.data[s.parseOffset:]))
return true, true
} else {
// wait for more
return true, false
}
case MysqlStateEatFields:
if len(s.data[s.parseOffset:]) < 4 {
// wait for more
return true, false
}
hdr := s.data[s.parseOffset : s.parseOffset+4]
m.PacketLength = uint32(hdr[0]) | uint32(hdr[1])<<8 | uint32(hdr[2])<<16
m.Seq = uint8(hdr[3])
logp.Debug("mysqldetailed", "Fields: packet length %d, packet number %d", m.PacketLength, m.Seq)
if len(s.data[s.parseOffset:]) >= int(m.PacketLength)+4 {
s.parseOffset += 4 // header
if uint8(s.data[s.parseOffset]) == 0xfe {
logp.Debug("mysqldetailed", "Received EOF packet")
// EOF marker
s.parseOffset += int(m.PacketLength)
s.parseState = MysqlStateEatRows
} else {
_ /* catalog */, off, complete, err := read_lstring(s.data, s.parseOffset)
if !complete {
return true, false
}
if err != nil {
logp.Debug("mysql", "Error on read_lstring: %s", err)
return false, false
}
db /*schema */, off, complete, err := read_lstring(s.data, off)
if !complete {
return true, false
}
if err != nil {
logp.Debug("mysql", "Error on read_lstring: %s", err)
return false, false
}
table /* table */, off, complete, err := read_lstring(s.data, off)
if !complete {
return true, false
}
if err != nil {
logp.Debug("mysql", "Error on read_lstring: %s", err)
return false, false
}
db_table := string(db) + "." + string(table)
if len(m.Tables) == 0 {
m.Tables = db_table
} else if !strings.Contains(m.Tables, db_table) {
m.Tables = m.Tables + ", " + db_table
}
logp.Debug("mysqldetailed", "db=%s, table=%s", db, table)
s.parseOffset += int(m.PacketLength)
// go to next field
}
} else {
// wait for more
return true, false
}
break
case MysqlStateEatRows:
if len(s.data[s.parseOffset:]) < 4 {
// wait for more
return true, false
}
hdr := s.data[s.parseOffset : s.parseOffset+4]
m.PacketLength = uint32(hdr[0]) | uint32(hdr[1])<<8 | uint32(hdr[2])<<16
m.Seq = uint8(hdr[3])
logp.Debug("mysqldetailed", "Rows: packet length %d, packet number %d", m.PacketLength, m.Seq)
if len(s.data[s.parseOffset:]) >= int(m.PacketLength)+4 {
s.parseOffset += 4 //header
if uint8(s.data[s.parseOffset]) == 0xfe {
logp.Debug("mysqldetailed", "Received EOF packet")
// EOF marker
s.parseOffset += int(m.PacketLength)
if m.end == 0 {
m.end = s.parseOffset
} else {
m.IsTruncated = true
}
m.Size = uint64(s.parseOffset - m.start)
if !m.IsError {
// in case the response was sent successfully
m.IsOK = true
}
return true, true
} else {
s.parseOffset += int(m.PacketLength)
if m.end == 0 && s.parseOffset > MAX_PAYLOAD_SIZE {
// only send up to here, but read until the end
m.end = s.parseOffset
}
m.NumberOfRows += 1
// go to next row
}
} else {
// wait for more
return true, false
}
break
}
}
return true, false
}
type mysqlPrivateData struct {
Data [2]*MysqlStream
}
func (mysql *Mysql) Parse(pkt *protos.Packet, tcptuple *common.TcpTuple,
dir uint8, private protos.ProtocolData) protos.ProtocolData {
defer logp.Recover("ParseMysql exception")
priv := mysqlPrivateData{}
if private != nil {
var ok bool
priv, ok = private.(mysqlPrivateData)
if !ok {
priv = mysqlPrivateData{}
}
}
if priv.Data[dir] == nil {
priv.Data[dir] = &MysqlStream{
tcptuple: tcptuple,
data: pkt.Payload,
message: &MysqlMessage{Ts: pkt.Ts},
}
} else {
// concatenate bytes
priv.Data[dir].data = append(priv.Data[dir].data, pkt.Payload...)
if len(priv.Data[dir].data) > tcp.TCP_MAX_DATA_IN_STREAM {
logp.Debug("mysql", "Stream data too large, dropping TCP stream")
priv.Data[dir] = nil
return priv
}
}
stream := priv.Data[dir]
for len(stream.data) > 0 {
if stream.message == nil {
stream.message = &MysqlMessage{Ts: pkt.Ts}
}
ok, complete := mysqlMessageParser(priv.Data[dir])
if !ok {
// drop this tcp stream. Will retry parsing with the next
// segment in it
priv.Data[dir] = nil
logp.Debug("mysql", "Ignore MySQL message. Drop tcp stream. Try parsing with the next segment")
return priv
}
if complete {
// all ok, ship it
msg := stream.data[stream.message.start:stream.message.end]
if !stream.message.IgnoreMessage {
mysql.handleMysql(mysql, stream.message, tcptuple, dir, msg)
}
// and reset message
stream.PrepareForNewMessage()
} else {
// wait for more data
break
}
}
return priv
}
func (mysql *Mysql) GapInStream(tcptuple *common.TcpTuple, dir uint8,
private protos.ProtocolData) protos.ProtocolData {
// TODO
return private
}
func (mysql *Mysql) ReceivedFin(tcptuple *common.TcpTuple, dir uint8,
private protos.ProtocolData) protos.ProtocolData {
// TODO
return private
}
func handleMysql(mysql *Mysql, m *MysqlMessage, tcptuple *common.TcpTuple,
dir uint8, raw_msg []byte) {
m.TcpTuple = *tcptuple
m.Direction = dir
m.CmdlineTuple = procs.ProcWatcher.FindProcessesTuple(tcptuple.IpPort())
m.Raw = raw_msg
if m.IsRequest {
mysql.receivedMysqlRequest(m)
} else {
mysql.receivedMysqlResponse(m)
}
}
func (mysql *Mysql) receivedMysqlRequest(msg *MysqlMessage) {
// Add it to the HT
tuple := msg.TcpTuple
trans := mysql.transactionsMap[tuple.Hashable()]
if trans != nil {
if trans.Mysql != nil {
logp.Debug("mysql", "Two requests without a Response. Dropping old request: %s", trans.Mysql)
}
} else {
trans = &MysqlTransaction{Type: "mysql", tuple: tuple}
mysql.transactionsMap[tuple.Hashable()] = trans
}
trans.ts = msg.Ts
trans.Ts = int64(trans.ts.UnixNano() / 1000) // transactions have microseconds resolution
trans.JsTs = msg.Ts
trans.Src = common.Endpoint{
Ip: msg.TcpTuple.Src_ip.String(),
Port: msg.TcpTuple.Src_port,
Proc: string(msg.CmdlineTuple.Src),
}
trans.Dst = common.Endpoint{
Ip: msg.TcpTuple.Dst_ip.String(),
Port: msg.TcpTuple.Dst_port,
Proc: string(msg.CmdlineTuple.Dst),
}
if msg.Direction == tcp.TcpDirectionReverse {
trans.Src, trans.Dst = trans.Dst, trans.Src
}
// Extract the method, by simply taking the first word and
// making it upper case.
query := strings.Trim(msg.Query, " \n\t")
index := strings.IndexAny(query, " \n\t")
var method string
if index > 0 {
method = strings.ToUpper(query[:index])
} else {
method = strings.ToUpper(query)
}
trans.Query = query
trans.Method = method
trans.Mysql = common.MapStr{}
// save Raw message
trans.Request_raw = msg.Query
if trans.timer != nil {
trans.timer.Stop()
}
trans.timer = time.AfterFunc(TransactionTimeout, func() { mysql.expireTransaction(trans) })
}
func (mysql *Mysql) receivedMysqlResponse(msg *MysqlMessage) {
tuple := msg.TcpTuple
trans := mysql.transactionsMap[tuple.Hashable()]
if trans == nil {
logp.Warn("Response from unknown transaction. Ignoring.")
return
}
// check if the request was received
if trans.Mysql == nil {
logp.Warn("Response from unknown transaction. Ignoring.")
return
}
// save json details
trans.Mysql.Update(common.MapStr{
"affected_rows": msg.AffectedRows,
"insert_id": msg.InsertId,
"num_rows": msg.NumberOfRows,
"num_fields": msg.NumberOfFields,
"iserror": msg.IsError,
"error_code": msg.ErrorCode,
"error_message": msg.ErrorInfo,
})
trans.Size = msg.Size
trans.Path = msg.Tables
trans.ResponseTime = int32(msg.Ts.Sub(trans.ts).Nanoseconds() / 1e6) // resp_time in milliseconds
// save Raw message
if len(msg.Raw) > 0 {
fields, rows := mysql.parseMysqlResponse(msg.Raw)
trans.Response_raw = common.DumpInCSVFormat(fields, rows)
}
mysql.publishMysqlTransaction(trans)
logp.Debug("mysql", "Mysql transaction completed: %s", trans.Mysql)
logp.Debug("mysql", "%s", trans.Response_raw)
// remove from map
delete(mysql.transactionsMap, trans.tuple.Hashable())
if trans.timer != nil {
trans.timer.Stop()
}
}
func (mysql *Mysql) expireTransaction(trans *MysqlTransaction) {
// TODO: Here we need to PUBLISH an incomplete/timeout transaction
// remove from map
delete(mysql.transactionsMap, trans.tuple.Hashable())
}
func (mysql *Mysql) parseMysqlResponse(data []byte) ([]string, [][]string) {
length := read_length(data, 0)
if length < 1 {
logp.Warn("Warning: Skipping empty Response")
return []string{}, [][]string{}
}
fields := []string{}
rows := [][]string{}
if uint8(data[4]) == 0x00 {
// OK response
} else if uint8(data[4]) == 0xff {
// Error response
} else {
offset := 5
// Read fields
for {
length = read_length(data, offset)
if uint8(data[offset+4]) == 0xfe {
// EOF
offset += length + 4
break
}
_ /* catalog */, off, complete, err := read_lstring(data, offset+4)
if err != nil || !complete {
logp.Debug("mysql", "Reading field: %s %b", err, complete)
return fields, rows
}
_ /*database*/, off, complete, err = read_lstring(data, off)
if err != nil || !complete {
logp.Debug("mysql", "Reading field: %s %b", err, complete)
return fields, rows
}
_ /*table*/, off, complete, err = read_lstring(data, off)
if err != nil || !complete {
logp.Debug("mysql", "Reading field: %s %b", err, complete)
return fields, rows
}
_ /*org table*/, off, complete, err = read_lstring(data, off)
if err != nil || !complete {
logp.Debug("mysql", "Reading field: %s %b", err, complete)
return fields, rows
}
name, off, complete, err := read_lstring(data, off)
if err != nil || !complete {
logp.Debug("mysql", "Reading field: %s %b", err, complete)
return fields, rows
}
_ /* org name */, off, complete, err = read_lstring(data, off)
if err != nil || !complete {
logp.Debug("mysql", "Reading field: %s %b", err, complete)
return fields, rows
}
fields = append(fields, string(name))
offset += length + 4
}
// Read rows
for offset < len(data) {
var row []string
var row_len int
if uint8(data[offset+4]) == 0xfe {
// EOF
offset += length + 4
break
}
length = read_length(data, offset)
off := offset + 4 // skip length + packet number
start := off
for off < start+length {
var text []byte
if uint8(data[off]) == 0xfb {
text = []byte("NULL")
off++
} else {
var err error
var complete bool
text, off, complete, err = read_lstring(data, off)
if err != nil {
logp.Debug("mysql", "Error parsing rows: %s %b", err, complete)
// nevertheless, return what we have so far
return fields, rows
}
}
if row_len < mysql.maxRowLength {
if row_len+len(text) > mysql.maxRowLength {
text = text[:mysql.maxRowLength-row_len]
}
row = append(row, string(text))
row_len += len(text)
}
}
rows = append(rows, row)
if len(rows) >= mysql.maxStoreRows {
break
}
offset += length + 4
}
}
return fields, rows
}
func (mysql *Mysql) publishMysqlTransaction(t *MysqlTransaction) {
if mysql.results == nil {
return
}
logp.Debug("mysql", "mysql.results exists")
event := common.MapStr{}
event["type"] = "mysql"
if t.Mysql["iserror"].(bool) {
event["status"] = common.ERROR_STATUS
} else {
event["status"] = common.OK_STATUS
}
event["responsetime"] = t.ResponseTime
if mysql.Send_request {
event["request_raw"] = t.Request_raw
}
if mysql.Send_response {
event["response_raw"] = t.Response_raw
}
event["method"] = t.Method
event["query"] = t.Query
event["mysql"] = t.Mysql
event["path"] = t.Path
event["bytes_out"] = t.Size
event["@timestamp"] = common.Time(t.ts)
event["src"] = &t.Src
event["dst"] = &t.Dst
mysql.results <- event
}
func read_lstring(data []byte, offset int) ([]byte, int, bool, error) {
length, off, complete, err := read_linteger(data, offset)
if err != nil {
return nil, 0, false, err
}
if !complete || len(data[off:]) < int(length) {
return nil, 0, false, nil
}
return data[off : off+int(length)], off + int(length), true, nil
}
func read_linteger(data []byte, offset int) (uint64, int, bool, error) {
if len(data) == 0 {
return 0, 0, false, nil
}
switch uint8(data[offset]) {
case 0xfe:
if len(data[offset+1:]) < 8 {
return 0, 0, false, nil
}
return uint64(data[offset+1]) | uint64(data[offset+2])<<8 |
uint64(data[offset+2])<<16 | uint64(data[offset+3])<<24 |
uint64(data[offset+4])<<32 | uint64(data[offset+5])<<40 |
uint64(data[offset+6])<<48 | uint64(data[offset+7])<<56,
offset + 9, true, nil
case 0xfd:
if len(data[offset+1:]) < 3 {
return 0, 0, false, nil
}
return uint64(data[offset+1]) | uint64(data[offset+2])<<8 |
uint64(data[offset+3])<<16, offset + 4, true, nil
case 0xfc:
if len(data[offset+1:]) < 2 {
return 0, 0, false, nil
}
return uint64(data[offset+1]) | uint64(data[offset+2])<<8, offset + 3, true, nil
}
if uint64(data[offset]) >= 0xfb {
return 0, 0, false, fmt.Errorf("Unexpected value in read_linteger")
}
return uint64(data[offset]), offset + 1, true, nil
}
func read_length(data []byte, offset int) int {
length := uint32(data[offset]) |
uint32(data[offset+1])<<8 |
uint32(data[offset+2])<<16
return int(length)
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。