rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-25 21:32:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(deps): upgrade sqlx to 0.9 to pick up webpki-roots 1.0.10 (#9862)

Browse Source 
sqlx 0.8 pins the unmaintained webpki-roots 0.26 line for its rustls
trust store, which still contains the recently compromised e-Tugra CA
root. Upgrading sqlx to 0.9 switches its dependency to webpki-roots
1.x, and the lockfile now resolves to 1.0.10 where the distrusted
root has been removed from the compiled-in Mozilla trust store.

sqlx 0.9 requires dynamic SQL strings to be explicitly audited via
the new SqlSafeStr trait: only &'static str literals are accepted
as-is. All affected call sites (MySQL KV backend, MySQL election,
integration tests and fuzz targets) now wrap their internally
generated SQL templates in AssertSqlSafe; the wrapped SQL is built
from operator config or test-local identifiers only, and dynamic
data keeps flowing through bind parameters.

The mysql-rsa feature is enabled alongside mysql to preserve 0.8
authentication behavior: sqlx 0.9 gates non-TLS RSA password auth
(caching_sha2_password / sha256_password) behind it.

Signed-off-by: Ning Sun <sunning@greptime.com>
This commit is contained in:
Ning Sun
2026-10-22 07:27:19 +00:00
committed by GitHub
parent 8b1d68369b
commit af52dfafd8
32 changed files with 672 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/meta/Cargo.toml
+1
View File
@@ -81,6 +81,7 @@ serde_with.workspace = true
snafu.workspace = true
sqlx = { workspace = true, features = [
"mysql",
"mysql-rsa",
"chrono",
], optional = true }
store-api.workspace = true
src/common/meta/src/election/rds/mysql.rs
+26 -15
View File
@@ -22,7 +22,7 @@ use snafu::{OptionExt, ResultExt, ensure};
use sqlx::mysql::{MySqlArguments, MySqlRow};
use sqlx::pool::PoolConnection;
use sqlx::query::Query;
use sqlx::{MySql, MySqlPool, MySqlTransaction, Row};
use sqlx::{AssertSqlSafe, MySql, MySqlPool, MySqlTransaction, Row};
use tokio::sync::{Mutex, MutexGuard, broadcast};
use tokio::time::MissedTickBehavior;
@@ -38,6 +38,17 @@ use crate::error::{
};
use crate::key::{CANDIDATES_ROOT, ELECTION_KEY};
/// Builds a [`Query`] from a dynamic SQL template.
///
/// sqlx 0.9 requires `&'static str` or [`AssertSqlSafe`] to reject unaudited
/// dynamic SQL strings. The templates here are generated by
/// [`ElectionSqlFactory`] from the configured table name (operator config,
/// not user input), and all dynamic data is passed via bind parameters,
/// so it is safe to mark them with [`AssertSqlSafe`].
fn template_query(sql: &str) -> Query<'_, MySql, MySqlArguments> {
sqlx::query(AssertSqlSafe(sql))
}
struct ElectionSqlFactory<'a> {
table_name: &'a str,
}
@@ -268,11 +279,11 @@ impl ElectionMysqlClient {
async fn ensure_table_exists(&mut self) -> Result<()> {
let create_table_sql = self.create_table_sql();
let query = sqlx::query(&create_table_sql);
let query = template_query(&create_table_sql);
self.execute(query, &create_table_sql).await?;
// Insert at least one row for `SELECT * FOR UPDATE` to work.
let insert_once_sql = self.insert_once_sql();
let query = sqlx::query(&insert_once_sql);
let query = template_query(&insert_once_sql);
self.execute(query, &insert_once_sql).await?;
Ok(())
}
@@ -324,7 +335,7 @@ impl ElectionMysqlClient {
/// if `current` is `None`.
async fn check_version(&mut self, sql: &str) -> Result<()> {
// Check if the MySQL version is supported.
let query = sqlx::query(sql);
let query = template_query(sql);
// Safety: `maybe_init_client` ensures `current` is not `None`.
let client = self.current.as_mut().unwrap();
let row = tokio::time::timeout(self.execution_timeout, query.fetch_one(&mut **client))
@@ -692,7 +703,7 @@ impl MySqlElection {
executor: &mut Executor<'_>,
) -> Result<Option<Lease>> {
let key = key.as_bytes();
let query = sqlx::query(&self.sql_set.get_value_with_lease).bind(key);
let query = template_query(&self.sql_set.get_value_with_lease).bind(key);
let res = executor
.query(query, &self.sql_set.get_value_with_lease)
.await?;
@@ -728,7 +739,7 @@ impl MySqlElection {
executor: &mut Executor<'_>,
) -> Result<(Vec<(String, Timestamp)>, Timestamp)> {
let key_prefix = format!("{}%", key_prefix).as_bytes().to_vec();
let query = sqlx::query(&self.sql_set.get_value_with_lease_by_prefix).bind(key_prefix);
let query = template_query(&self.sql_set.get_value_with_lease_by_prefix).bind(key_prefix);
let res = executor
.query(query, &self.sql_set.get_value_with_lease_by_prefix)
.await?;
@@ -765,7 +776,7 @@ impl MySqlElection {
let prev = prev.as_bytes();
let updated = updated.as_bytes();
let query = sqlx::query(&self.sql_set.update_value_with_lease)
let query = template_query(&self.sql_set.update_value_with_lease)
.bind(updated)
.bind(lease_ttl as f64)
.bind(key)
@@ -794,7 +805,7 @@ impl MySqlElection {
) -> Result<bool> {
let key = key.as_bytes();
let lease_ttl_secs = lease_ttl_secs as f64;
let query = sqlx::query(&self.sql_set.put_value_with_lease)
let query = template_query(&self.sql_set.put_value_with_lease)
.bind(key)
.bind(value)
.bind(lease_ttl_secs);
@@ -808,7 +819,7 @@ impl MySqlElection {
/// Caution: Should only delete the key if the lease is expired.
async fn delete_value(&self, key: &str, executor: &mut Executor<'_>) -> Result<bool> {
let key = key.as_bytes();
let query = sqlx::query(&self.sql_set.delete_value).bind(key);
let query = template_query(&self.sql_set.delete_value).bind(key);
let res = executor.execute(query, &self.sql_set.delete_value).await?;
Ok(res == 1)
@@ -837,7 +848,7 @@ impl MySqlElection {
let mut client = self.client.lock().await;
let txn = client.transaction().await?;
let mut executor = Executor::Txn(txn);
let query = sqlx::query(&self.sql_set.campaign);
let query = template_query(&self.sql_set.campaign);
executor.query(query, &self.sql_set.campaign).await?;
// Safety: Checked if lease is not None above.
self.renew_lease(executor, lease.unwrap()).await?;
@@ -858,7 +869,7 @@ impl MySqlElection {
let mut client = self.client.lock().await;
let txn = client.transaction().await?;
let mut executor = Executor::Txn(txn);
let query = sqlx::query(&self.sql_set.campaign);
let query = template_query(&self.sql_set.campaign);
executor.query(query, &self.sql_set.campaign).await?;
self.elected(executor, lease).await?;
}
@@ -869,7 +880,7 @@ impl MySqlElection {
let mut client = self.client.lock().await;
let txn = client.transaction().await?;
let mut executor = Executor::Txn(txn);
let query = sqlx::query(&self.sql_set.campaign);
let query = template_query(&self.sql_set.campaign);
executor.query(query, &self.sql_set.campaign).await?;
// Safety: Checked if lease is not None above.
self.renew_lease(executor, lease.unwrap()).await?;
@@ -1027,7 +1038,7 @@ mod tests {
async fn drop_table(client: &Mutex<ElectionMysqlClient>, table_name: &str) {
let mut client = client.lock().await;
let sql = format!("DROP TABLE IF EXISTS `{}`;", table_name);
client.execute(sqlx::query(&sql), &sql).await.unwrap();
client.execute(template_query(&sql), &sql).await.unwrap();
}
#[tokio::test]
@@ -1052,7 +1063,7 @@ mod tests {
let txn = a.transaction().await.unwrap();
let mut executor = Executor::Txn(txn);
let raw_query = format!("SELECT * FROM `{}` FOR UPDATE;", table_name);
let query = sqlx::query(&raw_query);
let query = template_query(&raw_query);
let _ = executor.query(query, &raw_query).await.unwrap();
}
@@ -1258,7 +1269,7 @@ mod tests {
let txn = client.transaction().await.unwrap();
let mut executor = Executor::Txn(txn);
let raw_query = format!("SELECT * FROM `{}` FOR UPDATE;", table_name);
let query = sqlx::query(&raw_query);
let query = template_query(&raw_query);
let _ = executor.query(query, &raw_query).await.unwrap();
election.elected(executor, expected_lease).await
}
src/common/meta/src/kv_backend/rds/mysql.rs
+50 -27
View File
@@ -21,7 +21,7 @@ use regex::Regex;
use snafu::{OptionExt, ResultExt};
use sqlx::mysql::MySqlRow;
use sqlx::pool::Pool;
use sqlx::{MySql, MySqlPool, Row, Transaction as MySqlTransaction};
use sqlx::{AssertSqlSafe, MySql, MySqlPool, Row, Transaction as MySqlTransaction};
use strum::AsRefStr;
use crate::error::{
@@ -264,7 +264,9 @@ impl Executor for MySqlClient {
}
async fn query(&mut self, raw_query: &str, params: &[&Vec<u8>]) -> Result<Vec<KeyValue>> {
let query = sqlx::query(raw_query);
// Safety: `raw_query` is generated by `MySqlTemplateSet`, and all dynamic data
// is passed via bind parameters.
let query = sqlx::query(AssertSqlSafe(raw_query));
let query = params.iter().fold(query, |query, param| query.bind(param));
let rows = query
.fetch_all(&**self)
@@ -274,7 +276,9 @@ impl Executor for MySqlClient {
}
async fn execute(&mut self, raw_query: &str, params: &[&Vec<u8>]) -> Result<()> {
let query = sqlx::query(raw_query);
// Safety: `raw_query` is generated by `MySqlTemplateSet`, and all dynamic data
// is passed via bind parameters.
let query = sqlx::query(AssertSqlSafe(raw_query));
let query = params.iter().fold(query, |query, param| query.bind(param));
query
.execute(&**self)
@@ -303,7 +307,9 @@ impl Executor for MySqlClient {
#[async_trait::async_trait]
impl Transaction<'_> for MySqlTxnClient {
async fn query(&mut self, raw_query: &str, params: &[&Vec<u8>]) -> Result<Vec<KeyValue>> {
let query = sqlx::query(raw_query);
// Safety: `raw_query` is generated by `MySqlTemplateSet`, and all dynamic data
// is passed via bind parameters.
let query = sqlx::query(AssertSqlSafe(raw_query));
let query = params.iter().fold(query, |query, param| query.bind(param));
// As said in https://docs.rs/sqlx/latest/sqlx/trait.Executor.html, we need a `&mut *transaction`. Weird.
let rows = query
@@ -314,7 +320,9 @@ impl Transaction<'_> for MySqlTxnClient {
}
async fn execute(&mut self, raw_query: &str, params: &[&Vec<u8>]) -> Result<()> {
let query = sqlx::query(raw_query);
// Safety: `raw_query` is generated by `MySqlTemplateSet`, and all dynamic data
// is passed via bind parameters.
let query = sqlx::query(AssertSqlSafe(raw_query));
let query = params.iter().fold(query, |query, param| query.bind(param));
// As said in https://docs.rs/sqlx/latest/sqlx/trait.Executor.html, we need a `&mut *transaction`. Weird.
query
@@ -562,12 +570,14 @@ impl MySqlStore {
pool: &Pool<MySql>,
sql_template_set: &MySqlTemplateSet,
) -> Result<Option<String>> {
let row = sqlx::query(&sql_template_set.show_create_table_statement)
.fetch_optional(pool)
.await
.with_context(|_| MySqlExecutionSnafu {
sql: sql_template_set.show_create_table_statement.clone(),
})?;
let row = sqlx::query(AssertSqlSafe(
sql_template_set.show_create_table_statement.as_str(),
))
.fetch_optional(pool)
.await
.with_context(|_| MySqlExecutionSnafu {
sql: sql_template_set.show_create_table_statement.clone(),
})?;
Ok(row.map(|row| row.get(1)))
}
@@ -598,12 +608,14 @@ impl MySqlStore {
match Self::parse_value_column_blob_type(&create_table_sql) {
Some(ValueBlobType::Blob) => {
sqlx::query(&sql_template_set.alter_value_column_statement)
.execute(pool)
.await
.with_context(|_| MySqlExecutionSnafu {
sql: sql_template_set.alter_value_column_statement.clone(),
})?;
sqlx::query(AssertSqlSafe(
sql_template_set.alter_value_column_statement.as_str(),
))
.execute(pool)
.await
.with_context(|_| MySqlExecutionSnafu {
sql: sql_template_set.alter_value_column_statement.clone(),
})?;
info!("Upgraded MySQL metadata value column to MEDIUMBLOB for `{table_name}`");
}
Some(ValueBlobType::MediumBlob | ValueBlobType::LongBlob) => {
@@ -638,12 +650,14 @@ impl MySqlStore {
// We check if greptime_metakv table exists, and we will create a new table
// if it does not exist.
let sql_template_set = MySqlTemplateFactory::new(table_name).build();
sqlx::query(&sql_template_set.create_table_statement)
.execute(&pool)
.await
.with_context(|_| MySqlExecutionSnafu {
sql: sql_template_set.create_table_statement.clone(),
})?;
sqlx::query(AssertSqlSafe(
sql_template_set.create_table_statement.as_str(),
))
.execute(&pool)
.await
.with_context(|_| MySqlExecutionSnafu {
sql: sql_template_set.create_table_statement.clone(),
})?;
Self::maybe_upgrade_value_column_to_mediumblob(&pool, &sql_template_set).await?;
Ok(Arc::new(MySqlStore {
max_txn_ops,
@@ -695,7 +709,10 @@ mod tests {
async fn show_create_table(pool: &MySqlPool, table_name: &str) -> String {
let sql = format!("SHOW CREATE TABLE `{table_name}`");
let row = sqlx::query(&sql).fetch_one(pool).await.unwrap();
let row = sqlx::query(AssertSqlSafe(sql.as_str()))
.fetch_one(pool)
.await
.unwrap();
row.get::<String, _>(1)
}
@@ -703,18 +720,24 @@ mod tests {
let sql = format!(
"CREATE TABLE IF NOT EXISTS `{table_name}`(k VARBINARY(3072) PRIMARY KEY, v BLOB);"
);
sqlx::query(&sql).execute(pool).await.unwrap();
sqlx::query(AssertSqlSafe(sql.as_str()))
.execute(pool)
.await
.unwrap();
}
async fn drop_table(pool: &MySqlPool, table_name: &str) {
let sql = format!("DROP TABLE IF EXISTS `{table_name}`;");
sqlx::query(&sql).execute(pool).await.unwrap();
sqlx::query(AssertSqlSafe(sql.as_str()))
.execute(pool)
.await
.unwrap();
}
async fn build_mysql_kv_backend(table_name: &str) -> Option<MySqlStore> {
let pool = mysql_pool().await?;
let sql_templates = MySqlTemplateFactory::new(table_name).build();
sqlx::query(&sql_templates.create_table_statement)
sqlx::query(AssertSqlSafe(sql_templates.create_table_statement.as_str()))
.execute(&pool)
.await
.unwrap();
src/meta-srv/Cargo.toml
+1
View File
@@ -67,6 +67,7 @@ servers.workspace = true
snafu.workspace = true
sqlx = { workspace = true, features = [
"mysql",
"mysql-rsa",
"chrono",
], optional = true }
store-api.workspace = true