Address PR feedback: remove outbound rate limit, improve config docs

Remove rate limiting from StartHandshake (outbound) since DoS
protection only needs to limit inbound handshakes. This also avoids
returning nil from StartHandshake which historically always returned
non-nil. Update config comment to note openssl speed is single-core
and suggest scaling by routines.

Co-Authored-By: Claude <svc-devxp-claude@slack-corp.com>

CODEOWNERS

@@ -0,0 +1 @@
+#ECCN:Open Source

dist/wireshark/nebula.lua

@@ -84,30 +84,24 @@ end
 
 function nebula.prefs_changed()
     if default_settings.all_ports == nebula.prefs.all_ports and default_settings.port == nebula.prefs.port then
-        -- Nothing changed, bail
         return
     end
 
-    -- Remove our old dissector
+    -- Remove all existing registrations
     DissectorTable.get("udp.port"):remove_all(nebula)
 
-    if nebula.prefs.all_ports and default_settings.all_ports ~= nebula.prefs.all_ports then
+    if nebula.prefs.all_ports then
-        default_settings.all_port = nebula.prefs.all_ports
+        -- Register on every port for hole punch capture
-
         for i=0, 65535 do
             DissectorTable.get("udp.port"):add(i, nebula)
         end
-
+    else
-        -- no need to establish again on specific ports
+        -- Register on the configured port only
-        return
+        DissectorTable.get("udp.port"):add(nebula.prefs.port, nebula)
     end
 
-
+    default_settings.all_ports = nebula.prefs.all_ports
-    if default_settings.all_ports ~= nebula.prefs.all_ports then
+    default_settings.port = nebula.prefs.port
-        -- Add our new port dissector
-        default_settings.port = nebula.prefs.port
-        DissectorTable.get("udp.port"):add(default_settings.port, nebula)
-    end
 end
 
 DissectorTable.get("udp.port"):add(default_settings.port, nebula)

e2e/tunnels_test.go

@@ -12,6 +12,8 @@ import (
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/cert_test"
 	"github.com/slackhq/nebula/e2e/router"
+	"github.com/slackhq/nebula/header"
+	"github.com/slackhq/nebula/udp"
 	"github.com/stretchr/testify/assert"
 	"gopkg.in/yaml.v3"
 )
@@ -365,3 +367,106 @@ func TestCrossStackRelaysWork(t *testing.T) {
 	//theirControl.Stop()
 	//relayControl.Stop()
 }
