net/sockopts,net/udprelay: implement BindUDPGroup (#21783)

We're going to add a second SO_REUSEPORT use in the codebase. Extract
and generalize udprelay mechanism for future reuse. Also address the
Linux flaw of allowing same UID groups to merge if the ephemeral port
selection collides.

Exposes public SetReusePort, RawConn and ControlFunc signatures for
shared use.

Updates tailscale/corp#48168
Fixes tailscale/corp#49297

Signed-off-by: Alex Valiushko <alexvaliushko@tailscale.com>
Change-Id: Iba33142877513d72123c51cf1bcc33cf6a6a6964

net/sockopts/reuseport.go

@@ -0,0 +1,83 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package sockopts
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"syscall"
+
+	"tailscale.com/types/logger"
+	"tailscale.com/types/nettype"
+)
+
+// BindUDPGroup opens up to n UDP sockets sharing one local port,
+// across which the kernel spreads inbound datagrams.
+// Network must be "udp", "udp4", or "udp6".
+//
+// Control is applied after BindUDPGroup's own hook has set SO_REUSEPORT.
+// It may be nil.
+//
+// BindUDPGroup returns at least one socket and logs the shortfall.
+// It returns one socket and logs nothing where [CanShareUDPPort] is false.
+// An error is reported only when it could not bind even one socket.
+// The caller owns every returned socket.
+func BindUDPGroup(ctx context.Context, logf logger.Logf, n int, control ControlFunc, network string, port uint16) (pconns []nettype.PacketConn, boundPort uint16, err error) {
+	if n < 1 {
+		panic(fmt.Sprintf("must bind at least one socket, %d was requested", n))
+	}
+	first, err := listenUDP(ctx, control, network, port)
+	if err != nil {
+		return nil, 0, err
+	}
+	boundPort = uint16(first.LocalAddr().(*net.UDPAddr).Port)
+	pconns = append(make([]nettype.PacketConn, 0, n), first)
+	if n == 1 || !CanShareUDPPort {
+		return pconns, boundPort, nil
+	}
+
+	// The first socket binds the port exclusively, then shares it. The rest
+	// set SO_REUSEPORT before binding, joining it.
+	rc, err := RawConn(first)
+	if err == nil {
+		err = SetReusePort(rc)
+	}
+	if err != nil {
+		logf("sharing bound port %d: %v", boundPort, err)
+		return pconns, boundPort, nil
+	}
+	joining := func(network, address string, c syscall.RawConn) error {
+		if err := SetReusePort(c); err != nil {
+			return err
+		}
+		if control != nil {
+			return control(network, address, c)
+		}
+		return nil
+	}
+	for len(pconns) < n {
+		pconn, err := listenUDP(ctx, joining, network, boundPort)
+		if err != nil {
+			logf("binding socket %d of %d on port %d: %v", len(pconns)+1, n, boundPort, err)
+			break
+		}
+		pconns = append(pconns, pconn)
+	}
+	return pconns, boundPort, nil
+}
+
+func listenUDP(ctx context.Context, control ControlFunc, network string, port uint16) (*net.UDPConn, error) {
+	lc := net.ListenConfig{Control: control}
+	pconn, err := lc.ListenPacket(ctx, network, fmt.Sprintf(":%d", port))
+	if err != nil {
+		return nil, err
+	}
+	uc, ok := pconn.(*net.UDPConn)
+	if !ok {
+		pconn.Close()
+		return nil, fmt.Errorf("%q (%T) is not a UDP network", network, pconn)
+	}
+	return uc, nil
+}

net/sockopts/reuseport_default_test.go

