From df2b4be89823031a517846478dadd859bf7acd8b Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Fri, 16 Jul 2021 22:18:28 +0300 Subject: [PATCH] Fix detection of packet detection for complex bpf fiters Right now it depends on the defined ports, but you can have more compex bpf, so wee may stil need ti check it with Start function --- capture/capture.go | 2 +- input_raw.go | 2 +- tcp/tcp_message.go | 35 ++++++++++++++++++++++++++--------- tcp/tcp_packet.go | 10 +++++++++- tcp/tcp_test.go | 10 +++++----- 5 files changed, 42 insertions(+), 17 deletions(-) diff --git a/capture/capture.go b/capture/capture.go index b9a111a..dcdc30e 100644 --- a/capture/capture.go +++ b/capture/capture.go @@ -357,7 +357,7 @@ func (l *Listener) read(handler PacketHandler) { if pckt.DstPort == p { for _, ip := range hndl.ips { if pckt.DstIP.Equal(ip) { - pckt.Incoming = true + pckt.Direction = tcp.DirIncoming break } } diff --git a/input_raw.go b/input_raw.go index cec9a99..7c9a43a 100644 --- a/input_raw.go +++ b/input_raw.go @@ -122,7 +122,7 @@ func (i *RAWInput) PluginRead() (*Message, error) { } var msgType byte = ResponsePayload - if msgTCP.IsRequest { + if msgTCP.Direction == tcp.DirIncoming { msgType = RequestPayload if i.RealIPHeader != "" { msg.Data = proto.SetHeader(msg.Data, []byte(i.RealIPHeader), []byte(msgTCP.SrcAddr)) diff --git a/tcp/tcp_message.go b/tcp/tcp_message.go index 942bcfa..54cf44d 100644 --- a/tcp/tcp_message.go +++ b/tcp/tcp_message.go @@ -20,7 +20,7 @@ type Stats struct { End time.Time // last packet's timestamp SrcAddr string DstAddr string - IsRequest bool + Direction Dir TimedOut bool // timeout before getting the whole message Truncated bool // last packet truncated due to max message size IPversion byte @@ -40,7 +40,7 @@ func (m *Message) UUID() []byte { pckt := m.packets[0] // check if response or request have generated the ID before. - if m.IsRequest { + if m.Direction == DirIncoming { streamID = uint64(pckt.SrcPort)<<48 | uint64(pckt.DstPort)<<32 | uint64(ip2int(pckt.SrcIP)) } else { @@ -51,7 +51,7 @@ func (m *Message) UUID() []byte { id := make([]byte, 12) binary.BigEndian.PutUint64(id, streamID) - if m.IsRequest { + if m.Direction == DirIncoming { binary.BigEndian.PutUint32(id[8:], pckt.Ack) } else { binary.BigEndian.PutUint32(id[8:], pckt.Seq) @@ -152,7 +152,6 @@ func (m *Message) Sort() { } func (m *Message) Finalize() { - } // Emitter message handler @@ -242,8 +241,6 @@ func (parser *MessageParser) wait() { } func (parser *MessageParser) processPacket(pckt *Packet) { - var in bool - // Trying to build unique hash, but there is small chance of collision // No matter if it is request or response, all packets in the same message have same m, ok := parser.m[pckt.MessageID()] @@ -251,12 +248,32 @@ func (parser *MessageParser) processPacket(pckt *Packet) { case ok: parser.addPacket(m, pckt) return - default: - in = pckt.Incoming + case pckt.Direction == DirUnknown && parser.Start != nil: + if in, out := parser.Start(pckt); !(in || out) { + // Packet can be received out of order, so give it another chance + if pckt.Retry < 2 && len(pckt.Payload) > 0 { + // Requeue not known packets + pckt.Retry++ + + select { + case parser.packets <- pckt: + return + default: + } + } + + return + } else { + if in { + pckt.Direction = DirIncoming + } else { + pckt.Direction = DirOutcoming + } + } } m = new(Message) - m.IsRequest = in + m.Direction = pckt.Direction parser.m[pckt.MessageID()] = m m.Start = pckt.Timestamp m.parser = parser diff --git a/tcp/tcp_packet.go b/tcp/tcp_packet.go index 1cb1008..b9d35df 100644 --- a/tcp/tcp_packet.go +++ b/tcp/tcp_packet.go @@ -43,6 +43,14 @@ func init() { stats.Set("buffer_released", releasedCount) } +type Dir int + +const ( + DirUnknown = iota + DirIncoming + DirOutcoming +) + /* Packet represent data and layers of packet. parser extracts information from pcap Packet. functions of *Packet doesn't validate if packet is nil, @@ -50,7 +58,7 @@ calllers must make sure that ParsePacket has'nt returned any error before callin function. */ type Packet struct { - Incoming bool + Direction Dir messageID uint64 SrcIP, DstIP net.IP Version uint8 diff --git a/tcp/tcp_test.go b/tcp/tcp_test.go index d05a631..1af644c 100644 --- a/tcp/tcp_test.go +++ b/tcp/tcp_test.go @@ -49,7 +49,7 @@ func GetPackets(request bool, start uint32, _len int, payload []byte) []*Packet ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()} packets[i-start], err = ParsePacket(d, int(layers.LinkTypeLoop), 4, ci, true) - packets[i-start].Incoming = request + packets[i-start].Direction = DirIncoming if err != nil { panic(err) } @@ -93,10 +93,10 @@ func TestRequestResponseMapping(t *testing.T) { messages = append(messages, m) } - assert.Equal(t, messages[0].IsRequest, true) - assert.Equal(t, messages[1].IsRequest, false) - assert.Equal(t, messages[2].IsRequest, true) - assert.Equal(t, messages[3].IsRequest, false) + assert.Equal(t, messages[0].Direction, DirIncoming) + assert.Equal(t, messages[1].Direction, DirOutcoming) + assert.Equal(t, messages[2].Direction, DirIncoming) + assert.Equal(t, messages[3].Direction, DirOutcoming) assert.Equal(t, messages[0].UUID(), messages[1].UUID()) assert.Equal(t, messages[2].UUID(), messages[3].UUID())