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