udp: fall back to SO_RCVBUF/SO_SNDBUF when FORCE is not permitted

SO_RCVBUFFORCE/SO_SNDBUFFORCE need CAP_NET_ADMIN. Without it (non-root,
or root under Docker's default caps) listen.read_buffer/write_buffer
failed with EPERM and the socket kept the system default. On EPERM retry
with the plain option, which the kernel clamps to net.core.{r,w}mem_max,
and warn when the effective size is smaller than requested.

udp/udp_linux.go

@@ -135,12 +135,23 @@ func (u *StdConn) Rebind() error {
 	return nil
 }
 
+// SetRecvBuffer tries SO_RCVBUFFORCE, which needs CAP_NET_ADMIN. Without it we
+// fall back to SO_RCVBUF, which the kernel clamps to net.core.rmem_max.
 func (u *StdConn) SetRecvBuffer(n int) error {
-	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, n)
+	err := unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, n)
+	if errors.Is(err, unix.EPERM) {
+		err = unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_RCVBUF, n)
+	}
+	return err
 }
 
+// SetSendBuffer is SetRecvBuffer for SO_SNDBUFFORCE/SO_SNDBUF and net.core.wmem_max.
 func (u *StdConn) SetSendBuffer(n int) error {
-	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, n)
+	err := unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, n)
+	if errors.Is(err, unix.EPERM) {
+		err = unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUF, n)
+	}
+	return err
 }
 
 func (u *StdConn) SetSoMark(mark int) error {
@@ -396,8 +407,13 @@ func (u *StdConn) ReloadConfig(c *config.C) {
 	b := c.GetInt("listen.read_buffer", 0)
 	if b > 0 {
 		if err := u.SetRecvBuffer(b); err == nil {
-			if s, err := u.GetRecvBuffer(); err == nil {
+			// The kernel reports double the requested size
+			if s, err := u.GetRecvBuffer(); err == nil && s >= 2*b {
 				u.l.Info("listen.read_buffer was set", "size", s)
+			} else if err == nil {
+				u.l.Warn("listen.read_buffer was limited by the system",
+					"requested", b, "size", s,
+					"hint", "raise net.core.rmem_max or grant CAP_NET_ADMIN")
 			} else {
 				u.l.Warn("Failed to get listen.read_buffer", "error", err)
 			}
@@ -409,8 +425,12 @@ func (u *StdConn) ReloadConfig(c *config.C) {
 	b = c.GetInt("listen.write_buffer", 0)
 	if b > 0 {
 		if err := u.SetSendBuffer(b); err == nil {
-			if s, err := u.GetSendBuffer(); err == nil {
+			if s, err := u.GetSendBuffer(); err == nil && s >= 2*b {
 				u.l.Info("listen.write_buffer was set", "size", s)
+			} else if err == nil {
+				u.l.Warn("listen.write_buffer was limited by the system",
+					"requested", b, "size", s,
+					"hint", "raise net.core.wmem_max or grant CAP_NET_ADMIN")
 			} else {
 				u.l.Warn("Failed to get listen.write_buffer", "error", err)
 			}

udp/udp_linux_buffer_test.go

@@ -0,0 +1,124 @@
+//go:build linux && !android && !e2e_testing
+
+package udp
+
+import (
+	"bytes"
+	"fmt"
+	"log/slog"
+	"net/netip"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/slackhq/nebula/config"
+)
+
+func readSysctlInt(t *testing.T, path string) int {
+	t.Helper()
+	b, err := os.ReadFile(path)
+	if err != nil {
+		t.Skipf("can't read %s: %v", path, err)
+	}
+	n, err := strconv.Atoi(strings.TrimSpace(string(b)))
+	if err != nil {
+		t.Skipf("can't parse %s: %v", path, err)
+	}
+	return n
+}
+
+// TestSetBufferAboveSysctlMax asks for more than net.core.{r,w}mem_max. With
+// CAP_NET_ADMIN the FORCE variant takes it as is; without it we must fall back
+// to the plain option and get the clamped size instead of EPERM.
+func TestSetBufferAboveSysctlMax(t *testing.T) {
+	cases := []struct {
+		name   string
+		sysctl string
+		set    func(*StdConn, int) error
+		get    func(*StdConn) (int, error)
+	}{
+		{"read", "/proc/sys/net/core/rmem_max", (*StdConn).SetRecvBuffer, (*StdConn).GetRecvBuffer},
+		{"write", "/proc/sys/net/core/wmem_max", (*StdConn).SetSendBuffer, (*StdConn).GetSendBuffer},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			sysMax := readSysctlInt(t, tc.sysctl)
+			c, err := NewListener(testLogger(), Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0"), Batch: 1})
+			if err != nil {
+				t.Skipf("NewListener: %v", err)
+			}
+			sc := c.(*StdConn)
+			defer sc.Close()
+
+			want := sysMax + 1<<20
+			if err := tc.set(sc, want); err != nil {
+				t.Fatalf("set %d: %v", want, err)
+			}
+			got, err := tc.get(sc)
+			if err != nil {
+				t.Fatalf("get: %v", err)
+			}
+			// The kernel doubles the value; either path gives at least the clamped size.
+			if got < 2*sysMax {
+				t.Fatalf("effective size %d, want >= %d", got, 2*sysMax)
+			}
+			t.Logf("requested %d, %s=%d, effective %d (forced=%v)", want, tc.sysctl, sysMax, got, got >= 2*want)
+		})
+	}
+}
+
+// TestReloadConfigBufferLogLevel checks that a clamped buffer is a Warn, and a
+// buffer that fits is still the Info it always was.
+func TestReloadConfigBufferLogLevel(t *testing.T) {
+	sysMax := readSysctlInt(t, "/proc/sys/net/core/rmem_max")
+	if wmax := readSysctlInt(t, "/proc/sys/net/core/wmem_max"); wmax < sysMax {
+		sysMax = wmax
+	}
+	cases := []struct {
+		name string
+		size int
+		want []string
+	}{
+		{"fits", 64 * 1024, []string{"level=INFO msg=\"listen.read_buffer was set\"", "level=INFO msg=\"listen.write_buffer was set\""}},
+		{"too big", sysMax + 1<<20, []string{"level=WARN msg=\"listen.read_buffer was limited by the system\"", "level=WARN msg=\"listen.write_buffer was limited by the system\""}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			var buf bytes.Buffer
+			l := slog.New(slog.NewTextHandler(&buf, nil))
+			c, err := NewListener(l, Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0"), Batch: 1})
+			if err != nil {
+				t.Skipf("NewListener: %v", err)
+			}
+			sc := c.(*StdConn)
+			defer sc.Close()
+
+			if tc.size > sysMax {
+				// With CAP_NET_ADMIN the FORCE variant takes any size, nothing gets limited.
+				if err := sc.SetRecvBuffer(tc.size); err == nil {
+					if got, _ := sc.GetRecvBuffer(); got >= 2*tc.size {
+						t.Skip("FORCE path took the full size, can't observe the limit")
+					}
+				}
+			}
+
+			cfg := config.NewC(l)
+			if err := cfg.LoadString(fmt.Sprintf("listen:\n  read_buffer: %d\n  write_buffer: %d\n", tc.size, tc.size)); err != nil {
+				t.Fatal(err)
+			}
+			buf.Reset()
+			sc.ReloadConfig(cfg)
+			out := buf.String()
+			for _, w := range tc.want {
+				if !strings.Contains(out, w) {
+					t.Errorf("missing %s in:\n%s", w, out)
+				}
+			}
+			if strings.Contains(out, "level=ERROR") {
+				t.Errorf("unexpected error log:\n%s", out)
+			}
+			t.Log(out)
+		})
+	}
+}