From 448f9735933816b7d4d5ce7dd080988bc29fbb51 Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Fri, 31 Jul 2026 21:23:15 +0800 Subject: [PATCH] fix: sandbox SQL local filesystem access (#8708) * fix: sandbox SQL local filesystem access Signed-off-by: jeremyhi * fix: address local file sandbox review findings Signed-off-by: jeremyhi * fix: support Windows local copy paths Signed-off-by: jeremyhi * fix: improve sandbox path errors Signed-off-by: jeremyhi * refactor: simplify local path error context Signed-off-by: jeremyhi * perf: stream secure filesystem listings Signed-off-by: jeremyhi * style: derive local file access default Signed-off-by: jeremyhi * fix: improve local file access errors Signed-off-by: jeremyhi * fix: address local file access review findings Signed-off-by: jeremyhi * test: simplify local file access coverage Signed-off-by: jeremyhi * fix: harden sandboxed local file backends Signed-off-by: jeremyhi * fix: reject directory copy targets before creation Signed-off-by: jeremyhi * fix: avoid implicit string clone in file table listing Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- .github/workflows/develop.yml | 4 + Cargo.lock | 101 +++ Cargo.toml | 1 + config/config.md | 1 + config/standalone.example.toml | 9 + docs/how-to/migrate-local-sql-file-access.md | 30 + src/cmd/Cargo.toml | 1 + src/cmd/src/standalone.rs | 133 +++- src/common/datasource/src/error.rs | 62 ++ src/common/datasource/src/lister.rs | 2 +- src/common/datasource/src/object_store.rs | 669 ++++++++++++++++- src/common/datasource/src/object_store/fs.rs | 13 +- src/datanode/Cargo.toml | 1 + src/datanode/src/config.rs | 6 + src/datanode/src/datanode.rs | 9 + src/file-engine/src/engine.rs | 17 +- src/file-engine/src/query.rs | 12 +- src/file-engine/src/region.rs | 39 + src/flow/Cargo.toml | 1 + src/flow/src/server.rs | 2 + src/frontend/src/instance/builder.rs | 9 + src/object-store/Cargo.toml | 2 + src/object-store/src/lib.rs | 1 + src/object-store/src/secure_fs.rs | 682 ++++++++++++++++++ src/operator/src/error.rs | 22 +- src/operator/src/expr_helper.rs | 4 +- src/operator/src/statement.rs | 4 + src/operator/src/statement/copy_database.rs | 56 +- src/operator/src/statement/copy_table_from.rs | 36 +- src/operator/src/statement/copy_table_to.rs | 16 +- src/operator/src/statement/ddl.rs | 4 +- src/query/src/error.rs | 23 +- src/query/src/sql.rs | 23 +- tests-integration/Cargo.toml | 1 + tests-integration/src/standalone.rs | 5 + tests-integration/src/test_util.rs | 1 + tests-integration/src/tests/instance_test.rs | 105 ++- tests-integration/src/tests/test_util.rs | 10 + .../distributed/local_file_access.result | 49 ++ tests/cases/distributed/local_file_access.sql | 30 + .../copy/copy_database_from_fs_parquet.result | 0 .../copy/copy_database_from_fs_parquet.sql | 0 .../copy/copy_from_csv_compressed.result | 0 .../copy/copy_from_csv_compressed.sql | 0 .../{common => }/copy/copy_from_fs_csv.result | 0 .../{common => }/copy/copy_from_fs_csv.sql | 0 .../copy/copy_from_fs_json.result | 0 .../{common => }/copy/copy_from_fs_json.sql | 0 .../copy/copy_from_fs_parquet.result | 0 .../copy/copy_from_fs_parquet.sql | 0 .../copy/copy_from_json_compressed.result | 0 .../copy/copy_from_json_compressed.sql | 0 .../copy/copy_to_csv_compressed.result | 0 .../copy/copy_to_csv_compressed.sql | 0 .../{common => }/copy/copy_to_fs.result | 0 .../{common => }/copy/copy_to_fs.sql | 0 .../copy/copy_to_json_compressed.result | 0 .../copy/copy_to_json_compressed.sql | 0 .../cases/standalone/local_file_access.result | 55 ++ tests/cases/standalone/local_file_access.sql | 29 + tests/conf/standalone-test.toml.template | 1 + tests/runner/src/cmd/compat.rs | 5 +- tests/runner/src/env/bare.rs | 5 +- tests/runner/src/server_mode.rs | 2 + 64 files changed, 2153 insertions(+), 140 deletions(-) create mode 100644 docs/how-to/migrate-local-sql-file-access.md create mode 100644 src/object-store/src/secure_fs.rs create mode 100644 tests/cases/distributed/local_file_access.result create mode 100644 tests/cases/distributed/local_file_access.sql rename tests/cases/standalone/{common => }/copy/copy_database_from_fs_parquet.result (100%) rename tests/cases/standalone/{common => }/copy/copy_database_from_fs_parquet.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_from_csv_compressed.result (100%) rename tests/cases/standalone/{common => }/copy/copy_from_csv_compressed.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_from_fs_csv.result (100%) rename tests/cases/standalone/{common => }/copy/copy_from_fs_csv.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_from_fs_json.result (100%) rename tests/cases/standalone/{common => }/copy/copy_from_fs_json.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_from_fs_parquet.result (100%) rename tests/cases/standalone/{common => }/copy/copy_from_fs_parquet.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_from_json_compressed.result (100%) rename tests/cases/standalone/{common => }/copy/copy_from_json_compressed.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_to_csv_compressed.result (100%) rename tests/cases/standalone/{common => }/copy/copy_to_csv_compressed.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_to_fs.result (100%) rename tests/cases/standalone/{common => }/copy/copy_to_fs.sql (100%) rename tests/cases/standalone/{common => }/copy/copy_to_json_compressed.result (100%) rename tests/cases/standalone/{common => }/copy/copy_to_json_compressed.sql (100%) create mode 100644 tests/cases/standalone/local_file_access.result create mode 100644 tests/cases/standalone/local_file_access.sql diff --git a/.github/workflows/develop.yml b/.github/workflows/develop.yml index 8a4d423061..15792779a5 100644 --- a/.github/workflows/develop.yml +++ b/.github/workflows/develop.yml @@ -850,7 +850,10 @@ jobs: - name: Unzip binaries run: tar -xvf ./bins.tar.gz - name: Start GreptimeDB standalone + env: + GREPTIMEDB_STANDALONE__STORAGE__COPY_ROOT: ${{ runner.temp }}/greptime-export-import-v2 run: | + mkdir -p "${GREPTIMEDB_STANDALONE__STORAGE__COPY_ROOT}" ./bins/greptime standalone start > greptimedb.log 2>&1 & greptime_pid=$! echo "Waiting for GreptimeDB..." @@ -870,6 +873,7 @@ jobs: - name: Run export/import v2 E2E tests run: cargo test -p cli data::export_v2::tests --lib -- --ignored --test-threads=1 env: + TMPDIR: ${{ runner.temp }}/greptime-export-import-v2 GREPTIME_ADDR: 127.0.0.1:4000 GT_S3_BUCKET: greptime GT_S3_ACCESS_KEY_ID: superpower_ci_user diff --git a/Cargo.lock b/Cargo.lock index d24b38e033..f82073177c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -125,6 +125,12 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "ambient-authority" +version = "0.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9d4ee0d472d1cd2e28c97dfa124b3d8d992e10eb0a035f33f5d12e3a177ba3b" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -1603,6 +1609,36 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "acbc26382d871df4b7442e3df10a9402bf3cf5e55cbd66f12be38861425f0564" +[[package]] +name = "cap-primitives" +version = "4.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdadbd7c002d3a484b35243669abdae85a0ebaded5a61117169dc3400f9a7ff0" +dependencies = [ + "ambient-authority", + "fs-set-times", + "io-extras", + "io-lifetimes 3.0.1", + "ipnet", + "maybe-owned", + "rustix 1.0.7", + "rustix-linux-procfs", + "windows-sys 0.61.2", + "winx", +] + +[[package]] +name = "cap-std" +version = "4.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7281235d6e96d3544ca18bba9049be92f4190f8d923e3caef1b5f66cfa752608" +dependencies = [ + "cap-primitives", + "io-extras", + "io-lifetimes 3.0.1", + "rustix 1.0.7", +] + [[package]] name = "cargo-manifest" version = "0.19.1" @@ -2113,6 +2149,7 @@ dependencies = [ "common-base", "common-catalog", "common-config", + "common-datasource", "common-error", "common-grpc", "common-macro", @@ -4424,6 +4461,7 @@ dependencies = [ "common-base", "common-catalog", "common-config", + "common-datasource", "common-error", "common-function", "common-greptimedb-telemetry", @@ -5371,6 +5409,7 @@ dependencies = [ "common-base", "common-catalog", "common-config", + "common-datasource", "common-decimal", "common-error", "common-frontend", @@ -5643,6 +5682,17 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "fs-set-times" +version = "0.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94e7099f6313ecacbe1256e8ff9d617b75d1bcb16a6fddef94866d225a01a14a" +dependencies = [ + "io-lifetimes 2.0.4", + "rustix 1.0.7", + "windows-sys 0.59.0", +] + [[package]] name = "fs2" version = "0.4.3" @@ -7029,6 +7079,28 @@ dependencies = [ "derive_utils", ] +[[package]] +name = "io-extras" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20fd6de4ccfcc187e38bc21cfa543cb5a302cb86a8b114eb7f0bf0dc9f8ac00f" +dependencies = [ + "io-lifetimes 3.0.1", + "windows-sys 0.60.2", +] + +[[package]] +name = "io-lifetimes" +version = "2.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06432fb54d3be7964ecd3649233cddf80db2832f47fec34c01f65b3d9d774983" + +[[package]] +name = "io-lifetimes" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f0fb0570afe1fed943c5c3d4102d5358592d8625fda6a0007fdbe65a92fba96" + [[package]] name = "ipcrypt-rs" version = "0.9.4" @@ -8149,6 +8221,12 @@ dependencies = [ "rawpointer", ] +[[package]] +name = "maybe-owned" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4facc753ae494aeb6e3c22f839b158aebd4f9270f55cd3c79906c45476c47ab4" + [[package]] name = "md-5" version = "0.10.6" @@ -9282,10 +9360,12 @@ dependencies = [ "anyhow", "async-trait", "bytes", + "cap-std", "chrono", "common-base", "common-error", "common-macro", + "common-runtime", "common-telemetry", "common-test-util", "derive_builder 0.20.2", @@ -12632,6 +12712,16 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "rustix-linux-procfs" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2fc84bf7e9aa16c4f2c758f27412dc9841341e16aa682d9c7ac308fe3ee12056" +dependencies = [ + "once_cell", + "rustix 1.0.7", +] + [[package]] name = "rustls" version = "0.23.28" @@ -14723,6 +14813,7 @@ dependencies = [ "common-base", "common-catalog", "common-config", + "common-datasource", "common-error", "common-event-recorder", "common-frontend", @@ -16805,6 +16896,16 @@ dependencies = [ "memchr", ] +[[package]] +name = "winx" +version = "0.36.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f3fd376f71958b862e7afb20cfe5a22830e1963462f3a17f49d82a6c1d1f42d" +dependencies = [ + "bitflags 2.12.1", + "windows-sys 0.59.0", +] + [[package]] name = "wit-bindgen" version = "0.51.0" diff --git a/Cargo.toml b/Cargo.toml index 3c40e8df7a..ac7161aeaa 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -119,6 +119,7 @@ bigdecimal = "0.4.2" bitflags = "2.4.1" bytemuck = "1.12" bytes = { version = "1.11", features = ["serde"] } +cap-std = "4.0" chrono = { version = "0.4", features = ["serde"] } chrono-tz = { version = "0.10", features = ["case-insensitive"] } clap = { version = "4.4", features = ["derive"] } diff --git a/config/config.md b/config/config.md index 8650fcdf08..6aa18e66dc 100644 --- a/config/config.md +++ b/config/config.md @@ -124,6 +124,7 @@ | `query.memory_pool_size` | String | `50%` | Memory pool size for query execution operators (aggregation, sorting, join).
Supports absolute size (e.g., "2GB", "4GB") or percentage of system memory (e.g., "20%").
Setting it to 0 disables the limit (unbounded, default behavior).
When this limit is reached, queries will fail with ResourceExhausted error.
NOTE: This does NOT limit memory used by table scans. | | `storage` | -- | -- | The data storage options. | | `storage.data_home` | String | `./greptimedb_data` | The working home directory. | +| `storage.copy_root` | String | `./greptimedb_data/copy` | Root directory for standalone SQL access to local files.
Relative SQL paths are resolved below this directory. Absolute paths are accepted only when
they are inside this directory. Defaults to `/copy`.
Distributed deployments always reject SQL access to local files.
Upgrade note: COPY commands and existing external tables that reference paths outside this
directory will fail. Move those files below the copy root, set this option to a dedicated
directory containing them, or migrate the files to object storage before upgrading. | | `storage.type` | String | `File` | The storage type used to store the data.
- `File`: the data is stored in the local file system.
- `S3`: the data is stored in the S3 object storage.
- `Gcs`: the data is stored in the Google Cloud Storage.
- `Azblob`: the data is stored in the Azure Blob Storage.
- `Oss`: the data is stored in the Aliyun OSS. | | `storage.bucket` | String | Unset | The S3 bucket name.
**It's only used when the storage type is `S3`, `Oss` and `Gcs`**. | | `storage.root` | String | Unset | The S3 data will be stored in the specified prefix, for example, `s3://${bucket}/${root}`.
**It's only used when the storage type is `S3`, `Oss` and `Azblob`**. | diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 371b64864c..fccfe28695 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -442,6 +442,15 @@ memory_pool_size = "50%" ## The working home directory. data_home = "./greptimedb_data" +## Root directory for standalone SQL access to local files. +## Relative SQL paths are resolved below this directory. Absolute paths are accepted only when +## they are inside this directory. Defaults to `/copy`. +## Distributed deployments always reject SQL access to local files. +## Upgrade note: COPY commands and existing external tables that reference paths outside this +## directory will fail. Move those files below the copy root, set this option to a dedicated +## directory containing them, or migrate the files to object storage before upgrading. +#+ copy_root = "./greptimedb_data/copy" + ## The storage type used to store the data. ## - `File`: the data is stored in the local file system. ## - `S3`: the data is stored in the S3 object storage. diff --git a/docs/how-to/migrate-local-sql-file-access.md b/docs/how-to/migrate-local-sql-file-access.md new file mode 100644 index 0000000000..8051f21395 --- /dev/null +++ b/docs/how-to/migrate-local-sql-file-access.md @@ -0,0 +1,30 @@ +# Migrate Local SQL File Access + +SQL access to local files is sandboxed in standalone deployments and disabled in +distributed deployments. + +## Standalone + +The default sandbox is `/copy`. Relative paths in `COPY` and +external-table locations are resolved below this directory. Absolute paths work +only when they are inside the sandbox. + +Before upgrading, identify existing `COPY` workflows and external tables that +use local paths outside the default sandbox. Choose one of these migrations: + +- Move the files below `/copy` and update the SQL locations. +- Set `storage.copy_root` to a dedicated local directory containing the files. +- Move the files to S3, OSS, GCS, or AzBlob and update the SQL locations. + +Do not set `storage.copy_root` to `storage.data_home` or to a directory that +contains GreptimeDB data, WAL, manifests, or configuration files. GreptimeDB +rejects copy roots that expose its internal data directory. + +When `storage.data_home` is an object-storage URL, local SQL file access is +disabled unless `storage.copy_root` explicitly names a local directory. + +## Distributed + +Distributed frontend and datanode processes reject local paths for `COPY TABLE`, +`COPY QUERY`, `COPY DATABASE`, and external tables. Migrate these workflows and +tables to S3, OSS, GCS, or AzBlob before upgrading. diff --git a/src/cmd/Cargo.toml b/src/cmd/Cargo.toml index 8002ab742f..24edcd5633 100644 --- a/src/cmd/Cargo.toml +++ b/src/cmd/Cargo.toml @@ -47,6 +47,7 @@ colored = "2.1.0" common-base.workspace = true common-catalog.workspace = true common-config.workspace = true +common-datasource.workspace = true common-error.workspace = true common-grpc.workspace = true common-macro.workspace = true diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs index 5690c2d0bc..aec373612a 100644 --- a/src/cmd/src/standalone.rs +++ b/src/cmd/src/standalone.rs @@ -28,6 +28,7 @@ use clap::Parser; use common_base::Plugins; use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID}; use common_config::{Configurable, metadata_store_dir}; +use common_datasource::object_store::{LocalFileAccess, configured_local_path}; use common_error::ext::BoxedError; use common_meta::DatanodeId; use common_meta::cache::{LayeredCacheRegistryBuilder, LayeredCacheRegistryRef}; @@ -51,7 +52,7 @@ use common_telemetry::info; use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions}; use common_time::timezone::set_default_timezone; use common_version::{short_version, verbose_version}; -use datanode::config::DatanodeOptions; +use datanode::config::{DatanodeOptions, StorageConfig}; use datanode::datanode::{Datanode, DatanodeBuilder}; use datanode::region_server::RegionServer; use flow::{ @@ -69,7 +70,7 @@ use plugins::frontend::context::{ }; use plugins::standalone::context::DdlManagerConfigureContext; use servers::tls::{TlsMode, TlsOption, merge_tls_option}; -use snafu::ResultExt; +use snafu::{OptionExt, ResultExt}; use standalone::options::StandaloneOptions; use standalone::{StandaloneInformationExtension, StandaloneRepartitionProcedureFactory}; use tracing_appender::non_blocking::WorkerGuard; @@ -80,6 +81,58 @@ use crate::{App, create_resource_limit_metrics, error, log_versions, maybe_activ pub const APP_NAME: &str = "greptime-standalone"; +fn standalone_local_file_access( + storage: &StorageConfig, +) -> common_datasource::error::Result { + let data_home = configured_local_path(&storage.data_home)?; + let copy_root = match &storage.copy_root { + Some(root) => configured_local_path(root)?.with_context(|| { + common_datasource::error::InvalidLocalFileRootConfigSnafu { + root: root.clone(), + reason: "copy_root must be a local path or file URL".to_string(), + } + })?, + None => { + let Some(data_home) = &data_home else { + info!( + "SQL access to local files is disabled because storage.data_home is not a local path and storage.copy_root is unset" + ); + return Ok(LocalFileAccess::Disabled); + }; + data_home.join("copy") + } + }; + + let access = LocalFileAccess::sandboxed(©_root)?; + if let Some(data_home) = data_home { + let canonical_data_home = data_home.canonicalize().with_context(|_| { + common_datasource::error::InvalidLocalFileRootSnafu { + root: data_home.display().to_string(), + } + })?; + let canonical_copy_root = access.sandbox_root().with_context(|| { + common_datasource::error::InvalidLocalFileRootConfigSnafu { + root: copy_root.display().to_string(), + reason: "sandboxed local file access has no root".to_string(), + } + })?; + let default_copy_root = canonical_data_home.join("copy"); + let exposes_internal_files = canonical_data_home.starts_with(canonical_copy_root) + || (canonical_copy_root.starts_with(&canonical_data_home) + && !canonical_copy_root.starts_with(default_copy_root)); + if exposes_internal_files { + return common_datasource::error::InvalidLocalFileRootConfigSnafu { + root: copy_root.display().to_string(), + reason: "copy_root must not expose files in data_home outside data_home/copy" + .to_string(), + } + .fail(); + } + } + + Ok(access) +} + #[derive(Parser)] pub struct Command { #[clap(subcommand)] @@ -396,6 +449,9 @@ impl StartCommand { // Ensure the data_home directory exists. fs::create_dir_all(path::Path::new(data_home)) .context(error::CreateDirSnafu { dir: data_home })?; + let local_file_access = standalone_local_file_access(&dn_opts.storage) + .map_err(BoxedError::new) + .context(OtherSnafu)?; let metadata_dir = metadata_store_dir(data_home); let kv_backend = creator @@ -427,6 +483,7 @@ impl StartCommand { let mut builder = DatanodeBuilder::new(dn_opts, plugins.clone(), kv_backend.clone()); builder.with_cache_registry(layered_cache_registry.clone()); + builder.with_local_file_access(local_file_access.clone()); if let Some(writable) = creator.open_regions_writable_override { builder.with_open_regions_writable_override(writable); } @@ -595,7 +652,8 @@ impl StartCommand { node_manager.clone(), procedure_executor.clone(), process_manager, - ); + ) + .with_local_file_access(local_file_access); plugins::setup_frontend_plugins_post_build(&mut plugins, &plugin_opts, &fe_instance) .await @@ -966,7 +1024,7 @@ mod tests { use common_base::readable_size::ReadableSize; use common_config::ENV_VAR_SEP; use common_options::plugin_options::StandaloneFlag; - use common_test_util::temp_dir::create_named_temp_file; + use common_test_util::temp_dir::{create_named_temp_file, create_temp_dir}; use common_wal::config::DatanodeWalConfig; use frontend::frontend::FrontendOptions; use object_store::config::{FileConfig, GcsConfig}; @@ -975,6 +1033,73 @@ mod tests { use super::*; use crate::options::GlobalOptions; + #[test] + fn test_standalone_local_file_access_config() { + let data_home = create_temp_dir("standalone_copy_root"); + let storage = StorageConfig { + data_home: data_home.path().display().to_string(), + ..Default::default() + }; + let access = standalone_local_file_access(&storage).unwrap(); + assert_eq!( + access.sandbox_root().unwrap(), + data_home.path().join("copy").canonicalize().unwrap() + ); + + let remote_data_home = StorageConfig { + data_home: "s3://bucket/data".to_string(), + ..Default::default() + }; + assert!(matches!( + standalone_local_file_access(&remote_data_home).unwrap(), + LocalFileAccess::Disabled + )); + + let explicit_root = create_temp_dir("standalone_explicit_copy_root"); + let remote_with_explicit_root = StorageConfig { + data_home: "s3://bucket/data".to_string(), + copy_root: Some(explicit_root.path().display().to_string()), + ..Default::default() + }; + assert_eq!( + standalone_local_file_access(&remote_with_explicit_root) + .unwrap() + .sandbox_root() + .unwrap(), + explicit_root.path().canonicalize().unwrap() + ); + + let remote_copy_root = StorageConfig { + data_home: data_home.path().display().to_string(), + copy_root: Some("s3://bucket/copy".to_string()), + ..Default::default() + }; + assert!(matches!( + standalone_local_file_access(&remote_copy_root), + Err(common_datasource::error::Error::InvalidLocalFileRootConfig { .. }) + )); + + let exposes_internal = StorageConfig { + data_home: data_home.path().display().to_string(), + copy_root: Some(data_home.path().join("data").display().to_string()), + ..Default::default() + }; + assert!(matches!( + standalone_local_file_access(&exposes_internal), + Err(common_datasource::error::Error::InvalidLocalFileRootConfig { .. }) + )); + + let exposes_data_home = StorageConfig { + data_home: data_home.path().display().to_string(), + copy_root: Some(data_home.path().parent().unwrap().display().to_string()), + ..Default::default() + }; + assert!(matches!( + standalone_local_file_access(&exposes_data_home), + Err(common_datasource::error::Error::InvalidLocalFileRootConfig { .. }) + )); + } + #[tokio::test] async fn test_try_from_start_command_to_anymap() { let fe_opts = FrontendOptions { diff --git a/src/common/datasource/src/error.rs b/src/common/datasource/src/error.rs index 3c68bbc147..dbb0dec417 100644 --- a/src/common/datasource/src/error.rs +++ b/src/common/datasource/src/error.rs @@ -65,6 +65,62 @@ pub enum Error { location: Location, }, + #[snafu(display( + "SQL access to the local filesystem is disabled for '{}'; use S3, OSS, GCS, or AzBlob instead", + path + ))] + LocalFileAccessDisabled { + path: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "Local filesystem path '{}' is outside the configured copy root or is unsafe: {}; use a path relative to the copy root or use S3, OSS, GCS, or AzBlob", + path, + reason + ))] + LocalFileAccessDenied { + path: String, + reason: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display( + "Local filesystem path '{}' does not exist within the configured copy root", + path + ))] + LocalFilePathNotFound { + path: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display("Location must include a file or object name: '{}'", path))] + MissingObjectName { + path: String, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display("Invalid local filesystem root '{}'", root))] + InvalidLocalFileRoot { + root: String, + #[snafu(source)] + error: std::io::Error, + #[snafu(implicit)] + location: Location, + }, + + #[snafu(display("Invalid local filesystem root '{}': {}", root, reason))] + InvalidLocalFileRootConfig { + root: String, + reason: String, + #[snafu(implicit)] + location: Location, + }, + #[snafu(display("Failed to build backend"))] BuildBackend { #[snafu(source)] @@ -231,6 +287,12 @@ impl ErrorExt for Error { | UnsupportedFormat { .. } | InvalidConnection { .. } | InvalidUrl { .. } + | LocalFileAccessDisabled { .. } + | LocalFileAccessDenied { .. } + | LocalFilePathNotFound { .. } + | MissingObjectName { .. } + | InvalidLocalFileRoot { .. } + | InvalidLocalFileRootConfig { .. } | EmptyHostPath { .. } | InferSchema { .. } | ReadParquetSnafu { .. } diff --git a/src/common/datasource/src/lister.rs b/src/common/datasource/src/lister.rs index 97bbadfe12..2ffef475f7 100644 --- a/src/common/datasource/src/lister.rs +++ b/src/common/datasource/src/lister.rs @@ -72,7 +72,7 @@ impl Lister { // make sure this file exists let _ = self.object_store.stat(filename).await.with_context(|_| { error::ListObjectsSnafu { - path: format!("{}{}", &self.root, filename), + path: self.root.clone(), } })?; diff --git a/src/common/datasource/src/object_store.rs b/src/common/datasource/src/object_store.rs index 56eaf3968b..97bf779413 100644 --- a/src/common/datasource/src/object_store.rs +++ b/src/common/datasource/src/object_store.rs @@ -19,9 +19,13 @@ pub mod oss; pub mod s3; use std::collections::HashMap; +use std::path::{Component, Path, PathBuf}; +use std::sync::Arc; +use common_telemetry::debug; use lazy_static::lazy_static; use object_store::ObjectStore; +use object_store::secure_fs::SecureFsRoot; use regex::Regex; use snafu::{OptionExt, ResultExt}; use url::{ParseError, Url}; @@ -35,16 +39,244 @@ use crate::object_store::oss::build_oss_backend; use crate::util::find_dir_and_filename; pub const FS_SCHEMA: &str = "FS"; +pub const FILE_SCHEMA: &str = "FILE"; pub const S3_SCHEMA: &str = "S3"; pub const OSS_SCHEMA: &str = "OSS"; pub const GCS_SCHEMA: &str = "GCS"; pub const AZBLOB_SCHEMA: &str = "AZBLOB"; +/// An object store rooted at the target's parent, together with the optional +/// target path relative to that root. +pub struct BuiltBackend { + pub object_store: ObjectStore, + pub object_path: Option, +} + +/// Controls whether SQL paths may access the local filesystem. +#[derive(Clone, Debug, Default)] +pub enum LocalFileAccess { + /// Local filesystem paths are rejected. + #[default] + Disabled, + /// Local filesystem paths are confined to a server-configured root. + Sandboxed { root: LocalFileRoot }, +} + +/// An opened server-controlled root for sandboxed SQL file access. +#[derive(Clone, Debug)] +pub struct LocalFileRoot { + root: Arc, + configured_path: Arc, +} + +impl LocalFileAccess { + /// Creates a sandbox rooted at a server-controlled local directory. + pub fn sandboxed(root: impl AsRef) -> Result { + let root_path = root.as_ref(); + let configured_path = + std::path::absolute(root_path).with_context(|_| error::InvalidLocalFileRootSnafu { + root: root_path.display().to_string(), + })?; + let root = + SecureFsRoot::open(root_path).with_context(|_| error::InvalidLocalFileRootSnafu { + root: root_path.display().to_string(), + })?; + Ok(Self::Sandboxed { + root: LocalFileRoot { + root: Arc::new(root), + configured_path: Arc::new(configured_path), + }, + }) + } + + /// Returns the canonical path of the configured sandbox root. + pub fn sandbox_root(&self) -> Option<&Path> { + match self { + Self::Disabled => None, + Self::Sandboxed { root } => Some(root.root.path()), + } + } + + fn authorize(&self, location: &str, path: &Path, trailing_slash: bool) -> Result { + let LocalFileAccess::Sandboxed { root } = self else { + return error::LocalFileAccessDisabledSnafu { + path: location.to_string(), + } + .fail(); + }; + + let path = normalize_untrusted_path(path).map_err(|reason| { + error::LocalFileAccessDeniedSnafu { + path: location.to_string(), + reason, + } + .build() + })?; + let relative = if path.is_absolute() { + strip_local_prefix(&path, root.configured_path.as_path()) + .or_else(|| strip_local_prefix(&path, root.root.path())) + .ok_or_else(|| { + error::LocalFileAccessDeniedSnafu { + path: location.to_string(), + reason: "absolute path is outside the configured copy root".to_string(), + } + .build() + })? + } else { + path.as_path() + }; + + let mut authorized = relative + .components() + .filter_map(|component| match component { + Component::CurDir => None, + Component::Normal(value) => Some(value.to_string_lossy().into_owned()), + _ => None, + }) + .collect::>() + .join("/"); + if trailing_slash && !authorized.is_empty() { + authorized.push('/'); + } + Ok(authorized) + } + + async fn open_backend_root( + &self, + location: &str, + relative_root: &str, + create: bool, + ) -> Result { + let LocalFileAccess::Sandboxed { root } = self else { + return error::LocalFileAccessDisabledSnafu { + path: location.to_string(), + } + .fail(); + }; + + let root = root.root.clone(); + let relative_root = relative_root.trim_matches('/').to_string(); + common_runtime::spawn_blocking_global(move || { + if create { + root.create_subdir(relative_root) + } else { + root.open_subdir(relative_root) + } + }) + .await + .context(error::JoinHandleSnafu)? + .map_err(|error| { + debug!( + "Failed to open an authorized local SQL path inside the copy root, path: {location}, error: {error:?}" + ); + if error.kind() == std::io::ErrorKind::NotFound { + return error::LocalFilePathNotFoundSnafu { path: location }.build(); + } + error::LocalFileAccessDeniedSnafu { + path: location.to_string(), + reason: "path could not be safely resolved within the configured copy root" + .to_string(), + } + .build() + }) + } +} + +/// Converts a configured location into a local path. +/// +/// Bare paths and `file://` URLs are local. Other URL schemes return `None`. +pub fn configured_local_path(location: &str) -> Result> { + #[cfg(windows)] + if Path::new(location).is_absolute() { + return Ok(Some(PathBuf::from(location))); + } + + let (schema, _, path) = parse_url(location)?; + match schema.to_uppercase().as_str() { + FS_SCHEMA => Ok(Some(PathBuf::from(path))), + FILE_SCHEMA => { + let url = Url::parse(location).context(error::InvalidUrlSnafu { url: location })?; + url.to_file_path().map(Some).map_err(|_| { + error::InvalidLocalFileRootConfigSnafu { + root: location.to_string(), + reason: "file URL must contain a local absolute path".to_string(), + } + .build() + }) + } + _ => Ok(None), + } +} + +fn strip_local_prefix<'a>(path: &'a Path, prefix: &Path) -> Option<&'a Path> { + #[cfg(not(windows))] + { + path.strip_prefix(prefix).ok() + } + + #[cfg(windows)] + { + let mut path_components = path.components(); + for prefix_component in prefix.components() { + let path_component = path_components.next()?; + if !windows_component_eq(path_component, prefix_component) { + return None; + } + } + Some(path_components.as_path()) + } +} + +#[cfg(windows)] +fn windows_component_eq(left: Component<'_>, right: Component<'_>) -> bool { + match (left, right) { + (Component::Prefix(left), Component::Prefix(right)) => { + windows_os_str_eq(left.as_os_str(), right.as_os_str()) + } + (Component::Normal(left), Component::Normal(right)) => windows_os_str_eq(left, right), + (Component::RootDir, Component::RootDir) + | (Component::CurDir, Component::CurDir) + | (Component::ParentDir, Component::ParentDir) => true, + _ => false, + } +} + +#[cfg(windows)] +fn windows_os_str_eq(left: &std::ffi::OsStr, right: &std::ffi::OsStr) -> bool { + use std::os::windows::ffi::OsStrExt; + + fn ascii_lowercase(value: u16) -> u16 { + if (u16::from(b'A')..=u16::from(b'Z')).contains(&value) { + value + u16::from(b'a' - b'A') + } else { + value + } + } + + left.encode_wide() + .map(ascii_lowercase) + .eq(right.encode_wide().map(ascii_lowercase)) +} + +fn normalize_untrusted_path(path: &Path) -> std::result::Result { + let mut normalized = PathBuf::new(); + for component in path.components() { + match component { + Component::Prefix(prefix) => normalized.push(prefix.as_os_str()), + Component::RootDir => normalized.push(Path::new(std::path::MAIN_SEPARATOR_STR)), + Component::CurDir => {} + Component::Normal(value) => normalized.push(value), + Component::ParentDir => return Err("'..' path components are not allowed".to_string()), + } + } + Ok(normalized) +} + /// Returns `(schema, Option, path)` pub fn parse_url(url: &str) -> Result<(String, Option, String)> { #[cfg(windows)] { - // On Windows, the url may start with `C:/`. + // On Windows, the URL may start with `C:/` or `C:\`. if handle_windows_path(url).is_some() { return Ok((FS_SCHEMA.to_string(), None, url.to_string())); } @@ -63,47 +295,134 @@ pub fn parse_url(url: &str) -> Result<(String, Option, String)> { } } -pub fn build_backend(url: &str, connection: &HashMap) -> Result { - let (schema, host, path) = parse_url(url)?; - let (root, _) = find_dir_and_filename(&path); +pub async fn build_backend( + url: &str, + connection: &HashMap, + local_file_access: &LocalFileAccess, +) -> Result { + Ok( + build_backend_inner(url, connection, local_file_access, false, false) + .await? + .object_store, + ) +} - match schema.to_uppercase().as_str() { +/// Builds a backend and returns the target path relative to the backend root. +pub async fn build_backend_with_path( + url: &str, + connection: &HashMap, + local_file_access: &LocalFileAccess, +) -> Result { + build_backend_inner(url, connection, local_file_access, false, false).await +} + +/// Builds a backend for an operation that may create the target directory. +pub async fn build_backend_for_write( + url: &str, + connection: &HashMap, + local_file_access: &LocalFileAccess, +) -> Result { + Ok( + build_backend_inner(url, connection, local_file_access, true, false) + .await? + .object_store, + ) +} + +/// Builds a writable backend and returns the target path relative to the backend root. +pub async fn build_backend_for_write_with_path( + url: &str, + connection: &HashMap, + local_file_access: &LocalFileAccess, +) -> Result { + build_backend_inner(url, connection, local_file_access, true, true).await +} + +async fn build_backend_inner( + url: &str, + connection: &HashMap, + local_file_access: &LocalFileAccess, + create_local_root: bool, + require_object_path: bool, +) -> Result { + let (schema, host, path) = parse_url(url)?; + let normalized_schema = schema.to_uppercase(); + + if normalized_schema == FS_SCHEMA || normalized_schema == FILE_SCHEMA { + let (local_path, trailing_slash) = if normalized_schema == FILE_SCHEMA { + let url = Url::parse(url).context(error::InvalidUrlSnafu { url })?; + let path = url.to_file_path().map_err(|_| { + error::LocalFileAccessDeniedSnafu { + path: url.to_string(), + reason: "file URL must contain a local absolute path".to_string(), + } + .build() + })?; + (path, url.path().ends_with('/')) + } else { + ( + PathBuf::from(&path), + path.ends_with('/') || cfg!(windows) && path.ends_with(std::path::MAIN_SEPARATOR), + ) + }; + let authorized = local_file_access.authorize(url, &local_path, trailing_slash)?; + let (root, object_path) = find_dir_and_filename(&authorized); + if require_object_path && object_path.is_none() { + return error::MissingObjectNameSnafu { + path: url.to_string(), + } + .fail(); + } + let root = local_file_access + .open_backend_root(url, &root, create_local_root) + .await?; + return Ok(BuiltBackend { + object_store: build_fs_backend(&root)?, + object_path, + }); + } + + let (root, object_path) = find_dir_and_filename(&path); + + let object_store = match normalized_schema.as_str() { S3_SCHEMA => { let host = host.context(error::EmptyHostPathSnafu { url: url.to_string(), })?; - Ok(build_s3_backend(&host, &root, connection)?) + build_s3_backend(&host, &root, connection)? } OSS_SCHEMA => { let host = host.context(error::EmptyHostPathSnafu { url: url.to_string(), })?; - Ok(build_oss_backend(&host, &root, connection)?) + build_oss_backend(&host, &root, connection)? } GCS_SCHEMA => { let host = host.context(error::EmptyHostPathSnafu { url: url.to_string(), })?; - Ok(build_gcs_backend(&host, &root, connection)?) + build_gcs_backend(&host, &root, connection)? } AZBLOB_SCHEMA => { let host = host.context(error::EmptyHostPathSnafu { url: url.to_string(), })?; - Ok(build_azblob_backend(&host, &root, connection)?) + build_azblob_backend(&host, &root, connection)? } - FS_SCHEMA => Ok(build_fs_backend(&root)?), - _ => error::UnsupportedBackendProtocolSnafu { protocol: schema, url, } - .fail(), - } + .fail()?, + }; + Ok(BuiltBackend { + object_store, + object_path, + }) } lazy_static! { - static ref DISK_SYMBOL_PATTERN: Regex = Regex::new("^([A-Za-z]:/)").unwrap(); + static ref DISK_SYMBOL_PATTERN: Regex = Regex::new(r"^([A-Za-z]:[/\\])").unwrap(); } pub fn handle_windows_path(url: &str) -> Option { @@ -114,7 +433,19 @@ pub fn handle_windows_path(url: &str) -> Option { #[cfg(test)] mod tests { - use super::handle_windows_path; + use std::collections::HashMap; + use std::fs; + + use common_error::ext::{ErrorExt, RetryHint}; + use common_error::status_code::StatusCode; + use common_test_util::temp_dir::create_temp_dir; + use url::Url; + + use super::{ + LocalFileAccess, build_backend, build_backend_for_write, build_backend_for_write_with_path, + build_backend_with_path, handle_windows_path, + }; + use crate::error::Error; #[test] fn test_handle_windows_path() { @@ -122,7 +453,315 @@ mod tests { handle_windows_path("C:/to/path/file"), Some("C:/".to_string()) ); + assert_eq!( + handle_windows_path(r"C:\to\path\file"), + Some(r"C:\".to_string()) + ); assert_eq!(handle_windows_path("https://google.com"), None); assert_eq!(handle_windows_path("s3://bucket/path/to"), None); } + + #[cfg(windows)] + #[test] + fn test_windows_local_path_detection_and_prefix() { + use std::path::{Path, PathBuf}; + + let location = r"C:\gtdata"; + assert_eq!( + super::configured_local_path(location).unwrap(), + Some(PathBuf::from(location)) + ); + assert_eq!( + super::parse_url(location).unwrap(), + ("FS".to_string(), None, location.to_string()) + ); + assert_eq!( + super::strip_local_prefix( + Path::new(r"c:\Data\Copy\nested\data.parquet"), + Path::new(r"C:\data\copy"), + ), + Some(Path::new(r"nested\data.parquet")) + ); + } + + #[tokio::test] + async fn test_local_file_access_policy() { + let data_home = create_temp_dir("local_file_access_policy"); + let copy_root = data_home.path().join("copy"); + let internal_dir = data_home.path().join("data"); + fs::create_dir_all(&internal_dir).unwrap(); + fs::write(internal_dir.join("secret"), "secret").unwrap(); + + let access = LocalFileAccess::sandboxed(©_root).unwrap(); + let connection = HashMap::new(); + + let store = build_backend_for_write("nested/data.txt", &connection, &access) + .await + .unwrap(); + store.write("data.txt", "first").await.unwrap(); + store.write("data.txt", "second").await.unwrap(); + assert_eq!( + fs::read_to_string(copy_root.join("nested/data.txt")).unwrap(), + "second" + ); + + let missing = copy_root.join("missing/directory"); + let error = build_backend("missing/directory/data.txt", &connection, &access) + .await + .unwrap_err(); + assert!(matches!(&error, Error::LocalFilePathNotFound { .. })); + assert_eq!(error.status_code(), StatusCode::InvalidArguments); + assert_eq!(error.retry_hint(), RetryHint::NonRetryable); + assert!( + error.to_string().contains("does not exist"), + "unexpected error: {error}" + ); + assert!(!missing.exists()); + + let absolute = copy_root.join("nested/data.txt"); + let store = build_backend(absolute.to_str().unwrap(), &connection, &access) + .await + .unwrap(); + assert_eq!(store.read("data.txt").await.unwrap().to_vec(), b"second"); + + let file_url = Url::from_file_path(&absolute).unwrap().to_string(); + let store = build_backend(&file_url, &connection, &access) + .await + .unwrap(); + assert_eq!(store.read("data.txt").await.unwrap().to_vec(), b"second"); + + let internal_file = internal_dir.join("secret"); + assert!(matches!( + build_backend(internal_file.to_str().unwrap(), &connection, &access).await, + Err(Error::LocalFileAccessDenied { .. }) + )); + assert!( + build_backend("../escape/data.txt", &connection, &access) + .await + .is_err() + ); + + let outside = data_home.path().join("outside/new"); + assert!( + build_backend(outside.to_str().unwrap(), &connection, &access) + .await + .is_err() + ); + assert!(!outside.parent().unwrap().exists()); + + let prefix_escape = data_home.path().join("copy-not-the-root/new"); + assert!(matches!( + build_backend(prefix_escape.to_str().unwrap(), &connection, &access).await, + Err(Error::LocalFileAccessDenied { .. }) + )); + assert!(!prefix_escape.parent().unwrap().exists()); + + let disabled = LocalFileAccess::Disabled; + let error = build_backend(internal_file.to_str().unwrap(), &connection, &disabled) + .await + .unwrap_err(); + assert!(matches!(&error, Error::LocalFileAccessDisabled { .. })); + assert_eq!(error.status_code(), StatusCode::InvalidArguments); + assert_eq!(error.retry_hint(), RetryHint::NonRetryable); + assert!(matches!( + build_backend(&file_url, &connection, &disabled).await, + Err(Error::LocalFileAccessDisabled { .. }) + )); + assert!(matches!( + build_backend(outside.to_str().unwrap(), &connection, &disabled).await, + Err(Error::LocalFileAccessDisabled { .. }) + )); + assert!(!outside.parent().unwrap().exists()); + } + + #[tokio::test] + async fn test_file_url_returns_decoded_backend_relative_path() { + let temp_dir = create_temp_dir("file_url_backend_relative_path"); + let copy_root = temp_dir.path().join("copy root"); + let file = copy_root.join("nested dir/data file.txt"); + fs::create_dir_all(file.parent().unwrap()).unwrap(); + fs::write(&file, "data").unwrap(); + + let location = Url::from_file_path(&file).unwrap().to_string(); + assert!(location.contains("%20")); + let access = LocalFileAccess::sandboxed(©_root).unwrap(); + let backend = build_backend_with_path(&location, &HashMap::new(), &access) + .await + .unwrap(); + + assert_eq!(backend.object_path.as_deref(), Some("data file.txt")); + assert_eq!( + backend + .object_store + .read(backend.object_path.as_deref().unwrap()) + .await + .unwrap() + .to_vec(), + b"data" + ); + } + + #[tokio::test] + async fn test_write_with_path_rejects_directory_before_creation() { + let temp_dir = create_temp_dir("write_with_path_rejects_directory"); + let copy_root = temp_dir.path().join("copy"); + let target = copy_root.join("new directory"); + let location = Url::from_directory_path(&target).unwrap().to_string(); + let access = LocalFileAccess::sandboxed(©_root).unwrap(); + + let result = build_backend_for_write_with_path(&location, &HashMap::new(), &access).await; + + assert!(matches!(result, Err(Error::MissingObjectName { .. }))); + assert!(!target.exists()); + + let result = build_backend_for_write_with_path( + &location, + &HashMap::new(), + &LocalFileAccess::Disabled, + ) + .await; + assert!(matches!(result, Err(Error::LocalFileAccessDisabled { .. }))); + assert!(!target.exists()); + + let relative_target = copy_root.join("relative directory"); + let result = + build_backend_for_write_with_path("relative directory/", &HashMap::new(), &access) + .await; + assert!(matches!(result, Err(Error::MissingObjectName { .. }))); + assert!(!relative_target.exists()); + + let allowed_directory = copy_root.join("allowed directory"); + build_backend_for_write("allowed directory/", &HashMap::new(), &access) + .await + .unwrap(); + assert!(allowed_directory.is_dir()); + } + + #[cfg(windows)] + #[tokio::test] + async fn test_windows_backslash_path_returns_backend_relative_path() { + let temp_dir = create_temp_dir("windows_backend_relative_path"); + let copy_root = temp_dir.path().join("copy"); + let directory = copy_root.join("nested"); + let file = directory.join("data.txt"); + fs::create_dir_all(&directory).unwrap(); + fs::write(&file, "data").unwrap(); + + let access = LocalFileAccess::sandboxed(©_root).unwrap(); + let connection = HashMap::new(); + let file_location = file.to_str().unwrap(); + assert!(file_location.contains('\\')); + let backend = build_backend_with_path(file_location, &connection, &access) + .await + .unwrap(); + assert_eq!(backend.object_path.as_deref(), Some("data.txt")); + assert_eq!( + backend + .object_store + .read(backend.object_path.as_deref().unwrap()) + .await + .unwrap() + .to_vec(), + b"data" + ); + + let new_directory = copy_root.join("new directory"); + let directory_location = format!("{}\\", new_directory.display()); + assert!(matches!( + build_backend_for_write_with_path(&directory_location, &connection, &access).await, + Err(Error::MissingObjectName { .. }) + )); + assert!(!new_directory.exists()); + } + + #[tokio::test] + async fn test_object_storage_ignores_local_file_policy() { + let cases = [ + ( + "s3://bucket/path/data%20file.parquet", + HashMap::from([ + ("region".to_string(), "us-east-1".to_string()), + ("disable_ec2_metadata".to_string(), "true".to_string()), + ]), + "data%20file.parquet", + ), + ( + "oss://bucket/path/file.parquet", + HashMap::from([ + ("endpoint".to_string(), "http://oss.example.com".to_string()), + ("allow_anonymous".to_string(), "true".to_string()), + ]), + "file.parquet", + ), + ( + "gcs://bucket/path/file.parquet", + HashMap::from([( + "endpoint".to_string(), + "http://storage.example.com".to_string(), + )]), + "file.parquet", + ), + ( + "azblob://container/path/file.parquet", + HashMap::from([ + ( + "endpoint".to_string(), + "http://storage.example.com".to_string(), + ), + ("account_name".to_string(), "test".to_string()), + ]), + "file.parquet", + ), + ]; + for (location, connection, expected_path) in cases { + let backend = build_backend_for_write_with_path( + location, + &connection, + &LocalFileAccess::Disabled, + ) + .await + .unwrap(); + assert_eq!(backend.object_path.as_deref(), Some(expected_path)); + } + } + + #[cfg(unix)] + #[tokio::test] + async fn test_local_file_access_rejects_symlink_escape() { + use std::os::unix::fs::symlink; + + let temp_dir = create_temp_dir("local_file_access_symlink"); + let copy_root = temp_dir.path().join("copy"); + let outside = temp_dir.path().join("outside"); + fs::create_dir_all(©_root).unwrap(); + fs::create_dir_all(&outside).unwrap(); + fs::write(outside.join("secret"), "secret").unwrap(); + symlink(&outside, copy_root.join("escape")).unwrap(); + symlink(outside.join("secret"), copy_root.join("secret-link")).unwrap(); + + let access = LocalFileAccess::sandboxed(©_root).unwrap(); + let connection = HashMap::new(); + + assert!( + build_backend("escape/secret", &connection, &access) + .await + .is_err() + ); + assert!( + build_backend_for_write("escape/new", &connection, &access) + .await + .is_err() + ); + assert!(!outside.join("new").exists()); + + let store = build_backend("secret-link", &connection, &access) + .await + .unwrap(); + assert!(store.read("secret-link").await.is_err()); + assert!(store.write("secret-link", "overwritten").await.is_err()); + assert_eq!( + fs::read_to_string(outside.join("secret")).unwrap(), + "secret" + ); + } } diff --git a/src/common/datasource/src/object_store/fs.rs b/src/common/datasource/src/object_store/fs.rs index 0f537b1e69..3f3c060b60 100644 --- a/src/common/datasource/src/object_store/fs.rs +++ b/src/common/datasource/src/object_store/fs.rs @@ -13,16 +13,11 @@ // limitations under the License. use object_store::ObjectStore; -use object_store::services::Fs; +use object_store::secure_fs::SecureFsRoot; use object_store::util::with_instrument_layers; -use snafu::ResultExt; -use crate::error::{BuildBackendSnafu, Result}; +use crate::error::Result; -pub fn build_fs_backend(root: &str) -> Result { - let builder = Fs::default(); - let object_store = ObjectStore::new(builder.root(root)) - .context(BuildBackendSnafu)? - .finish(); - Ok(with_instrument_layers(object_store, true)) +pub fn build_fs_backend(root: &SecureFsRoot) -> Result { + Ok(with_instrument_layers(root.build_operator(), true)) } diff --git a/src/datanode/Cargo.toml b/src/datanode/Cargo.toml index 1bb5cdd92a..6b78858bd2 100644 --- a/src/datanode/Cargo.toml +++ b/src/datanode/Cargo.toml @@ -20,6 +20,7 @@ client.workspace = true common-base.workspace = true common-catalog.workspace = true common-config.workspace = true +common-datasource.workspace = true common-error.workspace = true common-function.workspace = true common-greptimedb-telemetry.workspace = true diff --git a/src/datanode/src/config.rs b/src/datanode/src/config.rs index b757c95121..e627d8dc24 100644 --- a/src/datanode/src/config.rs +++ b/src/datanode/src/config.rs @@ -39,6 +39,11 @@ use servers::http::HttpOptions; pub struct StorageConfig { /// The working directory of database pub data_home: String, + /// Root directory for standalone SQL access to local files. + /// + /// Defaults to `/copy` when `data_home` is a local path. + /// Distributed deployments always disable SQL access to local files. + pub copy_root: Option, #[serde(flatten)] pub store: ObjectStoreConfig, /// Object storage providers @@ -56,6 +61,7 @@ impl Default for StorageConfig { fn default() -> Self { Self { data_home: DEFAULT_DATA_HOME.to_string(), + copy_root: None, store: ObjectStoreConfig::default(), providers: vec![], } diff --git a/src/datanode/src/datanode.rs b/src/datanode/src/datanode.rs index f72bb04da7..03fcef4926 100644 --- a/src/datanode/src/datanode.rs +++ b/src/datanode/src/datanode.rs @@ -19,6 +19,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use common_base::Plugins; +use common_datasource::object_store::LocalFileAccess; use common_error::ext::BoxedError; use common_greptimedb_telemetry::GreptimeDBTelemetryTask; use common_meta::cache::{LayeredCacheRegistry, SchemaCacheRef, TableSchemaCacheRef}; @@ -164,6 +165,7 @@ pub struct DatanodeBuilder { cache_registry: Option>, topic_stats_reporter: Option>, open_regions_writable_override: Option, + local_file_access: LocalFileAccess, #[cfg(feature = "enterprise")] extension_range_provider_factory: Option, } @@ -178,6 +180,7 @@ impl DatanodeBuilder { kv_backend, cache_registry: None, open_regions_writable_override: None, + local_file_access: LocalFileAccess::Disabled, #[cfg(feature = "enterprise")] extension_range_provider_factory: None, topic_stats_reporter: None, @@ -198,6 +201,11 @@ impl DatanodeBuilder { self } + pub fn with_local_file_access(&mut self, local_file_access: LocalFileAccess) -> &mut Self { + self.local_file_access = local_file_access; + self + } + pub fn kv_backend(&self) -> &KvBackendRef { &self.kv_backend } @@ -528,6 +536,7 @@ impl DatanodeBuilder { let file_engine = FileRegionEngine::new( file_engine_config, object_store_manager.default_object_store().clone(), // TODO: implement custom storage for file engine + self.local_file_access.clone(), ); Ok(vec![ diff --git a/src/file-engine/src/engine.rs b/src/file-engine/src/engine.rs index 07d201b29a..fae7a1f2f0 100644 --- a/src/file-engine/src/engine.rs +++ b/src/file-engine/src/engine.rs @@ -19,6 +19,7 @@ use std::sync::{Arc, RwLock}; use api::region::RegionResponse; use async_trait::async_trait; use common_catalog::consts::FILE_ENGINE; +use common_datasource::object_store::LocalFileAccess; use common_error::ext::BoxedError; use common_recordbatch::SendableRecordBatchStream; use common_telemetry::{error, info}; @@ -48,9 +49,13 @@ pub struct FileRegionEngine { } impl FileRegionEngine { - pub fn new(_config: EngineConfig, object_store: ObjectStore) -> Self { + pub fn new( + _config: EngineConfig, + object_store: ObjectStore, + local_file_access: LocalFileAccess, + ) -> Self { Self { - inner: Arc::new(EngineInner::new(object_store)), + inner: Arc::new(EngineInner::new(object_store, local_file_access)), } } @@ -64,7 +69,8 @@ impl FileRegionEngine { .await .context(RegionNotFoundSnafu { region_id }) .map_err(BoxedError::new)? - .query(request) + .query(request, &self.inner.local_file_access) + .await .map_err(BoxedError::new) } } @@ -182,6 +188,8 @@ struct EngineInner { region_mutex: Mutex<()>, object_store: ObjectStore, + + local_file_access: LocalFileAccess, } type EngineInnerRef = Arc; @@ -205,11 +213,12 @@ fn ensure_region_requirements( } impl EngineInner { - fn new(object_store: ObjectStore) -> Self { + fn new(object_store: ObjectStore, local_file_access: LocalFileAccess) -> Self { Self { regions: RwLock::new(HashMap::new()), region_mutex: Mutex::new(()), object_store, + local_file_access, } } diff --git a/src/file-engine/src/query.rs b/src/file-engine/src/query.rs index 62ca3c7b66..0923bf2cfd 100644 --- a/src/file-engine/src/query.rs +++ b/src/file-engine/src/query.rs @@ -19,7 +19,7 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use common_datasource::object_store::build_backend; +use common_datasource::object_store::{LocalFileAccess, build_backend}; use common_recordbatch::adapter::RecordBatchMetrics; use common_recordbatch::error::{self as recordbatch_error, Result as RecordBatchResult}; use common_recordbatch::{ @@ -41,8 +41,14 @@ use crate::error::{BuildBackendSnafu, ProjectSchemaSnafu, ProjectionOutOfBoundsS use crate::region::FileRegion; impl FileRegion { - pub fn query(&self, request: ScanRequest) -> Result { - let store = build_backend(&self.url, &self.options).context(BuildBackendSnafu)?; + pub async fn query( + &self, + request: ScanRequest, + local_file_access: &LocalFileAccess, + ) -> Result { + let store = build_backend(&self.url, &self.options, local_file_access) + .await + .context(BuildBackendSnafu)?; let projection = request.projection.as_deref(); let file_projection = self.projection_pushdown_to_file(projection)?; diff --git a/src/file-engine/src/region.rs b/src/file-engine/src/region.rs index c132f4c6b8..6f7e4bb668 100644 --- a/src/file-engine/src/region.rs +++ b/src/file-engine/src/region.rs @@ -107,7 +107,11 @@ impl FileRegion { mod tests { use std::assert_matches; + use common_datasource::object_store::LocalFileAccess; + use common_error::ext::{ErrorExt, RetryHint}; + use common_error::status_code::StatusCode; use store_api::region_request::PathType; + use store_api::storage::ScanRequest; use super::*; use crate::error::Error; @@ -155,6 +159,41 @@ mod tests { assert_matches!(err, Error::ManifestExists { .. }); } + #[tokio::test] + async fn test_persisted_local_region_rejected_when_disabled() { + let (_dir, object_store) = new_test_object_store("test_disabled_local_region"); + let request = RegionCreateRequest { + engine: "file".to_string(), + column_metadatas: new_test_column_metadata(), + primary_key: vec![1], + options: new_test_options(), + table_dir: "disabled_local_region/".to_string(), + path_type: PathType::Bare, + partition_expr_json: Some("".to_string()), + requirements: Default::default(), + }; + let region = FileRegion::create(RegionId::new(1, 0), request, &object_store) + .await + .unwrap(); + + let error = match region + .query(ScanRequest::default(), &LocalFileAccess::Disabled) + .await + { + Ok(_) => panic!("local file query must be rejected"), + Err(error) => error, + }; + assert_matches!( + &error, + Error::BuildBackend { + source: common_datasource::error::Error::LocalFileAccessDisabled { .. }, + .. + } + ); + assert_eq!(error.status_code(), StatusCode::InvalidArguments); + assert_eq!(error.retry_hint(), RetryHint::NonRetryable); + } + #[tokio::test] async fn test_open_region() { let (_dir, object_store) = new_test_object_store("test_open_region"); diff --git a/src/flow/Cargo.toml b/src/flow/Cargo.toml index ee4a7f830f..e5603fd8d8 100644 --- a/src/flow/Cargo.toml +++ b/src/flow/Cargo.toml @@ -20,6 +20,7 @@ chrono.workspace = true client.workspace = true common-base.workspace = true common-config.workspace = true +common-datasource.workspace = true common-decimal.workspace = true common-error.workspace = true common-frontend.workspace = true diff --git a/src/flow/src/server.rs b/src/flow/src/server.rs index c44c9e94a3..c9a24a4880 100644 --- a/src/flow/src/server.rs +++ b/src/flow/src/server.rs @@ -22,6 +22,7 @@ use api::v1::{RowDeleteRequests, RowInsertRequests}; use cache::{PARTITION_INFO_CACHE_NAME, TABLE_FLOWNODE_SET_CACHE_NAME, TABLE_ROUTE_CACHE_NAME}; use catalog::CatalogManagerRef; use common_base::Plugins; +use common_datasource::object_store::LocalFileAccess; use common_error::ext::BoxedError; use common_meta::cache::{LayeredCacheRegistryRef, TableFlownodeSetCacheRef, TableRouteCacheRef}; use common_meta::key::TableMetadataManagerRef; @@ -632,6 +633,7 @@ impl FrontendInvoker { partition_manager, None, origin_frontend_addr, + LocalFileAccess::Disabled, )); let invoker = FrontendInvoker::new(inserter, deleter, statement_executor); diff --git a/src/frontend/src/instance/builder.rs b/src/frontend/src/instance/builder.rs index 32395efd2d..42d234fae4 100644 --- a/src/frontend/src/instance/builder.rs +++ b/src/frontend/src/instance/builder.rs @@ -19,6 +19,7 @@ use cache::{PARTITION_INFO_CACHE_NAME, TABLE_FLOWNODE_SET_CACHE_NAME, TABLE_ROUT use catalog::CatalogManagerRef; use catalog::process_manager::ProcessManagerRef; use common_base::Plugins; +use common_datasource::object_store::LocalFileAccess; use common_event_recorder::EventRecorderImpl; use common_meta::cache::{LayeredCacheRegistryRef, TableRouteCacheRef}; use common_meta::cache_invalidator::{CacheInvalidatorRef, DummyCacheInvalidator}; @@ -65,6 +66,7 @@ pub struct FrontendBuilder { plugins: Option, procedure_executor: ProcedureExecutorRef, process_manager: ProcessManagerRef, + local_file_access: LocalFileAccess, } impl FrontendBuilder { @@ -88,6 +90,7 @@ impl FrontendBuilder { plugins: None, procedure_executor, process_manager, + local_file_access: LocalFileAccess::Disabled, } } @@ -134,6 +137,11 @@ impl FrontendBuilder { } } + pub fn with_local_file_access(mut self, local_file_access: LocalFileAccess) -> Self { + self.local_file_access = local_file_access; + self + } + pub fn options(&self) -> &FrontendOptions { &self.options } @@ -271,6 +279,7 @@ impl FrontendBuilder { partition_manager, Some(process_manager.clone()), frontend_peer_addr.clone(), + self.local_file_access, ); let statement_executor = diff --git a/src/object-store/Cargo.toml b/src/object-store/Cargo.toml index 18b246ec8e..8e6febf32e 100644 --- a/src/object-store/Cargo.toml +++ b/src/object-store/Cargo.toml @@ -15,10 +15,12 @@ testing = ["derive_builder"] [dependencies] async-trait.workspace = true bytes.workspace = true +cap-std.workspace = true chrono.workspace = true common-base.workspace = true common-error.workspace = true common-macro.workspace = true +common-runtime.workspace = true common-telemetry.workspace = true datafusion_object_store.workspace = true derive_builder = { workspace = true, optional = true } diff --git a/src/object-store/src/lib.rs b/src/object-store/src/lib.rs index f1f8b59082..a9257c2423 100644 --- a/src/object-store/src/lib.rs +++ b/src/object-store/src/lib.rs @@ -24,6 +24,7 @@ pub mod factory; pub mod layers; pub mod manager; mod metrics; +pub mod secure_fs; pub mod test_util; pub mod util; diff --git a/src/object-store/src/secure_fs.rs b/src/object-store/src/secure_fs.rs new file mode 100644 index 0000000000..448949ea5e --- /dev/null +++ b/src/object-store/src/secure_fs.rs @@ -0,0 +1,682 @@ +// Copyright 2023 Greptime Team +// +// Licensed 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. + +//! A capability-based filesystem backend for untrusted object paths. + +use std::path::{Component, Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::vec::IntoIter; +use std::{fmt, io}; + +use cap_std::ambient_authority; +use cap_std::fs::{Dir, DirEntry, OpenOptions, ReadDir}; +use opendal::raw::*; +use opendal::{ + Buffer, Capability, EntryMode, Error, ErrorKind, Metadata, Operator, OperatorBuilder, Result, +}; +use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}; + +const LIST_BATCH_SIZE: usize = 128; + +/// An opened filesystem root that confines all descendant path resolution. +#[derive(Clone)] +pub struct SecureFsRoot { + dir: Arc, + path: Arc, +} + +impl fmt::Debug for SecureFsRoot { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("SecureFsRoot") + .field("path", &self.path) + .finish_non_exhaustive() + } +} + +impl SecureFsRoot { + /// Creates and opens `path` using ambient authority. + /// + /// Callers must only pass a server-controlled path. + pub fn open(path: impl AsRef) -> io::Result { + let path = path.as_ref(); + std::fs::create_dir_all(path)?; + if std::fs::symlink_metadata(path)?.file_type().is_symlink() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "filesystem sandbox root must not be a symbolic link", + )); + } + + let path = path.canonicalize()?; + let dir = Dir::open_ambient_dir(&path, ambient_authority())?; + Ok(Self { + dir: Arc::new(dir), + path: Arc::new(path), + }) + } + + /// Returns the canonical path used to open this root. + pub fn path(&self) -> &Path { + &self.path + } + + /// Opens a descendant directory without leaving this capability root. + pub fn open_subdir(&self, path: impl AsRef) -> io::Result { + let path = normalize_relative_path(path.as_ref())?; + if path.as_os_str().is_empty() { + return Ok(self.clone()); + } + + let dir = self.dir.open_dir(&path)?; + Ok(Self { + dir: Arc::new(dir), + path: Arc::new(self.path.join(path)), + }) + } + + /// Creates and opens a descendant directory without leaving this capability root. + pub fn create_subdir(&self, path: impl AsRef) -> io::Result { + let path = normalize_relative_path(path.as_ref())?; + if path.as_os_str().is_empty() { + return Ok(self.clone()); + } + + self.dir.create_dir_all(&path)?; + let dir = self.dir.open_dir(&path)?; + Ok(Self { + dir: Arc::new(dir), + path: Arc::new(self.path.join(path)), + }) + } + + /// Builds an OpenDAL operator confined to this root. + pub fn build_operator(&self) -> Operator { + OperatorBuilder::new(SecureFsBackend::new(self.clone())).finish() + } +} + +fn normalize_relative_path(path: &Path) -> io::Result { + let mut normalized = PathBuf::new(); + for component in path.components() { + match component { + Component::CurDir => {} + Component::Normal(value) => normalized.push(value), + Component::ParentDir | Component::RootDir | Component::Prefix(_) => { + return Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "path escapes the filesystem sandbox", + )); + } + } + } + Ok(normalized) +} + +fn backend_path(path: &str) -> io::Result { + let path = path.trim_matches('/'); + if path.is_empty() { + Ok(PathBuf::new()) + } else { + normalize_relative_path(Path::new(path)) + } +} + +fn parse_write_error(error: io::Error, if_not_exists: bool) -> Error { + if if_not_exists && error.kind() == io::ErrorKind::AlreadyExists { + Error::new( + ErrorKind::ConditionNotMatch, + "the file already exists in the filesystem", + ) + .set_source(error) + } else { + new_std_io_error(error) + } +} + +fn metadata_from_fs(metadata: cap_std::fs::Metadata) -> Result { + let mode = if metadata.is_dir() { + EntryMode::DIR + } else if metadata.is_file() { + EntryMode::FILE + } else { + EntryMode::Unknown + }; + + Ok(Metadata::new(mode) + .with_content_length(metadata.len()) + .with_last_modified(Timestamp::try_from( + metadata.modified().map_err(new_std_io_error)?.into_std(), + )?)) +} + +#[derive(Clone, Debug)] +struct SecureFsBackend { + root: SecureFsRoot, + info: Arc, +} + +impl SecureFsBackend { + fn new(root: SecureFsRoot) -> Self { + let info = AccessorInfo::default(); + info.set_scheme("fs") + .set_root(&root.path().to_string_lossy()) + .set_native_capability(Capability { + stat: true, + read: true, + write: true, + write_can_empty: true, + write_can_append: true, + write_can_multi: true, + write_with_if_not_exists: true, + create_dir: true, + delete: true, + delete_with_recursive: true, + list: true, + shared: true, + ..Default::default() + }); + Self { + root, + info: info.into(), + } + } +} + +impl Access for SecureFsBackend { + type Reader = SecureFsReader; + type Writer = SecureFsWriter; + type Lister = Option; + type Deleter = oio::OneShotDeleter; + type Copier = (); + + fn info(&self) -> Arc { + self.info.clone() + } + + async fn create_dir(&self, path: &str, _: OpCreateDir) -> Result { + let path = backend_path(path).map_err(new_std_io_error)?; + let root = self.root.clone(); + common_runtime::spawn_blocking_global(move || root.dir.create_dir_all(path)) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error)?; + Ok(RpCreateDir::default()) + } + + async fn stat(&self, path: &str, _: OpStat) -> Result { + let path = backend_path(path).map_err(new_std_io_error)?; + let root = self.root.clone(); + let metadata = common_runtime::spawn_blocking_global(move || { + if path.as_os_str().is_empty() { + root.dir.dir_metadata() + } else { + root.dir.metadata(path) + } + }) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error)?; + Ok(RpStat::new(metadata_from_fs(metadata)?)) + } + + async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> { + let path = backend_path(path).map_err(new_std_io_error)?; + let root = self.root.clone(); + let file = common_runtime::spawn_blocking_global(move || root.dir.open(path)) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error)?; + let mut file = tokio::fs::File::from_std(file.into_std()); + if args.range().offset() != 0 { + file.seek(io::SeekFrom::Start(args.range().offset())) + .await + .map_err(new_std_io_error)?; + } + Ok(( + RpRead::default(), + SecureFsReader { + file, + remaining: args.range().size().unwrap_or(u64::MAX), + }, + )) + } + + async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> { + let path = backend_path(path).map_err(new_std_io_error)?; + let root = self.root.clone(); + let if_not_exists = args.if_not_exists(); + let file = common_runtime::spawn_blocking_global(move || { + if let Some(parent) = path.parent() + && !parent.as_os_str().is_empty() + { + root.dir.create_dir_all(parent).map_err(new_std_io_error)?; + } + + let mut options = OpenOptions::new(); + options.write(true); + if args.if_not_exists() { + options.create_new(true); + } else { + options.create(true); + } + if args.append() { + options.append(true); + } else { + options.truncate(true); + } + root.dir + .open_with(path, &options) + .map_err(|error| parse_write_error(error, if_not_exists)) + }) + .await + .map_err(new_task_join_error)??; + + Ok(( + RpWrite::default(), + SecureFsWriter { + file: tokio::fs::File::from_std(file.into_std()), + }, + )) + } + + async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> { + Ok(( + RpDelete::default(), + oio::OneShotDeleter::new(SecureFsDeleter { + root: self.root.clone(), + }), + )) + } + + async fn list(&self, path: &str, _: OpList) -> Result<(RpList, Self::Lister)> { + let path = backend_path(path).map_err(new_std_io_error)?; + let display_prefix = if path.as_os_str().is_empty() { + String::new() + } else { + format!("{}/", path.to_string_lossy().replace('\\', "/")) + }; + let root = self.root.clone(); + let read_dir = common_runtime::spawn_blocking_global(move || { + let result = (|| { + let dir = if path.as_os_str().is_empty() { + root.dir.open_dir(".")? + } else { + root.dir.open_dir(&path)? + }; + dir.entries() + })(); + + match result { + Ok(read_dir) => Ok(Some(read_dir)), + Err(error) + if matches!( + error.kind(), + io::ErrorKind::NotFound | io::ErrorKind::NotADirectory + ) => + { + Ok(None) + } + Err(error) => Err(error), + } + }) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error)?; + + let Some(read_dir) = read_dir else { + return Ok((RpList::default(), None)); + }; + let current_path = oio::Entry::new( + if display_prefix.is_empty() { + "/" + } else { + &display_prefix + }, + Metadata::new(EntryMode::DIR), + ); + Ok(( + RpList::default(), + Some(SecureFsLister { + read_dir: Arc::new(Mutex::new(read_dir)), + display_prefix, + entries: vec![current_path].into_iter(), + done: false, + }), + )) + } +} + +struct SecureFsReader { + file: tokio::fs::File, + remaining: u64, +} + +impl oio::Read for SecureFsReader { + async fn read(&mut self) -> Result { + if self.remaining == 0 { + return Ok(Buffer::new()); + } + + let size = self.remaining.min(2 * 1024 * 1024) as usize; + let mut buffer = vec![0; size]; + let read = self + .file + .read(&mut buffer) + .await + .map_err(new_std_io_error)?; + self.remaining = self.remaining.saturating_sub(read as u64); + buffer.truncate(read); + Ok(Buffer::from(buffer)) + } +} + +struct SecureFsWriter { + file: tokio::fs::File, +} + +impl oio::Write for SecureFsWriter { + async fn write(&mut self, buffer: Buffer) -> Result<()> { + self.file + .write_all(&buffer.to_bytes()) + .await + .map_err(new_std_io_error) + } + + async fn close(&mut self) -> Result { + self.file.flush().await.map_err(new_std_io_error)?; + self.file.sync_all().await.map_err(new_std_io_error)?; + let metadata = self.file.metadata().await.map_err(new_std_io_error)?; + Ok(Metadata::new(EntryMode::FILE) + .with_content_length(metadata.len()) + .with_last_modified(Timestamp::try_from( + metadata.modified().map_err(new_std_io_error)?, + )?)) + } + + async fn abort(&mut self) -> Result<()> { + Err(Error::new( + ErrorKind::Unsupported, + "filesystem writes cannot be aborted without atomic writes", + )) + } +} + +struct SecureFsLister { + read_dir: Arc>, + display_prefix: String, + entries: IntoIter, + done: bool, +} + +impl oio::List for SecureFsLister { + async fn next(&mut self) -> Result> { + if let Some(entry) = self.entries.next() { + return Ok(Some(entry)); + } + if self.done { + return Ok(None); + } + + let read_dir = self.read_dir.clone(); + let display_prefix = self.display_prefix.clone(); + let (entries, done) = common_runtime::spawn_blocking_global(move || { + let mut read_dir = read_dir + .lock() + .map_err(|_| io::Error::other("filesystem directory iterator lock is poisoned"))?; + read_list_batch(&mut read_dir, &display_prefix) + }) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error)?; + + self.entries = entries.into_iter(); + self.done = done; + Ok(self.entries.next()) + } +} + +fn read_list_batch( + read_dir: &mut ReadDir, + display_prefix: &str, +) -> io::Result<(Vec, bool)> { + let mut entries = Vec::with_capacity(LIST_BATCH_SIZE); + while entries.len() < LIST_BATCH_SIZE { + let entry = match read_dir.next() { + Some(Ok(entry)) => entry, + Some(Err(error)) if error.kind() == io::ErrorKind::NotFound => { + return Ok((entries, true)); + } + Some(Err(error)) => return Err(error), + None => return Ok((entries, true)), + }; + + if let Some(entry) = read_list_entry(entry, display_prefix)? { + entries.push(entry); + } + } + Ok((entries, false)) +} + +fn read_list_entry(entry: DirEntry, display_prefix: &str) -> io::Result> { + let file_type = match entry.file_type() { + Ok(file_type) => file_type, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error), + }; + let name = entry.file_name().to_string_lossy().to_string(); + let (path, mode) = if file_type.is_dir() { + (format!("{display_prefix}{name}/"), EntryMode::DIR) + } else if file_type.is_file() { + (format!("{display_prefix}{name}"), EntryMode::FILE) + } else { + (format!("{display_prefix}{name}"), EntryMode::Unknown) + }; + let metadata = if mode == EntryMode::Unknown { + Metadata::new(mode) + } else { + match entry.metadata() { + Ok(metadata) => match metadata_from_fs(metadata) { + Ok(metadata) => metadata, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(io::Error::other(error.to_string())), + }, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error), + } + }; + Ok(Some(oio::Entry::new(&path, metadata))) +} + +struct SecureFsDeleter { + root: SecureFsRoot, +} + +impl oio::OneShotDelete for SecureFsDeleter { + async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> { + let path = backend_path(&path).map_err(new_std_io_error)?; + if path.as_os_str().is_empty() { + return Err(Error::new( + ErrorKind::Unsupported, + "deleting the filesystem sandbox root is not supported", + )); + } + let root = self.root.clone(); + common_runtime::spawn_blocking_global(move || { + let metadata = match root.dir.symlink_metadata(&path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error), + }; + + if metadata.is_dir() { + if args.recursive() { + root.dir.remove_dir_all(path) + } else { + root.dir.remove_dir(path) + } + } else { + root.dir.remove_file(path) + } + }) + .await + .map_err(new_task_join_error)? + .map_err(new_std_io_error) + } +} + +#[cfg(test)] +mod tests { + use bytes::Bytes; + use common_test_util::temp_dir::create_temp_dir; + use opendal::ErrorKind; + use opendal::raw::oio::List; + use opendal::raw::{Access, OpList}; + + use super::{LIST_BATCH_SIZE, SecureFsBackend, SecureFsRoot, read_list_entry}; + + #[tokio::test] + async fn test_lister_streams_entries() { + let temp_dir = create_temp_dir("secure_fs_lister_streams_entries"); + for index in 0..129 { + std::fs::write(temp_dir.path().join(format!("{index}.parquet")), []).unwrap(); + } + + let root = SecureFsRoot::open(temp_dir.path()).unwrap(); + let backend = SecureFsBackend::new(root); + let (_, lister) = backend.list("/", OpList::new()).await.unwrap(); + let mut lister = lister.unwrap(); + + assert_eq!(1, lister.entries.len()); + + let mut paths = Vec::new(); + while let Some(entry) = lister.next().await.unwrap() { + paths.push(entry.path().to_string()); + assert!(lister.entries.len() <= LIST_BATCH_SIZE); + } + assert_eq!(130, paths.len()); + assert!(paths.iter().any(|path| path == "/")); + assert!(paths.iter().any(|path| path == "128.parquet")); + } + + #[tokio::test] + async fn test_if_not_exists_returns_condition_not_match() { + let temp_dir = create_temp_dir("secure_fs_if_not_exists"); + let operator = SecureFsRoot::open(temp_dir.path()) + .unwrap() + .build_operator(); + operator + .write("existing", Bytes::from_static(b"original")) + .await + .unwrap(); + + let error = operator + .write_with("existing", Bytes::from_static(b"replacement")) + .if_not_exists(true) + .await + .unwrap_err(); + + assert_eq!(ErrorKind::ConditionNotMatch, error.kind()); + assert_eq!( + Bytes::from_static(b"original"), + operator.read("existing").await.unwrap().to_bytes() + ); + } + + #[tokio::test] + async fn test_if_not_exists_does_not_remap_parent_directory_error() { + let temp_dir = create_temp_dir("secure_fs_if_not_exists_parent_error"); + std::fs::write(temp_dir.path().join("parent"), []).unwrap(); + let operator = SecureFsRoot::open(temp_dir.path()) + .unwrap() + .build_operator(); + + let error = operator + .write_with("parent/file", Bytes::new()) + .if_not_exists(true) + .await + .unwrap_err(); + + assert_eq!(ErrorKind::AlreadyExists, error.kind()); + } + + #[tokio::test] + async fn test_list_missing_or_non_directory_is_empty() { + let temp_dir = create_temp_dir("secure_fs_list_missing_or_non_directory"); + std::fs::write(temp_dir.path().join("file"), []).unwrap(); + let operator = SecureFsRoot::open(temp_dir.path()) + .unwrap() + .build_operator(); + + assert!(operator.list("missing/").await.unwrap().is_empty()); + assert!(operator.list("file/").await.unwrap().is_empty()); + } + + #[test] + fn test_lister_skips_entry_removed_during_iteration() { + let temp_dir = create_temp_dir("secure_fs_lister_removed_entry"); + let path = temp_dir.path().join("removed"); + std::fs::write(&path, []).unwrap(); + let root = SecureFsRoot::open(temp_dir.path()).unwrap(); + let mut read_dir = root.dir.entries().unwrap(); + let entry = read_dir.next().unwrap().unwrap(); + std::fs::remove_file(path).unwrap(); + + assert!(read_list_entry(entry, "").unwrap().is_none()); + } + + #[tokio::test] + async fn test_delete_root_is_unsupported() { + let temp_dir = create_temp_dir("secure_fs_delete_root"); + let operator = SecureFsRoot::open(temp_dir.path()) + .unwrap() + .build_operator(); + operator + .write("nested/file", Bytes::from_static(b"data")) + .await + .unwrap(); + + let error = operator.delete_with("/").recursive(true).await.unwrap_err(); + + assert_eq!(ErrorKind::Unsupported, error.kind()); + assert!(temp_dir.path().join("nested/file").exists()); + + operator + .delete_with("nested/") + .recursive(true) + .await + .unwrap(); + assert!(!temp_dir.path().join("nested").exists()); + } + + #[tokio::test] + async fn test_writer_abort_is_unsupported_without_atomic_write() { + let temp_dir = create_temp_dir("secure_fs_writer_abort"); + std::fs::write(temp_dir.path().join("partial"), b"original").unwrap(); + let operator = SecureFsRoot::open(temp_dir.path()) + .unwrap() + .build_operator(); + let mut writer = operator.writer("partial").await.unwrap(); + writer.write(Bytes::from_static(b"partial")).await.unwrap(); + + let error = writer.abort().await.unwrap_err(); + + assert_eq!(ErrorKind::Unsupported, error.kind()); + assert_eq!( + b"partial", + std::fs::read(temp_dir.path().join("partial")) + .unwrap() + .as_slice() + ); + } +} diff --git a/src/operator/src/error.rs b/src/operator/src/error.rs index 66fb66d752..eabd647773 100644 --- a/src/operator/src/error.rs +++ b/src/operator/src/error.rs @@ -469,13 +469,6 @@ pub enum Error { source: table::error::Error, }, - #[snafu(display("Failed to parse data source url"))] - ParseUrl { - #[snafu(implicit)] - location: Location, - source: common_datasource::error::Error, - }, - #[snafu(display("Unsupported format: {:?}", format))] UnsupportedFormat { #[snafu(implicit)] @@ -852,13 +845,6 @@ pub enum Error { location: Location, }, - #[snafu(display("Path not found: {path}"))] - PathNotFound { - path: String, - #[snafu(implicit)] - location: Location, - }, - #[snafu(display("Invalid time index type: {}", ty))] InvalidTimeIndexType { ty: arrow::datatypes::DataType, @@ -1040,9 +1026,9 @@ impl ErrorExt for Error { Error::ReadObject { .. } | Error::ReadParquetMetadata { .. } | Error::ReadOrc { .. } => StatusCode::StorageUnavailable, - Error::ListObjects { source, .. } - | Error::ParseUrl { source, .. } - | Error::BuildBackend { source, .. } => source.status_code(), + Error::ListObjects { source, .. } | Error::BuildBackend { source, .. } => { + source.status_code() + } Error::ExecuteDdl { source, .. } => source.status_code(), Error::InvalidCopyParameter { .. } | Error::InvalidCopyDatabasePath { .. } => { StatusCode::InvalidArguments @@ -1063,7 +1049,6 @@ impl ErrorExt for Error { } Error::InvalidProcessId { .. } => StatusCode::InvalidArguments, Error::ProcessManagerMissing { .. } => StatusCode::Unexpected, - Error::PathNotFound { .. } => StatusCode::InvalidArguments, Error::TimestampFormatNotSupported { .. } => StatusCode::InvalidArguments, Error::SqlCommon { source, .. } => source.status_code(), #[cfg(feature = "enterprise")] @@ -1096,7 +1081,6 @@ impl ErrorExt for Error { Error::ParseFileFormat { source, .. } | Error::InferSchema { source, .. } | Error::ListObjects { source, .. } - | Error::ParseUrl { source, .. } | Error::BuildBackend { source, .. } | Error::ReadOrc { source, .. } => source.retry_hint(), diff --git a/src/operator/src/expr_helper.rs b/src/operator/src/expr_helper.rs index 7bc0800aa7..f6ed0d8092 100644 --- a/src/operator/src/expr_helper.rs +++ b/src/operator/src/expr_helper.rs @@ -31,6 +31,7 @@ use api::v1::{ UnsetIndex, UnsetIndexes, UnsetInverted, UnsetSkipping, UnsetTableOptions, set_index, unset_index, }; +use common_datasource::object_store::LocalFileAccess; use common_error::ext::BoxedError; use common_grpc_expr::util::ColumnExpr; use common_time::Timezone; @@ -140,6 +141,7 @@ pub fn extract_add_columns_expr( pub(crate) async fn create_external_expr( create: CreateExternalTable, query_ctx: &QueryContextRef, + local_file_access: &LocalFileAccess, ) -> Result { let (catalog_name, schema_name, table_name) = table_idents_to_full_name(&create.name, query_ctx) @@ -148,7 +150,7 @@ pub(crate) async fn create_external_expr( let mut table_options = create.options.into_map(); - let (object_store, files) = prepare_file_table_files(&table_options) + let (object_store, files) = prepare_file_table_files(&table_options, local_file_access) .await .context(PrepareFileTableSnafu)?; diff --git a/src/operator/src/statement.rs b/src/operator/src/statement.rs index 388e258ceb..4fa33109cd 100644 --- a/src/operator/src/statement.rs +++ b/src/operator/src/statement.rs @@ -37,6 +37,7 @@ use catalog::process_manager::ProcessManagerRef; use client::RecordBatches; use client::error::{ExternalSnafu as ClientExternalSnafu, Result as ClientResult}; use client::inserter::{InsertOptions, Inserter}; +use common_datasource::object_store::LocalFileAccess; use common_error::ext::BoxedError; use common_meta::cache_invalidator::CacheInvalidatorRef; use common_meta::key::flow::{FlowMetadataManager, FlowMetadataManagerRef}; @@ -137,6 +138,7 @@ pub struct StatementExecutor { inserter: InserterRef, process_manager: Option, origin_frontend_addr: String, + pub(crate) local_file_access: LocalFileAccess, #[cfg(feature = "enterprise")] create_database_handler: Option, #[cfg(feature = "enterprise")] @@ -175,6 +177,7 @@ impl StatementExecutor { partition_manager: PartitionRuleManagerRef, process_manager: Option, origin_frontend_addr: String, + local_file_access: LocalFileAccess, ) -> Self { Self { catalog_manager, @@ -188,6 +191,7 @@ impl StatementExecutor { inserter, process_manager, origin_frontend_addr, + local_file_access, #[cfg(feature = "enterprise")] create_database_handler: None, #[cfg(feature = "enterprise")] diff --git a/src/operator/src/statement/copy_database.rs b/src/operator/src/statement/copy_database.rs index cd8eeb6d79..3db6eb8e76 100644 --- a/src/operator/src/statement/copy_database.rs +++ b/src/operator/src/statement/copy_database.rs @@ -21,7 +21,9 @@ use client::{Output, OutputData, OutputMeta}; use common_catalog::format_full_table_name; use common_datasource::file_format::Format; use common_datasource::lister::{Lister, Source}; -use common_datasource::object_store::build_backend; +#[cfg(windows)] +use common_datasource::object_store::{FS_SCHEMA, parse_url}; +use common_datasource::object_store::{LocalFileAccess, build_backend, build_backend_for_write}; use common_stat::get_total_cpu_cores; use common_telemetry::{debug, error, info, tracing}; use futures::future::try_join_all; @@ -43,6 +45,24 @@ pub(crate) const COPY_DATABASE_TIME_END_KEY: &str = "end_time"; pub(crate) const CONTINUE_ON_ERROR_KEY: &str = "continue_on_error"; pub(crate) const PARALLELISM_KEY: &str = "parallelism"; +fn is_directory_location(location: &str) -> bool { + if location.ends_with('/') { + return true; + } + + #[cfg(windows)] + { + location.ends_with(std::path::MAIN_SEPARATOR) + && matches!( + parse_url(location), + Ok((schema, _, _)) if schema.eq_ignore_ascii_case(FS_SCHEMA) + ) + } + + #[cfg(not(windows))] + false +} + /// Get parallelism from options, default to total CPU cores. fn parse_parallelism_from_option_map(options: &HashMap) -> usize { options @@ -59,13 +79,16 @@ impl StatementExecutor { req: CopyDatabaseRequest, ctx: QueryContextRef, ) -> error::Result { - // location must end with / so that every table is exported to a file. + // Location must end with a separator so that every table is exported to a file. ensure!( - req.location.ends_with('/'), + is_directory_location(&req.location), InvalidCopyDatabasePathSnafu { value: req.location, } ); + build_backend_for_write(&req.location, &req.connection, &self.local_file_access) + .await + .context(error::BuildBackendSnafu)?; let parallelism = parse_parallelism_from_option_map(&req.with); info!( @@ -153,9 +176,9 @@ impl StatementExecutor { req: CopyDatabaseRequest, ctx: QueryContextRef, ) -> error::Result { - // location must end with / + // Location must end with a directory separator. ensure!( - req.location.ends_with('/'), + is_directory_location(&req.location), InvalidCopyDatabasePathSnafu { value: req.location, } @@ -170,7 +193,7 @@ impl StatementExecutor { .context(error::ParseFileFormatSnafu)? .suffix(); - let entries = list_files_to_copy(&req, suffix).await?; + let entries = list_files_to_copy(&req, suffix, &self.local_file_access).await?; let continue_on_error = req .with @@ -198,7 +221,7 @@ impl StatementExecutor { catalog_name: req.catalog_name.clone(), schema_name: req.schema_name.clone(), table_name: table_name.clone(), - location: format!("{}/{}", req.location, e.path()), + location: format!("{}{}", req.location, e.path()), with: req.with.clone(), connection: req.connection.clone(), pattern: None, @@ -255,9 +278,14 @@ fn parse_file_name_to_copy(e: &Entry) -> error::Result { } /// Lists all files with expected suffix that can be imported to database. -async fn list_files_to_copy(req: &CopyDatabaseRequest, suffix: &str) -> error::Result> { - let object_store = - build_backend(&req.location, &req.connection).context(error::BuildBackendSnafu)?; +async fn list_files_to_copy( + req: &CopyDatabaseRequest, + suffix: &str, + local_file_access: &LocalFileAccess, +) -> error::Result> { + let object_store = build_backend(&req.location, &req.connection, local_file_access) + .await + .context(error::BuildBackendSnafu)?; let pattern = Regex::try_from(format!(".*{}", suffix)).context(error::BuildRegexSnafu)?; let lister = Lister::new( @@ -273,10 +301,12 @@ async fn list_files_to_copy(req: &CopyDatabaseRequest, suffix: &str) -> error::R mod tests { use std::collections::{HashMap, HashSet}; + use common_datasource::object_store::LocalFileAccess; use common_stat::get_total_cpu_cores; use object_store::ObjectStore; use object_store::services::Fs; use object_store::util::normalize_dir; + #[cfg(not(windows))] use path_slash::PathExt; use table::requests::CopyDatabaseRequest; @@ -296,7 +326,10 @@ mod tests { object_store.write("d", "").await.unwrap(); object_store.write("e.f.parquet", "").await.unwrap(); + #[cfg(not(windows))] let location = normalize_dir(&dir.path().to_slash().unwrap()); + #[cfg(windows)] + let location = format!("{}\\", dir.path().display()); let request = CopyDatabaseRequest { catalog_name: "catalog_0".to_string(), schema_name: "schema_0".to_string(), @@ -307,7 +340,8 @@ mod tests { connection: Default::default(), time_range: None, }; - let listed = list_files_to_copy(&request, ".parquet") + let local_file_access = LocalFileAccess::sandboxed(dir.path()).unwrap(); + let listed = list_files_to_copy(&request, ".parquet", &local_file_access) .await .unwrap() .into_iter() diff --git a/src/operator/src/statement/copy_table_from.rs b/src/operator/src/statement/copy_table_from.rs index d2f8b3be90..39bc3dafcd 100644 --- a/src/operator/src/statement/copy_table_from.rs +++ b/src/operator/src/statement/copy_table_from.rs @@ -14,7 +14,6 @@ use std::collections::{BTreeSet, HashMap, HashSet}; use std::future::Future; -use std::path::Path; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; @@ -28,8 +27,7 @@ use common_datasource::file_format::json::JsonFormat; use common_datasource::file_format::orc::{ReaderAdapter, infer_orc_schema, new_orc_stream_reader}; use common_datasource::file_format::{FileFormat, Format, file_to_stream}; use common_datasource::lister::{Lister, Source}; -use common_datasource::object_store::{FS_SCHEMA, build_backend, parse_url}; -use common_datasource::util::find_dir_and_filename; +use common_datasource::object_store::build_backend_with_path; use common_query::{OutputCost, OutputRows}; use common_recordbatch::DfSendableRecordBatchStream; use common_recordbatch::adapter::RecordBatchStreamTypeAdapter; @@ -54,7 +52,7 @@ use table::requests::{CopyTableRequest, InsertRequest}; use table::table_reference::TableReference; use tokio_util::compat::FuturesAsyncReadCompatExt; -use crate::error::{self, IntoVectorsSnafu, PathNotFoundSnafu, Result}; +use crate::error::{self, IntoVectorsSnafu, Result}; use crate::statement::StatementExecutor; const DEFAULT_BATCH_SIZE: usize = 8192; @@ -99,16 +97,10 @@ impl StatementExecutor { &self, req: &CopyTableRequest, ) -> Result<(ObjectStore, Vec)> { - let (schema, _host, path) = parse_url(&req.location).context(error::ParseUrlSnafu)?; - - if schema.to_uppercase() == FS_SCHEMA { - ensure!(Path::new(&path).exists(), PathNotFoundSnafu { path }); - } - - let object_store = - build_backend(&req.location, &req.connection).context(error::BuildBackendSnafu)?; - - let (dir, filename) = find_dir_and_filename(&path); + let backend = + build_backend_with_path(&req.location, &req.connection, &self.local_file_access) + .await + .context(error::BuildBackendSnafu)?; let regex = req .pattern .as_ref() @@ -116,17 +108,25 @@ impl StatementExecutor { .transpose() .context(error::BuildRegexSnafu)?; - let source = if let Some(filename) = filename { + let source = if let Some(filename) = backend.object_path { Source::Filename(filename) } else { Source::Dir }; - let lister = Lister::new(object_store.clone(), source.clone(), dir.clone(), regex); + let lister = Lister::new( + backend.object_store.clone(), + source.clone(), + req.location.clone(), + regex, + ); let entries = lister.list().await.context(error::ListObjectsSnafu)?; - debug!("Copy from dir: {dir:?}, {source:?}, entries: {entries:?}"); - Ok((object_store, entries)) + debug!( + "Copy from location: {:?}, {source:?}, entries: {entries:?}", + req.location + ); + Ok((backend.object_store, entries)) } async fn collect_metadata( diff --git a/src/operator/src/statement/copy_table_to.rs b/src/operator/src/statement/copy_table_to.rs index 9dd478ba20..4cf9448d8b 100644 --- a/src/operator/src/statement/copy_table_to.rs +++ b/src/operator/src/statement/copy_table_to.rs @@ -21,8 +21,7 @@ use common_datasource::file_format::Format; use common_datasource::file_format::csv::stream_to_csv; use common_datasource::file_format::json::stream_to_json; use common_datasource::file_format::parquet::stream_to_parquet; -use common_datasource::object_store::{build_backend, parse_url}; -use common_datasource::util::find_dir_and_filename; +use common_datasource::object_store::build_backend_for_write_with_path; use common_query::Output; use common_recordbatch::adapter::DfRecordBatchStreamAdapter; use common_recordbatch::{ @@ -178,13 +177,14 @@ impl StatementExecutor { _ => unreachable!(), }; - let (_schema, _host, path) = parse_url(location).context(error::ParseUrlSnafu)?; - let (_, filename) = find_dir_and_filename(&path); - let filename = filename.context(error::UnexpectedSnafu { - violated: format!("Expected filename, path: {path}"), + let backend = + build_backend_for_write_with_path(location, connection, &self.local_file_access) + .await + .context(error::BuildBackendSnafu)?; + let filename = backend.object_path.context(error::UnexpectedSnafu { + violated: format!("Expected filename, path: {location}"), })?; - let object_store = build_backend(location, connection).context(error::BuildBackendSnafu)?; - self.stream_to_file(stream, format, object_store, &filename) + self.stream_to_file(stream, format, backend.object_store, &filename) .await } } diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index 4502ebe36b..b671bab886 100644 --- a/src/operator/src/statement/ddl.rs +++ b/src/operator/src/statement/ddl.rs @@ -361,7 +361,9 @@ impl StatementExecutor { create_expr: CreateExternalTable, ctx: QueryContextRef, ) -> Result { - let create_expr = &mut expr_helper::create_external_expr(create_expr, &ctx).await?; + let create_expr = + &mut expr_helper::create_external_expr(create_expr, &ctx, &self.local_file_access) + .await?; self.create_table_inner(create_expr, None, ctx).await } diff --git a/src/query/src/error.rs b/src/query/src/error.rs index d7b4a66703..fdfeaafde9 100644 --- a/src/query/src/error.rs +++ b/src/query/src/error.rs @@ -424,7 +424,8 @@ impl ErrorExt for Error { | InvalidQueryContextExtension { .. } | ConflictingSnapshotSequence { .. } => StatusCode::InvalidArguments, - BuildBackend { .. } | ListObjects { .. } => StatusCode::StorageUnavailable, + BuildBackend { source, .. } => source.status_code(), + ListObjects { .. } => StatusCode::StorageUnavailable, TableNotFound { .. } => StatusCode::TableNotFound, @@ -494,3 +495,23 @@ impl From for DataFusionError { DataFusionError::External(Box::new(e)) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_build_backend_delegates_error_metadata() { + let source = common_datasource::error::LocalFileAccessDisabledSnafu { + path: "file:///tmp/data.parquet", + } + .build(); + let error = Error::BuildBackend { + source, + location: Location::default(), + }; + + assert_eq!(error.status_code(), StatusCode::InvalidArguments); + assert_eq!(error.retry_hint(), RetryHint::NonRetryable); + } +} diff --git a/src/query/src/sql.rs b/src/query/src/sql.rs index 8dda64cfd2..90c3b4ccd4 100644 --- a/src/query/src/sql.rs +++ b/src/query/src/sql.rs @@ -30,8 +30,7 @@ use common_catalog::consts::{ use common_catalog::format_full_table_name; use common_datasource::file_format::{FileFormat, Format, infer_schemas}; use common_datasource::lister::{Lister, Source}; -use common_datasource::object_store::build_backend; -use common_datasource::util::find_dir_and_filename; +use common_datasource::object_store::{LocalFileAccess, build_backend_with_path}; use common_meta::SchemaOptions; use common_meta::ddl::create_flow::FlowType; use common_meta::key::flow::flow_info::FlowInfoValue; @@ -1149,6 +1148,7 @@ fn describe_column_semantic_types( // lists files in the frontend to reduce unnecessary scan requests repeated in each datanode. pub async fn prepare_file_table_files( options: &HashMap, + local_file_access: &LocalFileAccess, ) -> Result<(ObjectStore, Vec)> { let url = options .get(FILE_TABLE_LOCATION_KEY) @@ -1156,19 +1156,20 @@ pub async fn prepare_file_table_files( name: FILE_TABLE_LOCATION_KEY, })?; - let (dir, filename) = find_dir_and_filename(url); - let source = if let Some(filename) = filename { - Source::Filename(filename) - } else { - Source::Dir - }; let regex = options .get(FILE_TABLE_PATTERN_KEY) .map(|x| Regex::new(x)) .transpose() .context(error::BuildRegexSnafu)?; - let object_store = build_backend(url, options).context(error::BuildBackendSnafu)?; - let lister = Lister::new(object_store.clone(), source, dir, regex); + let backend = build_backend_with_path(url, options, local_file_access) + .await + .context(error::BuildBackendSnafu)?; + let source = if let Some(filename) = backend.object_path { + Source::Filename(filename) + } else { + Source::Dir + }; + let lister = Lister::new(backend.object_store.clone(), source, url.clone(), regex); // If we scan files in a directory every time the database restarts, // then it might lead to a potential undefined behavior: // If a user adds a file with an incompatible schema to that directory, @@ -1186,7 +1187,7 @@ pub async fn prepare_file_table_files( } }) .collect::>(); - Ok((object_store, files)) + Ok((backend.object_store, files)) } pub async fn infer_file_table_schema( diff --git a/tests-integration/Cargo.toml b/tests-integration/Cargo.toml index 6ee801a6d8..ef29b7439d 100644 --- a/tests-integration/Cargo.toml +++ b/tests-integration/Cargo.toml @@ -33,6 +33,7 @@ cmd.workspace = true common-base.workspace = true common-catalog.workspace = true common-config.workspace = true +common-datasource.workspace = true common-error.workspace = true common-event-recorder.workspace = true common-frontend.workspace = true diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 2aff75dac5..f93d7da804 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -26,6 +26,7 @@ use cmd::error::StartFlownodeSnafu; use common_base::Plugins; use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID}; use common_config::KvBackendConfig; +use common_datasource::object_store::LocalFileAccess; use common_meta::cache::LayeredCacheRegistryBuilder; use common_meta::ddl::flow_meta::FlowMetadataAllocator; use common_meta::ddl::table_meta::TableMetadataAllocator; @@ -43,6 +44,7 @@ use common_procedure::ProcedureManagerRef; use common_procedure::local::EventRecorderHandle; use common_procedure::options::ProcedureConfig; use common_telemetry::logging::SlowQueryOptions; +use common_test_util::find_workspace_path; use common_wal::config::{DatanodeWalConfig, MetasrvWalConfig}; use datanode::datanode::DatanodeBuilder; use flow::{FlownodeBuilder, FrontendClient, GrpcQueryHandlerWithBoxedError}; @@ -172,7 +174,9 @@ impl GreptimeDbStandaloneBuilder { let mut builder = DatanodeBuilder::new(opts.datanode_options(), plugins.clone(), kv_backend.clone()); + let local_file_access = LocalFileAccess::sandboxed(find_workspace_path(".")).unwrap(); builder.with_cache_registry(layered_cache_registry); + builder.with_local_file_access(local_file_access.clone()); let datanode = builder.build().await.unwrap(); let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend.clone())); @@ -278,6 +282,7 @@ impl GreptimeDbStandaloneBuilder { procedure_executor.clone(), Arc::new(ProcessManager::new(server_addr, None)), ) + .with_local_file_access(local_file_access) .with_plugin(plugins.clone()) .try_build() .await diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs index acf0e2b522..107f2712d9 100644 --- a/tests-integration/src/test_util.rs +++ b/tests-integration/src/test_util.rs @@ -362,6 +362,7 @@ pub(crate) fn create_datanode_opts( require_lease_before_startup: true, storage: StorageConfig { data_home: home_dir, + copy_root: None, providers, store: default_store, }, diff --git a/tests-integration/src/tests/instance_test.rs b/tests-integration/src/tests/instance_test.rs index 90c3708b53..9c64e84125 100644 --- a/tests-integration/src/tests/instance_test.rs +++ b/tests-integration/src/tests/instance_test.rs @@ -17,11 +17,11 @@ use std::sync::Arc; use client::{DEFAULT_SCHEMA_NAME, OutputData}; use common_catalog::consts::DEFAULT_CATALOG_NAME; -use common_error::ext::ErrorExt; +use common_error::ext::{ErrorExt, RetryHint}; +use common_error::status_code::StatusCode; use common_query::Output; use common_recordbatch::util; use common_test_util::recordbatch::check_output_stream; -use common_test_util::temp_dir; use datatypes::arrow::array::{ ArrayRef, AsArray, StringArray, TimestampMillisecondArray, UInt64Array, }; @@ -37,9 +37,9 @@ use session::context::{QueryContext, QueryContextRef}; use crate::tests::test_util::{ MockInstance, both_instances_cases, both_instances_cases_with_custom_storages, - check_unordered_output_stream, distributed, distributed_with_multiple_object_stores, - find_testing_resource, prepare_path, standalone, standalone_instance_case, - standalone_with_multiple_object_stores, + check_unordered_output_stream, create_local_file_test_dir, distributed, + distributed_with_multiple_object_stores, find_testing_resource, prepare_path, standalone, + standalone_instance_case, standalone_with_multiple_object_stores, }; #[apply(both_instances_cases)] @@ -250,7 +250,7 @@ PARTITION ON COLUMNS (n) ( check_output_stream(output, expected).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_extra_external_table_options(instance: Arc) { let frontend = instance.frontend(); let format = "json"; @@ -277,7 +277,7 @@ async fn test_extra_external_table_options(instance: Arc) { )); } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_show_create_external_table(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -661,11 +661,52 @@ async fn test_execute_create(instance: Arc) { assert!(matches!(output, OutputData::AffectedRows(0))); } -#[apply(both_instances_cases)] +#[tokio::test(flavor = "multi_thread")] +async fn test_distributed_local_file_access_disabled() { + let instance = distributed().await.frontend(); + execute_sql( + &instance, + "CREATE TABLE local_file_access_distributed ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE + );", + ) + .await; + + let statements = [ + "COPY local_file_access_distributed TO 'local_file_access/table.parquet';", + "COPY local_file_access_distributed FROM 'local_file_access/table.parquet';", + "COPY (SELECT * FROM local_file_access_distributed) TO 'local_file_access/query.parquet';", + "COPY DATABASE public TO 'local_file_access/database/';", + "COPY DATABASE public FROM 'local_file_access/database/';", + "CREATE EXTERNAL TABLE local_file_access_external WITH ( + location = 'local_file_access/table.parquet', + format = 'parquet' + );", + ]; + + for statement in statements { + let error = try_execute_sql(&instance, statement).await.unwrap_err(); + assert_eq!(error.status_code(), StatusCode::InvalidArguments); + assert_eq!(error.retry_hint(), RetryHint::NonRetryable); + let message = error.output_msg(); + assert!( + message.contains("SQL access to the local filesystem is disabled"), + "{message}" + ); + assert!( + message.contains("use S3, OSS, GCS, or AzBlob instead"), + "{message}" + ); + } +} + +#[apply(standalone_instance_case)] async fn test_execute_external_create(instance: Arc) { let instance = instance.frontend(); - let tmp_dir = temp_dir::create_temp_dir("test_execute_external_create"); + let tmp_dir = create_local_file_test_dir("test_execute_external_create"); let location = prepare_path(tmp_dir.path().to_str().unwrap()); let output = execute_sql( @@ -700,11 +741,11 @@ async fn test_execute_external_create(instance: Arc) { assert!(matches!(output, OutputData::AffectedRows(0))); } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_external_create_infer_format(instance: Arc) { let instance = instance.frontend(); - let tmp_dir = temp_dir::create_temp_dir("test_execute_external_create_infer_format"); + let tmp_dir = create_local_file_test_dir("test_execute_external_create_infer_format"); let location = prepare_path(tmp_dir.path().to_str().unwrap()); let output = execute_sql( @@ -716,11 +757,11 @@ async fn test_execute_external_create_infer_format(instance: Arc) { let instance = instance.frontend(); - let tmp_dir = temp_dir::create_temp_dir("test_execute_external_create_without_ts"); + let tmp_dir = create_local_file_test_dir("test_execute_external_create_without_ts"); let location = prepare_path(tmp_dir.path().to_str().unwrap()); let result = try_execute_sql( @@ -741,11 +782,11 @@ async fn test_execute_external_create_without_ts(instance: Arc )); } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_external_create_with_invalid_ts(instance: Arc) { let instance = instance.frontend(); - let tmp_dir = temp_dir::create_temp_dir("test_execute_external_create_with_invalid_ts"); + let tmp_dir = create_local_file_test_dir("test_execute_external_create_with_invalid_ts"); let location = prepare_path(tmp_dir.path().to_str().unwrap()); let result = try_execute_sql( @@ -785,7 +826,7 @@ async fn test_execute_external_create_with_invalid_ts(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -860,7 +901,7 @@ async fn test_execute_query_external_table_parquet(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -945,7 +986,7 @@ async fn test_execute_query_external_table_orc(instance: Arc) check_output_stream(output, expect).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_query_external_table_orc_with_schema(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -1003,7 +1044,7 @@ async fn test_execute_query_external_table_orc_with_schema(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -1058,7 +1099,7 @@ async fn test_execute_query_external_table_csv(instance: Arc) check_output_stream(output, expect).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_copy_from_headerless_csv(instance: Arc) { let instance = instance.frontend(); let csv_path = find_testing_resource("/tests/data/csv/headerless.csv"); @@ -1183,10 +1224,10 @@ async fn test_execute_copy_from_headerless_csv(instance: Arc) check_output_stream(output, expect).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_copy_from_csv_strict_headers(instance: Arc) { let instance = instance.frontend(); - let tmp_dir = temp_dir::create_temp_dir("test_execute_copy_from_csv_strict_headers"); + let tmp_dir = create_local_file_test_dir("test_execute_copy_from_csv_strict_headers"); let matching_path = tmp_dir.path().join("matching.csv"); let unknown_path = tmp_dir.path().join("unknown.csv"); let missing_path = tmp_dir.path().join("missing.csv"); @@ -1315,7 +1356,7 @@ async fn test_execute_copy_from_csv_strict_headers(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -1377,7 +1418,7 @@ async fn test_execute_query_external_table_json(instance: Arc) check_output_stream(output, expect).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_query_external_table_json_with_schema(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -1448,7 +1489,7 @@ async fn test_execute_query_external_table_json_with_schema(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -1523,7 +1564,7 @@ async fn test_execute_query_external_table_json_type_cast(instance: Arc) { unsafe { std::env::set_var("TZ", "UTC"); @@ -2584,7 +2625,7 @@ async fn test_execute_copy_from_azblob(instance: Arc) { } } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_copy_from_orc_with_cast(instance: Arc) { common_telemetry::init_default_ut_logging(); let instance = instance.frontend(); @@ -2623,7 +2664,7 @@ async fn test_execute_copy_from_orc_with_cast(instance: Arc) { check_output_stream(output, expected).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_execute_copy_from_orc(instance: Arc) { common_telemetry::init_default_ut_logging(); let instance = instance.frontend(); @@ -2661,7 +2702,7 @@ async fn test_execute_copy_from_orc(instance: Arc) { check_output_stream(output, expected).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_cast_type_issue_1594(instance: Arc) { let instance = instance.frontend(); @@ -2698,7 +2739,7 @@ async fn test_cast_type_issue_1594(instance: Arc) { check_output_stream(output, expected).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_copy_from_csv_skip_bad_records(instance: Arc) { let instance = instance.frontend(); @@ -3116,7 +3157,7 @@ WITH( } } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_copy_parquet_map_to_json(instance: Arc) { let instance = instance.frontend(); @@ -3183,7 +3224,7 @@ async fn test_copy_parquet_map_to_json(instance: Arc) { check_output_stream(output, &expected).await; } -#[apply(both_instances_cases)] +#[apply(standalone_instance_case)] async fn test_copy_parquet_map_to_binary(instance: Arc) { let instance = instance.frontend(); diff --git a/tests-integration/src/tests/test_util.rs b/tests-integration/src/tests/test_util.rs index 473cb3ce2b..ab86036750 100644 --- a/tests-integration/src/tests/test_util.rs +++ b/tests-integration/src/tests/test_util.rs @@ -439,6 +439,16 @@ pub fn prepare_path(p: &str) -> String { p.to_string() } +/// Creates a temporary local-file test directory inside the standalone test sandbox. +pub fn create_local_file_test_dir(prefix: &str) -> tempfile::TempDir { + let parent = find_workspace_path("target/local-file-tests"); + std::fs::create_dir_all(&parent).unwrap(); + tempfile::Builder::new() + .prefix(prefix) + .tempdir_in(parent) + .unwrap() +} + /// Find the testing file resource under workspace root to be used in object store. pub fn find_testing_resource(path: &str) -> String { let p = find_workspace_path(path).display().to_string(); diff --git a/tests/cases/distributed/local_file_access.result b/tests/cases/distributed/local_file_access.result new file mode 100644 index 0000000000..16a97d7a67 --- /dev/null +++ b/tests/cases/distributed/local_file_access.result @@ -0,0 +1,49 @@ +CREATE TABLE local_file_access_distributed ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE +); + +Affected Rows: 0 + +INSERT INTO local_file_access_distributed VALUES (1, 'host-1', 1.0); + +Affected Rows: 1 + +COPY local_file_access_distributed +TO 'local_file_access/table.parquet'; + +Error: 1004(InvalidArguments), SQL access to the local filesystem is disabled for 'local_file_access/table.parquet'; use S3, OSS, GCS, or AzBlob instead + +COPY local_file_access_distributed +FROM 'local_file_access/table.parquet'; + +Error: 1004(InvalidArguments), SQL access to the local filesystem is disabled for 'local_file_access/table.parquet'; use S3, OSS, GCS, or AzBlob instead + +COPY (SELECT * FROM local_file_access_distributed) +TO 'local_file_access/query.parquet'; + +Error: 1004(InvalidArguments), SQL access to the local filesystem is disabled for 'local_file_access/query.parquet'; use S3, OSS, GCS, or AzBlob instead + +COPY DATABASE public +TO 'local_file_access/database/'; + +Error: 1004(InvalidArguments), SQL access to the local filesystem is disabled for 'local_file_access/database/'; use S3, OSS, GCS, or AzBlob instead + +COPY DATABASE public +FROM 'local_file_access/database/'; + +Error: 1004(InvalidArguments), SQL access to the local filesystem is disabled for 'local_file_access/database/'; use S3, OSS, GCS, or AzBlob instead + +CREATE EXTERNAL TABLE local_file_access_external +WITH ( + location = 'local_file_access/table.parquet', + format = 'parquet' +); + +Error: 1004(InvalidArguments), SQL access to the local filesystem is disabled for 'local_file_access/table.parquet'; use S3, OSS, GCS, or AzBlob instead + +DROP TABLE local_file_access_distributed; + +Affected Rows: 0 + diff --git a/tests/cases/distributed/local_file_access.sql b/tests/cases/distributed/local_file_access.sql new file mode 100644 index 0000000000..a36930d893 --- /dev/null +++ b/tests/cases/distributed/local_file_access.sql @@ -0,0 +1,30 @@ +CREATE TABLE local_file_access_distributed ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE +); + +INSERT INTO local_file_access_distributed VALUES (1, 'host-1', 1.0); + +COPY local_file_access_distributed +TO 'local_file_access/table.parquet'; + +COPY local_file_access_distributed +FROM 'local_file_access/table.parquet'; + +COPY (SELECT * FROM local_file_access_distributed) +TO 'local_file_access/query.parquet'; + +COPY DATABASE public +TO 'local_file_access/database/'; + +COPY DATABASE public +FROM 'local_file_access/database/'; + +CREATE EXTERNAL TABLE local_file_access_external +WITH ( + location = 'local_file_access/table.parquet', + format = 'parquet' +); + +DROP TABLE local_file_access_distributed; diff --git a/tests/cases/standalone/common/copy/copy_database_from_fs_parquet.result b/tests/cases/standalone/copy/copy_database_from_fs_parquet.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_database_from_fs_parquet.result rename to tests/cases/standalone/copy/copy_database_from_fs_parquet.result diff --git a/tests/cases/standalone/common/copy/copy_database_from_fs_parquet.sql b/tests/cases/standalone/copy/copy_database_from_fs_parquet.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_database_from_fs_parquet.sql rename to tests/cases/standalone/copy/copy_database_from_fs_parquet.sql diff --git a/tests/cases/standalone/common/copy/copy_from_csv_compressed.result b/tests/cases/standalone/copy/copy_from_csv_compressed.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_csv_compressed.result rename to tests/cases/standalone/copy/copy_from_csv_compressed.result diff --git a/tests/cases/standalone/common/copy/copy_from_csv_compressed.sql b/tests/cases/standalone/copy/copy_from_csv_compressed.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_csv_compressed.sql rename to tests/cases/standalone/copy/copy_from_csv_compressed.sql diff --git a/tests/cases/standalone/common/copy/copy_from_fs_csv.result b/tests/cases/standalone/copy/copy_from_fs_csv.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_fs_csv.result rename to tests/cases/standalone/copy/copy_from_fs_csv.result diff --git a/tests/cases/standalone/common/copy/copy_from_fs_csv.sql b/tests/cases/standalone/copy/copy_from_fs_csv.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_fs_csv.sql rename to tests/cases/standalone/copy/copy_from_fs_csv.sql diff --git a/tests/cases/standalone/common/copy/copy_from_fs_json.result b/tests/cases/standalone/copy/copy_from_fs_json.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_fs_json.result rename to tests/cases/standalone/copy/copy_from_fs_json.result diff --git a/tests/cases/standalone/common/copy/copy_from_fs_json.sql b/tests/cases/standalone/copy/copy_from_fs_json.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_fs_json.sql rename to tests/cases/standalone/copy/copy_from_fs_json.sql diff --git a/tests/cases/standalone/common/copy/copy_from_fs_parquet.result b/tests/cases/standalone/copy/copy_from_fs_parquet.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_fs_parquet.result rename to tests/cases/standalone/copy/copy_from_fs_parquet.result diff --git a/tests/cases/standalone/common/copy/copy_from_fs_parquet.sql b/tests/cases/standalone/copy/copy_from_fs_parquet.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_fs_parquet.sql rename to tests/cases/standalone/copy/copy_from_fs_parquet.sql diff --git a/tests/cases/standalone/common/copy/copy_from_json_compressed.result b/tests/cases/standalone/copy/copy_from_json_compressed.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_json_compressed.result rename to tests/cases/standalone/copy/copy_from_json_compressed.result diff --git a/tests/cases/standalone/common/copy/copy_from_json_compressed.sql b/tests/cases/standalone/copy/copy_from_json_compressed.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_from_json_compressed.sql rename to tests/cases/standalone/copy/copy_from_json_compressed.sql diff --git a/tests/cases/standalone/common/copy/copy_to_csv_compressed.result b/tests/cases/standalone/copy/copy_to_csv_compressed.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_to_csv_compressed.result rename to tests/cases/standalone/copy/copy_to_csv_compressed.result diff --git a/tests/cases/standalone/common/copy/copy_to_csv_compressed.sql b/tests/cases/standalone/copy/copy_to_csv_compressed.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_to_csv_compressed.sql rename to tests/cases/standalone/copy/copy_to_csv_compressed.sql diff --git a/tests/cases/standalone/common/copy/copy_to_fs.result b/tests/cases/standalone/copy/copy_to_fs.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_to_fs.result rename to tests/cases/standalone/copy/copy_to_fs.result diff --git a/tests/cases/standalone/common/copy/copy_to_fs.sql b/tests/cases/standalone/copy/copy_to_fs.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_to_fs.sql rename to tests/cases/standalone/copy/copy_to_fs.sql diff --git a/tests/cases/standalone/common/copy/copy_to_json_compressed.result b/tests/cases/standalone/copy/copy_to_json_compressed.result similarity index 100% rename from tests/cases/standalone/common/copy/copy_to_json_compressed.result rename to tests/cases/standalone/copy/copy_to_json_compressed.result diff --git a/tests/cases/standalone/common/copy/copy_to_json_compressed.sql b/tests/cases/standalone/copy/copy_to_json_compressed.sql similarity index 100% rename from tests/cases/standalone/common/copy/copy_to_json_compressed.sql rename to tests/cases/standalone/copy/copy_to_json_compressed.sql diff --git a/tests/cases/standalone/local_file_access.result b/tests/cases/standalone/local_file_access.result new file mode 100644 index 0000000000..f5cfebba24 --- /dev/null +++ b/tests/cases/standalone/local_file_access.result @@ -0,0 +1,55 @@ +CREATE TABLE local_file_access_source ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE +); + +Affected Rows: 0 + +INSERT INTO local_file_access_source VALUES + (1, 'host-1', 1.0), + (2, 'host-2', 2.0); + +Affected Rows: 2 + +COPY local_file_access_source TO 'local_file_access/table.parquet'; + +Affected Rows: 2 + +CREATE EXTERNAL TABLE local_file_access_external +WITH ( + location = 'local_file_access/table.parquet', + format = 'parquet' +); + +Affected Rows: 0 + +SELECT COUNT(*) FROM local_file_access_external; + ++----------+ +| count(*) | ++----------+ +| 2 | ++----------+ + +COPY (SELECT * FROM local_file_access_source) +TO 'local_file_access/query.parquet'; + +Affected Rows: 2 + +COPY DATABASE public TO 'local_file_access/database/'; + +Affected Rows: 4 + +COPY local_file_access_source FROM '../escape.parquet'; + +Error: 1004(InvalidArguments), Local filesystem path '../escape.parquet' is outside the configured copy root or is unsafe: '..' path components are not allowed; use a path relative to the copy root or use S3, OSS, GCS, or AzBlob + +DROP TABLE local_file_access_external; + +Affected Rows: 0 + +DROP TABLE local_file_access_source; + +Affected Rows: 0 + diff --git a/tests/cases/standalone/local_file_access.sql b/tests/cases/standalone/local_file_access.sql new file mode 100644 index 0000000000..a0d9676b9b --- /dev/null +++ b/tests/cases/standalone/local_file_access.sql @@ -0,0 +1,29 @@ +CREATE TABLE local_file_access_source ( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + val DOUBLE +); + +INSERT INTO local_file_access_source VALUES + (1, 'host-1', 1.0), + (2, 'host-2', 2.0); + +COPY local_file_access_source TO 'local_file_access/table.parquet'; + +CREATE EXTERNAL TABLE local_file_access_external +WITH ( + location = 'local_file_access/table.parquet', + format = 'parquet' +); + +SELECT COUNT(*) FROM local_file_access_external; + +COPY (SELECT * FROM local_file_access_source) +TO 'local_file_access/query.parquet'; + +COPY DATABASE public TO 'local_file_access/database/'; + +COPY local_file_access_source FROM '../escape.parquet'; + +DROP TABLE local_file_access_external; +DROP TABLE local_file_access_source; diff --git a/tests/conf/standalone-test.toml.template b/tests/conf/standalone-test.toml.template index 827cd2d0f6..88e3b1018c 100644 --- a/tests/conf/standalone-test.toml.template +++ b/tests/conf/standalone-test.toml.template @@ -24,6 +24,7 @@ broker_endpoints = {kafka_wal_broker_endpoints | unescaped} [storage] type = 'File' data_home = '{data_home}' +copy_root = '{copy_root}' [grpc] bind_addr = '{addrs.grpc_addr}' diff --git a/tests/runner/src/cmd/compat.rs b/tests/runner/src/cmd/compat.rs index 241d7b9eee..308d7f19fa 100644 --- a/tests/runner/src/cmd/compat.rs +++ b/tests/runner/src/cmd/compat.rs @@ -328,7 +328,10 @@ impl CompatCommand { .unwrap(); let sqlness_home = temp_dir.keep(); unsafe { - std::env::set_var("SQLNESS_HOME", sqlness_home.display().to_string()); + std::env::set_var( + "SQLNESS_HOME", + sqlness_home.join("copy").display().to_string(), + ); } // ---- 7. Build interceptor registry ---- diff --git a/tests/runner/src/env/bare.rs b/tests/runner/src/env/bare.rs index 7930bec243..9f9c0c7983 100644 --- a/tests/runner/src/env/bare.rs +++ b/tests/runner/src/env/bare.rs @@ -116,7 +116,10 @@ impl EnvController for Env { } unsafe { - std::env::set_var("SQLNESS_HOME", self.sqlness_home.display().to_string()); + std::env::set_var( + "SQLNESS_HOME", + self.sqlness_home.join("copy").display().to_string(), + ); } match mode { "standalone" => self.start_standalone(id).await, diff --git a/tests/runner/src/server_mode.rs b/tests/runner/src/server_mode.rs index 1f77f5d137..715d97f0d7 100644 --- a/tests/runner/src/server_mode.rs +++ b/tests/runner/src/server_mode.rs @@ -141,6 +141,7 @@ pub enum ServerMode { struct ConfigContext { wal_dir: String, data_home: String, + copy_root: String, procedure_dir: String, is_raft_engine: bool, kafka_wal_broker_endpoints: String, @@ -345,6 +346,7 @@ impl ServerMode { let ctx = ConfigContext { wal_dir, data_home: data_home.display().to_string(), + copy_root: sqlness_home.join("copy").display().to_string(), procedure_dir, is_raft_engine: db_ctx.is_raft_engine(), kafka_wal_broker_endpoints: db_ctx.kafka_wal_broker_endpoints(),