add locks around replay window updates (#1802)

connection_state.go

@@ -2,11 +2,13 @@ package nebula
 
 import (
 	"encoding/json"
+	"log/slog"
 	"sync"
 	"sync/atomic"
 
 	"github.com/slackhq/nebula/cert"
 	"github.com/slackhq/nebula/handshake"
+	"github.com/slackhq/nebula/header"
 	"github.com/slackhq/nebula/noiseutil"
 )
 
@@ -20,6 +22,7 @@ type ConnectionState struct {
 	initiator      bool
 	messageCounter atomic.Uint64
 	window         *Bits
+	decryptLock    sync.Mutex
 	writeLock      sync.Mutex
 }
 
@@ -54,3 +57,52 @@ func (cs *ConnectionState) MarshalJSON() ([]byte, error) {
 func (cs *ConnectionState) Curve() cert.Curve {
 	return cs.myCert.Curve()
 }
+
+func (cs *ConnectionState) Decrypt(l *slog.Logger, messageCounter uint64, out []byte, packet []byte, nb []byte) ([]byte, error) {
+	var err error
+	cs.decryptLock.Lock()
+	result := cs.window.Check(l, messageCounter)
+	cs.decryptLock.Unlock()
+	if !result {
+		return nil, ErrAlreadySeen
+	}
+
+	out, err = cs.dKey.DecryptDanger(out, packet[:header.Len], packet[header.Len:], messageCounter, nb)
+	if err != nil {
+		return nil, err
+	}
+
+	cs.decryptLock.Lock()
+	result = cs.window.Update(l, messageCounter)
+	cs.decryptLock.Unlock()
+	if !result {
+		return nil, ErrAlreadySeen
+	}
+	return out, nil
+}
+
+// VerifyRelay verifies AEAD protected (but not encrypted) relay frames. packet must be length-checked by the caller.
+func (cs *ConnectionState) VerifyRelay(l *slog.Logger, messageCounter uint64, packet []byte, nb []byte) error {
+	cs.decryptLock.Lock()
+	result := cs.window.Check(l, messageCounter)
+	cs.decryptLock.Unlock()
+	if !result {
+		return ErrAlreadySeen
+	}
+
+	signedPayload := packet[:len(packet)-cs.dKey.Overhead()]
+	signatureValue := packet[len(packet)-cs.dKey.Overhead():]
+	_, err := cs.dKey.DecryptDanger(nil, signedPayload, signatureValue, messageCounter, nb)
+	if err != nil {
+		return err
+	}
+
+	cs.decryptLock.Lock()
+	result = cs.window.Update(l, messageCounter)
+	cs.decryptLock.Unlock()
+	if !result {
+		return ErrAlreadySeen
+	}
+
+	return nil
+}

outside.go

@@ -102,27 +102,31 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
 		return
 	}
 
-	// All remaining packets are encrypted
-	ci := hostinfo.ConnectionState
-	if !ci.window.Check(f.l, h.MessageCounter) {
+	if len(packet) < header.Len+hostinfo.ConnectionState.dKey.Overhead() {
+		f.messageMetrics.RxInvalid(1)
+		if f.l.Enabled(context.Background(), slog.LevelDebug) {
+			f.l.Debug("packet too small", "from", via, "length", len(packet))
+		}
 		return
 	}
 
-	// Relay packets are special
+	// All remaining packets are encrypted
 	if isMessageRelay {
+		// Relay packets are special, this branch should always early-return
+		if err = hostinfo.ConnectionState.VerifyRelay(f.l, h.MessageCounter, packet, nb); err != nil {
+			if f.l.Enabled(context.Background(), slog.LevelDebug) {
+				hostinfo.logger(f.l).Debug("Failed to verify relay packet", "error", err, "from", via, "header", h)
+			}
+			return
+		}
 		f.handleOutsideRelayPacket(hostinfo, via, out, packet, h, fwPacket, lhf, nb, q, localCache)
-
 		return
 	}
 
-	out, err = f.decrypt(hostinfo, h.MessageCounter, out, packet, h, nb)
+	out, err = hostinfo.ConnectionState.Decrypt(f.l, h.MessageCounter, out, packet, nb)
 	if err != nil {
 		if f.l.Enabled(context.Background(), slog.LevelDebug) {
-			hostinfo.logger(f.l).Debug("Failed to decrypt packet",
-				"error", err,
-				"from", via,
-				"header", h,
-			)
+			hostinfo.logger(f.l).Debug("Failed to decrypt packet", "error", err, "from", via, "header", h)
 		}
 		return
 	}
@@ -151,7 +155,7 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
 			// No-op, useful for the Roaming and connectionManager side-effects above
 		case header.TestRequest:
 			//recycle the input packet ciphertext as our output buffer
-			f.send(header.Test, header.TestReply, ci, hostinfo, out, nb, packet)
+			f.send(header.Test, header.TestReply, hostinfo.ConnectionState, hostinfo, out, nb, packet)
 		default:
 			hostinfo.logger(f.l).Error("IsValidSubType was true, but unexpected test subtype seen", "from", via, "header", h)
 			return
@@ -170,27 +174,8 @@ func (f *Interface) readOutsidePackets(via ViaSender, out []byte, packet []byte,
 }
 
 func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender, out []byte, packet []byte, h *header.H, fwPacket *firewall.Packet, lhf *LightHouseHandler, nb []byte, q int, localCache firewall.ConntrackCache) {
-	// The entire body is sent as AD, not encrypted.
-	// The packet consists of a 16-byte parsed Nebula header, Associated Data-protected payload, and a trailing 16-byte AEAD signature value.
-	// The packet is guaranteed to be at least 16 bytes at this point, b/c it got past the h.Parse() call above. If it's
-	// otherwise malformed (meaning, there is no trailing 16 byte AEAD value), then this will result in at worst a 0-length slice
-	// which will gracefully fail in the DecryptDanger call.
-	signedPayload := packet[:len(packet)-hostinfo.ConnectionState.dKey.Overhead()]
-	signatureValue := packet[len(packet)-hostinfo.ConnectionState.dKey.Overhead():]
-	var err error
-	out, err = hostinfo.ConnectionState.dKey.DecryptDanger(out, signedPayload, signatureValue, h.MessageCounter, nb)
-	if err != nil {
-		return
-	}
-	// Advance the replay window now that the frame is authenticated
-	if !hostinfo.ConnectionState.window.Update(f.l, h.MessageCounter) {
-		if f.l.Enabled(context.Background(), slog.LevelDebug) {
-			hostinfo.logger(f.l).Debug("dropping out of window relay packet", "header", h)
-		}
-		return
-	}
-	// Successfully validated the thing. Get rid of the Relay header.
-	signedPayload = signedPayload[header.Len:]
+	// Successfully validated the thing. Get rid of the Relay header and the AEAD tag
+	signedPayload := packet[header.Len : len(packet)-hostinfo.ConnectionState.dKey.Overhead()]
 	// Pull the Roaming parts up here, and return in all call paths.
 	f.handleHostRoaming(hostinfo, via)
 	// Track usage of both the HostInfo and the Relay for the received & authenticated packet
@@ -234,9 +219,10 @@ func (f *Interface) handleOutsideRelayPacket(hostinfo *HostInfo, via ViaSender,
 		if targetRelay.State == Established {
 			switch targetRelay.Type {
 			case ForwardingType:
-				// Forward this packet through the relay tunnel
-				// Find the target HostInfo
-				f.SendVia(targetHI, targetRelay, signedPayload, nb, out, false)
+				// Forward this packet through the relay tunnel, rebuilding it in place.
+				// Encode overwrites the old outer header, and the new AEAD tag lands where the old one was
+				fwdBuf := packet[:0:len(packet)] // Cap to len(packet) to protect memory from a larger parent buffer
+				f.SendVia(targetHI, targetRelay, signedPayload, nb, fwdBuf, true)
 			case TerminalType:
 				hostinfo.logger(f.l).Error("Unexpected Relay Type of Terminal")
 				return
@@ -503,20 +489,6 @@ func parseV4(data []byte, incoming bool, fp *firewall.Packet) error {
 	return nil
 }
 
-func (f *Interface) decrypt(hostinfo *HostInfo, mc uint64, out []byte, packet []byte, h *header.H, nb []byte) ([]byte, error) {
-	var err error
-	out, err = hostinfo.ConnectionState.dKey.DecryptDanger(out, packet[:header.Len], packet[header.Len:], mc, nb)
-	if err != nil {
-		return nil, err
-	}
-
-	if !hostinfo.ConnectionState.window.Update(f.l, mc) {
-		return nil, ErrOutOfWindow
-	}
-
-	return out, nil
-}
-
 func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, packet []byte, fwPacket *firewall.Packet, nb []byte, q int, localCache firewall.ConntrackCache) {
 	err := newPacket(out, true, fwPacket)
 	if err != nil {