Repository: nats-io/nats-server
Pull request #5904 (2024-09-18 23:34:48)

Add support for setting TLS min version number in tls block
For example, to set minimum version to TLS v1.3:

```
  tls { min_version = "1.3" }
```

Signed-off-by: Waldemar Quevedo <wally@nats.io>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index bc16f6fb..a0a77b5f 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -1748,9 +1748,25 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 		return errPriorState
 	}
 	if matched = bytes.Equal(mb.lastChecksum(), lchk[:]); !matched {
-		// Detected a stale index.db, we didn't write it upon shutdown so can't rely on it being correct.
-		fs.warn("Stream state outdated, will rebuild")
-		return errPriorState
+		// If we are tracking max msgs per subject and we are not up to date we should rebuild.
+		if fs.cfg.MaxMsgsPer > 0 {
+			fs.warn("Stream state block state outdated, will rebuild")
+			return errPriorState
+		}
+
+		// Remove the last message block since recover will add in the new one.
+		fs.removeMsgBlockFromList(mb)
+		// Reverse update of tracking state for this mb, will add new state in below.
+		mstate.Msgs -= mb.msgs
+		mstate.Bytes -= mb.bytes
+		if nmb, err := fs.recoverMsgBlock(mb.index); err != nil && !os.IsNotExist(err) {
+			fs.warn("Stream state could not recover last msg block")
+			os.Remove(fn)
+			return errCorruptState
+		} else if nmb != nil {
+			fs.adjustAccounting(mb, nmb)
+			updateTrackingState(&mstate, nmb)
+		}
 	}
 
 	// On success double check our state.
@@ -1801,6 +1817,67 @@ func (fs *fileStore) recoverFullState() (rerr error) {
 	}
 }
 
