Repository: nats-io/nats-server
Pull request #4244 (2023-06-13 19:32:09)

Support TLS 1.2 RSA signature with PKCS#1 v1.5 padding
Cert Store (aka wincert) feature wasn't properly handling TLS 1.2 handshake with TLS 1.2 clients that do not support RSA signature with PSS padding.

With this update, Cert Store will perform either PKCS#1 v1.5 or PSS padding for RSA signature depending on what type is negotiated by the TLS 1.2 client.

Issue surfaces with the NATS .NET v1 client which supports TLS 1.2 only (.NET 4.6.2 dependency) only when the client application was hosted on Windows 10 Enterprise LTSC 2019 (equivalent also to Windows 10 1809 and Windows Server 2019).  If the same client was executed on a more modern Windows 10 release, RSA signature with PSS padding was negotiated and succeeded normally.

The Go NATS client as well as any client operating at TLS 1.3 level would not exhibit the issue as TLS 1.3 requires PSS.

Fix tested good on Windows 10 Enterprise LTSC 2019 host and in confirmed fixed in user's Windows environment where the issue was originally detected.



Patch:
diff --git a/server/certstore/certstore_windows.go b/server/certstore/certstore_windows.go
index 7c1054dc..57adc187 100644
--- a/server/certstore/certstore_windows.go
+++ b/server/certstore/certstore_windows.go
@@ -62,6 +62,7 @@ const (
 
 	winNcryptKeySpec = 0xFFFFFFFF // CERT_NCRYPT_KEY_SPEC
 
+	winBCryptPadPKCS1   uintptr = 0x2
 	winBCryptPadPSS     uintptr = 0x8 // Modern TLS 1.2+
 	winBCryptPadPSSSalt uint32  = 32  // default 20, 32 optimal for typical SHA256 hash
 
@@ -127,6 +128,10 @@ var (
 	winFnGetProperty = winGetProperty
 )
 
+type winPKCS1PaddingInfo struct {
+	pszAlgID *uint16
+}
+
 type winPSSPaddingInfo struct {
 	pszAlgID *uint16
 	cbSalt   uint32
@@ -405,7 +410,12 @@ func (k winKey) Sign(_ io.Reader, digest []byte, opts crypto.SignerOpts) ([]byte
 		if !ok {
 			return nil, ErrBadRSAHashAlgorithm
 		}
-		return winSignRSA(k.handle, digest, algID)
+		switch opts.(type) {
+		case *rsa.PSSOptions:
+			return winSignRSAPSSPadding(k.handle, digest, algID)
+		default:
+			return winSignRSAPKCS1Padding(k.handle, digest, algID)
+		}
 	default:
 		return nil, ErrBadSigningAlgorithm
 	}
@@ -467,8 +477,44 @@ func winPackECDSASigValue(r io.Reader, digestLength int) ([]byte, error) {
 	return b.Bytes()
 }
 
-func winSignRSA(kh uintptr, digest []byte, algID *uint16) ([]byte, error) {
-	// PSS padding for TLS 1.2+
+func winSignRSAPKCS1Padding(kh uintptr, digest []byte, algID *uint16) ([]byte, error) {
+	// PKCS#1 v1.5 padding for some TLS 1.2
+	padInfo := winPKCS1PaddingInfo{pszAlgID: algID}
+	var size uint32
+	// Obtain the size of the signature
+	r, _, _ := winNCryptSignHash.Call(
+		kh,
+		uintptr(unsafe.Pointer(&padInfo)),
+		uintptr(unsafe.Pointer(&digest[0])),
+		uintptr(len(digest)),
+		0,
+		0,
+		uintptr(unsafe.Pointer(&size)),
+		winBCryptPadPKCS1)
+	if r != 0 {
+		return nil, ErrStoreRSASigningError
+	}
+
+	// Obtain the signature data
+	sig := make([]byte, size)
+	r, _, _ = winNCryptSignHash.Call(
+		kh,
+		uintptr(unsafe.Pointer(&padInfo)),
+		uintptr(unsafe.Pointer(&digest[0])),
+		uintptr(len(digest)),
+		uintptr(unsafe.Pointer(&sig[0])),
+		uintptr(size),
+		uintptr(unsafe.Pointer(&size)),
+		winBCryptPadPKCS1)
+	if r != 0 {
+		return nil, ErrStoreRSASigningError
+	}
+
+	return sig[:size], nil
+}
+
+func winSignRSAPSSPadding(kh uintptr, digest []byte, algID *uint16) ([]byte, error) {
+	// PSS padding for TLS 1.3 and some TLS 1.2
 	padInfo := winPSSPaddingInfo{pszAlgID: algID, cbSalt: winBCryptPadPSSSalt}
 
 	var size uint32
diff --git a/server/mqtt.go b/server/mqtt.go
index 75f3f8d4..018d3577 100644
--- a/server/mqtt.go
+++ b/server/mqtt.go
@@ -23,7 +23,6 @@ import (
 	"io"
 	"net"
 	"net/http"
-	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -222,13 +221,13 @@ type mqttSessionManager struct {
 
 type mqttAccountSessionManager struct {
 	mu         sync.RWMutex
-	sessions   map[string]*mqttSession        // key is MQTT client ID
-	sessByHash map[string]*mqttSession        // key is MQTT client ID hash
-	sessLocked map[string]struct{}            // key is MQTT client ID and indicate that a session can not be taken by a new client at this time
-	flappers   map[string]int64               // When connection connects with client ID already in use
-	flapTimer  *time.Timer                    // Timer to perform some cleanup of the flappers map
-	sl         *Sublist                       // sublist allowing to find retained messages for given subscription
-	retmsgs    map[string]*mqttRetainedMsgRef // retained messages
+	sessions   map[string]*mqttSession     // key is MQTT client ID
+	sessByHash map[string]*mqttSession     // key is MQTT client ID hash
+	sessLocked map[string]struct{}         // key is MQTT client ID and indicate that a session can not be taken by a new client at this time
+	flappers   map[string]int64            // When connection connects with client ID already in use
+	flapTimer  *time.Timer                 // Timer to perform some cleanup of the flappers map
+	sl         *Sublist                    // sublist allowing to find retained messages for given subscription
+	retmsgs    map[string]*mqttRetainedMsg // retained messages
 	jsa        mqttJSA
 	rrmLastSeq uint64           // Restore retained messages expected last sequence
 	rrmDoneCh  chan struct{}    // To notify the caller that all retained messages have been loaded
@@ -294,9 +293,8 @@ type mqttRetainedMsg struct {
 	Msg     []byte `json:"msg,omitempty"`
 	Flags   byte   `json:"flags,omitempty"`
 	Source  string `json:"source,omitempty"`
-}
 
-type mqttRetainedMsgRef struct {
+	// non exported
 	sseq  uint64
 	floor uint64
 	sub   *subscription
@@ -1606,9 +1604,8 @@ func (as *mqttAccountSessionManager) processRetainedMsg(_ *subscription, c *clie
 	seq, _, _ := ackReplyInfo(reply)
 
 	// Handle this retained message
-	rf := &mqttRetainedMsgRef{}
-	rf.sseq = seq
-	as.handleRetainedMsg(rm.Subject, rf)
+	rm.sseq = seq
+	as.handleRetainedMsg(rm.Subject, rm)
 
 	// If we were recovering (lastSeq > 0), then check if we are done.
 	if as.rrmLastSeq > 0 && seq >= as.rrmLastSeq {
@@ -1876,11 +1873,11 @@ func (as *mqttAccountSessionManager) sendJSAPIrequests(s *Server, c *client, acc
 // or 0 if the record was added instead of updated.
 //
 // Lock not held on entry.
-func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetainedMsgRef) {
+func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetainedMsg) uint64 {
 	as.mu.Lock()
 	defer as.mu.Unlock()
 	if as.retmsgs == nil {
-		as.retmsgs = make(map[string]*mqttRetainedMsgRef)
+		as.retmsgs = make(map[string]*mqttRetainedMsg)
 		as.sl = NewSublistWithCache()
 	} else {
 		// Check if we already had one. If so, update the existing one.
@@ -1888,9 +1885,15 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetai
 			// If the new sequence is below the floor or the existing one,
 			// then ignore the new one.
 			if rm.sseq <= erm.sseq || rm.sseq <= erm.floor {
-				return
+				return 0
 			}
+			// Update the existing retained message record with the new rm record.
+			erm.Origin = rm.Origin
+			erm.Msg = rm.Msg
+			erm.Flags = rm.Flags
+			erm.Source = rm.Source
 			// Capture existing sequence number so we can return it as the old sequence.
+			oldSeq := erm.sseq
 			erm.sseq = rm.sseq
 			// Clear the floor
 			erm.floor = 0
@@ -1900,11 +1903,13 @@ func (as *mqttAccountSessionManager) handleRetainedMsg(key string, rm *mqttRetai
 				erm.sub = &subscription{subject: []byte(key)}
 				as.sl.Insert(erm.sub)
 			}
+			return oldSeq
 		}
 	}
 	rm.sub = &subscription{subject: []byte(key)}
 	as.retmsgs[key] = rm
 	as.sl.Insert(rm.sub)
+	return 0
 }
 
 // Removes the retained message for the given `subject` if present, and returns the
@@ -1917,7 +1922,7 @@ func (as *mqttAccountSessionManager) handleRetainedMsgDel(subject string, seq ui
 	var seqToRemove uint64
 	as.mu.Lock()
 	if as.retmsgs == nil {
-		as.retmsgs = make(map[string]*mqttRetainedMsgRef)
+		as.retmsgs = make(map[string]*mqttRetainedMsg)
 		as.sl = NewSublistWithCache()
 	}
 	if erm, ok := as.retmsgs[subject]; ok {
@@ -1936,8 +1941,8 @@ func (as *mqttAccountSessionManager) handleRetainedMsgDel(subject string, seq ui
 			seqToRemove = erm.sseq
 		}
 	} else if seq != 0 {
-		rf := &mqttRetainedMsgRef{floor: seq}
-		as.retmsgs[subject] = rf
+		rm := &mqttRetainedMsg{Subject: subject, floor: seq}
+		as.retmsgs[subject] = rm
 	}
 	as.mu.Unlock()
 	return seqToRemove
@@ -2188,16 +2193,11 @@ func (as *mqttAccountSessionManager) getRetainedPublishMsgs(subject string, rms
 		return
 	}
 	for _, sub := range result.psubs {
-		subj := mqttRetainedMsgsStreamSubject + as.domainTk + string(sub.subject)
-		jsm, err := as.jsa.loadLastMsgFor(mqttRetainedMsgsStreamName, subj)
-		if err != nil || jsm == nil {
-			continue
+		// Since this is a reverse match, the subscription objects here
+		// contain literals corresponding to the published subjects.
+		if rm, ok := as.retmsgs[string(sub.subject)]; ok {
+			*rms = append(*rms, rm)
 		}
-		var rm mqttRetainedMsg
-		if err := json.Unmarshal(jsm.Data, &rm); err != nil {
-			continue
-		}
-		*rms = append(*rms, &rm)
 	}
 }
 
@@ -3213,11 +3213,13 @@ func (c *client) mqttHandlePubRetain() {
 		smr, err := asm.jsa.storeMsg(mqttRetainedMsgsStreamSubject+asm.domainTk+key, -1, rmBytes)
 		if err == nil {
 			// Update the new sequence
-			rf := &mqttRetainedMsgRef{
-				sseq: smr.Sequence,
-			}
+			rm.sseq = smr.Sequence
 			// Add/update the map
-			asm.handleRetainedMsg(key, rf)
+			oldSeq := asm.handleRetainedMsg(key, rm)
+			// If this is a new message on the same subject, delete the old one.
+			if oldSeq != 0 {
+				asm.deleteRetainedMsg(oldSeq)
+			}
 		} else {
 			c.mu.Lock()
 			acc := c.acc
@@ -3254,50 +3256,14 @@ func (s *Server) mqttCheckPubRetainedPerms() {
 	}
 	s.mu.Unlock()
 
-	// First get a list of all of the sessions.
 	sm.mu.RLock()
-	asms := make([]*mqttAccountSessionManager, 0, len(sm.sessions))
-	for _, asm := range sm.sessions {
-		asms = append(asms, asm)
-	}
-	sm.mu.RUnlock()
-
-	type retainedMsg struct {
-		subj string
-		rmsg *mqttRetainedMsgRef
-	}
-
-	// For each session we will obtain a list of retained messages.
-	var _rms [128]retainedMsg
-	rms := _rms[:0]
-	for _, asm := range asms {
-		// Get all of the retained messages. Then we will sort them so
-		// that they are in sequence order, which should help the file
-		// store to not have to load out-of-order blocks so often.
-		asm.mu.RLock()
-		rms = rms[:0] // reuse slice
-		for subj, rf := range asm.retmsgs {
-			rms = append(rms, retainedMsg{
-				subj: subj,
-				rmsg: rf,
-			})
-		}
-		asm.mu.RUnlock()
-		sort.Slice(rms, func(i, j int) bool {
-			return rms[i].rmsg.sseq < rms[j].rmsg.sseq
-		})
+	defer sm.mu.RUnlock()
 
+	for _, asm := range sm.sessions {
 		perms := map[string]*perm{}
 		deletes := map[string]uint64{}
-		for _, rf := range rms {
-			jsm, err := asm.jsa.loadMsg(mqttRetainedMsgsStreamName, rf.rmsg.sseq)
-			if err != nil || jsm == nil {
-				continue
-			}
-			var rm mqttRetainedMsg
-			if err := json.Unmarshal(jsm.Data, &rm); err != nil {
-				continue
-			}
+		asm.mu.Lock()
+		for subject, rm := range asm.retmsgs {
 			if rm.Source == _EMPTY_ {
 				continue
 			}
@@ -3311,7 +3277,7 @@ func (s *Server) mqttCheckPubRetainedPerms() {
 				}
 				// If there is permission and no longer allowed to publish in
 				// the subject, remove the publish retained message from the map.
-				if p != nil && !pubAllowed(p, rf.subj) {
+				if p != nil && !pubAllowed(p, subject) {
 					u = nil
 				}
 			}
@@ -3319,14 +3285,12 @@ func (s *Server) mqttCheckPubRetainedPerms() {
 			// Not present or permissions have changed such that the source can't
 			// publish on that subject anymore: remove it from the map.
 			if u == nil {
-				asm.mu.Lock()
-				delete(asm.retmsgs, rf.subj)
-				asm.sl.Remove(rf.rmsg.sub)
-				asm.mu.Unlock()
-				deletes[rf.subj] = rf.rmsg.sseq
+				delete(asm.retmsgs, subject)
+				asm.sl.Remove(rm.sub)
+				deletes[subject] = rm.sseq
 			}
 		}
-
+		asm.mu.Unlock()
 		for subject, seq := range deletes {
 			asm.deleteRetainedMsg(seq)
 			asm.notifyRetainedMsgDeleted(subject, seq)


Test patch:
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index a4e793eb..31f59da9 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -2975,8 +2975,8 @@ func TestMQTTRetainedMsgNetworkUpdates(t *testing.T) {
 		t.Run(test.subject, func(t *testing.T) {
 			for _, a := range test.order {
 				if a.add {
-					rf := &mqttRetainedMsgRef{sseq: a.seq}
-					asm.handleRetainedMsg(test.subject, rf)
+					rm := &mqttRetainedMsg{sseq: a.seq}
+					asm.handleRetainedMsg(test.subject, rm)
 				} else {
 					asm.handleRetainedMsgDel(test.subject, a.seq)
 				}
@@ -2988,8 +2988,8 @@ func TestMQTTRetainedMsgNetworkUpdates(t *testing.T) {
 	for _, subject := range []string{"foo.5", "foo.6"} {
 		t.Run("clear_"+subject, func(t *testing.T) {
 			// Now add a new message, which should clear the floor.
-			rf := &mqttRetainedMsgRef{sseq: 3}
-			asm.handleRetainedMsg(subject, rf)
+			rm := &mqttRetainedMsg{sseq: 3}
+			asm.handleRetainedMsg(subject, rm)
 			check(t, subject, true, 3, 0)
 			// Now do a non network delete and make sure it is gone.
 			asm.handleRetainedMsgDel(subject, 0)