[studio][ISSUE #10901] Harden Proxy metrics header parsing

proxy/src/main/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManager.java

@@ -34,6 +34,7 @@
 import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.resources.Resource;
 import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -146,15 +147,7 @@ public void start() throws Exception {
 
         String labels = proxyConfig.getMetricsLabel();
         if (StringUtils.isNotBlank(labels)) {
-            List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(labels);
-            for (String item : kvPairs) {
-                String[] split = item.split(":");
-                if (split.length != 2) {
-                    log.warn("metricsLabel is not valid: {}", labels);
-                    continue;
-                }
-                LABEL_MAP.put(split[0], split[1]);
-            }
+            LABEL_MAP.putAll(parseKeyValuePairs("metricsLabel", labels));
         }
         if (proxyConfig.isMetricsInDelta()) {
             LABEL_MAP.put(LABEL_AGGREGATION, AGGREGATION_DELTA);
@@ -185,16 +178,7 @@ public void start() throws Exception {
 
             String headers = proxyConfig.getMetricsGrpcExporterHeader();
             if (StringUtils.isNotBlank(headers)) {
-                Map<String, String> headerMap = new HashMap<>();
-                List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(headers);
-                for (String item : kvPairs) {
-                    String[] split = item.split(":");
-                    if (split.length != 2) {
-                        log.warn("metricsGrpcExporterHeader is not valid: {}", headers);
-                        continue;
-                    }
-                    headerMap.put(split[0], split[1]);
-                }
+                Map<String, String> headerMap = parseKeyValuePairs("metricsGrpcExporterHeader", headers);
                 headerMap.forEach(metricExporterBuilder::addHeader);
             }
 
@@ -238,6 +222,23 @@ public void start() throws Exception {
         initMetrics(proxyMeter, null);
     }
 
+    static Map<String, String> parseKeyValuePairs(String configName, String config) {
+        Map<String, String> result = new LinkedHashMap<>();
+        List<String> kvPairs = Splitter.on(',').omitEmptyStrings().splitToList(config);
+        for (int i = 0; i < kvPairs.size(); i++) {
+            String item = kvPairs.get(i);
+            int separatorIndex = item.indexOf(':');
+            if (separatorIndex < 0 || StringUtils.isBlank(item.substring(0, separatorIndex))) {
+                log.warn("{} contains an invalid entry at position {}", configName, i);
+                continue;
+            }
+            String key = item.substring(0, separatorIndex).trim();
+            String value = item.substring(separatorIndex + 1).trim();
+            result.put(key, value);
+        }
+        return result;
+    }
+
     @Override
     public void shutdown() throws Exception {
         if (proxyConfig.getMetricsExporterType() == MetricsExporterType.OTLP_GRPC) {

proxy/src/test/java/org/apache/rocketmq/proxy/metrics/ProxyMetricsManagerTest.java

@@ -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 org.apache.rocketmq.proxy.metrics;
+
+import java.util.Map;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.entry;
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class ProxyMetricsManagerTest {
+
+    @Test
+    public void parseKeyValuePairsShouldPreserveColonsInValues() {
+        Map<String, String> result = ProxyMetricsManager.parseKeyValuePairs(
+            "metricsGrpcExporterHeader", "Authorization:Bearer token:part,endpoint:https://collector:4317");
+
+        assertThat(result)
+            .containsEntry("Authorization", "Bearer token:part")
+            .containsEntry("endpoint", "https://collector:4317");
+    }
+
+    @Test
+    public void parseKeyValuePairsShouldTrimEntriesAndAllowEmptyValues() {
+        Map<String, String> result = ProxyMetricsManager.parseKeyValuePairs(
+            "metricsLabel", " cluster : production ,optional:");
+
+        assertThat(result)
+            .containsEntry("cluster", "production")
+            .containsEntry("optional", "");
+    }
+
+    @Test
+    public void parseKeyValuePairsShouldSkipMalformedEntriesAndEmptyKeys() {
+        Map<String, String> result = ProxyMetricsManager.parseKeyValuePairs(
+            "metricsGrpcExporterHeader", "missing-separator, :secret,valid:value");
+
+        assertThat(result).containsExactly(entry("valid", "value"));
+    }
+}