Enforce outbound message counter limits to prevent nonce reuse (#1841)

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
 	}

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")

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()
 }

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)

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 {

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)

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

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),
 	}
 }
 

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

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

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.

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))