Add bidirectional packet capture

build/yamls/antrea.yml

@@ -3035,7 +3035,9 @@ spec:
                               type: integer
                               minimum: 1
                               maximum: 65535
-
+                bidirection:
+                  type: boolean
+                  default: false
                 timeout:
                   type: integer
                   minimum: 1

go.mod

@@ -40,7 +40,7 @@ require (
 	github.com/mdlayher/packet v1.1.2
 	github.com/miekg/dns v1.1.62
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822
-	github.com/onsi/ginkgo/v2 v2.22.0
+	github.com/onsi/ginkgo/v2 v2.22.1
 	github.com/onsi/gomega v1.36.1
 	github.com/osrg/gobgp/v3 v3.32.0
 	github.com/pkg/sftp v1.13.7
@@ -57,7 +57,7 @@ require (
 	go.uber.org/mock v0.5.0
 	golang.org/x/crypto v0.31.0
 	golang.org/x/mod v0.22.0
-	golang.org/x/net v0.32.0
+	golang.org/x/net v0.33.0
 	golang.org/x/sync v0.10.0
 	golang.org/x/sys v0.28.0
 	golang.org/x/time v0.8.0
@@ -147,7 +147,7 @@ require (
 	github.com/google/gnostic-models v0.6.8 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
-	github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db // indirect
+	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7 // indirect

go.sum

@@ -367,8 +367,8 @@ github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0=
 github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/pprof v0.0.0-20181206194817-3ea8567a2e57/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
-github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db h1:097atOisP2aRj7vFgYQBbFN4U4JNXUNYpxael3UzMyo=
-github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
+github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad h1:a6HEuzUHeKH6hwfN/ZoQgRgVIWFJljSWa/zetS2WTvg=
+github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad/go.mod h1:vavhavw2zAxS5dIdcRluK6cSGGPlZynqzFM8NdvU144=
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
 github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
@@ -588,8 +588,8 @@ github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108
 github.com/onsi/ginkgo v1.14.0/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
-github.com/onsi/ginkgo/v2 v2.22.0 h1:Yed107/8DjTr0lKCNt7Dn8yQ6ybuDRQoMGrNFKzMfHg=
-github.com/onsi/ginkgo/v2 v2.22.0/go.mod h1:7Du3c42kxCUegi0IImZ1wUQzMBVecgIHjR1C+NkhLQo=
+github.com/onsi/ginkgo/v2 v2.22.1 h1:QW7tbJAUDyVDVOM5dFa7qaybo+CRfR7bemlQUN6Z8aM=
+github.com/onsi/ginkgo/v2 v2.22.1/go.mod h1:S6aTpoRsSq2cZOd+pssHAlKW/Q/jZt6cPrPlnj4a1xM=
 github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=
 github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
@@ -900,8 +900,8 @@ golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.14.0/go.mod h1:PpSgVXXLK0OxS0F31C1/tv6XNguvCrnXIDrFMspZIUI=
 golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
-golang.org/x/net v0.32.0 h1:ZqPmj8Kzc+Y6e0+skZsuACbx+wzMgo5MQsJh9Qd6aYI=
-golang.org/x/net v0.32.0/go.mod h1:CwU0IoeOlnQQWJ6ioyFrfRuomB8GKF6KbYXZVyeXNfs=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
 golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=

pkg/agent/packetcapture/capture/bpf.go

@@ -75,17 +75,17 @@ func compareProtocol(protocol uint32, skipTrue, skipFalse uint8) bpf.Instruction
 // compilePacketFilter compiles the CRD spec to bpf instructions. For now, we only focus on
 // ipv4 traffic. Compared to the raw BPF filter supported by libpcap, we only need to support
 // limited use cases, so an expression parser is not needed.
-func compilePacketFilter(packetSpec *crdv1alpha1.Packet, srcIP, dstIP net.IP) []bpf.Instruction {
-	size := uint8(calculateInstructionsSize(packetSpec))
+func compilePacketFilter(packetSpec *crdv1alpha1.Packet, srcIP, dstIP net.IP, bidirection bool) []bpf.Instruction {
+	size := uint8(calculateInstructionsSize(packetSpec, bidirection))
 
 	// ipv4 check
-	inst := []bpf.Instruction{loadEtherKind}
+	inst := []bpf.Instruction{loadEtherKind} //(000)
 	// skip means how many instructions we need to skip if the compare fails.
 	// for example, for now we have 2 instructions, and the total size is 17, if ipv4
 	// check failed, we need to jump to the end (ret #0), skip 17-3=14 instructions.
 	// if check succeed, skipTrue means we jump to the next instruction. Here 3 means we
 	// have 3 instructions so far.
-	inst = append(inst, compareProtocolIP4(0, size-3))
+	inst = append(inst, compareProtocolIP4(0, size-3)) //(001)
 
 	if packetSpec != nil {
 		if packetSpec.Protocol != nil {
@@ -96,24 +96,27 @@ func compilePacketFilter(packetSpec *crdv1alpha1.Packet, srcIP, dstIP net.IP) []
 				proto = ProtocolMap[strings.ToUpper(packetSpec.Protocol.StrVal)]
 			}
 
-			inst = append(inst, loadIPv4Protocol)
-			inst = append(inst, compareProtocol(proto, 0, size-5))
+			inst = append(inst, loadIPv4Protocol)                  //(002)
+			inst = append(inst, compareProtocol(proto, 0, size-5)) //(003) 27-5=22
 		}
 	}
 
 	// source ip
 	if srcIP != nil {
-		inst = append(inst, loadIPv4SourceAddress)
+		inst = append(inst, loadIPv4SourceAddress) //(004)
 		addrVal := binary.BigEndian.Uint32(srcIP[len(srcIP)-4:])
 		// from here we need to check the inst length to calculate skipFalse. If no protocol is set, there will be no related bpf instructions.
-		inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: addrVal, SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2})
-
+		if bidirection {
+			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: addrVal, SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 13}) //(005) 27-5-13=9
+		} else {
+			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: addrVal, SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(005) 17-5-2=10
+		}
 	}
 	// dst ip
 	if dstIP != nil {
-		inst = append(inst, loadIPv4DestinationAddress)
+		inst = append(inst, loadIPv4DestinationAddress) //(006)
 		addrVal := binary.BigEndian.Uint32(dstIP[len(dstIP)-4:])
-		inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: addrVal, SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2})
+		inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: addrVal, SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(007) 18, 8
 	}
 
 	// ports
