diff --git a/capture/capture.go b/capture/capture.go index b06cbef..92e36e3 100644 --- a/capture/capture.go +++ b/capture/capture.go @@ -44,6 +44,7 @@ type Listener struct { Activate func() error // function is used to activate the engine. it must be called before reading packets Handles map[string]gopacket.PacketDataSource Interfaces []NetInterface + loopIndex int Reading chan bool // this channel is closed when the listener has started reading packets PcapOptions Engine EngineType @@ -73,7 +74,7 @@ func (eng *EngineType) Set(v string) error { *eng = EnginePcap case "pcap_file": *eng = EnginePcapFile - case "sock_raw", "af_packet": + case "raw_socket", "af_packet": *eng = EngineRawSocket default: return fmt.Errorf("invalid engine %s", v) @@ -88,7 +89,7 @@ func (eng *EngineType) String() (e string) { case EnginePcap: e = "libpcap" case EngineRawSocket: - e = "sock_raw" + e = "raw_socket" default: e = "" } @@ -308,6 +309,7 @@ func (l *Listener) SocketHandle(ifi NetInterface) (handle *SockRaw, err error) { handle.Close() return nil, fmt.Errorf("BPF filter error: %q%s, interface: %q", err, l.BPFFilter, ifi.Name) } + handle.SetLoopbackIndex(int32(l.loopIndex)) return } @@ -426,6 +428,9 @@ func (l *Listener) setInterfaces() (err error) { } for i := 0; i < len(ifis); i++ { + if ifis[i].Flags&net.FlagLoopback != 0 { + l.loopIndex = ifis[i].Index + } if ifis[i].Flags&net.FlagUp == 0 { continue } diff --git a/capture/capture_test.go b/capture/capture_test.go index 63bcf99..d17deea 100644 --- a/capture/capture_test.go +++ b/capture/capture_test.go @@ -11,6 +11,7 @@ import ( "github.com/google/gopacket" "github.com/google/gopacket/layers" + "github.com/google/gopacket/pcap" ) var LoopBack = func() net.Interface { @@ -164,21 +165,7 @@ func TestPcapHandler(t *testing.T) { t.Errorf("expected error to be nil, got %v", err) return } - quit := make(chan bool, 1) - pckts := 0 - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - errCh := l.ListenBackground(ctx, func(packet gopacket.Packet) { - pckts++ - if pckts == 10 { - quit <- true - } - }) - select { - case err = <-errCh: - t.Error(err) - case <-l.Reading: - } + defer l.Handles[LoopBack.Name].(*pcap.Handle).Close() if err != nil { t.Errorf("expected error to be nil, got %v", err) return @@ -186,10 +173,9 @@ func TestPcapHandler(t *testing.T) { for i := 0; i < 5; i++ { _, _ = net.Dial("tcp", "127.0.0.1:8000") } - select { - case <-time.After(time.Second * 2): - t.Error("failed to parse packets in time") - case <-quit: + sts, _ := l.Handles[LoopBack.Name].(*pcap.Handle).Stats() + if sts.PacketsReceived < 5 { + t.Errorf("expected >=5 packets got %d", sts.PacketsReceived) } } @@ -204,21 +190,7 @@ func TestSocketHandler(t *testing.T) { t.Errorf("expected error to be nil, got %v", err) return } - quit := make(chan bool, 1) - pckts := 0 - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - errCh := l.ListenBackground(ctx, func(packet gopacket.Packet) { - pckts++ - if pckts == 10 { - quit <- true - } - }) - select { - case err = <-errCh: - t.Error(err) - case <-l.Reading: - } + defer l.Handles[LoopBack.Name].(*SockRaw).Close() if err != nil { t.Errorf("expected error to be nil, got %v", err) return @@ -226,10 +198,9 @@ func TestSocketHandler(t *testing.T) { for i := 0; i < 5; i++ { _, _ = net.Dial("tcp", "127.0.0.1:8000") } - select { - case <-time.After(time.Second * 2): - t.Error("failed to parse packets in time") - case <-quit: + sts, _ := l.Handles[LoopBack.Name].(*SockRaw).Stats() + if sts.Packets < 5 { + t.Errorf("expected >=5 packets got %d", sts.Packets) } } @@ -305,36 +276,19 @@ func BenchmarkPcap(b *testing.B) { b.Errorf("expected error to be nil, got %v", err) return } - quit := make(chan bool, 1) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - pckts := 0 - errCh := l.ListenBackground(ctx, func(_ gopacket.Packet) { - pckts++ - if pckts == b.N*2 { - quit <- true - } - }) - select { - case err = <-errCh: - b.Error(err) - case <-l.Reading: - } + defer l.Handles[LoopBack.Name].(*pcap.Handle).Close() for i := 0; i < b.N; i++ { _, _ = net.Dial("tcp", "127.0.0.1:8000") } - select { - case <-time.After(time.Second): - case <-quit: - } - b.Logf("%d/%d packets in %s", pckts, b.N*2, time.Since(now)) + sts, _ := l.Handles[LoopBack.Name].(*pcap.Handle).Stats() + b.Logf("%d packets in %s", sts.PacketsReceived, time.Since(now)) } func BenchmarkRawSocket(b *testing.B) { now := time.Now() var err error - l, err := NewListener(LoopBack.Name, 8000, "", EngineRawSocket, true) + l, err := NewListener(LoopBack.Name, 0, "", EngineRawSocket, true) if err != nil { b.Errorf("expected error to be nil, got %v", err) return @@ -344,33 +298,10 @@ func BenchmarkRawSocket(b *testing.B) { b.Errorf("expected error to be nil, got %v", err) return } - sock := l.Handles[LoopBack.Name].(*SockRaw) - quit := make(chan bool, 1) - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - pckts := 0 - errCh := l.ListenBackground(ctx, func(_ gopacket.Packet) { - pckts++ - if pckts == b.N*2 { - quit <- true - } - }) - select { - case err = <-errCh: - b.Error(err) - case <-l.Reading: - } + defer l.Handles[LoopBack.Name].(*SockRaw).Close() for i := 0; i < b.N; i++ { - buf := generateHeaders(1, 1<<10) - err = sock.WritePacketData(buf[:]) - if err != nil { - b.Error(err) - } - } - select { - case <-time.After(time.Second): - case <-quit: + _, _ = net.Dial("tcp", "127.0.0.1:8000") } sts, _ := l.Handles[LoopBack.Name].(*SockRaw).Stats() - b.Logf("%d/%d packets in %s", sts.Packets-sts.Drops, sts.Packets, time.Since(now)) + b.Logf("%d packets in %s", sts.Packets, time.Since(now)) } diff --git a/capture/sock.go b/capture/sock.go index 4674999..796c9d0 100644 --- a/capture/sock.go +++ b/capture/sock.go @@ -37,9 +37,10 @@ type SockRaw struct { fd int ifindex int snaplen int - pollTimeout int - frame int + pollTimeout uintptr + frame uint32 // current frame buf []byte // points to the memory space of the ring buffer shared with the kernel. + loopIndex int32 // this field must filled to avoid reading packet twice on a loopback device } // NewSockRaw returns new M'maped sock_raw on packet version 2. @@ -53,7 +54,7 @@ func NewSockRaw(ifi net.Interface) (*SockRaw, error) { fd: fd, ifindex: ifi.Index, snaplen: unix.IP_MAXPACKET, - pollTimeout: -1, + pollTimeout: ^uintptr(0), } // set packet version @@ -115,23 +116,32 @@ func (sock *SockRaw) ReadPacketData() (buf []byte, ci gopacket.CaptureInfo, err Fd: int32(sock.fd), Events: unix.POLLIN, } - i := sock.frame * FRAMESIZE + var i int +read: + i = int(sock.frame * FRAMESIZE) tpHdr = (*unix.Tpacket2Hdr)(unsafe.Pointer(&sock.buf[i])) for tpHdr.Status&unix.TP_STATUS_USER == 0 { - _, _, e := unix.Syscall(unix.SYS_POLL, uintptr(unsafe.Pointer(poll)), 1, uintptr(sock.pollTimeout)) + _, _, e := unix.Syscall(unix.SYS_POLL, uintptr(unsafe.Pointer(poll)), 1, sock.pollTimeout) if e != 0 && e != unix.EINTR { return buf, ci, e } } + + sock.frame = (sock.frame + 1) % FRAMENR tpHdr.Status = unix.TP_STATUS_KERNEL sockAddr := (*unix.RawSockaddrLinklayer)(unsafe.Pointer(&sock.buf[i+tpacket2hdrlen])) + + // parse out repeating packets on loopback, 4 frames will be wasted obviously! + if sockAddr.Ifindex == sock.loopIndex && sock.frame%2 != 0 { + goto read + } + ci.Length = int(tpHdr.Len) ci.Timestamp = time.Unix(int64(tpHdr.Sec), int64(tpHdr.Nsec)) ci.InterfaceIndex = int(sockAddr.Ifindex) buf = make([]byte, tpHdr.Snaplen) ci.CaptureLength = copy(buf, sock.buf[i+int(tpHdr.Mac):]) - sock.frame = (sock.frame + 1) % FRAMENR return } @@ -169,7 +179,7 @@ func (sock *SockRaw) SetSnapLen(snap int) error { func (sock *SockRaw) SetTimeout(t time.Duration) error { sock.mu.Lock() defer sock.mu.Unlock() - sock.pollTimeout = int(t) + sock.pollTimeout = uintptr(t) return nil } @@ -204,7 +214,8 @@ func (sock *SockRaw) SetBPFFilter(expr string) error { return unix.SetsockoptSockFprog(sock.fd, unix.SOL_SOCKET, unix.SO_ATTACH_FILTER, fprog) } -// SetPromiscuous sets promiscous mode to the required value. If it is enabled, traffic not destined for the interface will also be captured. +// SetPromiscuous sets promiscous mode to the required value. for better result capture on all interfaces instead. +// If it is enabled, traffic not destined for the interface will also be captured. func (sock *SockRaw) SetPromiscuous(b bool) error { sock.mu.Lock() defer sock.mu.Unlock() @@ -228,6 +239,13 @@ func (sock *SockRaw) Stats() (*unix.TpacketStats, error) { return unix.GetsockoptTpacketStats(sock.fd, unix.SOL_PACKET, unix.PACKET_STATISTICS) } +// SetLoopbackIndex necessary to avoid reading packet twice on a loopback device +func (sock *SockRaw) SetLoopbackIndex(i int32) { + sock.mu.Lock() + defer sock.mu.Unlock() + sock.loopIndex = i +} + // WritePacketData transmits a raw packet. func (sock *SockRaw) WritePacketData(pkt []byte) error { _, err := unix.Write(sock.fd, pkt)