datapath: fix 12 correctness findings from tun/UDP offload review

Multi-disciplinary correctness review of the batched tun / GSO-GRO / sendmmsg rework. Each fix has a regression test; the merged tree builds on linux/darwin/openbsd/windows/freebsd/netbsd, vets clean, passes the unit and e2e suites, and is -race clean. Critical: - C1 zero-length inner UDP datagram no longer panics the process (remote DoS): the UDP coalescer routes payLen==0 to passthrough instead of seeding a GSO slot, and WriteGSO skips empty payload iovecs as defense in depth. - C2 segmenter no longer corrupts inner headers when gsoSize < headerLen: the L3+L4 header is snapshotted once and each segment stamped from the copy, replacing the destructive overlapping in-place slide (SegmentTCP + SegmentUDP). High: - H1 applyOuterECN updates the IPv4 header checksum (RFC 1624 incremental) when folding outer CE into the inner ToS, so passthrough packets are no longer dropped by the peer stack. - H2 the GRO reject path caps the borrowed RX segment ([:n:n]) so a reject can no longer overrun into the next coalesced segment's Nebula header. Note: oversized ICMPv6 rejects that need >16B beyond the segment are now refused rather than sent under GRO (safe; see TOFIX.md for the scratch-buffer follow-up). - H3 WriteBatch falls back to per-packet WriteTo for a chunk when writeSockaddr fails, so one bad-family destination costs only its own packet, not the batch. - H4 UserDevice.Readers returns N distinct queue wrappers with private buffers (sharing the pipes) so concurrent readers no longer race/overwrite borrowed packet bytes. - H5 Poll.Close / Offload.Close no longer null t.fd (matching master's tunFile.Close), removing the data race with a concurrent readOne load. Medium/Low: - M1 the UDP GSO 127-segment gate moved from kernel >=5.5 to >=6.9 (the real UDP_MAX_SEGMENTS 64->128 threshold), avoiding EINVAL + per-packet fallback on 5.5-6.8 kernels. - M2 NewMultiQueueReader replays the offload mask newTun actually negotiated instead of the TSO-only mask, so adding a queue no longer disables USO device-wide; the advertised USO capability derives from the same mask. - M3 the shutdown eventfd is closed in pollQueueSet.Close / offloadQueueSet.Close (double-close guarded), fixing the per-lifecycle fd leak. - M4 dual-stack ECN selects the cmsg by address family, not socket family: RX parseRecvCmsg reads both IP_TOS and IPV6_TCLASS; TX writeEntryCmsg stamps IP_TOS for v4/v4-mapped dests and IPV6_TCLASS for v6 (on-host verified). - L1 newPoll no longer closes the fd on failure (matching newOffload), removing the double-close on QueueSet.Add error.
2026-08-15 15:37:03 +02:00 · 2026-07-13 16:35:20 -05:00
parent 9e61269935
commit 0a44376403
19 changed files with 1346 additions and 72 deletions
@@ -167,6 +167,10 @@ func (u *StdConn) prepareWriteMessages(n int) {
 	u.writeCmsgSpace = u.writeCmsgSegSpace + u.writeCmsgEcnSpace
 	u.writeCmsg = make([]byte, n*u.writeCmsgSpace)

+	// Default the ECN header to the socket's own family. writeEntryCmsg
+	// finalizes Level/Type per entry from the destination address (a v4-mapped
+	// dst on a dual-stack v6 socket needs IP_TOS, not IPV6_TCLASS), so this is
+	// only the value used before the first per-entry rewrite.
 	ecnLevel := int32(unix.IPPROTO_IP)
 	ecnType := int32(unix.IP_TOS)
 	if !u.isV4 {
@@ -225,16 +229,29 @@ func (u *StdConn) prepareGSO() {
 		recordCapability("udp.gso.enabled", false)
 		return
 	}
-	major, minor := parseRelease(string(un.Release[:]))
-	if major > 5 || (major == 5 && minor >= 5) {
-		u.maxGSOSegments = 127
-	}
+	u.maxGSOSegments = gsoMaxSegments(string(un.Release[:]))

 	u.gsoSupported = true
 	u.l.Info("udp: GSO enabled", "maxGSOSegments", u.maxGSOSegments)
 	recordCapability("udp.gso.enabled", true)
 }

+// gsoMaxSegments returns the largest number of UDP_SEGMENT segments a single
+// sendmsg may carry on the running kernel, reserving one segment for the
+// header. UDP_MAX_SEGMENTS was 64 until Linux v6.9 (commit 1382e3b6a350,
+// "udp: change maximum number of UDP segments to 128") raised it to 128;
+// nothing about this changed in 5.5. On kernels older than 6.9 packing more
+// than 64 segments gets the sendmsg rejected with EINVAL, so cap at 63 there
+// and only use 127 from 6.9 on. (Maintainer stance: update your kernel if you
+// want to go fast — this is a plain version gate, not a runtime probe.)
+func gsoMaxSegments(release string) int {
+	major, minor := parseRelease(release)
+	if major > 6 || (major == 6 && minor >= 9) {
+		return 127
+	}
+	return 63
+}
+
 // udpGROBufferSize sizes the per-entry recvmmsg buffer when UDP_GRO is on.
 // The kernel stitches a run of same-flow datagrams into a single skb whose
 // length is bounded by sk_gso_max_size (typically 65535); anything larger
@@ -501,7 +518,7 @@ func (u *StdConn) listenOutBatch(r EncReader, flush func()) error {
 			segSize := 0
 			outerECN := byte(0)
 			if cmsgSpace > 0 {
-				segSize, outerECN = parseRecvCmsg(&msgs[i].Hdr, u.groSupported, u.ecnRecvSupported, u.isV4)
+				segSize, outerECN = parseRecvCmsg(&msgs[i].Hdr, u.groSupported, u.ecnRecvSupported)
 			}

 			if segSize <= 0 || segSize >= len(payload) {
@@ -535,9 +552,14 @@ func headerCounter(buf []byte) uint64 {
 // two values of interest in a single pass: the UDP_GRO gso_size (when
 // wantGRO is true) and the outer IP-level ECN codepoint stamped on the
 // carrier (when wantECN is true). Returns zeros for whichever field is not
-// requested or not present. isV4 selects between IP_TOS (1-byte) and
-// IPV6_TCLASS (4-byte int) cmsg payloads.
-func parseRecvCmsg(hdr *msghdr, wantGRO, wantECN bool, isV4 bool) (gso int, ecn byte) {
+// requested or not present.
+//
+// The outer ECN is accepted from EITHER an IP_TOS (IPPROTO_IP, 1-byte) or an
+// IPV6_TCLASS (IPPROTO_IPV6, 4-byte int) cmsg, regardless of the socket's
+// family: a dual-stack v6 socket (isV4 == false) delivers IPv4 peers' outer
+// ECN as an IP_TOS cmsg — gating on socket family here dropped v4-underlay
+// ECN entirely. Whichever cmsg the kernel delivered carries the value.
+func parseRecvCmsg(hdr *msghdr, wantGRO, wantECN bool) (gso int, ecn byte) {
 	controllen := int(hdr.Controllen)
 	if controllen < unix.SizeofCmsghdr || hdr.Control == nil {
 		return 0, 0
@@ -556,12 +578,13 @@ func parseRecvCmsg(hdr *msghdr, wantGRO, wantECN bool, isV4 bool) (gso int, ecn
 			if dataOff+udpGROCmsgPayload <= len(ctrl) {
 				gso = int(int32(binary.NativeEndian.Uint32(ctrl[dataOff : dataOff+udpGROCmsgPayload])))
 			}
-		case wantECN && isV4 && ch.Level == unix.IPPROTO_IP && ch.Type == unix.IP_TOS:
+		case wantECN && ch.Level == unix.IPPROTO_IP && ch.Type == unix.IP_TOS:
 			// IP_TOS arrives as a single byte; only the low 2 bits are ECN.
+			// A dual-stack v6 socket carries v4 peers' outer ECN here.
 			if dataOff+1 <= len(ctrl) {
 				ecn = ctrl[dataOff] & 0x03
 			}
-		case wantECN && !isV4 && ch.Level == unix.IPPROTO_IPV6 && ch.Type == unix.IPV6_TCLASS:
+		case wantECN && ch.Level == unix.IPPROTO_IPV6 && ch.Type == unix.IPV6_TCLASS:
 			// IPV6_TCLASS arrives as a 4-byte int; ECN is the low 2 bits.
 			if dataOff+4 <= len(ctrl) {
 				ecn = byte(binary.NativeEndian.Uint32(ctrl[dataOff:dataOff+4])) & 0x03
@@ -611,6 +634,7 @@ func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte)
 	// providing no observed reordering benefit.

 	i := 0
+sendChunks:
 	for i < len(bufs) {
 		baseI := i
 		entry := 0
@@ -638,7 +662,24 @@ func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte)

 			nlen, err := writeSockaddr(u.writeNames[entry], addrs[i], u.isV4)
 			if err != nil {
-				return err
+				// One destination in this chunk has an address family the
+				// socket can't send to (e.g. an IPv6 remote on a v4-bound
+				// socket → ErrInvalidIPv6RemoteForSocket). Abandoning the whole
+				// sendmmsg here would drop every packet already packed for this
+				// chunk plus every packet still ahead of us in bufs. Instead
+				// fall back to per-packet WriteTo for the packets packed so far
+				// in this chunk and the offending one: WriteTo delivers each
+				// good destination and only errors on the bad one, which we
+				// drop and keep going. One bad destination costs one packet,
+				// never the batch. (Same fallback the zero-sent sendmmsg path
+				// below uses, extended to cover the misaddressed packet.)
+				for k := baseI; k <= i; k++ {
+					if werr := u.WriteTo(bufs[k], addrs[k]); werr != nil && k != i {
+						return werr
+					}
+				}
+				i++
+				continue sendChunks
 			}

 			hdr := &u.writeMsgs[entry].Hdr
@@ -650,7 +691,11 @@ func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort, ecns []byte)
 			if ecns != nil {
 				ecn = ecns[i]
 			}
-			u.writeEntryCmsg(entry, runLen, segSize, ecn)
+			// ECN cmsg family follows the destination, not the socket: a
+			// v4-mapped dst on a dual-stack v6 socket must be stamped via
+			// IP_TOS. addrs[i] is this run's destination (i advances below).
+			dstIsV4 := addrs[i].Addr().Unmap().Is4()
+			u.writeEntryCmsg(entry, runLen, segSize, ecn, dstIsV4)

 			i += runLen
 			iovIdx += runLen
@@ -767,7 +812,15 @@ func (u *StdConn) planRun(bufs [][]byte, addrs []netip.AddrPort, ecns []byte, st
 // entry. It writes the UDP_SEGMENT payload when runLen >= 2 and the
 // IP_TOS/IPV6_TCLASS payload when ecn != 0, then points hdr.Control at the
 // smallest contiguous span that covers whichever cmsg(s) actually apply.
-func (u *StdConn) writeEntryCmsg(entry, runLen, segSize int, ecn byte) {
+//
+// The outer-ECN cmsg family must match the *destination*, not the socket: on
+// the default dual-stack v6 bind, a v4-mapped destination is routed through
+// the kernel's IPv4 path, which parses IP_TOS (IPPROTO_IP) and ignores an
+// IPV6_TCLASS cmsg. prepareWriteMessages pre-fills a default header; here we
+// rewrite its Level/Type (and Len) per entry from dstIsV4 so v4 peers get
+// IP_TOS and v6 peers get IPV6_TCLASS. The data payload is a 4-byte int for
+// both families, so the pre-computed cmsg space is unchanged.
+func (u *StdConn) writeEntryCmsg(entry, runLen, segSize int, ecn byte, dstIsV4 bool) {
 	hdr := &u.writeMsgs[entry].Hdr
 	useSeg := runLen >= 2
 	useEcn := ecn != 0
@@ -778,6 +831,15 @@ func (u *StdConn) writeEntryCmsg(entry, runLen, segSize int, ecn byte) {
 		binary.NativeEndian.PutUint16(u.writeCmsg[dataOff:dataOff+2], uint16(segSize))
 	}
 	if useEcn {
+		ecnHdr := (*unix.Cmsghdr)(unsafe.Pointer(&u.writeCmsg[base+u.writeCmsgSegSpace]))
+		if dstIsV4 {
+			ecnHdr.Level = int32(unix.IPPROTO_IP)
+			ecnHdr.Type = int32(unix.IP_TOS)
+		} else {
+			ecnHdr.Level = int32(unix.IPPROTO_IPV6)
+			ecnHdr.Type = int32(unix.IPV6_TCLASS)
+		}
+		setCmsgLen(ecnHdr, unix.CmsgLen(4))
 		dataOff := base + u.writeCmsgSegSpace + unix.CmsgLen(0)
 		binary.NativeEndian.PutUint32(u.writeCmsg[dataOff:dataOff+4], uint32(ecn))
 	}
@@ -0,0 +1,234 @@
+//go:build linux && !android && !e2e_testing
+
+package udp
+
+import (
+	"encoding/binary"
+	"io"
+	"log/slog"
+	"net"
+	"net/netip"
+	"syscall"
+	"testing"
+	"time"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+)
+
+// TestGSOMaxSegmentsKernelGate pins the corrected kernel-version gate: the
+// 128-segment cap (127 usable) only lands in Linux v6.9 (commit 1382e3b6a350),
+// not 5.5. Everything older stays at the conservative 63.
+func TestGSOMaxSegmentsKernelGate(t *testing.T) {
+	cases := []struct {
+		release string
+		want    int
+	}{
+		{"5.4.0", 63},
+		{"5.5.0-generic", 63}, // the old bug bumped here — it must not now
+		{"5.15.0", 63},
+		{"6.1.0", 63},
+		{"6.8.0-generic", 63},
+		{"6.9.0", 127},
+		{"6.10.1-arch1-1", 127},
+		{"7.0.5-arch1-1", 127},
+		{"garbage", 63},
+		{"", 63},
+	}
+	for _, c := range cases {
+		if got := gsoMaxSegments(c.release); got != c.want {
+			t.Errorf("gsoMaxSegments(%q) = %d, want %d", c.release, got, c.want)
+		}
+	}
+}
+
+// buildCmsg lays out a single ancillary cmsg (header + data) in a fresh buffer
+// the way the kernel would deliver it, so parseRecvCmsg can be exercised
+// without a live socket.
+func buildCmsg(level, typ int32, data []byte) []byte {
+	buf := make([]byte, unix.CmsgSpace(len(data)))
+	h := (*unix.Cmsghdr)(unsafe.Pointer(&buf[0]))
+	h.Level = level
+	h.Type = typ
+	setCmsgLen(h, unix.CmsgLen(len(data)))
+	copy(buf[unix.CmsgLen(0):], data)
+	return buf
+}
+
+// TestParseRecvCmsgOuterECNFamily is the RX half of the dual-stack ECN fix:
+// parseRecvCmsg must read the outer ECN from whichever family the kernel
+// delivered, not from the socket family. On the default `::` dual-stack bind
+// a v4 peer's outer ECN arrives as an IP_TOS cmsg, which the old socket-family
+// gate ignored entirely.
+func TestParseRecvCmsgOuterECNFamily(t *testing.T) {
+	tc := make([]byte, 4)
+	binary.NativeEndian.PutUint32(tc, 0x02)
+
+	cases := []struct {
+		name string
+		ctrl []byte
+		want byte
+	}{
+		{"ip_tos_ce", buildCmsg(int32(unix.IPPROTO_IP), int32(unix.IP_TOS), []byte{0x03}), 0x03},
+		{"ip_tos_ect0", buildCmsg(int32(unix.IPPROTO_IP), int32(unix.IP_TOS), []byte{0x02}), 0x02},
+		{"ipv6_tclass_ect0", buildCmsg(int32(unix.IPPROTO_IPV6), int32(unix.IPV6_TCLASS), tc), 0x02},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			hdr := &msghdr{Control: &c.ctrl[0]}
+			setMsgControllen(hdr, len(c.ctrl))
+			gso, ecn := parseRecvCmsg(hdr, false, true)
+			if gso != 0 {
+				t.Errorf("gso = %d, want 0 (no UDP_GRO cmsg present)", gso)
+			}
+			if ecn != c.want {
+				t.Errorf("ecn = 0x%02x, want 0x%02x", ecn, c.want)
+			}
+		})
+	}
+}
+
+func testLogger() *slog.Logger {
+	return slog.New(slog.NewTextHandler(io.Discard, nil))
+}
+
+// TestWriteBatchBadFamilyDeliversOthers is the H3 regression: a batch that
+// contains one destination the socket can't reach (an IPv6 remote on a
+// v4-bound socket) must still deliver every other packet. Before the fix the
+// writeSockaddr error returned early and dropped the whole chunk.
+func TestWriteBatchBadFamilyDeliversOthers(t *testing.T) {
+	rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	if err != nil {
+		t.Skipf("cannot open v4 receiver (sandbox?): %v", err)
+	}
+	defer rx.Close()
+	rxPort := rx.LocalAddr().(*net.UDPAddr).Port
+
+	// Bind a *non-wildcard* v4 address so Go gives us a genuine AF_INET
+	// socket. A wildcard v4 bind (0.0.0.0) via network "udp" comes up as a
+	// dual-stack AF_INET6 socket on Linux, for which a v6 dest is not a bad
+	// family — which would defeat the point of this test.
+	c, err := NewListener(testLogger(), netip.MustParseAddr("127.0.0.1"), 0, false, 1)
+	if err != nil {
+		t.Skipf("cannot open v4 sender (sandbox?): %v", err)
+	}
+	defer c.Close()
+	sender := c.(*StdConn)
+	if !sender.isV4 {
+		t.Fatalf("expected a v4-bound sender socket, got isV4=false")
+	}
+
+	good := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(rxPort))
+	bad := netip.MustParseAddrPort("[2001:db8::1]:9999") // genuine v6, unreachable on v4 socket
+
+	bufs := [][]byte{[]byte("AAA"), []byte("BBB"), []byte("CCC")}
+	addrs := []netip.AddrPort{good, bad, good}
+
+	if err := sender.WriteBatch(bufs, addrs, nil); err != nil {
+		t.Fatalf("WriteBatch returned error, want nil (bad dest should be isolated): %v", err)
+	}
+
+	got := map[string]bool{}
+	rx.SetReadDeadline(time.Now().Add(2 * time.Second))
+	buf := make([]byte, 64)
+	for i := 0; i < 2; i++ {
+		n, _, rerr := rx.ReadFromUDPAddrPort(buf)
+		if rerr != nil {
+			t.Fatalf("expected 2 delivered packets, read #%d failed: %v", i+1, rerr)
+		}
+		got[string(buf[:n])] = true
+	}
+	if !got["AAA"] || !got["CCC"] {
+		t.Errorf("delivered set = %v, want AAA and CCC both present", got)
+	}
+	if got["BBB"] {
+		t.Errorf("the bad-family packet BBB was somehow delivered")
+	}
+}
+
+// TestWriteBatchOuterTOSToV4Mapped is the TX half of the dual-stack ECN fix,
+// verified against a live kernel: WriteBatch on the default `::` dual-stack
+// socket, sending to a v4-mapped destination, must stamp the outer ECN via an
+// IP_TOS cmsg (not IPV6_TCLASS, which the kernel's v4 path ignores) so a v4
+// receiver actually sees it.
+func TestWriteBatchOuterTOSToV4Mapped(t *testing.T) {
+	rx, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	if err != nil {
+		t.Skipf("cannot open v4 receiver (sandbox?): %v", err)
+	}
+	defer rx.Close()
+	rxPort := rx.LocalAddr().(*net.UDPAddr).Port
+
+	// Ask the kernel to deliver the received outer TOS as ancillary data.
+	rxRaw, err := rx.SyscallConn()
+	if err != nil {
+		t.Fatalf("SyscallConn: %v", err)
+	}
+	var soErr error
+	if err := rxRaw.Control(func(fd uintptr) {
+		soErr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_RECVTOS, 1)
+	}); err != nil || soErr != nil {
+		t.Skipf("cannot enable IP_RECVTOS (sandbox/kernel?): ctrl=%v so=%v", err, soErr)
+	}
+
+	c, err := NewListener(testLogger(), netip.IPv6Unspecified(), 0, false, 1)
+	if err != nil {
+		t.Skipf("cannot open dual-stack sender (sandbox?): %v", err)
+	}
+	defer c.Close()
+	sender := c.(*StdConn)
+	if sender.isV4 {
+		t.Skipf("sender came up v4-only; need a dual-stack v6 socket for this test")
+	}
+
+	// v4-mapped-in-v6 destination: routed through the kernel's IPv4 path.
+	dst := netip.AddrPortFrom(netip.AddrFrom4([4]byte{127, 0, 0, 1}), uint16(rxPort))
+	const wantECN = byte(0x02) // ECT(0)
+
+	if err := sender.WriteBatch([][]byte{[]byte("tos-probe")}, []netip.AddrPort{dst}, []byte{wantECN}); err != nil {
+		t.Fatalf("WriteBatch: %v", err)
+	}
+
+	// Read the datagram plus its ancillary TOS.
+	rx.SetReadDeadline(time.Now().Add(3 * time.Second))
+	payload := make([]byte, 128)
+	oob := make([]byte, 512)
+	var n, oobn int
+	var rerr error
+	if err := rxRaw.Read(func(fd uintptr) bool {
+		n, oobn, _, _, rerr = unix.Recvmsg(int(fd), payload, oob, 0)
+		if rerr == syscall.EAGAIN || rerr == syscall.EWOULDBLOCK {
+			return false
+		}
+		return true
+	}); err != nil {
+		t.Fatalf("waiting for datagram failed (no delivery?): %v", err)
+	}
+	if rerr != nil {
+		t.Fatalf("Recvmsg: %v", rerr)
+	}
+	if string(payload[:n]) != "tos-probe" {
+		t.Fatalf("payload = %q, want %q", string(payload[:n]), "tos-probe")
+	}
+
+	cmsgs, err := unix.ParseSocketControlMessage(oob[:oobn])
+	if err != nil {
+		t.Fatalf("ParseSocketControlMessage: %v", err)
+	}
+	found := false
+	var gotTOS byte
+	for _, m := range cmsgs {
+		if m.Header.Level == unix.IPPROTO_IP && m.Header.Type == unix.IP_TOS && len(m.Data) >= 1 {
+			found = true
+			gotTOS = m.Data[0]
+		}
+	}
+	if !found {
+		t.Fatalf("no IP_TOS cmsg delivered to v4 receiver — outer ECN did not land (%d cmsgs)", len(cmsgs))
+	}
+	if gotTOS&0x03 != wantECN {
+		t.Errorf("received outer TOS = 0x%02x, want low-2-bits = 0x%02x", gotTOS, wantECN)
+	} else {
+		t.Logf("verified: v4 receiver saw outer TOS 0x%02x (ECN=0x%02x) from dual-stack sender", gotTOS, gotTOS&0x03)
+	}
+}