+// adjustAccounting will be called when a stream state was only partially accounted for
+// within a message block, e.g. additional records were added after the stream state.
+// Lock should be held.
+func (fs *fileStore) adjustAccounting(mb, nmb *msgBlock) {
+	nmb.mu.Lock()
+	defer nmb.mu.Unlock()
+
+	// First make sure the new block is loaded.
+	if nmb.cacheNotLoaded() {
+		nmb.loadMsgsWithLock()
+	}
+	nmb.ensurePerSubjectInfoLoaded()
+
+	var smv StoreMsg
+
+	// Need to walk previous messages and undo psim stats.
+	// We already undid msgs and bytes accounting.
+	for seq, lseq := atomic.LoadUint64(&mb.first.seq), atomic.LoadUint64(&mb.last.seq); seq <= lseq; seq++ {
+		// Lookup the message. If an error will be deleted, so can skip.
+		sm, err := nmb.cacheLookup(seq, &smv)
+		if err != nil {
+			continue
+		}
+		if len(sm.subj) > 0 && fs.psim != nil {
+			if info, ok := fs.psim.Find(stringToBytes(sm.subj)); ok {
+				info.total--
+			}
+		}
+	}
+
+	// Walk only new messages and update accounting at fs level. Any messages that should have
+	// triggered limits exceeded will be handled after the recovery and prior to the stream
+	// being available to the system.
+	for seq, lseq := atomic.LoadUint64(&mb.last.seq)+1, atomic.LoadUint64(&nmb.last.seq); seq <= lseq; seq++ {
+		// Lookup the message. If an error will be deleted, so can skip.
+		sm, err := nmb.cacheLookup(seq, &smv)
+		if err != nil {
+			continue
+		}
+		// Since we found it we just need to adjust fs totals and psim.
+		fs.state.Msgs++
+		fs.state.Bytes += fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
+	}
+
+	// Now check to see if we had a higher first for the recovered state mb vs nmb.
+	if atomic.LoadUint64(&nmb.first.seq) < atomic.LoadUint64(&mb.first.seq) {
+		// Now set first for nmb.
+		atomic.StoreUint64(&nmb.first.seq, atomic.LoadUint64(&mb.first.seq))
+	}
+
+	// Update top level accounting.
+	if fseq := atomic.LoadUint64(&nmb.first.seq); fs.state.FirstSeq == 0 || fseq < fs.state.FirstSeq {
+		fs.state.FirstSeq = fseq
+		fs.state.FirstTime = time.Unix(0, nmb.first.ts).UTC()
+	}
+	if lseq := atomic.LoadUint64(&nmb.last.seq); lseq > fs.state.LastSeq {
+		fs.state.LastSeq = lseq
+		fs.state.LastTime = time.Unix(0, nmb.last.ts).UTC()
+	}
+}
+
 // Grabs last checksum for the named block file.
 // Takes into account encryption etc.
 func (mb *msgBlock) lastChecksum() []byte {
diff --git a/server/opts.go b/server/opts.go
index 0d1848b3..6bf2cc8e 100644
--- a/server/opts.go
+++ b/server/opts.go
@@ -729,6 +729,7 @@ type TLSConfigOpts struct {
 	CaCertsMatch      []string
 	OCSPPeerConfig    *certidp.OCSPPeerConfig
 	Certificates      []*TLSCertPairOpt
+	MinVersion        uint16
 }
 
 // TLSCertPairOpt are the paths to a certificate and private key.
@@ -4560,6 +4561,24 @@ func parseCurvePreferences(curveName string) (tls.CurveID, error) {
 	return curve, nil
 }
 
+func parseTLSVersion(v any) (uint16, error) {
+	var tlsVersionNumber uint16
+	switch v := v.(type) {
+	case string:
+		n, err := tlsVersionFromString(v)
+		if err != nil {
+			return 0, err
+		}
+		tlsVersionNumber = n
+	default:
+		return 0, fmt.Errorf("'min_version' wrong type: %v", v)
+	}
+	if tlsVersionNumber < tls.VersionTLS12 {
+		return 0, fmt.Errorf("unsupported TLS version: %s", tls.VersionName(tlsVersionNumber))
+	}
+	return tlsVersionNumber, nil
+}
+
 // Helper function to parse TLS configs.
 func parseTLS(v any, isClientCtx bool) (t *TLSConfigOpts, retErr error) {
 	var (
@@ -4825,6 +4844,12 @@ func parseTLS(v any, isClientCtx bool) (t *TLSConfigOpts, retErr error) {
 				}
 				tc.Certificates[i] = certPair
 			}
+		case "min_version":
+			minVersion, err := parseTLSVersion(mv)
+			if err != nil {
+				return nil, &configErr{tk, fmt.Sprintf("error parsing tls config: %v", err)}
+			}
+			tc.MinVersion = minVersion
 		default:
 			return nil, &configErr{tk, fmt.Sprintf("error parsing tls config, unknown field %q", mk)}
 		}
@@ -5199,6 +5224,13 @@ func GenTLSConfig(tc *TLSConfigOpts) (*tls.Config, error) {
 		}
 		config.ClientCAs = pool
 	}
+	// Allow setting TLS minimum version.
+	if tc.MinVersion > 0 {
+		if tc.MinVersion < tls.VersionTLS12 {
+			return nil, fmt.Errorf("unsupported minimum TLS version: %s", tls.VersionName(tc.MinVersion))
+		}
+		config.MinVersion = tc.MinVersion
+	}
 
 	return &config, nil
 }