+
+func TestCloseTunnelAuthenticated(t *testing.T) {
+	ca, _, caKey, _ := cert_test.NewTestCaCert(cert.Version1, cert.Curve_CURVE25519, time.Now(), time.Now().Add(10*time.Minute), nil, nil, []string{})
+	myControl, myVpnIpNet, myUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "me", "10.128.0.1/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "5s"}})
+	theirControl, theirVpnIpNet, theirUdpAddr, _ := newSimpleServer(cert.Version1, ca, caKey, "them", "10.128.0.2/24", m{"tunnels": m{"drop_inactive": true, "inactivity_timeout": "10m"}})
+
+	// Share our underlay information
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+
+	// Start the servers
+	myControl.Start()
+	theirControl.Start()
+
+	r := router.NewR(t, myControl, theirControl)
+
+	r.Log("Assert the tunnel between me and them works")
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+
+	r.Log("Close the tunnel")
+	myControl.CloseTunnel(theirVpnIpNet[0].Addr(), false)
+	r.FlushAll()
+
+	waitStart := time.Now()
+	for {
+		myIndexes := len(myControl.GetHostmap().Indexes)
+		theirIndexes := len(theirControl.GetHostmap().Indexes)
+		if myIndexes == 0 && theirIndexes == 0 {
+			break
+		}
+
+		since := time.Since(waitStart)
+		r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
+		if since > time.Second*6 {
+			t.Fatal("Tunnel should have been declared inactive after 2 seconds and before 6 seconds")
+		}
+
+		time.Sleep(1 * time.Second)
+		//r.FlushAll()
+	}
+
+	r.Logf("Happy path success, tunnels were dropped within %v", time.Since(waitStart))
+
+	myControl.InjectLightHouseAddr(theirVpnIpNet[0].Addr(), theirUdpAddr)
+	theirControl.InjectLightHouseAddr(myVpnIpNet[0].Addr(), myUdpAddr)
+	r.Log("Assert another tunnel between me and them works")
+	assertTunnel(t, myVpnIpNet[0].Addr(), theirVpnIpNet[0].Addr(), myControl, theirControl, r)
+	hi := myControl.GetHostInfoByVpnAddr(theirVpnIpNet[0].Addr(), false)
+	if hi == nil {
+		t.Fatal("There is no hostinfo for this tunnel")
+	}
+	myHi := theirControl.GetHostInfoByVpnAddr(myVpnIpNet[0].Addr(), false)
+	if myHi == nil {
+		t.Fatal("There is no hostinfo for my tunnel")
+	}
+	r.Log("It does")
+
+	buf := make([]byte, 1024)
+	hdr := header.H{
+		Version:        1,
+		Type:           header.CloseTunnel,
+		Subtype:        0,
+		Reserved:       0,
+		RemoteIndex:    hi.RemoteIndex,
+		MessageCounter: 5,
+	}
+	out, err := hdr.Encode(buf)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	pkt := &udp.Packet{
+		To:   hi.CurrentRemote,
+		From: myHi.CurrentRemote,
+		Data: out,
+	}
+	r.InjectUDPPacket(myControl, theirControl, pkt)
+	r.Log("Injected bogus close tunnel. Let's see!")
+	waitStart = time.Now()
+	for {
+		myIndexes := len(myControl.GetHostmap().Indexes)
+		theirIndexes := len(theirControl.GetHostmap().Indexes)
+		if myIndexes == 0 {
+			t.Fatal("myIndexes should not be 0")
+		}
+		if theirIndexes == 0 {
+			t.Fatal("theirIndexes should not be 0, they should have rejected this bogus packet")
+		}
+
+		since := time.Since(waitStart)
+		r.Logf("my tunnels: %v; their tunnels: %v; duration: %v", myIndexes, theirIndexes, since)
+		if since > time.Second*4 {
+			t.Log("The tunnel would have been gone by now")
+			break
+		}
+
+		time.Sleep(1 * time.Second)
+		r.FlushAll()
+	}
+
+	myControl.Stop()
+	theirControl.Stop()
+}

examples/config.yml

@@ -342,6 +342,14 @@ logging:
   # after receiving the response for lighthouse queries
   #trigger_buffer: 64
 
+  # max_rate limits the number of new inbound handshakes per second. Once the limit is reached,
+  # new handshakes are dropped until the next second. A value of 0 means unlimited (default).
+  # This is useful for preventing DoS attacks that attempt to exhaust CPU with handshake crypto.
+  # Running `openssl speed ecdhp256` on your hardware can be a good rule of thumb for choosing
+  # a max, as each handshake performs similar DH operations. Note that this benchmarks a single
+  # core, so you may wish to scale the value by the number of `routines` configured.
+  #max_rate: 0
+
 # Tunnel manager settings
 #tunnels:
   # drop_inactive controls whether inactive tunnels are maintained or dropped after the inactive_timeout period has

firewall.go

