diff --git a/control.go b/control.go index 41bc4f97..e2df9de8 100644 --- a/control.go +++ b/control.go @@ -203,7 +203,8 @@ func (c *Control) RebindUDPServer() { } // A failure here means we are likely still pinned to the interface we came up on, so the rest of this is - // unlikely to help. Say so instead of silently carrying on as if we rebound. + // unlikely to help. Say so instead of silently carrying on as if we rebound. Only the first writer is rebound; + // darwin's Rebind clears every listen routine's listener. if err := c.f.outside.Rebind(); err != nil { c.l.Error("Failed to rebind udp socket", "error", err) } diff --git a/established_remotes.go b/established_remotes.go new file mode 100644 index 00000000..9ce287b3 --- /dev/null +++ b/established_remotes.go @@ -0,0 +1,90 @@ +package nebula + +import ( + "net/netip" + "sync" + + "github.com/slackhq/nebula/udp" +) + +// establishedRemotes tells the udp conns that implement udp.EstablishedPeerConn which addresses are the current +// direct remote of at least one established tunnel. A hostinfo holds a reference on its remote while it is in the +// main hostmap, and SetRemote moves the reference; a relayed tunnel has no direct remote and holds none. +type establishedRemotes struct { + mu sync.Mutex + conns []udp.EstablishedPeerConn + refs map[netip.AddrPort]int +} + +// newEstablishedRemotes returns nil when no conn wants to know, which every method accepts. +func newEstablishedRemotes(conns []udp.Conn) *establishedRemotes { + var pc []udp.EstablishedPeerConn + for _, c := range conns { + if p, ok := c.(udp.EstablishedPeerConn); ok { + pc = append(pc, p) + } + } + if len(pc) == 0 { + return nil + } + return &establishedRemotes{conns: pc, refs: map[netip.AddrPort]int{}} +} + +// add makes hostinfo hold a reference on its remote, as it joins the main hostmap. A hostinfo holds references in +// one establishedRemotes at a time, so an add while it is in another is ignored. +func (e *establishedRemotes) add(hostinfo *HostInfo) { + if e == nil { + return + } + e.mu.Lock() + defer e.mu.Unlock() + if !hostinfo.establishedIn.CompareAndSwap(nil, e) && hostinfo.establishedIn.Load() != e { + return + } + e.syncLocked(hostinfo) +} + +// remove drops hostinfo's reference, as it leaves the main hostmap. +func (e *establishedRemotes) remove(hostinfo *HostInfo) { + if e == nil { + return + } + e.mu.Lock() + defer e.mu.Unlock() + if hostinfo.establishedIn.CompareAndSwap(e, nil) { + e.syncLocked(hostinfo) + } +} + +// syncLocked moves hostinfo's reference to the remote it should hold now: its current remote while it is +// established, none otherwise. Reading both under mu makes the last call win however add, remove and SetRemote +// interleave. Only the establishedRemotes that hostinfo is in, or has just left, may call it, since +// hostinfo.establishedRemote is a reference in that one's refs. +func (e *establishedRemotes) syncLocked(hostinfo *HostInfo) { + var want netip.AddrPort + if hostinfo.establishedIn.Load() == e { + want = hostinfo.GetRemote() + } + held := hostinfo.establishedRemote + if want == held { + return + } + if held.IsValid() { + e.refs[held]-- + if e.refs[held] == 0 { + delete(e.refs, held) + for _, c := range e.conns { + c.SetEstablishedPeer(held, false) + } + } + } + if want.IsValid() { + e.refs[want]++ + if e.refs[want] == 1 { + for _, c := range e.conns { + c.SetEstablishedPeer(want, true) + } + } + } + hostinfo.establishedRemote = want +} diff --git a/established_remotes_test.go b/established_remotes_test.go new file mode 100644 index 00000000..c500e035 --- /dev/null +++ b/established_remotes_test.go @@ -0,0 +1,173 @@ +package nebula + +import ( + "net/netip" + "sync" + "testing" + + "github.com/slackhq/nebula/test" + "github.com/slackhq/nebula/udp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// establishedPeerConn records which destinations the hostmap has told it are established peers, and fails the test +// on a call that doesn't change anything, which would mean a reference count went wrong. +type establishedPeerConn struct { + udp.NoopConn + t *testing.T + mu sync.Mutex + peers map[netip.AddrPort]bool +} + +func (c *establishedPeerConn) SetEstablishedPeer(dst netip.AddrPort, established bool) { + c.mu.Lock() + defer c.mu.Unlock() + assert.NotEqual(c.t, established, c.peers[dst], "SetEstablishedPeer(%v, %v) repeated", dst, established) + if established { + c.peers[dst] = true + } else { + delete(c.peers, dst) + } +} + +func (c *establishedPeerConn) established() []netip.AddrPort { + c.mu.Lock() + defer c.mu.Unlock() + var out []netip.AddrPort + for dst := range c.peers { + out = append(out, dst) + } + return out +} + +// newEstablishedTestHostMap returns a hostmap whose conns are a recording conn and one that doesn't track peers. +func newEstablishedTestHostMap(t *testing.T) (*HostMap, *establishedPeerConn) { + hm := newHostMap(test.NewLogger()) + hm.preferredRanges.Store(&[]netip.Prefix{netip.MustParsePrefix("192.168.0.0/16")}) + c := &establishedPeerConn{t: t, peers: map[netip.AddrPort]bool{}} + hm.established = newEstablishedRemotes([]udp.Conn{c, udp.NoopConn{}}) + require.NotNil(t, hm.established) + return hm, c +} + +func newEstablishedTestHostInfo(vpnAddr netip.Addr, index uint32) *HostInfo { + return &HostInfo{ + vpnAddrs: []netip.Addr{vpnAddr}, + localIndexId: index, + remotes: NewRemoteList([]netip.Addr{vpnAddr}, nil), + } +} + +func TestNewEstablishedRemotesWithoutTrackingConns(t *testing.T) { + assert.Nil(t, newEstablishedRemotes([]udp.Conn{udp.NoopConn{}})) +} + +// A conn hears about a remote while at least one hostinfo in the main hostmap has it as its direct remote: from +// the add, through roaming and SetRemoteIfPreferred, until the last such hostinfo is deleted or retired by +// MaxHostInfosPerVpnIp. Pending and relayed hostinfos never count. +func TestHostMap_EstablishedRemotes(t *testing.T) { + hm, c := newEstablishedTestHostMap(t) + f := &Interface{} + a := netip.MustParseAddr("10.0.0.1") + r1 := netip.MustParseAddrPort("203.0.113.1:4242") + r2 := netip.MustParseAddrPort("203.0.113.2:4242") + r3 := netip.MustParseAddrPort("192.168.1.3:4242") + + // A pending hostinfo, as a handshake sets its remote, isn't established. + h1 := newEstablishedTestHostInfo(a, 1) + h1.SetRemote(r1) + assert.Empty(t, c.established()) + + // Handshake completion adds it to the main hostmap. + hm.unlockedAddHostInfo(h1, f) + assert.ElementsMatch(t, []netip.AddrPort{r1}, c.established()) + + // A relayed tunnel has no direct remote. + relayed := newEstablishedTestHostInfo(netip.MustParseAddr("10.0.0.2"), 2) + hm.unlockedAddHostInfo(relayed, f) + assert.ElementsMatch(t, []netip.AddrPort{r1}, c.established()) + + // A rehandshake's new hostinfo shares the remote, which stays established until both are gone. + h3 := newEstablishedTestHostInfo(a, 3) + h3.SetRemote(r1) + hm.unlockedAddHostInfo(h3, f) + hm.DeleteHostInfo(h1) + assert.ElementsMatch(t, []netip.AddrPort{r1}, c.established()) + + // Roaming moves the reference. + h3.SetRemote(r2) + assert.ElementsMatch(t, []netip.AddrPort{r2}, c.established()) + require.True(t, h3.SetRemoteIfPreferred(hm, ViaSender{UdpAddr: r3})) + assert.ElementsMatch(t, []netip.AddrPort{r3}, c.established()) + + // A relayed tunnel that hears from its peer directly roams onto a direct remote. + relayed.SetRemote(r1) + assert.ElementsMatch(t, []netip.AddrPort{r1, r3}, c.established()) + hm.DeleteHostInfo(relayed) + + // A deleted hostinfo lets go, and a SetRemote after that changes nothing. + hm.DeleteHostInfo(h3) + assert.Empty(t, c.established()) + h3.SetRemote(r2) + assert.Empty(t, c.established()) + hm.DeleteHostInfo(h3) + + // Adding past MaxHostInfosPerVpnIp retires the oldest hostinfo and its remote. + var hs []*HostInfo + for i := range MaxHostInfosPerVpnIp + 1 { + h := newEstablishedTestHostInfo(a, uint32(10+i)) + h.SetRemote(netip.AddrPortFrom(netip.MustParseAddr("198.51.100.1"), uint16(1000+i))) + hm.unlockedAddHostInfo(h, f) + hs = append(hs, h) + } + assert.Len(t, c.established(), MaxHostInfosPerVpnIp) + assert.NotContains(t, c.established(), hs[0].GetRemote()) +} + +// Roaming on the receive path races tunnel teardown on the connection manager; whatever the order, a deleted +// hostinfo holds no reference. +func TestHostMap_EstablishedRemotesRoamRacesDelete(t *testing.T) { + hm, c := newEstablishedTestHostMap(t) + f := &Interface{} + a := netip.MustParseAddr("10.0.0.1") + for i := range 200 { + h := newEstablishedTestHostInfo(a, uint32(i+1)) + h.SetRemote(netip.MustParseAddrPort("203.0.113.1:4242")) + hm.Lock() + hm.unlockedAddHostInfo(h, f) + hm.Unlock() + var wg sync.WaitGroup + wg.Go(func() { h.SetRemote(netip.AddrPortFrom(netip.MustParseAddr("203.0.113.2"), uint16(i+1))) }) + wg.Go(func() { hm.DeleteHostInfo(h) }) + wg.Wait() + require.Empty(t, c.established(), "round %d", i) + } +} + +// A hostinfo holds its reference in the first tracking hostmap it joins; another's add and delete of it change +// nothing, so neither hostmap's counts go wrong. +func TestHostMap_EstablishedRemotesSingleOwner(t *testing.T) { + hm1, c1 := newEstablishedTestHostMap(t) + hm2, c2 := newEstablishedTestHostMap(t) + f := &Interface{} + r := netip.MustParseAddrPort("203.0.113.1:4242") + h := newEstablishedTestHostInfo(netip.MustParseAddr("10.0.0.1"), 1) + h.SetRemote(r) + + hm1.unlockedAddHostInfo(h, f) + hm2.unlockedAddHostInfo(h, f) + assert.ElementsMatch(t, []netip.AddrPort{r}, c1.established()) + assert.Empty(t, c2.established()) + + hm2.DeleteHostInfo(h) + assert.Equal(t, map[netip.AddrPort]int{r: 1}, hm1.established.refs) + assert.Empty(t, hm2.established.refs) + assert.ElementsMatch(t, []netip.AddrPort{r}, c1.established()) + + hm1.DeleteHostInfo(h) + assert.Empty(t, hm1.established.refs) + assert.Empty(t, hm2.established.refs) + assert.Empty(t, c1.established()) + assert.Empty(t, c2.established()) +} diff --git a/hostmap.go b/hostmap.go index 71814a2f..2fe2ffc5 100644 --- a/hostmap.go +++ b/hostmap.go @@ -69,7 +69,9 @@ type HostMap struct { Hosts map[netip.Addr]*HostInfo moreHosts map[netip.Addr][]*HostInfo preferredRanges atomic.Pointer[[]netip.Prefix] - l *slog.Logger + // established hears about each hostinfo that joins or leaves the hostmap; nil when no conn wants to know. + established *establishedRemotes + l *slog.Logger } // For synchronization, treat the pointed-to Relay struct as immutable. To edit the Relay @@ -276,6 +278,11 @@ type HostInfo struct { // This value will be behind against actual tunnel utilization in the hot path. // This should only be used by the ConnectionManagers ticker routine. lastUsed time.Time + + // establishedIn is set while the hostinfo is in a main hostmap whose conns track established remotes, and + // establishedRemote is the remote it holds a reference on there, guarded by establishedIn's mu. + establishedIn atomic.Pointer[establishedRemotes] + establishedRemote netip.AddrPort } type ViaSender struct { @@ -516,6 +523,7 @@ func (hm *HostMap) unlockedDeleteHostInfo(hostinfo *HostInfo) bool { if len(hm.Indexes) == 0 { hm.Indexes = map[uint32]*HostInfo{} } + hm.established.remove(hostinfo) if hm.l.Enabled(context.Background(), slog.LevelDebug) { hm.l.Debug("Hostmap hostInfo deleted", @@ -662,6 +670,7 @@ func (hm *HostMap) unlockedAddHostInfo(hostinfo *HostInfo, f *Interface) { hm.Indexes[hostinfo.localIndexId] = hostinfo hm.RemoteIndexes[hostinfo.remoteIndexId] = hostinfo + hm.established.add(hostinfo) hostinfo.markOut(f.rebindEpoch.Load()) if f.connectionManager != nil { // f.connectionManager is only nil in some unit tests @@ -842,6 +851,13 @@ func (i *HostInfo) SetRemote(remote netip.AddrPort) { if i.GetRemote() != remote { i.remote.Store(&remote) i.remotes.LearnRemote(i.vpnAddrs[0], remote) + if e := i.establishedIn.Load(); e != nil { + e.mu.Lock() + if i.establishedIn.Load() == e { + e.syncLocked(i) + } + e.mu.Unlock() + } } } diff --git a/main.go b/main.go index 52f7bea8..08c15cec 100644 --- a/main.go +++ b/main.go @@ -213,6 +213,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev } hostMap := NewHostMapFromConfig(l, c) + hostMap.established = newEstablishedRemotes(udpConns) punchy := NewPunchyFromConfig(l, c, udpConns[0]) connManager := newConnectionManagerFromConfig(l, c, hostMap, punchy) lightHouse, err := NewLightHouseFromConfig(ctx, l, c, pki.getCertState(), udpConns[0], punchy) diff --git a/udp/conn.go b/udp/conn.go index 6d20cbbf..2b8e0751 100644 --- a/udp/conn.go +++ b/udp/conn.go @@ -48,6 +48,14 @@ type Conn interface { Close() error } +// EstablishedPeerConn is a Conn that treats the remotes of established tunnels differently from other +// destinations, as darwin's gives busy ones connected sockets. +type EstablishedPeerConn interface { + // SetEstablishedPeer reports whether dst is the current remote of at least one established tunnel. Calls for + // one dst must not race each other. + SetEstablishedPeer(dst netip.AddrPort, established bool) +} + type NoopConn struct{} func (NoopConn) Rebind() error { diff --git a/udp/connsock_bench_darwin_test.go b/udp/connsock_bench_darwin_test.go new file mode 100644 index 00000000..a35003b1 --- /dev/null +++ b/udp/connsock_bench_darwin_test.go @@ -0,0 +1,115 @@ +//go:build !e2e_testing + +package udp + +import ( + "log/slog" + "net" + "net/netip" + "testing" + "time" +) + +// benchConnSocks opens a wildcard listener with ListenOut running, with the connected socket knobs pinned to +// their defaults and then cfg applied, and a loopback peer that never reads, so its sends cost only the sender. +func benchConnSocks(b *testing.B, s Settings, cfg func(*connSockConfig)) (*StdConn, netip.AddrPort) { + b.Helper() + for k, v := range map[string]string{ + "NEBULA_CONNSOCKS": "8", + "NEBULA_CONNSOCK_RUN": "64", + "NEBULA_CONNSOCK_WINDOW": "1s", + "NEBULA_CONNSOCK_IDLE": "30s", + "NEBULA_CONNSOCK_OPEN_EVERY": "1s", + } { + b.Setenv(k, v) + } + c, err := NewListener(slog.New(slog.DiscardHandler), s) + if err != nil { + b.Fatal(err) + } + b.Cleanup(func() { _ = c.Close() }) + u := c.(*StdConn) + if cfg != nil { + cfg(&u.socks.cfg) + } + go func() { _ = u.ListenOut(func(netip.AddrPort, []byte) {}, func() {}) }() + for u.reader.Load() == nil { + time.Sleep(time.Millisecond) + } + pc, err := net.ListenPacket("udp4", "127.0.0.1:0") + if err != nil { + b.Fatal(err) + } + b.Cleanup(func() { _ = pc.Close() }) + return u, pc.LocalAddr().(*net.UDPAddr).AddrPort() +} + +// BenchmarkConnSockWriteTo measures one WriteTo of a full-size datagram down each path: the listener with +// connected sockets off, to a destination that is no tunnel's remote while another is, to an established peer +// that never gets busy, and on a connected socket. +func BenchmarkConnSockWriteTo(b *testing.B) { + pkt := make([]byte, 1400) + cases := []struct { + name string + cfg func(*connSockConfig) + setup func(b *testing.B, u *StdConn, dst netip.AddrPort) + }{ + {name: "listener/off", cfg: func(c *connSockConfig) { c.max = 0 }}, + {name: "listener/ineligible", setup: func(b *testing.B, u *StdConn, dst netip.AddrPort) { + u.SetEstablishedPeer(netip.AddrPortFrom(dst.Addr(), dst.Port()+1), true) + }}, + {name: "listener/eligible-not-busy", cfg: func(c *connSockConfig) { c.run = 1 << 30 }, + setup: func(b *testing.B, u *StdConn, dst netip.AddrPort) { u.SetEstablishedPeer(dst, true) }}, + {name: "connected", setup: func(b *testing.B, u *StdConn, dst netip.AddrPort) { + u.SetEstablishedPeer(dst, true) + for range u.socks.cfg.run { + _ = u.WriteTo(pkt, dst) + } + if u.connSockFor(dst) == nil { + b.Fatal("no connected socket") + } + }}, + } + for _, c := range cases { + b.Run(c.name, func(b *testing.B) { + u, dst := benchConnSocks(b, Settings{Listen: netip.MustParseAddrPort("0.0.0.0:0")}, c.cfg) + if c.setup != nil { + c.setup(b, u, dst) + } + b.ReportAllocs() + b.ResetTimer() + for range b.N { + _ = u.WriteTo(pkt, dst) + } + }) + } +} + +// BenchmarkConnSockDispatch measures what WriteTo adds before the listener's sendto, without the syscall: a v4 +// listener refuses a v6 destination before sending, with connected sockets off or with another peer established. +func BenchmarkConnSockDispatch(b *testing.B) { + pkt := make([]byte, 1400) + dst := netip.MustParseAddrPort("[::1]:4242") + for _, c := range []struct { + name string + cfg func(*connSockConfig) + establish bool + }{ + {name: "off", cfg: func(c *connSockConfig) { c.max = 0 }}, + {name: "ineligible", establish: true}, + } { + b.Run(c.name, func(b *testing.B) { + u, peer := benchConnSocks(b, Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0"), Multi: true}, c.cfg) + if c.establish { + u.SetEstablishedPeer(peer, true) + } + b.ReportAllocs() + b.ResetTimer() + for range b.N { + if u.WriteTo(pkt, dst) == nil { + b.Fatal("sent a v6 datagram on a v4 listener") + } + } + }) + } +} diff --git a/udp/connsock_darwin.go b/udp/connsock_darwin.go new file mode 100644 index 00000000..e4c9fb62 --- /dev/null +++ b/udp/connsock_darwin.go @@ -0,0 +1,803 @@ +//go:build !e2e_testing + +package udp + +import ( + "context" + "errors" + "fmt" + "log/slog" + "maps" + "net" + "net/netip" + "os" + "slices" + "strconv" + "sync" + "sync/atomic" + "syscall" + "time" + + "golang.org/x/sys/unix" +) + +// A connected socket is a UDP socket connected to one busy peer, on the listener's port and bound to the source +// address the listener's route to that peer picks, so the peer sees the same address and port from either. A send +// on a connected socket skips the per-datagram address and MAC policy checks, and the EndpointSecurity send check +// that xprotectd can hold every unconnected sendto on, and it is the only UDP send that reports the interface +// queue's flow control, as ENOBUFS. A connected socket wins xnu's lookup for its peer's datagrams; it never sees +// anyone else's. +// +// A connected socket sets SO_REUSEPORT and binds a specific address beside a wildcard listener, which sets +// SO_REUSEPORT only for listen routines. xnu refuses a wildcard SO_REUSEPORT bind beside a holder without the +// option, and refuses another uid's bind to an address a socket holds specifically, so no other user can join the +// port while connected sockets open and close: uid nobody got EADDRINUSE on every bind against a root listener +// with connected sockets in the 09-25 probes. A connected socket is closed, never disconnected, because xnu widens +// a disconnected socket's local address to the wildcard. +// +// Only the current remote of an established tunnel may get one: Nebula names those with SetEstablishedPeer, so an +// unauthenticated sender can't earn a connected socket by drawing replies such as recv_error or handshake +// responses. +type connSock struct { + dst netip.AddrPort + peer *connSockPeer + conn *net.UDPConn + rc syscall.RawConn + // used is the monotonic time in nanoseconds of the connected socket's last send or receive. + used atomic.Int64 + // stallAt is the monotonic time of the first ENOBUFS or EAGAIN since a send last went through, or 0, and + // dropping is set once that stall has dropped a datagram, so each stall logs its drops once. + stallAt atomic.Int64 + dropping atomic.Bool + // parkMu serializes the writes that wait for a full send buffer, since the write deadline bounding each wait + // is the descriptor's, not the writer's. + parkMu sync.Mutex + // sys writes one datagram on the descriptor: writeFd, or a stand-in for the interface in tests. + sys func(fd int, b []byte) error +} + +func writeFd(fd int, b []byte) error { + _, err := syscall.Write(fd, b) + return err +} + +// errStallSpent is a write that found the connected socket's stall budget spent, so its datagram is dropped. +var errStallSpent = errors.New("connected socket stall budget spent") + +// connSockWriter carries one write through RawConn.Write. They are pooled with fn bound once, since a closure +// passed through the RawConn interface escapes and would allocate on every send. +type connSockWriter struct { + s *connSock + b []byte + park bool + err error + fn func(fd uintptr) bool +} + +var connSockWriters = sync.Pool{New: func() any { + w := &connSockWriter{} + w.fn = w.write + return w +}} + +func (w *connSockWriter) write(fd uintptr) bool { + for { + w.err = w.s.sys(int(fd), w.b) + switch { + case w.err == syscall.EINTR: + case w.err == syscall.EAGAIN && w.park: + return false + default: + return true + } + } +} + +// try writes b on s once and returns nil, the write's errno, os.ErrDeadlineExceeded, or net.ErrClosed. With park +// set, EAGAIN waits in the netpoller until the socket is writable or the write deadline passes; without it, EAGAIN +// is returned. +func (s *connSock) try(b []byte, park bool) error { + w := connSockWriters.Get().(*connSockWriter) + w.s, w.b, w.park = s, b, park + rerr := s.rc.Write(w.fn) + err := w.err + *w = connSockWriter{fn: w.fn} + connSockWriters.Put(w) + switch { + case rerr == nil: + return err + case errors.Is(rerr, os.ErrDeadlineExceeded): + return os.ErrDeadlineExceeded + default: + return net.ErrClosed + } +} + +// connSockConfig holds the knobs. connSockConfigFromEnv reads all but backoff and enobufsMax from the environment, +// so one binary can be tuned without a rebuild. +type connSockConfig struct { + // max is how many connected sockets the listen routines may have open at once; 0 turns connected sockets off. + max int + // run is how many datagrams to one destination within one window earn it a connected socket. darwin's tun + // hands over one packet per read, so a peer's sends arrive one WriteTo or one-datagram batch at a time and + // busyness has to be counted across calls. + run int + window time.Duration + // idle closes a connected socket after it has neither sent nor received for this long. + idle time.Duration + // backoff keeps a destination whose connected socket failed on the listener for this long. + backoff time.Duration + // openEvery is how often the open bucket earns back an open once max are spent, which bounds connected socket + // churn however fast senders earn connected sockets and errors close them. + openEvery time.Duration + // enobufsMax is how long a connected socket's sends may wait out ENOBUFS without one going through before they + // drop instead, so a suspended interface costs one write per datagram rather than stalling the tun reader. + enobufsMax time.Duration +} + +func connSockConfigFromEnv() connSockConfig { + c := connSockConfig{ + max: 8, + run: 64, + window: time.Second, + idle: 30 * time.Second, + backoff: 30 * time.Second, + openEvery: time.Second, + enobufsMax: 10 * time.Millisecond, + } + if v, err := strconv.Atoi(os.Getenv("NEBULA_CONNSOCKS")); err == nil { + c.max = v + } + if v, err := strconv.Atoi(os.Getenv("NEBULA_CONNSOCK_RUN")); err == nil && v > 0 { + c.run = v + } + if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_WINDOW")); err == nil && v > 0 { + c.window = v + } + if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_IDLE")); err == nil && v > 0 { + c.idle = v + } + if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_OPEN_EVERY")); err == nil && v > 0 { + c.openEvery = v + } + return c +} + +// connSockGroups holds the connSocks that the listen routines bound to one address share, so the limits, the +// connected sockets and Rebind cover every routine, and a peer has at most one connected socket. +var connSockGroups = struct { + sync.Mutex + m map[netip.AddrPort]*connSocks +}{m: map[netip.AddrPort]*connSocks{}} + +// joinConnSocks returns the connSocks u shares with the other listen routines bound to la, and makes it for the +// first. +func joinConnSocks(u *StdConn, la netip.AddrPort, s Settings) *connSocks { + g := &connSockGroups + g.Lock() + defer g.Unlock() + if cs := g.m[la]; s.Multi && cs != nil { + cs.mu.Lock() + cs.members = append(cs.members, u) + cs.mu.Unlock() + return cs + } + cs := &connSocks{cfg: connSockConfigFromEnv(), key: la, members: []*StdConn{u}, clock: monoNow, sleep: time.Sleep} + if cs.cfg.max > 0 && !u.listenHost.IsUnspecified() && !s.Multi { + // A connected socket on a specific listen.host binds the listener's own address, which xnu allows only when + // both sockets set SO_REUSEPORT, and a listener with it lets another uid bind a dual-stack wildcard on the + // port. + cs.off.Store(true) + u.l.Debug("connected sockets: off for a specific listen.host", "addr", s.Listen) + } + if s.Multi { + g.m[la] = cs + } + return cs +} + +type connSocks struct { + cfg connSockConfig + // key is the listeners' bound address, which connSockGroups files this under. + key netip.AddrPort + // off keeps connected sockets from opening, when the listener's address can't be shared with them. + off atomic.Bool + // table maps each destination that is the remote of an established tunnel, and that a connected socket could + // reach, to its peer; it is nil when there are none, so a send with nothing to consider costs one load. Every + // send reads it without a lock, and mu guards copying it. An open that holds mu and still finds its peer here + // can't outlive the tunnel. + table atomic.Pointer[map[netip.AddrPort]*connSockPeer] + // reserved counts the connected sockets that are open, opening or closing, against cfg.max; mu guards changing it. + reserved atomic.Int32 + mu sync.Mutex + closed bool + // members are the listen routines' listeners that share these connected sockets. + members []*StdConn + // pending holds each destination whose connected socket is opening or closing, and not on its peer. xnu + // refuses to connect a 4-tuple that is still bound, so a destination in pending can't open another. + pending map[netip.AddrPort]struct{} + // rebinds counts Rebinds, so an open that copied the listener's interface before one is discarded. + rebinds uint64 + // closers tracks the goroutines that close detached connected sockets. + closers sync.WaitGroup + // done stops the reaper; it is made when ListenOut starts the connected sockets. + done chan struct{} + // opens bounds connected socket opens to cfg.max at once and one per cfg.openEvery after. + opens gcra + // logs bounds connected socket log lines at info or warn; see connSockLogBurst. + logs gcra + // clock returns monotonic nanoseconds and sleep waits out ENOBUFS: monoNow and time.Sleep, or stand-ins in + // tests. + clock func() int64 + sleep func(time.Duration) + // dialHook and closeHook, when set, run just before an open's dials and a close's Close, outside mu; tests + // use them to hold an open or close in flight. + dialHook, closeHook func(dst netip.AddrPort) +} + +// connSockPeer is what connSocks keeps about an established tunnel's remote: its connected socket, how busy it is, +// and how long a failed connected socket keeps it on the listener. +type connSockPeer struct { + sock atomic.Pointer[connSock] + // windowMu guards start, the monotonic time the current window began, and n, the datagrams sent in it, so a + // window's reset and its counts can't interleave with another sender's. + windowMu sync.Mutex + start int64 + n int + // retryAt is the monotonic time before which a peer whose connected socket failed stays on the listener; + // connSocks.mu guards it. + retryAt int64 +} + +// busy counts n more datagrams to p and reports whether it has now sent cfg.run within one window. A busy peer's +// count starts over, so a peer that can't get a connected socket retries at most once per cfg.run sends. +func (p *connSockPeer) busy(now int64, n int, cfg *connSockConfig) bool { + p.windowMu.Lock() + defer p.windowMu.Unlock() + if time.Duration(now-p.start) > cfg.window { + p.start, p.n = now, 0 + } + p.n += n + if p.n < cfg.run { + return false + } + p.n = 0 + return true +} + +// connSockLogBurst and connSockLogEvery size the bucket that lets connected socket opens, closes and failures log +// at info or warn; past it they log at debug, so churn can't flood the log. +const ( + connSockLogBurst = 32 + connSockLogEvery = 10 * time.Second +) + +var monoStart = time.Now() + +func monoNow() int64 { return int64(time.Since(monoStart)) } + +// gcra is a token bucket kept as the monotonic time it is next full (the generic cell rate algorithm), so it needs +// no refill tick. +type gcra struct{ tat int64 } + +// take spends a token from a bucket of burst that earns one back per every, and reports whether there was one to +// spend. +func (g *gcra) take(now int64, burst int, every time.Duration) bool { + e := int64(every) + tat := max(g.tat, now) + if tat-now > int64(burst-1)*e { + return false + } + g.tat = tat + e + return true +} + +// logLevelLocked returns level while the log bucket has a token, and debug once it is spent. +func (cs *connSocks) logLevelLocked(level slog.Level) slog.Level { + if cs.logs.take(cs.clock(), connSockLogBurst, connSockLogEvery) { + return level + } + return slog.LevelDebug +} + +// startConnSocks starts the idle reaper. +func (u *StdConn) startConnSocks() { + cs := u.socks + cs.mu.Lock() + defer cs.mu.Unlock() + if cs.closed || cs.done != nil || cs.cfg.max <= 0 || cs.off.Load() { + return + } + cs.done = make(chan struct{}) + go u.reapConnSocks(cs.done) +} + +// stopConnSocks closes every connected socket and keeps new ones from opening, once any listen routine's listener +// closes, as they do together at shutdown. It returns once their descriptors are closed. +func (u *StdConn) stopConnSocks() { + cs := u.socks + cs.mu.Lock() + cs.members = slices.DeleteFunc(cs.members, func(m *StdConn) bool { return m == u }) + if !cs.closed { + cs.closed = true + if cs.done != nil { + close(cs.done) + } + u.detachAllLocked("close") + } + cs.mu.Unlock() + connSockGroups.Lock() + if connSockGroups.m[cs.key] == cs { + delete(connSockGroups.m, cs.key) + } + connSockGroups.Unlock() + cs.closers.Wait() +} + +// siblings returns the other listen routines' listeners. +func (cs *connSocks) siblings(u *StdConn) []*StdConn { + cs.mu.Lock() + defer cs.mu.Unlock() + return slices.DeleteFunc(slices.Clone(cs.members), func(m *StdConn) bool { return m == u }) +} + +// connSockFor returns dst's connected socket, or nil. +func (u *StdConn) connSockFor(dst netip.AddrPort) *connSock { + if p := u.socks.peer(dst); p != nil { + return p.sock.Load() + } + return nil +} + +func (cs *connSocks) peer(dst netip.AddrPort) *connSockPeer { + if t := cs.table.Load(); t != nil { + return (*t)[dst] + } + return nil +} + +// SetEstablishedPeer makes dst eligible for a connected socket while it is the remote of an established tunnel, +// and closes the one it has once it no longer is. Nebula calls it for every listen routine under the hostmap's +// write lock, so repeats are no-ops and it makes no syscalls: the descriptor is closed by another goroutine. +func (u *StdConn) SetEstablishedPeer(dst netip.AddrPort, established bool) { + cs := u.socks + if cs.cfg.max <= 0 || cs.off.Load() || !u.connSockable(dst) { + return + } + cs.mu.Lock() + defer cs.mu.Unlock() + t := cs.table.Load() + p := cs.peer(dst) + if cs.closed || (p != nil) == established { + return + } + next := map[netip.AddrPort]*connSockPeer{} + if t != nil { + next = maps.Clone(*t) + } + if established { + p = &connSockPeer{start: cs.clock()} + next[dst] = p + } else { + delete(next, dst) + if s := p.sock.Load(); s != nil { + u.detachLocked(s, "tunnel gone", false) + } + } + cs.storeTableLocked(next) +} + +// storeTableLocked publishes t, or nil for an empty one. +func (cs *connSocks) storeTableLocked(t map[netip.AddrPort]*connSockPeer) { + if len(t) == 0 { + cs.table.Store(nil) + return + } + cs.table.Store(&t) +} + +// connSockable reports whether a connected socket to dst could send what the listener would. A v4 listener can't +// reach a v6 destination at all, and a connected socket bound to a specific listen.host must share its family. +func (u *StdConn) connSockable(dst netip.AddrPort) bool { + if !dst.IsValid() || dst.Port() == 0 { + return false + } + if u.isV4 { + return dst.Addr().Is4() + } + return u.listenHost.IsUnspecified() || u.listenHost.Unmap().Is4() == dst.Addr().Unmap().Is4() +} + +// connSockRun sends bufs, all to p's destination, on p's connected socket, opening one if they make p busy. It +// returns how many went out; when that is short of len(bufs), err is the failure of bufs[sent], or nil to leave +// bufs[sent:] to the listener. +func (u *StdConn) connSockRun(p *connSockPeer, dst netip.AddrPort, bufs [][]byte) (sent int, err error) { + s := p.sock.Load() + if s == nil { + if s = u.maybeOpenConnSock(p, dst, len(bufs)); s == nil { + return 0, nil + } + } + sent, errno, closed := u.connSockWrite(s, bufs) + switch { + case sent == len(bufs): + case connSockDead(errno): + u.closeConnSock(s, errno.Error(), true) + case !closed: + return sent, &net.OpError{Op: "write", Err: errno} + } + return sent, nil +} + +// maybeOpenConnSock counts n datagrams to p, and opens a connected socket to it once it is busy and nothing rules +// one out. +func (u *StdConn) maybeOpenConnSock(p *connSockPeer, dst netip.AddrPort, n int) *connSock { + cs := u.socks + if n <= 0 || u.reader.Load() == nil || int(cs.reserved.Load()) >= cs.cfg.max || !p.busy(cs.clock(), n, &cs.cfg) { + return nil + } + cs.mu.Lock() + if cs.closed || cs.peer(dst) != p { + cs.mu.Unlock() + return nil + } + if s := p.sock.Load(); s != nil { + cs.mu.Unlock() + return s + } + now := cs.clock() + if _, ok := cs.pending[dst]; ok || int(cs.reserved.Load()) >= cs.cfg.max || now < p.retryAt || + !cs.opens.take(now, cs.cfg.max, cs.cfg.openEvery) { + cs.mu.Unlock() + return nil + } + if cs.pending == nil { + cs.pending = map[netip.AddrPort]struct{}{} + } + cs.pending[dst] = struct{}{} + cs.reserved.Add(1) + rebinds := cs.rebinds + cs.mu.Unlock() + + if cs.dialHook != nil { + cs.dialHook(dst) + } + s, err := u.dialConnSock(p, dst) + + cs.mu.Lock() + if err != nil { + delete(cs.pending, dst) + cs.reserved.Add(-1) + p.retryAt = cs.clock() + int64(cs.cfg.backoff) + level := cs.logLevelLocked(slog.LevelWarn) + cs.mu.Unlock() + u.l.Log(context.Background(), level, "connected socket: open failed, staying on the listener", "udpAddr", dst, "error", err) + return nil + } + if cs.closed || cs.rebinds != rebinds || cs.peer(dst) != p { + // The tunnel went away, or a Rebind or shutdown ran, while the socket was opening; it stays in pending until + // it is closed. + cs.mu.Unlock() + u.finishClose(s) + u.l.Debug("connected socket: discarded, its tunnel or route changed while it opened", "udpAddr", dst) + return nil + } + delete(cs.pending, dst) + p.sock.Store(s) + go u.readConnSock(s) + level := cs.logLevelLocked(slog.LevelInfo) + cs.mu.Unlock() + u.l.Log(context.Background(), level, "connected socket: opened", "udpAddr", dst, "local", s.conn.LocalAddr(), "open", cs.reserved.Load()) + return s +} + +// dialConnSock opens a socket on the listener's port with SO_REUSEPORT, bound to the configured listen.host or, +// for a wildcard listener, to the source address the route to dst picks, and connected to dst. It carries the +// listener's IP_BOUND_IF, so both route the same way. +func (u *StdConn) dialConnSock(p *connSockPeer, dst netip.AddrPort) (*connSock, error) { + raddr := net.UDPAddrFromAddrPort(netip.AddrPortFrom(dst.Addr().Unmap(), dst.Port())) + network := "udp6" + if raddr.IP.To4() != nil { + network = "udp4" + } + ifindex, err := u.boundIf() + if err != nil { + return nil, err + } + local := &net.UDPAddr{IP: u.listenHost.AsSlice(), Zone: u.listenHost.Zone()} + if u.listenHost.IsUnspecified() { + // A connected socket gets the source address xnu would pick for an unconnected send to dst. + d := net.Dialer{Control: connSockControl(false, ifindex)} + probe, err := d.Dial(network, raddr.String()) + if err != nil { + return nil, err + } + local = probe.LocalAddr().(*net.UDPAddr) + _ = probe.Close() + } + local.Port = int(u.port) + d := net.Dialer{LocalAddr: local, Control: connSockControl(true, ifindex)} + c, err := d.Dial(network, raddr.String()) + if err != nil { + return nil, err + } + uc := c.(*net.UDPConn) + rc, err := uc.SyscallConn() + if err != nil { + _ = uc.Close() + return nil, err + } + s := &connSock{dst: dst, peer: p, conn: uc, rc: rc, sys: writeFd} + s.used.Store(u.socks.clock()) + return s, nil +} + +// connSockControl sets SO_REUSEPORT when reuse is set, and IP_BOUND_IF or IPV6_BOUND_IF when ifindex isn't 0, +// before the socket binds. +func connSockControl(reuse bool, ifindex int) func(network, address string, c syscall.RawConn) error { + return func(network, _ string, c syscall.RawConn) error { + var serr error + err := c.Control(func(fd uintptr) { + if reuse { + serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1) + } + if serr == nil && ifindex != 0 { + if network == "udp4" { + serr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_BOUND_IF, ifindex) + } else { + serr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF, ifindex) + } + } + }) + if err != nil { + return err + } + return serr + } +} + +// boundIf returns the interface index the listener is scoped to, or 0. An error keeps the peer on the listener, +// since a connected socket opened without the listener's scope could route out an interface the listener can't. +func (u *StdConn) boundIf() (int, error) { + level, opt := unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF + if u.isV4 { + level, opt = unix.IPPROTO_IP, unix.IP_BOUND_IF + } + v, err := unix.GetsockoptInt(int(u.sysFd), level, opt) + if err != nil { + return 0, fmt.Errorf("reading the listener's bound interface: %w", err) + } + return v, nil +} + +// closeConnSock closes s if it is still open. With backoff set, the peer stays on the listener until the backoff +// runs out. +func (u *StdConn) closeConnSock(s *connSock, why string, backoff bool) { + u.socks.mu.Lock() + u.detachLocked(s, why, backoff) + u.socks.mu.Unlock() +} + +// detachLocked takes s off its peer, if it is still there, and closes it on another goroutine, since callers such +// as SetEstablishedPeer hold locks that no socket syscall should run under. s.dst stays in pending until the close +// completes. +func (u *StdConn) detachLocked(s *connSock, why string, backoff bool) { + cs := u.socks + if !s.peer.sock.CompareAndSwap(s, nil) { + return + } + if backoff { + s.peer.retryAt = cs.clock() + int64(cs.cfg.backoff) + } + if cs.pending == nil { + cs.pending = map[netip.AddrPort]struct{}{} + } + cs.pending[s.dst] = struct{}{} + level := cs.logLevelLocked(slog.LevelInfo) + cs.closers.Add(1) + go func() { + defer cs.closers.Done() + open := u.finishClose(s) + u.l.Log(context.Background(), level, "connected socket: closed", "udpAddr", s.dst, "reason", why, "open", open) + }() +} + +// finishClose closes s, whose destination is in pending, then lets the destination open again, and returns how +// many connected sockets are left. Closing waits only for a write already in a syscall: a write parked on a full +// buffer is woken, and one waiting out ENOBUFS holds no reference on the descriptor. +func (u *StdConn) finishClose(s *connSock) int32 { + cs := u.socks + if cs.closeHook != nil { + cs.closeHook(s.dst) + } + _ = s.conn.Close() + cs.mu.Lock() + defer cs.mu.Unlock() + delete(cs.pending, s.dst) + return cs.reserved.Add(-1) +} + +// detachAllLocked detaches every connected socket, for a network change or shutdown. +func (u *StdConn) detachAllLocked(why string) { + if t := u.socks.table.Load(); t != nil { + for _, p := range *t { + if s := p.sock.Load(); s != nil { + u.detachLocked(s, why, false) + } + } + } +} + +// closeConnSocks detaches every connected socket and discards the ones still opening, for a network change. +func (u *StdConn) closeConnSocks(why string) { + u.socks.mu.Lock() + u.socks.rebinds++ + u.detachAllLocked(why) + u.socks.mu.Unlock() +} + +func (u *StdConn) reapConnSocks(done <-chan struct{}) { + cs := u.socks + idle := cs.cfg.idle + tick := time.NewTicker(max(min(idle/4, 5*time.Second), time.Millisecond)) + defer tick.Stop() + for { + select { + case <-done: + return + case <-tick.C: + } + t := cs.table.Load() + if t == nil { + continue + } + now := cs.clock() + for _, p := range *t { + if s := p.sock.Load(); s != nil && time.Duration(now-s.used.Load()) > idle { + u.closeConnSock(s, "idle", false) + } + } + } +} + +// connSockDead reports whether a connected socket send or receive error means the connected path to the peer is +// gone, so its traffic belongs back on the listener. A connected UDP socket reports a peer's ICMP unreachable as +// ECONNREFUSED on its next call, which the unconnected listener never sees. +func connSockDead(errno syscall.Errno) bool { + switch errno { + case unix.ECONNREFUSED, unix.EHOSTUNREACH, unix.EHOSTDOWN, unix.ENETUNREACH, unix.ENETDOWN, + unix.EADDRNOTAVAIL, unix.EPIPE, unix.ENOTCONN, unix.EDESTADDRREQ: + return true + } + return false +} + +// enobufsWait is how long a connected socket send sleeps when the interface queue is flow controlled. The kernel +// posts no wakeup when the advisory lifts, so this polls; 50us kept a gigabit link full without spinning a core in +// the 09-25 probes. +const enobufsWait = 50 * time.Microsecond + +// connSockWrite writes bufs on s, one datagram per write. A full socket buffer parks the caller in the netpoller +// until the socket is writable, and a flow-controlled interface makes it wait out ENOBUFS, so a busy link slows +// the tun reader instead of dropping. Once the connected socket has waited cfg.enobufsMax without a send going +// through, either one drops the datagram instead, and a dropped datagram counts as sent, as the listener's +// unreported interface drops do; it never falls back to the listener, which would reorder it. closed reports that +// the connected socket was closed before bufs[sent] went out; any other error stops the run and is returned with +// how many went out. +func (u *StdConn) connSockWrite(s *connSock, bufs [][]byte) (sent int, errno syscall.Errno, closed bool) { + cs := u.socks + defer func() { + if sent > 0 { + s.used.Store(cs.clock()) + } + }() + for sent < len(bufs) { + err := s.try(bufs[sent], false) + if err == syscall.EAGAIN || err == os.ErrDeadlineExceeded { + // os.ErrDeadlineExceeded here is another writer's parked deadline, passed but not yet cleared. + err = u.parkedWrite(s, bufs[sent]) + } + switch err { + case nil: + sent++ + if s.stallAt.Load() != 0 { + s.stallAt.Store(0) + s.dropping.Store(false) + } + continue + case errStallSpent: + u.dropStalled(s) + sent++ + continue + } + errno, ok := err.(syscall.Errno) + if !ok { + return sent, 0, true + } + if errno != syscall.ENOBUFS { + return sent, errno, false + } + // The wait holds no reference on the descriptor, so a close doesn't wait for it. + if s.stalledFor(cs.clock()) < cs.cfg.enobufsMax { + cs.sleep(enobufsWait) + } else { + u.dropStalled(s) + sent++ + } + } + return sent, 0, false +} + +// parkedWrite writes b after EAGAIN, parked in the netpoller for what is left of s's stall budget, and returns +// errStallSpent once none is. golang/go#73919: a darwin UDP write parked on EAGAIN can wait forever for a +// writability event that never comes, so the wait always has a deadline. +func (u *StdConn) parkedWrite(s *connSock, b []byte) error { + cs := u.socks + s.parkMu.Lock() + defer s.parkMu.Unlock() + left := cs.cfg.enobufsMax - s.stalledFor(cs.clock()) + if left <= 0 { + return errStallSpent + } + if err := s.conn.SetWriteDeadline(time.Now().Add(left)); err != nil { + return net.ErrClosed + } + err := s.try(b, true) + _ = s.conn.SetWriteDeadline(time.Time{}) + if err == os.ErrDeadlineExceeded { + return errStallSpent + } + return err +} + +// dropStalled logs the first datagram a stall drops. +func (u *StdConn) dropStalled(s *connSock) { + if !s.dropping.Swap(true) { + u.l.Debug("connected socket: sends stalled, dropping datagrams until one goes through", "udpAddr", s.dst) + } +} + +// stalledFor records an ENOBUFS or EAGAIN on s at now and returns how long its sends have stalled: since the first +// ENOBUFS or EAGAIN after a send last went through. Every writer measures from that one time, so concurrent writers +// spend the budget no faster than one, and a send going through ends the stall for every writer, including one +// asleep in it. +func (s *connSock) stalledFor(now int64) time.Duration { + if s.stallAt.CompareAndSwap(0, now) { + return 0 + } + at := s.stallAt.Load() + if at == 0 { + return 0 + } + return time.Duration(now - at) +} + +// readConnSock delivers a connected socket's datagrams through the listener's reader, serialized with ListenOut, +// until the connected socket is closed. Any receive error closes it: a connected socket reports its peer's +// unreachables here, and retrying an unexpected error could spin. +func (u *StdConn) readConnSock(s *connSock) { + rd := u.reader.Load() + buf := make([]byte, MTU) + for { + n, from, err := s.conn.ReadFromUDPAddrPort(buf) + if err != nil { + if errors.Is(err, net.ErrClosed) { + return + } + why := err.Error() + var errno syscall.Errno + if errors.As(err, &errno) && connSockDead(errno) { + why = errno.Error() + } else { + u.l.Warn("connected socket: unexpected receive error", "udpAddr", s.dst, "error", err) + } + u.closeConnSock(s, why, true) + return + } + s.used.Store(u.socks.clock()) + u.readMu.Lock() + rd.r(netip.AddrPortFrom(from.Addr().Unmap(), from.Port()), buf[:n:n]) + rd.flush() + u.readMu.Unlock() + } +} diff --git a/udp/connsock_darwin_test.go b/udp/connsock_darwin_test.go new file mode 100644 index 00000000..f568dd67 --- /dev/null +++ b/udp/connsock_darwin_test.go @@ -0,0 +1,1020 @@ +//go:build !e2e_testing + +package udp + +import ( + "context" + "fmt" + "log/slog" + "net" + "net/netip" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "golang.org/x/sys/unix" +) + +// newConnSockListener opens a listener with the connected socket knobs pinned to their defaults, so NEBULA_* +// variables in the environment can't change what a test sees, applies cfg, and starts ListenOut with r. +func newConnSockListener(t *testing.T, s Settings, h slog.Handler, r EncReader, cfg ...func(*connSockConfig)) *StdConn { + t.Helper() + for k, v := range map[string]string{ + "NEBULA_CONNSOCKS": "8", + "NEBULA_CONNSOCK_RUN": "64", + "NEBULA_CONNSOCK_WINDOW": "1s", + "NEBULA_CONNSOCK_IDLE": "30s", + "NEBULA_CONNSOCK_OPEN_EVERY": "1s", + } { + t.Setenv(k, v) + } + if h == nil { + h = slog.DiscardHandler + } + if r == nil { + r = func(netip.AddrPort, []byte) {} + } + c, err := NewListener(slog.New(h), s) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + u := c.(*StdConn) + for _, f := range cfg { + f(&u.socks.cfg) + } + go func() { _ = u.ListenOut(r, func() {}) }() + require.Eventually(t, func() bool { return u.reader.Load() != nil }, 5*time.Second, time.Millisecond) + return u +} + +func listen(s string) Settings { return Settings{Listen: netip.MustParseAddrPort(s)} } + +// run builds n datagrams to dst. +func run(dst netip.AddrPort, n int, tag string) ([][]byte, []netip.AddrPort) { + bufs := make([][]byte, n) + addrs := make([]netip.AddrPort, n) + for i := range bufs { + bufs[i] = fmt.Appendf(nil, "%s-%d", tag, i) + addrs[i] = dst + } + return bufs, addrs +} + +func newPeer(t *testing.T, network, addr string) (*net.UDPConn, netip.AddrPort) { + t.Helper() + pc, err := net.ListenPacket(network, addr) + require.NoError(t, err) + t.Cleanup(func() { _ = pc.Close() }) + uc := pc.(*net.UDPConn) + return uc, uc.LocalAddr().(*net.UDPAddr).AddrPort() +} + +// newEstablishedPeer opens a peer as newPeer does and tells u it is an established tunnel's remote. +func newEstablishedPeer(t *testing.T, u *StdConn, network, addr string) (*net.UDPConn, netip.AddrPort) { + t.Helper() + c, dst := newPeer(t, network, addr) + u.SetEstablishedPeer(dst, true) + return c, dst +} + +// readFrom reads n datagrams at peer and returns the source of each. +func readFrom(t *testing.T, peer *net.UDPConn, n int) []netip.AddrPort { + t.Helper() + buf := make([]byte, MTU) + var from []netip.AddrPort + for range n { + require.NoError(t, peer.SetReadDeadline(time.Now().Add(5*time.Second))) + _, f, err := peer.ReadFromUDPAddrPort(buf) + require.NoError(t, err) + from = append(from, netip.AddrPortFrom(f.Addr().Unmap(), f.Port())) + } + return from +} + +func reusePort(t *testing.T, c syscall.Conn) int { + t.Helper() + rc, err := c.SyscallConn() + require.NoError(t, err) + var v int + var serr error + require.NoError(t, rc.Control(func(fd uintptr) { + v, serr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT) + })) + require.NoError(t, serr) + return v +} + +// rogueBinds tries each of binds, "network address plain|reuseport", on port and returns the ones that bound. +func rogueBinds(t *testing.T, port uint16, binds []string) []string { + t.Helper() + var bound []string + for _, b := range binds { + f := strings.Fields(b) + network, addr, reuse := f[0], f[1], f[2] + lc := net.ListenConfig{Control: connSockControl(reuse == "reuseport", 0)} + pc, err := lc.ListenPacket(context.Background(), network, net.JoinHostPort(addr, fmt.Sprint(port))) + if err == nil { + _ = pc.Close() + bound = append(bound, b) + } + } + return bound +} + +// The listener never sets SO_REUSEPORT, and a connected socket binds with it only to a specific address, so every +// bind a lone listener refuses is refused after NewListener returns and while connected sockets are open. Same-uid +// SO_REUSEPORT binds to a specific address are left out: xnu allows them beside a lone wildcard listener too, +// which is how a connected socket binds, and refuses them to any other uid. +func TestConnSocksNoSharedBind(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + la, err := u.LocalAddr() + require.NoError(t, err) + binds := []string{ + "udp4 0.0.0.0 plain", "udp4 0.0.0.0 reuseport", "udp4 127.0.0.1 plain", + "udp6 :: plain", "udp6 :: reuseport", "udp6 ::1 plain", + } + assert.Zero(t, reusePort(t, u.UDPConn), "listener") + assert.Empty(t, rogueBinds(t, la.Port(), binds), "bound beside the listener") + + peer4, dst4 := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + peer6, dst6 := newEstablishedPeer(t, u, "udp6", "[::1]:0") + for _, p := range []struct { + c *net.UDPConn + dst netip.AddrPort + }{{peer4, dst4}, {peer6, dst6}} { + bufs, addrs := run(p.dst, u.socks.cfg.run, "p") + _, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + readFrom(t, p.c, len(bufs)) + s := u.connSockFor(p.dst) + require.NotNil(t, s, "no connected socket to %v", p.dst) + assert.NotZero(t, reusePort(t, s.conn), "connected socket to %v", p.dst) + } + assert.Empty(t, rogueBinds(t, la.Port(), binds), "bound beside the connected sockets") +} + +// A connected socket binds the source address the listener's route to its peer picks, for either family on a +// dual-stack listener, so the peer sees one address and port whichever socket sent. +func TestConnSocksSourceMatchesListener(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + la, err := u.LocalAddr() + require.NoError(t, err) + for _, p := range []struct{ network, addr string }{{"udp4", "127.0.0.1:0"}, {"udp6", "[::1]:0"}} { + peer, dst := newEstablishedPeer(t, u, p.network, p.addr) + require.NoError(t, u.WriteTo([]byte("probe"), dst)) + require.Nil(t, u.connSockFor(dst)) + viaListener := readFrom(t, peer, 1)[0] + assert.Equal(t, la.Port(), viaListener.Port()) + + bufs, addrs := run(dst, u.socks.cfg.run, p.network) + _, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + s := u.connSockFor(dst) + require.NotNil(t, s, "no connected socket to %v", dst) + for _, from := range readFrom(t, peer, len(bufs)) { + assert.Equal(t, viaListener, from, "source seen by %v", dst) + } + local := s.conn.LocalAddr().(*net.UDPAddr).AddrPort() + assert.Equal(t, viaListener, netip.AddrPortFrom(local.Addr().Unmap(), local.Port())) + } +} + +// A run long enough opens a connected socket per peer, its replies reach the listener's reader, and a peer that +// goes away sends its traffic back through the listener with its connected socket closed, while the next peer gets +// a connected socket of its own. +func TestConnSocks(t *testing.T) { + var mu sync.Mutex + got := map[netip.AddrPort]int{} + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, func(addr netip.AddrPort, _ []byte) { + mu.Lock() + got[addr]++ + mu.Unlock() + }) + la, err := u.LocalAddr() + require.NoError(t, err) + to := netip.AddrPortFrom(netip.MustParseAddr("127.0.0.1"), la.Port()) + + var peers [3]*net.UDPConn + var dsts [3]netip.AddrPort + for i := range peers { + peers[i], dsts[i] = newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + } + for i, dst := range dsts[:2] { + bufs, addrs := run(dst, u.socks.cfg.run, fmt.Sprint("p", i)) + n, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + assert.Equal(t, len(bufs), n) + require.NotNil(t, u.connSockFor(dst), "no connected socket to peer %d", i) + for _, from := range readFrom(t, peers[i], len(bufs)) { + assert.Equal(t, to, from) + } + for range 3 { + _, err := peers[i].WriteToUDPAddrPort([]byte("reply"), to) + require.NoError(t, err) + } + } + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return got[dsts[0]] == 3 && got[dsts[1]] == 3 + }, 5*time.Second, time.Millisecond) + + // The peer's ICMP port unreachable surfaces on the connected socket's next call. The deadline is generous + // because xnu rate-limits and delivers the unreachable on its own schedule. + gone := dsts[1] + goneSock := u.connSockFor(gone) + require.NoError(t, peers[1].Close()) + bufs, addrs := run(gone, u.socks.cfg.run, "gone") + require.Eventually(t, func() bool { + _, _ = u.WriteBatch(bufs, addrs) + return u.connSockFor(gone) == nil + }, 10*time.Second, 10*time.Millisecond) + assert.NotNil(t, u.connSockFor(dsts[0])) + u.socks.closers.Wait() + _, err = goneSock.conn.Write([]byte("x")) + assert.ErrorIs(t, err, net.ErrClosed, "closed connected socket left open") + + bufs, addrs = run(dsts[2], u.socks.cfg.run, "p2") + _, err = u.WriteBatch(bufs, addrs) + require.NoError(t, err) + require.NotNil(t, u.connSockFor(dsts[2])) + for _, from := range readFrom(t, peers[2], len(bufs)) { + assert.Equal(t, to, from) + } + + require.NoError(t, u.Rebind()) + for _, dst := range dsts { + assert.Nil(t, u.connSockFor(dst), "connected socket left after Rebind") + } +} + +// Listen routines on one port share one set of connected sockets: the cap and the open bucket are theirs together, +// a peer's connected socket carries every routine's sends to it, and Rebind on the first routine, which is all +// Control.RebindUDPServer calls, clears every routine's listener and closes the connected sockets. +func TestConnSocksSharedAcrossRoutines(t *testing.T) { + s := listen("0.0.0.0:0") + s.Multi = true + u0 := newConnSockListener(t, s, nil, nil, func(c *connSockConfig) { c.max = 1 }) + la, err := u0.LocalAddr() + require.NoError(t, err) + s.Listen = netip.AddrPortFrom(s.Listen.Addr(), la.Port()) + u1 := newConnSockListener(t, s, nil, nil) + require.Same(t, u0.socks, u1.socks) + lo, err := net.InterfaceByName("lo0") + require.NoError(t, err) + + peerA, a := newPeer(t, "udp4", "127.0.0.1:0") + peerB, b := newPeer(t, "udp4", "127.0.0.1:0") + for _, dst := range []netip.AddrPort{a, b} { + // Nebula tells every writer. + u0.SetEstablishedPeer(dst, true) + u1.SetEstablishedPeer(dst, true) + } + + bufs, addrs := run(a, u1.socks.cfg.run, "a") + _, err = u1.WriteBatch(bufs, addrs) + require.NoError(t, err) + readFrom(t, peerA, len(bufs)) + s1 := u1.connSockFor(a) + require.NotNil(t, s1) + var sends atomic.Int64 + s1.sys = func(fd int, b []byte) error { + sends.Add(1) + return writeFd(fd, b) + } + require.NoError(t, u0.WriteTo([]byte("via u0"), a)) + readFrom(t, peerA, 1) + assert.EqualValues(t, 1, sends.Load(), "routine 0 didn't send on routine 1's connected socket") + + bufs, addrs = run(b, 2*u0.socks.cfg.run, "b") + _, err = u0.WriteBatch(bufs, addrs) + require.NoError(t, err) + readFrom(t, peerB, len(bufs)) + assert.Nil(t, u0.connSockFor(b), "a second connected socket past a cap of 1") + + level, opt := unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF + if u0.isV4 { + level, opt = unix.IPPROTO_IP, unix.IP_BOUND_IF + } + boundIf := func(u *StdConn) int { + v, err := unix.GetsockoptInt(int(u.sysFd), level, opt) + require.NoError(t, err) + return v + } + for _, u := range []*StdConn{u0, u1} { + require.NoError(t, unix.SetsockoptInt(int(u.sysFd), level, opt, lo.Index)) + require.Equal(t, lo.Index, boundIf(u)) + } + require.NoError(t, u0.Rebind()) + assert.Zero(t, boundIf(u0)) + assert.Zero(t, boundIf(u1), "Rebind left routine 1 scoped") + assert.Nil(t, u1.connSockFor(a), "Rebind left routine 1's connected socket open") + u0.socks.closers.Wait() + _, err = s1.conn.Write([]byte("x")) + assert.ErrorIs(t, err, net.ErrClosed) +} + +// A connected socket on a specific listen.host would need SO_REUSEPORT on the listener, so connected sockets stay +// off there unless listen routines already set it. +func TestConnSocksSpecificHost(t *testing.T) { + for _, multi := range []bool{false, true} { + t.Run(fmt.Sprint("multi=", multi), func(t *testing.T) { + s := listen("127.0.0.1:0") + s.Multi = multi + u := newConnSockListener(t, s, nil, nil) + la, err := u.LocalAddr() + require.NoError(t, err) + if !multi { + assert.Zero(t, reusePort(t, u.UDPConn), "listener") + } + peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + bufs, addrs := run(dst, u.socks.cfg.run, "p") + _, err = u.WriteBatch(bufs, addrs) + require.NoError(t, err) + for _, from := range readFrom(t, peer, len(bufs)) { + assert.Equal(t, la, from) + } + if multi { + assert.NotNil(t, u.connSockFor(dst)) + } else { + assert.Nil(t, u.connSockFor(dst)) + } + }) + } +} + +// A v4 listener can't send to a v6 destination; however often it's asked, it says so and opens no connected +// socket. +func TestConnSocksV6DestinationOnV4Listener(t *testing.T) { + s := listen("127.0.0.1:0") + s.Multi = true + u := newConnSockListener(t, s, nil, nil) + require.True(t, u.isV4) + dst := netip.MustParseAddrPort("[::1]:9") + u.SetEstablishedPeer(dst, true) + bufs, _ := run(dst, 2*u.socks.cfg.run, "v6") + for _, b := range bufs { + assert.ErrorIs(t, u.WriteTo(b, dst), ErrInvalidIPv6RemoteForSocket) + } + assert.Nil(t, u.connSockFor(dst)) +} + +// darwin's tun hands over one packet per read, so a busy peer's traffic reaches the socket as single WriteTo +// calls. cfg.run of them inside one window earn a connected socket; the same count spread over more than a window +// doesn't. +func TestConnSocksOpenFromWriteTo(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + var now atomic.Int64 + now.Store(int64(time.Hour)) + u.socks.clock = now.Load + _, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + pkt := []byte("pkt") + + for range u.socks.cfg.run - 1 { + require.NoError(t, u.WriteTo(pkt, dst)) + } + now.Add(int64(u.socks.cfg.window) + 1) + require.NoError(t, u.WriteTo(pkt, dst)) + assert.Nil(t, u.connSockFor(dst), "a connected socket for sends spread over two windows") + + for range u.socks.cfg.run - 2 { + require.NoError(t, u.WriteTo(pkt, dst)) + } + assert.Nil(t, u.connSockFor(dst)) + require.NoError(t, u.WriteTo(pkt, dst)) + assert.NotNil(t, u.connSockFor(dst), "no connected socket after %d sends in one window", u.socks.cfg.run) +} + +// Only the remote of an established tunnel gets a connected socket, however busy another destination is, and one +// that stops being established loses its connected socket and has to be busy again to get a new one. +func TestConnSocksOnlyForEstablishedPeers(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + peer, dst := newPeer(t, "udp4", "127.0.0.1:0") + run1 := u.socks.cfg.run + send := func(n int) { + t.Helper() + bufs, addrs := run(dst, n, "p") + _, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + readFrom(t, peer, n) + } + + send(2 * run1) + assert.Nil(t, u.connSockFor(dst), "a connected socket for a busy destination with no tunnel") + + u.SetEstablishedPeer(dst, true) + send(run1) + s := u.connSockFor(dst) + require.NotNil(t, s, "no connected socket for a busy established peer") + + u.SetEstablishedPeer(dst, false) + assert.Nil(t, u.connSockFor(dst)) + u.socks.closers.Wait() + _, err := s.conn.Write([]byte("x")) + assert.ErrorIs(t, err, net.ErrClosed, "connected socket left open after its tunnel went away") + send(2 * run1) + assert.Nil(t, u.connSockFor(dst), "a connected socket after the tunnel went away") + + u.SetEstablishedPeer(dst, true) + send(run1 - 1) + assert.Nil(t, u.connSockFor(dst), "a re-established peer kept its old count") + send(1) + assert.NotNil(t, u.connSockFor(dst)) +} + +// A tunnel going away while its busy remote is opening a connected socket never leaves one open, whether or not +// the remote is established again before the open finishes. The dial hook holds the open between reserving its +// destination and dialing until the tunnel has changed, so every round races. +func TestConnSocksEstablishedRace(t *testing.T) { + h := &countingHandler{} + u := newConnSockListener(t, listen("0.0.0.0:0"), h, nil, func(c *connSockConfig) { + c.run = 1 + c.openEvery = time.Nanosecond + }) + cs := u.socks + peer, dst := newPeer(t, "udp4", "127.0.0.1:0") + for i, reestablish := range []bool{false, true} { + u.SetEstablishedPeer(dst, true) + dialing, release := make(chan struct{}), make(chan struct{}) + cs.dialHook = func(netip.AddrPort) { + close(dialing) + <-release + } + done := make(chan error) + go func() { done <- u.WriteTo([]byte("p"), dst) }() + <-dialing + u.SetEstablishedPeer(dst, false) + if reestablish { + u.SetEstablishedPeer(dst, true) + } + close(release) + require.NoError(t, <-done) + cs.dialHook = nil + readFrom(t, peer, 1) + + assert.Nil(t, u.connSockFor(dst), "reestablish=%v", reestablish) + assert.EqualValues(t, i+1, h.discarded.Load(), "reestablish=%v", reestablish) + cs.mu.Lock() + assert.Empty(t, cs.pending) + cs.mu.Unlock() + assert.Zero(t, cs.reserved.Load()) + u.SetEstablishedPeer(dst, false) + } + assert.Zero(t, h.opened.Load()) +} + +// Neither an open in flight nor a close holds anything SetEstablishedPeer waits on, since Nebula calls it under +// the hostmap's write lock; and a peer whose connected socket is still closing can't open another until the close +// completes. The hooks hold the open in its dials and the close before its Close. +func TestConnSocksSetEstablishedPeerMakesNoSyscalls(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil, func(c *connSockConfig) { c.openEvery = time.Nanosecond }) + cs := u.socks + peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + _, other := newPeer(t, "udp4", "127.0.0.1:0") + + dialing, releaseDial := make(chan struct{}), make(chan struct{}) + cs.dialHook = func(netip.AddrPort) { + close(dialing) + <-releaseDial + } + opened := make(chan *connSock) + go func() { + bufs, addrs := run(dst, cs.cfg.run, "p") + _, _ = u.WriteBatch(bufs, addrs) + opened <- u.connSockFor(dst) + }() + <-dialing + u.SetEstablishedPeer(other, true) + u.SetEstablishedPeer(other, false) + close(releaseDial) + s := <-opened + require.NotNil(t, s) + readFrom(t, peer, cs.cfg.run) + cs.dialHook = nil + + closing, releaseClose := make(chan struct{}), make(chan struct{}) + cs.closeHook = func(netip.AddrPort) { + close(closing) + <-releaseClose + } + u.SetEstablishedPeer(dst, false) + <-closing + assert.Nil(t, u.connSockFor(dst)) + u.SetEstablishedPeer(dst, true) + bufs, addrs := run(dst, 2*cs.cfg.run, "q") + _, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + readFrom(t, peer, len(bufs)) + assert.Nil(t, u.connSockFor(dst), "a connected socket opened while the last one to its peer was closing") + + close(releaseClose) + cs.closers.Wait() + cs.closeHook = nil + _, err = s.conn.Write([]byte("x")) + assert.ErrorIs(t, err, net.ErrClosed) + bufs, addrs = run(dst, cs.cfg.run, "r") + _, err = u.WriteBatch(bufs, addrs) + require.NoError(t, err) + readFrom(t, peer, len(bufs)) + assert.NotNil(t, u.connSockFor(dst), "no connected socket once the last one closed") +} + +// A listener whose interface scope can't be read gives a busy peer no connected socket, rather than one without +// the scope; the peer stays on the listener for the backoff. A pipe stands in for the listener's descriptor, so +// getsockopt fails with ENOTSOCK. +func TestConnSocksBoundIfFailsClosed(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + _, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + var pipe [2]int + require.NoError(t, unix.Pipe(pipe[:])) + t.Cleanup(func() { + _ = unix.Close(pipe[0]) + _ = unix.Close(pipe[1]) + }) + listener := u.sysFd + u.sysFd = uintptr(pipe[0]) + s := u.maybeOpenConnSock(u.socks.peer(dst), dst, u.socks.cfg.run) + u.sysFd = listener + assert.Nil(t, s) + assert.Nil(t, u.connSockFor(dst)) + assert.Greater(t, u.socks.peer(dst).retryAt, u.socks.clock(), "no backoff after a failed open") +} + +// readTags reads n datagrams at peer and returns their payloads. +func readTags(t *testing.T, peer *net.UDPConn, n int) []string { + t.Helper() + buf := make([]byte, MTU) + var tags []string + for range n { + require.NoError(t, peer.SetReadDeadline(time.Now().Add(5*time.Second))) + m, _, err := peer.ReadFromUDPAddrPort(buf) + require.NoError(t, err) + tags = append(tags, string(buf[:m])) + } + return tags +} + +// One batch can hold runs to a peer with a connected socket, to an established peer without one, and to a +// destination that is no tunnel's remote, interleaved; each run goes out its own way, in order, and all count. +func TestConnSocksWriteBatchMixedDestinations(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + s, peerA := openConnSock(t, u) + peerB, b := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + peerC, c := newPeer(t, "udp4", "127.0.0.1:0") + var onSock atomic.Int64 + s.sys = func(fd int, b []byte) error { + onSock.Add(1) + return writeFd(fd, b) + } + + var bufs [][]byte + var addrs []netip.AddrPort + for _, r := range []struct { + dst netip.AddrPort + tag string + n int + }{{s.dst, "a", 3}, {b, "b", 2}, {c, "c", 2}, {s.dst, "A", 2}, {c, "C", 1}} { + for i := range r.n { + bufs = append(bufs, fmt.Appendf(nil, "%s%d", r.tag, i)) + addrs = append(addrs, r.dst) + } + } + n, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + assert.Equal(t, len(bufs), n) + assert.EqualValues(t, 5, onSock.Load(), "datagrams on a's connected socket") + assert.Equal(t, []string{"a0", "a1", "a2", "A0", "A1"}, readTags(t, peerA, 5)) + assert.Equal(t, []string{"b0", "b1"}, readTags(t, peerB, 2)) + assert.Equal(t, []string{"c0", "c1", "C0"}, readTags(t, peerC, 3)) + assert.Nil(t, u.connSockFor(b)) +} + +// A connected socket error that doesn't mean the path is gone costs only its datagram: WriteBatch skips it and +// sends the rest of the run on the connected socket, WriteTo returns it, and the connected socket stays open. +func TestConnSocksWriteError(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + s, peer := openConnSock(t, u) + s.sys = func(fd int, b []byte) error { + if string(b) == "bad" { + return syscall.EMSGSIZE + } + return writeFd(fd, b) + } + bufs := [][]byte{[]byte("ok0"), []byte("bad"), []byte("ok2")} + n, err := u.WriteBatch(bufs, []netip.AddrPort{s.dst, s.dst, s.dst}) + require.NoError(t, err) + assert.Equal(t, 2, n) + assert.Equal(t, []string{"ok0", "ok2"}, readTags(t, peer, 2)) + assert.ErrorIs(t, u.WriteTo([]byte("bad"), s.dst), syscall.EMSGSIZE) + assert.Same(t, s, u.connSockFor(s.dst)) +} + +// A connected socket that turns out closed or dead partway through a run leaves the rest of the run to the +// listener, in order and all counted; a dead one is closed and its peer backs off to the listener. +func TestConnSocksWriteBatchFallsBackMidRun(t *testing.T) { + for _, fail := range []error{net.ErrClosed, syscall.ECONNREFUSED} { + t.Run(fail.Error(), func(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + s, peer := openConnSock(t, u) + var calls atomic.Int64 + s.sys = func(fd int, b []byte) error { + if calls.Add(1) > 2 { + return fail + } + return writeFd(fd, b) + } + bufs, addrs := run(s.dst, 5, "r") + n, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + assert.Equal(t, 5, n) + assert.Equal(t, []string{"r-0", "r-1", "r-2", "r-3", "r-4"}, readTags(t, peer, 5)) + assert.EqualValues(t, 3, calls.Load(), "the connected socket was tried again after it failed") + if fail == syscall.ECONNREFUSED { + assert.Nil(t, u.connSockFor(s.dst)) + u.socks.mu.Lock() + assert.Greater(t, u.socks.peer(s.dst).retryAt, u.socks.clock(), "no backoff after a dead connected socket") + u.socks.mu.Unlock() + } + }) + } +} + +// openConnSock earns a connected socket to a fresh loopback peer and returns it and the peer. +func openConnSock(t *testing.T, u *StdConn) (*connSock, *net.UDPConn) { + t.Helper() + peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + bufs, addrs := run(dst, u.socks.cfg.run, "p") + _, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + readFrom(t, peer, len(bufs)) + s := u.connSockFor(dst) + require.NotNil(t, s) + return s, peer +} + +// A flow-controlled interface makes a connected socket's sends wait, up to cfg.enobufsMax without one going +// through; after that each datagram is dropped without waiting until a send succeeds again, and each stall logs +// its drops once. s.sys stands in for an interface that answers every write with ENOBUFS until enobufs is +// cleared, and each sleep advances the clock. +func TestConnSockWriteENOBUFS(t *testing.T) { + h := &countingHandler{} + u := newConnSockListener(t, listen("0.0.0.0:0"), h, nil) + s, _ := openConnSock(t, u) + var now atomic.Int64 + now.Store(int64(time.Hour)) + sleeps := 0 + u.socks.clock = now.Load + u.socks.sleep = func(d time.Duration) { + sleeps++ + now.Add(int64(d)) + } + var enobufs atomic.Bool + s.sys = func(fd int, b []byte) error { + if enobufs.Load() { + return syscall.ENOBUFS + } + return writeFd(fd, b) + } + perStall := int(u.socks.cfg.enobufsMax / enobufsWait) + bufs, _ := run(s.dst, 3, "q") + + for stall := 1; stall <= 2; stall++ { + enobufs.Store(true) + sent, errno, closed := u.connSockWrite(s, bufs) + assert.Equal(t, 3, sent) + assert.Zero(t, errno) + assert.False(t, closed) + assert.Equal(t, stall*perStall, sleeps) + + sent, _, _ = u.connSockWrite(s, bufs) + assert.Equal(t, 3, sent) + assert.Equal(t, stall*perStall, sleeps, "a stalled connected socket waited again") + assert.EqualValues(t, stall, h.dropping.Load(), "drop log lines after stall %d", stall) + + enobufs.Store(false) + sent, _, _ = u.connSockWrite(s, bufs[:1]) + assert.Equal(t, 1, sent) + assert.Zero(t, s.stallAt.Load()) + } +} + +// Concurrent writers share one ENOBUFS budget measured from the first ENOBUFS since a send went through: two +// writers stalled together wait as long as one would, and a send going through while a writer sleeps gives that +// writer a fresh budget. The sleep hook parks each writer until the test has moved the clock. +func TestConnSockWriteENOBUFSConcurrentWriters(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + s, _ := openConnSock(t, u) + var now atomic.Int64 + u.socks.clock = now.Load + asleep, wake := make(chan struct{}), make(chan struct{}) + u.socks.sleep = func(time.Duration) { + asleep <- struct{}{} + <-wake + } + // Datagrams starting with 'e' get ENOBUFS while enobufs is set; the rest go out. + var enobufs atomic.Bool + var wrote atomic.Int64 + s.sys = func(fd int, b []byte) error { + if b[0] == 'e' && enobufs.Load() { + return syscall.ENOBUFS + } + wrote.Add(1) + return writeFd(fd, b) + } + write := func(tag string, done chan<- int) { + sent, _, _ := u.connSockWrite(s, [][]byte{[]byte(tag)}) + done <- sent + } + budget := int64(u.socks.cfg.enobufsMax) + + t.Run("a send going through ends the stall", func(t *testing.T) { + enobufs.Store(true) + now.Store(int64(time.Hour)) + done := make(chan int) + go write("e1", done) + <-asleep + now.Add(budget / 2) + sent, _, _ := u.connSockWrite(s, [][]byte{[]byte("ok")}) + require.Equal(t, 1, sent) + now.Add(budget) + wake <- struct{}{} + <-asleep // e1 hit ENOBUFS again 1.5x the budget after its first, and waited instead of dropping. + enobufs.Store(false) + before := wrote.Load() + wake <- struct{}{} + assert.Equal(t, 1, <-done) + assert.Equal(t, before+1, wrote.Load(), "e1 was dropped") + }) + + t.Run("two stalled writers spend one budget", func(t *testing.T) { + enobufs.Store(true) + now.Store(int64(2 * time.Hour)) + done := make(chan int) + go write("e2", done) + go write("e3", done) + <-asleep + <-asleep + now.Add(budget * 6 / 10) + wake <- struct{}{} + wake <- struct{}{} + <-asleep // Both are 0.6x the budget into one stall, not 1.2x, so both wait again. + <-asleep + now.Add(budget / 2) + before := wrote.Load() + wake <- struct{}{} + wake <- struct{}{} + assert.Equal(t, 1, <-done) + assert.Equal(t, 1, <-done) + assert.Equal(t, before, wrote.Load(), "a datagram went out while ENOBUFS lasted") + }) +} + +// A full send buffer that never drains parks a connected socket send only for cfg.enobufsMax, the stall budget +// ENOBUFS gets: golang/go#73919 has darwin UDP writes parked on EAGAIN waiting forever. Past it the datagrams are +// dropped and counted as sent, concurrent writers spend that one budget, and once a send goes through the +// connected socket works as before. s.sys stands in for a send buffer that stays full while eagain is set. +func TestConnSockWriteEAGAINBounded(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + s, peer := openConnSock(t, u) + var eagain atomic.Bool + eagain.Store(true) + s.sys = func(fd int, b []byte) error { + if eagain.Load() { + return syscall.EAGAIN + } + return writeFd(fd, b) + } + budget := u.socks.cfg.enobufsMax + + start := time.Now() + sent, errno, closed := u.connSockWrite(s, [][]byte{[]byte("x")}) + elapsed := time.Since(start) + assert.Equal(t, 1, sent) + assert.Zero(t, errno) + assert.False(t, closed) + assert.GreaterOrEqual(t, elapsed, budget, "the send didn't wait for the buffer to drain") + assert.Less(t, elapsed, time.Second, "the send waited past its deadline") + + // The budget is spent, so these drop without waiting, however many writers there are. + var wg sync.WaitGroup + start = time.Now() + for range 4 { + wg.Go(func() { + bufs, _ := run(s.dst, 8, "y") + sent, _, _ := u.connSockWrite(s, bufs) + assert.Equal(t, 8, sent) + }) + } + wg.Wait() + assert.Less(t, time.Since(start), time.Second) + + // A send going through starts a fresh budget, which two writers stalled together spend once; the upper bound + // is far looser than that, so a descheduled runner doesn't fail it. + eagain.Store(false) + sent, _, _ = u.connSockWrite(s, [][]byte{[]byte("ok")}) + require.Equal(t, 1, sent) + readTags(t, peer, 1) + eagain.Store(true) + start = time.Now() + for range 2 { + wg.Go(func() { + sent, _, _ := u.connSockWrite(s, [][]byte{[]byte("w")}) + assert.Equal(t, 1, sent) + }) + } + wg.Wait() + elapsed = time.Since(start) + assert.GreaterOrEqual(t, elapsed, budget) + assert.Less(t, elapsed, time.Second) + + eagain.Store(false) + bufs, addrs := run(s.dst, 3, "z") + n, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + assert.Equal(t, 3, n) + assert.Equal(t, []string{"z-0", "z-1", "z-2"}, readTags(t, peer, 3)) + assert.Zero(t, s.stallAt.Load()) + assert.Same(t, s, u.connSockFor(s.dst)) +} + +// A send waiting out ENOBUFS holds nothing a close or Close needs, and returns closed once it wakes to a closed +// connected socket. The sleep hook parks the writer until both closes have finished; after the close, s.sys +// writes for real, so the closed descriptor answers. +func TestConnSockWriteDoesNotBlockClose(t *testing.T) { + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil) + s, _ := openConnSock(t, u) + asleep, wake := make(chan struct{}), make(chan struct{}) + u.socks.sleep = func(time.Duration) { + asleep <- struct{}{} + <-wake + } + var gone atomic.Bool + s.sys = func(fd int, b []byte) error { + if gone.Load() { + return writeFd(fd, b) + } + return syscall.ENOBUFS + } + done := make(chan bool) + go func() { + bufs, _ := run(s.dst, 1, "q") + _, _, closed := u.connSockWrite(s, bufs) + done <- closed + }() + <-asleep + + u.closeConnSock(s, "test", false) + u.socks.closers.Wait() + require.NoError(t, u.Close()) + gone.Store(true) + close(wake) + assert.True(t, <-done, "the writer didn't see its connected socket close") +} + +// countingHandler counts connected socket open, close, discard and ENOBUFS drop log lines. +type countingHandler struct{ opened, closed, discarded, dropping atomic.Int64 } + +func (h *countingHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *countingHandler) Handle(_ context.Context, r slog.Record) error { + switch r.Message { + case "connected socket: opened": + h.opened.Add(1) + case "connected socket: closed": + h.closed.Add(1) + case "connected socket: discarded, its tunnel or route changed while it opened": + h.discarded.Add(1) + case "connected socket: sends stalled, dropping datagrams until one goes through": + h.dropping.Add(1) + } + return nil +} +func (h *countingHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *countingHandler) WithGroup(string) slog.Handler { return h } + +// Anyone can make Nebula reply to them, as recv_error does to every datagram with an unknown index; the reader +// below stands in for that reply. However many replies a sender draws, it never earns a connected socket without +// an established tunnel. +func TestConnSocksNoneForUnestablishedSenders(t *testing.T) { + h := &countingHandler{} + reply := []byte("reply") + var c *StdConn + var ready atomic.Bool + c = newConnSockListener(t, listen("0.0.0.0:0"), h, func(from netip.AddrPort, _ []byte) { + if ready.Load() { + _ = c.WriteTo(reply, from) + } + }) + ready.Store(true) + la, err := c.LocalAddr() + require.NoError(t, err) + to := netip.AddrPortFrom(netip.MustParseAddr("127.0.0.1"), la.Port()) + pkt := []byte("junk") + buf := make([]byte, 64) + for range 10 { + pc, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + require.NoError(t, err) + for range 2 * c.socks.cfg.run { + _, err := pc.WriteToUDPAddrPort(pkt, to) + require.NoError(t, err) + require.NoError(t, pc.SetReadDeadline(time.Now().Add(5*time.Second))) + _, _, err = pc.ReadFromUDPAddrPort(buf) + require.NoError(t, err) + } + _ = pc.Close() + } + assert.Zero(t, h.opened.Load()) + assert.Nil(t, c.socks.table.Load()) +} + +// Established peers that keep going away churn connected sockets at most as fast as the open bucket allows, and +// what connSocks keeps about each goes when its tunnel does. Each round a fresh peer is established, gets busy and +// goes away, and the clock moves 10ms. +func TestConnSocksChurnBounded(t *testing.T) { + h := &countingHandler{} + u := newConnSockListener(t, listen("0.0.0.0:0"), h, nil) + var now atomic.Int64 + now.Store(int64(time.Hour)) + u.socks.clock = now.Load + const rounds, step = 300, 10 * time.Millisecond + for range rounds { + peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0") + bufs, addrs := run(dst, u.socks.cfg.run, "p") + n, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + require.Equal(t, len(bufs), n) + readFrom(t, peer, len(bufs)) + u.SetEstablishedPeer(dst, false) + u.socks.closers.Wait() + require.NoError(t, peer.Close()) + now.Add(int64(step)) + } + elapsed := time.Duration(rounds) * step + budget := int64(u.socks.cfg.max) + int64(elapsed/u.socks.cfg.openEvery) + 1 + t.Logf("%d rounds over %v: %d opens, %d closes", rounds, elapsed, h.opened.Load(), h.closed.Load()) + require.GreaterOrEqual(t, h.opened.Load(), int64(u.socks.cfg.max), "the rounds never churned") + require.LessOrEqual(t, h.opened.Load(), budget) + assert.Equal(t, h.opened.Load(), h.closed.Load()) + assert.Nil(t, u.socks.table.Load(), "peers left after their tunnels went away") + assert.Zero(t, u.socks.reserved.Load()) +} + +// No send allocates, down any path: the listener with connected sockets off, to a destination that is no +// tunnel's remote, to an established peer that isn't busy, or on a connected socket, one datagram or a run. +func TestConnSockSendsDontAllocate(t *testing.T) { + if raceEnabled { + t.Skip("-race makes sync.Pool drop writers at random, so connected sends allocate") + } + pkt := make([]byte, 1400) + for _, c := range []struct { + name string + cfg func(*connSockConfig) + setup func(u *StdConn, dst netip.AddrPort) + }{ + {name: "off", cfg: func(c *connSockConfig) { c.max = 0 }}, + {name: "ineligible", setup: func(u *StdConn, dst netip.AddrPort) { + u.SetEstablishedPeer(netip.AddrPortFrom(dst.Addr(), dst.Port()+1), true) + }}, + {name: "eligible", cfg: func(c *connSockConfig) { c.run = 1 << 30 }, + setup: func(u *StdConn, dst netip.AddrPort) { u.SetEstablishedPeer(dst, true) }}, + {name: "connected", setup: func(u *StdConn, dst netip.AddrPort) { + u.SetEstablishedPeer(dst, true) + bufs, addrs := run(dst, u.socks.cfg.run, "p") + _, err := u.WriteBatch(bufs, addrs) + require.NoError(t, err) + require.NotNil(t, u.connSockFor(dst)) + }}, + } { + t.Run(c.name, func(t *testing.T) { + var cfg []func(*connSockConfig) + if c.cfg != nil { + cfg = append(cfg, c.cfg) + } + u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil, cfg...) + _, dst := newPeer(t, "udp4", "127.0.0.1:0") + if c.setup != nil { + c.setup(u, dst) + } + bufs, addrs := run(dst, 8, "b") + assert.Zero(t, testing.AllocsPerRun(100, func() { _ = u.WriteTo(pkt, dst) }), "WriteTo") + assert.Zero(t, testing.AllocsPerRun(100, func() { _, _ = u.WriteBatch(bufs, addrs) }), "WriteBatch") + }) + } +} + +// Once burst tokens are spent the bucket earns one back per interval. +func TestGCRA(t *testing.T) { + var g gcra + now := int64(time.Hour) + for i := range 3 { + require.True(t, g.take(now, 3, time.Second), "token %d of the burst", i) + } + require.False(t, g.take(now, 3, time.Second)) + require.False(t, g.take(now+int64(999*time.Millisecond), 3, time.Second)) + require.True(t, g.take(now+int64(time.Second), 3, time.Second)) + require.False(t, g.take(now+int64(time.Second), 3, time.Second)) + require.True(t, g.take(now+int64(time.Hour), 3, time.Second)) +} diff --git a/udp/race_off_darwin_test.go b/udp/race_off_darwin_test.go new file mode 100644 index 00000000..f2c1e7a2 --- /dev/null +++ b/udp/race_off_darwin_test.go @@ -0,0 +1,6 @@ +//go:build !race && !e2e_testing + +package udp + +// raceEnabled reports a -race build, whose sync.Pool drops a random share of what is put back. +const raceEnabled = false diff --git a/udp/race_on_darwin_test.go b/udp/race_on_darwin_test.go new file mode 100644 index 00000000..573e9085 --- /dev/null +++ b/udp/race_on_darwin_test.go @@ -0,0 +1,6 @@ +//go:build race && !e2e_testing + +package udp + +// raceEnabled reports a -race build, whose sync.Pool drops a random share of what is put back. +const raceEnabled = true diff --git a/udp/udp_darwin.go b/udp/udp_darwin.go index 2bb0cf0d..fad4ca5a 100644 --- a/udp/udp_darwin.go +++ b/udp/udp_darwin.go @@ -10,6 +10,8 @@ import ( "log/slog" "net" "net/netip" + "sync" + "sync/atomic" "syscall" "unsafe" @@ -22,9 +24,26 @@ type StdConn struct { isV4 bool sysFd uintptr l *slog.Logger + + // listenHost and port are the configured address and the bound port, which a connected socket shares. + listenHost netip.Addr + port uint16 + // rc reaches the listener's descriptor for another listen routine's Rebind. + rc syscall.RawConn + socks *connSocks + // reader is ListenOut's reader, which connected socket readers share; readMu serializes them, since an + // EncReader isn't safe for concurrent use. + reader atomic.Pointer[readerPair] + readMu sync.Mutex +} + +type readerPair struct { + r EncReader + flush func() } var _ Conn = &StdConn{} +var _ EstablishedPeerConn = &StdConn{} func NewListener(l *slog.Logger, s Settings) (Conn, error) { lc := NewListenConfig(s.Multi) @@ -34,12 +53,13 @@ func NewListener(l *slog.Logger, s Settings) (Conn, error) { } if uc, ok := pc.(*net.UDPConn); ok { - c := &StdConn{UDPConn: uc, l: l} + c := &StdConn{UDPConn: uc, l: l, listenHost: s.Listen.Addr()} rc, err := uc.SyscallConn() if err != nil { return nil, fmt.Errorf("failed to open udp socket: %w", err) } + c.rc = rc err = rc.Control(func(fd uintptr) { c.sysFd = fd @@ -53,6 +73,8 @@ func NewListener(l *slog.Logger, s Settings) (Conn, error) { return nil, err } c.isV4 = la.Addr().Is4() + c.port = la.Port() + c.socks = joinConnSocks(c, la, s) return c, nil } @@ -89,6 +111,19 @@ func NewListenConfig(multi bool) net.ListenConfig { func sendto(s int, buf []byte, flags int, to unsafe.Pointer, addrlen int32) (err error) func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error { + if t := u.socks.table.Load(); t != nil { + if p := (*t)[ap]; p != nil { + bufs := [1][]byte{b} + if sent, err := u.connSockRun(p, ap, bufs[:]); sent == 1 || err != nil { + return err + } + } + } + return u.writeTo(b, ap) +} + +// writeTo sends b through the listener. +func (u *StdConn) writeTo(b []byte, ap netip.AddrPort) error { var sa unsafe.Pointer var addrLen int32 @@ -141,15 +176,47 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error { func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) { // An un-sendable destination costs its own packet, never the ones behind it in the batch. - // TODO: WriteTo maps EWOULDBLOCK to an error, so a full send buffer - // silently drops the rest of a burst (linux blocks instead). Poll for - // writability on EAGAIN before giving up on the remainder. + // TODO: writeTo maps EWOULDBLOCK to an error, so a full listener send buffer + // silently drops those packets (linux blocks instead). Poll for + // writability on EAGAIN before giving up on them. written := 0 - for i, b := range bufs { - if err := u.WriteTo(b, addrs[i]); err == nil { - written++ - } else { - u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err) + t := u.socks.table.Load() + if t == nil { + for i, b := range bufs { + if err := u.writeTo(b, addrs[i]); err == nil { + written++ + } else { + u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err) + } + } + return written, nil + } + for off := 0; off < len(bufs); { + // A run of datagrams to one peer goes out together, on its connected socket if it has or earns one. + dst := addrs[off] + end := off + 1 + for end < len(bufs) && addrs[end] == dst { + end++ + } + if p := (*t)[dst]; p != nil { + for off < end { + sent, err := u.connSockRun(p, dst, bufs[off:end]) + written += sent + off += sent + if err == nil { + break + } + u.l.Debug("failed to write packet in batch", "udpAddr", dst, "error", err) + off++ + } + } + // The listener sends what no connected socket took. + for ; off < end; off++ { + if err := u.writeTo(bufs[off], dst); err == nil { + written++ + } else { + u.l.Debug("failed to write packet in batch", "udpAddr", dst, "error", err) + } } } return written, nil @@ -182,6 +249,8 @@ func NewUDPStatsEmitter(udpConns []Conn) func() { func (u *StdConn) ListenOut(r EncReader, flush func()) error { buffer := make([]byte, MTU) + u.reader.Store(&readerPair{r: r, flush: flush}) + u.startConnSocks() for { // Just read one packet at a time @@ -194,18 +263,29 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error { continue } + u.readMu.Lock() r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n:n]) flush() + u.readMu.Unlock() } } +// Close closes the connected sockets of every listen routine, then the listener. +func (u *StdConn) Close() error { + u.stopConnSocks() + return u.UDPConn.Close() +} + func (u *StdConn) SupportsMultipleReaders() bool { return false } // Rebind clears the interface the kernel scoped this socket to, so that sends are routed against the current // routing table instead of the interface we happened to be on when the socket was created. Darwin pins sockets -// this way on its own, which is what strands us after the underlying network changes. +// this way on its own, which is what strands us after the underlying network changes. Control.RebindUDPServer +// rebinds only the first writer, so this clears the other listen routines' listeners too. Connected sockets are +// closed after the listeners are cleared, and later ones copy the cleared scope and pick their source addresses +// afresh. func (u *StdConn) Rebind() error { var err error if u.isV4 { @@ -213,6 +293,23 @@ func (u *StdConn) Rebind() error { } else { err = syscall.SetsockoptInt(int(u.sysFd), syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF, 0) } - + for _, m := range u.socks.siblings(u) { + err = errors.Join(err, m.clearBoundIf()) + } + u.closeConnSocks("rebind") return err } + +// clearBoundIf clears the interface the listener is scoped to through its RawConn, which fails on a listener that +// closed concurrently instead of reaching whatever reused its descriptor number. +func (u *StdConn) clearBoundIf() error { + level, opt := syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF + if u.isV4 { + level, opt = syscall.IPPROTO_IP, syscall.IP_BOUND_IF + } + var serr error + if err := u.rc.Control(func(fd uintptr) { serr = syscall.SetsockoptInt(int(fd), level, opt, 0) }); err != nil { + return err + } + return serr +}