diff --git a/server/reload.go b/server/reload.go
index 80efcbb4..6d9af462 100644
--- a/server/reload.go
+++ b/server/reload.go
@@ -1151,7 +1151,7 @@ func imposeOrder(value any) error {
 		slices.SortFunc(value.Gateways, func(i, j *RemoteGatewayOpts) int { return cmp.Compare(i.Name, j.Name) })
 	case WebsocketOpts:
 		slices.Sort(value.AllowedOrigins)
-	case string, bool, uint8, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
+	case string, bool, uint8, uint16, int, int32, int64, time.Duration, float64, nil, LeafNodeOpts, ClusterOpts, *tls.Config, PinnedCertSet,
 		*URLAccResolver, *MemAccResolver, *DirAccResolver, *CacheDirAccResolver, Authentication, MQTTOpts, jwt.TagList,
 		*OCSPConfig, map[string]string, JSLimitOpts, StoreCipher, *OCSPResponseCacheConfig:
 		// explicitly skipped types
diff --git a/server/server.go b/server/server.go
index 120f70db..b198e248 100644
--- a/server/server.go
+++ b/server/server.go
@@ -3506,6 +3506,20 @@ func tlsVersion(ver uint16) string {
 	return fmt.Sprintf("Unknown [0x%x]", ver)
 }
 
+func tlsVersionFromString(ver string) (uint16, error) {
+	switch ver {
+	case "1.0":
+		return tls.VersionTLS10, nil
+	case "1.1":
+		return tls.VersionTLS11, nil
+	case "1.2":
+		return tls.VersionTLS12, nil
+	case "1.3":
+		return tls.VersionTLS13, nil
+	}
+	return 0, fmt.Errorf("Unknown version: %v", ver)
+}
+
 // We use hex here so we don't need multiple versions
 func tlsCipher(cs uint16) string {
 	name, present := cipherMapByID[cs]


Test patch:
diff --git a/server/norace_test.go b/server/norace_test.go
index f41aee25..4039e4ef 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -10842,111 +10842,67 @@ func TestNoRaceJetStreamStandaloneDontReplyToAckBeforeProcessingIt(t *testing.T)
 	}
 }
 
-// Under certain scenarios an old index.db with a stream that has msg limits set will not restore properly
-// due to an old index.db and compaction after the index.db took place which could lose per subject information.
-func TestNoRaceFileStoreMsgLimitsAndOldRecoverState(t *testing.T) {
-	for _, test := range []struct {
-		name             string
-		expectedFirstSeq uint64
-		expectedLastSeq  uint64
-		expectedMsgs     uint64
-		transform        func(StreamConfig) StreamConfig
-	}{
-		{
-			name:             "MaxMsgsPer",
-			expectedFirstSeq: 10_001,
-			expectedLastSeq:  1_010_001,
-			expectedMsgs:     1_000_001,
-			transform: func(config StreamConfig) StreamConfig {
-				config.MaxMsgsPer = 1
-				return config
-			},
-		},
-		{
-			name:             "MaxMsgs",
-			expectedFirstSeq: 10_001,
-			expectedLastSeq:  1_010_001,
-			expectedMsgs:     1_000_001,
-			transform: func(config StreamConfig) StreamConfig {
-				config.MaxMsgs = 1_000_001
-				return config
-			},
-		},
-		{
-			name:             "MaxBytes",
-			expectedFirstSeq: 8_624,
-			expectedLastSeq:  1_010_001,
-			expectedMsgs:     1_001_378,
-			transform: func(config StreamConfig) StreamConfig {
-				config.MaxBytes = 1_065_353_216
-				return config
-			},
-		},
-	} {
-		t.Run(test.name, func(t *testing.T) {
-			sd := t.TempDir()
-			fs, err := newFileStore(
-				FileStoreConfig{StoreDir: sd},
-				test.transform(StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage}),
-			)
-			require_NoError(t, err)
-			defer fs.Stop()
+// Under certain scenarios an old index.db with a stream that has max msgs per set will not restore properly
+// due to and old index.db and compaction after the index.db took place which could lose per subject information.
+func TestNoRaceFileStoreMaxMsgsPerSubjectAndOldRecoverState(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage, MaxMsgsPer: 1})
+	require_NoError(t, err)
+	defer fs.Stop()
 
-			msg := make([]byte, 1024)
+	msg := make([]byte, 1024)
 
-			for i := 0; i < 10_000; i++ {
-				subj := fmt.Sprintf("foo.%d", i)
-				fs.StoreMsg(subj, nil, msg)
-			}
+	for i := 0; i < 10_000; i++ {
+		subj := fmt.Sprintf("foo.%d", i)
+		fs.StoreMsg(subj, nil, msg)
+	}
 
-			// This will write the index.db file. We will capture this and use it to replace a new one.
-			sfile := filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile)
-			fs.Stop()
-			_, err = os.Stat(sfile)
-			require_NoError(t, err)
+	// This will write the index.db file. We will capture this and use it to replace a new one.
+	sfile := filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile)
+	fs.Stop()
+	_, err = os.Stat(sfile)
+	require_NoError(t, err)
 