@@ -0,0 +1,24 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build !linux
+
+package sockopts
+
+import (
+	"testing"
+
+	"tailscale.com/tstest"
+)
+
+func TestBindUDPGroupUnsupported(t *testing.T) {
+	var tc testControl
+	var log tstest.MemLogger
+	pconns, boundPort, err := BindUDPGroup(t.Context(), log.Logf, 4, tc.control, "udp4", 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer closeConns(pconns)
+	wantSharedPort(t, pconns, boundPort, 1)
+	wantLines(t, &log, 0)
+}

net/sockopts/reuseport_linux_test.go

@@ -0,0 +1,69 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build linux
+
+package sockopts
+
+import (
+	"testing"
+
+	"golang.org/x/sys/unix"
+	"tailscale.com/tstest"
+)
+
+func TestBindUDPGroup(t *testing.T) {
+	for _, tt := range []struct {
+		name    string
+		sockets int // passed to BindUDPGroup
+		failAt  int // bind at which the Control hook starts failing
+
+		wantSockets   int
+		wantReusePort bool
+		wantLines     int
+		wantErr       bool
+	}{
+		{name: "full", sockets: 4, wantSockets: 4, wantReusePort: true},
+		{name: "short", sockets: 4, failAt: 3, wantSockets: 2, wantReusePort: true, wantLines: 1},
+		{name: "lone", sockets: 1, wantSockets: 1},
+		{name: "unbindable", sockets: 4, failAt: 1, wantErr: true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := testControl{failAt: tt.failAt}
+			var log tstest.MemLogger
+
+			pconns, boundPort, err := BindUDPGroup(t.Context(), log.Logf, tt.sockets, tc.control, "udp4", 0)
+			defer closeConns(pconns)
+			if gotErr := err != nil; gotErr != tt.wantErr {
+				t.Fatalf("BindUDPGroup err = %v, want an error: %v", err, tt.wantErr)
+			}
+			if tt.wantErr {
+				if pconns != nil {
+					t.Errorf("returned %d sockets, want none", len(pconns))
+				}
+				return
+			}
+			wantSharedPort(t, pconns, boundPort, tt.wantSockets)
+			for i, pconn := range pconns {
+				rc, err := RawConn(pconn)
+				if err != nil {
+					t.Fatalf("socket %d: %v", i, err)
+				}
+				var val int
+				var serr error
+				if err := rc.Control(func(fd uintptr) {
+					val, serr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT)
+				}); err != nil {
+					t.Fatalf("socket %d: %v", i, err)
+				}
+				if serr != nil {
+					t.Fatalf("socket %d: getsockopt(SO_REUSEPORT): %v", i, serr)
+				}
+				if got := val == 1; got != tt.wantReusePort {
+					t.Errorf("socket %d: SO_REUSEPORT = %v, want %v", i, got, tt.wantReusePort)
+				}
+			}
+			wantLines(t, &log, tt.wantLines)
+		})
+	}
+}

net/sockopts/reuseport_test.go

@@ -0,0 +1,95 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package sockopts
+
+import (
+	"errors"
+	"net"
+	"strings"
+	"syscall"
+	"testing"
+
+	"tailscale.com/tstest"
+	"tailscale.com/types/nettype"
+)
+
+var errListen = errors.New("test listen failure")
+
+type testControl struct {
+	// failAt is the 1-based bind at which control starts returning errListen.
+	// Zero never fails.
+	failAt int
+
+	calls int
+}
+
+func (tc *testControl) control(_, _ string, _ syscall.RawConn) error {
+	tc.calls++
+	if tc.failAt != 0 && tc.calls >= tc.failAt {
+		return errListen
+	}
+	return nil
+}
+
+func wantLines(t *testing.T, log *tstest.MemLogger, want int) {
+	t.Helper()
+	if got := strings.Count(log.String(), "\n"); got != want {
+		t.Errorf("logged %d lines, want %d: %q", got, want, log.String())
+	}
+}
+
+func closeConns(pconns []nettype.PacketConn) {
+	for _, pconn := range pconns {
+		pconn.Close()
+	}
+}
+
+func wantSharedPort(t *testing.T, pconns []nettype.PacketConn, boundPort uint16, want int) {
+	t.Helper()
+	if got := len(pconns); got != want {
+		t.Fatalf("bound %d sockets, want %d", got, want)
+	}
+	if boundPort == 0 {
+		t.Error("boundPort is zero")
+	}
+	for i, pconn := range pconns {
+		if got := uint16(pconn.LocalAddr().(*net.UDPAddr).Port); got != boundPort {
+			t.Errorf("socket %d bound to port %d, want %d", i, got, boundPort)
+		}
+	}
+}
+
+func TestBindUDPGroupError(t *testing.T) {
+	for _, tt := range []struct {
+		name    string
+		failAt  int // bind at which the Control hook starts failing
+		network string
+
+		wantErr error // unwrapped from the returned error, if non-nil
+	}{
+		{name: "control-fails", failAt: 1, network: "udp4", wantErr: errListen},
+		{name: "bad-network", network: "tcp4"},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := testControl{failAt: tt.failAt}
+			var log tstest.MemLogger
+
+			pconns, boundPort, err := BindUDPGroup(t.Context(), log.Logf, 4, tc.control, tt.network, 0)
+			defer closeConns(pconns)
+			if err == nil {
+				t.Fatal("BindUDPGroup succeeded, want an error")
+			}
+			if tt.wantErr != nil && !errors.Is(err, tt.wantErr) {
+				t.Errorf("err = %v, want it to wrap %v", err, tt.wantErr)
+			}
+			if pconns != nil {
+				t.Errorf("returned %d sockets, want none", len(pconns))
+			}
+			if boundPort != 0 {
+				t.Errorf("boundPort = %d, want 0", boundPort)
+			}
+			wantLines(t, &log, 0)
+		})
+	}
+}