@@ -230,7 +230,7 @@ func NewFirewallFromConfig(l *logrus.Logger, cs *CertState, c *config.C) (*Firew
 	case "drop":
 		fw.OutSendReject = false
 	default:
-		l.WithField("action", inboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
+		l.WithField("action", outboundAction).Warn("invalid firewall.outbound_action, defaulting to `drop`")
 		fw.OutSendReject = false
 	}
 

handshake_manager.go

@@ -23,22 +23,25 @@ const (
 	DefaultHandshakeRetries       = 10
 	DefaultHandshakeTriggerBuffer = 64
 	DefaultUseRelays              = true
+	DefaultMaxHandshakeRate       = 0 // 0 means unlimited
 )
 
 var (
 	defaultHandshakeConfig = HandshakeConfig{
-		tryInterval:   DefaultHandshakeTryInterval,
+		tryInterval:      DefaultHandshakeTryInterval,
-		retries:       DefaultHandshakeRetries,
+		retries:          DefaultHandshakeRetries,
-		triggerBuffer: DefaultHandshakeTriggerBuffer,
+		triggerBuffer:    DefaultHandshakeTriggerBuffer,
-		useRelays:     DefaultUseRelays,
+		useRelays:        DefaultUseRelays,
+		maxHandshakeRate: DefaultMaxHandshakeRate,
 	}
 )
 
 type HandshakeConfig struct {
-	tryInterval   time.Duration
+	tryInterval      time.Duration
-	retries       int64
+	retries          int64
-	triggerBuffer int
+	triggerBuffer    int
-	useRelays     bool
+	useRelays        bool
+	maxHandshakeRate int
 
 	messageMetrics *MessageMetrics
 }
@@ -58,9 +61,15 @@ type HandshakeManager struct {
 	messageMetrics         *MessageMetrics
 	metricInitiated        metrics.Counter
 	metricTimedOut         metrics.Counter
+	metricRateLimited      metrics.Counter
 	f                      *Interface
 	l                      *logrus.Logger
 
+	// Rate limiting for new handshakes (token bucket)
+	rateBucket   int // tokens currently available
+	rateMax      int // max tokens (== max handshakes per second), 0 means unlimited
+	rateLastTick time.Time
+
 	// can be used to trigger outbound handshake for the given vpnIp
 	trigger chan netip.Addr
 }
@@ -116,10 +125,41 @@ func NewHandshakeManager(l *logrus.Logger, mainHostMap *HostMap, lightHouse *Lig
 		messageMetrics:         config.messageMetrics,
 		metricInitiated:        metrics.GetOrRegisterCounter("handshake_manager.initiated", nil),
 		metricTimedOut:         metrics.GetOrRegisterCounter("handshake_manager.timed_out", nil),
+		metricRateLimited:      metrics.GetOrRegisterCounter("handshake_manager.rate_limited", nil),
+		rateBucket:             config.maxHandshakeRate,
+		rateMax:                config.maxHandshakeRate,
+		rateLastTick:           time.Now(),
 		l:                      l,
 	}
 }
 