-			// Read it in and make sure len > 0.
-			buf, err := os.ReadFile(sfile)
-			require_NoError(t, err)
-			require_True(t, len(buf) > 0)
+	// Read it in and make sure len > 0.
+	buf, err := os.ReadFile(sfile)
+	require_NoError(t, err)
+	require_True(t, len(buf) > 0)
 
-			// Restart
-			fs, err = newFileStore(
-				FileStoreConfig{StoreDir: sd},
-				test.transform(StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage}),
-			)
-			require_NoError(t, err)
-			defer fs.Stop()
-
-			// Put in more messages with wider range. This will compact a bunch of the previous blocks.
-			for i := 0; i < 1_000_001; i++ {
-				subj := fmt.Sprintf("foo.%d", i)
-				fs.StoreMsg(subj, nil, msg)
-			}
-
-			var ss StreamState
-			fs.FastState(&ss)
-			require_Equal(t, ss.FirstSeq, test.expectedFirstSeq)
-			require_Equal(t, ss.LastSeq, test.expectedLastSeq)
-			require_Equal(t, ss.Msgs, test.expectedMsgs)
-
-			// Now stop again, but replace index.db with old one.
-			fs.Stop()
-			// Put back old stream state.
-			require_NoError(t, os.WriteFile(sfile, buf, defaultFilePerms))
-
-			// Restart
-			fs, err = newFileStore(
-				FileStoreConfig{StoreDir: sd},
-				test.transform(StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage}),
-			)
-			require_NoError(t, err)
-			defer fs.Stop()
+	// Restart
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: sd},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage, MaxMsgsPer: 1})
+	require_NoError(t, err)
+	defer fs.Stop()
 
-			fs.FastState(&ss)
-			require_Equal(t, ss.FirstSeq, test.expectedFirstSeq)
-			require_Equal(t, ss.LastSeq, test.expectedLastSeq)
-			require_Equal(t, ss.Msgs, test.expectedMsgs)
-		})
+	// Put in more messages with wider range. This will compact a bunch of the previous blocks.
+	for i := 0; i < 1_000_001; i++ {
+		subj := fmt.Sprintf("foo.%d", i)
+		fs.StoreMsg(subj, nil, msg)
 	}
+
+	var ss StreamState
+	fs.FastState(&ss)
+	require_Equal(t, ss.FirstSeq, 10_001)
+	require_Equal(t, ss.LastSeq, 1_010_001)
+	require_Equal(t, ss.Msgs, 1_000_001)
+
+	// Now stop again, but replace index.db with old one.
+	fs.Stop()
+	// Put back old stream state.
+	require_NoError(t, os.WriteFile(sfile, buf, defaultFilePerms))
+
+	// Restart
+	fs, err = newFileStore(
+		FileStoreConfig{StoreDir: sd},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage, MaxMsgsPer: 1})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	fs.FastState(&ss)
+	require_Equal(t, ss.FirstSeq, 10_001)
+	require_Equal(t, ss.LastSeq, 1_010_001)
+	require_Equal(t, ss.Msgs, 1_000_001)
 }
diff --git a/server/server_test.go b/server/server_test.go
index e0b6d6e7..bb300af4 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -217,7 +217,102 @@ func TestTLSVersions(t *testing.T) {
 	}
 }
 
