This is an automated email from the ASF dual-hosted git repository. erickguan pushed a commit to branch fix-stat-impl in repository https://gitbox.apache.org/repos/asf/opendal.git
commit eca6ec57ac496e3ba8204633d7af1d37d24510ad Author: Erick Guan <[email protected]> AuthorDate: Sun Jul 5 11:25:21 2026 +0800 harden sql services and improve d1 and gridfs --- core/services/d1/src/core.rs | 26 ++++++--- core/services/gridfs/src/backend.rs | 7 +-- core/services/gridfs/src/core.rs | 29 +++++++--- core/services/mysql/src/core.rs | 12 +++- core/services/postgresql/src/core.rs | 10 +++- core/services/redis/src/backend.rs | 1 + core/services/sqlite/src/backend.rs | 9 +-- core/services/sqlite/src/core.rs | 104 +++++++++++++++++++++++++---------- 8 files changed, 140 insertions(+), 58 deletions(-) diff --git a/core/services/d1/src/core.rs b/core/services/d1/src/core.rs index 96343b4ca..b4c8bfc39 100644 --- a/core/services/d1/src/core.rs +++ b/core/services/d1/src/core.rs @@ -47,6 +47,8 @@ impl Debug for D1Core { } } +const CLOUDFLARE_API_BASE_URL: &str = "https://api.cloudflare.com/client/v4";; + impl D1Core { fn create_d1_query_request( &self, @@ -61,7 +63,7 @@ impl D1Core { ); let url: String = format!( "{}{}", - "https://api.cloudflare.com/client/v4";, + CLOUDFLARE_API_BASE_URL, percent_encode_path(&p) ); @@ -84,8 +86,13 @@ impl D1Core { } pub async fn get(&self, ctx: &OperationContext, path: &str) -> Result<Option<Buffer>> { + // d1 follows SQLite SQL syntax, we use `"` for identifier quote. A quoted identifier is case-sensitive and + // can contain special characters. + // Read more https://www.sqlite.org/quirks.html#double_quoted_string_literals + // + // We uses identifier quote for trusted table and field configuration to ensure correctness. let query = format!( - "SELECT {} FROM {} WHERE {} = ? LIMIT 1", + r#"SELECT "{}" FROM "{}" WHERE "{}" = ? LIMIT 1"#, self.value_field, self.table, self.key_field ); let req = @@ -106,7 +113,7 @@ impl D1Core { pub async fn get_length(&self, ctx: &OperationContext, path: &str) -> Result<Option<usize>> { let query = format!( - "SELECT LENGTH(CAST({} AS BLOB)) AS content_length FROM {} WHERE {} = ? LIMIT 1", + r#"SELECT LENGTH(CAST("{}" AS BLOB)) AS "content_length" FROM "{}" WHERE "{}" = ? LIMIT 1"#, self.value_field, self.table, self.key_field ); let req = @@ -130,10 +137,10 @@ impl D1Core { let key_field = &self.key_field; let value_field = &self.value_field; let query = format!( - "INSERT INTO {table} ({key_field}, {value_field}) \ - VALUES (?, ?) \ - ON CONFLICT ({key_field}) \ - DO UPDATE SET {value_field} = EXCLUDED.{value_field}", + r#"INSERT INTO "{table}" ("{key_field}", "{value_field}") \ + VALUES ('?', ?) \ + ON CONFLICT ("{key_field}") \ + DO UPDATE SET "{value_field}" = EXCLUDED."{value_field}""#, ); let params = vec![path.into(), value.to_vec().into()]; @@ -148,7 +155,10 @@ impl D1Core { } pub async fn delete(&self, ctx: &OperationContext, path: &str) -> Result<()> { - let query = format!("DELETE FROM {} WHERE {} = ?", self.table, self.key_field); + let query = format!( + r#"DELETE FROM "{}" WHERE "{}" = ?"#, + self.table, self.key_field + ); let req = self.create_d1_query_request(&query, vec![path.into()], Operation::Delete, "Delete")?; diff --git a/core/services/gridfs/src/backend.rs b/core/services/gridfs/src/backend.rs index 2a14a879e..88200ca19 100644 --- a/core/services/gridfs/src/backend.rs +++ b/core/services/gridfs/src/backend.rs @@ -223,10 +223,9 @@ impl Service for GridfsBackend { if p == build_abs_path(&self.root, "") { Ok(RpStat::new(Metadata::new(EntryMode::DIR))) } else { - let bs = self.core.get(&p).await?; - match bs { - Some(bs) => Ok(RpStat::new( - Metadata::new(EntryMode::FILE).with_content_length(bs.len() as u64), + match self.core.get_length(&p).await? { + Some(len) => Ok(RpStat::new( + Metadata::new(EntryMode::FILE).with_content_length(len as u64), )), None => Err(Error::new(ErrorKind::NotFound, "kv not found in gridfs")), } diff --git a/core/services/gridfs/src/core.rs b/core/services/gridfs/src/core.rs index 3f5f9a9bb..2256afc3a 100644 --- a/core/services/gridfs/src/core.rs +++ b/core/services/gridfs/src/core.rs @@ -67,12 +67,17 @@ impl GridfsCore { .await } + async fn get_one_doc(bucket: &GridFsBucket, path: &str) -> Result<Option<FilesCollectionDocument>> { + let filter = doc! { "filename": path }; + let doc = bucket.find_one(filter).await.map_err(parse_mongodb_error)?; + Ok(doc) + } + pub async fn get(&self, path: &str) -> Result<Option<Buffer>> { let bucket = self.get_bucket().await?; - let filter = doc! { "filename": path }; - let Some(doc) = bucket.find_one(filter).await.map_err(parse_mongodb_error)? else { - return Ok(None); - }; + let Some(doc) = get_one_doc(&bucket, path).await? else { + return Ok(None) + } let mut destination = Vec::new(); let file_id = doc.id; @@ -87,12 +92,21 @@ impl GridfsCore { Ok(Some(Buffer::from(destination))) } + /// Get the byte length of the file. + pub async fn get_length(&self, path: &str) -> Result<Option<usize>> { + let bucket = self.get_bucket().await?; + let Some(doc) = get_one_doc(&bucket, path).await? else { + return Ok(None) + } + + Ok(Some(doc.length as usize)) + } + pub async fn set(&self, path: &str, value: Buffer) -> Result<()> { let bucket = self.get_bucket().await?; // delete old file if exists - let filter = doc! { "filename": path }; - if let Some(doc) = bucket.find_one(filter).await.map_err(parse_mongodb_error)? { + if let Some(doc) = get_one_doc(&bucket, path).await? { let file_id = doc.id; bucket.delete(file_id).await.map_err(parse_mongodb_error)?; }; @@ -113,8 +127,7 @@ impl GridfsCore { pub async fn delete(&self, path: &str) -> Result<()> { let bucket = self.get_bucket().await?; - let filter = doc! { "filename": path }; - let Some(doc) = bucket.find_one(filter).await.map_err(parse_mongodb_error)? else { + let Some(doc) = get_one_doc(&bucket, path).await? else { return Ok(()); }; diff --git a/core/services/mysql/src/core.rs b/core/services/mysql/src/core.rs index 807688300..961c29807 100644 --- a/core/services/mysql/src/core.rs +++ b/core/services/mysql/src/core.rs @@ -45,6 +45,16 @@ impl MysqlCore { pub async fn get(&self, path: &str) -> Result<Option<Buffer>> { let pool = self.get_client().await?; + // MySQL uses a backtick for identifier quote. An identifier may or may not be case-sensitive, + // depending on database configuration. Only a quoted identifier can have special characters. + // Read more: + // https://dev.mysql.com/doc/refman/9.7/en/identifiers.html + // https://dev.mysql.com/doc/refman/9.7/en/identifier-case-sensitivity.html + // + // We use: + // - formatted, quoted identifiers for trusted table and field configuration + // - bind parameters for values to avoid malformed SQL and SQL injection + // to ensure correctness. let value: Option<Vec<u8>> = sqlx::query_scalar(&format!( "SELECT `{}` FROM `{}` WHERE `{}` = ? LIMIT 1", self.value_field, self.table, self.key_field @@ -84,7 +94,7 @@ impl MysqlCore { sqlx::query(&format!( r#"INSERT INTO `{}` (`{}`, `{}`) VALUES (?, ?) - ON DUPLICATE KEY UPDATE `{}` = VALUES({})"#, + ON DUPLICATE KEY UPDATE `{}` = VALUES(`{}`)"#, self.table, self.key_field, self.value_field, self.value_field, self.value_field )) .bind(path) diff --git a/core/services/postgresql/src/core.rs b/core/services/postgresql/src/core.rs index 670d7c276..ff6107ab1 100644 --- a/core/services/postgresql/src/core.rs +++ b/core/services/postgresql/src/core.rs @@ -45,6 +45,14 @@ impl PostgresqlCore { pub async fn get(&self, path: &str) -> Result<Option<Buffer>> { let pool = self.get_client().await?; + // PostgreSQL uses `"` for identifier quote. A quoted identifier is case-sensitive and + // can contain special characters. + // Read more https://www.postgresql.org/docs/18/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS + // + // We use: + // - formatted, quoted identifiers for trusted table and field configuration + // - bind parameters for values to avoid malformed SQL and SQL injection + // to ensure correctness. let value: Option<Vec<u8>> = sqlx::query_scalar(&format!( r#"SELECT "{}" FROM "{}" WHERE "{}" = $1 LIMIT 1"#, self.value_field, self.table, self.key_field @@ -104,7 +112,7 @@ impl PostgresqlCore { let pool = self.get_client().await?; sqlx::query(&format!( - "DELETE FROM {} WHERE {} = $1", + r#"DELETE FROM "{}" WHERE "{}" = $1"#, self.table, self.key_field )) .bind(path) diff --git a/core/services/redis/src/backend.rs b/core/services/redis/src/backend.rs index 6afba8a8b..67aa3896d 100644 --- a/core/services/redis/src/backend.rs +++ b/core/services/redis/src/backend.rs @@ -348,6 +348,7 @@ impl Service for RedisBackend { } } } + fn read(&self, _ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> { let output: oio::StreamReader<RedisReader> = { Ok(oio::StreamReader::new(RedisReader::new( diff --git a/core/services/sqlite/src/backend.rs b/core/services/sqlite/src/backend.rs index 5a96dae0e..b32a9164e 100644 --- a/core/services/sqlite/src/backend.rs +++ b/core/services/sqlite/src/backend.rs @@ -240,14 +240,7 @@ impl Service for SqliteBackend { } else { format!("{}/", p) }; - let count: i64 = sqlx::query_scalar(&format!( - "SELECT COUNT(*) FROM `{}` WHERE `{}` LIKE $1 LIMIT 1", - self.core.table, self.core.key_field - )) - .bind(format!("{}%", dir_path)) - .fetch_one(self.core.get_client().await?) - .await - .map_err(parse_sqlite_error)?; + let count = self.core.count_under(&dir_path).await?; if count > 0 { // Directory exists (has children) diff --git a/core/services/sqlite/src/core.rs b/core/services/sqlite/src/core.rs index 16ee5f68a..478884595 100644 --- a/core/services/sqlite/src/core.rs +++ b/core/services/sqlite/src/core.rs @@ -48,8 +48,17 @@ impl SqliteCore { pub async fn get(&self, path: &str) -> Result<Option<Buffer>> { let pool = self.get_client().await?; + // SQLite prefers using standard `"` for identifier quotes instead of + // MySQL-compatible backtick ` or [] quotes. + // Identifier quotes are case-sensitive and allow special characters. + // Read more https://www.sqlite.org/quirks.html#double_quoted_string_literals_are_accepted + // + // We use: + // - formatted, quoted identifiers for trusted table and field configuration + // - bind parameters for values to avoid malformed SQL and SQL injection + // to ensure correctness. let value: Option<Vec<u8>> = sqlx::query_scalar(&format!( - "SELECT `{}` FROM `{}` WHERE `{}` = $1 LIMIT 1", + r#"SELECT "{}" FROM "{}" WHERE "{}" = $1 LIMIT 1"#, self.value_field, self.table, self.key_field )) .bind(path) @@ -64,7 +73,7 @@ impl SqliteCore { let pool = self.get_client().await?; let value: Option<i64> = sqlx::query_scalar(&format!( - "SELECT LENGTH(CAST(`{}` AS BLOB)) FROM `{}` WHERE `{}` = $1 LIMIT 1", + r#"SELECT LENGTH(CAST("{}" AS BLOB)) FROM "{}" WHERE "{}" = $1 LIMIT 1"#, self.value_field, self.table, self.key_field )) .bind(path) @@ -82,6 +91,19 @@ impl SqliteCore { .transpose() } + pub async fn count_under(&self, path: &str) -> Result<i64> { + let pool = self.get_client().await?; + + sqlx::query_scalar(&format!( + r#"SELECT COUNT(*) FROM "{}" WHERE "{}" LIKE $1 LIMIT 1"#, + self.table, self.key_field + )) + .bind(format!("{}%", path)) + .fetch_one(pool) + .await + .map_err(parse_sqlite_error) + } + pub async fn get_range( &self, path: &str, @@ -89,31 +111,57 @@ impl SqliteCore { limit: Option<isize>, ) -> Result<Option<(Buffer, u64)>> { let pool = self.get_client().await?; - let query = match limit { - Some(limit) => format!( - "SELECT SUBSTR(CAST(`{}` AS BLOB), {}, {}), LENGTH(CAST(`{}` AS BLOB)) FROM `{}` WHERE `{}` = $1 LIMIT 1", - self.value_field, - start + 1, - limit, - self.value_field, - self.table, - self.key_field - ), - None => format!( - "SELECT SUBSTR(CAST(`{}` AS BLOB), {}), LENGTH(CAST(`{}` AS BLOB)) FROM `{}` WHERE `{}` = $1 LIMIT 1", - self.value_field, - start + 1, - self.value_field, - self.table, - self.key_field - ), + if start < 0 || limit.is_some_and(|v| v < 0) { + return Err(Error::new( + ErrorKind::Unexpected, + "sqlite range contains negative value", + )); + } + + let start = start.checked_add(1).ok_or_else(|| { + Error::new( + ErrorKind::Unexpected, + "sqlite range start exceeds supported value", + ) + })?; + let start: i64 = start.try_into().map_err(|err| { + Error::new( + ErrorKind::Unexpected, + "sqlite range start exceeds supported value", + ) + .set_source(err) + })?; + let value = match limit { + Some(limit) => { + let limit: i64 = limit.try_into().map_err(|err| { + Error::new( + ErrorKind::Unexpected, + "sqlite range size exceeds supported value", + ) + .set_source(err) + })?; + sqlx::query_as(&format!( + r#"SELECT SUBSTR(CAST("{}" AS BLOB), $1, $2), LENGTH(CAST("{}" AS BLOB)) FROM "{}" WHERE "{}" = $3 LIMIT 1"#, + self.value_field, self.value_field, self.table, self.key_field + )) + .bind(start) + .bind(limit) + .bind(path) + .fetch_optional(pool) + .await + } + None => { + sqlx::query_as(&format!( + r#"SELECT SUBSTR(CAST("{}" AS BLOB), $1), LENGTH(CAST("{}" AS BLOB)) FROM "{}" WHERE "{}" = $2 LIMIT 1"#, + self.value_field, self.value_field, self.table, self.key_field + )) + .bind(start) + .bind(path) + .fetch_optional(pool) + .await + } }; - - let value: Option<(Vec<u8>, i64)> = sqlx::query_as(&query) - .bind(path) - .fetch_optional(pool) - .await - .map_err(parse_sqlite_error)?; + let value: Option<(Vec<u8>, i64)> = value.map_err(parse_sqlite_error)?; Ok(value.map(|(bs, size)| (Buffer::from(bs), size as u64))) } @@ -122,7 +170,7 @@ impl SqliteCore { let pool = self.get_client().await?; sqlx::query(&format!( - "INSERT OR REPLACE INTO `{}` (`{}`, `{}`) VALUES ($1, $2)", + r#"INSERT OR REPLACE INTO "{}" ("{}", "{}") VALUES ($1, $2)"#, self.table, self.key_field, self.value_field, )) .bind(path) @@ -138,7 +186,7 @@ impl SqliteCore { let pool = self.get_client().await?; sqlx::query(&format!( - "DELETE FROM `{}` WHERE `{}` = $1", + r#"DELETE FROM "{}" WHERE "{}" = $1"#, self.table, self.key_field )) .bind(path)
