common_datasource/
object_store.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15pub mod azblob;
16pub mod fs;
17pub mod gcs;
18pub mod oss;
19pub mod s3;
20
21use std::collections::HashMap;
22use std::path::{Component, Path, PathBuf};
23use std::sync::Arc;
24
25use common_telemetry::debug;
26use lazy_static::lazy_static;
27use object_store::ObjectStore;
28use object_store::secure_fs::SecureFsRoot;
29use regex::Regex;
30use snafu::{OptionExt, ResultExt};
31use url::{ParseError, Url};
32
33use self::azblob::build_azblob_backend;
34use self::fs::build_fs_backend;
35use self::gcs::build_gcs_backend;
36use self::s3::build_s3_backend;
37use crate::error::{self, Result};
38use crate::object_store::oss::build_oss_backend;
39use crate::util::find_dir_and_filename;
40
41pub const FS_SCHEMA: &str = "FS";
42pub const FILE_SCHEMA: &str = "FILE";
43pub const S3_SCHEMA: &str = "S3";
44pub const OSS_SCHEMA: &str = "OSS";
45pub const GCS_SCHEMA: &str = "GCS";
46pub const AZBLOB_SCHEMA: &str = "AZBLOB";
47
48/// An object store rooted at the target's parent, together with the optional
49/// target path relative to that root.
50pub struct BuiltBackend {
51    pub object_store: ObjectStore,
52    pub object_path: Option<String>,
53    local_root: Option<SecureFsRoot>,
54}
55
56impl BuiltBackend {
57    /// Checks an explicit input using the same file-type filter as directory listing.
58    pub async fn is_file(&self, path: &str) -> object_store::Result<bool> {
59        let meta = self.object_store.stat(path).await?;
60        self.is_file_with_mode(path, meta.mode()).await
61    }
62
63    /// Applies local file-type protections to a previously fetched object stat.
64    pub async fn is_file_with_mode(
65        &self,
66        path: &str,
67        mode: object_store::EntryMode,
68    ) -> object_store::Result<bool> {
69        if mode != object_store::EntryMode::FILE {
70            return Ok(false);
71        }
72        if let Some(root) = &self.local_root {
73            root.is_file(path).await
74        } else {
75            Ok(true)
76        }
77    }
78}
79
80/// Controls whether SQL paths may access the local filesystem.
81#[derive(Clone, Debug, Default)]
82pub enum LocalFileAccess {
83    /// Local filesystem paths are rejected.
84    #[default]
85    Disabled,
86    /// Local filesystem paths are confined to a server-configured root.
87    Sandboxed { root: LocalFileRoot },
88}
89
90/// An opened server-controlled root for sandboxed SQL file access.
91#[derive(Clone, Debug)]
92pub struct LocalFileRoot {
93    root: Arc<SecureFsRoot>,
94    configured_path: Arc<PathBuf>,
95}
96
97impl LocalFileAccess {
98    /// Creates a sandbox rooted at a server-controlled local directory.
99    pub fn sandboxed(root: impl AsRef<Path>) -> Result<Self> {
100        let root_path = root.as_ref();
101        let configured_path =
102            std::path::absolute(root_path).with_context(|_| error::InvalidLocalFileRootSnafu {
103                root: root_path.display().to_string(),
104            })?;
105        let root =
106            SecureFsRoot::open(root_path).with_context(|_| error::InvalidLocalFileRootSnafu {
107                root: root_path.display().to_string(),
108            })?;
109        Ok(Self::Sandboxed {
110            root: LocalFileRoot {
111                root: Arc::new(root),
112                configured_path: Arc::new(configured_path),
113            },
114        })
115    }
116
117    /// Returns the canonical path of the configured sandbox root.
118    pub fn sandbox_root(&self) -> Option<&Path> {
119        match self {
120            Self::Disabled => None,
121            Self::Sandboxed { root } => Some(root.root.path()),
122        }
123    }
124
125    fn authorize(&self, location: &str, path: &Path, trailing_slash: bool) -> Result<String> {
126        let LocalFileAccess::Sandboxed { root } = self else {
127            return error::LocalFileAccessDisabledSnafu {
128                path: location.to_string(),
129            }
130            .fail();
131        };
132
133        let path = normalize_untrusted_path(path).map_err(|reason| {
134            error::LocalFileAccessDeniedSnafu {
135                path: location.to_string(),
136                reason,
137            }
138            .build()
139        })?;
140        let relative = if path.is_absolute() {
141            strip_local_prefix(&path, root.configured_path.as_path())
142                .or_else(|| strip_local_prefix(&path, root.root.path()))
143                .ok_or_else(|| {
144                    error::LocalFileAccessDeniedSnafu {
145                        path: location.to_string(),
146                        reason: "absolute path is outside the configured copy root".to_string(),
147                    }
148                    .build()
149                })?
150        } else {
151            path.as_path()
152        };
153
154        let mut authorized = relative
155            .components()
156            .filter_map(|component| match component {
157                Component::CurDir => None,
158                Component::Normal(value) => Some(value.to_string_lossy().into_owned()),
159                _ => None,
160            })
161            .collect::<Vec<_>>()
162            .join("/");
163        if trailing_slash && !authorized.is_empty() {
164            authorized.push('/');
165        }
166        Ok(authorized)
167    }
168
169    async fn open_backend_root(
170        &self,
171        location: &str,
172        relative_root: &str,
173        create: bool,
174    ) -> Result<SecureFsRoot> {
175        let LocalFileAccess::Sandboxed { root } = self else {
176            return error::LocalFileAccessDisabledSnafu {
177                path: location.to_string(),
178            }
179            .fail();
180        };
181
182        let root = root.root.clone();
183        let relative_root = relative_root.trim_matches('/').to_string();
184        common_runtime::spawn_blocking_global(move || {
185            if create {
186                root.create_subdir(relative_root)
187            } else {
188                root.open_subdir(relative_root)
189            }
190        })
191        .await
192        .context(error::JoinHandleSnafu)?
193        .map_err(|error| {
194            debug!(
195                "Failed to open an authorized local SQL path inside the copy root, path: {location}, error: {error:?}"
196            );
197            if error.kind() == std::io::ErrorKind::NotFound {
198                return error::LocalFilePathNotFoundSnafu { path: location }.build();
199            }
200            error::LocalFileAccessDeniedSnafu {
201                path: location.to_string(),
202                reason: "path could not be safely resolved within the configured copy root"
203                    .to_string(),
204            }
205            .build()
206        })
207    }
208}
209
210/// Converts a configured location into a local path.
211///
212/// Bare paths and `file://` URLs are local. Other URL schemes return `None`.
213pub fn configured_local_path(location: &str) -> Result<Option<PathBuf>> {
214    #[cfg(windows)]
215    if Path::new(location).is_absolute() {
216        return Ok(Some(PathBuf::from(location)));
217    }
218
219    let (schema, _, path) = parse_url(location)?;
220    match schema.to_uppercase().as_str() {
221        FS_SCHEMA => Ok(Some(PathBuf::from(path))),
222        FILE_SCHEMA => {
223            let url = Url::parse(location).context(error::InvalidUrlSnafu { url: location })?;
224            url.to_file_path().map(Some).map_err(|_| {
225                error::InvalidLocalFileRootConfigSnafu {
226                    root: location.to_string(),
227                    reason: "file URL must contain a local absolute path".to_string(),
228                }
229                .build()
230            })
231        }
232        _ => Ok(None),
233    }
234}
235
236fn strip_local_prefix<'a>(path: &'a Path, prefix: &Path) -> Option<&'a Path> {
237    #[cfg(not(windows))]
238    {
239        path.strip_prefix(prefix).ok()
240    }
241
242    #[cfg(windows)]
243    {
244        let mut path_components = path.components();
245        for prefix_component in prefix.components() {
246            let path_component = path_components.next()?;
247            if !windows_component_eq(path_component, prefix_component) {
248                return None;
249            }
250        }
251        Some(path_components.as_path())
252    }
253}
254
255#[cfg(windows)]
256fn windows_component_eq(left: Component<'_>, right: Component<'_>) -> bool {
257    match (left, right) {
258        (Component::Prefix(left), Component::Prefix(right)) => {
259            windows_os_str_eq(left.as_os_str(), right.as_os_str())
260        }
261        (Component::Normal(left), Component::Normal(right)) => windows_os_str_eq(left, right),
262        (Component::RootDir, Component::RootDir)
263        | (Component::CurDir, Component::CurDir)
264        | (Component::ParentDir, Component::ParentDir) => true,
265        _ => false,
266    }
267}
268
269#[cfg(windows)]
270fn windows_os_str_eq(left: &std::ffi::OsStr, right: &std::ffi::OsStr) -> bool {
271    use std::os::windows::ffi::OsStrExt;
272
273    fn ascii_lowercase(value: u16) -> u16 {
274        if (u16::from(b'A')..=u16::from(b'Z')).contains(&value) {
275            value + u16::from(b'a' - b'A')
276        } else {
277            value
278        }
279    }
280
281    left.encode_wide()
282        .map(ascii_lowercase)
283        .eq(right.encode_wide().map(ascii_lowercase))
284}
285
286fn normalize_untrusted_path(path: &Path) -> std::result::Result<PathBuf, String> {
287    let mut normalized = PathBuf::new();
288    for component in path.components() {
289        match component {
290            Component::Prefix(prefix) => normalized.push(prefix.as_os_str()),
291            Component::RootDir => normalized.push(Path::new(std::path::MAIN_SEPARATOR_STR)),
292            Component::CurDir => {}
293            Component::Normal(value) => normalized.push(value),
294            Component::ParentDir => return Err("'..' path components are not allowed".to_string()),
295        }
296    }
297    Ok(normalized)
298}
299
300/// Returns `(schema, Option<host>, path)`
301pub fn parse_url(url: &str) -> Result<(String, Option<String>, String)> {
302    #[cfg(windows)]
303    {
304        // On Windows, the URL may start with `C:/` or `C:\`.
305        if handle_windows_path(url).is_some() {
306            return Ok((FS_SCHEMA.to_string(), None, url.to_string()));
307        }
308    }
309    let parsed_url = Url::parse(url);
310    match parsed_url {
311        Ok(url) => Ok((
312            url.scheme().to_string(),
313            url.host_str().map(|s| s.to_string()),
314            url.path().to_string(),
315        )),
316        Err(ParseError::RelativeUrlWithoutBase) => {
317            Ok((FS_SCHEMA.to_string(), None, url.to_string()))
318        }
319        Err(err) => Err(err).context(error::InvalidUrlSnafu { url }),
320    }
321}
322
323pub async fn build_backend(
324    url: &str,
325    connection: &HashMap<String, String>,
326    local_file_access: &LocalFileAccess,
327) -> Result<ObjectStore> {
328    Ok(
329        build_backend_inner(url, connection, local_file_access, false, false)
330            .await?
331            .object_store,
332    )
333}
334
335/// Builds a backend and returns the target path relative to the backend root.
336pub async fn build_backend_with_path(
337    url: &str,
338    connection: &HashMap<String, String>,
339    local_file_access: &LocalFileAccess,
340) -> Result<BuiltBackend> {
341    build_backend_inner(url, connection, local_file_access, false, false).await
342}
343
344/// Builds a backend for an operation that may create the target directory.
345pub async fn build_backend_for_write(
346    url: &str,
347    connection: &HashMap<String, String>,
348    local_file_access: &LocalFileAccess,
349) -> Result<ObjectStore> {
350    Ok(
351        build_backend_inner(url, connection, local_file_access, true, false)
352            .await?
353            .object_store,
354    )
355}
356
357/// Builds a writable backend and returns the target path relative to the backend root.
358pub async fn build_backend_for_write_with_path(
359    url: &str,
360    connection: &HashMap<String, String>,
361    local_file_access: &LocalFileAccess,
362) -> Result<BuiltBackend> {
363    build_backend_inner(url, connection, local_file_access, true, true).await
364}
365
366async fn build_backend_inner(
367    url: &str,
368    connection: &HashMap<String, String>,
369    local_file_access: &LocalFileAccess,
370    create_local_root: bool,
371    require_object_path: bool,
372) -> Result<BuiltBackend> {
373    let (schema, host, path) = parse_url(url)?;
374    let normalized_schema = schema.to_uppercase();
375
376    if normalized_schema == FS_SCHEMA || normalized_schema == FILE_SCHEMA {
377        let (local_path, trailing_slash) = if normalized_schema == FILE_SCHEMA {
378            let url = Url::parse(url).context(error::InvalidUrlSnafu { url })?;
379            let path = url.to_file_path().map_err(|_| {
380                error::LocalFileAccessDeniedSnafu {
381                    path: url.to_string(),
382                    reason: "file URL must contain a local absolute path".to_string(),
383                }
384                .build()
385            })?;
386            (path, url.path().ends_with('/'))
387        } else {
388            (
389                PathBuf::from(&path),
390                path.ends_with('/') || cfg!(windows) && path.ends_with(std::path::MAIN_SEPARATOR),
391            )
392        };
393        let authorized = local_file_access.authorize(url, &local_path, trailing_slash)?;
394        let (root, object_path) = find_dir_and_filename(&authorized);
395        if require_object_path && object_path.is_none() {
396            return error::MissingObjectNameSnafu {
397                path: url.to_string(),
398            }
399            .fail();
400        }
401        let root = local_file_access
402            .open_backend_root(url, &root, create_local_root)
403            .await?;
404        return Ok(BuiltBackend {
405            object_store: build_fs_backend(&root)?,
406            object_path,
407            local_root: Some(root),
408        });
409    }
410
411    let (root, object_path) = find_dir_and_filename(&path);
412
413    let object_store = match normalized_schema.as_str() {
414        S3_SCHEMA => {
415            let host = host.context(error::EmptyHostPathSnafu {
416                url: url.to_string(),
417            })?;
418            build_s3_backend(&host, &root, connection)?
419        }
420        OSS_SCHEMA => {
421            let host = host.context(error::EmptyHostPathSnafu {
422                url: url.to_string(),
423            })?;
424            build_oss_backend(&host, &root, connection)?
425        }
426        GCS_SCHEMA => {
427            let host = host.context(error::EmptyHostPathSnafu {
428                url: url.to_string(),
429            })?;
430            build_gcs_backend(&host, &root, connection)?
431        }
432        AZBLOB_SCHEMA => {
433            let host = host.context(error::EmptyHostPathSnafu {
434                url: url.to_string(),
435            })?;
436            build_azblob_backend(&host, &root, connection)?
437        }
438        _ => error::UnsupportedBackendProtocolSnafu {
439            protocol: schema,
440            url,
441        }
442        .fail()?,
443    };
444    Ok(BuiltBackend {
445        object_store,
446        object_path,
447        local_root: None,
448    })
449}
450
451lazy_static! {
452    static ref DISK_SYMBOL_PATTERN: Regex = Regex::new(r"^([A-Za-z]:[/\\])").unwrap();
453}
454
455pub fn handle_windows_path(url: &str) -> Option<String> {
456    DISK_SYMBOL_PATTERN
457        .captures(url)
458        .map(|captures| captures[0].to_string())
459}
460
461#[cfg(test)]
462mod tests {
463    use std::collections::HashMap;
464    use std::fs;
465
466    use common_error::ext::{ErrorExt, RetryHint};
467    use common_error::status_code::StatusCode;
468    use common_test_util::temp_dir::create_temp_dir;
469    use url::Url;
470
471    use super::{
472        LocalFileAccess, build_backend, build_backend_for_write, build_backend_for_write_with_path,
473        build_backend_with_path, handle_windows_path,
474    };
475    use crate::error::Error;
476
477    #[test]
478    fn test_handle_windows_path() {
479        assert_eq!(
480            handle_windows_path("C:/to/path/file"),
481            Some("C:/".to_string())
482        );
483        assert_eq!(
484            handle_windows_path(r"C:\to\path\file"),
485            Some(r"C:\".to_string())
486        );
487        assert_eq!(handle_windows_path("https://google.com"), None);
488        assert_eq!(handle_windows_path("s3://bucket/path/to"), None);
489    }
490
491    #[cfg(windows)]
492    #[test]
493    fn test_windows_local_path_detection_and_prefix() {
494        use std::path::{Path, PathBuf};
495
496        let location = r"C:\gtdata";
497        assert_eq!(
498            super::configured_local_path(location).unwrap(),
499            Some(PathBuf::from(location))
500        );
501        assert_eq!(
502            super::parse_url(location).unwrap(),
503            ("FS".to_string(), None, location.to_string())
504        );
505        assert_eq!(
506            super::strip_local_prefix(
507                Path::new(r"c:\Data\Copy\nested\data.parquet"),
508                Path::new(r"C:\data\copy"),
509            ),
510            Some(Path::new(r"nested\data.parquet"))
511        );
512    }
513
514    #[tokio::test]
515    async fn test_local_file_access_policy() {
516        let data_home = create_temp_dir("local_file_access_policy");
517        let copy_root = data_home.path().join("copy");
518        let internal_dir = data_home.path().join("data");
519        fs::create_dir_all(&internal_dir).unwrap();
520        fs::write(internal_dir.join("secret"), "secret").unwrap();
521
522        let access = LocalFileAccess::sandboxed(&copy_root).unwrap();
523        let connection = HashMap::new();
524
525        let store = build_backend_for_write("nested/data.txt", &connection, &access)
526            .await
527            .unwrap();
528        store.write("data.txt", "first").await.unwrap();
529        store.write("data.txt", "second").await.unwrap();
530        assert_eq!(
531            fs::read_to_string(copy_root.join("nested/data.txt")).unwrap(),
532            "second"
533        );
534
535        let missing = copy_root.join("missing/directory");
536        let error = build_backend("missing/directory/data.txt", &connection, &access)
537            .await
538            .unwrap_err();
539        assert!(matches!(&error, Error::LocalFilePathNotFound { .. }));
540        assert_eq!(error.status_code(), StatusCode::InvalidArguments);
541        assert_eq!(error.retry_hint(), RetryHint::NonRetryable);
542        assert!(
543            error.to_string().contains("does not exist"),
544            "unexpected error: {error}"
545        );
546        assert!(!missing.exists());
547
548        let absolute = copy_root.join("nested/data.txt");
549        let store = build_backend(absolute.to_str().unwrap(), &connection, &access)
550            .await
551            .unwrap();
552        assert_eq!(store.read("data.txt").await.unwrap().to_vec(), b"second");
553
554        let file_url = Url::from_file_path(&absolute).unwrap().to_string();
555        let store = build_backend(&file_url, &connection, &access)
556            .await
557            .unwrap();
558        assert_eq!(store.read("data.txt").await.unwrap().to_vec(), b"second");
559
560        let internal_file = internal_dir.join("secret");
561        assert!(matches!(
562            build_backend(internal_file.to_str().unwrap(), &connection, &access).await,
563            Err(Error::LocalFileAccessDenied { .. })
564        ));
565        assert!(
566            build_backend("../escape/data.txt", &connection, &access)
567                .await
568                .is_err()
569        );
570
571        let outside = data_home.path().join("outside/new");
572        assert!(
573            build_backend(outside.to_str().unwrap(), &connection, &access)
574                .await
575                .is_err()
576        );
577        assert!(!outside.parent().unwrap().exists());
578
579        let prefix_escape = data_home.path().join("copy-not-the-root/new");
580        assert!(matches!(
581            build_backend(prefix_escape.to_str().unwrap(), &connection, &access).await,
582            Err(Error::LocalFileAccessDenied { .. })
583        ));
584        assert!(!prefix_escape.parent().unwrap().exists());
585
586        let disabled = LocalFileAccess::Disabled;
587        let error = build_backend(internal_file.to_str().unwrap(), &connection, &disabled)
588            .await
589            .unwrap_err();
590        assert!(matches!(&error, Error::LocalFileAccessDisabled { .. }));
591        assert_eq!(error.status_code(), StatusCode::InvalidArguments);
592        assert_eq!(error.retry_hint(), RetryHint::NonRetryable);
593        assert!(matches!(
594            build_backend(&file_url, &connection, &disabled).await,
595            Err(Error::LocalFileAccessDisabled { .. })
596        ));
597        assert!(matches!(
598            build_backend(outside.to_str().unwrap(), &connection, &disabled).await,
599            Err(Error::LocalFileAccessDisabled { .. })
600        ));
601        assert!(!outside.parent().unwrap().exists());
602    }
603
604    #[tokio::test]
605    async fn test_file_url_returns_decoded_backend_relative_path() {
606        let temp_dir = create_temp_dir("file_url_backend_relative_path");
607        let copy_root = temp_dir.path().join("copy root");
608        let file = copy_root.join("nested dir/data file.txt");
609        fs::create_dir_all(file.parent().unwrap()).unwrap();
610        fs::write(&file, "data").unwrap();
611
612        let location = Url::from_file_path(&file).unwrap().to_string();
613        assert!(location.contains("%20"));
614        let access = LocalFileAccess::sandboxed(&copy_root).unwrap();
615        let backend = build_backend_with_path(&location, &HashMap::new(), &access)
616            .await
617            .unwrap();
618
619        assert_eq!(backend.object_path.as_deref(), Some("data file.txt"));
620        assert_eq!(
621            backend
622                .object_store
623                .read(backend.object_path.as_deref().unwrap())
624                .await
625                .unwrap()
626                .to_vec(),
627            b"data"
628        );
629    }
630
631    #[tokio::test]
632    async fn test_write_with_path_rejects_directory_before_creation() {
633        let temp_dir = create_temp_dir("write_with_path_rejects_directory");
634        let copy_root = temp_dir.path().join("copy");
635        let target = copy_root.join("new directory");
636        let location = Url::from_directory_path(&target).unwrap().to_string();
637        let access = LocalFileAccess::sandboxed(&copy_root).unwrap();
638
639        let result = build_backend_for_write_with_path(&location, &HashMap::new(), &access).await;
640
641        assert!(matches!(result, Err(Error::MissingObjectName { .. })));
642        assert!(!target.exists());
643
644        let result = build_backend_for_write_with_path(
645            &location,
646            &HashMap::new(),
647            &LocalFileAccess::Disabled,
648        )
649        .await;
650        assert!(matches!(result, Err(Error::LocalFileAccessDisabled { .. })));
651        assert!(!target.exists());
652
653        let relative_target = copy_root.join("relative directory");
654        let result =
655            build_backend_for_write_with_path("relative directory/", &HashMap::new(), &access)
656                .await;
657        assert!(matches!(result, Err(Error::MissingObjectName { .. })));
658        assert!(!relative_target.exists());
659
660        let allowed_directory = copy_root.join("allowed directory");
661        build_backend_for_write("allowed directory/", &HashMap::new(), &access)
662            .await
663            .unwrap();
664        assert!(allowed_directory.is_dir());
665    }
666
667    #[cfg(windows)]
668    #[tokio::test]
669    async fn test_windows_backslash_path_returns_backend_relative_path() {
670        let temp_dir = create_temp_dir("windows_backend_relative_path");
671        let copy_root = temp_dir.path().join("copy");
672        let directory = copy_root.join("nested");
673        let file = directory.join("data.txt");
674        fs::create_dir_all(&directory).unwrap();
675        fs::write(&file, "data").unwrap();
676
677        let access = LocalFileAccess::sandboxed(&copy_root).unwrap();
678        let connection = HashMap::new();
679        let file_location = file.to_str().unwrap();
680        assert!(file_location.contains('\\'));
681        let backend = build_backend_with_path(file_location, &connection, &access)
682            .await
683            .unwrap();
684        assert_eq!(backend.object_path.as_deref(), Some("data.txt"));
685        assert_eq!(
686            backend
687                .object_store
688                .read(backend.object_path.as_deref().unwrap())
689                .await
690                .unwrap()
691                .to_vec(),
692            b"data"
693        );
694
695        let new_directory = copy_root.join("new directory");
696        let directory_location = format!("{}\\", new_directory.display());
697        assert!(matches!(
698            build_backend_for_write_with_path(&directory_location, &connection, &access).await,
699            Err(Error::MissingObjectName { .. })
700        ));
701        assert!(!new_directory.exists());
702    }
703
704    #[tokio::test]
705    async fn test_object_storage_ignores_local_file_policy() {
706        let cases = [
707            (
708                "s3://bucket/path/data%20file.parquet",
709                HashMap::from([
710                    ("region".to_string(), "us-east-1".to_string()),
711                    ("disable_ec2_metadata".to_string(), "true".to_string()),
712                ]),
713                "data%20file.parquet",
714            ),
715            (
716                "oss://bucket/path/file.parquet",
717                HashMap::from([
718                    ("endpoint".to_string(), "http://oss.example.com".to_string()),
719                    ("allow_anonymous".to_string(), "true".to_string()),
720                ]),
721                "file.parquet",
722            ),
723            (
724                "gcs://bucket/path/file.parquet",
725                HashMap::from([(
726                    "endpoint".to_string(),
727                    "http://storage.example.com".to_string(),
728                )]),
729                "file.parquet",
730            ),
731            (
732                "azblob://container/path/file.parquet",
733                HashMap::from([
734                    (
735                        "endpoint".to_string(),
736                        "http://storage.example.com".to_string(),
737                    ),
738                    ("account_name".to_string(), "test".to_string()),
739                ]),
740                "file.parquet",
741            ),
742        ];
743        for (location, connection, expected_path) in cases {
744            let backend = build_backend_for_write_with_path(
745                location,
746                &connection,
747                &LocalFileAccess::Disabled,
748            )
749            .await
750            .unwrap();
751            assert_eq!(backend.object_path.as_deref(), Some(expected_path));
752        }
753    }
754
755    #[cfg(unix)]
756    #[tokio::test]
757    async fn test_local_file_access_rejects_symlink_escape() {
758        use std::os::unix::fs::symlink;
759
760        let temp_dir = create_temp_dir("local_file_access_symlink");
761        let copy_root = temp_dir.path().join("copy");
762        let outside = temp_dir.path().join("outside");
763        fs::create_dir_all(&copy_root).unwrap();
764        fs::create_dir_all(&outside).unwrap();
765        fs::write(outside.join("secret"), "secret").unwrap();
766        symlink(&outside, copy_root.join("escape")).unwrap();
767        symlink(outside.join("secret"), copy_root.join("secret-link")).unwrap();
768
769        let access = LocalFileAccess::sandboxed(&copy_root).unwrap();
770        let connection = HashMap::new();
771
772        assert!(
773            build_backend("escape/secret", &connection, &access)
774                .await
775                .is_err()
776        );
777        assert!(
778            build_backend_for_write("escape/new", &connection, &access)
779                .await
780                .is_err()
781        );
782        assert!(!outside.join("new").exists());
783
784        let store = build_backend("secret-link", &connection, &access)
785            .await
786            .unwrap();
787        assert!(store.read("secret-link").await.is_err());
788        assert!(store.write("secret-link", "overwritten").await.is_err());
789        assert_eq!(
790            fs::read_to_string(outside.join("secret")).unwrap(),
791            "secret"
792        );
793    }
794}
common_datasource/object_store.rs

common_datasource/
object_store.rs
