Add type check before metadata decode

src/storage/redis_metadata.cc

@@ -372,29 +372,31 @@ rocksdb::Status StreamMetadata::Decode(Slice *input) {
     return s;
   }
 
-  if (input->size() < 8 * 11) {
-    return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
-  }
+  if (Type() == kRedisStream) {
+    if (input->size() < 8 * 11) {
+      return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+    }
 
-  GetFixed64(input, &last_generated_id.ms);
-  GetFixed64(input, &last_generated_id.seq);
+    GetFixed64(input, &last_generated_id.ms);
+    GetFixed64(input, &last_generated_id.seq);
 
-  GetFixed64(input, &recorded_first_entry_id.ms);
-  GetFixed64(input, &recorded_first_entry_id.seq);
+    GetFixed64(input, &recorded_first_entry_id.ms);
+    GetFixed64(input, &recorded_first_entry_id.seq);
 
-  GetFixed64(input, &max_deleted_entry_id.ms);
-  GetFixed64(input, &max_deleted_entry_id.seq);
+    GetFixed64(input, &max_deleted_entry_id.ms);
+    GetFixed64(input, &max_deleted_entry_id.seq);
 
-  GetFixed64(input, &first_entry_id.ms);
-  GetFixed64(input, &first_entry_id.seq);
+    GetFixed64(input, &first_entry_id.ms);
+    GetFixed64(input, &first_entry_id.seq);
 
-  GetFixed64(input, &last_entry_id.ms);
-  GetFixed64(input, &last_entry_id.seq);
+    GetFixed64(input, &last_entry_id.ms);
+    GetFixed64(input, &last_entry_id.seq);
 
-  GetFixed64(input, &entries_added);
+    GetFixed64(input, &entries_added);
 
-  if (input->size() >= 8) {
-    GetFixed64(input, &group_number);
+    if (input->size() >= 8) {
+      GetFixed64(input, &group_number);
+    }
   }
 
   return rocksdb::Status::OK();
@@ -416,16 +418,18 @@ rocksdb::Status BloomChainMetadata::Decode(Slice *input) {
     return s;
   }
 
-  if (input->size() < 20) {
-    return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
-  }
+  if (Type() == kRedisBloomFilter) {
+    if (input->size() < 20) {
+      return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+    }
 
-  GetFixed16(input, &n_filters);
-  GetFixed16(input, &expansion);
+    GetFixed16(input, &n_filters);
+    GetFixed16(input, &expansion);
 
-  GetFixed32(input, &base_capacity);
-  GetDouble(input, &error_rate);
-  GetFixed32(input, &bloom_bytes);
+    GetFixed32(input, &base_capacity);
+    GetDouble(input, &error_rate);
+    GetFixed32(input, &bloom_bytes);
+  }
 
   return rocksdb::Status::OK();
 }
@@ -454,7 +458,7 @@ rocksdb::Status JsonMetadata::Decode(Slice *input) {
     return s;
   }
 
-  if (!GetFixed8(input, reinterpret_cast<uint8_t *>(&format))) {
+  if (Type() == kRedisJson && !GetFixed8(input, reinterpret_cast<uint8_t *>(&format))) {
     return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
   }
 

tests/gocase/unit/type/types_test.go

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package types
+
+import (
+	"context"
+	"testing"
+
+	"github.com/apache/kvrocks/tests/gocase/util"
+	"github.com/stretchr/testify/require"
+)
+
+func TestTypesError(t *testing.T) {
+	srv := util.StartServer(t, map[string]string{})
+	defer srv.Close()
+	ctx := context.Background()
+	rdb := srv.NewClient()
+	defer func() { require.NoError(t, rdb.Close()) }()
+
+	t.Run("Operate with wrong type", func(t *testing.T) {
+		message := "ERR Invalid argument: WRONGTYPE Operation against a key holding the wrong kind of value"
+		require.NoError(t, rdb.Set(ctx, "a", "hello", 0).Err())
+		require.EqualError(t, rdb.Do(ctx, "XADD", "a", "*", "a", "test").Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "LPUSH", "a", 1).Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "HSET", "a", "1", "2").Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "SADD", "a", "1", "2").Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "ZADD", "a", "1", "2").Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "JSON.SET", "a", "$", "{}").Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "BF.ADD", "a", "test").Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "SADD", "a", 100).Err(), message)
+
+		require.NoError(t, rdb.LPush(ctx, "a1", "hello", 0).Err())
+		require.EqualError(t, rdb.Do(ctx, "SETBIT", "a1", 1, 1).Err(), message)
+		require.EqualError(t, rdb.Do(ctx, "GET", "a1").Err(), message)
+
+	})
+}