fix panic on defrag

README.md

@@ -1,9 +1,9 @@
 <p align="center">
 <img src="https://goreportcard.com/badge/github.com/dmachard/go-dns-collector" alt="Go Report"/>
 <img src="https://img.shields.io/badge/go%20version-min%201.21-green" alt="Go version"/>
-<img src="https://img.shields.io/badge/go%20tests-446-green" alt="Go tests"/>
+<img src="https://img.shields.io/badge/go%20tests-450-green" alt="Go tests"/>
 <img src="https://img.shields.io/badge/go%20bench-20-green" alt="Go bench"/>
-<img src="https://img.shields.io/badge/go%20lines-39776-green" alt="Go lines"/>
+<img src="https://img.shields.io/badge/go%20lines-37761-green" alt="Go lines"/>
 </p>
 
 <p align="center">

collectors/file_ingestor.go

@@ -114,9 +114,9 @@ func (c *FileIngestor) ProcessPcap(filePath string) {
 	packetSource.NoCopy = true
 
 	// defrag ipv4
-	go netutils.IPDefragger(fragIP4Chan, udpChan, tcpChan)
+	go netutils.IPDefragger(fragIP4Chan, udpChan, tcpChan, c.GetConfig().Collectors.FileIngestor.PcapDNSPort)
 	// defrag ipv6
-	go netutils.IPDefragger(fragIP6Chan, udpChan, tcpChan)
+	go netutils.IPDefragger(fragIP6Chan, udpChan, tcpChan, c.GetConfig().Collectors.FileIngestor.PcapDNSPort)
 	// tcp assembly
 	go netutils.TCPAssembler(tcpChan, dnsChan, c.GetConfig().Collectors.FileIngestor.PcapDNSPort)
 	// udp processor

collectors/sniffer_afpacket_linux.go

@@ -102,9 +102,9 @@ func (c *AfpacketSniffer) Run() {
 	netDecoder := &netutils.NetDecoder{}
 
 	// defrag ipv4
-	go netutils.IPDefragger(fragIP4Chan, udpChan, tcpChan)
+	go netutils.IPDefragger(fragIP4Chan, udpChan, tcpChan, c.GetConfig().Collectors.AfpacketLiveCapture.Port)
 	// defrag ipv6
-	go netutils.IPDefragger(fragIP6Chan, udpChan, tcpChan)
+	go netutils.IPDefragger(fragIP6Chan, udpChan, tcpChan, c.GetConfig().Collectors.AfpacketLiveCapture.Port)
 	// tcp assembly
 	go netutils.TCPAssembler(tcpChan, dnsChan, 0)
 	// udp processor
@@ -198,18 +198,24 @@ func (c *AfpacketSniffer) Run() {
 
 				// ipv4 fragmented packet ?
 				if packet.NetworkLayer().LayerType() == layers.LayerTypeIPv4 {
+					if !c.GetConfig().Collectors.AfpacketLiveCapture.FragmentSupport {
+						continue
+					}
 					ip4 := packet.NetworkLayer().(*layers.IPv4)
 					if ip4.Flags&layers.IPv4MoreFragments == 1 || ip4.FragOffset > 0 {
-						// fragIP4Chan <- packet
+						fragIP4Chan <- packet
 						continue
 					}
 				}
 
 				// ipv6 fragmented packet ?
 				if packet.NetworkLayer().LayerType() == layers.LayerTypeIPv6 {
+					if !c.GetConfig().Collectors.AfpacketLiveCapture.FragmentSupport {
+						continue
+					}
 					v6frag := packet.Layer(layers.LayerTypeIPv6Fragment)
 					if v6frag != nil {
-						// fragIP6Chan <- packet
+						fragIP6Chan <- packet
 						continue
 					}
 				}

docs/collectors/collector_afpacket.md

@@ -21,6 +21,9 @@ Options:
 * `device` (str)
   > Interface name to sniff. If value is empty, bind on all interfaces.
 
+* `enable-fragment-support` (bool)
+  > Enable IP defrag support
+
 * `chan-buffer-size` (int)
   > Specifies the maximum number of packets that can be buffered before dropping additional packets.
 
@@ -31,5 +34,6 @@ Defaults:
   afpacket-sniffer:
     port: 53
     device: wlp2s0
+    enable-fragment-support: true
     chan-buffer-size: 65535
 ```

netutils/bpf.go

@@ -23,34 +23,35 @@ func GetBpfFilter(port int) []bpf.Instruction {
 		// Load eth.type (2 bytes at offset 12) and push-it in register A
 		bpf.LoadAbsolute{Off: 12, Size: 2},
 		// if eth.type == IPv4 continue with the next instruction
-		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x0800, SkipTrue: 0, SkipFalse: 10},
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x0800, SkipTrue: 0, SkipFalse: 11},
 		// Load ip.proto (1 byte at offset 23) and push-it in register A
 		bpf.LoadAbsolute{Off: 23, Size: 1},
 		// ip.proto == UDP ?
 		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x11, SkipTrue: 1, SkipFalse: 0},
 		// ip.proto == TCP ?
-		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x6, SkipTrue: 0, SkipFalse: 16},
-		// load flags and fragment offset (2 bytes at offset 20) to ignore fragmented packet
-		bpf.LoadAbsolute{Off: 20, Size: 2},
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x6, SkipTrue: 0, SkipFalse: 17},
+		// load flags and fragment offset (2 bytes at offset 20)
 		// Only look at the last 13 bits of the data saved in regiter A
 		//  0x1fff == 0001 1111 1111 1111 (fragment offset)
-		// If any of the data in fragment offset is true, ignore the packet
+		bpf.LoadAbsolute{Off: 20, Size: 2},
 		bpf.JumpIf{Cond: bpf.JumpBitsSet, Val: 0x1fff, SkipTrue: 14, SkipFalse: 0},
-		// Load ip.length
 		// Register X = ip header len * 4
 		bpf.LoadMemShift{Off: 14},
 		// Load source port in tcp or udp (2 bytes at offset x+14)
 		bpf.LoadIndirect{Off: 14, Size: 2},
 		// source port equal to 53 ?
-		bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(port), SkipTrue: 10, SkipFalse: 0},
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(port), SkipTrue: 11, SkipFalse: 0},
 		// Load destination port in tcp or udp  (2 bytes at offset x+16)
 		bpf.LoadIndirect{Off: 16, Size: 2},
 		// destination port equal to 53 ?
-		bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(port), SkipTrue: 8, SkipFalse: 9},
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(port), SkipTrue: 9, SkipFalse: 10},
+
 		// if eth.type == IPv6 continue with the next instruction
-		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x86dd, SkipTrue: 0, SkipFalse: 8},
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x86dd, SkipTrue: 0, SkipFalse: 9},
 		// Load ipv6.nxt (2 bytes at offset 12) and push-it in register A
 		bpf.LoadAbsolute{Off: 20, Size: 1},
+		// fragment ?
+		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x2c, SkipTrue: 6, SkipFalse: 0},
 		// ip.proto == UDP ?
 		bpf.JumpIf{Cond: bpf.JumpEqual, Val: 0x11, SkipTrue: 1, SkipFalse: 0},
 		// ip.proto == TCP ?
@@ -63,6 +64,7 @@ func GetBpfFilter(port int) []bpf.Instruction {
 		bpf.LoadAbsolute{Off: 56, Size: 2},
 		// destination port equal to 53 ?
 		bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(port), SkipTrue: 0, SkipFalse: 1},
+
 		// Keep the packet and send up to 65k of the packet to userspace
 		bpf.RetConstant{Val: 0xFFFF},
 		// Ignore packet

netutils/ipdefrag.go

@@ -56,11 +56,14 @@ func (f *fragments) insert(in gopacket.Packet) (gopacket.Packet, error) {
 
 			var fragOffset uint16
 
-			frag6 := packet.Layer(layers.LayerTypeIPv6Fragment).(*layers.IPv6Fragment)
-			ip4, _ := e.Value.(*layers.IPv4)
-			if frag6 != nil {
-				fragOffset = frag6.FragmentOffset * 8
-			} else {
+			switch packet.NetworkLayer().LayerType() {
+			case layers.LayerTypeIPv6:
+				if frag6Layer := packet.Layer(layers.LayerTypeIPv6Fragment); frag6Layer != nil {
+					frag6 := frag6Layer.(*layers.IPv6Fragment)
+					fragOffset = frag6.FragmentOffset * 8
+				}
+			case layers.LayerTypeIPv4:
+				ip4, _ := packet.Layer(layers.LayerTypeIPv4).(*layers.IPv4)
 				fragOffset = ip4.FragOffset * 8
 			}
 

netutils/ipdefrag_test.go

@@ -1,40 +1,101 @@
 package netutils
 
-// TODO
-/*
-func createIPv6FragmentPacketWithNilLayer() gopacket.Packet {
-	// IPv6 layer
-	ipLayer := &layers.IPv6{
-		Version:    6,
-		NextHeader: layers.IPProtocolIPv6Fragment, // Next header is Fragmentation Header
-		HopLimit:   64,
-		SrcIP:      net.ParseIP("2001:db8::1"),
-		DstIP:      net.ParseIP("2001:db8::2"),
-	}
-
-	// Create a packet with nil IPv6Fragment layer
-	builder := gopacket.NewSerializeBuffer()
-	ipLayer.SerializeTo(builder, gopacket.SerializeOptions{})
-	// Set the IPv6 layer manually
-	packet := gopacket.NewPacket(builder.Bytes(), layers.LayerTypeIPv6, gopacket.Default)
-	// Remove IPv6Fragment layer
-	packet.Layer(layers.LayerTypeIPv6Fragment).(*layers.IPv6Fragment).Payload = nil
-
-	return packet
-}
+import (
+	"testing"
+
+	"github.com/google/gopacket"
+	"github.com/google/gopacket/layers"
+)
 
+func TestIpDefrag_IPv4Fragment(t *testing.T) {
+	// fragmented packet, see the packet in testsdata/pcap/dnsdump_ip4_fragmented_query.pcap
+	frag1Bytes := []byte{0x58, 0x1d, 0xd8, 0x12, 0x84, 0x10, 0xb0, 0x35, 0x9f, 0xd4, 0x03, 0x91, 0x08, 0x00,
+		0x45, 0x00, 0x00, 0x4c, 0x00, 0x01, 0x20, 0x00, 0x40, 0x11, 0xcb, 0x8f, 0x7f, 0x00, 0x00, 0x01,
+		0x08, 0x08, 0x08, 0x08, 0x30, 0x39, 0x00, 0x35, 0x00, 0x5d, 0xf9, 0x10, 0xaa, 0xaa, 0x01, 0x00,
+		0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x3f, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41}
+	frag1 := gopacket.NewPacket(frag1Bytes, layers.LayerTypeEthernet, gopacket.Default)
+	frag2Bytes := []byte{0x58, 0x1d, 0xd8, 0x12, 0x84, 0x10, 0xb0, 0x35, 0x9f, 0xd4, 0x03, 0x91, 0x08, 0x00,
+		0x45, 0x00, 0x00, 0x39, 0x00, 0x01, 0x00, 0x07, 0x40, 0x11, 0xeb, 0x9b, 0x7f, 0x00, 0x00, 0x01,
+		0x08, 0x08, 0x08, 0x08, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x03, 0x63, 0x6f, 0x6d,
+		0x00, 0x00, 0x01, 0x00, 0x01}
+	frag2 := gopacket.NewPacket(frag2Bytes, layers.LayerTypeEthernet, gopacket.Default)
 
-func TestIpDefrag_WithNilIPv6Fragment(t *testing.T) {
 	defragger := NewIPDefragmenter()
+	_, err := defragger.DefragIP(frag1)
+	if err != nil {
+		t.Errorf("Unexpected error on the 1er defrag: %v", err)
+	}
 
-	// Create an IPv6 packet with nil IPv6Fragment layer
-	packet := createIPv6FragmentPacketWithNilLayer()
+	pkt, err := defragger.DefragIP(frag2)
+	if err != nil {
+		t.Errorf("Unexpected error on the 2nd defrag: %v", err)
+	}
 
+	if pkt.Metadata().Length != 113 {
+		t.Errorf("Invalid reassembled packet size: %v", err)
+	}
+}
 
-	// This packet has a nil IPv6Fragment layer, which should trigger an error
+func TestIpDefrag_IPv4FragmentWithRetransmission(t *testing.T) {
+
+	// fragmented packet, see the packet in testsdata/pcap/dnsdump_ip4_fragmented_query.pcap
+	packetBytes := []byte{
+		0x58, 0x1d, 0xd8, 0x12, 0x84, 0x10, 0xb0, 0x35, 0x9f, 0xd4, 0x03, 0x91, 0x08, 0x00, 0x45, 0x00,
+		0x00, 0x39, 0x00, 0x01, 0x00, 0x07, 0x40, 0x11, 0xeb, 0x9b, 0x7f, 0x00, 0x00, 0x01, 0x08, 0x08,
+		0x08, 0x08, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x03, 0x63,
+		0x6f, 0x6d, 0x00, 0x00, 0x01, 0x00, 0x01,
+	}
+	packet := gopacket.NewPacket(packetBytes, layers.LayerTypeEthernet, gopacket.Default)
+
+	// This packet is just a fragment
+	defragger := NewIPDefragmenter()
 	_, err := defragger.DefragIP(packet)
-	if err == nil {
-		t.Errorf("Expected error, got nil")
+	if err != nil {
+		t.Errorf("Unexpected error on the 1er defrag: %v", err)
+	}
+
+	// Try to defrag the same packet
+	_, err = defragger.DefragIP(packet)
+	if err != nil {
+		t.Errorf("Unexpected error for the 2nd defrag: %v", err)
+	}
+}
+
+func TestIpDefrag_IPv6Fragment(t *testing.T) {
+	// fragmented packet, see the packet in testsdata/pcap/dnsdump_ip4_fragmented_query.pcap
+	frag1Bytes := []byte{0x33, 0x33, 0x00, 0x00, 0x00, 0x01, 0xb0, 0x35, 0x9f, 0xd4, 0x03, 0x91, 0x86, 0xdd, 0x60, 0x00,
+		0x00, 0x00, 0x00, 0x38, 0x2c, 0x40, 0xfe, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x5b, 0x20,
+		0xbf, 0xa1, 0x87, 0x8c, 0x3a, 0x68, 0xff, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x11, 0x00, 0x00, 0x01, 0xee, 0x8f, 0x5f, 0xf8, 0x30, 0x39,
+		0x00, 0x35, 0x00, 0x61, 0x98, 0xc4, 0xaa, 0xaa, 0x01, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x07, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x3f, 0x63, 0x6f, 0x6d, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41}
+	frag1 := gopacket.NewPacket(frag1Bytes, layers.LayerTypeEthernet, gopacket.Default)
+	frag2Bytes := []byte{0x33, 0x33, 0x00, 0x00, 0x00, 0x01, 0xb0, 0x35, 0x9f, 0xd4, 0x03, 0x91, 0x86, 0xdd, 0x60, 0x00,
+		0x00, 0x00, 0x00, 0x39, 0x2c, 0x40, 0xfe, 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x5b, 0x20,
+		0xbf, 0xa1, 0x87, 0x8c, 0x3a, 0x68, 0xff, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x11, 0x00, 0x00, 0x30, 0xee, 0x8f, 0x5f, 0xf8, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41,
+		0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x41, 0x00, 0x00, 0x01, 0x00, 0x01}
+	frag2 := gopacket.NewPacket(frag2Bytes, layers.LayerTypeEthernet, gopacket.Default)
+
+	defragger := NewIPDefragmenter()
+	_, err := defragger.DefragIP(frag1)
+	if err != nil {
+		t.Errorf("Unexpected error on the 1er defrag: %v", err)
+	}
+
+	pkt, err := defragger.DefragIP(frag2)
+	if err != nil {
+		t.Errorf("Unexpected error on the 2nd defrag: %v", err)
+	}
+
+	if pkt.Metadata().Length != 137 {
+		t.Errorf("Invalid reassembled packet size: %v", err)
 	}
 }
-*/

netutils/packetproccesor.go

@@ -86,7 +86,7 @@ FLUSHALL:
 	assembler.FlushAll()
 }
 
-func IPDefragger(ipInput chan gopacket.Packet, udpOutput chan gopacket.Packet, tcpOutput chan gopacket.Packet) {
+func IPDefragger(ipInput chan gopacket.Packet, udpOutput chan gopacket.Packet, tcpOutput chan gopacket.Packet, port int) {
 	defragger := NewIPDefragmenter()
 	for fragment := range ipInput {
 		reassembled, err := defragger.DefragIP(fragment)
@@ -96,10 +96,23 @@ func IPDefragger(ipInput chan gopacket.Packet, udpOutput chan gopacket.Packet, t
 		if reassembled == nil {
 			continue
 		}
+
 		if reassembled.TransportLayer() != nil && reassembled.TransportLayer().LayerType() == layers.LayerTypeUDP {
+			// ignore packet regarding udp port
+			pkt := reassembled.TransportLayer().(*layers.UDP)
+			if pkt.DstPort != layers.UDPPort(port) && pkt.SrcPort != layers.UDPPort(port) {
+				continue
+			}
+			// valid reassembled packet
 			udpOutput <- reassembled
 		}
 		if reassembled.TransportLayer() != nil && reassembled.TransportLayer().LayerType() == layers.LayerTypeTCP {
+			// ignore packet regarding udp port
+			pkt := reassembled.TransportLayer().(*layers.TCP)
+			if pkt.DstPort != layers.TCPPort(port) && pkt.SrcPort != layers.TCPPort(port) {
+				continue
+			}
+			// valid reassembled packet
 			tcpOutput <- reassembled
 		}
 	}

netutils/packetprocessor_test.go

@@ -50,9 +50,9 @@ func Test_IpDefrag(t *testing.T) {
 			outputChan := make(chan gopacket.Packet, 2)
 
 			// defrag ipv4
-			go IPDefragger(fragIP4Chan, outputChan, outputChan)
+			go IPDefragger(fragIP4Chan, outputChan, outputChan, 53)
 			// defrag ipv6
-			go IPDefragger(fragIP6Chan, outputChan, outputChan)
+			go IPDefragger(fragIP6Chan, outputChan, outputChan, 53)
 
 			packetSource := gopacket.NewPacketSource(pcapHandler, pcapHandler.LinkType())
 			packetSource.DecodeOptions.Lazy = true

pkgconfig/collectors.go

@@ -49,6 +49,7 @@ type ConfigCollectors struct {
 		Port              int    `yaml:"port"`
 		Device            string `yaml:"device"`
 		ChannelBufferSize int    `yaml:"chan-buffer-size"`
+		FragmentSupport   bool   `yaml:"enable-defrag-ip"`
 	} `yaml:"afpacket-sniffer"`
 	XdpLiveCapture struct {
 		Enable            bool   `yaml:"enable"`
@@ -127,6 +128,7 @@ func (c *ConfigCollectors) SetDefault() {
 	c.AfpacketLiveCapture.Port = 53
 	c.AfpacketLiveCapture.Device = ""
 	c.AfpacketLiveCapture.ChannelBufferSize = 65535
+	c.AfpacketLiveCapture.FragmentSupport = true
 
 	c.PowerDNS.Enable = false
 	c.PowerDNS.ListenIP = AnyIP