fix(storage): block path traversal in filters API on localfs

src/handlers/http/users/filters.rs

@@ -20,11 +20,12 @@ use crate::{
     handlers::http::rbac::RBACError,
     metastore::MetastoreError,
     parseable::PARSEABLE,
-    storage::ObjectStorageError,
+    storage::{ObjectStorageError, StreamType},
     users::filters::{CURRENT_FILTER_VERSION, FILTERS, Filter},
     utils::{
         actix::extract_session_key_from_req, get_hash, get_user_and_tenant_from_request, is_admin,
     },
+    validator,
 };
 use actix_web::http::StatusCode;
 use actix_web::{
@@ -65,6 +66,8 @@ pub async fn post(
     req: HttpRequest,
     Json(mut filter): Json<Filter>,
 ) -> Result<impl Responder, FiltersError> {
+    validate_stream_name(&filter.stream_name)?;
+
     let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?;
     user_id = get_hash(&user_id);
     let filter_id = Ulid::new().to_string();
@@ -83,6 +86,8 @@ pub async fn update(
     filter_id: Path<String>,
     Json(mut filter): Json<Filter>,
 ) -> Result<impl Responder, FiltersError> {
+    validate_stream_name(&filter.stream_name)?;
+
     let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?;
     user_id = get_hash(&user_id);
     let filter_id = filter_id.into_inner();
@@ -132,6 +137,11 @@ pub async fn delete(
     Ok(HttpResponse::Ok().finish())
 }
 
+fn validate_stream_name(stream_name: &str) -> Result<(), FiltersError> {
+    validator::stream_name(stream_name, StreamType::UserDefined)
+        .map_err(FiltersError::InvalidStreamName)
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum FiltersError {
     #[error("Failed to connect to storage: {0}")]
@@ -144,6 +154,8 @@ pub enum FiltersError {
     UserDoesNotExist(#[from] RBACError),
     #[error("Error: {0}")]
     Custom(String),
+    #[error("Invalid stream name: {0}")]
+    InvalidStreamName(#[from] validator::error::StreamNameValidationError),
     #[error(transparent)]
     MetastoreError(#[from] MetastoreError),
 }
@@ -156,6 +168,7 @@ impl actix_web::ResponseError for FiltersError {
             Self::Metadata(_) => StatusCode::BAD_REQUEST,
             Self::UserDoesNotExist(_) => StatusCode::NOT_FOUND,
             Self::Custom(_) => StatusCode::INTERNAL_SERVER_ERROR,
+            Self::InvalidStreamName(_) => StatusCode::BAD_REQUEST,
             Self::MetastoreError(e) => e.status_code(),
         }
     }
@@ -173,3 +186,28 @@ impl actix_web::ResponseError for FiltersError {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::validate_stream_name;
+    use crate::validator::error::StreamNameValidationError;
+
+    #[test]
+    fn accepts_flat_stream_names() {
+        assert!(validate_stream_name("team1").is_ok());
+        assert!(validate_stream_name("orders-prod").is_ok());
+        assert!(validate_stream_name("user_events_v2").is_ok());
+    }
+
+    #[test]
+    fn rejects_path_like_stream_names() {
+        let err =
+            validate_stream_name("team1/serviceA").expect_err("path-like names must be rejected");
+        assert!(matches!(
+            err,
+            super::FiltersError::InvalidStreamName(
+                StreamNameValidationError::NameSpecialChar { c: '/' }
+            )
+        ));
+    }
+}

src/storage/localfs.rs

@@ -128,6 +128,7 @@ impl ObjectStorage for LocalFS {
             }
         }
     }
+
     async fn upload_multipart(
         &self,
         key: &RelativePath,
@@ -139,6 +140,7 @@ impl ObjectStorage for LocalFS {
         file.read_to_end(&mut data).await?;
         self.put_object(key, data.into(), tenant_id).await
     }
+
     async fn get_buffered_reader(
         &self,
         _path: &RelativePath,
@@ -151,6 +153,7 @@ impl ObjectStorage for LocalFS {
             ),
         )))
     }
+
     async fn head(
         &self,
         path: &RelativePath,
@@ -181,6 +184,7 @@ impl ObjectStorage for LocalFS {
             Err(e) => Err(ObjectStorageError::IoError(e)),
         }
     }
+
     async fn get_object(
         &self,
         path: &RelativePath,
@@ -424,6 +428,16 @@ impl ObjectStorage for LocalFS {
         path: &RelativePath,
         tenant_id: &Option<String>,
     ) -> Result<(), ObjectStorageError> {
+        if path
+            .components()
+            .any(|component| matches!(component, relative_path::Component::ParentDir))
+        {
+            return Err(ObjectStorageError::PathTraversal {
+                attempted: path.to_path(&self.root),
+                root: self.root.clone(),
+            });
+        }
+
         let path = self.path_in_root(path);
         let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
 
@@ -735,6 +749,7 @@ impl ObjectStorage for LocalFS {
             ..CopyOptions::default()
         };
         let to_path = self.root.join(key);
+
         let tenant_str = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
         if let Some(path) = to_path.parent() {
             fs::create_dir_all(path).await?;

src/storage/mod.rs

@@ -378,6 +378,12 @@ pub enum ObjectStorageError {
 
     #[error("MetastoreError: {0:?}")]
     MetastoreError(Box<MetastoreErrorDetail>),
+
+    #[error("Path traversal detected: attempted path '{attempted}' escapes root '{root}'")]
+    PathTraversal {
+        attempted: std::path::PathBuf,
+        root: std::path::PathBuf,
+    },
 }
 
 pub fn to_object_store_path(path: &RelativePath) -> Path {