diff --git a/connection_manager.go b/connection_manager.go index af6d7d06..467d6b3a 100644 --- a/connection_manager.go +++ b/connection_manager.go @@ -14,6 +14,7 @@ import ( "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/udp" ) type trafficDecision int @@ -171,7 +172,7 @@ func (cm *connectionManager) Start(ctx context.Context) { } func (cm *connectionManager) doTrafficCheck(localIndex uint32, p, nb, out []byte, now time.Time) { - decision, hostinfo, primary := cm.makeTrafficDecision(localIndex, now) + decision, hostinfo, primary, punch := cm.makeTrafficDecision(localIndex, now) switch decision { case deleteTunnel: @@ -196,6 +197,20 @@ func (cm *connectionManager) doTrafficCheck(localIndex uint32, p, nb, out []byte } cm.resetRelayTrafficCheck(hostinfo) + + if punch != nil { + cm.sendPunch(punch) + } +} + +// sendPunch keeps a tunnel's NAT state alive when we are not sending on it. When we have no route to its remote any +// more, typically after moving to a network without that address family, the tunnel probes its other remotes so the +// peer roams over to us. Nothing else would notice, a tunnel we are not sending on has nothing else that could fail. +func (cm *connectionManager) sendPunch(hostinfo *HostInfo) { + remote, err := cm.punchy.SendPunch(hostinfo) + if udp.IsUnreachable(err) { + hostinfo.remoteUnreachable(cm.intf, remote) + } } func (cm *connectionManager) resetRelayTrafficCheck(hostinfo *HostInfo) { @@ -313,7 +328,9 @@ func (cm *connectionManager) migrateRelayUsed(oldhostinfo, newhostinfo *HostInfo } } -func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Time) (trafficDecision, *HostInfo, *HostInfo) { +// makeTrafficDecision decides what to do with a tunnel under the hostmap lock. It also returns the tunnel to punch when +// we are not sending on it, for the caller to send once the lock is released. +func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Time) (trafficDecision, *HostInfo, *HostInfo, *HostInfo) { // Read lock the main hostmap to order decisions based on tunnels being the primary tunnel cm.hostMap.RLock() defer cm.hostMap.RUnlock() @@ -321,17 +338,17 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim hostinfo := cm.hostMap.Indexes[localIndex] if hostinfo == nil { cm.l.Debug("Not found in hostmap", "localIndex", localIndex) - return doNothing, nil, nil + return doNothing, nil, nil, nil } if cm.isInvalidCertificate(now, hostinfo) { - return closeTunnel, hostinfo, nil + return closeTunnel, hostinfo, nil, nil } if hostinfo.ConnectionState != nil && hostinfo.ConnectionState.messageCounter.Load() >= RejectAfterMessages { // Send path can't encrypt a CloseTunnel notify, so just delete locally; the peer recovers via recv_error. hostinfo.logger(cm.l).Error("Dropping tunnel, message counter is exhausted") - return deleteTunnel, hostinfo, nil + return deleteTunnel, hostinfo, nil, nil } primary := cm.hostMap.Hosts[hostinfo.vpnAddrs[0]] @@ -368,10 +385,10 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim if !outTraffic { // Send a punch packet to keep the NAT state alive - cm.punchy.SendPunch(hostinfo) + return decision, hostinfo, primary, hostinfo } - return decision, hostinfo, primary + return decision, hostinfo, primary, nil } if hostinfo.isPendingDeletion() { @@ -380,7 +397,7 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim "tunnelCheck", m{"state": "dead", "method": "active"}, ) - return deleteTunnel, hostinfo, nil + return deleteTunnel, hostinfo, nil, nil } decision := doNothing @@ -394,14 +411,13 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim "primary", mainHostInfo, ) - return closeTunnel, hostinfo, primary + return closeTunnel, hostinfo, primary, nil } // If we aren't sending or receiving traffic then its an unused tunnel and we don't to test the tunnel. // Just maintain NAT state if configured to do so. - cm.punchy.SendPunch(hostinfo) cm.trafficTimer.Add(hostinfo.localIndexId, cm.checkInterval) - return doNothing, nil, nil + return doNothing, nil, nil, hostinfo } // We aren't receiving traffic but we are sending it. The outbound @@ -427,7 +443,7 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim hostinfo.setPendingDeletion(true) cm.trafficTimer.Add(hostinfo.localIndexId, cm.pendingDeletionInterval) - return decision, hostinfo, nil + return decision, hostinfo, nil, nil } func (cm *connectionManager) isInactive(hostinfo *HostInfo, now time.Time) (time.Duration, bool) { diff --git a/connection_manager_test.go b/connection_manager_test.go index 68b4dc01..96d583ad 100644 --- a/connection_manager_test.go +++ b/connection_manager_test.go @@ -3,6 +3,7 @@ package nebula import ( "crypto/ed25519" "crypto/rand" + "io" "net/netip" "testing" "time" @@ -266,12 +267,12 @@ func Test_NewConnectionManager_CounterLimits(t *testing.T) { // Still below the reject limit, the tunnel stays up nc.In(hostinfo) - decision, _, _ := nc.makeTrafficDecision(hostinfo.localIndexId, time.Now()) + decision, _, _, _ := nc.makeTrafficDecision(hostinfo.localIndexId, time.Now()) assert.Equal(t, tryRehandshake, decision) // At the reject limit, the tunnel is deleted locally without a doomed CloseTunnel notify hostinfo.ConnectionState.messageCounter.Store(RejectAfterMessages) - decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, time.Now()) + decision, _, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, time.Now()) assert.Equal(t, deleteTunnel, decision) } @@ -333,14 +334,14 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) { assert.True(t, (hostinfo.state.Load()&stateIn != 0)) now := time.Now() - decision, _, _ := nc.makeTrafficDecision(hostinfo.localIndexId, now) + decision, _, _, _ := nc.makeTrafficDecision(hostinfo.localIndexId, now) assert.Equal(t, tryRehandshake, decision) assert.Equal(t, now, hostinfo.lastUsed) assert.False(t, hostinfo.isPendingDeletion()) assert.False(t, hostinfo.sentSinceCheck()) assert.False(t, (hostinfo.state.Load()&stateIn != 0)) - decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*5)) + decision, _, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*5)) assert.Equal(t, doNothing, decision) assert.Equal(t, now, hostinfo.lastUsed) assert.False(t, hostinfo.isPendingDeletion()) @@ -348,7 +349,7 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) { assert.False(t, (hostinfo.state.Load()&stateIn != 0)) // Do another traffic check tick, should still not be pending deletion - decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*10)) + decision, _, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Second*10)) assert.Equal(t, doNothing, decision) assert.Equal(t, now, hostinfo.lastUsed) assert.False(t, hostinfo.isPendingDeletion()) @@ -358,7 +359,7 @@ func Test_NewConnectionManager_DisconnectInactive(t *testing.T) { assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0]) // Finally advance beyond the inactivity timeout - decision, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Minute*10)) + decision, _, _, _ = nc.makeTrafficDecision(hostinfo.localIndexId, now.Add(time.Minute*10)) assert.Equal(t, closeTunnel, decision) assert.Equal(t, now, hostinfo.lastUsed) assert.False(t, hostinfo.isPendingDeletion()) @@ -578,3 +579,100 @@ func (d *dummyCert) MarshalJSON() ([]byte, error) { func (d *dummyCert) Copy() cert.Certificate { return d } + +// newPunchTestConnectionManager is a connection manager punching through conn, holding the tunnel from newPunchTarget +func newPunchTestConnectionManager(t *testing.T, conn udp.Conn) (*connectionManager, *HostInfo) { + l := test.NewLogger() + hostMap := newHostMap(l) + hostMap.preferredRanges.Store(&[]netip.Prefix{}) + + lh := newTestLighthouse() + ifce := &Interface{ + hostMap: hostMap, + inside: &overlaytest.NoopTun{}, + outside: &udp.NoopConn{}, + firewall: &Firewall{}, + lightHouse: lh, + pki: &PKI{}, + handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig), + l: l, + } + ifce.pki.cs.Store(&CertState{ + initiatingVersion: cert.Version1, + privateKey: []byte{}, + v1Cert: &dummyCert{version: cert.Version1}, + }) + + punchy := newPunchyForTest(t, map[string]any{"punch": true}, conn) + nc := newConnectionManagerFromConfig(l, config.NewC(l), hostMap, punchy) + nc.intf = ifce + ifce.connectionManager = nc + + hostinfo := newPunchTarget() + hostinfo.localIndexId = 1099 + hostinfo.remoteIndexId = 9901 + hostinfo.ConnectionState = &ConnectionState{myCert: &dummyCert{version: cert.Version1}} + nc.hostMap.unlockedAddHostInfo(hostinfo, ifce) + return nc, hostinfo +} + +// A tunnel we are not sending on gets punched whether or not it is receiving, the punch is the only thing that +// would notice its remote has become unreachable. +func Test_makeTrafficDecision_punchesWhenNotSending(t *testing.T) { + nc, hostinfo := newPunchTestConnectionManager(t, &recordingConn{}) + now := time.Now() + + // Receiving but not sending + hostinfo.takeTraffic() + nc.In(hostinfo) + decision, _, _, punch := nc.makeTrafficDecision(hostinfo.localIndexId, now) + assert.Equal(t, tryRehandshake, decision) + assert.Same(t, hostinfo, punch) + + // Sending and receiving, our own traffic keeps the NAT state alive + nc.Out(hostinfo) + nc.In(hostinfo) + decision, _, _, punch = nc.makeTrafficDecision(hostinfo.localIndexId, now) + assert.Equal(t, tryRehandshake, decision) + assert.Nil(t, punch) + + // Unused + decision, decided, _, punch := nc.makeTrafficDecision(hostinfo.localIndexId, now) + assert.Equal(t, doNothing, decision) + assert.Nil(t, decided, "an unused tunnel has no relay usage to reset") + assert.Same(t, hostinfo, punch) + + // Sending but not receiving gets a test packet instead, which finds an unreachable remote on its own + nc.Out(hostinfo) + decision, _, _, punch = nc.makeTrafficDecision(hostinfo.localIndexId, now) + assert.Equal(t, sendTestPacket, decision) + assert.Nil(t, punch) +} + +func Test_sendPunch_probesWhenRemoteUnreachable(t *testing.T) { + t.Run("unreachable", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + + nc.sendPunch(hostinfo) + assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote) + assert.NotZero(t, hostinfo.nextUnreachableProbe.Load(), "the tunnel should have probed its other remotes") + }) + + t.Run("delivered", func(t *testing.T) { + conn := &recordingConn{} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + + nc.sendPunch(hostinfo) + assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote) + assert.Zero(t, hostinfo.nextUnreachableProbe.Load()) + }) + + t.Run("other failure", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: io.ErrShortWrite}} + nc, hostinfo := newPunchTestConnectionManager(t, conn) + + nc.sendPunch(hostinfo) + assert.Zero(t, hostinfo.nextUnreachableProbe.Load(), "only a missing route says to look elsewhere") + }) +} diff --git a/e2e/dual_stack_roam_test.go b/e2e/dual_stack_roam_test.go index 9dedec2f..a8730403 100644 --- a/e2e/dual_stack_roam_test.go +++ b/e2e/dual_stack_roam_test.go @@ -31,6 +31,8 @@ func TestDualStackRoamToV4Only(t *testing.T) { {name: "peer sends", themSends: true}, {name: "peer sends, rebind", themSends: true, rebind: true}, {name: "peer sends, lh over v4", themSends: true, lhOverV4: true}, + {name: "peer sends, punchy", themSends: true, punchy: true}, + {name: "peer sends, punchy all", themSends: true, punchy: true, punchAll: true}, } { t.Run(tc.name, func(t *testing.T) { t.Parallel() @@ -51,6 +53,8 @@ type dualStackRoamCase struct { // simultaneous has both sides initiate at once at the office, which can leave each side a different primary // tunnel for the other simultaneous bool + // punchy turns on punchy.punch everywhere, punchAll adds punchy.target_all_remotes + punchy, punchAll bool } func fmtRecovery(d time.Duration) string { @@ -115,6 +119,10 @@ func measureDualStackRoam(t *testing.T, tc dualStackRoamCase) (time.Duration, ti "connection_alive_interval": 5, "pending_deletion_interval": 10, }, + "punchy": m{ + "punch": tc.punchy, + "target_all_remotes": tc.punchAll, + }, } me := newDualStackNode(ca, caKey, "me", "10.128.0.2/24", ap("10.0.0.2:4242"), ap("[fd00::2]:4242"), peerCfg) diff --git a/punchy.go b/punchy.go index 67876269..373a0572 100644 --- a/punchy.go +++ b/punchy.go @@ -165,20 +165,24 @@ func (p *Punchy) scheduleJob(job holepunchJob, delay time.Duration) { // SendPunch sends an immediate keepalive punch for an idle hostinfo. // The configured punchy.target_all_remotes mode picks the targets. Gated on punchy.punch and the lighthouse-skip rule // (lighthouses don't get keepalive punches because the regular update interval keeps their NAT state warm). -func (p *Punchy) SendPunch(hostinfo *HostInfo) { +// Returns the hostinfo's current remote and the error from punching it, an invalid remote if it was not punched. +func (p *Punchy) SendPunch(hostinfo *HostInfo) (netip.AddrPort, error) { if !p.punch.Load() { - return + return netip.AddrPort{}, nil } if p.lh.IsAnyLighthouseAddr(hostinfo.vpnAddrs) { - return + return netip.AddrPort{}, nil } + hr := hostinfo.GetRemote() if p.punchEverything.Load() { - p.sendPunchToAllRemotes(hostinfo) - } else if hr := hostinfo.GetRemote(); hr.IsValid() { - p.metricPunchyTx.Inc(1) - p.punchConn.WriteTo([]byte{1}, hr) + return hr, p.sendPunchToAllRemotes(hostinfo, hr) } + if !hr.IsValid() { + return hr, nil + } + p.metricPunchyTx.Inc(1) + return hr, p.punchConn.WriteTo([]byte{1}, hr) } // SendPunchToAll punches every known remote for hostinfo, but only when punchy.target_all_remotes is enabled. @@ -195,14 +199,19 @@ func (p *Punchy) SendPunchToAll(hostinfo *HostInfo) { if p.lh.IsAnyLighthouseAddr(hostinfo.vpnAddrs) { return } - p.sendPunchToAllRemotes(hostinfo) + p.sendPunchToAllRemotes(hostinfo, netip.AddrPort{}) } -func (p *Punchy) sendPunchToAllRemotes(hostinfo *HostInfo) { +// sendPunchToAllRemotes punches every known remote for hostinfo and returns the error from punching current +func (p *Punchy) sendPunchToAllRemotes(hostinfo *HostInfo, current netip.AddrPort) error { + var currentErr error hostinfo.remotes.ForEach(p.hm.GetPreferredRanges(), func(addr netip.AddrPort, preferred bool) { p.metricPunchyTx.Inc(1) - p.punchConn.WriteTo([]byte{1}, addr) + if err := p.punchConn.WriteTo([]byte{1}, addr); addr == current { + currentErr = err + } }) + return currentErr } // Start wires the runtime dependencies and spawns the scheduler worker. diff --git a/punchy_test.go b/punchy_test.go index 25f8de79..47f09af0 100644 --- a/punchy_test.go +++ b/punchy_test.go @@ -3,11 +3,13 @@ package nebula import ( "context" "log/slog" + "net/netip" "testing" "time" "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/test" + "github.com/slackhq/nebula/udp" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -242,3 +244,93 @@ func TestPunchyReloadRestartRequired(t *testing.T) { require.NoError(t, res.Err) assert.True(t, res.RestartRequired) } + +// errNoRoute is the unreachable error every platform classifies the same way +var errNoRoute = udp.ErrInvalidIPv6RemoteForSocket + +// recordingConn records where every write went and fails the ones to an address in fail +type recordingConn struct { + udp.NoopConn + fail map[netip.AddrPort]error + wrote []netip.AddrPort +} + +func (c *recordingConn) WriteTo(_ []byte, addr netip.AddrPort) error { + c.wrote = append(c.wrote, addr) + return c.fail[addr] +} + +type noLighthouses struct{} + +func (noLighthouses) IsAnyLighthouseAddr([]netip.Addr) bool { return false } + +var ( + punchTargetV4 = netip.MustParseAddrPort("10.0.0.3:4242") + punchTargetV6 = netip.MustParseAddrPort("[fd00::3]:4242") +) + +// newPunchTarget is a tunnel on its IPv6 remote that also knows the peer's IPv4 one +func newPunchTarget() *HostInfo { + vpnAddr := netip.MustParseAddr("10.128.0.3") + hostinfo := &HostInfo{ + vpnAddrs: []netip.Addr{vpnAddr}, + remotes: NewRemoteList([]netip.Addr{vpnAddr}, nil), + } + hostinfo.remotes.LearnRemote(vpnAddr, punchTargetV4) + hostinfo.SetRemote(punchTargetV6) + return hostinfo +} + +func newPunchyForTest(t *testing.T, settings map[string]any, conn udp.Conn) *Punchy { + c := config.NewC(test.NewLogger()) + c.Settings["punchy"] = settings + p := NewPunchyFromConfig(test.NewLogger(), c, conn) + + hm := newHostMap(test.NewLogger()) + hm.preferredRanges.Store(&[]netip.Prefix{}) + p.hm = hm + p.lh = noLighthouses{} + return p +} + +func TestPunchy_SendPunchReportsCurrentRemote(t *testing.T) { + t.Run("disabled", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + p := newPunchyForTest(t, map[string]any{"punch": false}, conn) + + remote, err := p.SendPunch(newPunchTarget()) + assert.False(t, remote.IsValid()) + require.NoError(t, err) + assert.Empty(t, conn.wrote) + }) + + t.Run("current remote", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + p := newPunchyForTest(t, map[string]any{"punch": true}, conn) + + remote, err := p.SendPunch(newPunchTarget()) + assert.Equal(t, punchTargetV6, remote) + require.ErrorIs(t, err, errNoRoute) + assert.Equal(t, []netip.AddrPort{punchTargetV6}, conn.wrote) + }) + + t.Run("all remotes, current fails", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV6: errNoRoute}} + p := newPunchyForTest(t, map[string]any{"punch": true, "target_all_remotes": true}, conn) + + remote, err := p.SendPunch(newPunchTarget()) + assert.Equal(t, punchTargetV6, remote) + require.ErrorIs(t, err, errNoRoute) + assert.ElementsMatch(t, []netip.AddrPort{punchTargetV4, punchTargetV6}, conn.wrote) + }) + + t.Run("all remotes, only another fails", func(t *testing.T) { + conn := &recordingConn{fail: map[netip.AddrPort]error{punchTargetV4: errNoRoute}} + p := newPunchyForTest(t, map[string]any{"punch": true, "target_all_remotes": true}, conn) + + remote, err := p.SendPunch(newPunchTarget()) + assert.Equal(t, punchTargetV6, remote) + require.NoError(t, err, "a failure on a remote we are not using says nothing about the current one") + assert.ElementsMatch(t, []netip.AddrPort{punchTargetV4, punchTargetV6}, conn.wrote) + }) +} diff --git a/udp/unreachable_test.go b/udp/unreachable_test.go new file mode 100644 index 00000000..d8dea75d --- /dev/null +++ b/udp/unreachable_test.go @@ -0,0 +1,37 @@ +//go:build !windows + +package udp + +import ( + "net" + "syscall" + "testing" +) + +func TestIsUnreachable(t *testing.T) { + wrap := func(err error) error { return &net.OpError{Op: "write", Net: "udp", Err: err} } + + for _, err := range []error{ + wrap(syscall.ENETUNREACH), + wrap(syscall.EHOSTUNREACH), + wrap(syscall.EADDRNOTAVAIL), + wrap(syscall.EAFNOSUPPORT), + ErrInvalidIPv6RemoteForSocket, + } { + if !IsUnreachable(err) { + t.Errorf("IsUnreachable(%v) = false, want true", err) + } + } + + for _, err := range []error{ + nil, + wrap(syscall.EPERM), + wrap(syscall.ENOBUFS), + wrap(syscall.EMSGSIZE), + net.ErrClosed, + } { + if IsUnreachable(err) { + t.Errorf("IsUnreachable(%v) = true, want false", err) + } + } +}