[Bugfix][Store] Reject invalid metadata client IDs

mooncake-store/src/master_service.cpp

@@ -7852,7 +7852,12 @@ MasterService::MetadataSerializer::DeserializeMetadata(
     // Deserialize client_id string
     std::string client_id_str = array[index++].as<std::string>();
     UUID client_id;
-    StringToUuid(client_id_str, client_id);
+    if (!StringToUuid(client_id_str, client_id)) {
+        return tl::unexpected(SerializationError(
+            ErrorCode::DESERIALIZE_FAIL,
+            fmt::format("deserialize ObjectMetadata invalid client_id UUID: {}",
+                        client_id_str)));
+    }
 
     // Deserialize put_start_time
     uint64_t put_start_time_timestamp = array[index++].as<uint64_t>();

mooncake-store/tests/ha/snapshot/snapshot_child_process_test.cpp

@@ -666,6 +666,57 @@ TEST_F(SnapshotChildProcessTest,
     EXPECT_FALSE(ObjectIsGroupedInMetadata(key, shard_idx));
 }
 
+TEST_F(SnapshotChildProcessTest, DeserializeMetadataSkipsInvalidClientId) {
+    CreateDefaultService();
+    const std::string key = "invalid_client_id_snapshot_key";
+    const uint32_t shard_idx = GetShardIndexForTest(key);
+
+    msgpack::sbuffer shard_buffer;
+    MsgpackPacker shard_packer(&shard_buffer);
+    shard_packer.pack_map(1);
+    shard_packer.pack(std::string("metadata"));
+    shard_packer.pack_array(1);
+    shard_packer.pack_array(2);
+    shard_packer.pack(key);
+
+    shard_packer.pack_array(8);
+    shard_packer.pack(std::string("not-a-uuid"));
+    shard_packer.pack(kDefaultTestPutStartTimeMs);
+    shard_packer.pack(kDefaultTestObjectSize);
+    shard_packer.pack(kDefaultTestLeaseTimeoutMs);
+    shard_packer.pack(false);
+    shard_packer.pack(uint64_t{0});
+    shard_packer.pack(uint32_t{1});
+    PackDiskReplica(shard_packer, kDefaultTestDiskFilePath,
+                    kDefaultTestObjectSize);
+
+    auto compressed_shard =
+        zstd_compress(reinterpret_cast<const uint8_t*>(shard_buffer.data()),
+                      shard_buffer.size(), 3);
+
+    msgpack::sbuffer root_buffer;
+    MsgpackPacker root_packer(&root_buffer);
+    root_packer.pack_map(3);
+    root_packer.pack(std::string("shards"));
+    root_packer.pack_map(1);
+    root_packer.pack(shard_idx);
+    root_packer.pack_bin(compressed_shard.size());
+    root_packer.pack_bin_body(
+        reinterpret_cast<const char*>(compressed_shard.data()),
+        compressed_shard.size());
+    root_packer.pack(std::string("discarded_replicas"));
+    root_packer.pack_array(0);
+    root_packer.pack(std::string("replica_next_id"));
+    root_packer.pack(uint64_t{10});
+
+    auto deserialize_result =
+        DeserializeMetadataForTest(ToByteVector(root_buffer));
+    ASSERT_TRUE(deserialize_result.has_value())
+        << deserialize_result.error().message;
+
+    EXPECT_FALSE(KeyExistsInMetadata(service_.get(), key));
+}
+
 TEST_F(SnapshotChildProcessTest, LegacyEtcdConnstringFallbackIsPreserved) {
     MasterConfig legacy_config;
     legacy_config.enable_ha = true;

mooncake-store/tests/ha/snapshot/snapshot_test_utils.h

@@ -167,8 +167,9 @@ inline std::vector<uint8_t> WrapShardIntoMetadataRoot(
     return ToByteVector(root_buffer);
 }
 
-inline std::vector<uint8_t> BuildMetadataPayload(
-    const UUID& client_id, std::string_view object_key = kDefaultTestObjectKey,
+inline std::vector<uint8_t> BuildMetadataPayloadWithClientIdString(
+    std::string_view client_id,
+    std::string_view object_key = kDefaultTestObjectKey,
     std::string_view disk_file_path = kDefaultTestDiskFilePath,
     uint64_t object_size = kDefaultTestObjectSize,
     uint64_t put_start_time_ms = kDefaultTestPutStartTimeMs,
@@ -199,7 +200,7 @@ inline std::vector<uint8_t> BuildMetadataPayload(
     shard_packer.pack(std::string(object_key));
 
     shard_packer.pack_array(array_size);
-    shard_packer.pack(UuidToString(client_id));
+    shard_packer.pack(std::string(client_id));
     shard_packer.pack(put_start_time_ms);
     shard_packer.pack(object_size);
     shard_packer.pack(lease_timeout_ms);
@@ -220,6 +221,18 @@ inline std::vector<uint8_t> BuildMetadataPayload(
     return WrapShardIntoMetadataRoot(shard_buffer);
 }
 
+inline std::vector<uint8_t> BuildMetadataPayload(
+    const UUID& client_id, std::string_view object_key = kDefaultTestObjectKey,
+    std::string_view disk_file_path = kDefaultTestDiskFilePath,
+    uint64_t object_size = kDefaultTestObjectSize,
+    uint64_t put_start_time_ms = kDefaultTestPutStartTimeMs,
+    uint64_t lease_timeout_ms = kDefaultTestLeaseTimeoutMs,
+    SnapshotMetadataFormat format = SnapshotMetadataFormat::kLegacy) {
+    return BuildMetadataPayloadWithClientIdString(
+        UuidToString(client_id), object_key, disk_file_path, object_size,
+        put_start_time_ms, lease_timeout_ms, format);
+}
+
 // Builds a metadata payload whose declared replica_count field is set to
 // `declared_replica_count` while no replicas are actually packed (the entry
 // array stays at the 7 leading fields). Used to verify the deserializer