+// handshakeRateAllow checks the token bucket rate limiter and returns true if a
+// new handshake is allowed. Must be called with hm.Lock held.
+func (hm *HandshakeManager) handshakeRateAllow(now time.Time) bool {
+	if hm.rateMax == 0 {
+		return true
+	}
+
+	// Refill tokens based on elapsed time
+	elapsed := now.Sub(hm.rateLastTick)
+	if elapsed >= time.Second {
+		// Add tokens for full seconds elapsed
+		tokens := int(elapsed/time.Second) * hm.rateMax
+		hm.rateBucket += tokens
+		if hm.rateBucket > hm.rateMax {
+			hm.rateBucket = hm.rateMax
+		}
+		hm.rateLastTick = now
+	}
+
+	if hm.rateBucket > 0 {
+		hm.rateBucket--
+		return true
+	}
+
+	return false
+}
+
 func (hm *HandshakeManager) Run(ctx context.Context) {
 	clockSource := time.NewTicker(hm.config.tryInterval)
 	defer clockSource.Stop()
@@ -149,6 +189,15 @@ func (hm *HandshakeManager) HandleIncoming(via ViaSender, packet []byte, h *head
 	case header.HandshakeIXPSK0:
 		switch h.MessageCounter {
 		case 1:
+			// Check rate limit for new incoming handshakes
+			hm.Lock()
+			allowed := hm.handshakeRateAllow(time.Now())
+			hm.Unlock()
+			if !allowed {
+				hm.metricRateLimited.Inc(1)
+				hm.l.WithField("from", via).Debug("Handshake rate limit reached, dropping incoming handshake")
+				return
+			}
 			ixHandshakeStage1(hm.f, via, packet, h)
 
 		case 2:

handshake_manager_test.go

@@ -65,6 +65,68 @@ func Test_NewHandshakeManagerVpnIp(t *testing.T) {
 	assert.NotContains(t, blah.vpnIps, ip)
 }
 
+func Test_HandshakeManagerRateLimit(t *testing.T) {
+	l := test.NewLogger()
+	localrange := netip.MustParsePrefix("10.1.1.1/24")
+	preferredRanges := []netip.Prefix{localrange}
+	mainHM := newHostMap(l)
+	mainHM.preferredRanges.Store(&preferredRanges)
+
+	lh := newTestLighthouse()
+
+	config := defaultHandshakeConfig
+	config.maxHandshakeRate = 2
+
+	hm := NewHandshakeManager(l, mainHM, lh, &udp.NoopConn{}, config)
+	hm.f = &Interface{handshakeManager: hm, pki: &PKI{}, l: l}
+
+	now := time.Now()
+
+	// Should allow up to maxHandshakeRate handshakes
+	hm.Lock()
+	assert.True(t, hm.handshakeRateAllow(now), "first handshake should be allowed")
+	assert.True(t, hm.handshakeRateAllow(now), "second handshake should be allowed")
+	assert.False(t, hm.handshakeRateAllow(now), "third handshake should be rate limited")
+	hm.Unlock()
+
+	// After advancing time by 1 second, tokens should refill
+	hm.Lock()
+	assert.True(t, hm.handshakeRateAllow(now.Add(time.Second)), "handshake should be allowed after token refill")
+	hm.Unlock()
+}
+
+func Test_HandshakeManagerRateLimitUnlimited(t *testing.T) {
+	l := test.NewLogger()
+	localrange := netip.MustParsePrefix("10.1.1.1/24")
+	preferredRanges := []netip.Prefix{localrange}
+	mainHM := newHostMap(l)
+	mainHM.preferredRanges.Store(&preferredRanges)
+
+	lh := newTestLighthouse()
+
+	cs := &CertState{
+		initiatingVersion: cert.Version1,
+		privateKey:        []byte{},
+		v1Cert:            &dummyCert{version: cert.Version1},
+		v1HandshakeBytes:  []byte{},
+	}
+
+	// Default config has maxHandshakeRate=0 (unlimited)
+	hm := NewHandshakeManager(l, mainHM, lh, &udp.NoopConn{}, defaultHandshakeConfig)
+	hm.f = &Interface{handshakeManager: hm, pki: &PKI{}, l: l}
+	hm.f.pki.cs.Store(cs)
+
+	// Should allow many handshakes with no limit
+	// Limited to 10 due to test lighthouse query channel buffer
+	for i := 0; i < 10; i++ {
+		ip := netip.MustParseAddr("172.1.1.1").As16()
+		ip[15] = byte(i + 1)
+		addr := netip.AddrFrom16(ip)
+		h := hm.StartHandshake(addr, nil)
+		assert.NotNil(t, h, "handshake %d should be allowed with unlimited rate", i)
+	}
+}
+
 func testCountTimerWheelEntries(tw *LockingTimerWheel[netip.Addr]) (c int) {
 	for _, i := range tw.t.wheel {
 		n := i.Head

interface.go

@@ -263,6 +263,8 @@ func (f *Interface) run() {
 }
 
 func (f *Interface) listenOut(i int) {
+	runtime.LockOSThread()
+
 	var li udp.Conn
 	if i > 0 {
 		li = f.writers[i]

main.go

@@ -204,10 +204,11 @@ func Main(c *config.C, configTest bool, buildVersion string, logger *logrus.Logg
 	useRelays := c.GetBool("relay.use_relays", DefaultUseRelays) && !c.GetBool("relay.am_relay", false)
 
 	handshakeConfig := HandshakeConfig{
-		tryInterval:   c.GetDuration("handshakes.try_interval", DefaultHandshakeTryInterval),
+		tryInterval:      c.GetDuration("handshakes.try_interval", DefaultHandshakeTryInterval),
-		retries:       int64(c.GetInt("handshakes.retries", DefaultHandshakeRetries)),
+		retries:          int64(c.GetInt("handshakes.retries", DefaultHandshakeRetries)),
-		triggerBuffer: c.GetInt("handshakes.trigger_buffer", DefaultHandshakeTriggerBuffer),
+		triggerBuffer:    c.GetInt("handshakes.trigger_buffer", DefaultHandshakeTriggerBuffer),
-		useRelays:     useRelays,
+		useRelays:        useRelays,
+		maxHandshakeRate: c.GetInt("handshakes.max_rate", DefaultMaxHandshakeRate),
 
 		messageMetrics: messageMetrics,
 	}

outside.go

@@ -190,6 +190,13 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
 		if !f.handleEncrypted(ci, via, h) {
 			return
 		}
+		_, err = f.decrypt(hostinfo, h.MessageCounter, out, packet, h, nb)
+		if err != nil {
+			hostinfo.logger(f.l).WithError(err).WithField("from", via).
+				WithField("packet", packet).
+				Error("Failed to decrypt CloseTunnel packet")
+			return
+		}
 
 		hostinfo.logger(f.l).WithField("from", via).
 			Info("Close tunnel received, tearing down.")

ssh.go

@@ -632,6 +632,9 @@ func sshCreateTunnel(ifce *Interface, fs any, a []string, w sshd.StringWriter) e
 	}
 
 	hostInfo = ifce.handshakeManager.StartHandshake(vpnAddr, nil)
+	if hostInfo == nil {
+		return w.WriteLine("Handshake rate limit reached")
+	}
 	if addr.IsValid() {
 		hostInfo.SetRemote(addr)
 	}

udp/udp_linux.go

@@ -4,7 +4,6 @@
 package udp
 
 import (
-	"context"
 	"encoding/binary"
 	"fmt"
 	"net"
@@ -19,66 +18,58 @@ import (
 )
 
 type StdConn struct {
-	udpConn *net.UDPConn
+	sysFd int
-	rawConn syscall.RawConn
+	isV4  bool
-	sockFd  int
+	l     *logrus.Logger
-	isV4    bool
+	batch int
-	l       *logrus.Logger
-	batch   int
 }
 
-func setReusePort(network, address string, c syscall.RawConn) error {
+func maybeIPV4(ip net.IP) (net.IP, bool) {
-	var opErr error
+	ip4 := ip.To4()
-	err := c.Control(func(fd uintptr) {
+	if ip4 != nil {
-		opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
+		return ip4, true
-		//CloseOnExec already set by the runtime
-	})
-	if err != nil {
-		return err
 	}
-	return opErr
+	return ip, false
 }
 
 func NewListener(l *logrus.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) {
-	listen := netip.AddrPortFrom(ip, uint16(port))
+	af := unix.AF_INET6
-	lc := net.ListenConfig{}
+	if ip.Is4() {
-	if multi {
+		af = unix.AF_INET
-		lc.Control = setReusePort
 	}
-	network := "udp"
+	syscall.ForkLock.RLock()
-	if ip.Is4() { //todo I am not a fan of this, would prefer a default where we always bind dual-stack unless given a non-wildcard v4 addr
+	fd, err := unix.Socket(af, unix.SOCK_DGRAM, unix.IPPROTO_UDP)
-		network = "udp4"
+	if err == nil {
+		unix.CloseOnExec(fd)
 	}
-	//this context is only used during the bind operation, you can't cancel it to kill the socket
+	syscall.ForkLock.RUnlock()
-	pc, err := lc.ListenPacket(context.Background(), network, listen.String())
+
 	if err != nil {
+		unix.Close(fd)
 		return nil, fmt.Errorf("unable to open socket: %s", err)
 	}
-	udpConn := pc.(*net.UDPConn)
+
-	rawConn, err := udpConn.SyscallConn()
+	if multi {
-	if err != nil {
+		if err = unix.SetsockoptInt(fd, unix.SOL_SOCKET, unix.SO_REUSEPORT, 1); err != nil {
-		_ = udpConn.Close()
+			return nil, fmt.Errorf("unable to set SO_REUSEPORT: %s", err)
-		return nil, err
+		}
 	}
 
-	//steal the socket's fd for sending
+	var sa unix.Sockaddr
-	var sockFd uintptr
+	if ip.Is4() {
-	err = rawConn.Control(func(fd uintptr) {
+		sa4 := &unix.SockaddrInet4{Port: port}
-		sockFd = fd
+		sa4.Addr = ip.As4()
-	})
+		sa = sa4
-	if err != nil {
+	} else {
-		_ = udpConn.Close()
+		sa6 := &unix.SockaddrInet6{Port: port}
-		return nil, err
+		sa6.Addr = ip.As16()
+		sa = sa6
+	}
+	if err = unix.Bind(fd, sa); err != nil {
+		return nil, fmt.Errorf("unable to bind to socket: %s", err)
 	}
 
-	return &StdConn{
+	return &StdConn{sysFd: fd, isV4: ip.Is4(), l: l, batch: batch}, err
-		udpConn: udpConn,
-		rawConn: rawConn,
-		sockFd:  int(sockFd),
-		isV4:    ip.Is4(),
-		l:       l,
-		batch:   batch,
-	}, err
 }
 
 func (u *StdConn) SupportsMultipleReaders() bool {
@@ -89,126 +80,63 @@ func (u *StdConn) Rebind() error {
 	return nil
 }
 
-func (u *StdConn) getSockOptInt(opt int) (int, error) {
-	if u.rawConn == nil {
-		return 0, fmt.Errorf("no UDP connection")
-	}
-	var out int
-	var opErr error
-	err := u.rawConn.Control(func(fd uintptr) {
-		out, opErr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, opt)
-	})
-	if err != nil {
-		return 0, err
-	}
-	return out, opErr
-}
-
-func (u *StdConn) setSockOptInt(opt int, n int) error {
-	if u.rawConn == nil {
-		return fmt.Errorf("no UDP connection")
-	}
-	var opErr error
-	err := u.rawConn.Control(func(fd uintptr) {
-		opErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, opt, n)
-	})
-	if err != nil {
-		return err
-	}
-	return opErr
-}
-
 func (u *StdConn) SetRecvBuffer(n int) error {
-	return u.setSockOptInt(unix.SO_RCVBUFFORCE, n)
+	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_RCVBUFFORCE, n)
 }
 
 func (u *StdConn) SetSendBuffer(n int) error {
-	return u.setSockOptInt(unix.SO_SNDBUFFORCE, n)
+	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_SNDBUFFORCE, n)
 }
 
 func (u *StdConn) SetSoMark(mark int) error {
-	return u.setSockOptInt(unix.SO_MARK, mark)
+	return unix.SetsockoptInt(u.sysFd, unix.SOL_SOCKET, unix.SO_MARK, mark)
 }
 
 func (u *StdConn) GetRecvBuffer() (int, error) {
-	return u.getSockOptInt(unix.SO_RCVBUF)
+	return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_RCVBUF)
 }
 
 func (u *StdConn) GetSendBuffer() (int, error) {
-	return u.getSockOptInt(unix.SO_SNDBUF)
+	return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_SNDBUF)
 }
 
 func (u *StdConn) GetSoMark() (int, error) {
-	return u.getSockOptInt(unix.SO_MARK)
+	return unix.GetsockoptInt(int(u.sysFd), unix.SOL_SOCKET, unix.SO_MARK)
 }
 
 func (u *StdConn) LocalAddr() (netip.AddrPort, error) {
-	addr := u.udpConn.LocalAddr()
+	sa, err := unix.Getsockname(u.sysFd)
-	return netip.ParseAddrPort(addr.String())
+	if err != nil {
-}
+		return netip.AddrPort{}, err
-
-func recvmmsg(fd uintptr, msgs []rawMessage) (int, bool, error) {
-	var errno syscall.Errno
-	n, _, errno := unix.Syscall6(
-		unix.SYS_RECVMMSG,
-		fd,
-		uintptr(unsafe.Pointer(&msgs[0])),
-		uintptr(len(msgs)),
-		unix.MSG_WAITFORONE,
-		0,
-		0,
-	)
-	if errno == syscall.EAGAIN || errno == syscall.EWOULDBLOCK {
-		// No data available, block for I/O and try again.
-		return int(n), false, nil
 	}
-	if errno != 0 {
-		return int(n), true, &net.OpError{Op: "recvmmsg", Err: errno}
-	}
-	return int(n), true, nil
-}
 
-func (u *StdConn) listenOutSingle(r EncReader) {
+	switch sa := sa.(type) {
-	var err error
+	case *unix.SockaddrInet4:
-	var n int
+		return netip.AddrPortFrom(netip.AddrFrom4(sa.Addr), uint16(sa.Port)), nil
-	var from netip.AddrPort
-	buffer := make([]byte, MTU)
 
-	for {
+	case *unix.SockaddrInet6:
-		n, from, err = u.udpConn.ReadFromUDPAddrPort(buffer)
+		return netip.AddrPortFrom(netip.AddrFrom16(sa.Addr), uint16(sa.Port)), nil
-		if err != nil {
+
-			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
+	default:
-			return
+		return netip.AddrPort{}, fmt.Errorf("unsupported sock type: %T", sa)
-		}
-		from = netip.AddrPortFrom(from.Addr().Unmap(), from.Port())
-		r(from, buffer[:n])
 	}
 }
 
-func (u *StdConn) listenOutBatch(r EncReader) {
+func (u *StdConn) ListenOut(r EncReader) {
 	var ip netip.Addr
-	var n int
-	var operr error
 
 	msgs, buffers, names := u.PrepareRawMessages(u.batch)
-
+	read := u.ReadMulti
-	//reader needs to capture variables from this function, since it's used as a lambda with rawConn.Read
+	if u.batch == 1 {
-	//defining it outside the loop so it gets re-used
+		read = u.ReadSingle
-	reader := func(fd uintptr) (done bool) {
-		n, done, operr = recvmmsg(fd, msgs)
-		return done
 	}
 
 	for {
-		err := u.rawConn.Read(reader)
+		n, err := read(msgs)
 		if err != nil {
 			u.l.WithError(err).Debug("udp socket is closed, exiting read loop")
 			return
 		}
-		if operr != nil {
-			u.l.WithError(operr).Debug("operr: udp socket is closed, exiting read loop")
-			return
-		}
 
 		for i := 0; i < n; i++ {
 			// Its ok to skip the ok check here, the slicing is the only error that can occur and it will panic
@@ -222,14 +150,44 @@ func (u *StdConn) listenOutBatch(r EncReader) {
 	}
 }
 
-func (u *StdConn) ListenOut(r EncReader) {
+func (u *StdConn) ReadSingle(msgs []rawMessage) (int, error) {
-	if u.batch == 1 {
+	for {
-		//save some ram by not calling PrepareRawMessages for fields we won't use
+		n, _, err := unix.Syscall6(
-		//we could also make this path more common by calling recvmmsg with msgs[:1],
+			unix.SYS_RECVMSG,
-		//but that's still the recvmmsg syscall, which would be a change
+			uintptr(u.sysFd),
-		u.listenOutSingle(r)
+			uintptr(unsafe.Pointer(&(msgs[0].Hdr))),
-	} else {
+			0,
-		u.listenOutBatch(r)
+			0,
+			0,
+			0,
+		)
+
+		if err != 0 {
+			return 0, &net.OpError{Op: "recvmsg", Err: err}
+		}
+
+		msgs[0].Len = uint32(n)
+		return 1, nil
+	}
+}
+
+func (u *StdConn) ReadMulti(msgs []rawMessage) (int, error) {
+	for {
+		n, _, err := unix.Syscall6(
+			unix.SYS_RECVMMSG,
+			uintptr(u.sysFd),
+			uintptr(unsafe.Pointer(&msgs[0])),
+			uintptr(len(msgs)),
+			unix.MSG_WAITFORONE,
+			0,
+			0,
+		)
+
+		if err != 0 {
+			return 0, &net.OpError{Op: "recvmmsg", Err: err}
+		}
+
+		return int(n), nil
 	}
 }
 
@@ -249,7 +207,7 @@ func (u *StdConn) writeTo6(b []byte, ip netip.AddrPort) error {
 	for {
 		_, _, err := unix.Syscall6(
 			unix.SYS_SENDTO,
-			uintptr(u.sockFd),
+			uintptr(u.sysFd),
 			uintptr(unsafe.Pointer(&b[0])),
 			uintptr(len(b)),
 			uintptr(0),
@@ -278,7 +236,7 @@ func (u *StdConn) writeTo4(b []byte, ip netip.AddrPort) error {
 	for {
 		_, _, err := unix.Syscall6(
 			unix.SYS_SENDTO,
-			uintptr(u.sockFd),
+			uintptr(u.sysFd),
 			uintptr(unsafe.Pointer(&b[0])),
 			uintptr(len(b)),
 			uintptr(0),
@@ -344,28 +302,15 @@ func (u *StdConn) ReloadConfig(c *config.C) {
 
 func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error {
 	var vallen uint32 = 4 * unix.SK_MEMINFO_VARS
-
+	_, _, err := unix.Syscall6(unix.SYS_GETSOCKOPT, uintptr(u.sysFd), uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
-	if u.rawConn == nil {
+	if err != 0 {
-		return fmt.Errorf("no UDP connection")
-	}
-	var opErr error
-	err := u.rawConn.Control(func(fd uintptr) {
-		_, _, syserr := unix.Syscall6(unix.SYS_GETSOCKOPT, fd, uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)
-		if syserr != 0 {
-			opErr = syserr
-		}
-	})
-	if err != nil {
 		return err
 	}
-	return opErr
+	return nil
 }
 
 func (u *StdConn) Close() error {
-	if u.udpConn != nil {
+	return syscall.Close(u.sysFd)
-		return u.udpConn.Close()
-	}
-	return nil
 }
 
 func NewUDPStatsEmitter(udpConns []Conn) func() {