Repository: nats-io/nats-server
Pull request #3314 (2022-07-31 15:53:41)

[FIXED] Instability with encrypted systems
Fixed and improved several system instability issues with encrypted state that were a result of block counters being incorrect for message blocks either due to compaction or when messages expired and we would keep markers for sequence state.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/.github/workflows/nightly.yaml b/.github/workflows/nightly.yaml
index d19efbbe..9f17332c 100644
--- a/.github/workflows/nightly.yaml
+++ b/.github/workflows/nightly.yaml
@@ -23,7 +23,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v2
         with:
-          go-version: 1.18
+          go-version: 1.17
 
       - name: goreleaser
         uses: goreleaser/goreleaser-action@v2
diff --git a/docker/Dockerfile.nightly b/docker/Dockerfile.nightly
index ba08ba8f..21521d8d 100644
--- a/docker/Dockerfile.nightly
+++ b/docker/Dockerfile.nightly
@@ -1,4 +1,4 @@
-FROM golang:1.18-alpine AS builder
+FROM golang:1.17-alpine AS builder
 
 ARG VERSION="nightly"
 
@@ -9,7 +9,7 @@ RUN mkdir -p src/github.com/nats-io && \
     cd natscli/nats && \
     go build -ldflags "-w -X main.version=${VERSION}" -o /nats
 
-RUN go install github.com/nats-io/nsc@latest
+RUN go get github.com/nats-io/nsc
 
 FROM alpine:latest
 
diff --git a/server/filestore.go b/server/filestore.go
index 3c83a584..d469b32c 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -652,11 +652,7 @@ func (fs *fileStore) recoverMsgBlock(fi os.FileInfo, index uint64) (*msgBlock, e
 		}
 		// Undo cache from above for later.
 		mb.cache = nil
-		wbek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce)
-		if err != nil {
-			return nil, err
-		}
-		wbek.XORKeyStream(buf, buf)
+		mb.bek.XORKeyStream(buf, buf)
 		if err := ioutil.WriteFile(mb.mfn, buf, defaultFilePerms); err != nil {
 			return nil, err
 		}
@@ -3532,24 +3528,22 @@ checkCache:
 
 	// Check if we need to decrypt.
 	if mb.bek != nil && len(buf) > 0 {
-		rbek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce)
+		bek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce)
 		if err != nil {
 			return err
 		}
