Add support for re-encrypting streams with new key

server/filestore.go

@@ -165,6 +165,7 @@ type fileStore struct {
 	cfg         FileStreamInfo
 	fcfg        FileStoreConfig
 	prf         keyGen
+	oldprf      keyGen
 	aek         cipher.AEAD
 	lmb         *msgBlock
 	blks        []*msgBlock
@@ -332,10 +333,10 @@ const (
 )
 
 func newFileStore(fcfg FileStoreConfig, cfg StreamConfig) (*fileStore, error) {
-	return newFileStoreWithCreated(fcfg, cfg, time.Now().UTC(), nil)
+	return newFileStoreWithCreated(fcfg, cfg, time.Now().UTC(), nil, nil)
 }
 
-func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created time.Time, prf keyGen) (*fileStore, error) {
+func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created time.Time, prf, oldprf keyGen) (*fileStore, error) {
 	if cfg.Name == _EMPTY_ {
 		return nil, fmt.Errorf("name required")
 	}
@@ -375,12 +376,13 @@ func newFileStoreWithCreated(fcfg FileStoreConfig, cfg StreamConfig, created tim
 	dios <- struct{}{}
 
 	fs := &fileStore{
-		fcfg: fcfg,
-		psim: make(map[string]*psi),
-		bim:  make(map[uint32]*msgBlock),
-		cfg:  FileStreamInfo{Created: created, StreamConfig: cfg},
-		prf:  prf,
-		qch:  make(chan struct{}),
+		fcfg:   fcfg,
+		psim:   make(map[string]*psi),
+		bim:    make(map[uint32]*msgBlock),
+		cfg:    FileStreamInfo{Created: created, StreamConfig: cfg},
+		prf:    prf,
+		oldprf: oldprf,
+		qch:    make(chan struct{}),
 	}
 
 	// Set flush in place to AsyncFlush which by default is false.
@@ -948,52 +950,67 @@ func (mb *msgBlock) convertCipher() error {
 	if len(ekey) < minBlkKeySize {
 		return errBadKeySize
 	}
-	// Recover key encryption key.
-	rb, err := fs.prf([]byte(fmt.Sprintf("%s:%d", fs.cfg.Name, mb.index)))
-	if err != nil {
-		return err
+	type prfWithCipher struct {
+		keyGen
+		StoreCipher
 	}
-	kek, err := genEncryptionKey(osc, rb)
-	if err != nil {
-		return err
+	var prfs []prfWithCipher
+	if fs.prf != nil {
+		prfs = append(prfs, prfWithCipher{fs.prf, sc})
+		prfs = append(prfs, prfWithCipher{fs.prf, osc})
 	}
-	ns := kek.NonceSize()
-	seed, err := kek.Open(nil, ekey[:ns], ekey[ns:], nil)
-	if err != nil {
-		return err
+	if fs.oldprf != nil {
+		prfs = append(prfs, prfWithCipher{fs.oldprf, sc})
+		prfs = append(prfs, prfWithCipher{fs.oldprf, osc})
 	}
-	nonce := ekey[:ns]
 
-	bek, err := genBlockEncryptionKey(osc, seed, nonce)
-	if err != nil {
-		return err
-	}
+	for _, prf := range prfs {
+		// Recover key encryption key.
+		rb, err := prf.keyGen([]byte(fmt.Sprintf("%s:%d", fs.cfg.Name, mb.index)))
+		if err != nil {
+			continue
+		}
+		kek, err := genEncryptionKey(prf.StoreCipher, rb)
+		if err != nil {
+			continue
+		}
+		ns := kek.NonceSize()
+		seed, err := kek.Open(nil, ekey[:ns], ekey[ns:], nil)
+		if err != nil {
+			continue
+		}
+		nonce := ekey[:ns]
+		bek, err := genBlockEncryptionKey(prf.StoreCipher, seed, nonce)
+		if err != nil {
+			return err
+		}
 
-	buf, _ := mb.loadBlock(nil)
-	bek.XORKeyStream(buf, buf)
-	// Make sure we can parse with old cipher and key file.
-	if err = mb.indexCacheBuf(buf); err != nil {
-		return err
-	}
-	// Reset the cache since we just read everything in.
-	mb.cache = nil
+		buf, _ := mb.loadBlock(nil)
+		bek.XORKeyStream(buf, buf)
+		// Make sure we can parse with old cipher and key file.
+		if err = mb.indexCacheBuf(buf); err != nil {
+			return err
+		}
+		// Reset the cache since we just read everything in.
+		mb.cache = nil
 
-	// Generate new keys based on our
-	if err := fs.genEncryptionKeysForBlock(mb); err != nil {
-		// Put the old keyfile back.
-		keyFile := filepath.Join(mdir, fmt.Sprintf(keyScan, mb.index))
-		os.WriteFile(keyFile, ekey, defaultFilePerms)
-		return err
-	}
-	mb.bek.XORKeyStream(buf, buf)
-	if err := os.WriteFile(mb.mfn, buf, defaultFilePerms); err != nil {
-		return err
+		// Generate new keys. If we error for some reason then we will put
+		// the old keyfile back.
+		if err := fs.genEncryptionKeysForBlock(mb); err != nil {
+			keyFile := filepath.Join(mdir, fmt.Sprintf(keyScan, mb.index))
+			os.WriteFile(keyFile, ekey, defaultFilePerms)
+			return err
+		}
+		mb.bek.XORKeyStream(buf, buf)
+		if err := os.WriteFile(mb.mfn, buf, defaultFilePerms); err != nil {
+			return err
+		}
+		// If we are here we want to delete other meta, e.g. idx, fss.
+		os.Remove(mb.ifn)
+		os.Remove(mb.sfn)
+		return nil
 	}
-	// If we are here we want to delete other meta, e.g. idx, fss.
-	os.Remove(mb.ifn)
-	os.Remove(mb.sfn)
-
-	return nil
+	return fmt.Errorf("unable to recover keys")
 }
 
 // Convert a plaintext block to encrypted.

server/filestore_test.go

@@ -820,7 +820,7 @@ func TestFileStoreCompact(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
@@ -865,7 +865,7 @@ func TestFileStoreCompact(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
@@ -1019,7 +1019,7 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Subjects: []string{"*"}, Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
@@ -1083,7 +1083,7 @@ func TestFileStoreStreamTruncate(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Subjects: []string{"foo"}, Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
@@ -3489,7 +3489,7 @@ func TestFileStoreSparseCompaction(t *testing.T) {
 			prf = nil
 		}
 
-		fs, err = newFileStoreWithCreated(fcfg, cfg, time.Now(), prf)
+		fs, err = newFileStoreWithCreated(fcfg, cfg, time.Now(), prf, nil)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -3814,7 +3814,7 @@ func TestFileStoreCompactReclaimHeadSpace(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "TEST", Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -3844,7 +3844,7 @@ func TestFileStoreCompactReclaimHeadSpace(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "TEST", Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -4072,7 +4072,7 @@ func TestFileStoreShortIndexWriteBug(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "TEST", Storage: FileStorage, MaxAge: time.Second},
 			created,
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -4108,7 +4108,7 @@ func TestFileStoreShortIndexWriteBug(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "TEST", Storage: FileStorage, MaxAge: time.Second},
 			created,
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -4136,7 +4136,7 @@ func TestFileStoreDoubleCompactWithWriteInBetweenEncryptedBug(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -4188,7 +4188,7 @@ func TestFileStoreEncryptedKeepIndexNeedBekResetBug(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Storage: FileStorage, MaxAge: ttl},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -4478,7 +4478,7 @@ func TestFileStoreEncrypted(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -4505,7 +4505,7 @@ func TestFileStoreEncrypted(t *testing.T) {
 			fcfg,
 			StreamConfig{Name: "zzz", Storage: FileStorage},
 			time.Now(),
-			prf,
+			prf, nil,
 		)
 		require_NoError(t, err)
 		defer fs.Stop()
@@ -5622,7 +5622,7 @@ func TestFileStoreRestoreEncryptedWithNoKeyFuncFails(t *testing.T) {
 	fs, err := newFileStoreWithCreated(
 		fcfg, scfg,
 		time.Now(),
-		prf,
+		prf, nil,
 	)
 	require_NoError(t, err)
 
@@ -5638,7 +5638,7 @@ func TestFileStoreRestoreEncryptedWithNoKeyFuncFails(t *testing.T) {
 	_, err = newFileStoreWithCreated(
 		fcfg, scfg,
 		time.Now(),
-		nil,
+		nil, nil,
 	)
 	require_Error(t, err, errNoMainKey)
 }