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
The following commit(s) were added to refs/heads/fix-stat-impl by this push:
new e42d2cc9a fix: format sql identifiers in sql services
e42d2cc9a is described below
commit e42d2cc9a97b45f7a386c761ad47328494ba5bf4
Author: Erick Guan <[email protected]>
AuthorDate: Sun Jul 5 12:30:36 2026 +0800
fix: format sql identifiers in sql services
---
core/services/mysql/src/core.rs | 81 ++++++++----------
core/services/postgresql/src/core.rs | 67 +++++++--------
core/services/sqlite/src/core.rs | 162 +++++++++++++++++++----------------
3 files changed, 160 insertions(+), 150 deletions(-)
diff --git a/core/services/mysql/src/core.rs b/core/services/mysql/src/core.rs
index 00f3564a8..961c29807 100644
--- a/core/services/mysql/src/core.rs
+++ b/core/services/mysql/src/core.rs
@@ -51,21 +51,18 @@ impl MysqlCore {
// https://dev.mysql.com/doc/refman/9.7/en/identifiers.html
//
https://dev.mysql.com/doc/refman/9.7/en/identifier-case-sensitivity.html
//
- // When ANSI_QUOTES SQL mode is enabled, a string literal must be
quoted as well.
- //
// We use:
- // - parameterized query to avoid SQL injection
- // - quoted identifiers and string literal quote
+ // - 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("SELECT `?` FROM `?` WHERE `?` = '?' LIMIT 1")
- .bind(self.value_field.clone())
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_mysql_error)?;
+ let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
+ "SELECT `{}` FROM `{}` WHERE `{}` = ? LIMIT 1",
+ self.value_field, self.table, self.key_field
+ ))
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_mysql_error)?;
Ok(value.map(Buffer::from))
}
@@ -73,15 +70,14 @@ impl MysqlCore {
pub async fn get_length(&self, path: &str) -> Result<Option<usize>> {
let pool = self.get_client().await?;
- let value: Option<i64> =
- sqlx::query_scalar("SELECT OCTET_LENGTH(`?`) FROM `?` WHERE `?` =
'?' LIMIT 1")
- .bind(self.value_field.clone())
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_mysql_error)?;
+ let value: Option<i64> = sqlx::query_scalar(&format!(
+ "SELECT OCTET_LENGTH(`{}`) FROM `{}` WHERE `{}` = ? LIMIT 1",
+ self.value_field, self.table, self.key_field
+ ))
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_mysql_error)?;
value
.map(|v| {
@@ -96,16 +92,12 @@ impl MysqlCore {
pub async fn set(&self, path: &str, value: Buffer) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(
- "INSERT INTO `?` (`?`, `?`) VALUES ('?', ?)
- ON DUPLICATE KEY UPDATE `?` = VALUES(`?`)",
- )
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(self.value_field.clone())
+ sqlx::query(&format!(
+ r#"INSERT INTO `{}` (`{}`, `{}`) VALUES (?, ?)
+ ON DUPLICATE KEY UPDATE `{}` = VALUES(`{}`)"#,
+ self.table, self.key_field, self.value_field, self.value_field,
self.value_field
+ ))
.bind(path)
- .bind(self.value_field.clone())
- .bind(self.value_field.clone())
.bind(value.to_vec())
.execute(pool)
.await
@@ -117,13 +109,14 @@ impl MysqlCore {
pub async fn delete(&self, path: &str) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query("DELETE FROM `?` WHERE `?` = '?'")
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(path)
- .execute(pool)
- .await
- .map_err(parse_mysql_error)?;
+ sqlx::query(&format!(
+ "DELETE FROM `{}` WHERE `{}` = ?",
+ self.table, self.key_field
+ ))
+ .bind(path)
+ .execute(pool)
+ .await
+ .map_err(parse_mysql_error)?;
Ok(())
}
@@ -131,13 +124,15 @@ impl MysqlCore {
pub async fn list(&self, path: &str) -> Result<Vec<String>> {
let pool = self.get_client().await?;
+ let mut sql = format!(
+ "SELECT `{}` FROM `{}` WHERE `{}` LIKE ?",
+ self.key_field, self.table, self.key_field
+ );
+ sql.push_str(&format!(" ORDER BY `{}`", self.key_field));
+
let escaped = escape_like(path);
- sqlx::query_scalar("SELECT `?` FROM `?` WHERE `?` LIKE '?' ORDER BY
`?`")
- .bind(self.key_field.clone())
- .bind(self.table.clone())
- .bind(self.key_field.clone())
+ sqlx::query_scalar(&sql)
.bind(format!("{escaped}%"))
- .bind(self.key_field.clone())
.fetch_all(pool)
.await
.map_err(parse_mysql_error)
diff --git a/core/services/postgresql/src/core.rs
b/core/services/postgresql/src/core.rs
index 7eae9415b..ff6107ab1 100644
--- a/core/services/postgresql/src/core.rs
+++ b/core/services/postgresql/src/core.rs
@@ -45,24 +45,22 @@ impl PostgresqlCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
- // PostgreSQL uses `"` for identifier quote. An quoted identifier is
case-sensitive and
+ // PostgreSQL uses `"` for identifier quote. A quoted identifier is
case-sensitive and
// can contain special characters.
- // PostresSQL uses `'` for string literal quote.
// Read more
https://www.postgresql.org/docs/18/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS
//
// We use:
- // - parameterized query to avoid SQL injection
- // - quoted identifiers and string literal quote
+ // - 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(r#"SELECT "$1" FROM "$2" WHERE "$3" = '$4'
LIMIT 1"#)
- .bind(self.value_field.clone())
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_postgres_error)?;
+ let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
+ r#"SELECT "{}" FROM "{}" WHERE "{}" = $1 LIMIT 1"#,
+ self.value_field, self.table, self.key_field
+ ))
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_postgres_error)?;
Ok(value.map(Buffer::from))
}
@@ -70,12 +68,10 @@ impl PostgresqlCore {
pub async fn get_length(&self, path: &str) -> Result<Option<usize>> {
let pool = self.get_client().await?;
- let value: Option<i64> = sqlx::query_scalar(
- r#"SELECT OCTET_LENGTH("$1")::BIGINT FROM "$2" WHERE "$3" = '$4'
LIMIT 1"#,
- )
- .bind(self.value_field.clone())
- .bind(self.table.clone())
- .bind(self.key_field.clone())
+ let value: Option<i64> = sqlx::query_scalar(&format!(
+ r#"SELECT OCTET_LENGTH("{}")::BIGINT FROM "{}" WHERE "{}" = $1
LIMIT 1"#,
+ self.value_field, self.table, self.key_field
+ ))
.bind(path)
.fetch_optional(pool)
.await
@@ -94,15 +90,15 @@ impl PostgresqlCore {
pub async fn set(&self, path: &str, value: Buffer) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(
- r#"INSERT INTO "$1" ("$2", "$3")
- VALUES ('$4', $5)
- ON CONFLICT ("$2")
- DO UPDATE SET "$3" = EXCLUDED."$3""#,
- )
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(self.value_field.clone())
+ let table = &self.table;
+ let key_field = &self.key_field;
+ let value_field = &self.value_field;
+ sqlx::query(&format!(
+ r#"INSERT INTO "{table}" ("{key_field}", "{value_field}")
+ VALUES ($1, $2)
+ ON CONFLICT ("{key_field}")
+ DO UPDATE SET "{value_field}" = EXCLUDED."{value_field}""#,
+ ))
.bind(path)
.bind(value.to_vec())
.execute(pool)
@@ -115,13 +111,14 @@ impl PostgresqlCore {
pub async fn delete(&self, path: &str) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(r#"DELETE FROM "$1" WHERE "$2" = '$3'"#)
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(path)
- .execute(pool)
- .await
- .map_err(parse_postgres_error)?;
+ sqlx::query(&format!(
+ r#"DELETE FROM "{}" WHERE "{}" = $1"#,
+ self.table, self.key_field
+ ))
+ .bind(path)
+ .execute(pool)
+ .await
+ .map_err(parse_postgres_error)?;
Ok(())
}
diff --git a/core/services/sqlite/src/core.rs b/core/services/sqlite/src/core.rs
index 4bd52b5bf..478884595 100644
--- a/core/services/sqlite/src/core.rs
+++ b/core/services/sqlite/src/core.rs
@@ -34,12 +34,6 @@ pub struct SqliteCore {
pub value_field: String,
}
-#[derive(Debug)]
-struct SqliteBlobSized {
- bs: Vec<u8>,
- size: i64,
-}
-
impl SqliteCore {
pub async fn get_client(&self) -> Result<&SqlitePool> {
self.pool
@@ -54,24 +48,23 @@ impl SqliteCore {
pub async fn get(&self, path: &str) -> Result<Option<Buffer>> {
let pool = self.get_client().await?;
- // SQLite prefer using standard `"` for identifier quotes, `'` for
string literal quotes,
- // instead of MySQL-compatible backtick ` or [] quotes.
+ // 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:
- // - parameterized query to avoid SQL injection
- // - quoted identifiers and string literal quote
+ // - 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(r#"SELECT "$3" FROM "$1" WHERE "$2" = "$4"
LIMIT 1"#)
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(self.value_field.clone())
- .bind(path)
- .fetch_optional(pool)
- .await
- .map_err(parse_sqlite_error)?;
+ let value: Option<Vec<u8>> = sqlx::query_scalar(&format!(
+ r#"SELECT "{}" FROM "{}" WHERE "{}" = $1 LIMIT 1"#,
+ self.value_field, self.table, self.key_field
+ ))
+ .bind(path)
+ .fetch_optional(pool)
+ .await
+ .map_err(parse_sqlite_error)?;
Ok(value.map(Buffer::from))
}
@@ -79,12 +72,10 @@ impl SqliteCore {
pub async fn get_length(&self, path: &str) -> Result<Option<usize>> {
let pool = self.get_client().await?;
- let value: Option<i64> = sqlx::query_scalar(
- r#"SELECT LENGTH(CAST("$3" AS BLOB)) FROM "$1" WHERE "$2" = '$4'
LIMIT 1"#,
- )
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(self.value_field.clone())
+ let value: Option<i64> = sqlx::query_scalar(&format!(
+ r#"SELECT LENGTH(CAST("{}" AS BLOB)) FROM "{}" WHERE "{}" = $1
LIMIT 1"#,
+ self.value_field, self.table, self.key_field
+ ))
.bind(path)
.fetch_optional(pool)
.await
@@ -103,13 +94,14 @@ impl SqliteCore {
pub async fn count_under(&self, path: &str) -> Result<i64> {
let pool = self.get_client().await?;
- sqlx::query_scalar(r#"SELECT COUNT(*) FROM "$1" WHERE "$2" LIKE "$3"
LIMIT 1"#)
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(format!("{}%", path))
- .fetch_one(pool)
- .await
- .map_err(parse_sqlite_error)
+ 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(
@@ -119,48 +111,73 @@ impl SqliteCore {
limit: Option<isize>,
) -> Result<Option<(Buffer, u64)>> {
let pool = self.get_client().await?;
- let query = match limit {
- Some(limit) => sqlx::query_as(
- r#"SELECT SUBSTR(CAST("$3" AS BLOB), $4, $5), LENGTH(CAST("$3"
AS BLOB)) FROM "$1" WHERE "$2" = '$6' LIMIT 1"#,
+ 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",
)
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(self.value_field.clone())
- .bind(start+1)
- .bind(limit)
- .bind(path)
- ,
- None => sqlx::query_as(
- r#"SELECT SUBSTR(CAST("$3" AS BLOB), $4), LENGTH(CAST("$3" AS
BLOB)) FROM "$1" WHERE "$2" = '$5' LIMIT 1"#,
+ })?;
+ let start: i64 = start.try_into().map_err(|err| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "sqlite range start exceeds supported value",
)
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(self.value_field.clone())
- .bind(start+1)
- .bind(path)
- ,
+ .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)> = value.map_err(parse_sqlite_error)?;
- let value: Option<SqliteBlobSized> = query
- .fetch_optional(pool)
- .await
- .map_err(parse_sqlite_error)?;
-
- Ok(value.map(|s| (Buffer::from(s.bs), s.size as u64)))
+ Ok(value.map(|(bs, size)| (Buffer::from(bs), size as u64)))
}
pub async fn set(&self, path: &str, value: Buffer) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(r#"INSERT OR REPLACE INTO "$1" ("$2", "$3") VALUES ('$4',
$5)"#)
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(self.value_field.clone())
- .bind(path)
- .bind(value.to_vec())
- .execute(pool)
- .await
- .map_err(parse_sqlite_error)?;
+ sqlx::query(&format!(
+ r#"INSERT OR REPLACE INTO "{}" ("{}", "{}") VALUES ($1, $2)"#,
+ self.table, self.key_field, self.value_field,
+ ))
+ .bind(path)
+ .bind(value.to_vec())
+ .execute(pool)
+ .await
+ .map_err(parse_sqlite_error)?;
Ok(())
}
@@ -168,13 +185,14 @@ impl SqliteCore {
pub async fn delete(&self, path: &str) -> Result<()> {
let pool = self.get_client().await?;
- sqlx::query(r#"DELETE FROM "$1" WHERE "$2" = '$3'"#)
- .bind(self.table.clone())
- .bind(self.key_field.clone())
- .bind(path)
- .execute(pool)
- .await
- .map_err(parse_sqlite_error)?;
+ sqlx::query(&format!(
+ r#"DELETE FROM "{}" WHERE "{}" = $1"#,
+ self.table, self.key_field
+ ))
+ .bind(path)
+ .execute(pool)
+ .await
+ .map_err(parse_sqlite_error)?;
Ok(())
}