net/sockopts/sockopts.go

@@ -5,12 +5,62 @@
 package sockopts
 
 import (
+	"errors"
+	"fmt"
 	"net"
 	"runtime"
+	"syscall"
 
 	"tailscale.com/types/nettype"
 )
 
+// RawConn returns the socket underlying pconn.
+//
+// The error wraps [ErrNoSocket] if pconn has no socket.
+func RawConn(pconn nettype.PacketConn) (syscall.RawConn, error) {
+	sc, ok := pconn.(syscall.Conn)
+	if !ok {
+		return nil, fmt.Errorf("%T: %w", pconn, ErrNoSocket)
+	}
+	return sc.SyscallConn()
+}
+
+// ErrNoSocket reports that there is no socket to apply an option to,
+// distinct from the kernel refusing the option on one.
+var ErrNoSocket = errors.New("no socket to configure")
+
+// ControlEach runs the system call fn on the socket of every conn in pconns.
+// It reports [ErrNoSocket] without calling fn if pconns is empty or if any
+// conn has no socket; otherwise it calls fn for each socket before returning,
+// and returns the join of fn's errors.
+func ControlEach(pconns []nettype.PacketConn, fn func(fd uintptr) error) error {
+	if len(pconns) == 0 {
+		return fmt.Errorf("empty group: %w", ErrNoSocket)
+	}
+	rcs := make([]syscall.RawConn, len(pconns))
+	for i, pconn := range pconns {
+		rc, err := RawConn(pconn)
+		if err != nil {
+			return fmt.Errorf("socket %d of %d: %w", i, len(pconns), err)
+		}
+		rcs[i] = rc
+	}
+	var errs []error
+	for _, rc := range rcs {
+		if err := rc.Control(func(fd uintptr) {
+			if err := fn(fd); err != nil {
+				errs = append(errs, err)
+			}
+		}); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	return errors.Join(errs...)
+}
+
+// ControlFunc is the Control hook of a [net.ListenConfig].
+type ControlFunc func(network, address string, c syscall.RawConn) error
+
 // BufferDirection represents either the read/receive or write/send direction
 // of a socket buffer.
 type BufferDirection string

net/sockopts/sockopts_default.go

@@ -6,9 +6,21 @@
 package sockopts
 
 import (
+	"errors"
+	"syscall"
+
 	"tailscale.com/types/nettype"
 )
 
+// CanShareUDPPort reports whether this platform spreads inbound datagrams
+// across several UDP sockets sharing one local port.
+const CanShareUDPPort = false
+
+// SetReusePort returns [errors.ErrUnsupported].
+func SetReusePort(c syscall.RawConn) error {
+	return errors.ErrUnsupported
+}
+
 // SetBufferSize sets pconn's buffer to size for direction. size may be silently
 // capped depending on platform.
 //

net/sockopts/sockopts_linux.go

@@ -9,9 +9,25 @@ import (
 	"net"
 	"syscall"
 
+	"golang.org/x/sys/unix"
 	"tailscale.com/types/nettype"
 )
 
+// CanShareUDPPort reports whether this platform spreads inbound datagrams
+// across several UDP sockets sharing one local port.
+const CanShareUDPPort = true
+
+// SetReusePort sets SO_REUSEPORT on the socket underlying c.
+func SetReusePort(c syscall.RawConn) error {
+	var serr error
+	if err := c.Control(func(fd uintptr) {
+		serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
+	}); err != nil {
+		return err
+	}
+	return serr
+}
+
 // SetBufferSize sets pconn's buffer to size for direction. It attempts
 // (errForce) to set SO_SNDBUFFORCE or SO_RECVBUFFORCE which can overcome the
 // limit of net.core.{r,w}mem_max, but require CAP_NET_ADMIN. It falls back to

net/udprelay/server.go

@@ -13,11 +13,9 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
-	"net"
 	"net/netip"
 	"runtime"
 	"slices"
-	"strconv"
 	"sync"
 	"syscall"
 	"time"
@@ -565,11 +563,11 @@ var _ batching.Conn = (*singlePacketConn)(nil)
 // singlePacketConn implements [batching.Conn] with single packet syscall
 // operations.
 type singlePacketConn struct {
-	*net.UDPConn
+	nettype.PacketConn
 }
 
 func (c *singlePacketConn) ReadBatch(slab []byte, packets []batching.ReceivedPacket) (int, error) {
-	n, ap, err := c.UDPConn.ReadFromUDPAddrPort(slab)
+	n, ap, err := c.PacketConn.ReadFromUDPAddrPort(slab)
 	if err != nil {
 		return 0, err
 	}
@@ -586,7 +584,7 @@ func (c *singlePacketConn) WriteBatchTo(buffs [][]byte, addr netip.AddrPort, gen
 		} else {
 			buff = buff[offset:]
 		}
-		_, err := c.UDPConn.WriteToUDPAddrPort(buff, addr)
+		_, err := c.PacketConn.WriteToUDPAddrPort(buff, addr)
 		if err != nil {
 			return err
 		}
@@ -642,10 +640,8 @@ func trySetSOMark(logf logger.Logf, netMon *netmon.Monitor, network, address str
 // may be bound per address family, e.g. SO_REUSEPORT, depending on platform.
 //
 // desiredPort may be zero, in which case port selection is left up to the host
-// networking stack. We make no attempt to bind a consistent port between udp4
-// and udp6 if the requested port is zero, but a consistent port is used
-// across multiple sockets within a given address family if SO_REUSEPORT is
-// supported.
+// networking stack.
+// If more than one udp4 socket was bound udp6 reuses the same port number.
 //
 // TODO: make these "re-bindable" in similar fashion to magicsock as a means to
 // deal with EDR software closing them. http://go/corp/30118. We could re-use
@@ -657,56 +653,44 @@ func (s *Server) bindSockets(desiredPort uint16) error {
 	// arbitrary. Use GOMAXPROCS rather than NumCPU as it is container-aware
 	// and respects CPU limits/quotas set via cgroups.
 	maxSocketsPerAF := min(16, runtime.GOMAXPROCS(0))
-	listenConfig := &net.ListenConfig{
-		Control: func(network, address string, c syscall.RawConn) error {
-			trySetReusePort(network, address, c)
-			trySetSOMark(s.logf, s.netMon, network, address, c)
-			return nil
-		},
+	control := func(network, address string, c syscall.RawConn) error {
+		trySetSOMark(s.logf, s.netMon, network, address, c)
+		return nil
 	}
 	for _, network := range []string{"udp4", "udp6"} {
-	SocketsLoop:
-		for i := range maxSocketsPerAF {
-			if i > 0 {
-				// Use a consistent port per address family if the user-supplied
-				// port was zero, and we are binding multiple sockets.
-				if network == "udp4" {
-					desiredPort = s.uc4Port
-				} else {
-					desiredPort = s.uc6Port
-				}
+		pconns, boundPort, err := sockopts.BindUDPGroup(
+			context.Background(),
+			logger.WithPrefix(s.logf, network+": "),
+			maxSocketsPerAF,
+			control,
+			network,
+			desiredPort,
+		)
+		if err != nil {
+			if network == "udp4" {
+				return err
 			}
-			uc, boundPort, err := s.bindSocketTo(listenConfig, network, desiredPort)
-			if err != nil {
-				switch {
-				case i == 0 && network == "udp4":
-					// At least one udp4 socket is required.
-					return err
-				case i == 0 && network == "udp6":
-					// A udp6 socket is not required.
-					s.logf("ignoring IPv6 bind failure: %v", err)
-					break SocketsLoop
-				default: // i > 0
-					// Reusable sockets are not required.
-					s.logf("ignoring reusable (index=%d network=%v) socket bind failure: %v", i, network, err)
-					break SocketsLoop
-				}
-			}
-			pc := batching.TryUpgradeToConn(uc, network, "udprelay_rxq_overflows", s.controlKnobs)
+			// A udp6 socket is not required.
+			s.logf("ignoring IPv6 bind failure: %v", err)
+			continue
+		}
+		bcs := make([]batching.Conn, 0, len(pconns))
+		for _, pconn := range pconns {
+			trySetUDPSocketOptions(pconn, s.logf)
+			pc := batching.TryUpgradeToConn(pconn, network, "udprelay_rxq_overflows", s.controlKnobs)
 			bc, ok := pc.(batching.Conn)
 			if !ok {
-				bc = &singlePacketConn{uc}
+				bc = &singlePacketConn{pconn}
 			}
-			if network == "udp4" {
-				s.uc4 = append(s.uc4, bc)
-				s.uc4Port = boundPort
-			} else {
-				s.uc6 = append(s.uc6, bc)
-				s.uc6Port = boundPort
-			}
-			if !isReusableSocket(uc) {
-				break
+			bcs = append(bcs, bc)
+		}
+		if network == "udp4" {
+			s.uc4, s.uc4Port = bcs, boundPort
+			if len(pconns) > 1 {
+				desiredPort = boundPort
 			}
+		} else {
+			s.uc6, s.uc6Port = bcs, boundPort
 		}
 	}
 	s.logf("listening on udp4:%d sockets=%d", s.uc4Port, len(s.uc4))
@@ -716,26 +700,6 @@ func (s *Server) bindSockets(desiredPort uint16) error {
 	return nil
 }
 
-func (s *Server) bindSocketTo(listenConfig *net.ListenConfig, network string, port uint16) (*net.UDPConn, uint16, error) {
-	lis, err := listenConfig.ListenPacket(context.Background(), network, fmt.Sprintf(":%d", port))
-	if err != nil {
-		return nil, 0, err
-	}
-	uc := lis.(*net.UDPConn)
-	trySetUDPSocketOptions(uc, s.logf)
-	_, boundPortStr, err := net.SplitHostPort(uc.LocalAddr().String())
-	if err != nil {
-		uc.Close()
-		return nil, 0, err
-	}
-	portUint, err := strconv.ParseUint(boundPortStr, 10, 16)
-	if err != nil {
-		uc.Close()
-		return nil, 0, err
-	}
-	return uc, uint16(portUint), nil
-}
-
 // Close closes the server.
 func (s *Server) Close() error {
 	s.closeOnce.Do(func() {

net/udprelay/server_linux.go

@@ -1,34 +0,0 @@
-// Copyright (c) Tailscale Inc & contributors
-// SPDX-License-Identifier: BSD-3-Clause
-
-//go:build linux
-
-package udprelay
-
-import (
-	"net"
-	"syscall"
-
-	"golang.org/x/sys/unix"
-)
-
-func trySetReusePort(_ string, _ string, c syscall.RawConn) {
-	c.Control(func(fd uintptr) {
-		unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
-	})
-}
-
-func isReusableSocket(uc *net.UDPConn) bool {
-	rc, err := uc.SyscallConn()
-	if err != nil {
-		return false
-	}
-	var reusable bool
-	rc.Control(func(fd uintptr) {
-		val, err := unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT)
-		if err == nil && val == 1 {
-			reusable = true
-		}
-	})
-	return reusable
-}

net/udprelay/server_notlinux.go

@@ -1,17 +0,0 @@
-// Copyright (c) Tailscale Inc & contributors
-// SPDX-License-Identifier: BSD-3-Clause
-
-//go:build !linux
-
-package udprelay
-
-import (
-	"net"
-	"syscall"
-)
-
-func trySetReusePort(_ string, _ string, _ syscall.RawConn) {}
-
-func isReusableSocket(*net.UDPConn) bool {
-	return false
-}