Canonicalize filesystem paths in user-facing APIs (#2370) (#2371)

* Canonicalize LocalFileSystem::root (#2370)

* Canonicalize paths passed to Path::from_filesystem_path

* Add test

object_store/src/local.rs

@@ -18,7 +18,7 @@
 //! An object store implementation for a local filesystem
 use crate::{
     maybe_spawn_blocking,
-    path::{filesystem_path_to_url, Path},
+    path::{absolute_path_to_url, Path},
     GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore, Result,
 };
 use async_trait::async_trait;
@@ -129,6 +129,12 @@ pub(crate) enum Error {
         path: String,
         source: io::Error,
     },
+
+    #[snafu(display("Unable to canonicalize filesystem root: {}", path.display()))]
+    UnableToCanonicalize {
+        path: PathBuf,
+        source: io::Error,
+    },
 }
 
 impl From<Error> for super::Error {
@@ -214,17 +220,24 @@ impl LocalFileSystem {
     }
 
     /// Create new filesystem storage with `prefix` applied to all paths
+    ///
+    /// Returns an error if the path does not exist
+    ///
     pub fn new_with_prefix(prefix: impl AsRef<std::path::Path>) -> Result<Self> {
+        let path = std::fs::canonicalize(&prefix).context(UnableToCanonicalizeSnafu {
+            path: prefix.as_ref(),
+        })?;
+
         Ok(Self {
             config: Arc::new(Config {
-                root: filesystem_path_to_url(prefix)?,
+                root: absolute_path_to_url(path)?,
             }),
         })
     }
 }
 
 impl Config {
-    /// Return filesystem path of the given location
+    /// Return an absolute filesystem path of the given location
     fn path_to_filesystem(&self, location: &Path) -> Result<PathBuf> {
         let mut url = self.root.clone();
         url.path_segments_mut()
@@ -238,8 +251,9 @@ impl Config {
             .map_err(|_| Error::InvalidUrl { url }.into())
     }
 
+    /// Resolves the provided absolute filesystem path to a [`Path`] prefix
     fn filesystem_to_path(&self, location: &std::path::Path) -> Result<Path> {
-        Ok(Path::from_filesystem_path_with_base(
+        Ok(Path::from_absolute_path_with_base(
             location,
             Some(&self.root),
         )?)
@@ -1286,4 +1300,15 @@ mod tests {
         assert_eq!(res.objects.len(), 1);
         assert_eq!(res.objects[0].location.as_ref(), filename);
     }
+
+    #[tokio::test]
+    async fn relative_paths() {
+        LocalFileSystem::new_with_prefix(".").unwrap();
+        LocalFileSystem::new_with_prefix("..").unwrap();
+        LocalFileSystem::new_with_prefix("../..").unwrap();
+
+        let integration = LocalFileSystem::new();
+        let path = Path::from_filesystem_path(".").unwrap();
+        integration.list_with_delimiter(Some(&path)).await.unwrap();
+    }
 }

object_store/src/path/mod.rs

@@ -162,23 +162,38 @@ impl Path {
 
     /// Convert a filesystem path to a [`Path`] relative to the filesystem root
     ///
-    /// This will return an error if the path contains illegal
-    /// character sequences as defined by [`Path::parse`]
+    /// This will return an error if the path contains illegal character sequences
+    /// as defined by [`Path::parse`] or does not exist
+    ///
+    /// Note: this will canonicalize the provided path, resolving any symlinks
     pub fn from_filesystem_path(
         path: impl AsRef<std::path::Path>,
     ) -> Result<Self, Error> {
-        Self::from_filesystem_path_with_base(path, None)
+        let absolute = std::fs::canonicalize(&path).context(CanonicalizeSnafu {
+            path: path.as_ref(),
+        })?;
+
+        Self::from_absolute_path(absolute)
+    }
+
+    /// Convert an absolute filesystem path to a [`Path`] relative to the filesystem root
+    ///
+    /// This will return an error if the path contains illegal character sequences
+    /// as defined by [`Path::parse`], or `base` is not an absolute path
+    pub fn from_absolute_path(path: impl AsRef<std::path::Path>) -> Result<Self, Error> {
+        Self::from_absolute_path_with_base(path, None)
     }
 
     /// Convert a filesystem path to a [`Path`] relative to the provided base
     ///
     /// This will return an error if the path contains illegal character sequences
-    /// as defined by [`Path::parse`], or `base` does not refer to a parent path of `path`
-    pub(crate) fn from_filesystem_path_with_base(
+    /// as defined by [`Path::parse`], or `base` does not refer to a parent path of `path`,
+    /// or `base` is not an absolute path
+    pub(crate) fn from_absolute_path_with_base(
         path: impl AsRef<std::path::Path>,
         base: Option<&Url>,
     ) -> Result<Self, Error> {
-        let url = filesystem_path_to_url(path)?;
+        let url = absolute_path_to_url(path)?;
         let path = match base {
             Some(prefix) => url.path().strip_prefix(prefix.path()).ok_or_else(|| {
                 Error::PrefixMismatch {
@@ -293,8 +308,8 @@ where
     }
 }
 
-/// Given a filesystem path convert it to a URL representation
-pub(crate) fn filesystem_path_to_url(
+/// Given an absolute filesystem path convert it to a URL representation without canonicalization
+pub(crate) fn absolute_path_to_url(
     path: impl AsRef<std::path::Path>,
 ) -> Result<Url, Error> {
     Url::from_file_path(&path).map_err(|_| Error::InvalidPath {