@@ -136,21 +139,48 @@ func compilePacketFilter(packetSpec *crdv1alpha1.Packet, srcIP, dstIP net.IP) []
 
 	if srcPort > 0 || dstPort > 0 {
 		skipTrue := size - uint8(len(inst)) - 3
-		inst = append(inst, loadIPv4HeaderOffset(skipTrue)...)
+		inst = append(inst, loadIPv4HeaderOffset(skipTrue)...) //(008), (009), (010)
 		if srcPort > 0 {
-			inst = append(inst, loadIPv4SourcePort)
-			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(srcPort), SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2})
+			inst = append(inst, loadIPv4SourcePort)                                                                                         //(011)
+			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(srcPort), SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(012)
 		}
 		if dstPort > 0 {
-			inst = append(inst, loadIPv4DestinationPort)
-			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(dstPort), SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2})
+			inst = append(inst, loadIPv4DestinationPort)                                                                                    //(013)
+			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(dstPort), SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(014)
+		}
+	}
+
+	if bidirection {
+		// src ip (return traffic)
+		if dstIP != nil {
+			addrVal := binary.BigEndian.Uint32(dstIP[len(dstIP)-4:])
+			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: addrVal, SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(015)
+		}
+
+		// dst ip (return traffic)
+		if srcIP != nil {
+			inst = append(inst, loadIPv4SourceAddress) //(016)
+			addrVal := binary.BigEndian.Uint32(srcIP[len(srcIP)-4:])
+			inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: addrVal, SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(017)
 		}
 
+		if srcPort > 0 || dstPort > 0 {
+			skipTrue := size - uint8(len(inst)) - 3
+			inst = append(inst, loadIPv4HeaderOffset(skipTrue)...) //(018), (019), (020)
+			if dstPort > 0 {
+				inst = append(inst, loadIPv4SourcePort)                                                                                         //(021)
+				inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(dstPort), SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(022)
+			}
+			if srcPort > 0 {
+				inst = append(inst, loadIPv4DestinationPort)                                                                                    //(023)
+				inst = append(inst, bpf.JumpIf{Cond: bpf.JumpEqual, Val: uint32(srcPort), SkipTrue: 0, SkipFalse: size - uint8(len(inst)) - 2}) //(024)
+			}
+		}
 	}
 
 	// return
-	inst = append(inst, returnKeep)
-	inst = append(inst, returnDrop)
+	inst = append(inst, returnKeep) //(015), (025)
+	inst = append(inst, returnDrop) //(016), (026)
 
 	return inst
 
