From f15d10fc54d8624610eb63631740ea2ba8456dc4 Mon Sep 17 00:00:00 2001 From: Nate Brown Date: Wed, 19 Aug 2026 20:48:02 -0500 Subject: [PATCH] Enforce outbound message counter limits to prevent nonce reuse (#1841) --- connection_manager.go | 20 ++++++++++ connection_manager_test.go | 73 ++++++++++++++++++++++++++++++++++ connection_state.go | 33 +++++++++++++-- connection_state_test.go | 55 ++++++++++++++++++++++++- handshake_manager.go | 16 +++++++- inside.go | 26 +++++++++++- main.go | 16 ++++++-- message_metrics.go | 17 ++++++-- noiseutil/aesgcm.go | 3 ++ noiseutil/chachapoly.go | 3 ++ noiseutil/cipher_state.go | 11 +++++ noiseutil/cipher_state_test.go | 19 +++++++++ 12 files changed, 276 insertions(+), 16 deletions(-) diff --git a/connection_manager.go b/connection_manager.go index 88f31321..cd711c8e 100644 --- a/connection_manager.go +++ b/connection_manager.go @@ -323,6 +323,12 @@ func (cm *connectionManager) makeTrafficDecision(localIndex uint32, now time.Tim return closeTunnel, hostinfo, 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 + } + primary := cm.hostMap.Hosts[hostinfo.vpnAddrs[0]] mainHostInfo := true if primary != nil && primary != hostinfo { @@ -448,6 +454,11 @@ func (cm *connectionManager) shouldSwapPrimary(current *HostInfo) bool { return false } + if current.ConnectionState.messageCounter.Load() >= RehandshakeAfterMessages { + // This tunnel is being rolled for counter exhaustion, never swap back onto its spent key. + return false + } + crt := cm.intf.pki.getCertState().getCertificate(current.ConnectionState.myCert.Version()) if crt == nil { //my cert was reloaded away. We should definitely swap from this tunnel @@ -544,6 +555,15 @@ func (cm *connectionManager) tryRehandshake(hostinfo *HostInfo) { "reason", "current cert version < pki.initiatingVersion", ) + cm.intf.handshakeManager.StartHandshake(hostinfo.vpnAddrs[0], nil) + return + } + if hostinfo.ConnectionState.messageCounter.Load() >= RehandshakeAfterMessages { + cm.l.Info("Re-handshaking with remote", + "vpnAddrs", hostinfo.vpnAddrs, + "reason", "message counter rehandshake threshold reached", + ) + cm.intf.handshakeManager.StartHandshake(hostinfo.vpnAddrs[0], nil) return } diff --git a/connection_manager_test.go b/connection_manager_test.go index 25637c25..0511e69e 100644 --- a/connection_manager_test.go +++ b/connection_manager_test.go @@ -199,6 +199,79 @@ func Test_NewConnectionManagerTest2(t *testing.T) { assert.Contains(t, nc.hostMap.Hosts, hostinfo.vpnAddrs[0]) } +func Test_NewConnectionManager_CounterLimits(t *testing.T) { + l := test.NewLogger() + localrange := netip.MustParsePrefix("10.1.1.1/24") + vpnIp := netip.MustParseAddr("172.1.1.2") + preferredRanges := []netip.Prefix{localrange} + + // Very incomplete mock objects + hostMap := newHostMap(l) + hostMap.preferredRanges.Store(&preferredRanges) + + cs := &CertState{ + initiatingVersion: cert.Version1, + privateKey: []byte{}, + v1Cert: &dummyCert{version: cert.Version1}, + v1Credential: nil, + } + + lh := newTestLighthouse() + ifce := &Interface{ + hostMap: hostMap, + inside: &overlaytest.NoopTun{}, + outside: &udp.NoopConn{}, + firewall: &Firewall{}, + lightHouse: lh, + pki: &PKI{}, + myVpnAddrs: []netip.Addr{netip.MustParseAddr("172.1.1.1")}, // sorts below vpnIp so shouldSwapPrimary can proceed + handshakeManager: NewHandshakeManager(l, hostMap, lh, &udp.NoopConn{}, defaultHandshakeConfig), + l: l, + } + ifce.pki.cs.Store(cs) + + conf := config.NewC(test.NewLogger()) + punchy := NewPunchyFromConfig(test.NewLogger(), conf, nil) + nc := newConnectionManagerFromConfig(test.NewLogger(), conf, hostMap, punchy) + nc.intf = ifce + + hostinfo := &HostInfo{ + vpnAddrs: []netip.Addr{vpnIp}, + localIndexId: 1099, + remoteIndexId: 9901, + } + hostinfo.ConnectionState = &ConnectionState{ + myCert: &dummyCert{version: cert.Version1}, + } + nc.hostMap.unlockedAddHostInfo(hostinfo, ifce) + + // Below the rehandshake threshold, no handshake is started + hostinfo.ConnectionState.messageCounter.Store(RehandshakeAfterMessages - 1) + nc.tryRehandshake(hostinfo) + assert.Nil(t, ifce.handshakeManager.QueryVpnAddr(vpnIp)) + + // A tunnel on its current cert would normally swap to primary + assert.True(t, nc.shouldSwapPrimary(hostinfo)) + + // At the rehandshake threshold, a new handshake is started + hostinfo.ConnectionState.messageCounter.Store(RehandshakeAfterMessages) + nc.tryRehandshake(hostinfo) + assert.NotNil(t, ifce.handshakeManager.QueryVpnAddr(vpnIp)) + + // An exhausted tunnel being rolled must never swap back to primary onto its spent key + assert.False(t, nc.shouldSwapPrimary(hostinfo)) + + // Still below the reject limit, the tunnel stays up + nc.In(hostinfo) + 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()) + assert.Equal(t, deleteTunnel, decision) +} + func Test_NewConnectionManager_DisconnectInactive(t *testing.T) { l := test.NewLogger() localrange := netip.MustParsePrefix("10.1.1.1/24") diff --git a/connection_state.go b/connection_state.go index 0d6f59e1..7d1f0091 100644 --- a/connection_state.go +++ b/connection_state.go @@ -2,6 +2,7 @@ package nebula import ( "encoding/json" + "fmt" "log/slog" "sync" "sync/atomic" @@ -12,7 +13,18 @@ import ( "github.com/slackhq/nebula/noiseutil" ) -const ReplayWindow = 1024 +const ( + ReplayWindow = 1024 + + // RehandshakeAfterMessages rolls keys inside the AES-GCM data-volume margin (~2^-36 advantage at 64KB frames). + RehandshakeAfterMessages = uint64(1) << 34 + + // RejectAfterMessages is the nonce ceiling enforced by noiseutil; a tunnel here is deleted locally, not notified. + RejectAfterMessages = noiseutil.RejectAfterMessages +) + +// RehandshakeAfterMessages must stay below RejectAfterMessages so tunnels roll before the hard send stop. +const _ = RejectAfterMessages - RehandshakeAfterMessages type ConnectionState struct { eKey noiseutil.CipherState @@ -30,7 +42,12 @@ type ConnectionState struct { // completed handshake.Result. It seeds messageCounter and the replay window so // that the post-handshake message indices already used on the wire don't count // as missed traffic in the data plane. -func newConnectionStateFromResult(r *handshake.Result) *ConnectionState { +func newConnectionStateFromResult(r *handshake.Result) (*ConnectionState, error) { + // Refuse a MessageIndex too big for the replay window: it can only be a bug, and would spin the seed loop below. + if r.MessageIndex >= ReplayWindow { + return nil, fmt.Errorf("handshake message index %d exceeds replay window", r.MessageIndex) + } + ci := &ConnectionState{ myCert: r.MyCert, initiator: r.Initiator, @@ -43,7 +60,7 @@ func newConnectionStateFromResult(r *handshake.Result) *ConnectionState { for i := uint64(1); i <= r.MessageIndex; i++ { ci.window.Update(nil, i) } - return ci + return ci, nil } func (cs *ConnectionState) MarshalJSON() ([]byte, error) { @@ -54,6 +71,16 @@ func (cs *ConnectionState) MarshalJSON() ([]byte, error) { }) } +// NextMessageCounter reserves the next 1-based counter; RejectAfterMessages is the first we refuse, pinned to not wrap. +func (cs *ConnectionState) NextMessageCounter() (uint64, bool) { + c := cs.messageCounter.Add(1) + if c >= RejectAfterMessages { + cs.messageCounter.Store(RejectAfterMessages) + return c, false + } + return c, true +} + func (cs *ConnectionState) Curve() cert.Curve { return cs.myCert.Curve() } diff --git a/connection_state_test.go b/connection_state_test.go index dea60d39..766e88eb 100644 --- a/connection_state_test.go +++ b/connection_state_test.go @@ -6,10 +6,12 @@ import ( "time" "github.com/flynn/noise" + "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/cert" ct "github.com/slackhq/nebula/cert_test" "github.com/slackhq/nebula/handshake" "github.com/slackhq/nebula/header" + "github.com/slackhq/nebula/test" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -79,11 +81,51 @@ func runTestHandshake(t *testing.T) (initR, respR *handshake.Result) { return initR, respR } +func TestConnectionState_NextMessageCounter(t *testing.T) { + cs := &ConnectionState{} + cs.messageCounter.Store(RejectAfterMessages - 2) + + c, ok := cs.NextMessageCounter() + assert.True(t, ok) + assert.Equal(t, RejectAfterMessages-1, c) + + // Hitting the limit refuses and pins the counter there + c, ok = cs.NextMessageCounter() + assert.False(t, ok) + assert.Equal(t, RejectAfterMessages, c) + assert.Equal(t, RejectAfterMessages, cs.messageCounter.Load()) + + // Continued send attempts stay refused and the counter never wraps + for i := 0; i < 10; i++ { + _, ok = cs.NextMessageCounter() + assert.False(t, ok) + } + assert.Equal(t, RejectAfterMessages, cs.messageCounter.Load()) +} + +// TestSendNoMetricsDropsExhausted drives the send path to the exhausted drop; metric and out flag prove it. +func TestSendNoMetricsDropsExhausted(t *testing.T) { + initR, _ := runTestHandshake(t) + ci, err := newConnectionStateFromResult(initR) + require.NoError(t, err) + ci.messageCounter.Store(RejectAfterMessages - 1) + + f := &Interface{l: test.NewLogger(), messageMetrics: &MessageMetrics{txExhausted: metrics.NewCounter()}} + hostinfo := &HostInfo{vpnAddrs: []netip.Addr{netip.MustParseAddr("10.0.0.1")}, ConnectionState: ci} + + f.sendNoMetrics(header.Message, 0, ci, hostinfo, netip.AddrPort{}, []byte{}, make([]byte, 12), make([]byte, mtu), 0) + + // The crossing send is refused: it records an exhaustion drop and never reaches connectionManager.Out. + assert.Equal(t, int64(1), f.messageMetrics.txExhausted.Count()) + assert.False(t, hostinfo.out.Load()) +} + func TestNewConnectionStateFromResult(t *testing.T) { initR, respR := runTestHandshake(t) t.Run("initiator", func(t *testing.T) { - ci := newConnectionStateFromResult(initR) + ci, err := newConnectionStateFromResult(initR) + require.NoError(t, err) assert.True(t, ci.initiator) assert.Equal(t, initR.MyCert, ci.myCert) assert.Equal(t, initR.RemoteCert, ci.peerCert) @@ -102,8 +144,17 @@ func TestNewConnectionStateFromResult(t *testing.T) { assert.True(t, ci.window.Check(nil, 3), "counter 3 must not be pre-seeded") }) + t.Run("message index too large is refused", func(t *testing.T) { + bad := *initR + bad.MessageIndex = ReplayWindow + ci, err := newConnectionStateFromResult(&bad) + require.Error(t, err) + assert.Nil(t, ci) + }) + t.Run("responder", func(t *testing.T) { - ci := newConnectionStateFromResult(respR) + ci, err := newConnectionStateFromResult(respR) + require.NoError(t, err) assert.False(t, ci.initiator) assert.Equal(t, respR.MyCert, ci.myCert) assert.Equal(t, respR.RemoteCert, ci.peerCert) diff --git a/handshake_manager.go b/handshake_manager.go index b4bebdc7..f3e801d3 100644 --- a/handshake_manager.go +++ b/handshake_manager.go @@ -749,8 +749,14 @@ func (hm *HandshakeManager) beginHandshake(via ViaSender, packet []byte, h *head return } + connState, err := newConnectionStateFromResult(result) + if err != nil { + f.l.Error("Discarding handshake with an invalid message index", "error", err, "vpnAddrs", vpnAddrs) + return + } + hostinfo := &HostInfo{ - ConnectionState: newConnectionStateFromResult(result), + ConnectionState: connState, localIndexId: result.LocalIndex, remoteIndexId: result.RemoteIndex, vpnAddrs: vpnAddrs, @@ -868,7 +874,13 @@ func (hm *HandshakeManager) continueHandshake(via ViaSender, hh *HandshakeHostIn } // Handshake complete; build the ConnectionState now that we have keys and a verified peer cert. - hostinfo.ConnectionState = newConnectionStateFromResult(result) + cs, err := newConnectionStateFromResult(result) + if err != nil { + f.l.Error("Discarding handshake with an invalid message index", "error", err, "vpnAddrs", hostinfo.vpnAddrs) + hm.DeleteHostInfo(hostinfo) + return + } + hostinfo.ConnectionState = cs remoteCert := result.RemoteCert if remoteCert == nil { diff --git a/inside.go b/inside.go index a80b2e96..c85afc2f 100644 --- a/inside.go +++ b/inside.go @@ -275,6 +275,14 @@ func (f *Interface) sendTo(t header.MessageType, st header.MessageSubType, ci *C f.sendNoMetrics(t, st, ci, hostinfo, remote, p, nb, out, 0) } +// dropExhausted records an exhaustion drop and logs once, on the crossing send, for a spent tunnel. +func (f *Interface) dropExhausted(hostinfo *HostInfo, c uint64, msg string) { + f.messageMetrics.TxExhausted(1) + if c == RejectAfterMessages { + hostinfo.logger(f.l).Error(msg) + } +} + // SendVia sends a payload through a Relay tunnel. No authentication or encryption is done // to the payload for the ultimate target host, making this a useful method for sending // handshake messages to peers through relay tunnels. @@ -294,7 +302,14 @@ func (f *Interface) SendVia(via *HostInfo, // NOTE: for goboring AESGCMTLS we need to lock because of the nonce check via.ConnectionState.writeLock.Lock() } - c := via.ConnectionState.messageCounter.Add(1) + c, ok := via.ConnectionState.NextMessageCounter() + if !ok { + if noiseutil.EncryptLockNeeded { + via.ConnectionState.writeLock.Unlock() + } + f.dropExhausted(via, c, "Dropping outbound relay packets, tunnel message counter is exhausted") + return + } out = header.Encode(out, header.Version, header.Message, header.MessageRelay, relay.RemoteIndex, c) f.connectionManager.Out(via) @@ -361,7 +376,14 @@ func (f *Interface) sendNoMetrics(t header.MessageType, st header.MessageSubType // NOTE: for goboring AESGCMTLS we need to lock because of the nonce check ci.writeLock.Lock() } - c := ci.messageCounter.Add(1) + c, ok := ci.NextMessageCounter() + if !ok { + if noiseutil.EncryptLockNeeded { + ci.writeLock.Unlock() + } + f.dropExhausted(hostinfo, c, "Dropping outbound packets, tunnel message counter is exhausted") + return + } //l.WithField("trace", string(debug.Stack())).Error("out Header ", &Header{Version, t, st, 0, hostinfo.remoteIndexId, c}, p) out = header.Encode(out, header.Version, t, st, hostinfo.remoteIndexId, c) diff --git a/main.go b/main.go index 2ef2031b..da2776f1 100644 --- a/main.go +++ b/main.go @@ -11,6 +11,7 @@ import ( "time" "github.com/slackhq/nebula/config" + "github.com/slackhq/nebula/noiseutil" "github.com/slackhq/nebula/overlay" "github.com/slackhq/nebula/sshd" "github.com/slackhq/nebula/udp" @@ -20,6 +21,12 @@ import ( type m = map[string]any +// maxRoutines caps routines below the RejectHeadroom nonce gap so concurrent senders can't race the counter past wrap. +const maxRoutines = 1 << 16 + +// The reject headroom must exceed every sender that can be mid-reservation at once, about two per routine. +const _ = noiseutil.RejectHeadroom - 4*maxRoutines + func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, deviceFactory overlay.DeviceFactory) (retcon *Control, reterr error) { ctx, cancel := context.WithCancel(context.Background()) // Automatically cancel the context if Main returns an error, to signal all created goroutines to quit. @@ -81,9 +88,6 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev if routines < 1 { routines = 1 } - if routines > 1 { - l.Info("Using multiple routines", "routines", routines) - } } else { // deprecated and undocumented tunQueues := c.GetInt("tun.routines", 1) @@ -93,6 +97,12 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev l.Warn("Setting tun.routines and listen.routines is deprecated. Use `routines` instead", "routines", routines) } } + if routines > maxRoutines { + l.Warn("Using multiple routines", "routines", maxRoutines, "clamped", true, "requestedRoutines", routines) + routines = maxRoutines + } else if routines > 1 { + l.Info("Using multiple routines", "routines", routines) + } // EXPERIMENTAL // Intentionally not documented yet while we do more testing and determine diff --git a/message_metrics.go b/message_metrics.go index 45de9a5c..0ceafc4b 100644 --- a/message_metrics.go +++ b/message_metrics.go @@ -14,7 +14,8 @@ type MessageMetrics struct { rxUnknown metrics.Counter txUnknown metrics.Counter - rxInvalid metrics.Counter + rxInvalid metrics.Counter + txExhausted metrics.Counter } func (m *MessageMetrics) Rx(t header.MessageType, s header.MessageSubType, i int64) { @@ -41,6 +42,13 @@ func (m *MessageMetrics) RxInvalid(i int64) { } } +// TxExhausted counts outbound packets dropped because the tunnel's message counter is spent. +func (m *MessageMetrics) TxExhausted(i int64) { + if m != nil && m.txExhausted != nil { + m.txExhausted.Inc(i) + } +} + func newMessageMetrics() *MessageMetrics { gen := func(t string) [][]metrics.Counter { return [][]metrics.Counter{ @@ -61,9 +69,10 @@ func newMessageMetrics() *MessageMetrics { rx: gen("rx"), tx: gen("tx"), - rxUnknown: metrics.GetOrRegisterCounter("messages.rx.other", nil), - txUnknown: metrics.GetOrRegisterCounter("messages.tx.other", nil), - rxInvalid: metrics.GetOrRegisterCounter("messages.rx.invalid", nil), + rxUnknown: metrics.GetOrRegisterCounter("messages.rx.other", nil), + txUnknown: metrics.GetOrRegisterCounter("messages.tx.other", nil), + rxInvalid: metrics.GetOrRegisterCounter("messages.rx.invalid", nil), + txExhausted: metrics.GetOrRegisterCounter("messages.tx.exhausted", nil), } } diff --git a/noiseutil/aesgcm.go b/noiseutil/aesgcm.go index dcbd5693..cf035913 100644 --- a/noiseutil/aesgcm.go +++ b/noiseutil/aesgcm.go @@ -25,6 +25,9 @@ func (s *CipherStateAESGCM) EncryptDanger(out, ad, plaintext []byte, n uint64, n if s == nil { return nil, errors.New("no cipher state available to encrypt") } + if n >= RejectAfterMessages { + return nil, ErrMessageCounterExhausted + } nb[0] = 0 nb[1] = 0 nb[2] = 0 diff --git a/noiseutil/chachapoly.go b/noiseutil/chachapoly.go index 31ab3bfe..2cec73a6 100644 --- a/noiseutil/chachapoly.go +++ b/noiseutil/chachapoly.go @@ -24,6 +24,9 @@ func (s *CipherStateChaChaPoly) EncryptDanger(out, ad, plaintext []byte, n uint6 if s == nil { return nil, errors.New("no cipher state available to encrypt") } + if n >= RejectAfterMessages { + return nil, ErrMessageCounterExhausted + } nb[0] = 0 nb[1] = 0 nb[2] = 0 diff --git a/noiseutil/cipher_state.go b/noiseutil/cipher_state.go index bb316385..789c8ed5 100644 --- a/noiseutil/cipher_state.go +++ b/noiseutil/cipher_state.go @@ -1,11 +1,22 @@ package noiseutil import ( + "errors" "fmt" + "math" "github.com/flynn/noise" ) +// RejectHeadroom is the wrap gap for senders racing the counter, sized large enough for any routine count. +const RejectHeadroom = uint64(1) << 40 + +// RejectAfterMessages is the nonce ceiling: encrypting stops RejectHeadroom short of the wrap. +const RejectAfterMessages = math.MaxUint64 - RejectHeadroom + +// ErrMessageCounterExhausted is returned by EncryptDanger once the nonce reaches RejectAfterMessages. +var ErrMessageCounterExhausted = errors.New("message counter exhausted") + // CipherState is the post-handshake AEAD cipher used for the data plane. // Each supported cipher has its own concrete implementation in this package with the nonce endianness hardcoded, // so the encrypt/decrypt fast path avoids interface dispatch on the byte order. diff --git a/noiseutil/cipher_state_test.go b/noiseutil/cipher_state_test.go index a4df01e9..01cb959f 100644 --- a/noiseutil/cipher_state_test.go +++ b/noiseutil/cipher_state_test.go @@ -1,6 +1,7 @@ package noiseutil import ( + "math" "testing" "github.com/flynn/noise" @@ -89,6 +90,24 @@ func roundtrip(t *testing.T, enc, dec CipherState) { assert.Equal(t, 16, enc.Overhead()) } +func TestEncryptRejectsExhaustedCounter(t *testing.T) { + // Pin the headroom below the uint64 wrap so a typo can't silently move the ceiling. + require.Equal(t, uint64(1)<<40, RejectHeadroom) + require.Equal(t, math.MaxUint64-RejectHeadroom, RejectAfterMessages) + + encA, _ := buildCipherStates(t, CipherAESGCM) + encC, _ := buildCipherStates(t, noise.CipherChaChaPoly) + nb := make([]byte, 12) + + for _, cs := range []CipherState{NewCipherStateAESGCM(encA), NewCipherStateChaChaPoly(encC)} { + _, err := cs.EncryptDanger(nil, nil, []byte("x"), RejectAfterMessages-1, nb) + require.NoError(t, err) + + _, err = cs.EncryptDanger(nil, nil, []byte("x"), RejectAfterMessages, nb) + require.ErrorIs(t, err, ErrMessageCounterExhausted) + } +} + func BenchmarkCipherStateEncryptAESGCM(b *testing.B) { enc, _ := buildCipherStatesB(b, CipherAESGCM) benchEncryptCipherState(b, NewCipherState(enc, CipherAESGCM))