[fix](tvf) Secure LOCAL TVF access (4.1)

be/src/format_v2/table/lance_reader.cpp

@@ -35,11 +35,36 @@
 #include "format_v2/lance/lance_reader_helper.h"
 #include "format_v2/lance/lance_runtime_filter_helper.h"
 #include "format_v2/lance/lance_session_manager.h"
+#include "io/fs/local_file_system.h"
 #include "runtime/file_scan_profile.h"
 #include "runtime/runtime_state.h"
 
 namespace doris::format::lance {
 
+namespace {
+Status resolve_dataset_uri(const TFileRangeDesc& range, const TFileScanRangeParams& scan_params,
+                           std::string* uri) {
+    const auto& requested_uri = range.table_format_params.lance_params.dataset_uri;
+    // Modern plans tag each range; older plans put the type in the scan parameters.
+    // Untagged catalog scans retain their original storage-provider selection.
+    const bool is_local = range.__isset.file_type
+                                  ? range.file_type == TFileType::FILE_LOCAL
+                                  : scan_params.__isset.file_type &&
+                                            scan_params.file_type == TFileType::FILE_LOCAL;
+    // LOCAL TVF uses version zero for both schema discovery and whole-dataset scans.
+    // Catalog scans pin a positive snapshot version, including file:// catalogs whose
+    // operator-configured warehouse is independent of the LOCAL TVF secure directory.
+    if (is_local && range.table_format_params.lance_params.version == 0) {
+        // LOCAL Lance bypasses glob because it opens a dataset directory. Apply the same
+        // canonical secure-root check before either schema discovery or a cached dataset open.
+        return io::global_local_filesystem()->canonicalize_local_file(
+                config::user_files_secure_path, requested_uri, uri);
+    }
+    *uri = requested_uri;
+    return Status::OK();
+}
+} // namespace
+
 LanceTableReader::~LanceTableReader() {
     static_cast<void>(close());
 }
@@ -52,6 +77,8 @@ Status LanceTableReader::fetch_schema(const TFileRangeDesc& range,
         return Status::InvalidArgument("Lance schema output must not be null");
     }
     const auto& params = range.table_format_params.lance_params;
+    std::string dataset_uri;
+    RETURN_IF_ERROR(resolve_dataset_uri(range, scan_params, &dataset_uri));
     std::vector<std::string> storage_options;
     RETURN_IF_ERROR(build_lance_storage_options(&scan_params, &storage_options));
     std::vector<const char*> storage_option_ptrs;
@@ -61,10 +88,9 @@ Status LanceTableReader::fetch_schema(const TFileRangeDesc& range,
     }
     storage_option_ptrs.emplace_back(nullptr);
 
-    std::unique_ptr<LanceDataset, LanceDatasetDeleter> dataset(
-            lance_dataset_open(params.dataset_uri.c_str(),
-                               storage_options.empty() ? nullptr : storage_option_ptrs.data(),
-                               static_cast<uint64_t>(params.version)));
+    std::unique_ptr<LanceDataset, LanceDatasetDeleter> dataset(lance_dataset_open(
+            dataset_uri.c_str(), storage_options.empty() ? nullptr : storage_option_ptrs.data(),
+            static_cast<uint64_t>(params.version)));
     if (dataset == nullptr) {
         return lance_error("open Lance dataset for schema");
     }
@@ -1329,7 +1355,7 @@ Status LanceTableReader::_fill_block_from_lance_batch(LanceBatch* batch, Block*
 
 Status LanceTableReader::_dataset_key(const TFileRangeDesc& range, DatasetKey* key) const {
     const auto& params = range.table_format_params.lance_params;
-    key->uri = params.dataset_uri;
+    RETURN_IF_ERROR(resolve_dataset_uri(range, *_scan_params, &key->uri));
     key->version = params.version;
     return build_lance_storage_options(_scan_params, &key->storage_options);
 }

be/src/io/fs/local_file_system.cpp

@@ -455,28 +455,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

@@ -70,6 +70,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/format_v2/table/lance_reader_test.cpp

@@ -49,6 +49,7 @@
 #include <utility>
 #include <vector>
 
+#include "common/config.h"
 #include "common/consts.h"
 #include "core/assert_cast.h"
 #include "core/block/block.h"
@@ -4313,3 +4314,105 @@ TEST(LanceTableReaderTypeTest, ReadsTemporalTypesFromAllTypesFixture) {
 }
 
 } // namespace doris::format::lance
+
+namespace doris::format::lance {
+TEST(LanceTableReaderLocalPathTest, SchemaAndScanUseTheSameSecurePath) {
+    const auto original_secure_path = config::user_files_secure_path;
+    DEFER(config::user_files_secure_path = original_secure_path);
+    const auto root = std::filesystem::canonical("./be/test/format_v2/table/lance/data");
+    config::user_files_secure_path = root.string();
+    TFileScanRangeParams scan_params;
+    LanceTableReader reader;
+    reader._scan_params = &scan_params;
+    auto range = make_latest_lance_range("all_types.lance");
+    range.__set_file_type(TFileType::FILE_LOCAL);
+    std::vector<std::string> names;
+    std::vector<DataTypePtr> types;
+    ASSERT_TRUE(reader.fetch_schema(range, scan_params, &names, &types).ok());
+    EXPECT_FALSE(names.empty());
+    LanceTableReader::DatasetKey key;
+    ASSERT_TRUE(reader._dataset_key(range, &key).ok());
+    EXPECT_EQ((root / "all_types.lance").string(), key.uri);
+
+    // Legacy scan-level tags must enforce the same restriction as per-range tags.
+    range.__isset.file_type = false;
+    scan_params.__set_file_type(TFileType::FILE_LOCAL);
+    ASSERT_TRUE(reader._dataset_key(range, &key).ok());
+    EXPECT_EQ((root / "all_types.lance").string(), key.uri);
+
+    for (const auto* uri : {"https://invalid/dataset", "http://invalid/dataset",
+                            "s3://bucket/dataset", "../data-other/dataset"}) {
+        range.table_format_params.lance_params.__set_dataset_uri(uri);
+        const auto schema_status = reader.fetch_schema(range, scan_params, &names, &types);
+        const auto scan_status = reader._dataset_key(range, &key);
+        EXPECT_FALSE(schema_status.ok()) << uri;
+        EXPECT_EQ(schema_status.to_string(), scan_status.to_string());
+    }
+}
+
+TEST(LanceTableReaderLocalPathTest, ScansRelativeDatasetInsideSecureRoot) {
+    const auto original_secure_path = config::user_files_secure_path;
+    DEFER(config::user_files_secure_path = original_secure_path);
+    config::user_files_secure_path =
+            std::filesystem::canonical("./be/test/format_v2/table/lance/data").string();
+    TQueryGlobals query_globals;
+    RuntimeState state(query_globals);
+    RuntimeProfile profile("lance_secure_local_scan");
+    TFileScanRangeParams scan_params;
+    const Columns columns {projected_column("row_id", TYPE_BIGINT, false)};
+    LanceTableReader reader;
+    ASSERT_TRUE(init_reader(&reader, columns, &state, &profile, &scan_params).ok());
+    auto range = make_latest_lance_range("all_types.lance");
+    range.__set_file_type(TFileType::FILE_LOCAL);
+    ASSERT_TRUE(prepare_range(&reader, range).ok());
+
+    // Exercise data reads through the validated relative LOCAL range.
+    Block block;
+    add_output_columns(&block, columns);
+    std::vector<int64_t> row_ids;
+    bool eos = false;
+    while (!eos) {
+        ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+        if (eos) {
+            continue;
+        }
+        const auto& values =
+                assert_cast<const ColumnInt64&>(*block.get_by_position(0).column).get_data();
+        for (size_t row = 0; row < block.rows(); ++row) {
+            row_ids.push_back(values[row]);
+        }
+    }
+    std::sort(row_ids.begin(), row_ids.end());
+    EXPECT_EQ((std::vector<int64_t> {1, 2, 3, 4}), row_ids);
+    ASSERT_TRUE(reader.close().ok());
+}
+
+TEST(LanceTableReaderLocalPathTest, FixedVersionLocalCatalogKeepsItsUri) {
+    TFileScanRangeParams scan_params;
+    LanceTableReader reader;
+    reader._scan_params = &scan_params;
+    // Catalog warehouses use their own location, not the LOCAL TVF secure directory.
+    auto range = make_lance_range("file:///catalog-owned/dataset.lance", 7, {0});
+    range.__set_file_type(TFileType::FILE_LOCAL);
+    LanceTableReader::DatasetKey key;
+    ASSERT_TRUE(reader._dataset_key(range, &key).ok());
+    EXPECT_EQ("file:///catalog-owned/dataset.lance", key.uri);
+    EXPECT_EQ(7, key.version);
+}
+
+TEST(LanceTableReaderLocalPathTest, RemoteRangeKeepsItsUriAndOptions) {
+    TFileScanRangeParams scan_params;
+    scan_params.__set_file_type(TFileType::FILE_LOCAL);
+    TLanceScanParams lance_params;
+    lance_params.__set_lance_storage_options({{"aws_region", "us-east-1"}});
+    scan_params.__set_lance_scan_params(lance_params);
+    LanceTableReader reader;
+    reader._scan_params = &scan_params;
+    auto range = make_latest_lance_range("s3://bucket/dataset.lance");
+    range.__set_file_type(TFileType::FILE_S3);
+    LanceTableReader::DatasetKey key;
+    ASSERT_TRUE(reader._dataset_key(range, &key).ok());
+    EXPECT_EQ("s3://bucket/dataset.lance", key.uri);
+    EXPECT_EQ((std::vector<std::string> {"aws_region", "us-east-1"}), key.storage_options);
+}
+} // namespace doris::format::lance

be/test/io/fs/local_file_system_test.cpp

@@ -30,12 +30,14 @@
 #include <filesystem>
 #include <vector>
 
+#include "common/config.h"
 #include "common/status.h"
 #include "cpp/sync_point.h"
 #include "gtest/gtest_pred_impl.h"
 #include "io/fs/file_reader.h"
 #include "io/fs/file_writer.h"
 #include "io/fs/local_file_writer.h"
+#include "util/defer_op.h"
 #include "util/slice.h"
 
 namespace doris {
@@ -547,6 +549,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()
@@ -578,6 +584,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

@@ -3101,6 +3101,10 @@ public class Config extends ConfigBase {
     })
     public static boolean enable_java_udf = true;
 
+    @ConfField(description = "Whether to enable LOCAL TVF and FILE TVF with local storage. 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(mutable = true, masterOnly = true, description = {
         "开启后，可以在导入时，利用创建的全局 java_udf 函数处理数据，默认为 false。",
         "When enabled, data can be processed using the globally created java_udf function during import."

fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java

@@ -241,6 +241,9 @@ protected void setScanParams(TFileRangeDesc rangeDesc, Split split) {
             if (!(split instanceof LanceSplit)) {
                 throw new IllegalArgumentException("Expected LanceSplit but got " + split.getClass().getName());
             }
+            // A bare local path has no URI scheme, so split inference cannot supply its type.
+            // Preserve the TVF's LOCAL/S3 identity for the BE's pre-open path validation.
+            rangeDesc.setFileType(tableValuedFunction.getTFileType());
             LanceSplit lanceSplit = (LanceSplit) split;
             TLanceFileDesc lanceParams = new TLanceFileDesc();
             lanceParams.setDatasetUri(lanceSplit.getDatasetUri());