[improvement](fe) Add connector property validators

fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/HMSBaseProperties.java

@@ -24,6 +24,7 @@
 import org.apache.doris.common.security.authentication.KerberosAuthenticationConfig;
 import org.apache.doris.foundation.property.ConnectorPropertiesUtils;
 import org.apache.doris.foundation.property.ConnectorProperty;
+import org.apache.doris.foundation.property.NoPathTraversalValidator;
 import org.apache.doris.foundation.property.ParamRules;
 
 import com.google.common.base.Strings;
@@ -57,6 +58,7 @@ public class HMSBaseProperties {
 
     @ConnectorProperty(names = {"hive.conf.resources"},
             required = false,
+            validator = NoPathTraversalValidator.class,
             description = "The conf resources of the hive metastore.")
     private String hiveConfResourcesConfig = "";
 

fe/fe-filesystem/fe-filesystem-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/properties/HdfsProperties.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.filesystem.FileSystemType;
 import org.apache.doris.foundation.property.ConnectorProperty;
+import org.apache.doris.foundation.property.NoPathTraversalValidator;
 
 import com.google.common.collect.ImmutableSet;
 import org.apache.commons.collections4.MapUtils;
@@ -45,6 +46,7 @@ public class HdfsProperties extends HdfsCompatibleProperties {
 
     @ConnectorProperty(names = {"hadoop.config.resources"},
             required = false,
+            validator = NoPathTraversalValidator.class,
             description = "The xml files of Hadoop configuration.")
     private String hadoopConfigResources = "";
 

fe/fe-filesystem/fe-filesystem-hdfs/src/test/java/org/apache/doris/filesystem/hdfs/properties/HdfsPropertiesTest.java

@@ -111,6 +111,20 @@ void xmlResourcesAreLoadedFromAbsolutePathWhenNoConfigDirInjected(@TempDir Path
         Assertions.assertEquals("v2", resolved.get("dfs.custom.abs.key"));
     }
 
+    @Test
+    void xmlResourcesRejectParentDirectoryTraversal(@TempDir Path tmp) {
+        Map<String, String> raw = new HashMap<>();
+        raw.put("fs.defaultFS", "hdfs://ns");
+        raw.put("hadoop.config.resources", "../outside.xml");
+        raw.put("_HADOOP_CONFIG_DIR_", tmp.toString() + "/");
+
+        IllegalArgumentException exception = Assertions.assertThrows(
+                IllegalArgumentException.class, () -> resolve(raw));
+
+        Assertions.assertTrue(exception.getMessage().contains("parent-directory references"),
+                exception.getMessage());
+    }
+
     @Test
     void translationIsIdempotentOnAlreadyResolvedMap() {
         Map<String, String> raw = new HashMap<>();

fe/fe-filesystem/fe-filesystem-jfs/src/main/java/org/apache/doris/filesystem/jfs/properties/JfsProperties.java

@@ -21,6 +21,7 @@
 import org.apache.doris.filesystem.hdfs.properties.HdfsCompatibleProperties;
 import org.apache.doris.filesystem.hdfs.properties.HdfsPropertiesUtils;
 import org.apache.doris.foundation.property.ConnectorProperty;
+import org.apache.doris.foundation.property.NoPathTraversalValidator;
 
 import com.google.common.collect.ImmutableSet;
 import org.apache.commons.collections4.MapUtils;
@@ -45,6 +46,7 @@ public class JfsProperties extends HdfsCompatibleProperties {
     private String fsDefaultFS = "";
 
     @ConnectorProperty(names = {"hadoop.config.resources"}, required = false,
+            validator = NoPathTraversalValidator.class,
             description = "The xml files of Hadoop configuration.")
     private String hadoopConfigResources = "";
 

fe/fe-filesystem/fe-filesystem-oss-hdfs/src/main/java/org/apache/doris/filesystem/hdfs/properties/OssHdfsProperties.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.filesystem.FileSystemType;
 import org.apache.doris.foundation.property.ConnectorProperty;
+import org.apache.doris.foundation.property.NoPathTraversalValidator;
 
 import com.google.common.collect.ImmutableSet;
 import org.apache.commons.lang3.StringUtils;
@@ -94,6 +95,7 @@ public class OssHdfsProperties extends HdfsCompatibleProperties {
 
     @ConnectorProperty(names = {"oss.hdfs.hadoop.config.resources"},
             required = false,
+            validator = NoPathTraversalValidator.class,
             description = "The xml files of Hadoop configuration.")
     private String hadoopConfigResources = "";
 

fe/fe-foundation/src/main/java/org/apache/doris/foundation/property/ConnectorPropertiesUtils.java

@@ -23,12 +23,15 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Utility class for handling fields annotated with {@link ConnectorProperty}.
  * Provides methods to extract supported connector properties from a class and bind them to an object instance.
  */
 public class ConnectorPropertiesUtils {
+    private static final Map<Class<? extends ConnectorPropertyValidator>, ConnectorPropertyValidator> VALIDATORS =
+            new ConcurrentHashMap<>();
 
     /**
      * Retrieves all fields annotated with {@link ConnectorProperty} from the given class and its superclasses,
@@ -74,6 +77,7 @@ public static void bindConnectorProperties(Object target, Map<String, String> pr
                 try {
                     Object rawValue = props.get(matchedName);
                     Object convertedValue = convertValue(rawValue, field.getType());
+                    validateValue(target, field, matchedName, convertedValue, props);
                     field.set(target, convertedValue);
                 } catch (Exception e) {
                     throw new IllegalArgumentException(
@@ -85,6 +89,25 @@ public static void bindConnectorProperties(Object target, Map<String, String> pr
         }
     }
 
+    private static void validateValue(Object target, Field field, String propertyName, Object value,
+            Map<String, String> props) {
+        Class<? extends ConnectorPropertyValidator> validatorClass =
+                field.getAnnotation(ConnectorProperty.class).validator();
+        ConnectorPropertyValidator validator = VALIDATORS.computeIfAbsent(
+                validatorClass, ConnectorPropertiesUtils::createValidator);
+        validator.validate(target, field, propertyName, value, props);
+    }
+
+    private static ConnectorPropertyValidator createValidator(
+            Class<? extends ConnectorPropertyValidator> validatorClass) {
+        try {
+            return validatorClass.getDeclaredConstructor().newInstance();
+        } catch (ReflectiveOperationException e) {
+            throw new IllegalArgumentException(
+                    "Failed to create connector property validator " + validatorClass.getName(), e);
+        }
+    }
+
     /**
      * Finds the first matching property name from the field's {@code @ConnectorProperty#names()} list
      * that exists in the provided property map.

fe/fe-foundation/src/main/java/org/apache/doris/foundation/property/ConnectorProperty.java

@@ -31,4 +31,6 @@
     boolean sensitive() default false;
 
     boolean isRegionField() default false;
+
+    Class<? extends ConnectorPropertyValidator> validator() default ConnectorPropertyValidator.None.class;
 }

fe/fe-foundation/src/main/java/org/apache/doris/foundation/property/ConnectorPropertyValidator.java

@@ -0,0 +1,50 @@
+// 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.doris.foundation.property;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
+/**
+ * Validates a converted connector property before it is assigned to its target field.
+ *
+ * <p>Implementations must be stateless because one validator instance may be reused across
+ * property bindings.
+ */
+public interface ConnectorPropertyValidator {
+
+    /**
+     * Validates one connector property value.
+     *
+     * @param target target object being populated
+     * @param field annotated target field
+     * @param propertyName matched property name
+     * @param value converted property value
+     * @param properties complete raw property map
+     * @throws IllegalArgumentException if the value is invalid
+     */
+    void validate(Object target, Field field, String propertyName, Object value, Map<String, String> properties);
+
+    /** Default validator used by properties that do not declare validation rules. */
+    final class None implements ConnectorPropertyValidator {
+        @Override
+        public void validate(Object target, Field field, String propertyName, Object value,
+                Map<String, String> properties) {
+        }
+    }
+}

fe/fe-foundation/src/main/java/org/apache/doris/foundation/property/NoPathTraversalValidator.java

@@ -0,0 +1,45 @@
+// 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.doris.foundation.property;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+
+/**
+ * Rejects parent-directory path components in a string property.
+ *
+ * <p>Both slash forms are treated as separators so a value cannot become unsafe when it is
+ * forwarded to a platform or library with different path-separator semantics. Comma-separated
+ * resource lists are supported naturally because commas cannot hide a path component.
+ */
+public final class NoPathTraversalValidator implements ConnectorPropertyValidator {
+
+    @Override
+    public void validate(Object target, Field field, String propertyName, Object value,
+            Map<String, String> properties) {
+        if (!(value instanceof String)) {
+            throw new IllegalArgumentException("Path property must be a string");
+        }
+        for (String component : ((String) value).split("[,/\\\\]")) {
+            if ("..".equals(component.trim())) {
+                throw new IllegalArgumentException(
+                        "Property " + propertyName + " must not contain parent-directory references");
+            }
+        }
+    }
+}

fe/fe-foundation/src/test/java/org/apache/doris/foundation/property/ConnectorPropertiesUtilsTest.java

@@ -104,6 +104,30 @@ void testPartialBindingWithTypeErrorThrowsException() {
                 () -> ConnectorPropertiesUtils.bindConnectorProperties(config, props));
     }
 
+    @Test
+    void testPropertyValidatorRejectsInvalidValue() {
+        Map<String, String> props = new HashMap<>();
+        props.put("path.key", "core-site.xml, ..\\secret.xml");
+
+        SampleConfig config = new SampleConfig();
+        IllegalArgumentException ex = Assertions.assertThrows(IllegalArgumentException.class,
+                () -> ConnectorPropertiesUtils.bindConnectorProperties(config, props));
+
+        Assertions.assertTrue(ex.getMessage().contains("parent-directory references"), ex.getMessage());
+        Assertions.assertNull(config.getPathValue());
+    }
+
+    @Test
+    void testPropertyValidatorAcceptsSafeValue() {
+        Map<String, String> props = new HashMap<>();
+        props.put("path.key", "conf/core-site.xml");
+
+        SampleConfig config = new SampleConfig();
+        ConnectorPropertiesUtils.bindConnectorProperties(config, props);
+
+        Assertions.assertEquals("conf/core-site.xml", config.getPathValue());
+    }
+
     @Test
     void testGetConnectorPropertiesIncludesSuperclassFields() {
         List<java.lang.reflect.Field> fields = ConnectorPropertiesUtils.getConnectorProperties(SampleConfig.class);
@@ -188,6 +212,9 @@ static class SampleConfig extends BaseConfig {
         @ConnectorProperty(names = {"secret.key", "secret.alias"}, sensitive = true)
         private String secretValue;
 
+        @ConnectorProperty(names = {"path.key"}, validator = NoPathTraversalValidator.class)
+        private String pathValue;
+
         public String getStringValue() {
             return stringValue;
         }
@@ -223,6 +250,10 @@ public UnsupportedFieldType getUnsupportedField() {
         public String getSecretValue() {
             return secretValue;
         }
+
+        public String getPathValue() {
+            return pathValue;
+        }
     }
 
     static class UnsupportedFieldType {