diff --git a/src/handlers/http/users/filters.rs b/src/handlers/http/users/filters.rs index 90a326240..92d5f52bb 100644 --- a/src/handlers/http/users/filters.rs +++ b/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, ) -> Result { + 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, Json(mut filter): Json, ) -> Result { + 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: '/' } + ) + )); + } +} diff --git a/src/storage/localfs.rs b/src/storage/localfs.rs index 97d909eb6..8bc642e8e 100644 --- a/src/storage/localfs.rs +++ b/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, ) -> 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?; diff --git a/src/storage/mod.rs b/src/storage/mod.rs index ac010edf6..fac2dd087 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -378,6 +378,12 @@ pub enum ObjectStorageError { #[error("MetastoreError: {0:?}")] MetastoreError(Box), + + #[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 {