@@ -178,7 +208,38 @@ func compilePacketFilter(packetSpec *crdv1alpha1.Packet, srcIP, dstIP net.IP) []
 // (015) ret      #262144                                  # MATCH
 // (016) ret      #0                                       # NOMATCH
 
-func calculateInstructionsSize(packet *crdv1alpha1.Packet) int {
+// When capturing return traffic also (i.e., both src -> dst and dst -> src), the filter might look like this:
+// 'ip proto 6 and ((src host 10.244.1.2 and dst host 10.244.1.3 and src port 123 and dst port 124) or (src host 10.244.1.3 and dst host 10.244.1.2 and src port 124 and dst port 123))'
+// And using `tcpdump -i <device> '<filter>' -d` will generate the following BPF instructions:
+// (000) ldh      [12]									   # Load 2B at 12 (Ethertype)
+// (001) jeq      #0x800           jt 2	jf 26			   # Ethertype: If IPv4, goto #2, else #26
+// (002) ldb      [23]									   # Load 1B at 23 (IPv4 Protocol)
+// (003) jeq      #0x6             jt 4	jf 26			   # IPv4 Protocol: If TCP, goto #4, #26
+// (004) ld       [26]									   # Load 4B at 26 (source address)
+// (005) jeq      #0xaf40102       jt 6	jf 15			   # If bytes match(10.244.0.2), goto #6, else #15
+// (006) ld       [30]									   # Load 4B at 30 (dest address)
+// (007) jeq      #0xaf40103       jt 8	jf 26			   # If bytes match(10.244.0.3), goto #8, else #26
+// (008) ldh      [20]									   # Load 2B at 20 (13b Fragment Offset)
+// (009) jset     #0x1fff          jt 26	jf 10	   # Use 0x1fff as a mask for fragment offset; If fragment offset != 0, #10, else #26
+// (010) ldxb     4*([14]&0xf)							   # x = IP header length
+// (011) ldh      [x + 14]								   # Load 2B at x+14 (TCP Source Port)
+// (012) jeq      #0x7b            jt 13	jf 26	   # TCP Source Port: If 123, goto #13, else #26
+// (013) ldh      [x + 16]								   # Load 2B at x+16 (TCP dst port)
+// (014) jeq      #0x7c            jt 25	jf 26	   # TCP dst port: If 123, goto #25, else #26
+// (015) jeq      #0xaf40103       jt 16	jf 26		   # If bytes match(10.244.0.3), goto #16, else #26
+// (016) ld       [30]									   # Load 4B at 30 (return traffic dest address)
+// (017) jeq      #0xaf40102       jt 18	jf 26		   # If bytes match(10.244.0.2), goto #18, else #26
+// (018) ldh      [20]									   # Load 2B at 20 (13b Fragment Offset)
+// (019) jset     #0x1fff          jt 26	jf 20	   # Use 0x1fff as a mask for fragment offset; If fragment offset != 0, #20, else #26
+// (020) ldxb     4*([14]&0xf)							   # x = IP header length
+// (021) ldh      [x + 14]								   # Load 2B at x+14 (TCP Source Port)
+// (022) jeq      #0x7c            jt 23	jf 26	   # TCP Source Port: If 124, goto #23, else #26
+// (023) ldh      [x + 16]								   # Load 2B at x+16 (TCP dst port)
+// (024) jeq      #0x7b            jt 25	jf 26	   # TCP dst port: If 123, goto #25, else #26
+// (025) ret      #262144								   # MATCH
+// (026) ret      #0									   # NOMATCH
+
+func calculateInstructionsSize(packet *crdv1alpha1.Packet, bidirection bool) int {
 	count := 0
 	// load ethertype
 	count++
@@ -214,6 +275,10 @@ func calculateInstructionsSize(packet *crdv1alpha1.Packet) int {
 	// src and dst ip
 	count += 4
 
+	if bidirection {
+		count += 10
+	}
+
 	// ret command
 	count += 2
 	return count

pkg/agent/packetcapture/capture/bpf_test.go

@@ -34,9 +34,10 @@ var (
 
 func TestCalculateInstructionsSize(t *testing.T) {
 	tt := []struct {
-		name   string
-		packet *crdv1alpha1.Packet
-		count  int
+		name        string
+		packet      *crdv1alpha1.Packet
+		count       int
+		bidirection bool
 	}{
 		{
 			name: "proto and host and port",
@@ -92,7 +93,7 @@ func TestCalculateInstructionsSize(t *testing.T) {
 
 	for _, item := range tt {
 		t.Run(item.name, func(t *testing.T) {
-			assert.Equal(t, item.count, calculateInstructionsSize(item.packet))
+			assert.Equal(t, item.count, calculateInstructionsSize(item.packet, item.bidirection))
 		})
 	}
 }
@@ -177,7 +178,7 @@ func TestPacketCaptureCompileBPF(t *testing.T) {
 
 	for _, item := range tt {
 		t.Run(item.name, func(t *testing.T) {
-			result := compilePacketFilter(item.spec.Packet, item.srcIP, item.dstIP)
+			result := compilePacketFilter(item.spec.Packet, item.srcIP, item.dstIP, item.spec.Bidirection)
 			assert.Equal(t, item.inst, result)
 		})
 	}

pkg/agent/packetcapture/capture/pcap_linux.go

@@ -41,9 +41,9 @@ func zeroFilter() []bpf.Instruction {
 	return []bpf.Instruction{returnDrop}
 }
 
-func (p *pcapCapture) Capture(ctx context.Context, device string, snapLen int, srcIP, dstIP net.IP, packet *crdv1alpha1.Packet) (chan gopacket.Packet, error) {
+func (p *pcapCapture) Capture(ctx context.Context, device string, snapLen int, srcIP, dstIP net.IP, packet *crdv1alpha1.Packet, bidirection bool) (chan gopacket.Packet, error) {
 	// Compile the BPF filter in advance to reduce the time window between starting the capture and applying the filter.
-	inst := compilePacketFilter(packet, srcIP, dstIP)
+	inst := compilePacketFilter(packet, srcIP, dstIP, bidirection)
 	klog.V(5).InfoS("Generated bpf instructions for PacketCapture", "device", device, "srcIP", srcIP, "dstIP", dstIP, "packetSpec", packet, "bpf", inst)
 	rawInst, err := bpf.Assemble(inst)
 	if err != nil {

pkg/agent/packetcapture/capture_interface.go

@@ -24,5 +24,5 @@ import (
 )
 
 type PacketCapturer interface {
-	Capture(ctx context.Context, device string, snapLen int, srcIP, dstIP net.IP, packet *crdv1alpha1.Packet) (chan gopacket.Packet, error)
+	Capture(ctx context.Context, device string, snapLen int, srcIP, dstIP net.IP, packet *crdv1alpha1.Packet, bidirection bool) (chan gopacket.Packet, error)
 }

pkg/agent/packetcapture/packetcapture_controller.go

@@ -464,7 +464,7 @@ func (c *Controller) performCapture(
 	}
 	defer pcapngWriter.Flush()
 	updateRateLimiter := rate.NewLimiter(rate.Every(captureStatusUpdatePeriod), 1)
-	packets, err := c.captureInterface.Capture(ctx, device, snapLen, srcIP, dstIP, pc.Spec.Packet)
+	packets, err := c.captureInterface.Capture(ctx, device, snapLen, srcIP, dstIP, pc.Spec.Packet, pc.Spec.Bidirection)
 	if err != nil {
 		return false, err
 	}

pkg/agent/packetcapture/packetcapture_controller_test.go

@@ -193,7 +193,7 @@ func craftTestPacket() gopacket.Packet {
 type testCapture struct {
 }
 
-func (p *testCapture) Capture(ctx context.Context, device string, snapLen int, srcIP, dstIP net.IP, packet *crdv1alpha1.Packet) (chan gopacket.Packet, error) {
+func (p *testCapture) Capture(ctx context.Context, device string, snapLen int, srcIP, dstIP net.IP, packet *crdv1alpha1.Packet, bidirection bool) (chan gopacket.Packet, error) {
 	ch := make(chan gopacket.Packet, testCaptureNum)
 	for i := 0; i < 15; i++ {
 		ch <- craftTestPacket()

pkg/apis/crd/v1alpha1/types.go

@@ -439,6 +439,9 @@ type PacketCaptureSpec struct {
 	// for a capture session, and at least one `Pod` should be present either in the source or the destination.
 	Source      Source      `json:"source"`
 	Destination Destination `json:"destination"`
+	// Bidirection specifies whether to capture the return (response) traffic from the destination back to the source.
+	// If not specified, defaults to false.
+	Bidirection bool `json:"bidirection"`
 	// Packet defines what kind of traffic we want to capture between the source and destination. If not specified,
 	// all kinds of traffic will count.
 	Packet *Packet `json:"packet,omitempty"`