From a526f3a5f1d6ffe4643c76e7722901146a23fdf3 Mon Sep 17 00:00:00 2001 From: Urban Ishimwe Date: Mon, 21 Dec 2020 18:41:21 +0200 Subject: [PATCH] lazy packet parser (#870) benchmarks of packet parser with `-cpu=1` packet (IPv6 with 2 extension header) master: ``` 337463 3300 ns/op 1624 B/op 24 allocs/op ``` current: ``` 2014885 576 ns/op 384 B/op 3 allocs/op ``` --- capture/capture.go | 194 ++++++++++++++++++--------------------- capture/capture_test.go | 79 ++++------------ capture/packet.go | 180 ++++++++++++++++++++++++++++++++++++ capture/packet_test.go | 197 ++++++++++++++++++++++++++++++++++++++++ input_raw.go | 21 +++-- tcp/tcp_message.go | 9 +- tcp/tcp_packet.go | 193 ++++++++------------------------------- tcp/tcp_test.go | 100 +++++++++++--------- 8 files changed, 592 insertions(+), 381 deletions(-) create mode 100644 capture/packet.go create mode 100644 capture/packet_test.go diff --git a/capture/capture.go b/capture/capture.go index d8c67e4..d77d1f7 100644 --- a/capture/capture.go +++ b/capture/capture.go @@ -4,6 +4,8 @@ import ( "context" "errors" "fmt" + "io" + "log" "net" "os" "runtime" @@ -14,10 +16,11 @@ import ( "github.com/google/gopacket" "github.com/google/gopacket/layers" "github.com/google/gopacket/pcap" + "golang.org/x/sys/unix" ) // Handler is a function that is used to handle packets -type Handler func(gopacket.Packet) +type Handler func(*Packet) // PcapOptions options that can be set on a pcap capture handle, // these options take effect on inactive pcap handles @@ -31,19 +34,13 @@ type PcapOptions struct { Snaplen bool `json:"input-raw-override-snaplen"` } -// NetInterface represents network interface -type NetInterface struct { - net.Interface - IPs []string -} - // Listener handle traffic capture, this is its representation. type Listener struct { sync.Mutex Transport string // transport layer default to tcp Activate func() error // function is used to activate the engine. it must be called before reading packets Handles map[string]gopacket.PacketDataSource - Interfaces []NetInterface + Interfaces []net.Interface loopIndex int Reading chan bool // this channel is closed when the listener has started reading packets PcapOptions @@ -53,8 +50,8 @@ type Listener struct { host string // pcap file name or interface (name, hardware addr, index or ip address) - quit chan bool - packets chan gopacket.Packet + closeDone chan struct{} + quit chan struct{} } // EngineType ... @@ -110,9 +107,9 @@ func NewListener(host string, port uint16, transport string, engine EngineType, } l.Handles = make(map[string]gopacket.PacketDataSource) l.trackResponse = trackResponse - l.packets = make(chan gopacket.Packet, 1000) - l.quit = make(chan bool, 1) - l.Reading = make(chan bool, 1) + l.closeDone = make(chan struct{}) + l.quit = make(chan struct{}) + l.Reading = make(chan bool) switch engine { default: l.Engine = EnginePcap @@ -139,30 +136,19 @@ func (l *Listener) SetPcapOptions(opts PcapOptions) { } // Listen listens for packets from the handles, and call handler on every packet received -// until the context done signal is sent or EOF on handles. -// this function should be called after activating pcap handles +// until the context done signal is sent or there is unrecoverable error on all handles. +// this function must be called after activating pcap handles func (l *Listener) Listen(ctx context.Context, handler Handler) (err error) { - l.read() + l.read(handler) done := ctx.Done() - var p gopacket.Packet - var ok bool - for { - select { - case <-done: - l.quit <- true - close(l.quit) - err = ctx.Err() - done = nil - case p, ok = <-l.packets: - if !ok { - return - } - if p == nil { - continue - } - handler(p) - } + select { + case <-done: + close(l.quit) // signal close on all handles + <-l.closeDone // wait all handles to be closed + err = ctx.Err() + case <-l.closeDone: // all handles closed voluntarily } + return } // ListenBackground is like listen but can run concurrently and signal error through channel @@ -179,7 +165,7 @@ func (l *Listener) ListenBackground(ctx context.Context, handler Handler) chan e // Filter returns automatic filter applied by goreplay // to a pcap handle of a specific interface -func (l *Listener) Filter(ifi NetInterface) (filter string) { +func (l *Listener) Filter(ifi net.Interface) (filter string) { // https://www.tcpdump.org/manpages/pcap-filter.7.html port := fmt.Sprintf("portrange 0-%d", 1<<16-1) @@ -201,7 +187,7 @@ func (l *Listener) Filter(ifi NetInterface) (filter string) { // PcapDumpHandler returns a handler to write packet data in PCAP // format, See http://wiki.wireshark.org/Development/LibpcapFileFormathandler. // if link layer is invalid Ethernet is assumed -func PcapDumpHandler(file *os.File, link layers.LinkType, debugger func(int, ...interface{})) (handler func(packet gopacket.Packet), err error) { +func PcapDumpHandler(file *os.File, link layers.LinkType) (handler func(packet *Packet) error, err error) { if link.String() == "" { link = layers.LinkTypeEthernet } @@ -210,25 +196,20 @@ func PcapDumpHandler(file *os.File, link layers.LinkType, debugger func(int, ... if err != nil { return nil, err } - return func(packet gopacket.Packet) { - err = w.WritePacket(packet.Metadata().CaptureInfo, packet.Data()) - if err != nil && debugger != nil { - go debugger(3, err) - } + return func(packet *Packet) error { + return w.WritePacket(*packet.Info, packet.Data) }, nil } // PcapHandle returns new pcap Handle from dev on success. // this function should be called after setting all necessary options for this listener -func (l *Listener) PcapHandle(ifi NetInterface) (handle *pcap.Handle, err error) { +func (l *Listener) PcapHandle(ifi net.Interface) (handle *pcap.Handle, err error) { var inactive *pcap.InactiveHandle inactive, err = pcap.NewInactiveHandle(ifi.Name) - if inactive != nil && err != nil { - defer inactive.CleanUp() - } if err != nil { return nil, fmt.Errorf("inactive handle error: %q, interface: %q", err, ifi.Name) } + defer inactive.CleanUp() if l.TimestampType != "" { var ts pcap.TimestampSource ts, err = pcap.TimestampSourceFromString(l.TimestampType) @@ -290,8 +271,8 @@ func (l *Listener) PcapHandle(ifi NetInterface) (handle *pcap.Handle, err error) } // SocketHandle returns new unix ethernet handle associated with this listener settings -func (l *Listener) SocketHandle(ifi NetInterface) (handle Socket, err error) { - handle, err = NewSocket(ifi.Interface) +func (l *Listener) SocketHandle(ifi net.Interface) (handle Socket, err error) { + handle, err = NewSocket(ifi) if err != nil { return nil, fmt.Errorf("sock raw error: %q, interface: %q", err, ifi.Name) } @@ -313,35 +294,54 @@ func (l *Listener) SocketHandle(ifi NetInterface) (handle Socket, err error) { return } -func (l *Listener) read() { +func (l *Listener) read(handler Handler) { l.Lock() defer l.Unlock() for key, handle := range l.Handles { - var source *gopacket.PacketSource - linkType := layers.LinkTypeEthernet - if _, ok := handle.(*pcap.Handle); ok { - linkType = handle.(*pcap.Handle).LinkType() - } - source = gopacket.NewPacketSource(handle, linkType) - source.Lazy = true - source.NoCopy = true - ch := source.Packets() - go func(key string) { + go func(key string, hndl gopacket.PacketDataSource) { defer l.closeHandles(key) + linkSize := 14 + linkType := int(layers.LinkTypeEthernet) + if _, ok := hndl.(*pcap.Handle); ok { + linkType = int(hndl.(*pcap.Handle).LinkType()) + linkSize, ok = pcapLinkTypeLength(linkType) + if !ok { + if os.Getenv("GORDEBUG") != "0" { + log.Printf("can not identify link type of an interface '%s'\n", key) + } + return // can't find the linktype size + } + } for { select { case <-l.quit: return - case p, ok := <-ch: - if !ok { + default: + data, ci, err := hndl.ReadPacketData() + if err == nil { + handler(NewPacket(data, linkType, linkSize, &ci)) + continue + } + if enext, ok := err.(pcap.NextError); ok && enext == pcap.NextErrorTimeoutExpired { + continue + } + if eno, ok := err.(unix.Errno); ok && eno.Temporary() { + continue + } + if enet, ok := err.(*net.OpError); ok && (enet.Temporary() || enet.Timeout()) { + continue + } + if err == io.EOF || err == io.ErrClosedPipe { return } - l.packets <- p + if os.Getenv("GORDEBUG") != "0" { + log.Printf("stopped reading from %s interface with error %s\n", key, err) + } + return } } - }(key) + }(key, handle) } - l.Reading <- true close(l.Reading) } @@ -356,7 +356,7 @@ func (l *Listener) closeHandles(key string) { } delete(l.Handles, key) if len(l.Handles) == 0 { - close(l.packets) + close(l.closeDone) } } } @@ -394,7 +394,10 @@ func (l *Listener) activateRawSocket() error { } l.Handles[ifi.Name] = handle } - return e + if len(l.Handles) == 0 { + return fmt.Errorf("raw socket handles error:%s", msg) + } + return nil } func (l *Listener) activatePcapFile() (err error) { @@ -410,7 +413,7 @@ func (l *Listener) activatePcapFile() (err error) { } else { addr := l.host l.host = "" - l.BPFFilter = l.Filter(NetInterface{}) + l.BPFFilter = l.Filter(net.Interface{}) l.host = addr } if e = handle.SetBPFFilter(l.BPFFilter); e != nil { @@ -422,59 +425,36 @@ func (l *Listener) activatePcapFile() (err error) { } func (l *Listener) setInterfaces() (err error) { - var Ifis []NetInterface var ifis []net.Interface ifis, err = net.Interfaces() if err != nil { - return err + return } - - for i := 0; i < len(ifis); i++ { + for i := range ifis { if ifis[i].Flags&net.FlagLoopback != 0 { l.loopIndex = ifis[i].Index } if ifis[i].Flags&net.FlagUp == 0 { continue } - var addrs []net.Addr - addrs, err = ifis[i].Addrs() - if err != nil { - return err - } - if len(addrs) == 0 { - continue - } - ifi := NetInterface{} - ifi.Interface = ifis[i] - ifi.IPs = make([]string, len(addrs)) - for j, addr := range addrs { - ifi.IPs[j] = cutMask(addr) - } - Ifis = append(Ifis, ifi) - } - - if listenAll(l.host) { - l.Interfaces = Ifis - return - } - found := false - for _, ifi := range Ifis { - if isDevice(l.host, ifi) { - found = true - } - for _, ip := range ifi.IPs { - if ip == l.host { - found = true - break - } - } - if found { - l.Interfaces = []NetInterface{ifi} + if isDevice(l.host, ifis[i]) { + l.Interfaces = []net.Interface{ifis[i]} return } + addrs, e := ifis[i].Addrs() + if e != nil { + // don't give up on a failure from a single interface + continue + } + for _, addr := range addrs { + if cutMask(addr) == l.host { + l.Interfaces = []net.Interface{ifis[i]} + return + } + } } - err = fmt.Errorf("can not find interface with addr, name or index %s", l.host) - return err + l.Interfaces = ifis + return } func cutMask(addr net.Addr) string { @@ -487,7 +467,7 @@ func cutMask(addr net.Addr) string { return mask } -func isDevice(addr string, ifi NetInterface) bool { +func isDevice(addr string, ifi net.Interface) bool { return addr == ifi.Name || addr == fmt.Sprintf("%d", ifi.Index) || addr == ifi.HardwareAddr.String() } diff --git a/capture/capture_test.go b/capture/capture_test.go index 9ca8d22..a8532f0 100644 --- a/capture/capture_test.go +++ b/capture/capture_test.go @@ -10,7 +10,6 @@ import ( "testing" "time" - "github.com/google/gopacket" "github.com/google/gopacket/layers" "github.com/google/gopacket/pcap" ) @@ -61,69 +60,23 @@ func TestBPFFilter(t *testing.T) { } } -var decodeOpts = gopacket.DecodeOptions{Lazy: true, NoCopy: true} - -func generateHeaders(seq uint32, length uint16) (headers [44]byte) { - // set ethernet headers - binary.BigEndian.PutUint32(headers[0:4], uint32(layers.ProtocolFamilyIPv4)) - - // set ip header - ip := headers[4:] - copy(ip[0:2], []byte{4<<4 | 5, 0x28<<2 | 0x00}) - binary.BigEndian.PutUint16(ip[2:4], length+54) - ip[9] = uint8(layers.IPProtocolTCP) - copy(ip[12:16], []byte{127, 0, 0, 1}) - copy(ip[16:], []byte{127, 0, 0, 1}) - - // set tcp header - tcp := ip[20:] - binary.BigEndian.PutUint16(tcp[0:2], 45678) - binary.BigEndian.PutUint16(tcp[2:4], 8000) - tcp[12] = 5 << 4 - return -} - -func randomPackets(start uint32, _len int, length uint16) []gopacket.Packet { - var packets = make([]gopacket.Packet, _len) - for i := start; i < start+uint32(_len); i++ { - h := generateHeaders(i, length) - d := make([]byte, int(length)+len(h)) - copy(d, h[0:]) - packet := gopacket.NewPacket(d, layers.LinkTypeLoop, decodeOpts) - packets[i-start] = packet - inf := packets[i-start].Metadata() - _len := len(d) - inf.CaptureInfo = gopacket.CaptureInfo{CaptureLength: _len, Length: _len, Timestamp: time.Now()} - } - return packets -} - func TestPcapDump(t *testing.T) { f, err := ioutil.TempFile("", "pcap_file") if err != nil { t.Error(err) } - waiter := make(chan bool, 1) - h, _ := PcapDumpHandler(f, layers.LinkTypeLoop, func(level int, a ...interface{}) { - if level != 3 { - t.Errorf("expected debug level to be 3, got %d", level) - } - waiter <- true - }) - packets := randomPackets(1, 5, 5) + h, _ := PcapDumpHandler(f, layers.LinkTypeLoop) + packets := Packets(1, 5, 5, 4) for i := 0; i < len(packets); i++ { if i == 1 { - tcp := packets[i].Data()[4:][20:] // change dst port - binary.BigEndian.PutUint16(tcp[2:], 8001) + binary.BigEndian.PutUint16(packets[i].TransLayer[2:], 8001) } if i == 4 { - inf := packets[i].Metadata() - inf.CaptureLength = 40 + packets[i].Info.CaptureLength = 40 } h(packets[i]) } - <-waiter name := f.Name() f.Close() testPcapDumpEngine(name, t) @@ -140,9 +93,9 @@ func testPcapDumpEngine(f string, t *testing.T) { pckts := 0 ctx, cancel := context.WithCancel(context.Background()) defer cancel() - err = l.Listen(ctx, func(packet gopacket.Packet) { - if packet.Metadata().CaptureLength != 49 { - t.Errorf("expected packet length to be %d, got %d", 49, packet.Metadata().CaptureLength) + err = l.Listen(ctx, func(packet *Packet) { + if packet.Info.CaptureLength != 57 { + t.Errorf("expected packet length to be %d, got %d", 57, packet.Info.CaptureLength) } pckts++ }) @@ -208,8 +161,8 @@ func BenchmarkPcapDump(b *testing.B) { } now := time.Now() defer os.Remove(f.Name()) - h, _ := PcapDumpHandler(f, layers.LinkTypeLoop, nil) - packets := randomPackets(1, b.N, 5) + h, _ := PcapDumpHandler(f, layers.LinkTypeLoop) + packets := Packets(1, b.N, 5, 4) for i := 0; i < len(packets); i++ { h(packets[i]) } @@ -224,8 +177,8 @@ func BenchmarkPcapFile(b *testing.B) { return } defer os.Remove(f.Name()) - h, _ := PcapDumpHandler(f, layers.LinkTypeLoop, nil) - packets := randomPackets(1, b.N, 5) + h, _ := PcapDumpHandler(f, layers.LinkTypeLoop) + packets := Packets(1, b.N, 5, 4) for i := 0; i < len(packets); i++ { h(packets[i]) } @@ -247,9 +200,9 @@ func BenchmarkPcapFile(b *testing.B) { pckts := 0 ctx, cancel := context.WithCancel(context.Background()) defer cancel() - if err = l.Listen(ctx, func(packet gopacket.Packet) { - if packet.Metadata().CaptureLength != 49 { - b.Errorf("expected packet length to be %d, got %d", 49, packet.Metadata().CaptureLength) + if err = l.Listen(ctx, func(packet *Packet) { + if packet.Info.CaptureLength != 49 { + b.Errorf("expected packet length to be %d, got %d", 49, packet.Info.CaptureLength) } pckts++ }); err != nil { @@ -268,8 +221,8 @@ func init() { } func handler(n, counter *int32) Handler { - return func(p gopacket.Packet) { - nn := int32(len(p.Data())) + return func(p *Packet) { + nn := int32(len(p.Data)) atomic.AddInt32(n, nn) atomic.AddInt32(counter, 1) } diff --git a/capture/packet.go b/capture/packet.go new file mode 100644 index 0000000..6f197ef --- /dev/null +++ b/capture/packet.go @@ -0,0 +1,180 @@ +package capture + +import ( + "github.com/google/gopacket" + "github.com/google/gopacket/layers" +) + +// Packet properties of a pcaket +type Packet struct { + Data []byte + LinkLayer []byte + NetLayer []byte + NetOptsLen int // length of extension headers(IPV6) or options(IPV4) + TransLayer []byte + TransOptsLen int // length of tcp options + Payload []byte + + LinkType int + Info *gopacket.CaptureInfo + Err error +} + +// NewPacket parses packet and stop at the first error encountered +// pckt.Error will be nil if packet was parsed successfully +func NewPacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo) (p *Packet) { + p = new(Packet) + p.Info = cp + p.Data = data + if len(data) < lTypeLen { + p.Err = ErrHdrLength("Link") + return + } + p.LinkType = lType + p.LinkLayer = data[:lTypeLen] + if len(data) <= lTypeLen { + p.Err = ErrHdrMissing("IPv4 or IPv6") + return + } + data = data[lTypeLen:] + var proto byte + if data[0]>>4 == 4 { + // IPv4 header + if len(data) < 20 { + p.Err = ErrHdrLength("IPv4") + return + } + proto = data[9] + ihl := int(data[0]&0x0F) * 4 + if ihl < 20 { + p.Err = ErrHdrInvalid("IPv4's IHL") + return + } + if len(data) < ihl { + p.Err = ErrHdrLength("IPv4 opts") + p.NetLayer = data[:20] + return + } + p.NetOptsLen = ihl - 20 + p.NetLayer = data[:ihl] + } else if data[0]>>4 == 6 { + if len(data) < 40 { + p.Err = ErrHdrLength("IPv6") + return + } + proto = data[6] + totalLen := 40 + for ipv6ExtensionHdr(proto) { + hdr := len(data) - totalLen + if hdr < 8 { + p.Err = ErrHdrExpected("IPv6 opts") + break + } + extLen := 8 + if proto != 44 { + extLen = int(data[totalLen+1]+1) * 8 + } + if hdr < extLen { + p.Err = ErrHdrLength("IPv6 opts") + break + } + p.NetOptsLen += extLen + proto = data[totalLen] + totalLen += extLen + } + p.NetLayer = data[:totalLen] + if p.Err != nil { + return + } + } else { + p.Err = ErrHdrExpected("IPv4 or IPv6") + return + } + if proto != 6 { + p.Err = ErrHdrExpected("TCP") + return + } + if len(data) <= len(p.NetLayer) { + p.Err = ErrHdrMissing("TCP") + return + } + data = data[len(p.NetLayer):] + // TCP header + if len(data) < 20 { + p.Err = ErrHdrLength("TCP") + return + } + dOf := int(data[12]>>4) * 4 + if dOf < 20 { + p.Err = ErrHdrInvalid("TCP's data offset") + return + } + if len(data) < dOf { + p.Err = ErrHdrLength("TCP opts") + p.TransLayer = data[:20] + return + } + p.TransLayer = data[:dOf] + p.TransOptsLen = dOf - 20 + if len(data) > dOf { + p.Payload = data[dOf:] + } + return +} + +// ErrHdrLength returned on short header length +type ErrHdrLength string + +func (err ErrHdrLength) Error() string { + return "short " + string(err) + " length" +} + +// ErrHdrMissing returned on missing header(s) +type ErrHdrMissing string + +func (err ErrHdrMissing) Error() string { + return "missing " + string(err) + " header(s)" +} + +// ErrHdrExpected returned when header(s) are different from the one expected +type ErrHdrExpected string + +func (err ErrHdrExpected) Error() string { + return "expected " + string(err) + " header(s)" +} + +// ErrHdrInvalid returned when header(s) are different from the one expected +type ErrHdrInvalid string + +func (err ErrHdrInvalid) Error() string { + return "invalid " + string(err) + " value" +} + +func pcapLinkTypeLength(lType int) (int, bool) { + switch layers.LinkType(lType) { + case layers.LinkTypeEthernet: + return 14, true + case layers.LinkTypeNull, layers.LinkTypeLoop: + return 4, true + case layers.LinkTypeRaw, 12, 14: + return 0, true + case layers.LinkTypeIPv4, layers.LinkTypeIPv6: + // (TODO:) look out for IP encapsulation? + return 0, true + case layers.LinkTypeLinuxSLL: + return 16, true + case layers.LinkTypeFDDI: + return 13, true + case 226 /*DLT_IPNET*/ : + // https://www.tcpdump.org/linktypes/LINKTYPE_IPNET.html + return 24, true + default: + return 0, false + } +} + +// https://en.wikipedia.org/wiki/IPv6_packet#Extension_headers +func ipv6ExtensionHdr(b byte) bool { + // TODO: support all extension headers + return b == 0 || b == 43 || b == 44 +} diff --git a/capture/packet_test.go b/capture/packet_test.go new file mode 100644 index 0000000..7a05de8 --- /dev/null +++ b/capture/packet_test.go @@ -0,0 +1,197 @@ +package capture + +import ( + "encoding/binary" + "errors" + "testing" + "time" + + "github.com/google/gopacket" + "github.com/google/gopacket/layers" +) + +func generateHeader4(seq uint32, length uint16) []byte { + hdr := make([]byte, 4+24+24, 4+24+24) + binary.BigEndian.PutUint32(hdr, uint32(layers.ProtocolFamilyIPv4)) + + ip := hdr[4:] + ip[0] = 4<<4 | 6 + binary.BigEndian.PutUint16(ip[2:4], length+24+24) + ip[9] = uint8(layers.IPProtocolTCP) + copy(ip[12:16], []byte{127, 0, 0, 1}) + copy(ip[16:], []byte{127, 0, 0, 1}) + + // set tcp header + tcp := ip[24:] + tcp[12] = 6 << 4 + binary.BigEndian.PutUint16(tcp, 5535) + binary.BigEndian.PutUint16(tcp[2:], 8000) + binary.BigEndian.PutUint32(tcp[4:], seq) + return hdr +} + +func generateHeader6(seq uint32, length uint16) []byte { + hdr := make([]byte, 4+40+32+24, 4+40+32+24) + binary.BigEndian.PutUint32(hdr, uint32(layers.ProtocolFamilyIPv6Linux)) + + ip := hdr[4:] + ip[0] = 6 << 4 + binary.BigEndian.PutUint16(ip[4:], length+32+24) + var ipAddr [16]byte + ipAddr[15] = 0x01 + copy(ip[8:], ip[:]) + copy(ip[24:], ip[:]) + copy(ip[40:], []byte{ + // net-layer (IPv6-Opts) + 0x2b, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + uint8(layers.IPProtocolTCP), 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + }) + + // set tcp header + tcp := ip[40+32:] + tcp[12] = 6 << 4 + binary.BigEndian.PutUint16(tcp, 5535) + binary.BigEndian.PutUint16(tcp[2:], 8000) + binary.BigEndian.PutUint32(tcp[4:], seq) + return hdr +} + +func Packets(start uint32, _len int, length uint16, version byte) []*Packet { + var packets = make([]*Packet, _len) + for i := start; i < start+uint32(_len); i++ { + var h []byte + if version == 4 { + h = generateHeader4(i, length) + } else { + h = generateHeader6(i, length) + } + d := append(h, make([]byte, int(length))...) + ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()} + packets[i-start] = NewPacket(d, int(layers.LinkTypeLoop), 4, ci) + } + return packets +} + +func TestIPv4Packet(t *testing.T) { + pckt := packet(append(generateHeader4(1024, 10), make([]byte, 10)...)) + if pckt.Err != nil { + t.Error(pckt) + return + } + if err := packet(pckt.Data[:2]).Err; !errors.Is(err, ErrHdrLength("Link")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("Link"), err) + return + } + if err := packet(pckt.Data[:20]).Err; !errors.Is(err, ErrHdrLength("IPv4")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv4"), err) + return + } + if err := packet(pckt.Data[:27]).Err; !errors.Is(err, ErrHdrLength("IPv4 opts")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv4 opts"), err) + return + } + if err := packet(pckt.Data[:40]).Err; !errors.Is(err, ErrHdrLength("TCP")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err) + return + } + if err := packet(pckt.Data[:50]).Err; !errors.Is(err, ErrHdrLength("TCP opts")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err) + return + } + pckt.TransLayer[12] = 0x10 + if err := packet(pckt.Data[:50]).Err; !errors.Is(err, ErrHdrInvalid("TCP's data offset")) { + t.Errorf("should fail with %q, got %q", ErrHdrInvalid("TCP's data offset"), err) + return + } + pckt.TransLayer[12] = 0x60 + if err := packet(pckt.Data[:28]).Err; !errors.Is(err, ErrHdrMissing("TCP")) { + t.Errorf("should fail with %q, got %q", ErrHdrMissing("TCP"), err) + return + } + pckt.NetLayer[9] = 0x02 + if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("TCP")) { + t.Errorf("should fail with %q, got %q", ErrHdrExpected("TCP"), err) + return + } + pckt.NetLayer[9] = 0x06 + pckt.NetLayer[0] = 0x44 + if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrInvalid("IPv4's IHL")) { + t.Errorf("should fail with %q, got %q", ErrHdrInvalid("IPv4's IHL"), err) + return + } + pckt.NetLayer[0] = 0x56 + if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("IPv4 or IPv6")) { + t.Errorf("should fail with %q, got %q", ErrHdrExpected("IPv4 or IPv6"), err) + return + } +} + +func TestIPv6Packet(t *testing.T) { + pckt := packet(append(generateHeader6(1024, 10), make([]byte, 10)...)) + if pckt.Err != nil { + t.Error(pckt) + return + } + if err := packet(pckt.Data[:4]).Err; !errors.Is(err, ErrHdrMissing("IPv4 or IPv6")) { + t.Errorf("should fail with %q, got %q", ErrHdrMissing("IPv4 or IPv6"), err) + return + } + if err := packet(pckt.Data[:40]).Err; !errors.Is(err, ErrHdrLength("IPv6")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv6"), err) + return + } + if err := packet(pckt.Data[:52]).Err; !errors.Is(err, ErrHdrLength("IPv6 opts")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("IPv6 opts"), err) + return + } + if err := packet(pckt.Data[:80]).Err; !errors.Is(err, ErrHdrLength("TCP")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err) + return + } + if err := packet(pckt.Data[:98]).Err; !errors.Is(err, ErrHdrLength("TCP opts")) { + t.Errorf("should fail with %q, got %q", ErrHdrLength("TCP opts"), err) + return + } + pckt.TransLayer[12] = 0x10 + if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrInvalid("TCP's data offset")) { + t.Errorf("should fail with %q, got %q", ErrHdrInvalid("TCP's data offset"), err) + return + } + pckt.TransLayer[12] = 0x60 + if err := packet(pckt.Data[:76]).Err; !errors.Is(err, ErrHdrMissing("TCP")) { + t.Errorf("should fail with %q, got %q", ErrHdrMissing("TCP"), err) + return + } + pckt.NetLayer[56] = 0x02 + if err := packet(pckt.Data).Err; !errors.Is(err, ErrHdrExpected("TCP")) { + t.Errorf("should fail with %q, got %q", ErrHdrExpected("TCP"), err) + return + } + pckt.NetLayer[56] = 0x06 +} + +func packet(data []byte) *Packet { + return NewPacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{}) +} + +func BenchmarkNewPacketIPv4(b *testing.B) { + data := append(generateHeader4(1204, 10), make([]byte, 10)...) + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := packet(data).Err; err != nil { + b.Error(err) + } + } +} + +func BenchmarkNewPacketIPv6(b *testing.B) { + data := append(generateHeader6(1024, 10), make([]byte, 10)...) + b.ResetTimer() + for i := 0; i < b.N; i++ { + if err := packet(data).Err; err != nil { + b.Error(err) + } + } +} diff --git a/input_raw.go b/input_raw.go index 26dc2de..66062e8 100644 --- a/input_raw.go +++ b/input_raw.go @@ -72,13 +72,14 @@ type RAWInput struct { listener *capture.Listener message chan *tcp.Message cancelListener context.CancelFunc + closed bool } // NewRAWInput constructor for RAWInput. Accepts raw input config as arguments. func NewRAWInput(address string, config RAWInputConfig) (i *RAWInput) { i = new(RAWInput) i.RAWInputConfig = config - i.message = make(chan *tcp.Message, 1000) + i.message = make(chan *tcp.Message, 10000) i.quit = make(chan bool) var host, _port string var err error @@ -157,12 +158,12 @@ func (i *RAWInput) listen(address string) { var ctx context.Context ctx, i.cancelListener = context.WithCancel(context.Background()) errCh := i.listener.ListenBackground(ctx, pool.Handler) - select { - case err := <-errCh: - log.Fatal(err) - case <-i.listener.Reading: - Debug(1, i) - } + <-i.listener.Reading + Debug(1, i) + go func() { + <-errCh // the listener closed voluntarily + i.Close() + }() } func (i *RAWInput) handler(m *tcp.Message) { @@ -185,8 +186,14 @@ func (i *RAWInput) GetStats() []tcp.Stats { // Close closes the input raw listener func (i *RAWInput) Close() error { + i.Lock() + defer i.Unlock() + if i.closed { + return nil + } i.cancelListener() close(i.quit) + i.closed = true return nil } diff --git a/tcp/tcp_message.go b/tcp/tcp_message.go index f857ef1..07889c6 100644 --- a/tcp/tcp_message.go +++ b/tcp/tcp_message.go @@ -9,8 +9,8 @@ import ( "sync" "time" + "github.com/buger/goreplay/capture" "github.com/buger/goreplay/size" - "github.com/google/gopacket" ) // Stats every message carry its own stats object @@ -164,14 +164,11 @@ func NewMessagePool(maxSize size.Size, messageExpire time.Duration, debugger Deb } // Handler returns packet handler -func (pool *MessagePool) Handler(packet gopacket.Packet) { +func (pool *MessagePool) Handler(packet *capture.Packet) { var in, out bool pckt, err := ParsePacket(packet) if err != nil { - go pool.say(4, fmt.Sprintf("error decoding packet(%dBytes):%s\n", packet.Metadata().CaptureLength, err)) - return - } - if pckt == nil { + go pool.say(4, fmt.Sprintf("error decoding packet(%dBytes):%s\n", packet.Info.CaptureLength, err)) return } pool.Lock() diff --git a/tcp/tcp_packet.go b/tcp/tcp_packet.go index 445fe58..68e8461 100644 --- a/tcp/tcp_packet.go +++ b/tcp/tcp_packet.go @@ -2,12 +2,12 @@ package tcp import ( "encoding/binary" + "errors" "fmt" "net" "time" - "github.com/google/gopacket" - "github.com/google/gopacket/layers" + "github.com/buger/goreplay/capture" ) /* @@ -17,81 +17,51 @@ calllers must make sure that ParsePacket has'nt returned any error before callin function. */ type Packet struct { - // Link layer - gopacket.LinkLayer - - // IP Header - Version uint8 // Ip version - SrcIP net.IP - DstIP net.IP - IHL uint8 - Length uint16 - - // TCP Segment Header - *layers.TCP - - // Data info - Lost uint16 - Timestamp time.Time + SrcIP, DstIP net.IP + Version uint8 + SrcPort, DstPort uint16 + Ack, Seq uint32 + ACK, SYN, FIN, RST bool + Lost uint32 + Timestamp time.Time + Payload []byte } // ParsePacket parse raw packets -func ParsePacket(packet gopacket.Packet) (pckt *Packet, err error) { +func ParsePacket(packet *capture.Packet) (pckt *Packet, err error) { // early check of error if packet == nil { - return + return nil, errors.New("empty packet") } - defer func() { - if packet.ErrorLayer() != nil { - err = packet.ErrorLayer().Error() - return - } - }() - - // initialization - pckt = new(Packet) - pckt.Timestamp = packet.Metadata().Timestamp - if pckt.Timestamp.IsZero() { - pckt.Timestamp = time.Now() + if packet.Err != nil { + return nil, packet.Err } - // parsing link layer - pckt.LinkLayer = packet.LinkLayer() - - // parsing network layer - if net4, ok := packet.NetworkLayer().(*layers.IPv4); ok { + var t Packet + pckt = &t + // TODO: check resolution + pckt.Timestamp = packet.Info.Timestamp + if (packet.NetLayer[0] >> 4) == 4 { + // IPv4 header pckt.Version = 4 - pckt.SrcIP = net4.SrcIP - pckt.DstIP = net4.DstIP - pckt.IHL = net4.IHL * 4 - pckt.Length = net4.Length - } else if net6, ok := packet.NetworkLayer().(*layers.IPv6); ok { + pckt.SrcIP = packet.NetLayer[12:16] + pckt.DstIP = packet.NetLayer[16:20] + } else { + // IPv6 header pckt.Version = 6 - pckt.SrcIP = net6.SrcIP - pckt.DstIP = net6.DstIP - pckt.IHL = 40 - pckt.Length = net6.Length - } else { - pckt = nil - return + pckt.SrcIP = packet.NetLayer[8:24] + pckt.DstIP = packet.NetLayer[24:40] } - - // parsing tcp header(transportation layer) - if tcp, ok := packet.TransportLayer().(*layers.TCP); ok { - pckt.TCP = tcp - } else { - pckt = nil - return - } - pckt.DataOffset *= 4 - - // calculating lost data - headerSize := int(uint32(pckt.DataOffset) + uint32(pckt.IHL)) - if pckt.Version == 6 { - headerSize -= 40 // in ipv6 the length of payload doesn't include the IPheader size - } - pckt.Lost = pckt.Length - uint16(headerSize+len(pckt.Payload)) - + pckt.SrcPort = binary.BigEndian.Uint16(packet.TransLayer[0:2]) + pckt.DstPort = binary.BigEndian.Uint16(packet.TransLayer[2:4]) + pckt.Seq = binary.BigEndian.Uint32(packet.TransLayer[4:8]) + pckt.Ack = binary.BigEndian.Uint32(packet.TransLayer[8:12]) + pckt.FIN = packet.TransLayer[13]&0x01 != 0 + pckt.SYN = packet.TransLayer[13]&0x02 != 0 + pckt.RST = packet.TransLayer[13]&0x04 != 0 + pckt.ACK = packet.TransLayer[13]&0x10 != 0 + pckt.Lost = uint32(packet.Info.Length - packet.Info.CaptureLength) + pckt.Payload = packet.Payload return } @@ -104,94 +74,3 @@ func (pckt *Packet) Src() string { func (pckt *Packet) Dst() string { return fmt.Sprintf("%s:%d", pckt.DstIP, pckt.DstPort) } - -// SYNOptions returns MSS and windowscale of syn packets -func (pckt *Packet) SYNOptions() (mss uint16, windowscale byte) { - if !pckt.SYN { - return - } - for _, v := range pckt.Options { - if v.OptionType == layers.TCPOptionKindMSS { - mss = binary.BigEndian.Uint16(v.OptionData) - continue - } - if v.OptionType == layers.TCPOptionKindWindowScale { - if v.OptionLength > 0 { - windowscale = v.OptionData[0] - } - } - } - return -} - -// LinkInfo returns info about the link layer -func (pckt *Packet) LinkInfo() string { - if l, ok := pckt.LinkLayer.(*layers.Ethernet); ok { - return fmt.Sprintf( - "Source Mac: %s\nDestination Mac: %s\nProtocol: %s", - l.SrcMAC, - l.DstMAC, - l.EthernetType, - ) - } - return "" -} - -// Flag returns formatted tcp flags -func (pckt *Packet) Flag() (flag string) { - if pckt.FIN { - flag += "FIN, " - } - if pckt.SYN { - flag += "SYN, " - } - if pckt.RST { - flag += "RST, " - } - if pckt.PSH { - flag += "PSH, " - } - if pckt.ACK { - flag += "ACK, " - } - if pckt.URG { - flag += "URG, " - } - if len(flag) != 0 { - return flag[:len(flag)-2] - } - return flag -} - -// String output for a TCP Packet -func (pckt *Packet) String() string { - return fmt.Sprintf(`Time: %s -%s -Source: %s -Destination: %s -IHL: %d -Total Length: %d -Sequence: %d -Acknowledgment: %d -DataOffset: %d -Window: %d -Flag: %s -Options: %s -Data Size: %d -Lost Data: %d`, - pckt.Timestamp.Format(time.StampNano), - pckt.LinkInfo(), - pckt.Src(), - pckt.Dst(), - pckt.IHL, - pckt.Length, - pckt.Seq, - pckt.Ack, - pckt.DataOffset, - pckt.Window, - pckt.Flag(), - pckt.Options, - len(pckt.Payload), - pckt.Lost, - ) -} diff --git a/tcp/tcp_test.go b/tcp/tcp_test.go index fb7208e..22a4ed0 100644 --- a/tcp/tcp_test.go +++ b/tcp/tcp_test.go @@ -6,41 +6,38 @@ import ( "testing" "time" + "github.com/buger/goreplay/capture" "github.com/buger/goreplay/proto" "github.com/google/gopacket" "github.com/google/gopacket/layers" ) -var decodeOpts = gopacket.DecodeOptions{Lazy: true, NoCopy: true} +func generateHeader(seq uint32, length uint16) []byte { + hdr := make([]byte, 4+24+24, 4+24+24) + binary.BigEndian.PutUint32(hdr, uint32(layers.ProtocolFamilyIPv4)) -func headersIP4(seq uint32, length uint16) (headers [54]byte) { - // set ethernet headers - binary.BigEndian.PutUint16(headers[12:14], uint16(layers.EthernetTypeIPv4)) - - // set ip header - ip := headers[14:] - copy(ip[0:2], []byte{4<<4 | 5, 0x28<<2 | 0x00}) - binary.BigEndian.PutUint16(ip[2:4], length+40) + ip := hdr[4:] + ip[0] = 4<<4 | 6 + binary.BigEndian.PutUint16(ip[2:4], length+24+24) ip[9] = uint8(layers.IPProtocolTCP) - copy(ip[12:16], []byte{192, 168, 1, 2}) - copy(ip[16:], []byte{192, 168, 1, 3}) + copy(ip[12:16], []byte{127, 0, 0, 1}) + copy(ip[16:], []byte{127, 0, 0, 1}) // set tcp header - tcp := ip[20:] - binary.BigEndian.PutUint16(tcp[0:2], 45678) - binary.BigEndian.PutUint16(tcp[2:4], 8001) - tcp[12] = 5 << 4 - return + tcp := ip[24:] + tcp[12] = 6 << 4 + binary.BigEndian.PutUint16(tcp, 5535) + binary.BigEndian.PutUint16(tcp[2:], 8000) + binary.BigEndian.PutUint32(tcp[4:], seq) + return hdr } -func GetPackets(start uint32, _len int, payload []byte) []gopacket.Packet { - var packets = make([]gopacket.Packet, _len) +func GetPackets(start uint32, _len int, payload []byte) []*capture.Packet { + var packets = make([]*capture.Packet, _len) for i := start; i < start+uint32(_len); i++ { - data := make([]byte, 54+len(payload)) - h := headersIP4(i, uint16(len(payload))) - copy(data, h[:]) - copy(data[len(h):], payload) - packets[i-start] = gopacket.NewPacket(data, layers.LinkTypeEthernet, decodeOpts) + d := append(generateHeader(i, uint16(len(payload))), payload...) + ci := &gopacket.CaptureInfo{Length: len(d), CaptureLength: len(d), Timestamp: time.Now()} + packets[i-start] = capture.NewPacket(d, int(layers.LinkTypeLoop), 4, ci) } return packets } @@ -55,9 +52,9 @@ func TestMessageParserWithHint(t *testing.T) { return proto.HasFullPayload(m.Data(), m) } packets := GetPackets(1, 30, nil) - packets[0].Data()[14:][20:][13] = 2 // SYN flag - packets[10].Data()[14:][20:][13] = 2 // SYN flag - packets[29].Data()[14:][20:][13] = 1 // FIN flag + packets[0].TransLayer[13] = 2 // SYN flag + packets[10].TransLayer[13] = 2 // SYN flag + packets[29].TransLayer[13] = 1 // FIN flag packets[4] = GetPackets(5, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7"))[0] packets[5] = GetPackets(6, 1, []byte("\r\nMozilla\r\n9\r\nDeveloper\r"))[0] packets[6] = GetPackets(7, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0] @@ -105,8 +102,8 @@ func TestMessageParserWithoutHint(t *testing.T) { var mssg = make(chan *Message, 1) var data [63 << 10]byte packets := GetPackets(1, 10, data[:]) - packets[0].Data()[14:][20:][13] = 2 // SYN flag - packets[9].Data()[14:][20:][13] = 1 // FIN flag + packets[0].TransLayer[13] = 2 // SYN flag + packets[9].TransLayer[13] = 1 // FIN flag p := NewMessagePool(63<<10*10, time.Second, nil, func(m *Message) { mssg <- m }) for _, v := range packets { p.Handler(v) @@ -128,9 +125,9 @@ func TestMessageMaxSizeReached(t *testing.T) { var data [63 << 10]byte packets := GetPackets(1, 2, data[:]) packets = append(packets, GetPackets(3, 1, make([]byte, 63<<10+10))...) - packets[0].Data()[14:][20:][13] = 2 // SYN flag - packets[2].Data()[14:][20:][13] = 2 // SYN flag - packets[2].Data()[14:][15] = 3 // changing address + packets[0].TransLayer[13] = 2 // SYN flag + packets[2].TransLayer[13] = 2 // SYN flag + packets[2].NetLayer[15] = 3 // changing address p := NewMessagePool(63<<10+10, time.Second, nil, func(m *Message) { mssg <- m }) for _, v := range packets { p.Handler(v) @@ -167,7 +164,7 @@ func TestMessageTimeoutReached(t *testing.T) { var mssg = make(chan *Message, 2) var data [63 << 10]byte packets := GetPackets(1, 2, data[:]) - packets[0].Data()[14:][20:][13] = 2 // SYN flag + packets[0].TransLayer[13] = 2 // SYN flag p := NewMessagePool(1<<20, 0, nil, func(m *Message) { mssg <- m }) p.Handler(packets[0]) time.Sleep(time.Millisecond * 200) @@ -183,10 +180,10 @@ func TestMessageTimeoutReached(t *testing.T) { func TestMessageUUID(t *testing.T) { packets := GetPackets(1, 10, nil) - packets[0].Data()[14:][20:][13] = 2 // SYN flag - packets[4].Data()[14:][20:][13] = 1 // FIN flag - packets[5].Data()[14:][20:][13] = 2 // SYN flag - packets[9].Data()[14:][20:][13] = 1 // FIN flag + packets[0].TransLayer[13] = 2 // SYN flag + packets[4].TransLayer[13] = 1 // FIN flag + packets[5].TransLayer[13] = 2 // SYN flag + packets[9].TransLayer[13] = 1 // FIN flag var uuid, uuid1 []byte pool := NewMessagePool(0, 0, nil, func(msg *Message) { if len(uuid) == 0 { @@ -207,8 +204,8 @@ func TestMessageUUID(t *testing.T) { func BenchmarkMessageUUID(b *testing.B) { packets := GetPackets(1, 5, nil) - packets[0].Data()[14:][20:][13] = 2 // SYN flag - packets[4].Data()[14:][20:][13] = 1 // FIN flag + packets[0].TransLayer[13] = 2 // SYN flag + packets[4].TransLayer[13] = 1 // FIN flag var uuid []byte var msg *Message pool := NewMessagePool(0, 0, nil, func(m *Message) { @@ -241,8 +238,8 @@ func BenchmarkMessageParserWithoutHint(b *testing.B) { var mssg = make(chan *Message, 1) var chunk = []byte("111111111111111111111111111111") packets := GetPackets(1, 1000, chunk) - packets[0].Data()[14:][20:][13] = 2 // SYN flag - packets[1000-1].Data()[14:][20:][13] = 1 // FIN flag + packets[0].TransLayer[13] = 2 // SYN flag + packets[1000-1].TransLayer[13] = 1 // FIN flag p := NewMessagePool(1<<20, time.Second*2, nil, func(m *Message) { mssg <- m }) @@ -264,7 +261,7 @@ func BenchmarkMessageParserWithHint(b *testing.B) { buf[i] = chunk } buf[1001] = []byte("0\r\n\r\n") - packets := make([]gopacket.Packet, len(buf)) + packets := make([]*capture.Packet, len(buf)) for i := 0; i < len(buf); i++ { packets[i] = GetPackets(uint32(i+10), 1, buf[i])[0] } @@ -286,3 +283,24 @@ func BenchmarkMessageParserWithHint(b *testing.B) { <-mssg } } + +func BenchmarkNewAndParsePacket(b *testing.B) { + data := append(generateHeader(1024, 10), make([]byte, 10)...) + b.ResetTimer() + for i := 0; i < b.N; i++ { + ParsePacket(capture.NewPacket(data, int(layers.LinkTypeLoop), 4, &gopacket.CaptureInfo{})) + } +} + +func BenchmarkNewPacket(b *testing.B) { + packet := capture.NewPacket( + append(generateHeader(1024, 10), make([]byte, 10)...), + int(layers.LinkTypeLoop), + 4, + &gopacket.CaptureInfo{}, + ) + b.ResetTimer() + for i := 0; i < b.N; i++ { + ParsePacket(packet) + } +}