-		rbek.XORKeyStream(buf, buf)
+		mb.bek = bek
+		mb.bek.XORKeyStream(buf, buf)
 	}
 
 	if err := mb.indexCacheBuf(buf); err != nil {
 		if err == errCorruptState {
-			fs := mb.fs
-			mb.mu.Unlock()
 			var ld *LostStreamData
-			if ld, err = mb.rebuildState(); ld != nil {
+			if ld, err = mb.rebuildStateLocked(); ld != nil {
 				// We do not know if fs is locked or not at this point.
 				// This should be an exceptional condition so do so in Go routine.
-				go fs.rebuildState(ld)
+				go mb.fs.rebuildState(ld)
 			}
-			mb.mu.Lock()
 		}
 		if err != nil {
 			return err
@@ -4545,16 +4539,15 @@ func (fs *fileStore) Compact(seq uint64) (uint64, error) {
 			// Check for encryption.
 			if smb.bek != nil && len(nbuf) > 0 {
 				// Recreate to reset counter.
-				rbek, err := chacha20.NewUnauthenticatedCipher(smb.seed, smb.nonce)
+				bek, err := chacha20.NewUnauthenticatedCipher(smb.seed, smb.nonce)
 				if err != nil {
 					goto SKIP
 				}
-				cbuf := make([]byte, len(nbuf))
-				rbek.XORKeyStream(cbuf, nbuf)
-				if err = ioutil.WriteFile(smb.mfn, cbuf, defaultFilePerms); err != nil {
-					goto SKIP
-				}
-			} else if err = ioutil.WriteFile(smb.mfn, nbuf, defaultFilePerms); err != nil {
+				// For future writes make sure to set smb.bek to keep counter correct.
+				smb.bek = bek
+				smb.bek.XORKeyStream(nbuf, nbuf)
+			}
+			if err = ioutil.WriteFile(smb.mfn, nbuf, defaultFilePerms); err != nil {
 				goto SKIP
 			}
 			smb.clearCacheAndOffset()
@@ -4730,6 +4723,13 @@ func (mb *msgBlock) closeAndKeepIndex() {
 	if mb.sfn != _EMPTY_ {
 		os.Remove(mb.sfn)
 	}
+
+	// If we are encrypted we should reset our bek counter.
+	if mb.bek != nil {
+		if bek, err := chacha20.NewUnauthenticatedCipher(mb.seed, mb.nonce); err == nil {
+			mb.bek = bek
+		}
+	}
 }
 
 // Called by purge to simply get rid of the cache and close our fds.


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 185be8eb..43b1a7fd 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -4025,3 +4025,104 @@ func TestFileStoreShortIndexWriteBug(t *testing.T) {
 		t.Fatalf("Expected first sequence of 101 vs %d", state.FirstSeq)
 	}
 }
+
+func TestFileStoreDoubleCompactWithWriteInBetweenEncryptedBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer os.RemoveAll(storeDir)
+
+	prf := func(context []byte) ([]byte, error) {
+		h := hmac.New(sha256.New, []byte("dlc22"))
+		if _, err := h.Write(context); err != nil {
+			return nil, err
+		}
+		return h.Sum(nil), nil
+	}
+
+	fs, err := newFileStoreWithCreated(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Storage: FileStorage},
+		time.Now(),
+		prf,
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("ouch")
+	for i := 0; i < 5; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+	_, err = fs.Compact(5)
+	require_NoError(t, err)
+
+	if state := fs.State(); state.LastSeq != 5 {
+		t.Fatalf("Expected last sequence to be 5 but got %d", state.LastSeq)
+	}
+	for i := 0; i < 5; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+	_, err = fs.Compact(10)
+	require_NoError(t, err)
+
+	if state := fs.State(); state.LastSeq != 10 {
+		t.Fatalf("Expected last sequence to be 10 but got %d", state.LastSeq)
+	}
+}
+
+// When we kept the empty block for tracking sequence, we needed to reset the bek
+// counter when encrypted for subsequent writes to be correct. The bek in place could
+// possibly still have a non-zero counter from previous writes.
+// Happens when all messages expire and the are flushed and then subsequent writes occur.
+func TestFileStoreEncryptedKeepIndexNeedBekResetBug(t *testing.T) {
+	storeDir := createDir(t, JetStreamStoreDir)
+	defer os.RemoveAll(storeDir)
+
+	prf := func(context []byte) ([]byte, error) {
+		h := hmac.New(sha256.New, []byte("dlc22"))
+		if _, err := h.Write(context); err != nil {
+			return nil, err
+		}
+		return h.Sum(nil), nil
+	}
+
+	ttl := 250 * time.Millisecond
+	fs, err := newFileStoreWithCreated(
+		FileStoreConfig{StoreDir: storeDir},
+		StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: ttl},
+		time.Now(),
+		prf,
+	)
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	subj, msg := "foo", []byte("ouch")
+	for i := 0; i < 5; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+
+	// What to go to 0.
+	// This will leave the marker.
+	checkFor(t, time.Second, ttl, func() error {
+		if state := fs.State(); state.Msgs != 0 {
+			return fmt.Errorf("Expected no msgs, got %d", state.Msgs)
+		}
+		return nil
+	})
+
+	// Now write additional messages.
+	for i := 0; i < 5; i++ {
+		fs.StoreMsg(subj, nil, msg)
+	}
+
+	// Make sure the buffer is cleared.
+	fs.mu.RLock()
+	mb := fs.lmb
+	fs.mu.RUnlock()
+	mb.mu.Lock()
+	mb.clearCacheAndOffset()
+	mb.mu.Unlock()
+
+	// Now make sure we can read.
+	var smv StoreMsg
+	_, err = fs.LoadMsg(10, &smv)
+	require_NoError(t, err)
+}
diff --git a/server/jetstream_cluster_test.go b/server/jetstream_cluster_test.go
index 3c25de39..bd99bf09 100644
--- a/server/jetstream_cluster_test.go
+++ b/server/jetstream_cluster_test.go
@@ -11662,3 +11662,49 @@ func TestJetStreamClusterPullConsumerMaxWaiting(t *testing.T) {
 		t.Fatalf(`expected "cannot be updated" error, got %s`, err)
 	}
 }
+
+func TestJetStreamClusterEncryptedDoubleSnapshotBug(t *testing.T) {
+	c := createJetStreamClusterWithTemplate(t, jsClusterEncryptedTempl, "JSC", 3)
+	defer c.shutdown()
+
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	_, err := js.AddStream(&nats.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		MaxAge:   time.Second,
+		Replicas: 3,
+	})
+	require_NoError(t, err)
+
+	numMsgs := 50
+	for i := 0; i < numMsgs; i++ {
+		js.PublishAsync("foo", []byte("SNAP"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	// Perform a snapshot on a follower.
+	nl := c.randomNonStreamLeader("$G", "TEST")
+	mset, err := nl.GlobalAccount().lookupStream("TEST")
+	require_NoError(t, err)
+	err = mset.raftNode().InstallSnapshot(mset.stateSnapshot())
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", []byte("SNAP2"))
+	require_NoError(t, err)
+
+	for _, seq := range []uint64{1, 11, 22, 51} {
+		js.DeleteMsg("TEST", seq)
+	}
+
+	err = mset.raftNode().InstallSnapshot(mset.stateSnapshot())
+	require_NoError(t, err)
+
+	_, err = js.Publish("foo", []byte("SNAP3"))
+	require_NoError(t, err)
+}
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index ba8356c3..325aa698 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -130,6 +130,25 @@ var jsClusterTempl = `
 	accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
 `
 
+var jsClusterEncryptedTempl = `
+	listen: 127.0.0.1:-1
+	server_name: %s
+	jetstream: {max_mem_store: 256MB, max_file_store: 2GB, store_dir: '%s', key: "s3cr3t!"}
+
+	leaf {
+		listen: 127.0.0.1:-1
+	}
+
+	cluster {
+		name: %s
+		listen: 127.0.0.1:%d
+		routes = [%s]
+	}
+
+	# For access to system account.
+	accounts { $SYS { users = [ { user: "admin", pass: "s3cr3t!" } ] } }
+`
+
 var jsClusterMaxBytesTempl = `
 	listen: 127.0.0.1:-1
 	server_name: %s