[fix](tvf) Restrict local paths and require explicit startup opt-in

be/src/io/fs/local_file_system.cpp

@@ -426,28 +426,48 @@ const std::shared_ptr<LocalFileSystem>& global_local_filesystem() {
 Status LocalFileSystem::canonicalize_local_file(const std::string& dir,
                                                 const std::string& file_path,
                                                 std::string* full_path) {
-    const std::string absolute_path = dir + "/" + file_path;
+    // An empty root would turn a relative request into an unrestricted absolute path.
+    if (dir.empty()) {
+        return Status::InvalidArgument("user_files_secure_path must not be empty");
+    }
+    // Reject embedded NULs before passing paths to filesystem APIs that use C strings.
+    if (file_path.empty() || file_path.find('\0') != std::string::npos) {
+        return Status::InvalidArgument("local file path must not be empty or contain NUL");
+    }
+    Path local_path;
+    // Convert file: URIs to filesystem paths and reject remote schemes before any reader opens them.
+    RETURN_IF_ERROR(convert_to_abs_path(file_path, local_path));
+    std::string secure_path;
+    RETURN_IF_ERROR(canonicalize(dir, &secure_path));
     std::string canonical_path;
-    RETURN_IF_ERROR(canonicalize(absolute_path, &canonical_path));
-    if (!contain_path(dir, canonical_path)) {
-        return Status::InvalidArgument("file path is not allowed: {}", canonical_path);
+    const auto status = canonicalize(Path(secure_path) / local_path, &canonical_path);
+    // A glob result can reveal external names even when resolution fails on a dangling symlink.
+    // Reject both failed resolution and out-of-root targets without exposing expanded names.
+    if (!status.ok() || !contain_path(secure_path, canonical_path)) {
+        return Status::InvalidArgument("file path is not allowed");
     }
-
-    *full_path = canonical_path;
+    *full_path = std::move(canonical_path);
     return Status::OK();
 }
 
 Status LocalFileSystem::safe_glob(const std::string& path, std::vector<FileInfo>* res) {
     if (path.find("..") != std::string::npos) {
         return Status::InvalidArgument("can not contain '..' in path");
     }
-    std::string full_path = config::user_files_secure_path + "/" + path;
+    // An empty root would turn the prefix below into "/" and allow any local file.
+    if (config::user_files_secure_path.empty()) {
+        return Status::InvalidArgument("user_files_secure_path must not be empty");
+    }
+    std::string secure_path;
+    RETURN_IF_ERROR(canonicalize(config::user_files_secure_path, &secure_path));
+    std::string full_path = secure_path + "/" + path;
     std::vector<std::string> files;
     RETURN_IF_ERROR(_glob(full_path, &files));
     for (auto& file : files) {
         FileInfo fi;
         fi.is_file = true;
-        RETURN_IF_ERROR(canonicalize_local_file("", file, &(fi.file_name)));
+        // Glob classes and symlinks can escape without a literal ".." in the pattern.
+        RETURN_IF_ERROR(canonicalize_local_file(secure_path, file, &(fi.file_name)));
         RETURN_IF_ERROR(file_size_impl(fi.file_name, &(fi.file_size)));
         res->push_back(std::move(fi));
     }

be/src/io/fs/local_file_system.h

@@ -67,6 +67,8 @@ class LocalFileSystem final : public FileSystem {
 
     static std::filesystem::perms PERMS_OWNER_RW;
 
+    // Resolve a local path relative to a nonempty secure root, then check canonical containment.
+    // Absolute paths and file: URIs must resolve inside the same root; remote schemes are rejected.
     Status canonicalize_local_file(const std::string& dir, const std::string& file_path,
                                    std::string* full_path);
 

be/test/io/fs/local_file_system_test.cpp

@@ -29,11 +29,13 @@
 #include <filesystem>
 #include <vector>
 
+#include "common/config.h"
 #include "common/status.h"
 #include "common/sync_point.h"
 #include "gtest/gtest_pred_impl.h"
 #include "io/fs/file_reader.h"
 #include "io/fs/file_writer.h"
+#include "util/defer_op.h"
 #include "util/slice.h"
 
 namespace doris {
@@ -376,6 +378,10 @@ TEST_F(LocalFileSystemTest, AbnormalWriteRead) {
 }
 
 TEST_F(LocalFileSystemTest, TestGlob) {
+    const std::string original_secure_path = config::user_files_secure_path;
+    DEFER(config::user_files_secure_path = original_secure_path);
+    // Bind this fixture to its own files, independent of the runner's build-profile directory.
+    config::user_files_secure_path = std::filesystem::absolute("./be/ut_build_ASAN/test").string();
     std::string path = "./be/ut_build_ASAN/test/file_path/";
     EXPECT_TRUE(io::global_local_filesystem()->delete_directory(path).ok());
     EXPECT_TRUE(io::global_local_filesystem()
@@ -407,6 +413,101 @@ TEST_F(LocalFileSystemTest, TestGlob) {
     EXPECT_TRUE(io::global_local_filesystem()->delete_directory(path).ok());
 }
 
+TEST_F(LocalFileSystemTest, TestSafeGlobPathContainment) {
+    auto fs = io::global_local_filesystem();
+    const std::string original_secure_path = config::user_files_secure_path;
+    DEFER(config::user_files_secure_path = original_secure_path);
+    const auto root = std::filesystem::canonical(test_dir);
+    const auto secure = root / "secure";
+    const auto outside = root / "secure-other";
+    std::filesystem::create_directories(secure / "nested");
+    std::filesystem::create_directory(outside);
+    ASSERT_TRUE(save_string_file((secure / "nested/inside.csv").string(), "inside").ok());
+    ASSERT_TRUE(save_string_file((outside / "outside.csv").string(), "outside").ok());
+    std::filesystem::create_directory_symlink(outside, secure / "escape");
+    std::filesystem::create_symlink(outside / "outside.csv", secure / "outside.csv");
+    std::filesystem::create_symlink(outside / "missing.csv", outside / "broken.csv");
+    std::filesystem::create_directory_symlink(secure / "nested", secure / "inside");
+    config::user_files_secure_path = secure.string();
+
+    for (const auto* pattern : {".[.]/secure-other/outside.csv", ".?/secure-other/outside.csv",
+                                ".[.-.]/secure-other/outside.csv", "escape/*.csv", "outside.csv",
+                                ".[.]/secure-other/*.csv", "escape/broken*.csv"}) {
+        std::vector<io::FileInfo> files;
+        const auto status = fs->safe_glob(pattern, &files);
+        EXPECT_FALSE(status.ok()) << pattern;
+        EXPECT_NE(status.to_string().find("file path is not allowed"), std::string::npos) << status;
+        // The pattern need not name the matched file, so errors must not reveal that name.
+        const auto message = status.to_string();
+        EXPECT_EQ(std::string::npos, message.find(root.string())) << message;
+        for (const auto* name : {"outside.csv", "broken.csv", "missing.csv"}) {
+            EXPECT_EQ(std::string::npos, message.find(name)) << message;
+        }
+        EXPECT_TRUE(files.empty());
+    }
+
+    for (const auto* pattern : {"nested/*.csv", "inside/*.csv"}) {
+        std::vector<io::FileInfo> files;
+        ASSERT_TRUE(fs->safe_glob(pattern, &files).ok()) << pattern;
+        ASSERT_EQ(1, files.size());
+        EXPECT_EQ((secure / "nested/inside.csv").string(), files[0].file_name);
+        EXPECT_EQ(6, files[0].file_size);
+    }
+}
+
+TEST_F(LocalFileSystemTest, TestSafeGlobSecureRoot) {
+    auto fs = io::global_local_filesystem();
+    const std::string original_secure_path = config::user_files_secure_path;
+    DEFER(config::user_files_secure_path = original_secure_path);
+    const auto root = std::filesystem::canonical(test_dir);
+    std::filesystem::create_directory(root / "secure");
+    std::filesystem::create_directory_symlink(root / "secure", root / "secure-link");
+    ASSERT_TRUE(save_string_file((root / "secure/inside.csv").string(), "inside").ok());
+
+    for (const auto& secure : {std::string(test_dir) + "/secure", (root / "secure/").string(),
+                               (root / "secure-link").string()}) {
+        config::user_files_secure_path = secure;
+        std::vector<io::FileInfo> files;
+        ASSERT_TRUE(fs->safe_glob("*.csv", &files).ok()) << secure;
+        ASSERT_EQ(1, files.size());
+        EXPECT_EQ((root / "secure/inside.csv").string(), files[0].file_name);
+    }
+    config::user_files_secure_path = "";
+    std::vector<io::FileInfo> files;
+    EXPECT_FALSE(fs->safe_glob("*.csv", &files).ok());
+    config::user_files_secure_path = (root / "missing").string();
+    EXPECT_FALSE(fs->safe_glob("*.csv", &files).ok());
+}
+
+TEST_F(LocalFileSystemTest, TestCanonicalizeSecureLocalPath) {
+    auto fs = io::global_local_filesystem();
+    const auto root = std::filesystem::canonical(test_dir);
+    const auto secure = root / "secure";
+    std::filesystem::create_directories(secure / "nested");
+    std::filesystem::create_directory(root / "secure-other");
+    std::filesystem::create_directory_symlink(root / "secure-other", secure / "escape");
+    std::filesystem::create_directory_symlink(secure / "nested", secure / "inside");
+    for (const auto& path :
+         {std::string("nested"), std::string("nested/../inside"), (secure / "nested").string(),
+          "file://" + (secure / "nested").string()}) {
+        std::string resolved;
+        ASSERT_TRUE(fs->canonicalize_local_file(secure.string(), path, &resolved).ok()) << path;
+        EXPECT_EQ((secure / "nested").string(), resolved);
+    }
+    for (const auto& path :
+         {std::string("../secure-other"), std::string("escape"), (root / "secure-other").string(),
+          std::string("https://invalid/dataset"), std::string("s3://bucket/dataset"),
+          std::string("http://invalid/dataset"), std::string(""),
+          std::string("nested\0ignored", 14)}) {
+        std::string resolved;
+        const auto status = fs->canonicalize_local_file(secure.string(), path, &resolved);
+        EXPECT_FALSE(status.ok()) << path;
+        EXPECT_EQ(std::string::npos, status.to_string().find(root.string())) << status;
+    }
+    std::string resolved;
+    EXPECT_FALSE(fs->canonicalize_local_file("", secure.string(), &resolved).ok());
+}
+
 TEST_F(LocalFileSystemTest, TestConvertToAbsPath) {
     io::Path abs_path;
     Status st;

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2619,6 +2619,10 @@ public class Config extends ConfigBase {
     })
     public static boolean enable_java_udf = true;
 
+    @ConfField(description = "Whether to enable LOCAL TVF. Disabled by default. "
+            + "Set this in fe.conf and restart every FE to enable it. Existing access checks still apply.")
+    public static boolean enable_local_tvf = false;
+
     @ConfField(description = {
             "是否忽略 Image 文件中未知的模块。如果为 true，不在 PersistMetaModules.MODULE_NAMES 中的元数据模块将被忽略并跳过。"
                     + "默认为 false，如果 Image 文件中包含未知的模块，Doris 将会抛出异常。"

fe/fe-core/src/main/java/org/apache/doris/tablefunction/LocalTableValuedFunction.java

@@ -21,9 +21,13 @@
 import org.apache.doris.analysis.StorageBackend.StorageType;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.Config;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.proto.InternalService;
 import org.apache.doris.proto.InternalService.PGlobResponse;
 import org.apache.doris.proto.InternalService.PGlobResponse.PFileInfo;
+import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.rpc.BackendServiceProxy;
 import org.apache.doris.system.Backend;
 import org.apache.doris.thrift.TBrokerFileStatus;
@@ -65,6 +69,14 @@ public class LocalTableValuedFunction extends ExternalFileTableValuedFunction {
     private boolean sharedStorage = false;
 
     public LocalTableValuedFunction(Map<String, String> properties) throws AnalysisException {
+        if (!Config.enable_local_tvf) {
+            throw new org.apache.doris.nereids.exceptions.AnalysisException(
+                    "LOCAL TVF is disabled. Set enable_local_tvf=true in fe.conf and restart every FE to enable it.");
+        }
+
+        // Check privileges before listing files or discovering their schema.
+        checkAuth(ConnectContext.get());
+
         // 1. analyze common properties
         Map<String, String> otherProps = super.parseCommonProperties(properties);
 
@@ -129,6 +141,16 @@ private void getFileListFromBackend() throws AnalysisException {
         }
     }
 
+    @Override
+    public void checkAuth(ConnectContext ctx) {
+        if (!Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.ADMIN)
+                && !Env.getCurrentEnv().getAccessManager().checkGlobalPriv(ctx, PrivPredicate.OPERATOR)) {
+            String message = ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR.formatErrorMsg("ADMIN/OPERATOR");
+            throw new org.apache.doris.nereids.exceptions.AnalysisException(message);
+        }
+        super.checkAuth(ctx);
+    }
+
     @Override
     public TFileType getTFileType() {
         return TFileType.FILE_LOCAL;

fe/fe-core/src/test/java/org/apache/doris/tablefunction/LocalTableValuedFunctionEnableTest.java

@@ -0,0 +1,142 @@
+// 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.tablefunction;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.proto.InternalService.PGlobResponse;
+import org.apache.doris.proto.Types.PStatus;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TNetworkAddress;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+public class LocalTableValuedFunctionEnableTest {
+    private static final Map<String, String> PROPERTIES = ImmutableMap.of(
+            "file_path", "local_privilege_test.csv", "backend_id", "1", "format", "csv");
+
+    private final Env env = Mockito.mock(Env.class);
+    private final ConnectContext context = Mockito.mock(ConnectContext.class);
+    private final AccessControllerManager accessManager = Mockito.mock(AccessControllerManager.class);
+    private final SystemInfoService systemInfo = Mockito.mock(SystemInfoService.class);
+    private final BackendServiceProxy proxy = Mockito.mock(BackendServiceProxy.class);
+    private boolean originalEnableLocalTvf;
+    private MockedStatic<Env> mockedEnv;
+    private MockedStatic<ConnectContext> mockedContext;
+    private MockedStatic<BackendServiceProxy> mockedProxy;
+
+    @BeforeEach
+    public void setUp() throws Exception {
+        originalEnableLocalTvf = Config.enable_local_tvf;
+        Config.enable_local_tvf = false;
+        mockedEnv = Mockito.mockStatic(Env.class);
+        mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+        mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfo);
+        mockedContext = Mockito.mockStatic(ConnectContext.class);
+        mockedContext.when(ConnectContext::get).thenReturn(context);
+        mockedProxy = Mockito.mockStatic(BackendServiceProxy.class);
+        mockedProxy.when(BackendServiceProxy::getInstance).thenReturn(proxy);
+        Mockito.when(env.getAccessManager()).thenReturn(accessManager);
+        Mockito.when(context.getSessionVariable()).thenReturn(new SessionVariable());
+        Mockito.when(context.getQueryTimeoutS()).thenReturn(60);
+
+        Backend backend = Mockito.mock(Backend.class);
+        Mockito.when(systemInfo.getBackend(1L)).thenReturn(backend);
+        Mockito.when(backend.getBrpcAddress()).thenReturn(new TNetworkAddress("127.0.0.1", 8060));
+        Mockito.when(proxy.glob(Mockito.any(), Mockito.any())).thenReturn(CompletableFuture.completedFuture(
+                PGlobResponse.newBuilder().setStatus(PStatus.newBuilder().setStatusCode(0)).build()));
+    }
+
+    @AfterEach
+    public void tearDown() {
+        Config.enable_local_tvf = originalEnableLocalTvf;
+        mockedProxy.close();
+        mockedContext.close();
+        mockedEnv.close();
+    }
+
+    @Test
+    public void testDisabledRejectsLocalBeforePropertyParsing() {
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                () -> new LocalTableValuedFunction(ImmutableMap.of()));
+        Assertions.assertTrue(exception.getMessage().contains("enable_local_tvf=true"));
+        Mockito.verifyNoInteractions(accessManager, systemInfo, proxy);
+    }
+
+    @Test
+    public void testDisabledRejectsCsvAndLanceBeforeFileAccess() {
+        for (String format : new String[] {"csv", "lance"}) {
+            Map<String, String> properties = ImmutableMap.of(
+                    "file_path", "local_enable_test.csv", "backend_id", "1", "format", format);
+            AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                    () -> new LocalTableValuedFunction(properties));
+            Assertions.assertTrue(exception.getMessage().contains("LOCAL TVF is disabled"));
+        }
+        Mockito.verifyNoInteractions(accessManager, systemInfo, proxy);
+    }
+
+    @Test
+    public void testEnabledLocalReachesExistingFileAccess() throws Exception {
+        Config.enable_local_tvf = true;
+        Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(true);
+        new LocalTableValuedFunction(PROPERTIES);
+        Mockito.verify(proxy).glob(Mockito.any(), Mockito.any());
+    }
+
+    @Test
+    public void testEnabledLocalRejectsUnprivilegedUserBeforeFileAccess() {
+        Config.enable_local_tvf = true;
+        AnalysisException exception = Assertions.assertThrows(AnalysisException.class,
+                () -> new LocalTableValuedFunction(PROPERTIES));
+        Assertions.assertTrue(exception.getMessage().contains("ADMIN/OPERATOR"));
+        Mockito.verifyNoInteractions(systemInfo, proxy);
+    }
+
+    @Test
+    public void testOperatorCanAccessLocalFiles() throws Exception {
+        Config.enable_local_tvf = true;
+        Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.OPERATOR)).thenReturn(true);
+        new LocalTableValuedFunction(PROPERTIES);
+        Mockito.verify(proxy).glob(Mockito.any(), Mockito.any());
+    }
+
+    @Test
+    public void testCheckAuthRejectsRevokedPrivileges() throws Exception {
+        Config.enable_local_tvf = true;
+        Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(true);
+        LocalTableValuedFunction local = new LocalTableValuedFunction(PROPERTIES);
+        Mockito.when(accessManager.checkGlobalPriv(context, PrivPredicate.ADMIN)).thenReturn(false);
+        Assertions.assertThrows(AnalysisException.class, () -> local.checkAuth(context));
+    }
+}

regression-test/pipeline/external/conf/fe.conf

@@ -1,3 +1,6 @@
+# Exercise LOCAL TVF explicitly; production defaults to disabled.
+enable_local_tvf=true
+
 # 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