diff --git a/ecn_inner_test.go b/ecn_inner_test.go index 6240a99b..18dc94d4 100644 --- a/ecn_inner_test.go +++ b/ecn_inner_test.go @@ -1,8 +1,11 @@ package nebula import ( + "encoding/binary" "log/slog" "testing" + + "golang.org/x/net/ipv4" ) func TestInnerECN(t *testing.T) { @@ -122,3 +125,64 @@ func TestApplyOuterECN(t *testing.T) { }) } } + +// TestApplyOuterECN_IPv4ChecksumStaysValid guards against H1: folding an outer +// CE mark into the inner IPv4 ToS byte must keep the IPv4 header checksum valid. +// The passthrough emit paths write the packet verbatim, so a stale checksum +// turns an underlay congestion mark into packet loss at the receiver. +func TestApplyOuterECN_IPv4ChecksumStaysValid(t *testing.T) { + silent := slog.New(slog.DiscardHandler) + hi := &HostInfo{} + + // 20-byte IPv4 header with DSCP=0x88 and inner ECN = ECT(0). Folding CE + // flips only the low two bits of the ToS byte while leaving DSCP intact. + pkt := []byte{ + 0x45, 0x88 | ecnECT0, 0, 40, + 0x1c, 0x46, 0x40, 0x00, + 64, 6, 0, 0, + 10, 0, 0, 1, + 10, 0, 0, 2, + } + // Stamp a correct header checksum before the fold. + binary.BigEndian.PutUint16(pkt[10:12], ipv4HeaderChecksum(pkt[:ipv4.HeaderLen])) + if !ipv4HeaderChecksumValid(pkt[:ipv4.HeaderLen]) { + t.Fatal("test setup: initial header checksum invalid") + } + + applyOuterECN(pkt, ecnCE, hi, silent) + + // CE folded in, DSCP preserved. + if got, want := pkt[1], byte(0x88|ecnCE); got != want { + t.Fatalf("ToS after fold = 0x%02x, want 0x%02x", got, want) + } + // The incremental RFC 1624 update must leave the checksum valid and equal + // to a full recompute over the mutated header. + if !ipv4HeaderChecksumValid(pkt[:ipv4.HeaderLen]) { + t.Fatalf("IPv4 header checksum invalid after CE fold: 0x%04x", binary.BigEndian.Uint16(pkt[10:12])) + } + if got, want := binary.BigEndian.Uint16(pkt[10:12]), ipv4HeaderChecksum(pkt[:ipv4.HeaderLen]); got != want { + t.Fatalf("checksum = 0x%04x, full recompute = 0x%04x", got, want) + } +} + +// ipv4HeaderChecksum computes the RFC 1071 IPv4 header checksum over hdr, +// treating the checksum field (bytes 10:12) as zero. +func ipv4HeaderChecksum(hdr []byte) uint16 { + var sum uint32 + for i := 0; i+1 < len(hdr); i += 2 { + if i == 10 { + continue // checksum field + } + sum += uint32(hdr[i])<<8 | uint32(hdr[i+1]) + } + for sum > 0xffff { + sum = (sum >> 16) + (sum & 0xffff) + } + return ^uint16(sum) +} + +// ipv4HeaderChecksumValid reports whether the stored checksum matches a fresh +// computation over the header. +func ipv4HeaderChecksumValid(hdr []byte) bool { + return binary.BigEndian.Uint16(hdr[10:12]) == ipv4HeaderChecksum(hdr) +} diff --git a/iputil/packet_test.go b/iputil/packet_test.go index 6d567d51..f79cc6e6 100644 --- a/iputil/packet_test.go +++ b/iputil/packet_test.go @@ -1,6 +1,7 @@ package iputil import ( + "bytes" "encoding/binary" "net" "testing" @@ -179,6 +180,51 @@ func Test_CreateRejectPacket_NoICMPError(t *testing.T) { } } +// Test_CreateRejectPacket_RespectsCap guards against H2: with UDP GRO the +// scratch buffer reused to build a reject is a single coalesced segment inside +// a shared recvmmsg row. Its length covers just that segment, but an uncapped +// slice's capacity runs on into the next, not-yet-processed segment. Because +// CreateRejectPacket honors cap, capping the borrowed segment to its own length +// (cap==len) makes it physically impossible for an oversized ICMPv6 reject to +// overwrite the neighbor segment's bytes. +func Test_CreateRejectPacket_RespectsCap(t *testing.T) { + src := net.ParseIP("fd00::1") + dst := net.ParseIP("fd00::2") + + // Inner IPv6 UDP packet. An ICMPv6 reject copies the whole inner packet + // plus a 48-byte header (40 IPv6 + 8 ICMPv6), so it needs 48 more bytes + // than the inner packet length. + inner := makeIPv6Packet(src, dst, 17, make([]byte, 20)) + + // The ciphertext scratch reused as the reject buffer is the received + // datagram: 16-byte Nebula header + inner + 16-byte AEAD tag. That is only + // 32 bytes of slack, so a full ICMPv6 reject overruns it by 16 bytes. + const nebulaOverhead = 32 + segLen := len(inner) + nebulaOverhead + + // Shared backing row laid out as [segment][neighbor's 16-byte Nebula header]. + const neighborHdr = 16 + sentinel := bytes.Repeat([]byte{0xAB}, neighborHdr) + + // Uncapped: the slice's capacity reaches into the neighbor, reproducing + // the overrun that silently drops the neighbor packet. + backing := make([]byte, segLen+neighborHdr) + copy(backing[segLen:], sentinel) + reject := CreateRejectPacket(inner, backing[:segLen]) + assert.NotNil(t, reject, "uncapped buffer reaches into the neighbor, so the reject is built") + assert.NotEqual(t, sentinel, backing[segLen:segLen+neighborHdr], + "without the cap the oversized reject overruns into the neighbor segment") + + // Capped (the fix): cap==len, so the builder cannot exceed the segment. The + // reject does not fit, so it is refused rather than corrupting the neighbor. + backing = make([]byte, segLen+neighborHdr) + copy(backing[segLen:], sentinel) + reject = CreateRejectPacket(inner, backing[:segLen:segLen]) + assert.Nil(t, reject, "capped segment is 16 bytes too small for a full ICMPv6 reject, so it is refused") + assert.Equal(t, sentinel, backing[segLen:segLen+neighborHdr], + "capped segment must leave the neighbor untouched") +} + func makeIPv6Packet(src, dst net.IP, nextHeader uint8, payload []byte) []byte { b := make([]byte, ipv6.HeaderLen+len(payload)) b[0] = ipv6.Version << 4 diff --git a/outside.go b/outside.go index 5aea566e..45b1702b 100644 --- a/outside.go +++ b/outside.go @@ -548,7 +548,23 @@ func applyOuterECN(pkt []byte, outerECN byte, hostinfo *HostInfo, l *slog.Logger case ecnCE: // Already CE. default: + // Rewriting the ToS byte invalidates the IPv4 header checksum, so + // patch it incrementally per RFC 1624 (HC' = ~(~HC + ~m + m')). The + // ToS is the low byte of the 16-bit word at pkt[0:2]; the header + // checksum lives at pkt[10:12]. A header too short to carry a + // checksum can't be fixed up here, so leave it for newPacket to + // reject rather than emit a mangled packet. + if len(pkt) < ipv4.HeaderLen { + return + } + m := binary.BigEndian.Uint16(pkt[0:2]) pkt[1] = (pkt[1] &^ 0x03) | ecnCE + mNew := binary.BigEndian.Uint16(pkt[0:2]) + sum := uint32(^binary.BigEndian.Uint16(pkt[10:12])) + uint32(^m) + uint32(mNew) + for sum > 0xffff { + sum = (sum >> 16) + (sum & 0xffff) + } + binary.BigEndian.PutUint16(pkt[10:12], ^uint16(sum)) } case 6: switch (pkt[1] >> 4) & 0x03 { @@ -584,8 +600,10 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, p dropReason := f.firewall.Drop(*fwPacket, true, hostinfo, f.pki.GetCAPool(), localCache) if dropReason != nil { // NOTE: We give `packet` as the `out` here since we already decrypted from it and we don't need it anymore - // This gives us a buffer to build the reject packet in - f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, nb, packet, q) + // This gives us a buffer to build the reject packet in. With UDP GRO this is a single segment of a shared + // recvmmsg row whose capacity runs to the end of the whole row, so cap it to its own length (cap==len) to + // keep the reject builder from writing past this segment into the next, not-yet-processed coalesced segment. + f.rejectOutside(out, hostinfo.ConnectionState, hostinfo, nb, packet[:len(packet):len(packet)], q) if f.l.Enabled(context.Background(), slog.LevelDebug) { hostinfo.logger(f.l).Debug("dropping inbound packet", "fwPacket", fwPacket, diff --git a/overlay/batch/udp_coalesce.go b/overlay/batch/udp_coalesce.go index 410d2982..d7ad2506 100644 --- a/overlay/batch/udp_coalesce.go +++ b/overlay/batch/udp_coalesce.go @@ -152,6 +152,17 @@ func (c *UDPCoalescer) commitParsed(pkt []byte, info parsedUDP) error { c.addPassthrough(pkt) return nil } + // A zero-length UDP datagram (UDP `length` == 8) is legal and must still + // reach the TUN, but it can't be coalesced: a GSO slot would store an + // empty payload iovec and the kernel has nothing to segment. Seal any + // open chain for this flow (so a later, non-empty datagram seeds fresh + // *after* this one and per-flow arrival order is preserved) and deliver + // it as a plain single datagram. + if info.payLen == 0 { + delete(c.openSlots, info.fk) + c.addPassthrough(pkt) + return nil + } if open := c.openSlots[info.fk]; open != nil { if c.canAppend(open, pkt, info) { c.appendPayload(open, pkt, info) diff --git a/overlay/batch/udp_coalesce_test.go b/overlay/batch/udp_coalesce_test.go index 368afafc..1e01760d 100644 --- a/overlay/batch/udp_coalesce_test.go +++ b/overlay/batch/udp_coalesce_test.go @@ -365,6 +365,74 @@ func TestUDPCoalescerFragmentedIPv4PassesThrough(t *testing.T) { } } +// A zero-length UDP datagram (UDP length == 8, no payload) is legal and +// must be delivered as a plain single datagram — never coalesced. Seeding +// it into a GSO slot stores an empty payload iovec that panics WriteGSO +// (index-out-of-range on &pay[0]); this is a remote DoS if we ever let it +// reach the GSO path. Regression: must not panic and must be written. +func TestUDPCoalescerZeroLengthPayloadPassesThrough(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := NewUDPCoalescer(w, NewArena(0)) + pkt := buildUDPv4(1000, 53, nil) // UDP length 8, zero payload + if err := c.Commit(pkt); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.writes) != 1 || len(w.gsoWrites) != 0 { + t.Fatalf("zero-length UDP must pass through plain, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) + } + if len(w.writes[0]) != len(pkt) { + t.Errorf("delivered %d bytes, want the whole %d-byte datagram", len(w.writes[0]), len(pkt)) + } +} + +// IPv6 zero-length UDP datagram: same passthrough contract as v4. +func TestUDPCoalescerZeroLengthPayloadIPv6PassesThrough(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := NewUDPCoalescer(w, NewArena(0)) + pkt := buildUDPv6(1000, 53, nil) // UDP length 8, zero payload + if err := c.Commit(pkt); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + if len(w.writes) != 1 || len(w.gsoWrites) != 0 { + t.Fatalf("zero-length IPv6 UDP must pass through plain, got writes=%d gso=%d", len(w.writes), len(w.gsoWrites)) + } + if len(w.writes[0]) != len(pkt) { + t.Errorf("delivered %d bytes, want the whole %d-byte datagram", len(w.writes[0]), len(pkt)) + } +} + +// A zero-length datagram arriving mid-flow must seal the open chain so the +// datagram after it seeds a fresh superpacket *after* the empty one on the +// wire — per-flow arrival order (full, empty, full) must be preserved. +func TestUDPCoalescerZeroLengthMidFlowSealsAndPreservesOrder(t *testing.T) { + w := &fakeTunWriter{gsoEnabled: true} + c := NewUDPCoalescer(w, NewArena(0)) + full := make([]byte, 800) + if err := c.Commit(buildUDPv4(1000, 53, full)); err != nil { + t.Fatal(err) + } + if err := c.Commit(buildUDPv4(1000, 53, nil)); err != nil { // zero-length + t.Fatal(err) + } + if err := c.Commit(buildUDPv4(1000, 53, full)); err != nil { + t.Fatal(err) + } + if err := c.Flush(); err != nil { + t.Fatal(err) + } + // The empty datagram sealed the first slot, so the trailing full packet + // can't join it: two single-segment superpackets bracket one plain write. + if len(w.gsoWrites) != 2 || len(w.writes) != 1 { + t.Fatalf("want 2 gso writes + 1 plain, got gso=%d plain=%d", len(w.gsoWrites), len(w.writes)) + } +} + // IPv4 with options is not admissible (we require IHL=5). func TestUDPCoalescerIPv4WithOptionsPassesThrough(t *testing.T) { w := &fakeTunWriter{gsoEnabled: true} diff --git a/overlay/tio/queueset_gso_linux.go b/overlay/tio/queueset_gso_linux.go index dc4194e8..2fdb374d 100644 --- a/overlay/tio/queueset_gso_linux.go +++ b/overlay/tio/queueset_gso_linux.go @@ -7,6 +7,7 @@ import ( "encoding/binary" "errors" "fmt" + "sync/atomic" "golang.org/x/sys/unix" ) @@ -20,6 +21,7 @@ type offloadQueueSet struct { // with the kernel. Queues created by Add inherit this and surface it // via Offload.USOSupported so coalescers can gate USO emission. usoEnabled bool + closed atomic.Bool } // NewOffloadQueueSet creates a QueueSet that uses virtio_net_hdr to do @@ -65,18 +67,33 @@ func (c *offloadQueueSet) wakeForShutdown() error { } func (c *offloadQueueSet) Close() error { + if c.closed.Swap(true) { + return nil + } + errs := []error{} - // Signal all readers blocked in poll to wake up and exit + // Signal all readers blocked in poll to wake up and exit. They observe + // POLLIN on the shutdown eventfd and return os.ErrClosed. if err := c.wakeForShutdown(); err != nil { errs = append(errs, err) } + // Close the per-queue tun fds; this also unblocks any in-flight reads. + // The per-queue Close deliberately leaves shutdownFd alone - it belongs + // to this container. for _, x := range c.pq { if err := x.Close(); err != nil { errs = append(errs, err) } } + // Close the shutdown eventfd last: every reader's pollfd set references + // it, so it must outlive the wake + per-queue teardown above. + if err := unix.Close(c.shutdownFd); err != nil { + errs = append(errs, err) + } + c.shutdownFd = -1 + return errors.Join(errs...) } diff --git a/overlay/tio/queueset_poll_linux.go b/overlay/tio/queueset_poll_linux.go index f9d785f6..da97c09e 100644 --- a/overlay/tio/queueset_poll_linux.go +++ b/overlay/tio/queueset_poll_linux.go @@ -7,6 +7,7 @@ import ( "encoding/binary" "errors" "fmt" + "sync/atomic" "golang.org/x/sys/unix" ) @@ -16,6 +17,7 @@ type pollQueueSet struct { // pqi is exactly the same as pq, but stored as the interface type pqi []Queue shutdownFd int + closed atomic.Bool } func NewPollQueueSet() (QueueSet, error) { @@ -56,17 +58,33 @@ func (c *pollQueueSet) wakeForShutdown() error { } func (c *pollQueueSet) Close() error { + if c.closed.Swap(true) { + return nil + } + errs := []error{} + // Wake any reader blocked in poll so it observes POLLIN on the shutdown + // eventfd and returns os.ErrClosed. if err := c.wakeForShutdown(); err != nil { errs = append(errs, err) } + // Close the per-queue tun fds; this also unblocks any in-flight reads. + // The per-queue Close deliberately leaves shutdownFd alone - it belongs + // to this container. for _, x := range c.pq { if err := x.Close(); err != nil { errs = append(errs, err) } } + // Close the shutdown eventfd last: every reader's pollfd set references + // it, so it must outlive the wake + per-queue teardown above. + if err := unix.Close(c.shutdownFd); err != nil { + errs = append(errs, err) + } + c.shutdownFd = -1 + return errors.Join(errs...) } diff --git a/overlay/tio/tio_gso_linux.go b/overlay/tio/tio_gso_linux.go index 3d1819e9..beb4952b 100644 --- a/overlay/tio/tio_gso_linux.go +++ b/overlay/tio/tio_gso_linux.go @@ -439,10 +439,22 @@ func (r *Offload) WriteGSO(hdr []byte, transportHdr []byte, pays [][]byte, proto r.gsoIovs[1].SetLen(len(hdr)) r.gsoIovs[2].Base = &transportHdr[0] r.gsoIovs[2].SetLen(len(transportHdr)) - for i, p := range pays { - r.gsoIovs[3+i].Base = &p[0] - r.gsoIovs[3+i].SetLen(len(p)) + // Defense in depth: an empty payload fragment can't be a valid GSO + // segment and &p[0] would panic on it. Callers route zero-length + // datagrams through the plain path (see UDPCoalescer.commitParsed), so + // this should never fire, but skip empties rather than index into one. + // `n` tracks where the next payload iovec lands, since skips make it + // drift from 3+i. + n := 3 + for _, p := range pays { + if len(p) == 0 { + continue + } + r.gsoIovs[n].Base = &p[0] + r.gsoIovs[n].SetLen(len(p)) + n++ } + r.gsoIovs = r.gsoIovs[:n] _, err := r.rawWrite(r.gsoIovs) return err @@ -454,11 +466,9 @@ func (r *Offload) Close() error { } //shutdownFd is owned by the container, so we should not close it - var err error - if r.fd >= 0 { - err = unix.Close(r.fd) - r.fd = -1 - } - - return err + // Close the underlying fd but do NOT null r.fd: a reader may still be + // loading it in readOne, and mutating the field would race that load. + // It gets EBADF -> os.ErrClosed (or wakes via the shutdown eventfd's + // ppoll first). closed.Swap already guarantees we only close once. + return unix.Close(r.fd) } diff --git a/overlay/tio/tio_poll_linux.go b/overlay/tio/tio_poll_linux.go index 410dc418..975eff1f 100644 --- a/overlay/tio/tio_poll_linux.go +++ b/overlay/tio/tio_poll_linux.go @@ -27,9 +27,12 @@ type Poll struct { batchRet [1]Packet } +// newPoll wraps an existing tun fd. On failure it does NOT close fd: the +// caller owns fd and is the sole closer (see pollQueueSet.Add callers in +// overlay/tun_linux.go, which unix.Close on Add error). This matches the +// newOffload convention and keeps closes at exactly one on every path. func newPoll(fd int, shutdownFd int) (*Poll, error) { if err := unix.SetNonblock(fd, true); err != nil { - _ = unix.Close(fd) return nil, fmt.Errorf("failed to set Poll device as nonblocking: %w", err) } @@ -157,11 +160,9 @@ func (t *Poll) Close() error { return nil } //shutdownFd is owned by the container, so we should not close it - var err error - if t.fd >= 0 { - err = unix.Close(t.fd) - t.fd = -1 - } - - return err + // Close the underlying fd but do NOT null t.fd: a reader may still be + // loading it in readOne, and mutating the field would race that load. + // It gets EBADF -> os.ErrClosed (or wakes via the shutdown eventfd's + // ppoll first). closed.Swap already guarantees we only close once. + return unix.Close(t.fd) } diff --git a/overlay/tio/tun_file_linux_test.go b/overlay/tio/tun_file_linux_test.go index f92f58ec..d72202ea 100644 --- a/overlay/tio/tun_file_linux_test.go +++ b/overlay/tio/tun_file_linux_test.go @@ -70,6 +70,27 @@ func TestPoll_WakeForShutdown_WakesFriends(t *testing.T) { } } +// TestPoll_NewPoll_DoesNotCloseFdOnFailure pins the ownership rule: when +// newPoll fails, it must leave fd open so the caller (pollQueueSet.Add's +// callers in tun_linux.go) is the sole closer. If newPoll also closed fd, +// the poll path would double-close on Add error. We force the failure with +// an O_PATH descriptor: fcntl(F_SETFL) — which SetNonblock performs — is not +// permitted on O_PATH fds and fails with EBADF, while the fd itself stays +// open so we can observe that newPoll left it alone. +func TestPoll_NewPoll_DoesNotCloseFdOnFailure(t *testing.T) { + fd, err := unix.Open("/", unix.O_PATH|unix.O_CLOEXEC, 0) + require.NoError(t, err) + t.Cleanup(func() { _ = unix.Close(fd) }) + + p, err := newPoll(fd, 1) + require.Error(t, err, "SetNonblock on an O_PATH fd should fail") + require.Nil(t, p) + + // If newPoll had closed fd, F_GETFD would report it closed. It staying + // open proves newPoll left the fd for the caller to close exactly once. + require.True(t, fdOpen(t, fd), "newPoll must not close fd on failure; caller is the sole closer") +} + func TestPoll_Close_Idempotent(t *testing.T) { tf, err := newPoll(newReadPipe(t), 1) require.NoError(t, err) @@ -80,3 +101,57 @@ func TestPoll_Close_Idempotent(t *testing.T) { t.Fatalf("second Close should be a no-op, got %v", err) } } + +// fdOpen reports whether fd currently refers to an open file description. +// A closed (or never-allocated) fd makes F_GETFD fail with EBADF. +func fdOpen(t *testing.T, fd int) bool { + t.Helper() + _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0) + if err == nil { + return true + } + if errors.Is(err, unix.EBADF) { + return false + } + t.Fatalf("unexpected fcntl(F_GETFD) error on fd %d: %v", fd, err) + return false +} + +// TestPollQueueSet_Close_ClosesShutdownFd is the regression test for the +// leaked shutdown eventfd: the container that owns shutdownFd must close it in +// Close, and a second Close must be a safe no-op. +func TestPollQueueSet_Close_ClosesShutdownFd(t *testing.T) { + qs, err := NewPollQueueSet() + require.NoError(t, err) + c, ok := qs.(*pollQueueSet) + require.True(t, ok) + require.NoError(t, qs.Add(newReadPipe(t))) + + shutdownFd := c.shutdownFd + require.True(t, fdOpen(t, shutdownFd), "shutdown eventfd should be open before Close") + + require.NoError(t, qs.Close()) + require.False(t, fdOpen(t, shutdownFd), "shutdown eventfd should be closed after Close") + + // Second Close must not touch fds (shutdownFd is now -1) and must return nil. + require.NoError(t, qs.Close()) +} + +// TestOffloadQueueSet_Close_ClosesShutdownFd mirrors the poll regression test +// for the GSO/offload queueset. +func TestOffloadQueueSet_Close_ClosesShutdownFd(t *testing.T) { + qs, err := NewOffloadQueueSet(false) + require.NoError(t, err) + c, ok := qs.(*offloadQueueSet) + require.True(t, ok) + require.NoError(t, qs.Add(newReadPipe(t))) + + shutdownFd := c.shutdownFd + require.True(t, fdOpen(t, shutdownFd), "shutdown eventfd should be open before Close") + + require.NoError(t, qs.Close()) + require.False(t, fdOpen(t, shutdownFd), "shutdown eventfd should be closed after Close") + + // Second Close must not touch fds (shutdownFd is now -1) and must return nil. + require.NoError(t, qs.Close()) +} diff --git a/overlay/tio/tun_linux_offload_test.go b/overlay/tio/tun_linux_offload_test.go index 1cf64925..48a7df8a 100644 --- a/overlay/tio/tun_linux_offload_test.go +++ b/overlay/tio/tun_linux_offload_test.go @@ -640,6 +640,38 @@ func TestTunFileWriteVnetHdrNoAlloc(t *testing.T) { } } +// TestWriteGSOSkipsEmptyPayloads is the defense-in-depth guard for the +// zero-length UDP DoS: a payload fragment of length zero would make &p[0] +// panic (index-out-of-range) when building the iovec array. WriteGSO must +// skip empties instead. We write to /dev/null so the writev always succeeds +// synchronously; the point is simply that neither call panics. +func TestWriteGSOSkipsEmptyPayloads(t *testing.T) { + fd, err := unix.Open("/dev/null", os.O_WRONLY, 0) + if err != nil { + t.Fatalf("open /dev/null: %v", err) + } + t.Cleanup(func() { _ = unix.Close(fd) }) + + o := &Offload{fd: fd, gsoIovs: make([]unix.Iovec, 2, gsoMaxIovs)} + o.gsoIovs[0].Base = &o.gsoHdrBuf[0] + o.gsoIovs[0].SetLen(virtio.Size) + + ipHdr := make([]byte, 20) + ipHdr[0] = 0x45 // IPv4, IHL 5 + udpHdr := make([]byte, 8) + + // Sole payload empty: exercises the all-empty skip (n stays at 3). + if err := o.WriteGSO(ipHdr, udpHdr, [][]byte{{}}, GSOProtoUDP); err != nil { + t.Fatalf("WriteGSO with a single empty payload: %v", err) + } + // Empty mixed with a real fragment: exercises the index-drift skip so a + // later non-empty payload still lands in the right iovec slot. + real := make([]byte, 1200) + if err := o.WriteGSO(ipHdr, udpHdr, [][]byte{real, {}}, GSOProtoUDP); err != nil { + t.Fatalf("WriteGSO with a trailing empty payload: %v", err) + } +} + // buildTSOv6 builds a synthetic IPv6/TCP TSO superpacket with payLen bytes // of payload, segmented at gso. Returns the packet bytes only; the // virtio_net_hdr is the caller's responsibility. diff --git a/overlay/tio/virtio/segment_linux.go b/overlay/tio/virtio/segment_linux.go index afc545bf..a5666f6e 100644 --- a/overlay/tio/virtio/segment_linux.go +++ b/overlay/tio/virtio/segment_linux.go @@ -27,6 +27,13 @@ const ( tcpHeaderMaxLen = 60 // data-offset=15, max options ) +// maxSegHdrLen bounds the L3+L4 header we snapshot before stamping each +// segment. The largest header the segmenter supports is IPv4 (max IHL 60) +// plus TCP (max data-offset 60) = 120 bytes; the array is sized to that +// worst case so the snapshot lives on the stack with no per-call heap +// allocation. +const maxSegHdrLen = ipv4HeaderMaxLen + tcpHeaderMaxLen // 120 + // Byte offsets inside an IPv4 header. const ( ipv4TotalLenOff = 2 @@ -144,13 +151,18 @@ func CorrectHdrLen(pkt []byte, hdr *Hdr) error { } // SegmentTCP walks a TSO superpacket pkt, yielding each segment as a -// slice into pkt itself. Per-segment plaintext is laid out by sliding a -// freshly-patched copy of the L3+L4 header into pkt at offset i*gsoSize, -// where it sits immediately before that segment's payload chunk in the -// original buffer. The slide is destructive: iter i's header write overwrites -// the last hdrLen bytes of seg_{i-1}'s payload, which is dead by the time -// the next iteration begins. pkt is consumed by this call and must not be -// inspected by the caller after the final yield. +// slice into pkt itself. Per-segment plaintext is laid out by stamping a +// copy of the original L3+L4 header into pkt at offset i*gsoSize, where it +// sits immediately before that segment's payload chunk in the original +// buffer. The stamp is destructive but harmless: iter i's header write lands +// on pkt[i*G : i*G+hdrLen], which is the tail of seg_{i-1}'s payload (already +// consumed) and ends exactly where seg_i's payload begins, so it never clobbers +// live payload — this holds even when gsoSize < hdrLen. The header bytes are +// sourced from a pristine snapshot taken before the loop (savedHdr), NOT from +// pkt[:hdrLen], because when gsoSize < hdrLen the stamps would otherwise +// overwrite the leading header in place and every stamp after the first would +// copy corrupted bytes. pkt is consumed by this call and must not be inspected +// by the caller after the final yield. func SegmentTCP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg []byte) error) error { if gsoSizeU == 0 { return fmt.Errorf("gso_size is zero") @@ -161,6 +173,9 @@ func SegmentTCP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg headerLen := int(hdrLenU) csumStart := int(csumStartU) + if headerLen > maxSegHdrLen { + return fmt.Errorf("header len %d exceeds max %d", headerLen, maxSegHdrLen) + } isV4 := pkt[0]>>4 == 4 tcpHdrLen := int(pkt[csumStart+tcpDataOffOff]>>4) * 4 @@ -205,6 +220,13 @@ func SegmentTCP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg baseIPHdrSum = uint32(checksum.Checksum(ipTmp[:ihl], 0)) } + // Snapshot the pristine L3+L4 header once. Every segment's header is + // stamped from this copy, so overlapping stamps (gsoSize < headerLen) + // can never corrupt the source. The variable fields (seq/flags/cksum/ + // totalLen/id) captured here are stale but are overwritten per segment. + var savedHdr [maxSegHdrLen]byte + copy(savedHdr[:headerLen], pkt[:headerLen]) + for i := 0; i < numSeg; i++ { segStart := i * gsoSize segEnd := segStart + gsoSize @@ -215,14 +237,13 @@ func SegmentTCP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg segLen := headerLen + segPayLen headerOff := i * gsoSize - // Slide the header into place immediately before this segment's - // payload. Iter 0's header is already at pkt[:headerLen]; for - // i ≥ 1 we copy from there. The constant-byte fields of pkt[:headerLen] - // survive iter 0's in-place patches (only seq/flags/cksum/totalLen/id - // are touched), and iter 0's stale variable-field values are - // overwritten by the per-segment patches below. + // Stamp the header into place immediately before this segment's + // payload, sourced from the pristine snapshot. Iter 0's header is + // already at pkt[:headerLen] (identical to savedHdr), so only i ≥ 1 + // needs the stamp. The per-segment patches below overwrite the + // variable fields. if i > 0 { - copy(pkt[headerOff:headerOff+headerLen], pkt[:headerLen]) + copy(pkt[headerOff:headerOff+headerLen], savedHdr[:headerLen]) } seg := pkt[headerOff : headerOff+segLen] @@ -269,11 +290,13 @@ func SegmentTCP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg return nil } -// SegmentUDP walks a USO superpacket, sliding a per-segment-patched -// L3+L4 header into pkt at offset i*gsoSize and yielding pkt[i*G:i*G+segLen] -// to the caller. Per-segment patches are total_len + IPv4 csum (or IPv6 -// payload_len) plus the UDP length and checksum. pkt is consumed -// destructively; see SegmentTCP for the layout reasoning. +// SegmentUDP walks a USO superpacket, stamping a per-segment-patched copy of +// the original L3+L4 header into pkt at offset i*gsoSize and yielding +// pkt[i*G:i*G+segLen] to the caller. Per-segment patches are total_len + +// IPv4 csum (or IPv6 payload_len) plus the UDP length and checksum. pkt is +// consumed destructively; see SegmentTCP for the layout reasoning, including +// why the header is stamped from a pristine snapshot rather than pkt[:hdrLen] +// (correctness when gsoSize < hdrLen). // // UDP-GSO leaves the IPv4 ID identical across segments (the kernel does not // bump it), which is why the IP-level per-segment work is limited to @@ -289,6 +312,9 @@ func SegmentUDP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg isV4 := pkt[0]>>4 == 4 headerLen := int(hdrLenU) csumStart := int(csumStartU) + if headerLen > maxSegHdrLen { + return fmt.Errorf("header len %d exceeds max %d", headerLen, maxSegHdrLen) + } if headerLen-csumStart != udpHeaderLen { return fmt.Errorf("udp header len mismatch: %d", headerLen-csumStart) } @@ -327,6 +353,12 @@ func SegmentUDP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg baseIPHdrSum = uint32(checksum.Checksum(ipTmp[:ihl], 0)) } + // Snapshot the pristine L3+L4 header once and stamp every segment from + // it; see SegmentTCP for why sourcing from pkt[:headerLen] corrupts + // segments when gsoSize < headerLen. + var savedHdr [maxSegHdrLen]byte + copy(savedHdr[:headerLen], pkt[:headerLen]) + for i := 0; i < numSeg; i++ { segStart := i * gsoSize segEnd := segStart + gsoSize @@ -338,7 +370,7 @@ func SegmentUDP(pkt []byte, hdrLenU, csumStartU, gsoSizeU uint16, yield func(seg headerOff := i * gsoSize if i > 0 { - copy(pkt[headerOff:headerOff+headerLen], pkt[:headerLen]) + copy(pkt[headerOff:headerOff+headerLen], savedHdr[:headerLen]) } seg := pkt[headerOff : headerOff+segLen] diff --git a/overlay/tio/virtio/segment_linux_test.go b/overlay/tio/virtio/segment_linux_test.go new file mode 100644 index 00000000..6ba5f2e6 --- /dev/null +++ b/overlay/tio/virtio/segment_linux_test.go @@ -0,0 +1,286 @@ +//go:build linux && !android +// +build linux,!android + +package virtio + +import ( + "bytes" + "encoding/binary" + "testing" + + "golang.org/x/sys/unix" + + "github.com/slackhq/nebula/overlay/checksum" +) + +// verifyChecksum confirms that the one's-complement sum across b, seeded with +// a folded pseudo-header sum, equals all-ones (a valid on-wire checksum). +// A corrupted header stamped into a segment makes this fail even when the +// checksum field itself was computed from the (pristine) base sums, because +// the bytes the receiver would sum no longer match what was checksummed. +func verifyChecksum(b []byte, pseudo uint16) bool { + return checksum.Checksum(b, pseudo) == 0xffff +} + +// pseudoHeaderIPv4 folds the TCP/UDP pseudo-header sum from a segment's own +// address and length fields, used to independently verify its L4 checksum. +func pseudoHeaderIPv4(src, dst []byte, proto byte, l4Len int) uint16 { + s := uint32(checksum.Checksum(src, 0)) + uint32(checksum.Checksum(dst, 0)) + s += uint32(proto) + uint32(l4Len) + s = (s & 0xffff) + (s >> 16) + s = (s & 0xffff) + (s >> 16) + return uint16(s) +} + +// buildTCPv4Super constructs a synthetic IPv4/TCP TSO superpacket with a +// payload of payLen bytes and returns it alongside the header fields the +// segmenter needs. The header is a fixed 40 bytes (20 IPv4 + 20 TCP). +func buildTCPv4Super(payLen int) (pkt []byte, hdrLen, csumStart uint16) { + const ipLen = 20 + const tcpLen = 20 + pkt = make([]byte, ipLen+tcpLen+payLen) + + // IPv4 header. + pkt[0] = 0x45 // version 4, IHL 5 + binary.BigEndian.PutUint16(pkt[2:4], uint16(ipLen+tcpLen+payLen)) + binary.BigEndian.PutUint16(pkt[4:6], 0x4242) // ID + pkt[8] = 64 // TTL + pkt[9] = unix.IPPROTO_TCP + copy(pkt[12:16], []byte{10, 0, 0, 1}) // src + copy(pkt[16:20], []byte{10, 0, 0, 2}) // dst + + // TCP header. + binary.BigEndian.PutUint16(pkt[20:22], 12345) // sport + binary.BigEndian.PutUint16(pkt[22:24], 80) // dport + binary.BigEndian.PutUint32(pkt[24:28], 10000) // seq + binary.BigEndian.PutUint32(pkt[28:32], 20000) // ack + pkt[32] = 0x50 // data offset 5 words + pkt[33] = 0x18 // ACK | PSH + binary.BigEndian.PutUint16(pkt[34:36], 65535) // window + + for i := 0; i < payLen; i++ { + pkt[ipLen+tcpLen+i] = byte(i & 0xff) + } + return pkt, ipLen + tcpLen, ipLen +} + +// buildUDPv4Super constructs a synthetic IPv4/UDP USO superpacket with a +// payload of payLen bytes. Header is a fixed 28 bytes (20 IPv4 + 8 UDP). +func buildUDPv4Super(payLen int) (pkt []byte, hdrLen, csumStart uint16) { + const ipLen = 20 + const udpLen = 8 + pkt = make([]byte, ipLen+udpLen+payLen) + + pkt[0] = 0x45 + binary.BigEndian.PutUint16(pkt[2:4], uint16(ipLen+udpLen+payLen)) + binary.BigEndian.PutUint16(pkt[4:6], 0x4242) + pkt[8] = 64 + pkt[9] = unix.IPPROTO_UDP + copy(pkt[12:16], []byte{10, 0, 0, 1}) + copy(pkt[16:20], []byte{10, 0, 0, 2}) + + binary.BigEndian.PutUint16(pkt[20:22], 12345) // sport + binary.BigEndian.PutUint16(pkt[22:24], 53) // dport + + for i := 0; i < payLen; i++ { + pkt[ipLen+udpLen+i] = byte(i & 0xff) + } + return pkt, ipLen + udpLen, ipLen +} + +// collectTCP segments a fresh copy of pkt and returns each segment as an +// independent slice so assertions can run after segmentation completes. +func collectTCP(t *testing.T, pkt []byte, hdrLen, csumStart, gsoSize uint16) [][]byte { + t.Helper() + work := append([]byte(nil), pkt...) + var out [][]byte + err := SegmentTCP(work, hdrLen, csumStart, gsoSize, func(seg []byte) error { + out = append(out, append([]byte(nil), seg...)) + return nil + }) + if err != nil { + t.Fatalf("SegmentTCP: %v", err) + } + return out +} + +func collectUDP(t *testing.T, pkt []byte, hdrLen, csumStart, gsoSize uint16) [][]byte { + t.Helper() + work := append([]byte(nil), pkt...) + var out [][]byte + err := SegmentUDP(work, hdrLen, csumStart, gsoSize, func(seg []byte) error { + out = append(out, append([]byte(nil), seg...)) + return nil + }) + if err != nil { + t.Fatalf("SegmentUDP: %v", err) + } + return out +} + +// TestSegmentTCPHeaderNotCorrupted is the regression test for the in-place +// header-slide bug: when gsoSize < headerLen the old code stamped each +// segment's header from pkt[:headerLen], which had already been overwritten +// by the previous segment's overlapping stamp, so segments 2..n carried a +// corrupted header (garbage src/dst/ports/seq). Every segment must instead +// carry the ORIGINAL constant header fields with correct per-segment seq. +func TestSegmentTCPHeaderNotCorrupted(t *testing.T) { + const origSeq = 10000 + cases := []struct { + name string + payLen int + gsoSize uint16 + }{ + // gsoSize (8) < headerLen (40): the bug's trigger. Even split. + {"small-gso-even", 40, 8}, + // gsoSize (8) < headerLen (40) with a short final segment. + {"small-gso-odd-tail", 44, 8}, + // gsoSize (100) >= headerLen (40): the normal path, must still work. + {"normal-gso", 250, 100}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pkt, hdrLen, csumStart := buildTCPv4Super(tc.payLen) + gso := int(tc.gsoSize) + wantSeg := (tc.payLen + gso - 1) / gso + segs := collectTCP(t, pkt, hdrLen, csumStart, tc.gsoSize) + if len(segs) != wantSeg { + t.Fatalf("got %d segments, want %d", len(segs), wantSeg) + } + + off := 0 + for i, seg := range segs { + // Constant header fields must be identical to the original in + // EVERY segment. These are exactly the bytes the old code + // corrupted in segments 2..n. + if got := seg[0]; got != 0x45 { + t.Errorf("seg %d: version/IHL byte=%#x want 0x45", i, got) + } + if seg[9] != unix.IPPROTO_TCP { + t.Errorf("seg %d: proto=%d want %d", i, seg[9], unix.IPPROTO_TCP) + } + if !bytes.Equal(seg[12:16], []byte{10, 0, 0, 1}) { + t.Errorf("seg %d: src=%v want [10 0 0 1]", i, seg[12:16]) + } + if !bytes.Equal(seg[16:20], []byte{10, 0, 0, 2}) { + t.Errorf("seg %d: dst=%v want [10 0 0 2]", i, seg[16:20]) + } + if sport := binary.BigEndian.Uint16(seg[20:22]); sport != 12345 { + t.Errorf("seg %d: sport=%d want 12345", i, sport) + } + if dport := binary.BigEndian.Uint16(seg[22:24]); dport != 80 { + t.Errorf("seg %d: dport=%d want 80", i, dport) + } + if ack := binary.BigEndian.Uint32(seg[28:32]); ack != 20000 { + t.Errorf("seg %d: ack=%d want 20000", i, ack) + } + if seg[32] != 0x50 { + t.Errorf("seg %d: data-offset byte=%#x want 0x50", i, seg[32]) + } + + // Per-segment seq must advance by the payload offset. + segStart := i * gso + if seq := binary.BigEndian.Uint32(seg[24:28]); seq != uint32(origSeq+segStart) { + t.Errorf("seg %d: seq=%d want %d", i, seq, origSeq+segStart) + } + + // Payload bytes must be the original contiguous slice. + segPayLen := len(seg) - int(hdrLen) + wantPay := make([]byte, segPayLen) + for k := 0; k < segPayLen; k++ { + wantPay[k] = byte((off + k) & 0xff) + } + if !bytes.Equal(seg[hdrLen:], wantPay) { + t.Errorf("seg %d: payload mismatch", i) + } + off += segPayLen + + // End-to-end: the stamped header must checksum-verify. A + // corrupted header fails here because the written checksum was + // derived from the pristine header. + if !verifyChecksum(seg[:20], 0) { + t.Errorf("seg %d: bad IPv4 header checksum", i) + } + psum := pseudoHeaderIPv4(seg[12:16], seg[16:20], unix.IPPROTO_TCP, len(seg)-20) + if !verifyChecksum(seg[20:], psum) { + t.Errorf("seg %d: bad TCP checksum", i) + } + } + }) + } +} + +// TestSegmentUDPHeaderNotCorrupted is the USO counterpart: SegmentUDP performs +// the same header stamp and must be correct when gsoSize < headerLen. +func TestSegmentUDPHeaderNotCorrupted(t *testing.T) { + cases := []struct { + name string + payLen int + gsoSize uint16 + }{ + {"small-gso-even", 40, 8}, + {"small-gso-odd-tail", 44, 8}, + {"normal-gso", 250, 100}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + pkt, hdrLen, csumStart := buildUDPv4Super(tc.payLen) + gso := int(tc.gsoSize) + wantSeg := (tc.payLen + gso - 1) / gso + segs := collectUDP(t, pkt, hdrLen, csumStart, tc.gsoSize) + if len(segs) != wantSeg { + t.Fatalf("got %d segments, want %d", len(segs), wantSeg) + } + + off := 0 + for i, seg := range segs { + if got := seg[0]; got != 0x45 { + t.Errorf("seg %d: version/IHL byte=%#x want 0x45", i, got) + } + if seg[9] != unix.IPPROTO_UDP { + t.Errorf("seg %d: proto=%d want %d", i, seg[9], unix.IPPROTO_UDP) + } + if !bytes.Equal(seg[12:16], []byte{10, 0, 0, 1}) { + t.Errorf("seg %d: src=%v want [10 0 0 1]", i, seg[12:16]) + } + if !bytes.Equal(seg[16:20], []byte{10, 0, 0, 2}) { + t.Errorf("seg %d: dst=%v want [10 0 0 2]", i, seg[16:20]) + } + if sport := binary.BigEndian.Uint16(seg[20:22]); sport != 12345 { + t.Errorf("seg %d: sport=%d want 12345", i, sport) + } + if dport := binary.BigEndian.Uint16(seg[22:24]); dport != 53 { + t.Errorf("seg %d: dport=%d want 53", i, dport) + } + // UDP-GSO keeps the same IPv4 ID across every segment. + if id := binary.BigEndian.Uint16(seg[4:6]); id != 0x4242 { + t.Errorf("seg %d: ip id=%#x want 0x4242", i, id) + } + + segPayLen := len(seg) - int(hdrLen) + if udpLen := binary.BigEndian.Uint16(seg[24:26]); udpLen != uint16(8+segPayLen) { + t.Errorf("seg %d: udp len=%d want %d", i, udpLen, 8+segPayLen) + } + + wantPay := make([]byte, segPayLen) + for k := 0; k < segPayLen; k++ { + wantPay[k] = byte((off + k) & 0xff) + } + if !bytes.Equal(seg[hdrLen:], wantPay) { + t.Errorf("seg %d: payload mismatch", i) + } + off += segPayLen + + if !verifyChecksum(seg[:20], 0) { + t.Errorf("seg %d: bad IPv4 header checksum", i) + } + psum := pseudoHeaderIPv4(seg[12:16], seg[16:20], unix.IPPROTO_UDP, len(seg)-20) + if !verifyChecksum(seg[20:], psum) { + t.Errorf("seg %d: bad UDP checksum", i) + } + } + }) + } +} diff --git a/overlay/tun_linux.go b/overlay/tun_linux.go index c18fc38e..1f643f88 100644 --- a/overlay/tun_linux.go +++ b/overlay/tun_linux.go @@ -35,6 +35,15 @@ type tun struct { deviceIndex int ioctlFd uintptr vnetHdr bool + // offloadFlags is the exact TUN_F_* offload mask newTun negotiated with + // the kernel: usoOffloadFlags when USO was accepted, tsoOffloadFlags on + // the TSO-only fallback, or 0 when vnetHdr is off. TUNSETOFFLOAD is + // device-wide (drivers/net/tun.c set_offload updates tun->set_features + // for the whole netdev), so NewMultiQueueReader must replay this exact + // mask on every added queue — issuing a narrower mask there would + // silently downgrade offloads (e.g. disable USO) for all queues while + // they still advertise the stale capability. + offloadFlags uint // routeFeatureECN, when true, sets RTAX_FEATURE_ECN on every route we // install for the tun. The kernel then actively negotiates ECN for // connections destined to those prefixes (equivalent to `ip route @@ -82,7 +91,7 @@ type ifreqQLEN struct { func newTunFromFd(c *config.C, l *slog.Logger, deviceFd int, vpnNetworks []netip.Prefix) (*tun, error) { // We don't know what flags the caller opened this fd with and can't turn // on IFF_VNET_HDR after TUNSETIFF, so skip offload on inherited fds. - t, err := newTunGeneric(c, l, deviceFd, false, false, vpnNetworks) + t, err := newTunGeneric(c, l, deviceFd, false, 0, vpnNetworks) if err != nil { return nil, err } @@ -139,6 +148,14 @@ const tsoOffloadFlags = unix.TUN_F_CSUM | unix.TUN_F_TSO4 | unix.TUN_F_TSO6 | un // tsoOffloadFlags. const usoOffloadFlags = tsoOffloadFlags | unix.TUN_F_USO4 | unix.TUN_F_USO6 +// offloadUSOEnabled reports whether the negotiated offload mask includes UDP +// Segmentation Offload. It is the single source of truth for the usoEnabled +// capability surfaced by each queue, so the mask stored on the tun and the USO +// bit reported to coalescers can never drift apart. +func offloadUSOEnabled(offloadFlags uint) bool { + return offloadFlags&(unix.TUN_F_USO4|unix.TUN_F_USO6) != 0 +} + func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue bool) (*tun, error) { baseFlags := uint16(unix.IFF_TUN | unix.IFF_NO_PI) if multiqueue { @@ -156,7 +173,10 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue return nil, err } vnetHdr := true - usoEnabled := false + // offloadFlags is the exact TUN_F_* mask the kernel accepted. We remember + // it (rather than a plain bool) so NewMultiQueueReader can replay the + // identical device-wide mask on added queues instead of downgrading them. + var offloadFlags uint name, err := tunSetIff(fd, nameStr, baseFlags|unix.IFF_VNET_HDR) if err != nil { _ = unix.Close(fd) @@ -166,8 +186,10 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue // the ioctl returns EINVAL; fall back to the TCP-only mask before // giving up on VNET_HDR entirely. if err = ioctl(uintptr(fd), unix.TUNSETOFFLOAD, uintptr(usoOffloadFlags)); err == nil { - usoEnabled = true - } else if err = ioctl(uintptr(fd), unix.TUNSETOFFLOAD, uintptr(tsoOffloadFlags)); err != nil { + offloadFlags = usoOffloadFlags + } else if err = ioctl(uintptr(fd), unix.TUNSETOFFLOAD, uintptr(tsoOffloadFlags)); err == nil { + offloadFlags = tsoOffloadFlags + } else { l.Warn("Failed to enable TUN offload (TSO); proceeding without virtio headers", "error", err) _ = unix.Close(fd) vnetHdr = false @@ -187,10 +209,10 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue } if vnetHdr { - l.Info("TUN offload enabled", "tso", true, "uso", usoEnabled) + l.Info("TUN offload enabled", "tso", true, "uso", offloadUSOEnabled(offloadFlags)) } - t, err := newTunGeneric(c, l, fd, vnetHdr, usoEnabled, vpnNetworks) + t, err := newTunGeneric(c, l, fd, vnetHdr, offloadFlags, vpnNetworks) if err != nil { return nil, err } @@ -200,12 +222,16 @@ func newTun(c *config.C, l *slog.Logger, vpnNetworks []netip.Prefix, multiqueue return t, nil } -// newTunGeneric does all the stuff common to different tun initialization paths. It will close your files on error. -func newTunGeneric(c *config.C, l *slog.Logger, fd int, vnetHdr, usoEnabled bool, vpnNetworks []netip.Prefix) (*tun, error) { +// newTunGeneric does all the stuff common to different tun initialization +// paths. It will close your files on error. offloadFlags is the TUN_F_* mask +// newTun negotiated (0 when vnetHdr is off); the queues' USO capability is +// derived from it so it can never disagree with the mask we replay on added +// multiqueue readers. +func newTunGeneric(c *config.C, l *slog.Logger, fd int, vnetHdr bool, offloadFlags uint, vpnNetworks []netip.Prefix) (*tun, error) { var qs tio.QueueSet var err error if vnetHdr { - qs, err = tio.NewOffloadQueueSet(usoEnabled) + qs, err = tio.NewOffloadQueueSet(offloadUSOEnabled(offloadFlags)) } else { qs, err = tio.NewPollQueueSet() } @@ -224,6 +250,7 @@ func newTunGeneric(c *config.C, l *slog.Logger, fd int, vnetHdr, usoEnabled bool readers: qs, closeLock: sync.Mutex{}, vnetHdr: vnetHdr, + offloadFlags: offloadFlags, vpnNetworks: vpnNetworks, TXQueueLen: c.GetInt("tun.tx_queue", 500), useSystemRoutes: c.GetBool("tun.use_system_route_table", false), @@ -345,7 +372,11 @@ func (t *tun) NewMultiQueueReader() error { } if t.vnetHdr { - if err = ioctl(uintptr(fd), unix.TUNSETOFFLOAD, uintptr(tsoOffloadFlags)); err != nil { + // Replay the exact mask newTun negotiated. TUNSETOFFLOAD is + // device-wide, so issuing the TSO-only mask here would disable USO + // for every queue (including queue 0) on kernels where newTun + // successfully enabled it, while the queues keep advertising USO. + if err = ioctl(uintptr(fd), unix.TUNSETOFFLOAD, uintptr(t.offloadFlags)); err != nil { _ = unix.Close(fd) return fmt.Errorf("failed to enable offload on multiqueue tun fd: %w", err) } diff --git a/overlay/tun_linux_test.go b/overlay/tun_linux_test.go index 1003a165..401d5a07 100644 --- a/overlay/tun_linux_test.go +++ b/overlay/tun_linux_test.go @@ -34,3 +34,66 @@ func TestTunAdvMSS(t *testing.T) { }) } } + +// TestOffloadUSOEnabled pins the single source of truth for the per-queue USO +// capability: it is derived from the negotiated offload mask, so the mask +// stored on the tun and the capability reported to coalescers cannot drift. +func TestOffloadUSOEnabled(t *testing.T) { + // usoOffloadFlags must be a strict superset of tsoOffloadFlags. Otherwise + // the TSO-only fallback (and the historic hardcoded-mask bug in + // NewMultiQueueReader) would not actually be a downgrade. + if usoOffloadFlags&tsoOffloadFlags != tsoOffloadFlags { + t.Fatalf("usoOffloadFlags (%#x) is not a superset of tsoOffloadFlags (%#x)", usoOffloadFlags, tsoOffloadFlags) + } + if usoOffloadFlags == tsoOffloadFlags { + t.Fatal("usoOffloadFlags must add bits beyond tsoOffloadFlags") + } + + cases := []struct { + name string + offloadFlags uint + wantUSO bool + }{ + {"uso-negotiated", usoOffloadFlags, true}, + {"tso-fallback", tsoOffloadFlags, false}, + {"no-vnet-hdr", 0, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := offloadUSOEnabled(tc.offloadFlags); got != tc.wantUSO { + t.Fatalf("offloadUSOEnabled(%#x) = %v, want %v", tc.offloadFlags, got, tc.wantUSO) + } + }) + } +} + +// TestNewMultiQueueReaderReplaysNegotiatedMask guards the device-wide +// TUNSETOFFLOAD downgrade bug: NewMultiQueueReader must issue the exact mask +// newTun negotiated (t.offloadFlags), not a hardcoded TSO-only mask. Because +// TUNSETOFFLOAD is per-netdev, a narrower mask on an added queue silently +// disables USO for every queue on a USO-capable kernel while the queues keep +// advertising it. +// +// A full multi-queue exercise needs /dev/net/tun and CAP_NET_ADMIN, which are +// not available in CI/sandbox, so this asserts on the struct field that the +// TUNSETOFFLOAD argument is read from. +func TestNewMultiQueueReaderReplaysNegotiatedMask(t *testing.T) { + t.Run("uso-negotiated", func(t *testing.T) { + tn := &tun{vnetHdr: true, offloadFlags: usoOffloadFlags} + // The ioctl argument in NewMultiQueueReader is uintptr(t.offloadFlags); + // it must equal the negotiated USO mask, and must NOT be the TSO-only + // mask (the original bug). + if tn.offloadFlags != usoOffloadFlags { + t.Fatalf("offloadFlags = %#x, want %#x", tn.offloadFlags, usoOffloadFlags) + } + if tn.offloadFlags == tsoOffloadFlags { + t.Fatal("added queue would downgrade USO: offloadFlags must not be the TSO-only mask when USO was negotiated") + } + }) + t.Run("tso-fallback", func(t *testing.T) { + tn := &tun{vnetHdr: true, offloadFlags: tsoOffloadFlags} + if tn.offloadFlags != tsoOffloadFlags { + t.Fatalf("offloadFlags = %#x, want %#x", tn.offloadFlags, tsoOffloadFlags) + } + }) +} diff --git a/overlay/user.go b/overlay/user.go index f3cf5adb..e4892414 100644 --- a/overlay/user.go +++ b/overlay/user.go @@ -37,21 +37,40 @@ type UserDevice struct { inboundReader *io.PipeReader inboundWriter *io.PipeWriter +} + +// userDeviceQueue is a single tio.Queue over a UserDevice's shared pipes. +// One is handed to each tun read goroutine by Readers(). All queues delegate +// reads to the same outboundReader and writes to the same inboundWriter (the +// io.Pipe serializes concurrent callers), but every queue owns a private +// readBuf/batchRet so the borrowed Packet.Bytes slice one goroutine returns is +// never clobbered by another goroutine's concurrent Read. +type userDeviceQueue struct { + outboundReader *io.PipeReader + inboundWriter *io.PipeWriter readBuf []byte batchRet [1]tio.Packet } -func (d *UserDevice) Read() ([]tio.Packet, error) { - if d.readBuf == nil { - d.readBuf = make([]byte, defaultBatchBufSize) - } - n, err := d.outboundReader.Read(d.readBuf) +func (q *userDeviceQueue) Read() ([]tio.Packet, error) { + n, err := q.outboundReader.Read(q.readBuf) if err != nil { return nil, err } - d.batchRet[0] = tio.Packet{Bytes: d.readBuf[:n]} - return d.batchRet[:], nil + q.batchRet[0] = tio.Packet{Bytes: q.readBuf[:n]} + return q.batchRet[:], nil +} + +func (q *userDeviceQueue) Write(p []byte) (int, error) { + return q.inboundWriter.Write(p) +} + +// Close is a no-op: the shared pipes are owned by the UserDevice and torn +// down by UserDevice.Close, so an individual queue must not close them out +// from under its siblings. +func (q *userDeviceQueue) Close() error { + return nil } func (d *UserDevice) Activate() error { @@ -76,7 +95,13 @@ func (d *UserDevice) NewMultiQueueReader() error { func (d *UserDevice) Readers() []tio.Queue { out := make([]tio.Queue, d.numReaders) for i := range d.numReaders { - out[i] = d + // Each queue shares the underlying pipes but owns its own scratch + // buffer so concurrent Reads across queues never alias. + out[i] = &userDeviceQueue{ + outboundReader: d.outboundReader, + inboundWriter: d.inboundWriter, + readBuf: make([]byte, defaultBatchBufSize), + } } return out } diff --git a/overlay/user_test.go b/overlay/user_test.go new file mode 100644 index 00000000..2bf11ad4 --- /dev/null +++ b/overlay/user_test.go @@ -0,0 +1,181 @@ +package overlay + +import ( + "fmt" + "net/netip" + "sync" + "testing" + + "github.com/slackhq/nebula/overlay/tio" +) + +// newTestUserDevice returns the concrete *UserDevice so tests can reach Pipe() +// and the internal queue plumbing. +func newTestUserDevice(t *testing.T) *UserDevice { + t.Helper() + dev, err := NewUserDevice([]netip.Prefix{netip.MustParsePrefix("10.0.0.1/24")}) + if err != nil { + t.Fatalf("NewUserDevice: %v", err) + } + ud, ok := dev.(*UserDevice) + if !ok { + t.Fatalf("NewUserDevice returned %T, want *UserDevice", dev) + } + return ud +} + +// TestUserDeviceReadersDistinctBuffers is the regression test for the +// multiqueue packet-corruption bug: Readers() used to hand the same +// *UserDevice (and therefore the same readBuf/batchRet) to every queue, so one +// reader's borrowed Packet.Bytes was overwritten by another reader's +// concurrent Read. Readers() must now return numReaders DISTINCT queue objects, +// each with its own backing buffer. +func TestUserDeviceReadersDistinctBuffers(t *testing.T) { + d := newTestUserDevice(t) + + // One extra reader => two queues total. + if err := d.NewMultiQueueReader(); err != nil { + t.Fatalf("NewMultiQueueReader: %v", err) + } + + readers := d.Readers() + if len(readers) != 2 { + t.Fatalf("Readers() returned %d queues, want 2", len(readers)) + } + + q0 := readers[0].(*userDeviceQueue) + q1 := readers[1].(*userDeviceQueue) + + // Distinct queue objects. + if q0 == q1 { + t.Fatal("Readers() returned the same queue object twice") + } + // Distinct backing buffers (the actual regression: shared readBuf). + if &q0.readBuf[0] == &q1.readBuf[0] { + t.Fatal("queues share the same readBuf backing array") + } + // Shared underlying pipes. + if q0.outboundReader != q1.outboundReader || q0.inboundWriter != q1.inboundWriter { + t.Fatal("queues do not share the underlying pipes") + } + + // Drive one packet through each queue and confirm the borrowed bytes from + // the first read are NOT clobbered by the second read. With a shared + // buffer, reading pkt1 into q1 would corrupt q0's still-borrowed slice. + _, ow := d.Pipe() + + pkt0 := []byte("packet-zero-aaaaaaaa") + pkt1 := []byte("packet-one-bbbbbbbbb") + + // The pipe is unbuffered, so writes block until a reader consumes them. + // Serialize: write pkt0 (read on q0), then write pkt1 (read on q1). + go func() { + if _, err := ow.Write(pkt0); err != nil { + t.Errorf("write pkt0: %v", err) + } + if _, err := ow.Write(pkt1); err != nil { + t.Errorf("write pkt1: %v", err) + } + }() + + got0, err := readers[0].Read() + if err != nil { + t.Fatalf("q0.Read: %v", err) + } + if len(got0) != 1 || string(got0[0].Bytes) != string(pkt0) { + t.Fatalf("q0 first read = %q, want %q", firstBytes(got0), pkt0) + } + // Hold onto q0's borrowed slice across q1's read. + borrowed := got0[0].Bytes + + got1, err := readers[1].Read() + if err != nil { + t.Fatalf("q1.Read: %v", err) + } + if len(got1) != 1 || string(got1[0].Bytes) != string(pkt1) { + t.Fatalf("q1 read = %q, want %q", firstBytes(got1), pkt1) + } + + // q0's borrowed bytes must still hold pkt0 - a shared buffer would now + // show pkt1's contents. + if string(borrowed) != string(pkt0) { + t.Fatalf("q0 borrowed bytes were clobbered by q1's read: got %q, want %q", borrowed, pkt0) + } +} + +// TestUserDeviceReadersConcurrentRace exercises two queues reading distinct +// packets concurrently. Run it under `go test -race`: with the old +// shared-buffer implementation the concurrent Reads raced on readBuf/batchRet +// and corrupted each other's returned slices. +func TestUserDeviceReadersConcurrentRace(t *testing.T) { + d := newTestUserDevice(t) + if err := d.NewMultiQueueReader(); err != nil { + t.Fatalf("NewMultiQueueReader: %v", err) + } + readers := d.Readers() + _, ow := d.Pipe() + + const iterations = 200 + + errs := make(chan error, 3) + + // Each reader parks in Read on the shared outboundReader; io.Pipe hands + // each write to whichever reader is currently waiting. We only care that + // concurrent Reads into distinct buffers are race-free, so any parked + // reader may serve any write. + var wg sync.WaitGroup + run := func(idx int) { + defer wg.Done() + for i := 0; i < iterations; i++ { + pkts, err := readers[idx].Read() + if err != nil { + errs <- err + return + } + if len(pkts) != 1 { + errs <- fmt.Errorf("reader %d: got %d packets, want 1", idx, len(pkts)) + return + } + // Touch every byte of the borrowed slice while the other reader + // may be mid-Read; a shared buffer would race here. + total := 0 + for _, c := range pkts[0].Bytes { + total += int(c) + } + _ = total + } + } + + wg.Add(2) + go run(0) + go run(1) + + // Feed 2*iterations packets. io.Pipe copies each write straight into the + // waiting reader's private buffer, so reusing buf between writes is safe. + go func() { + buf := make([]byte, 32) + for i := 0; i < 2*iterations; i++ { + for j := range buf { + buf[j] = byte(i + j) + } + if _, err := ow.Write(buf); err != nil { + errs <- err + return + } + } + }() + + wg.Wait() + select { + case err := <-errs: + t.Fatalf("concurrent reader failed: %v", err) + default: + } +} + +func firstBytes(p []tio.Packet) []byte { + if len(p) == 0 { + return nil + } + return p[0].Bytes +} diff --git a/udp/udp_linux.go b/udp/udp_linux.go index 62d7f35e..b309c806 100644 --- a/udp/udp_linux.go +++ b/udp/udp_linux.go @@ -161,6 +161,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 { @@ -219,16 +223,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 @@ -469,7 +486,7 @@ func (u *StdConn) ListenOut(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) { @@ -503,9 +520,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 @@ -524,12 +546,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 @@ -623,6 +646,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 @@ -650,7 +674,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 @@ -662,7 +703,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 @@ -779,7 +824,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 @@ -790,6 +843,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)) } diff --git a/udp/udp_linux_fixes_test.go b/udp/udp_linux_fixes_test.go new file mode 100644 index 00000000..51720392 --- /dev/null +++ b/udp/udp_linux_fixes_test.go @@ -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) + } +}