-func TestTlsCipher(t *testing.T) {
+func TestTLSMinVersionConfig(t *testing.T) {
+	tmpl := `
+		listen: "127.0.0.1:-1"
+		tls {
+			cert_file: 	"../test/configs/certs/server-cert.pem"
+			key_file:  	"../test/configs/certs/server-key.pem"
+			timeout: 	1
+			min_version: 	%s
+		}
+	`
+	conf := createConfFile(t, []byte(fmt.Sprintf(tmpl, `"1.3"`)))
+	s, o := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	connect := func(t *testing.T, tlsConf *tls.Config, expectedErr error) {
+		t.Helper()
+		opts := []nats.Option{}
+		if tlsConf != nil {
+			opts = append(opts, nats.Secure(tlsConf))
+		}
+		opts = append(opts, nats.RootCAs("../test/configs/certs/ca.pem"))
+		nc, err := nats.Connect(fmt.Sprintf("tls://localhost:%d", o.Port), opts...)
+		if expectedErr == nil {
+			if err != nil {
+				t.Fatalf("Unexpected error: %v", err)
+			}
+		} else if err == nil || err.Error() != expectedErr.Error() {
+			nc.Close()
+			t.Fatalf("Expected error %v, got: %v", expectedErr, err)
+		}
+	}
+
+	// Cannot connect with client requiring a lower minimum TLS Version.
+	connect(t, &tls.Config{
+		MaxVersion: tls.VersionTLS12,
+	}, errors.New(`remote error: tls: protocol version not supported`))
+
+	// Should connect since matching minimum TLS version.
+	connect(t, &tls.Config{
+		MinVersion: tls.VersionTLS13,
+	}, nil)
+
+	// Reloading with invalid values should fail.
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(tmpl, `"1.0"`)), 0666); err != nil {
+		t.Fatalf("Error creating config file: %v", err)
+	}
+	if err := s.Reload(); err == nil {
+		t.Fatalf("Expected reload to fail: %v", err)
+	}
+
+	// Reloading with original values and no changes should be ok.
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(tmpl, `"1.3"`)), 0666); err != nil {
+		t.Fatalf("Error creating config file: %v", err)
+	}
+	if err := s.Reload(); err != nil {
+		t.Fatalf("Unexpected error reloading TLS version: %v", err)
+	}
+
+	// Reloading with a new minimum lower version.
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(tmpl, `"1.2"`)), 0666); err != nil {
+		t.Fatalf("Error creating config file: %v", err)
+	}
+	if err := s.Reload(); err != nil {
+		t.Fatalf("Unexpected error reloading: %v", err)
+	}
+
+	// Should connect since now matching minimum TLS version.
+	connect(t, &tls.Config{
+		MaxVersion: tls.VersionTLS12,
+	}, nil)
+	connect(t, &tls.Config{
+		MinVersion: tls.VersionTLS13,
+	}, nil)
+
+	// Setting unsupported TLS versions
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(tmpl, `"1.4"`)), 0666); err != nil {
+		t.Fatalf("Error creating config file: %v", err)
+	}
+	if err := s.Reload(); err == nil || !strings.Contains(err.Error(), `Unknown version: 1.4`) {
+		t.Fatalf("Unexpected error reloading: %v", err)
+	}
+
+	tc := &TLSConfigOpts{
+		CertFile:   "../test/configs/certs/server-cert.pem",
+		KeyFile:    "../test/configs/certs/server-key.pem",
+		CaFile:     "../test/configs/certs/ca.pem",
+		Timeout:    4.0,
+		MinVersion: tls.VersionTLS11,
+	}
+	_, err := GenTLSConfig(tc)
+	if err == nil || err.Error() != `unsupported minimum TLS version: TLS 1.1` {
+		t.Fatalf("Expected error generating TLS config: %v", err)
+	}
+}
+
+func TestTLSCipher(t *testing.T) {
 	if strings.Compare(tlsCipher(0x0005), "TLS_RSA_WITH_RC4_128_SHA") != 0 {
 		t.Fatalf("Invalid tls cipher")
 	}